#!/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" # --------------------------------------------------------------------------- # Token usage tracking # --------------------------------------------------------------------------- import atexit, datetime as _dt _token_usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "calls": 0} def _flush_token_usage(): output_dir = os.path.join(SKILL_DIR, "output") os.makedirs(output_dir, exist_ok=True) log_path = os.path.join(output_dir, "token_usage.jsonl") record = { "time": _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "script": "step2_query", **_token_usage } with open(log_path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") atexit.register(_flush_token_usage) # --------------------------------------------------------------------------- # 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) result = json.loads(resp.read().decode("utf-8")) usage = result.get("usage", {}) _token_usage["input_tokens"] += usage.get("input_tokens", 0) _token_usage["output_tokens"] += usage.get("output_tokens", 0) _token_usage["total_tokens"] += usage.get("total_tokens", 0) _token_usage["calls"] += 1 return result["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 模式报错时，自动修复 SQL。 修复逻辑： 1. 如果 GROUP BY 使用函数表达式（如 DATE(col)），将 SELECT 中对应的裸列名替换为相同的函数表达式 例：SELECT `report_time` ... GROUP BY DATE(`report_time`) → SELECT DATE(`report_time`) AS `report_time` 2. 其他未聚合且不在 GROUP BY 中的列包裹 ANY_VALUE() """ import re # 提取 GROUP BY 子句（支持多行） gb_match = re.search( r'GROUP\s+BY\s+(.*?)(?=\s+(?:ORDER|HAVING|LIMIT|UNION|$))', sql, re.IGNORECASE | re.DOTALL ) if not gb_match: return sql gb_raw = gb_match.group(1).strip() # 解析 GROUP BY 中的每一项，分为两类： # - func_items: {col_name: full_expr} 如 {"report_time": "DATE(`report_time`)"} # - plain_cols: {col_name} 如 {"park_name"} func_items = {} # 裸列名 -> GROUP BY 函数表达式 plain_cols = set() for item in gb_raw.split(','): item = item.strip() func_m = re.match(r'(\w+)\s*\(\s*(`?[\w\u4e00-\u9fff]+`?)\s*\)', item) if func_m: func_name = func_m.group(1) # e.g. DATE inner_col = func_m.group(2).strip('`') # e.g. report_time full_expr = func_m.group(0) # e.g. DATE(`report_time`) func_items[inner_col] = (func_name, full_expr) else: plain_cols.add(item.strip('`')) all_gb_cols = set(func_items.keys()) | plain_cols # 提取 SELECT 子句 sel_match = re.search(r'SELECT\s+(.*?)\s+FROM\b', sql, re.IGNORECASE | re.DOTALL) if not sel_match: return sql sel_str = sel_match.group(1) # 智能分割 SELECT 列（处理函数内的逗号） def split_select_cols(s): cols, depth, current = [], 0, [] for ch in s: if ch == '(': depth += 1; current.append(ch) elif ch == ')': depth -= 1; current.append(ch) elif ch == ',' and depth == 0: cols.append(''.join(current).strip()); current = [] else: current.append(ch) if current: cols.append(''.join(current).strip()) return cols cols = split_select_cols(sel_str) new_cols = [] for col in cols: col = col.strip() # 已是聚合函数，不处理 is_agg = bool(re.search( r'\b(SUM|AVG|MAX|MIN|COUNT|ANY_VALUE|GROUP_CONCAT|STD|VARIANCE)\s*\(', col, re.IGNORECASE )) if is_agg: new_cols.append(col) continue # 提取列的裸名（去掉 AS 别名、反引号） as_match = re.search(r'\bAS\s+`?([\w\u4e00-\u9fff]+)`?\s*$', col, re.IGNORECASE) if as_match: col_name = as_match.group(1) else: idents = re.findall(r'`?([\w\u4e00-\u9fff]+)`?', col) col_name = idents[-1] if idents else '' # 情况1：列名在 GROUP BY 的函数表达式中（如 DATE(report_time)） # 将 SELECT `report_time` 替换为 DATE(`report_time`) AS `report_time` if col_name in func_items: func_name, full_expr = func_items[col_name] # 保留反引号格式 new_cols.append(f"{full_expr} AS `{col_name}`") continue # 情况2：列名在普通 GROUP BY 中，不处理 if col_name in plain_cols: new_cols.append(col) continue # 情况3：列名不在任何 GROUP BY 中，包裹 ANY_VALUE() new_cols.append(f'ANY_VALUE({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=180) 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) # 无论 SQL 是否变化，都尝试重试（fix_group_by_sql 可能已修复） retry_sql = fixed_sql if fixed_sql != sql else sql req2 = urllib.request.Request( DB_API_URL, data=json.dumps({"sql": retry_sql, "basejdbc": DB_JDBC}).encode("utf-8"), headers={"Content-Type": "application/json"} ) try: resp2 = urllib.request.urlopen(req2, timeout=180) result2 = json.loads(resp2.read().decode("utf-8")) # 如果重试仍然报错，返回原始错误信息 if (isinstance(result2, list) and result2 and isinstance(result2[0], dict) and 'error' in result2[0]): return result # 返回原始报错，让上层处理 return result2 except Exception as e2: # 重试超时或失败，返回原始结果 return result return result except Exception as e: print(json.dumps({"status": "error", "message": f"DB error: {e}"})) sys.exit(1) def convert_timestamps(rows): """ 将 DB API 返回的毫秒时间戳自动转换为可读时间格式。 - 13位整数（毫秒时间戳，范围在2000~2100年之间）→ "YYYY-MM-DD HH:MM:SS" - 10位整数（秒时间戳）→ "YYYY-MM-DD HH:MM:SS" - 其他类型不做处理 """ # 2000-01-01 到 2100-01-01 的毫秒时间戳范围 MS_MIN = 946684800000 # 2000-01-01 00:00:00 UTC 毫秒 MS_MAX = 4102444800000 # 2100-01-01 00:00:00 UTC 毫秒 S_MIN = 946684800 # 秒 S_MAX = 4102444800 converted = [] for row in rows: new_row = {} for k, v in row.items(): if isinstance(v, int): if MS_MIN <= v <= MS_MAX: # 毫秒时间戳 → 转为秒再格式化 try: new_row[k] = datetime.datetime.fromtimestamp(v / 1000).strftime("%Y-%m-%d %H:%M:%S") except Exception: new_row[k] = v elif S_MIN <= v <= S_MAX: # 秒时间戳 try: new_row[k] = datetime.datetime.fromtimestamp(v).strftime("%Y-%m-%d %H:%M:%S") except Exception: new_row[k] = v else: new_row[k] = v else: new_row[k] = v converted.append(new_row) return converted def extract_rows(db_result): if db_result is None: return [] if isinstance(db_result, list): return convert_timestamps(db_result) for k in ("data", "rows", "result", "records"): if k in db_result and isinstance(db_result[k], list): return convert_timestamps(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.\n5. [AGGREGATION RULE] For time-series queries (trend/daily/monthly/hourly), ALWAYS aggregate using GROUP BY with DATE() for daily or DATE_FORMAT() for monthly. NEVER return raw per-record rows for chart templates.\n6. [LIMIT RULE] ALWAYS end SQL with LIMIT: 365 for daily, 60 for monthly, 500 for tables. This is mandatory to prevent slow rendering.\n7. [TIME FIELD RULE - CRITICAL] ONLY use fields that actually exist in the VIEW SCHEMA above. NEVER invent or assume field names. When the view has a datetime field (e.g. `report_time`), use DATE(`report_time`) for daily grouping and the SELECT clause MUST use the EXACT SAME expression: SELECT DATE(`report_time`) AS `report_time` ... GROUP BY DATE(`report_time`). When the view stores date as separate integer columns (e.g. `year`, `month`, `day`), you MUST group by those columns directly: GROUP BY `year`, `month`, `day` — do NOT use DATE() or invent a `report_time` column that does not exist in the schema.\n8. [SCHEMA COMPLIANCE - CRITICAL] Before writing any SQL, check the VIEW SCHEMA fields carefully. If the view has `year`/`month`/`day` integer columns but NO `report_time` column, use GROUP BY `year`, `month`, `day` and ORDER BY `year`, `month`, `day`. NEVER use a field that is not listed in the VIEW SCHEMA.\n9. [VIEW SELECTION RULE - CRITICAL] When multiple views are provided, always choose the view whose schema best matches the query granularity. For daily historical trend queries, prefer views with `year`/`month`/`day` integer columns or a `report_time` datetime field (e.g. `光伏设备天数据`). NEVER use a real-time snapshot view (e.g. `光伏逆变器信息`) for time-series trend queries -- real-time views only store current-state fields (inverter_real_power, today_generation_total, data_update_time) and have NO historical time-series columns.\n10. [FILTER RULE] When the query mentions a specific device or entity name (e.g. '赤峰云铜逆变器1'), add a WHERE clause filtering by the matching name field in the chosen view (e.g. `inverterName` = '赤峰云铜逆变器1'). Always use the exact field name from the VIEW SCHEMA.", system="Expert MySQL developer. Return ONLY JSON {\"sql\": \"...\"}. The database enforces only_full_group_by strictly \u2014 every SELECT column must be in GROUP BY or aggregated. When GROUP BY uses DATE(col), SELECT must also use DATE(col) AS col, not the raw col. ALWAYS include a LIMIT clause. ALWAYS aggregate time-series data by day or month. CRITICAL: ONLY use field names that exist in the provided VIEW SCHEMA. If the view uses separate year/month/day integer columns instead of a datetime field, use those columns for grouping and ordering. NEVER fabricate field names like report_time if they are not in the schema.", 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"] # [方案四优化] LIMIT 兜底：若 LLM 生成的 SQL 未包含 LIMIT，自动追加 def add_limit_if_missing(raw_sql, template, default_limit=500): """根据模板类型智能追加 LIMIT，防止大数据量拖慢渲染""" sql_stripped = raw_sql.rstrip("; \n") if re.search(r'\bLIMIT\b', sql_stripped, re.IGNORECASE): return sql_stripped # 已有 LIMIT，不重复追加 # 根据模板类型设置合理上限 limit_map = { "lineChart": 365, # 日趋势最多1年 "lineChartCompare": 365, "barChart": 60, # 月对比最多5年 "pieChart": 50, # 饼图类别不宜过多 "dataTable": 500, # 表格最多500行 "infoCard": 100, "dashboardKpiBar": 60, "dashboardKpiBarCompare": 60, } limit = limit_map.get(template, default_limit) return sql_stripped + f" LIMIT {limit}" sql = add_limit_if_missing(sql, template_name) 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) # [SQL历史记录] 将提问和生成的SQL追加保存到固定JSON文件 sql_history_file = os.path.join(SKILL_DIR, "output", "sql_history.json") try: if os.path.exists(sql_history_file): with open(sql_history_file, 'r', encoding='utf-8') as _hf: _sql_history = json.load(_hf) else: _sql_history = [] _sql_history.append({ "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "query": args.query, "views": args.views, "template": template_name, "sql": sql, "row_count": len(rows) }) with open(sql_history_file, 'w', encoding='utf-8') as _hf: json.dump(_sql_history, _hf, ensure_ascii=False, indent=2) except Exception: pass # 保存历史失败不影响主流程 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()