#!/usr/bin/env python3 """ Step 3: Map query results to template and generate HTML report. For 园区概览 (dashboard_overview) and 企业概览 (dashboard_enterprise), step2_query.py already used LLM to produce a flat dict of placeholder_values (including INJECT_DATA). This script just substitutes them into the HTML. For all other templates, uses LLM to map data to template format as before. """ import os import sys import json import time import argparse import urllib.request import re SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) TEMPLATES_DIR = os.path.join(SKILL_DIR, "templates") OUTPUT_DIR = os.path.join(SKILL_DIR, "output") LLM_API_URL = os.environ.get("LLM_API_URL", "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation") LLM_API_KEY = os.environ.get("LLM_API_KEY", "sk-b36adcdd25d64b7c847e7c941b5c1e03") LLM_MODEL = os.environ.get("LLM_MODEL", "qwen-plus") def call_llm(prompt, system="You are a helpful assistant.", model=None, max_tokens=None): use_model = model or LLM_MODEL headers = {"Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json"} parameters = {"result_format": "message"} if max_tokens: parameters["max_tokens"] = max_tokens payload = {"model": use_model, "input": {"messages": [ {"role": "system", "content": system}, {"role": "user", "content": prompt} ]}, "parameters": parameters} req = urllib.request.Request(LLM_API_URL, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST") try: resp = urllib.request.urlopen(req, timeout=60) return json.loads(resp.read().decode("utf-8"))["output"]["choices"][0]["message"]["content"] except Exception as e: print(json.dumps({"status": "error", "message": f"LLM error: {e}"})) sys.exit(1) def extract_json(text): for pattern in [r'```json\s*(.*?)\s*```', r'```\s*([\[{].*?)\s*```']: m = re.search(pattern, text, re.DOTALL) if m: try: return json.loads(m.group(1)) except: pass try: return json.loads(text.strip()) except: pass brace_match = re.search(r'(\{.*\})', text, re.DOTALL) if brace_match: try: return json.loads(brace_match.group(1)) except: pass return None def read_file(path): if os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: return f.read() return "" def inject_placeholders(html, placeholder_values): """ Replace all {{KEY}} placeholders in html with the corresponding values from placeholder_values dict. Special handling for INJECT_DATA: the template uses {{INJECT_DATA}} as a standalone placeholder (not prefixed with // ), so we replace it directly with the full `const mockData = {...};` JS string. """ for key, value in placeholder_values.items(): if value is None: value = "" html = html.replace("{{" + key + "}}", str(value)) return html def main(): parser = argparse.ArgumentParser() parser.add_argument("--query", required=True) parser.add_argument("--template", required=True) parser.add_argument("--result_file", required=True) args = parser.parse_args() template_name = args.template # Read result file written by step2_query.py try: with open(args.result_file, 'r', encoding='utf-8') as f: result_data = json.load(f) except Exception as e: print(json.dumps({"status": "error", "message": f"Failed to read result file: {e}"})) sys.exit(1) # Load HTML template html_path = os.path.join(TEMPLATES_DIR, f"{template_name}.html") if not os.path.exists(html_path): print(json.dumps({"status": "error", "message": f"Template HTML not found: {html_path}"})) sys.exit(1) html = read_file(html_path) # ----------------------------------------------------------------------- # Fixed template branch: placeholder_values already prepared by step2 # ----------------------------------------------------------------------- if isinstance(result_data, dict) and result_data.get("is_fixed_template"): placeholder_values = result_data.get("placeholder_values", {}) summary = result_data.get("summary", "数据分析报告已生成。") if not placeholder_values: # Fallback: if LLM in step2 failed to produce values, error out print(json.dumps({"status": "error", "message": "Fixed template: placeholder_values is empty. Check step2 output."})) sys.exit(1) html = inject_placeholders(html, placeholder_values) # ----------------------------------------------------------------------- # Normal flow: call LLM to map raw rows to template format # ----------------------------------------------------------------------- else: rows = result_data.get("rows", []) template_doc = result_data.get("template_doc", "") summary = "数据分析报告已生成。" if not rows: print(json.dumps({"status": "error", "message": "Normal template: rows is empty. Check step2 output."})) sys.exit(1) # Truncate rows to avoid exceeding token limit rows_str = json.dumps(rows, ensure_ascii=False, separators=(',', ':')) if len(rows_str) > 12000: rows_str = rows_str[:12000] + "...(truncated)" map_resp = call_llm( f"USER QUERY: {args.query}\n\nQUERY RESULTS ({len(rows)} rows):\n{rows_str}\n\nTEMPLATE SPEC:\n{template_doc}\n\nMap ALL data to the template JSON format. Return ONLY valid JSON with all required template fields plus a 'summary' field (2-3 sentences Chinese analysis of the data).", system="You are a data visualization engineer. Map query results to the template format exactly as specified. Include ALL data rows. Return ONLY valid JSON.", model=LLM_MODEL, max_tokens=4096 ) if not map_resp: print(json.dumps({"status": "error", "message": "Failed to map data with LLM"})) sys.exit(1) mapped = extract_json(map_resp) if not mapped: print(json.dumps({"status": "error", "message": "Failed to parse LLM mapped data", "raw": map_resp[:500]})) sys.exit(1) summary = mapped.pop("summary", "数据分析报告已生成。") # Render: text_values + js_variables text_values = mapped.get("data", {}).get("text_values", {}) for key, val in text_values.items(): html = html.replace(f"{{{{{key}}}}}", str(val)) js_vars = mapped.get("data", {}).get("js_variables", {}) if js_vars: js_lines = [f"var {k} = {json.dumps(v, ensure_ascii=False)};" for k, v in js_vars.items()] injected = "\n ".join(js_lines) if "// {{INJECT_DATA}}" in html: html = html.replace("// {{INJECT_DATA}}", injected) elif "{{INJECT_DATA}}" in html: html = html.replace("{{INJECT_DATA}}", injected) # ----------------------------------------------------------------------- # Write output HTML # ----------------------------------------------------------------------- timestamp = int(time.time()) out_file = f"report_{timestamp}.html" out_path = os.path.join(OUTPUT_DIR, out_file) os.makedirs(OUTPUT_DIR, exist_ok=True) with open(out_path, 'w', encoding='utf-8') as f: f.write(html) # Clean up temp file try: os.remove(args.result_file) except: pass print(json.dumps({ "status": "success", "message": "Report generated successfully", "template": template_name, "address": f"/skill/sql-report-generator/output/{out_file}", "summary": summary.strip() if isinstance(summary, str) else str(summary) }, ensure_ascii=False)) if __name__ == "__main__": main()