#!/usr/bin/env python3 """ Step 2: Generate SQL based on selected views and template, then execute it. For 园区概览 (dashboard_overview) and 企业概览 (dashboard_enterprise), uses fixed SQL templates with dynamic parameters. Supports both year-level (date_type=year) and month-level (date_type=month) queries. Actual database views (confirmed): - 园区企业能耗年数据: year, park_name, enterprise_name, energy_consumption_electricity, energy_consumption_heat, energy_consumption_gas - 园区企业能耗月数据: year, month, park_name, enterprise_name, energy_consumption_electricity, energy_consumption_heat, energy_consumption_gas - 园区企业碳排年数据: year, park_name, enterprise_name, carbon_emissions_electricity, carbon_emissions_heat, carbon_emissions_gas - 园区企业碳排月数据: year, month, park_name, enterprise_name, carbon_emissions_electricity, carbon_emissions_heat, carbon_emissions_gas - 园区企业光伏节碳年数据: year, park_name, enterprise_name, save_carbon, power_generation - 园区企业光伏节碳月数据: year, month, park_name, enterprise_name, save_carbon, power_generation - energy_price_view: year, month, energy_type, hourly_price, pricing_name - unified_device_view: park_name, enterprise_name, terminal_category, terminal_type, device_status """ import os import sys import json import argparse import urllib.request import re import datetime from concurrent.futures import ThreadPoolExecutor, as_completed SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) TEMPLATES_DIR = os.path.join(SKILL_DIR, "templates") REFERENCES_DIR = os.path.join(SKILL_DIR, "references") VIEWS_DIR = os.path.join(REFERENCES_DIR, "views") 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_SQL = os.environ.get("LLM_MODEL_SQL", "qwen-turbo") LLM_MODEL = os.environ.get("LLM_MODEL", "qwen-plus") DB_API_URL = "http://47.97.121.32:8080/VOLERTECH/app/api/oauth/getmysql" DB_JDBC = "snowy_cloud_mysql_copy" # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def call_llm(prompt, system="You are a helpful assistant.", model=None, max_tokens=None): use_model = model or LLM_MODEL_SQL 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 load_view_schemas(view_names): schemas = [] for vname in view_names: safe = re.sub(r'[^\w\u4e00-\u9fff]', '_', vname.strip()) path = os.path.join(VIEWS_DIR, f"{safe}.md") if not os.path.exists(path): for fname in os.listdir(VIEWS_DIR): if vname.strip() in fname or fname.replace('.md', '') in vname: path = os.path.join(VIEWS_DIR, fname) break if os.path.exists(path): schemas.append(read_file(path)) return "\n\n---\n\n".join(schemas) def sanitize_sql(sql): sql = re.sub(r'"([^"]*[\u4e00-\u9fff][^"]*)"', lambda m: f'`{m.group(1)}`', sql) sql = re.sub(r'"([^"]+\s[^"]+)"', lambda m: f'`{m.group(1)}`', sql) return sql def fix_group_by_sql(sql): """ 当 MySQL only_full_group_by 模式报错时，自动将 SELECT 中未聚合且不在 GROUP BY 中的列包裹 ANY_VALUE()。 """ import re # 提取 GROUP BY 子句中的列名 gb_match = re.search(r'GROUP\s+BY\s+([^\n;ORDER]+)', sql, re.IGNORECASE) if not gb_match: return sql gb_cols = [c.strip().strip('`') for c in gb_match.group(1).split(',')] # 提取 SELECT 子句 sel_match = re.search(r'SELECT\s+(.*?)\s+FROM', sql, re.IGNORECASE | re.DOTALL) if not sel_match: return sql sel_str = sel_match.group(1) new_cols = [] for col in sel_str.split(','): col = col.strip() col_name = re.sub(r'^`|`$', '', col.split()[-1]) # 去掉反引号和别名 is_agg = bool(re.match(r'(SUM|AVG|MAX|MIN|COUNT|ANY_VALUE)\s*\(', col, re.IGNORECASE)) in_gb = any(col_name == g or col_name in g for g in gb_cols) if not is_agg and not in_gb: col = f'ANY_VALUE({col})' new_cols.append(col) new_sel = ', '.join(new_cols) return sql[:sel_match.start(1)] + new_sel + sql[sel_match.end(1):] def query_database(sql): sql = sanitize_sql(sql) req = urllib.request.Request( DB_API_URL, data=json.dumps({"sql": sql, "basejdbc": DB_JDBC}).encode("utf-8"), headers={"Content-Type": "application/json"} ) try: resp = urllib.request.urlopen(req, timeout=20) result = json.loads(resp.read().decode("utf-8")) # 检查是否是 only_full_group_by 报错，如果是则自动修复 SQL 并重试 if (isinstance(result, list) and result and isinstance(result[0], dict) and 'error' in result[0] and 'only_full_group_by' in result[0].get('error', '')): fixed_sql = fix_group_by_sql(sql) if fixed_sql != sql: req2 = urllib.request.Request( DB_API_URL, data=json.dumps({"sql": fixed_sql, "basejdbc": DB_JDBC}).encode("utf-8"), headers={"Content-Type": "application/json"} ) resp2 = urllib.request.urlopen(req2, timeout=20) return json.loads(resp2.read().decode("utf-8")) return result except Exception as e: print(json.dumps({"status": "error", "message": f"DB error: {e}"})) sys.exit(1) def extract_rows(db_result): if db_result is None: return [] if isinstance(db_result, list): return db_result for k in ("data", "rows", "result", "records"): if k in db_result and isinstance(db_result[k], list): return db_result[k] return [] # --------------------------------------------------------------------------- # Fixed SQL templates # --------------------------------------------------------------------------- def get_default_date(): today = datetime.date.today() first = today.replace(day=1) last_month = first - datetime.timedelta(days=1) return last_month.strftime("%Y-%m") def build_park_overview_sqls(date_type, date_str, park_name): """ Build fixed SQL queries for dashboard_overview. date_type: 'year' or 'month' date_str: '2025' (year) or '2025-03' (month) """ park_filter = f"AND park_name = '{park_name}'" if park_name else "" if date_type == "year": year = date_str # e.g. "2025" # Year-level views energy_view = "园区企业能耗年数据" carbon_view = "园区企业碳排年数据" pv_view = "园区企业光伏节碳年数据" time_filter_energy = f"WHERE year = {year} {park_filter}" time_filter_carbon = f"WHERE year = {year} {park_filter}" time_filter_pv = f"WHERE year = {year} {park_filter}" # For trend: monthly breakdown within the year trend_view = "园区企业能耗月数据" trend_filter = f"WHERE year = {year} {park_filter}" trend_group = "month" pv_trend_view = "园区企业光伏节碳月数据" pv_trend_filter = f"WHERE year = {year} {park_filter}" carbon_trend_view = "园区企业碳排月数据" carbon_trend_filter = f"WHERE year = {year} {park_filter}" rank_filter = f"WHERE year = {year} {park_filter}" price_filter = f"WHERE year = {year}" else: # Month-level: date_str = "YYYY-MM" parts = date_str.split("-") year = int(parts[0]) month = int(parts[1]) energy_view = "园区企业能耗月数据" carbon_view = "园区企业碳排月数据" pv_view = "园区企业光伏节碳月数据" time_filter_energy = f"WHERE year = {year} AND month = {month} {park_filter}" time_filter_carbon = f"WHERE year = {year} AND month = {month} {park_filter}" time_filter_pv = f"WHERE year = {year} AND month = {month} {park_filter}" # Trend: all months in the same year up to this month trend_view = "园区企业能耗月数据" trend_filter = f"WHERE year = {year} AND month <= {month} {park_filter}" trend_group = "month" pv_trend_view = "园区企业光伏节碳月数据" pv_trend_filter = f"WHERE year = {year} AND month <= {month} {park_filter}" carbon_trend_view = "园区企业碳排月数据" carbon_trend_filter = f"WHERE year = {year} AND month <= {month} {park_filter}" rank_filter = f"WHERE year = {year} AND month = {month} {park_filter}" price_filter = f"WHERE year = {year} AND month = {month}" return [ { "key": "current_energy", "desc": "能耗汇总（电/热/燃气）", "sql": f"""SELECT SUM(energy_consumption_electricity) AS ele_energy, SUM(energy_consumption_heat) AS heat_energy, SUM(energy_consumption_gas) AS gas_energy FROM `{energy_view}` {time_filter_energy}""" }, { "key": "current_carbon", "desc": "碳排汇总（电/热/燃气）", "sql": f"""SELECT SUM(carbon_emissions_electricity) AS ele_carbon, SUM(carbon_emissions_heat) AS heat_carbon, SUM(carbon_emissions_gas) AS gas_carbon FROM `{carbon_view}` {time_filter_carbon}""" }, { "key": "current_pv", "desc": "光伏节碳汇总", "sql": f"""SELECT SUM(save_carbon) AS pv_save_carbon, SUM(power_generation) AS pv_power FROM `{pv_view}` {time_filter_pv}""" }, { "key": "energy_price", "desc": "能源费用单价（电/热/燃气）", "sql": f"""SELECT energy_type, AVG(hourly_price) AS avg_price FROM `energy_price_view` {price_filter} GROUP BY energy_type""" }, { "key": "device_stats", "desc": "能耗终端设备分布统计（电力/热力/燃气）", "sql": f"""SELECT terminal_category AS device_type, COUNT(*) AS cnt FROM `unified_device_view` WHERE 1=1 {park_filter} GROUP BY terminal_category ORDER BY cnt DESC""" }, { "key": "pv_device_stats", "desc": "光伏设备统计", "sql": f"""SELECT COUNT(*) AS cnt FROM `光伏设备信息` WHERE 1=1 {park_filter.replace('park_name', 'parkName')}""" }, { "key": "enterprise_ranking", "desc": "企业能耗排名", "sql": f"""SELECT enterprise_name, park_name, SUM(energy_consumption_electricity + energy_consumption_heat + energy_consumption_gas) AS total_energy FROM `{energy_view}` {rank_filter} GROUP BY enterprise_name, park_name ORDER BY total_energy DESC LIMIT 20""" }, { "key": "monthly_energy_trend", "desc": "各月能耗趋势（电/热/燃气）", "sql": f"""SELECT {trend_group}, SUM(energy_consumption_electricity) AS ele_energy, SUM(energy_consumption_heat) AS heat_energy, SUM(energy_consumption_gas) AS gas_energy FROM `{trend_view}` {trend_filter} GROUP BY {trend_group} ORDER BY {trend_group}""" }, { "key": "monthly_carbon_trend", "desc": "各月碳排趋势（电/热/燃气）", "sql": f"""SELECT {trend_group}, SUM(carbon_emissions_electricity) AS ele_carbon, SUM(carbon_emissions_heat) AS heat_carbon, SUM(carbon_emissions_gas) AS gas_carbon FROM `{carbon_trend_view}` {carbon_trend_filter} GROUP BY {trend_group} ORDER BY {trend_group}""" }, { "key": "pv_trend", "desc": "各月光伏节碳趋势", "sql": f"""SELECT {trend_group}, SUM(save_carbon) AS pv_saving FROM `{pv_trend_view}` {pv_trend_filter} GROUP BY {trend_group} ORDER BY {trend_group}""" }, { "key": "energy_dist_by_park", "desc": "各园区能耗分布", "sql": f"""SELECT park_name, SUM(energy_consumption_electricity) AS ele_energy, SUM(energy_consumption_heat) AS heat_energy, SUM(energy_consumption_gas) AS gas_energy FROM `{energy_view}` {time_filter_energy} GROUP BY park_name ORDER BY (SUM(energy_consumption_electricity) + SUM(energy_consumption_heat) + SUM(energy_consumption_gas)) DESC""" } ] def build_enterprise_overview_sqls(date_type, date_str, enterprise_name, park_name): """Fixed SQL queries for dashboard_enterprise.""" ent_filter = f"AND enterprise_name = '{enterprise_name}'" if enterprise_name else "" park_filter = f"AND park_name = '{park_name}'" if park_name else "" combined = f"{ent_filter} {park_filter}".strip() if date_type == "year": year = date_str energy_view = "园区企业能耗年数据" carbon_view = "园区企业碳排年数据" pv_view = "园区企业光伏节碳年数据" time_filter = f"WHERE year = {year} {combined}" trend_view = "园区企业能耗月数据" trend_filter = f"WHERE year = {year} {combined}" trend_group = "month" rank_filter = f"WHERE year = {year} {park_filter}" else: parts = date_str.split("-") year = int(parts[0]) month = int(parts[1]) energy_view = "园区企业能耗月数据" carbon_view = "园区企业碳排月数据" pv_view = "园区企业光伏节碳月数据" time_filter = f"WHERE year = {year} AND month = {month} {combined}" trend_view = "园区企业能耗月数据" trend_filter = f"WHERE year = {year} AND month <= {month} {combined}" trend_group = "month" rank_filter = f"WHERE year = {year} AND month = {month} {park_filter}" return [ { "key": "current_energy", "desc": "当期能耗汇总", "sql": f"""SELECT enterprise_name, park_name, SUM(energy_consumption_electricity) AS ele_energy, SUM(energy_consumption_heat) AS heat_energy, SUM(energy_consumption_gas) AS gas_energy FROM `{energy_view}` {time_filter} GROUP BY enterprise_name, park_name LIMIT 1""" }, { "key": "current_carbon", "desc": "当期碳排汇总", "sql": f"""SELECT SUM(carbon_emissions_electricity) AS ele_carbon, SUM(carbon_emissions_heat) AS heat_carbon, SUM(carbon_emissions_gas) AS gas_carbon FROM `{carbon_view}` {time_filter}""" }, { "key": "current_pv", "desc": "当期光伏节碳", "sql": f"""SELECT SUM(save_carbon) AS pv_save_carbon, SUM(power_generation) AS pv_power FROM `{pv_view}` {time_filter}""" }, { "key": "monthly_trend", "desc": "各月能耗趋势", "sql": f"""SELECT {trend_group}, SUM(energy_consumption_electricity) AS ele_energy, SUM(energy_consumption_heat) AS heat_energy, SUM(energy_consumption_gas) AS gas_energy FROM `{trend_view}` {trend_filter} GROUP BY {trend_group} ORDER BY {trend_group}""" }, { "key": "rank_in_park", "desc": "在园区内的能耗排名", "sql": f"""SELECT enterprise_name, SUM(energy_consumption_electricity + energy_consumption_heat + energy_consumption_gas) AS total_energy FROM `{energy_view}` {rank_filter} GROUP BY enterprise_name ORDER BY total_energy DESC LIMIT 20""" } ] def execute_fixed_sqls(sql_list): """Execute all fixed SQLs concurrently, return dict keyed by 'key'.""" results = {} def run_one(item): db_result = query_database(item["sql"]) return item["key"], extract_rows(db_result) with ThreadPoolExecutor(max_workers=min(len(sql_list), 8)) as pool: futures = {pool.submit(run_one, item): item["key"] for item in sql_list} for future in as_completed(futures): key, rows = future.result() results[key] = rows return results def fill_template_with_llm(query, template_name, template_doc, data_results, date_type, date_str, park_name, enterprise_name): """ Use LLM to map query results into ALL template placeholders. Returns a flat dict where keys are placeholder names (without {{ }}) and values are strings to substitute into the HTML. The special key 'INJECT_DATA' holds the full `const mockData = {...};` string. Also returns a summary string. """ data_str = json.dumps(data_results, ensure_ascii=False, separators=(',', ':')) if len(data_str) > 8000: data_str = data_str[:8000] + "...(truncated)" params_parts = [] if date_str: if date_type == "year": params_parts.append(f"统计年份: {date_str}年") else: params_parts.append(f"统计月份: {date_str}") if park_name: params_parts.append(f"园区名称: {park_name}") if enterprise_name: params_parts.append(f"企业名称: {enterprise_name}") params_str = "、".join(params_parts) if params_parts else "全部数据" # Fixed default values for static label placeholders (no need to ask LLM) if template_name == "dashboard_overview": static_defaults = { "METRIC_CARD_TITLE": "综合能耗与碳排", "ENERGY_TITLE": "总能耗", "ENERGY_UNIT": "t 折标煤", "ENERGY_SUB_1_LABEL": "电能耗", "ENERGY_SUB_2_LABEL": "热能耗", "ENERGY_SUB_3_LABEL": "燃能耗", "CARBON_TITLE": "总碳排", "CARBON_UNIT": "tCO₂", "CARBON_SUB_1_LABEL": "电碳排", "CARBON_SUB_2_LABEL": "热碳排", "CARBON_SUB_3_LABEL": "燃碳排", "YEAR_TITLE": "年度累计", "YEAR_UNIT": "tCO₂", "YEAR_SUB_1_LABEL": "碳排", "YEAR_SUB_2_LABEL": "光伏节碳", "PRICE_CARD_TITLE": "能源费用单价", "PRICE_ELECTRIC_LABEL": "电价（¥）", "PRICE_HEAT_LABEL": "热价（¥）", "PRICE_GAS_LABEL": "燃价（¥）", "CARBON_DIST_TITLE": "碳排分布", "CARBON_DIST_UNIT": "单位/tCO₂", "PV_TITLE": "光伏节碳", "PV_UNIT": "单位/tCO₂", "ENERGY_DIST_TITLE": "能耗分布（t折标煤）", "DEVICE_DIST_TITLE": "设备分布", "DEVICE_TOTAL_LABEL": "总数", "DEVICE_TOTAL_UNIT": "台", "RANK_TABLE_TITLE": "企业能耗排名", "TABLE_HEADER_1": "序号", "TABLE_HEADER_2": "公司名称", "TABLE_HEADER_3": "所属园区", "TABLE_HEADER_4": "总能耗（t折标煤）", } placeholder_spec = f"""需要返回的动态占位符(字符串): PAGE_TITLE, ENERGY_VALUE(折标煤), ENERGY_SUB_1_VALUE(电), ENERGY_SUB_2_VALUE(热), ENERGY_SUB_3_VALUE(燃), CARBON_VALUE(tCO₂), CARBON_SUB_1_VALUE(电碳), CARBON_SUB_2_VALUE(热碳), CARBON_SUB_3_VALUE(燃碳), YEAR_VALUE(年度总碳排), YEAR_SUB_1_VALUE(年度碳排), YEAR_SUB_2_VALUE(年度光伏节碳), PRICE_ELECTRIC_VALUE(元/度), PRICE_HEAT_VALUE(元/GJ), PRICE_GAS_VALUE(元/m³), DEVICE_TOTAL_VALUE(设备总数) INJECT_DATA: const mockData={{...}}; 字段名必须严格按以下格式(模板JS直接引用这些字段名): allEnergy, eleEnergy, heatEnergy, gasEnergy, (折标煤) allCarbon, eleCarbon, heatCarbon, gasCarbon, (tCO₂) yearNet, yearCarbon, yearPv, elePrice, heatPrice, gasPrice, carbonMonths:[], (月份数字列表, 如[1,2,3]) carbonEle:[], (每月电碳排) carbonGas:[], (每月燃碳排) carbonHeat:[], (每月热碳排) pvMonths:[], (光伏月份数字列表) pvData:[], (每月光伏节碳量) pieParks:[], (园区名称列表) pieEle:[], (各园区电能耗分布) pieHeat:[], (各园区热能耗分布) pieGas:[], (各园区燃能耗分布) devices:[{{name,count}},...], rankings:[{{rank,name,park,value}},...] 注意: rankings不是enterpriseRanking！字段名必须精确。""" elif template_name == "dashboard_enterprise": static_defaults = {} placeholder_spec = """HTML占位符(字符串): PAGE_TITLE, DATE_LABEL, DATE_VALUE, DATE_OPTION_1, DATE_OPTION_2, DATE_OPTION_3, ENTERPRISE_NAME_LABEL, ENTERPRISE_NAME_VALUE, PARK_LABEL, PARK_VALUE, ENTERPRISE_ADDRESS_LABEL, ENTERPRISE_ADDRESS_VALUE, STAT_1_TITLE, STAT_1_VALUE, STAT_1_UNIT, STAT_1_SUBTITLE, STAT_1_SUBTITLE_VALUE, STAT_1_SUBTITLE_UNIT, STAT_1_TREND_LABEL, STAT_1_TREND_VALUE, STAT_1_FOOTER, STAT_2_TITLE, STAT_2_VALUE, STAT_2_UNIT, STAT_2_SUBTITLE, STAT_2_SUBTITLE_VALUE, STAT_2_SUBTITLE_UNIT, STAT_2_TREND_LABEL, STAT_2_TREND_VALUE, STAT_2_FOOTER, ENERGY_CHART_TITLE, CHART_1_UNIT, CHART_2_UNIT, ENERGY_1_LABEL, ENERGY_1_VALUE, ENERGY_1_PERCENT, ENERGY_1_TREND_LABEL, ENERGY_1_TREND_VALUE, ENERGY_2_LABEL, ENERGY_2_VALUE, ENERGY_2_PERCENT, ENERGY_2_TREND_LABEL, ENERGY_2_TREND_VALUE, ENERGY_3_LABEL, ENERGY_3_VALUE, ENERGY_3_PERCENT, ENERGY_3_TREND_LABEL, ENERGY_3_TREND_VALUE, CARBON_EMISSION_TITLE, CARBON_DIST_TITLE, CHART_CARBON_XAXIS, SERIES_1_NAME, SERIES_1_DATA, SERIES_2_NAME, SERIES_2_DATA, SERIES_3_NAME, SERIES_3_DATA, ENERGY_SAVING_TITLE, LINE_CHART_DATA, BAR_CHART_DATA, DEVICE_TOTAL_NAME, DEVICE_TOTAL_COUNT, DEVICE_TOTAL_ONLINE, DEVICE_TOTAL_ONLINE_LABEL, DEVICE_TOTAL_OFFLINE, DEVICE_TOTAL_OFFLINE_LABEL, DEVICE_TOTAL_UNIT, DEVICE_1_NAME~DEVICE_7_NAME, DEVICE_1_TOTAL~DEVICE_7_TOTAL, DEVICE_1_ONLINE~DEVICE_7_ONLINE, DEVICE_1_OFFLINE~DEVICE_7_OFFLINE, DEVICE_1_ONLINE_LABEL~DEVICE_7_ONLINE_LABEL, DEVICE_1_OFFLINE_LABEL~DEVICE_7_OFFLINE_LABEL, DEVICE_1_UNIT~DEVICE_7_UNIT""" else: static_defaults = {} placeholder_spec = template_doc[:2000] if template_doc else "" prompt = f"""参数: {params_str} 数据: {data_str} 需要返回的占位符: {placeholder_spec} 返回 JSON，包含所有占位符的值。规则: 1. HTML占位符直接给字符串值。 2. INJECT_DATA给完整`const mockData={{...}};`单行JS字符串。 3. 数值保留3位小数。 4. 加"summary"字段，2-3句中文分析摘要。 5. 空数据用0或[]。 6. PAGE_TITLE包含时间："{params_str}综合能耗概览"。 7. devices合并: device_stats每个terminal_category一项+pv_device_stats的cnt为光伏看板一项，DEVICE_TOTAL_VALUE=总和。 只返回 JSON。""" resp = call_llm( prompt, system="你是数据可视化工程师。将数据填入模板占位符，只返回 JSON。", model=LLM_MODEL_SQL, # qwen-turbo: faster and cheaper for structured output max_tokens=3000 ) if not resp: return {}, "数据查询完成" result = extract_json(resp) if not result: return {}, "数据查询完成" summary = result.pop("summary", "数据分析报告已生成。") # Merge static defaults first, then let LLM values override dynamic ones merged = dict(static_defaults) # start with static labels merged.update(result) # LLM values override / add dynamic values return merged, summary # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser() parser.add_argument("--query", required=True) parser.add_argument("--views", required=True) parser.add_argument("--template", required=True) parser.add_argument("--is_fixed_template", default="false") parser.add_argument("--date_type", default="month") parser.add_argument("--date", default=None) parser.add_argument("--park_name", default=None) parser.add_argument("--enterprise_name", default=None) args = parser.parse_args() is_fixed = args.is_fixed_template.lower() == "true" template_name = args.template # ----------------------------------------------------------------------- # Fixed template branch # ----------------------------------------------------------------------- if is_fixed: date_type = args.date_type if args.date_type else "month" date_str = args.date if args.date and args.date != "None" else None park_name = args.park_name if args.park_name and args.park_name != "None" else None enterprise_name = args.enterprise_name if args.enterprise_name and args.enterprise_name != "None" else None if not date_str: # Default: last month today = datetime.date.today() last_month = today.replace(day=1) - datetime.timedelta(days=1) date_str = last_month.strftime("%Y-%m") date_type = "month" template_doc = read_file(os.path.join(TEMPLATES_DIR, f"{template_name}.md")) if template_name == "dashboard_overview": sql_list = build_park_overview_sqls(date_type, date_str, park_name) elif template_name == "dashboard_enterprise": sql_list = build_enterprise_overview_sqls(date_type, date_str, enterprise_name, park_name) else: print(json.dumps({"status": "error", "message": f"Unknown fixed template: {template_name}"})) sys.exit(1) # Execute all fixed SQLs data_results = execute_fixed_sqls(sql_list) # Use LLM to map data → all template placeholder values (including INJECT_DATA) placeholder_values, summary = fill_template_with_llm( args.query, template_name, template_doc, data_results, date_type, date_str, park_name, enterprise_name ) # Save to temp file for step3_render.py temp_result_file = os.path.join(SKILL_DIR, "output", f"temp_result_{os.getpid()}.json") os.makedirs(os.path.dirname(temp_result_file), exist_ok=True) combined_output = { "is_fixed_template": True, "template_name": template_name, "date_type": date_type, "date": date_str, "park_name": park_name, "enterprise_name": enterprise_name, "placeholder_values": placeholder_values, "summary": summary } with open(temp_result_file, 'w', encoding='utf-8') as f: json.dump(combined_output, f, ensure_ascii=False) print(json.dumps({ "status": "success", "is_fixed_template": True, "template_name": template_name, "date_type": date_type, "date": date_str, "park_name": park_name, "enterprise_name": enterprise_name, "row_count": sum(len(v) for v in data_results.values()), "result_file": temp_result_file, "summary": summary }, ensure_ascii=False)) return # ----------------------------------------------------------------------- # Normal LLM-based flow: generate SQL, execute, save raw rows for step3 # ----------------------------------------------------------------------- selected_views = [v.strip() for v in args.views.split(",") if v.strip()] view_schemas = load_view_schemas(selected_views) template_doc = read_file(os.path.join(TEMPLATES_DIR, f"{template_name}.md")) if not view_schemas: print(json.dumps({"status": "error", "message": f"Could not load schemas for: {selected_views}"})) sys.exit(1) if not template_doc: print(json.dumps({"status": "error", "message": f"Template doc not found: {template_name}.md"})) sys.exit(1) # Generate SQL sql_resp = call_llm( f"QUERY: {args.query}\nTEMPLATE: {template_name}\nVIEW SCHEMAS:\n{view_schemas}\n\nReturn JSON: {{\"sql\": \"SELECT...\"}}\nCRITICAL SQL rules:\n1. Use backtick-quoted Chinese view/column names.\n2. The database uses sql_mode=only_full_group_by. Every column in SELECT must either be in GROUP BY or wrapped in an aggregate function (SUM/AVG/MAX/MIN/COUNT). NEVER select non-aggregated columns that are not in GROUP BY.\n3. When grouping by time, use only the time column in GROUP BY. Wrap all other columns with AVG() or SUM() as appropriate.\n4. ORDER BY the time column. SELECT minimum columns needed.", system="Expert MySQL developer. Return ONLY JSON {\"sql\": \"...\"}. The database enforces only_full_group_by strictly — every SELECT column must be in GROUP BY or aggregated.", max_tokens=500 ) if not sql_resp: print(json.dumps({"status": "error", "message": "Failed to generate SQL"})) sys.exit(1) sql_plan = extract_json(sql_resp) if not sql_plan or "sql" not in sql_plan: print(json.dumps({"status": "error", "message": "Failed to parse SQL", "raw": sql_resp})) sys.exit(1) sql = sql_plan["sql"] db_result = query_database(sql) if not db_result: print(json.dumps({"status": "error", "message": "Failed to execute SQL"})) sys.exit(1) rows = extract_rows(db_result) if not rows: print(json.dumps({"status": "error", "message": "SQL returned no data", "sql": sql})) sys.exit(1) # Save raw rows + template_doc for step3 to map with LLM temp_result_file = os.path.join(SKILL_DIR, "output", f"temp_result_{os.getpid()}.json") os.makedirs(os.path.dirname(temp_result_file), exist_ok=True) with open(temp_result_file, 'w', encoding='utf-8') as f: json.dump({ "is_fixed_template": False, "template_name": template_name, "template_doc": template_doc, "rows": rows, "sql": sql }, f, ensure_ascii=False) print(json.dumps({ "status": "success", "is_fixed_template": False, "sql": sql, "row_count": len(rows), "result_file": temp_result_file, "summary": f"已查询到 {len(rows)} 条数据记录。" }, ensure_ascii=False)) if __name__ == "__main__": main()