#!/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()。 修复点： 1. GROUP BY 解析支持多行 SQL（原正则 [^\n;ORDER] 遇换行截断） 2. 列名提取改为识别 AS 别名，并正确判断聚合函数 3. 对视图列（如 data_update_time）也能正确识别并包裹 ANY_VALUE() """ import re # 提取 GROUP BY 子句（支持多行，直到 ORDER BY / HAVING / LIMIT / 语句结束） 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 # 解析 GROUP BY 中的列名（去掉反引号、函数调用等，取裸列名） gb_raw = gb_match.group(1).strip() gb_cols = set() for item in gb_raw.split(','): item = item.strip() # 去掉反引号 item_clean = item.strip('`') # 如果是函数调用（如 DATE(col)），提取括号内的列名 func_inner = re.search(r'\w+\s*\(\s*`?([\w\u4e00-\u9fff]+)`?\s*\)', item) if func_inner: gb_cols.add(func_inner.group(1)) else: gb_cols.add(item_clean) # 提取 SELECT 子句（到 FROM 为止，支持多行） 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 = 0 current = [] 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() # 判断是否是聚合函数（SUM/AVG/MAX/MIN/COUNT/ANY_VALUE/GROUP_CONCAT/ROUND/IFNULL/IF/CASE等包含聚合的） 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 后面的别名作为列名判断 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 '' # 判断是否在 GROUP BY 中 in_gb = col_name in gb_cols or any(col_name == g for g in gb_cols) if 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=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] Time/datetime columns (report_time, create_time, update_time, stat_date, data_update_time, etc.) that appear in SELECT but NOT in GROUP BY MUST be wrapped with MAX() or ANY_VALUE(). Example: MAX(`report_time`) AS `latest_time`. NEVER select raw time columns alongside GROUP BY without aggregation.", 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. ALWAYS include a LIMIT clause. ALWAYS aggregate time-series data by day or month.", 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()