""" AI 蓝图生成服务：描述 + Excel(+可选图片) → AppBlueprint draft 无 LLM Key 时走启发式规则，保证可离线测试。 领域无关：不写死库存/沉降等业务字段，由表头规范化 + 类型推断 + 蓝图 widgets 驱动。 """ from __future__ import annotations import asyncio import hashlib import io import json import re import unicodedata from pathlib import Path from typing import Any from fastapi import FastAPI, File, Form, UploadFile from fastapi.middleware.cors import CORSMiddleware from openpyxl import load_workbook from generation_rules import ( DEFAULT_ACTION_LABELS, compose_effective_prompt, parse_action_bar, wants_screenshot_layout, ) from html_layout import layout_summary_text, merge_layout_into_ui_hints, parse_layout_file from llm import enhance_blueprint_with_llm, list_providers, load_ai_config, understand_images from fidelity_loop import run_fidelity_loop from generate_log import GenerateTrace, ensure_logging from demo_fixtures import file_response, fixtures_manifest, read_prompt_text from preview_store import create_preview, get_preview import os ensure_logging() # 优先加载仓库根 .env，其次 ai-service/.env（容错编码，避免整服务起不来） _root = Path(__file__).resolve().parents[1] def _safe_load_dotenv(path: Path, *, override: bool = False) -> None: if not path.is_file(): return raw = path.read_bytes() text: str | None = None for enc in ("utf-8-sig", "utf-8", "gbk", "cp936", "latin-1"): try: text = raw.decode(enc) break except UnicodeDecodeError: continue if text is None: print(f"WARN: cannot decode env file {path}, skip") return for line in text.splitlines(): s = line.strip() if not s or s.startswith("#") or "=" not in s: continue key, _, val = s.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if not key: continue if override or key not in os.environ: os.environ[key] = val # 若曾用非 utf-8 读入，尝试写回 utf-8，避免下次再炸 try: if raw[:3] != b"\xef\xbb\xbf": # 仅当原文件不是合法 utf-8 时重写 try: raw.decode("utf-8") except UnicodeDecodeError: path.write_text(text if text.endswith("\n") else text + "\n", encoding="utf-8") print(f"WARN: rewrote {path} as utf-8") except OSError: pass _safe_load_dotenv(_root / ".env") _safe_load_dotenv(Path(__file__).resolve().parent / ".env", override=True) app = FastAPI(title="aijianzhan-ai", version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) _SYSTEM_FIELDS = {"id", "tenant_id", "created_by", "created_at", "updated_at"} def slugify(text: str, fallback: str = "app") -> str: """通用标识符：驼峰/点号 → snake_case；纯中文用稳定短哈希，无业务词表。""" original = text or "" text = unicodedata.normalize("NFKC", original) text = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", text) text = text.lower() text = text.replace("-", "_").replace(".", "_") text = re.sub(r"[^a-z0-9_]+", "_", text) text = re.sub(r"_+", "_", text).strip("_") if not text or not re.match(r"^[a-z]", text): # 列名用 col_N；应用 slug 用短哈希，避免依赖中文词表 if re.match(r"^col_\d+$", fallback): text = fallback else: h = hashlib.sha1(original.encode("utf-8")).hexdigest()[:8] text = f"{fallback}_{h}" return text[:48] def to_field_name(header: str, idx: int) -> str: raw = slugify(header, fallback=f"col_{idx}") return raw if re.match(r"^[a-z]", raw) else f"col_{idx}" def infer_app_name(prompt: str) -> str: """从需求描述提取应用显示名。""" p = prompt or "" for pat in ( r"应用名称[推荐为是：:\s]*[「『\"]?([^」』\"\n，。；]{2,32})", r"[「『](.+?)[」』]", r"生成[一个]?(?:业务)?应用[：:\s]*([^\n，。；]{2,32})", ): m = re.search(pat, p) if m: name = m.group(1).strip() if name and name not in {"Excel", "xlsx", "CSV"}: return name[:32] for line in p.splitlines(): line = line.strip("# ").strip() if 4 <= len(line) <= 32 and not line.startswith("http"): return line[:32] return "智能应用" def infer_item_label(prompt: str, sheet_label: str) -> str: """业务对象称呼：提示词显式指定 > Excel 表名 >「记录」。""" p = prompt or "" m = re.search(r"(?:对象|实体|记录)名[称]?[推荐为是：:\s]*([^\n，。；]{1,16})", p) if m: return m.group(1).strip()[:16] if sheet_label and re.search(r"[\u4e00-\u9fff]", sheet_label): lab = re.sub(r"(表|列表|台账|数据|信息)$", "", sheet_label).strip() if lab and lab.lower() not in {"sheet1", "sheet"}: return lab[:16] if sheet_label and sheet_label.lower() not in {"sheet1", "sheet", "数据"}: return sheet_label[:16] return "记录" def parse_ui_hints(prompt: str, image_count: int, html_count: int = 0) -> dict[str, Any]: """从用户需求提取展示元信息；截图/HTML 忠实策略由 generation_rules 决定。""" p = prompt or "" hints: dict[str, Any] = {"ui": {}} if wants_screenshot_layout(p, image_count, html_count): hints["ui_preset"] = "screenshot_faithful" m = re.search(r"平台抬头[：:]\s*(.+)", p) if m: hints["ui"]["platform_title"] = m.group(1).strip().splitlines()[0][:64] m = re.search(r"系统名称[：:]\s*(.+)", p) if m: hints["app_name"] = m.group(1).strip().splitlines()[0][:64] m = re.search(r"应用名称推荐[：:]\s*(.+)", p) if m and not hints.get("app_name"): hints["app_name"] = m.group(1).strip().splitlines()[0][:64] m = re.search(r"推荐 slug[：:]\s*`?([a-z][a-z0-9_]{1,47})`?", p) if m: hints["slug"] = m.group(1) m = re.search(r"工程上下文(?:示例)?[：:]\s*(.+)", p) if m: ctx = m.group(1).strip().splitlines()[0] ctx = re.split(r"[（(]", ctx, 1)[0].strip() hints["project_context"] = ctx[:120] # 常见「XX平台」抬头 if "platform_title" not in hints["ui"]: m = re.search(r"([^\n]{2,20}管理平台)", p) if m: hints["ui"]["platform_title"] = m.group(1).strip() # 仅当提示词明文写出时才采纳（不写死行业） if "点击" in p and "查看" in p: m = re.search(r"(点击[^。\n]{2,40})", p) if m: hints["ui"]["filter_hint"] = m.group(1).strip() # 单选选项：全部 / A / B / C 或 **A / B / C** m = re.search(r"(?:全部\s*[/／]\s*)([^。\n]{2,40})", p) if m and ("单选" in p or "筛选" in p or "类型" in p): opts = [x.strip().strip("*` ") for x in re.split(r"[/／、，,]", m.group(1)) if x.strip()] opts = [o for o in opts if o and o != "全部" and len(o) <= 12][:8] if len(opts) >= 2: hints["ui"]["section_options"] = opts hints["filter_style"] = "section_radios" # 页面标题：仅匹配「标题：xxx」或明确业务页名行 for key, pat in ( ("list_title", r"(?:列表标题|列表页)[：:]\s*(.+)"), ("create_title", r"(?:新增页|表单标题|创建页)[：:]\s*(.+)"), ("dash_title", r"(?:看板标题|看板页)[：:]\s*(.+)"), ): m = re.search(pat, p) if m: hints[key] = m.group(1).strip().splitlines()[0][:64] action_bar = parse_action_bar(p) if action_bar: hints["actions"] = action_bar["actions"] hints["action_labels"] = action_bar["action_labels"] return hints def wants_dashboard(prompt: str, image_count: int, html_count: int = 0) -> bool: p = prompt or "" keys = ("图表", "看板", "统计", "概览", "dashboard", "折线", "曲线", "纵断", "KPI", "可视化") if any(k.lower() in p.lower() if k.isascii() else k in p for k in keys): return True return image_count > 0 or html_count > 0 def wants_line_chart(prompt: str) -> bool: p = prompt or "" return any(k in p for k in ("折线", "曲线", "纵断", "趋势", "line chart", "时序")) def build_dashboard_widgets( fields: list[dict[str, Any]], filters: list[str], list_cols: list[str], entity_name: str, item_label: str, prompt: str, ) -> list[dict[str, Any]]: """按字段类型与提示词组装 widgets；标题优先来自提示词/HTML，否则用通用名。""" usable = [f for f in fields if f["name"] not in _SYSTEM_FIELDS] names = {f["name"] for f in usable} numeric = [f for f in usable if f["type"] in {"int", "bigint", "decimal"}] categorical = [ f["name"] for f in usable if f.get("enum_values") or f["type"] == "enum" or f["name"] in filters ] x_candidates = [f["name"] for f in usable if f["type"] in {"string", "enum", "date", "datetime"}] p = prompt or "" def pick_title(*cands: str, default: str) -> str: for c in cands: if c and c in p: # 取含该关键词的短句 m = re.search(rf"([^\n：:]{{0,10}}{re.escape(c)}[^\n：:]{{0,16}})", p) if m: t = m.group(1).strip(" -—|｜") if 2 <= len(t) <= 32: return t return c return default # 多数值 + 类目/编码轴 → 堆叠看板（图 + 可选状态条 + 表），不绑定行业 stacked = ( len(numeric) >= 2 and (x_candidates or filters) and ( wants_line_chart(p) or "看板" in p or "示意图" in p or "监督" in p or "status_strip" in p or bool({"chainage", "dkilo"} & names) ) ) if stacked: x_field = next((n for n in ("chainage", "dkilo", "code", "name") if n in names), None) if not x_field: x_field = x_candidates[0] if x_candidates else (list_cols[0] if list_cols else "id") metrics = [f["name"] for f in numeric[:4]] # 优先常见度量名（若存在） preferred = [n for n in ( "value", "cjl_value", "amount", "qty", "score", "count", "design_value", "design_settlement_mm", "cum_value", "cum_settlement_mm", "pred_value", "pred_settlement_mm", ) if n in names] if preferred: # 主度量优先观测值，设计/累积作辅系列 primary = next( (n for n in ("cjl_value", "value", "cum_value", "cum_settlement_mm") if n in preferred), preferred[0], ) rest = [n for n in preferred if n != primary][:3] metrics = [primary] + rest widgets: list[dict[str, Any]] = [ { "type": "line_chart", "title": pick_title("示意图", "趋势图", "曲线", default="趋势图"), "x_field": x_field, "group_by": x_field, "metrics": metrics, "metric": metrics[0] if metrics else "count", "y_unit": "mm" if any("mm" in n or n.endswith("_value") for n in metrics) else "", "entity": entity_name, }, ] if "before_day" in names and "next_day" in names: widgets.append( { "type": "status_strip", "title": pick_title("监督", "状态条", "进度", default="状态"), "label_field": x_field, "value_field": "before_day", "secondary_field": "next_day", "warn_field": next((n for n in ("overdue_days", "exceed_days", "warn_flag") if n in names), ""), "variant": "stacked_days", "cycle_days": 30, "entity": entity_name, } ) elif "supervise_days" in names or "status_days" in names or "days" in names: value_f = next((n for n in ("supervise_days", "status_days", "days") if n in names), "supervise_days") widgets.append( { "type": "status_strip", "title": pick_title("状态条", "进度", "监督", default="状态"), "label_field": x_field, "value_field": value_f, "warn_field": next((n for n in ("overdue_days", "exceed_days", "warn_flag") if n in names), ""), "entity": entity_name, } ) table_cols = list_cols[:6] or [f["name"] for f in usable[:6]] widgets.append( { "type": "table", "title": pick_title("明细", "列表", "异常", default=f"{item_label}明细"), "entity": entity_name, "columns": table_cols, "filter_field": next((n for n in ("exceed_mm", "status", "flag") if n in names), ""), "filter_op": "gt" if "exceed_mm" in names else "eq", "filter_value": 0 if "exceed_mm" in names else "", } ) return widgets widgets = [ {"type": "kpi", "title": f"{item_label}数量", "metric": "count", "entity": entity_name}, ] if wants_line_chart(prompt) or (len(numeric) >= 2 and x_candidates): x_field = x_candidates[0] if x_candidates else (list_cols[0] if list_cols else "id") metrics = [f["name"] for f in numeric[:4]] widgets.append( { "type": "line_chart", "title": "趋势图", "x_field": x_field, "group_by": x_field, "metrics": metrics, "metric": metrics[0] if metrics else "count", "entity": entity_name, } ) g = categorical[0] if categorical else (filters[0] if filters else None) if g: chart_type = "bar_chart" if wants_line_chart(prompt) else "pie_chart" widgets.append( { "type": chart_type, "title": "分布", "metric": "count", "group_by": g, "entity": entity_name, } ) elif not any(w["type"] == "line_chart" for w in widgets): g2 = list_cols[0] if list_cols else "id" widgets.append( { "type": "pie_chart", "title": "分布", "metric": "count", "group_by": g2, "entity": entity_name, } ) return widgets _ALLOWED_TYPES = { "string", "text", "int", "bigint", "decimal", "boolean", "date", "datetime", "enum", "json", "file_ref", } _TYPE_ALIAS = { "float": "decimal", "float32": "decimal", "float64": "decimal", "double": "decimal", "number": "decimal", "numeric": "decimal", "integer": "int", "int32": "int", "int64": "bigint", "long": "bigint", "bool": "boolean", "varchar": "string", "str": "string", "timestamp": "datetime", "timestamptz": "datetime", } def looks_like_faithful_dashboard(draft: dict[str, Any]) -> bool: """看板是否应按截图壳渲染：已标 preset，或为「图+条/表」分区结构（不限行业）。""" for p in draft.get("pages") or []: if not isinstance(p, dict) or p.get("type") != "dashboard": continue lay = p.get("layout") or {} if not isinstance(lay, dict): continue if lay.get("preset") in {"screenshot_faithful", "ops_monitor"}: return True widgets = lay.get("widgets") or [] types = {str(w.get("type") or "") for w in widgets if isinstance(w, dict)} # 截图类看板常见：主图 +（状态条或明细表），而非仅 KPI if "line_chart" in types and ("status_strip" in types or "table" in types): return True if "bar_chart" in types and "table" in types and "kpi" in types and len(types) >= 3: return True return False def apply_screenshot_faithful(draft: dict[str, Any], *, force: bool = False) -> dict[str, Any]: """通用：写入 ui_preset / layout.preset，并补齐主图/监督条默认，避免主图区空白。""" if not force and not looks_like_faithful_dashboard(draft): return draft meta = draft.setdefault("meta", {}) meta["ui_preset"] = "screenshot_faithful" ui = meta.get("ui") if isinstance(meta.get("ui"), dict) else {} meta["ui"] = ui # 从主实体字段推断主图取值/颜色/设计曲线（领域无关：按常见后缀与颜色字段） fields: list[dict[str, Any]] = [] for ent in draft.get("entities") or []: if isinstance(ent, dict) and ent.get("fields"): fields = list(ent["fields"]) break names = {str(f.get("name") or "") for f in fields if isinstance(f, dict)} def first(*cands: str) -> str: for c in cands: if c in names: return c return "" if not ui.get("value_field"): vf = first("cjl_value", "value", "cum_value", "cum_settlement_mm", "amount", "qty", "score") if vf: ui["value_field"] = vf if not ui.get("color_field"): cf = first("cjl_color", "color", "mark_color") if cf: ui["color_field"] = cf if not ui.get("design_field"): df = first("design_value", "design_settlement_mm", "target", "design") if df: ui["design_field"] = df if not ui.get("category_field"): cat = first("section_type", "category", "type", "kind") if cat: ui["category_field"] = cat # 截图还原默认着色测点图（有取值字段时）；显式 line/multi_series 除外 style = str(ui.get("chart_style") or "").lower() if style not in {"line", "multi_series", "line_chart", "section_marks", "colored_marks"}: if ui.get("value_field") or first("cjl_value", "value"): ui["chart_style"] = "section_marks" if not ui.get("y_unit") and ( ui.get("value_field", "").endswith("_mm") or "settlement" in ui.get("value_field", "") or ui.get("value_field") in {"cjl_value", "cum_value", "design_value"} ): ui["y_unit"] = "mm" if not ui.get("y_axis_label") and ui.get("y_unit") == "mm": ui["y_axis_label"] = "沉降量" if "invert_y" not in ui and ui.get("chart_style") in {"section_marks", "colored_marks"}: ui["invert_y"] = True if not ui.get("legend_items"): # 用单选选项生成占位图例色，避免图例全空 radios = ui.get("filter_radios") or ui.get("section_options") or [] palette = ["#8000FF", "#214080", "#868e96", "#0f766e", "#e8590c"] items = [] for i, lab in enumerate(radios): if not lab or lab in {"全部", "所有", "all", "All"}: continue items.append({"label": f"{lab}累计", "color": palette[i % len(palette)]}) if ui.get("design_field"): items.append({"label": "设计曲线", "color": ui.get("design_series_color") or "#e8590c"}) if items: ui["legend_items"] = items for p in draft.get("pages") or []: if not isinstance(p, dict): continue if p.get("type") == "dashboard": lay = p.setdefault("layout", {}) if isinstance(lay, dict): lay["preset"] = "screenshot_faithful" # 截图主区通常无顶部 KPI 卡片条；有图/条/表分区时去掉 kpi 以免破坏还原 widgets = [w for w in (lay.get("widgets") or []) if isinstance(w, dict)] types = {w.get("type") for w in widgets} if ("line_chart" in types or "bar_chart" in types) and ( "status_strip" in types or "table" in types ): widgets = [w for w in widgets if w.get("type") != "kpi"] # 补齐主图/监督条字段，避免渲染侧整段跳过 x_fallback = first("chainage", "dkilo", "code", "name") or ( next( ( str(f.get("name")) for f in fields if isinstance(f, dict) and f.get("type") in {"string", "enum", "date"} and f.get("name") not in _SYSTEM_FIELDS ), "id", ) ) for w in widgets: if w.get("type") in {"line_chart", "bar_chart"}: w.setdefault("x_field", x_fallback) w.setdefault("group_by", w.get("x_field") or x_fallback) if not w.get("metrics") and not w.get("metric"): m0 = ui.get("value_field") or first( "cjl_value", "value", "cum_value", "cum_settlement_mm" ) if m0: w["metric"] = m0 w["metrics"] = [m0] if ui.get("design_field"): w["metrics"] = [m0, ui["design_field"]] if ui.get("y_unit"): w.setdefault("y_unit", ui["y_unit"]) if w.get("type") == "status_strip": w.setdefault("label_field", x_fallback) if "before_day" in names and "next_day" in names: w.setdefault("value_field", "before_day") w.setdefault("secondary_field", "next_day") w.setdefault("variant", "stacked_days") w.setdefault("cycle_days", 30) elif not w.get("value_field"): w["value_field"] = first( "supervise_days", "status_days", "before_day", "days" ) or "supervise_days" lay["widgets"] = widgets title = str(p.get("title") or "") # 去掉套话「概览」，保留业务名；不改写成固定行业标题 if title.endswith("概览") and len(title) > 2: p["title"] = title[: -len("概览")].rstrip() or title if not str(p["title"]).endswith("看板"): p["title"] = f"{p['title']}看板" if p["title"] else "看板" if p.get("type") == "list": lay = p.setdefault("layout", {}) if isinstance(lay, dict) and not lay.get("filter_style"): if ui.get("filter_style") or ui.get("section_options") or ui.get("filter_radios"): lay["filter_style"] = ui.get("filter_style") or "section_radios" return draft def harden_draft(draft: dict[str, Any], fallback: dict[str, Any]) -> dict[str, Any]: """发布前收口：非法 type/空结构回退，避免 LLM 润色后无法 publish。""" if not isinstance(draft, dict) or not draft.get("entities"): return fallback draft.setdefault("version", "1.0") draft.setdefault("storage", fallback.get("storage") or {"mode": "schema_per_app", "engine": "postgres"}) draft.setdefault("security", fallback.get("security") or {"visibility": "private", "roles": [], "row_policies": []}) draft.setdefault("apis", fallback.get("apis")) draft.setdefault("pages", fallback.get("pages") or []) meta = draft.setdefault("meta", {}) if not meta.get("slug"): meta["slug"] = (fallback.get("meta") or {}).get("slug") or "app" meta["slug"] = slugify(str(meta["slug"]), fallback="app") if not meta.get("name"): meta["name"] = (fallback.get("meta") or {}).get("name") or meta["slug"] entity_map: dict[str, str] = {} for ent in draft["entities"]: old = str(ent.get("name") or "") if not ent.get("name"): ent["name"] = "record" ent["name"] = slugify(str(ent["name"]), fallback="record") ent["table"] = slugify(str(ent.get("table") or ent["name"]), fallback=ent["name"]) if old: entity_map[old] = ent["name"] entity_map[slugify(old, fallback=ent["name"])] = ent["name"] ent.setdefault("primary_key", "id") fields = ent.get("fields") or [] if not fields: return fallback names = set() for i, f in enumerate(fields): fname = slugify(str(f.get("name") or f"col_{i}"), fallback=f"col_{i}") if fname in names: fname = f"{fname}_{i}" names.add(fname) f["name"] = fname t = str(f.get("type") or "string").lower().strip() t = _TYPE_ALIAS.get(t, t) if t not in _ALLOWED_TYPES: t = "string" f["type"] = t if not any(f.get("name") == ent["primary_key"] for f in fields): fields.insert(0, { "name": ent["primary_key"], "label": "ID", "type": "bigint", "nullable": False, "ui": {"widget": "hidden", "listable": False}, }) ent["fields"] = fields def resolve_entity(ref: Any) -> str: key = str(ref or "") if key in entity_map: return entity_map[key] sn = slugify(key, fallback="") if sn in entity_map: return entity_map[sn] return sn or key apis = draft.get("apis") or {} for r in apis.get("resources") or []: if isinstance(r, dict): r["entity"] = resolve_entity(r.get("entity")) path = str(r.get("path") or "").lstrip("/") r["path"] = "/" + (slugify(path, fallback="items") if path else "items") for p in draft.get("pages") or []: if isinstance(p, dict) and p.get("entity"): p["entity"] = resolve_entity(p.get("entity")) for rp in (draft.get("security") or {}).get("row_policies") or []: if isinstance(rp, dict) and rp.get("entity"): rp["entity"] = resolve_entity(rp.get("entity")) # 截图忠实：LLM 不得抹掉；有图或分区结构时强制恢复（不限行业） fb_meta = fallback.get("meta") or {} fb_src = fb_meta.get("source") or {} force_faithful = ( fb_meta.get("ui_preset") == "screenshot_faithful" or fb_src.get("screenshot_faithful") is True or bool(fb_src.get("image_refs")) or looks_like_faithful_dashboard(fallback) or looks_like_faithful_dashboard(draft) ) if force_faithful: if fb_meta.get("platform_title"): meta["platform_title"] = fb_meta["platform_title"] if fb_meta.get("ui"): meta["ui"] = {**(fb_meta.get("ui") or {}), **(meta.get("ui") or {})} if fb_meta.get("project_context") and not meta.get("project_context"): meta["project_context"] = fb_meta["project_context"] # 保留用户/启发式名称，禁止被润色成套话 fb_name = str(fb_meta.get("name") or "") cur_name = str(meta.get("name") or "") if fb_name and (not cur_name or "数据管理" in cur_name or cur_name.endswith("概览")): meta["name"] = fb_name apply_screenshot_faithful(draft, force=True) # 功能按键文案：LLM 不得抹掉用户/截图还原的 labels for p, fp in zip(draft.get("pages") or [], fallback.get("pages") or []): if not isinstance(p, dict) or not isinstance(fp, dict): continue flay = fp.get("layout") or {} if not isinstance(flay, dict): continue lay = p.setdefault("layout", {}) if not isinstance(lay, dict): continue if flay.get("action_labels"): merged = dict(flay.get("action_labels") or {}) merged.update(lay.get("action_labels") or {}) for k, v in (flay.get("action_labels") or {}).items(): merged[k] = v lay["action_labels"] = merged if flay.get("actions") and ( not lay.get("actions") or len(lay.get("actions") or []) < 2 ): lay["actions"] = list(flay["actions"]) if flay.get("preset") == "screenshot_faithful": lay["preset"] = "screenshot_faithful" if flay.get("filter_style") and not lay.get("filter_style"): lay["filter_style"] = flay["filter_style"] return draft def infer_type(values: list[Any]) -> tuple[str, list[str] | None, float]: samples = [v for v in values if v is not None and str(v).strip() != ""] if not samples: return "string", None, 0.4 strs = [str(v).strip() for v in samples] uniq = sorted(set(strs)) # 高基数字段不当 enum，避免导入新值失败 if len(uniq) <= 8 and len(uniq) <= max(2, int(len(strs) * 0.25)) and len(strs) >= 4: if all(len(u) <= 32 for u in uniq): return "enum", uniq[:50], 0.85 int_ok = True dec_ok = True for s in strs: try: if "." in s or "e" in s.lower(): float(s) int_ok = False else: int(s) except ValueError: int_ok = False try: float(s) except ValueError: dec_ok = False break if int_ok: return "int", None, 0.9 if dec_ok: return "decimal", None, 0.88 if all(re.match(r"^\d{4}-\d{2}-\d{2}", s) for s in strs): return "datetime", None, 0.8 avg_len = sum(len(s) for s in strs) / len(strs) if avg_len > 80: return "text", None, 0.7 return "string", None, 0.75 def parse_excel(content: bytes) -> dict[str, Any]: wb = load_workbook(io.BytesIO(content), read_only=True, data_only=True) # 默认首表（不按行业表名挑选） ws = wb[wb.sheetnames[0]] if wb.sheetnames else wb.active rows = list(ws.iter_rows(values_only=True)) if not rows: return {"sheets": []} headers = [str(h).strip() if h is not None else f"col_{i}" for i, h in enumerate(rows[0])] data_rows = rows[1:2001] # 最多采 2000 行做类型推断 / 蓝图 cols: dict[str, list[Any]] = {h: [] for h in headers} sample_rows: list[list[Any]] = [] for r in data_rows: if r is None or all(c is None or str(c).strip() == "" for c in r): continue vals = [] for i, h in enumerate(headers): v = r[i] if i < len(r) else None cols[h].append(v) vals.append(v) sample_rows.append(vals) inferred = {} for h in headers: t, enum_vals, _ = infer_type(cols[h]) inferred[h] = {"type": t, "enum_values": enum_vals} return { "sheets": [ { "name": ws.title, "headers": headers, "sample_rows": sample_rows[:5], "row_dicts": [ {headers[i]: (r[i] if i < len(r) else None) for i in range(len(headers))} for r in sample_rows[:500] ], "inferred": inferred, "columns": cols, "row_count": len(sample_rows), } ] } def _chart_rows_from_json(data: dict[str, Any]) -> list[dict[str, Any]]: """展开并行数组图表 JSON（mcljChartData / cljdChartData）→ 行字典。不写死行业文案。""" mclj = data.get("mcljChartData") or {} cljd = data.get("cljdChartData") or {} dkilos = list(mclj.get("dkilo") or []) cjls = list(mclj.get("cjl") or []) sjcjls = list(mclj.get("sjcjl") or []) workinfos = list(cljd.get("workinfo_kilo") or []) befores = list(cljd.get("beforeDay") or cljd.get("before_day") or []) nexts = list(cljd.get("nextDay") or cljd.get("next_day") or []) freqs = list(cljd.get("frequency") or []) # 颜色 → 分组N（按出现顺序）；不臆造路基/桥梁等业务词 color_to_group: dict[str, str] = {} rows: list[dict[str, Any]] = [] for i, dkilo in enumerate(dkilos): parts = str(dkilo).split("_") point_code = parts[0] if parts else str(dkilo) mileage_m = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else None chainage = "" if mileage_m is not None: km, rem = divmod(mileage_m, 1000) chainage = f"DK{km}+{rem:03d}" cjl = cjls[i] if i < len(cjls) else {} if not isinstance(cjl, dict): cjl = {"value": cjl, "color": ""} try: cjl_value = float(str(cjl.get("value") or 0).replace(",", "")) except ValueError: cjl_value = 0.0 color = str(cjl.get("color") or "").strip().lower() try: curve = abs(float(sjcjls[i])) if i < len(sjcjls) else 0.0 except (TypeError, ValueError): curve = 0.0 design = 3.0 if curve >= 10 else (curve or 3.0) cum_abs = abs(cjl_value) exceed = round(max(0.0, cum_abs - design), 2) if color and color not in color_to_group: color_to_group[color] = f"分组{len(color_to_group) + 1}" section = color_to_group.get(color) or f"分组{(i % 4) + 1}" before = float(befores[i]) if i < len(befores) else 0.0 nxt = float(nexts[i]) if i < len(nexts) else 0.0 freq = float(freqs[i]) if i < len(freqs) else 0.0 rows.append( { "dkilo": str(dkilo), "chainage": chainage or str(dkilo), "point_code": point_code, "section_type": section, "worksite": f"分组{i // 40 + 1}", "value": cjl_value, "color": color, "cjl_value": cjl_value, "cjl_color": color, "design_value": design, "cum_value": cjl_value, "pred_value": round(cjl_value * 0.9, 2), "exceed_mm": exceed, "exceed_days": int(round(exceed)) if exceed > 0 else 0, "supervise_days": int(round(before)) if before > 1 else 28, "overdue_days": max(0, int(round(nxt / 30)) if nxt > 50 else 0), "before_day": before, "next_day": nxt, "frequency": freq, "workinfo_kilo": workinfos[i] if i < len(workinfos) else "", "status": "异常" if exceed > 0 else "正常", } ) return rows def _flatten_json_object(obj: dict[str, Any], prefix: str = "") -> dict[str, Any]: out: dict[str, Any] = {} for k, v in obj.items(): key = slugify(f"{prefix}_{k}" if prefix else str(k), fallback="field") if isinstance(v, dict) and "value" in v: out[key] = v.get("value") if "color" in v: out[f"{key}_color"] = v.get("color") elif isinstance(v, dict): out.update(_flatten_json_object(v, key)) elif isinstance(v, list): out[key] = json.dumps(v, ensure_ascii=False) else: out[key] = v return out def _rows_to_sheet(name: str, row_dicts: list[dict[str, Any]], *, link_field: str | None = None, parent: str | None = None) -> dict[str, Any]: if not row_dicts: return {"name": name, "headers": [], "sample_rows": [], "inferred": {}, "columns": {}, "row_count": 0} headers = list(row_dicts[0].keys()) for r in row_dicts[1:]: for k in r: if k not in headers: headers.append(k) # 用全量做类型推断（上限 2000） use = row_dicts[:2000] cols: dict[str, list[Any]] = {h: [r.get(h) for r in use] for h in headers} inferred = {} for h in headers: t, enum_vals, _ = infer_type(cols[h]) inferred[h] = {"type": t, "enum_values": enum_vals} sheet: dict[str, Any] = { "name": name, "headers": headers, "sample_rows": [[r.get(h) for h in headers] for r in use[:5]], "row_dicts": use[:500], "inferred": inferred, "columns": cols, "source": "json", "row_count": len(row_dicts), } if link_field: sheet["link_field"] = link_field if parent: sheet["parent_entity"] = parent return sheet def parse_json_table(content: bytes) -> dict[str, Any]: """JSON → 一张或多张关联表（sheets），按结构拆实体。""" text = content.decode("utf-8-sig") data = json.loads(text) sheets: list[dict[str, Any]] = [] if isinstance(data, dict) and ("mcljChartData" in data or "cljdChartData" in data): points = _chart_rows_from_json(data) primary = "chart_points" sheets.append(_rows_to_sheet(primary, points)) # 并行状态数组拆成关联表（按 dkilo 关联） cljd = data.get("cljdChartData") or {} mclj = data.get("mcljChartData") or {} dkilos = list(mclj.get("dkilo") or []) workinfos = list(cljd.get("workinfo_kilo") or []) befores = list(cljd.get("beforeDay") or cljd.get("before_day") or []) nexts = list(cljd.get("nextDay") or cljd.get("next_day") or []) freqs = list(cljd.get("frequency") or []) superv = [] for i, dk in enumerate(dkilos): superv.append( { "dkilo": str(dk), "workinfo_kilo": workinfos[i] if i < len(workinfos) else "", "before_day": float(befores[i]) if i < len(befores) else 0, "next_day": float(nexts[i]) if i < len(nexts) else 0, "frequency": float(freqs[i]) if i < len(freqs) else 0, "supervise_days": int(round(float(befores[i]))) if i < len(befores) and float(befores[i]) > 1 else 28, } ) if superv: sheets.append( _rows_to_sheet( "status_series", superv, link_field="dkilo", parent=primary, ) ) exceed = [ { "dkilo": r["dkilo"], "worksite": r.get("worksite"), "point_code": r.get("point_code"), "design_value": r.get("design_value"), "cum_value": r.get("cum_value"), "exceed_mm": r.get("exceed_mm"), "exceed_days": r.get("exceed_days"), "status": r.get("status"), } for r in points if float(r.get("exceed_mm") or 0) > 0 ] if exceed: sheets.append( _rows_to_sheet( "alert_points", exceed, link_field="dkilo", parent=primary, ) ) return {"sheets": sheets, "relations": [ {"from": "status_series", "to": primary, "on": "dkilo"}, {"from": "alert_points", "to": primary, "on": "dkilo"}, ]} if isinstance(data, list) and data and all(isinstance(x, dict) for x in data): return {"sheets": [_rows_to_sheet("records", [_flatten_json_object(x) for x in data])]} if isinstance(data, dict): # 多个 list[object] → 多实体；尝试 *_id 关联 list_keys = [ k for k, v in data.items() if isinstance(v, list) and v and isinstance(v[0], dict) ] if len(list_keys) >= 1: for k in list_keys: rows = [_flatten_json_object(x) for x in data[k]] sheets.append(_rows_to_sheet(slugify(k, fallback="records"), rows)) relations = [] names = [s["name"] for s in sheets] for s in sheets: for h in s.get("headers") or []: if h.endswith("_id") or h.endswith("id") and h != "id": base = h[:-3] if h.endswith("_id") else h # 匹配其他实体名 for other in names: if other != s["name"] and (base in other or other.startswith(base)): relations.append({"from": s["name"], "to": other, "on": h}) s["link_field"] = h s["parent_entity"] = other # 同名字段关联 for other in sheets: if other["name"] == s["name"]: continue common = set(s.get("headers") or []) & set(other.get("headers") or []) common -= {"id", "created_at", "updated_at", "tenant_id"} if len(common) == 1: on = next(iter(common)) relations.append({"from": s["name"], "to": other["name"], "on": on}) s.setdefault("link_field", on) s.setdefault("parent_entity", other["name"]) return {"sheets": sheets, "relations": relations} for key in ("items", "rows", "data", "records", "list"): if isinstance(data.get(key), list) and data[key] and isinstance(data[key][0], dict): return { "sheets": [ _rows_to_sheet( slugify(key, fallback="records"), [_flatten_json_object(x) for x in data[key]], ) ] } return {"sheets": [_rows_to_sheet("records", [_flatten_json_object(data)])]} return {"sheets": []} def parse_tabular_upload(filename: str, content: bytes) -> dict[str, Any]: name = (filename or "").lower() # 首页抓包：key/value（菜单 HTML + 图表 JSON + 统计 JSON …） if name.endswith(".config") or name in {"url.config", "url_config", "capture.config"}: from url_capture import parse_url_config cap = parse_url_config(content) warnings = list(cap.get("warnings") or []) chart = cap.get("chart_json") if not chart: return { "sheets": [], "capture": cap, "warnings": warnings + ["url.config 中未找到图表 JSON（mcljChartData/cljdChartData）"], } raw = json.dumps(chart, ensure_ascii=False).encode("utf-8") meta = parse_json_table(raw) meta["capture"] = cap meta["warnings"] = warnings meta.setdefault("ui_hints", {}) if cap.get("nav_items"): meta["ui_hints"]["nav_items"] = cap["nav_items"] stats = cap.get("stats_json") or {} if isinstance(stats, dict) and stats: meta["ui_hints"]["stats_raw"] = stats if cap.get("exceed_json") is None and any( p.get("empty") and "gonghou" in (p.get("action") or "").lower() for p in (cap.get("pairs") or []) ): warnings.append( "hightchartGonghouCjlCX 返回为空：超限表由主图表数据按 exceed 条件推导（非原接口）" ) meta["warnings"] = warnings return meta if name.endswith(".json"): return parse_json_table(content) if name.endswith(".csv"): # 简易 CSV：首行表头 text = content.decode("utf-8-sig") import csv as _csv reader = _csv.DictReader(io.StringIO(text)) rows = list(reader)[:5000] if not rows: return {"sheets": []} headers = list(rows[0].keys()) use = rows[:2000] cols: dict[str, list[Any]] = {h: [r.get(h) for r in use] for h in headers} inferred = {h: {"type": infer_type(cols[h])[0], "enum_values": infer_type(cols[h])[1]} for h in headers} return { "sheets": [ { "name": "csv_import", "headers": headers, "sample_rows": [[r.get(h) for h in headers] for r in use[:5]], "inferred": inferred, "columns": cols, "source": "csv", "row_count": len(rows), } ] } return parse_excel(content) def _fields_from_sheet(sheet: dict[str, Any]) -> tuple[list[dict[str, Any]], list[str], list[str], list[str], list[float]]: """从一张 sheet 推断字段定义。""" headers = sheet.get("headers") or [] inferred = sheet.get("inferred") or {} cols = sheet.get("columns") or {} fields: list[dict[str, Any]] = [ { "name": "id", "label": "ID", "type": "bigint", "nullable": False, "ui": {"widget": "hidden", "listable": False}, } ] filters: list[str] = [] sorts: list[str] = [] list_cols: list[str] = [] confidences: list[float] = [] for i, h in enumerate(headers): fname = to_field_name(h, i) if fname == "id": fname = f"col_{i}" info = inferred.get(h, {"type": "string", "enum_values": None}) ftype = info["type"] _, _, c = infer_type(cols.get(h, [])) confidences.append(c) field: dict[str, Any] = { "name": fname, "label": str(h)[:64], "type": ftype, "nullable": True, "max_length": 128 if ftype == "string" else None, "ui": { "widget": "select" if ftype == "enum" else "input", "listable": True, "filterable": ftype in {"enum", "string"}, "sortable": ftype in {"int", "decimal", "datetime", "date", "string"}, "width": "md", }, "from_excel": { "column": h, "sample_values": [str(x) for x in cols.get(h, [])[:3] if x is not None], }, } if ftype == "string": uniq = sorted( {str(v).strip() for v in cols.get(h, []) if v is not None and str(v).strip() != ""} ) id_like = any(x in fname for x in ("code", "dkilo", "chainage", "kilo", "id", "name", "color")) if not id_like and 1 < len(uniq) <= 8 and len(uniq) <= max(2, int(len(cols.get(h, [])) * 0.25)): ftype = "enum" field["type"] = "enum" field["enum_values"] = uniq field["ui"]["widget"] = "select" field["ui"]["filterable"] = True if ftype == "enum" and info.get("enum_values"): field["enum_values"] = info["enum_values"][:24] field["nullable"] = True field["ui"]["widget"] = "select" if ftype == "decimal": field["precision"] = 12 field["scale"] = 2 field["ui"]["widget"] = "number" if ftype == "int": field["ui"]["widget"] = "number" if ftype == "datetime": field["ui"]["widget"] = "datetime" if ftype == "date": field["ui"]["widget"] = "datepicker" field = {k: v for k, v in field.items() if v is not None} fields.append(field) list_cols.append(fname) if field["ui"].get("filterable"): filters.append(fname) if field["ui"].get("sortable"): sorts.append(fname) return fields, filters, sorts, list_cols, confidences def build_blueprint( prompt: str, excel_meta: dict[str, Any] | None, storage_mode: str, image_count: int, html_count: int = 0, ) -> tuple[dict[str, Any], list[str], float]: warnings: list[str] = [] confidences: list[float] = [] app_name = infer_app_name(prompt or "") slug = slugify(app_name, fallback="app") entities: list[dict[str, Any]] = [] resources: list[dict[str, Any]] = [] pages: list[dict[str, Any]] = [] primary_entity = "record" primary_fields: list[dict[str, Any]] = [] primary_filters: list[str] = [] primary_list_cols: list[str] = [] label = "数据" sheets = (excel_meta or {}).get("sheets") or [] relations = (excel_meta or {}).get("relations") or [] if sheets: for si, sheet in enumerate(sheets): ename = slugify(sheet.get("name") or f"record_{si}", fallback=f"record_{si}") if ename in {"sheet1", "sheet", "sheet_1"}: ename = "record" if si == 0 else f"record_{si}" fields, filters, sorts, list_cols, confs = _fields_from_sheet(sheet) confidences.extend(confs) ent: dict[str, Any] = { "name": ename, "table": ename, "label": str(sheet.get("name") or ename)[:32], "primary_key": "id", "fields": fields, "indexes": [], } if sheet.get("parent_entity") and sheet.get("link_field"): ent["relation"] = { "parent": slugify(str(sheet["parent_entity"]), fallback="record"), "on": to_field_name(str(sheet["link_field"]), 0), } entities.append(ent) path = f"/{ename}" if ename.endswith("s") else f"/{ename}s" resources.append( { "entity": ename, "path": path, "operations": ["list", "get", "create", "update", "delete", "import", "export"], "list": { "default_page_size": 50, "max_page_size": 2000, "allowed_filters": filters[:12], "allowed_sorts": sorts[:12], }, } ) if si == 0: primary_entity = ename primary_fields = fields primary_filters = filters primary_list_cols = list_cols label = ent["label"] rc = sheet.get("row_count") if rc: warnings.append(f"主表 {ename} 识别到约 {rc} 行，发布后请导入完整 xlsx/json") else: pages.append( { "id": f"{ename}_list", "title": f"{ent['label']}", "route": f"/{ename}", "type": "list", "entity": ename, "layout": { "columns": list_cols[:8], "filters": filters[:5], "actions": ["create", "edit", "delete", "export", "import", "refresh"], "action_labels": dict(DEFAULT_ACTION_LABELS), }, } ) if relations: warnings.append(f"已按 JSON 字段关联生成 {len(relations)} 条表关系") else: warnings.append("未上传数据文件，已生成通用文本字段草案") confidences.append(0.55) primary_fields = [ {"name": "id", "label": "ID", "type": "bigint", "nullable": False, "ui": {"widget": "hidden", "listable": False}}, {"name": "title", "label": "标题", "type": "string", "nullable": False, "max_length": 128, "ui": {"widget": "input", "listable": True, "sortable": True}}, {"name": "content", "label": "内容", "type": "text", "nullable": True, "ui": {"widget": "textarea", "listable": True}}, {"name": "status", "label": "状态", "type": "enum", "nullable": True, "enum_values": ["草稿", "已发布"], "ui": {"widget": "select", "listable": True, "filterable": True}}, ] primary_filters = ["status"] primary_list_cols = ["title", "content", "status"] entities = [{"name": "record", "table": "record", "label": "数据", "primary_key": "id", "fields": primary_fields, "indexes": []}] resources = [{ "entity": "record", "path": "/records", "operations": ["list", "get", "create", "update", "delete", "import", "export"], "list": {"default_page_size": 50, "max_page_size": 2000, "allowed_filters": primary_filters, "allowed_sorts": ["title"]}, }] if image_count > 0 and wants_screenshot_layout(prompt or "", image_count, html_count): warnings.append(f"已收到 {image_count} 张截图：展示页将按截图分区还原") confidences.append(0.75) elif image_count > 0: warnings.append(f"已收到 {image_count} 张截图（用户要求不按截图布局）") confidences.append(0.7) if html_count > 0: warnings.append(f"已收到 {html_count} 个页面 HTML/MHTML：文案与控件优先按页面源码还原") confidences.append(0.8) ui_hints = parse_ui_hints(prompt or "", image_count, html_count) has_dashboard = ( wants_dashboard(prompt or "", image_count, html_count) or image_count > 0 or html_count > 0 or bool(ui_hints.get("ui_preset")) ) confidence = sum(confidences) / len(confidences) if confidences else 0.6 if prompt and excel_meta: confidence = min(0.95, confidence + 0.05) if ui_hints.get("app_name"): app_name = ui_hints["app_name"] if ui_hints.get("slug"): slug = slugify(str(ui_hints["slug"]), fallback=slug) item_label = infer_item_label(prompt or "", label) list_title = ui_hints.get("list_title") or f"{item_label}列表" create_title = ui_hints.get("create_title") or f"新增{item_label}" dash_title = ui_hints.get("dash_title") or f"{item_label}看板" filter_style = ui_hints.get("filter_style") or "default" # 有截图或用户要求还原 → 通用截图忠实（不绑定行业默认文案） faithful = ( ui_hints.get("ui_preset") == "screenshot_faithful" or wants_screenshot_layout(prompt or "", image_count, html_count) ) layout_preset = "screenshot_faithful" if faithful else "default" if faithful: ui_hints.setdefault("ui", {}) filter_style = ui_hints.get("filter_style") or filter_style if ui_hints.get("app_name"): app_name = ui_hints["app_name"] if ui_hints.get("list_title"): list_title = ui_hints["list_title"] if ui_hints.get("create_title"): create_title = ui_hints["create_title"] if ui_hints.get("dash_title"): dash_title = ui_hints["dash_title"] elif dash_title.endswith("概览"): dash_title = dash_title.replace("概览", "看板") # 从数据字段补齐筛选（不写死行业词：有 enum 才出单选，有 worksite 类字段才出下拉） ui_block = ui_hints.setdefault("ui", {}) field_by_name = {f["name"]: f for f in primary_fields} if not ui_block.get("section_options") and not ui_block.get("filter_radios"): for cand in ("section_type", "category", "type", "kind", "status"): f = field_by_name.get(cand) ev = (f or {}).get("enum_values") or [] if len(ev) >= 2: ui_block["section_options"] = list(ev)[:8] ui_block["filter_radios"] = list(ev)[:8] ui_block["filter_style"] = "section_radios" ui_block["radio_field"] = cand filter_style = "section_radios" break if not ui_block.get("select_field") and not ui_block.get("filter_select_field"): for cand in ("worksite", "site", "store", "warehouse", "dept", "org"): if cand in field_by_name: ui_block["select_field"] = cand break if ui_hints.get("ui", {}).get("filter_hint"): pass elif "点击" in (prompt or "") and "查看" in (prompt or ""): m = re.search(r"(点击[^。\n]{2,40})", prompt or "") if m: ui_block.setdefault("filter_hint", m.group(1).strip()) list_actions = ui_hints.get("actions") or ["create", "edit", "delete", "export", "import", "refresh"] action_labels = dict(DEFAULT_ACTION_LABELS) if ui_hints.get("action_labels"): action_labels.update(ui_hints["action_labels"]) # 列表列：优先通用 list_cols；若提示词/字段含常用业务列再穿插（不写死行业） preferred_cols = [c for c in primary_list_cols if c][:8] preferred_filters = primary_filters[:5] if ui_hints.get("filter_style") == "section_radios" and "section_type" in primary_filters: preferred_filters = [f for f in ("section_type", "worksite", "status") if f in primary_filters] or preferred_filters main_pages = [ { "id": f"{primary_entity}_list", "title": list_title, "route": f"/{primary_entity}", "type": "list", "entity": primary_entity, "layout": { "columns": preferred_cols, "filters": preferred_filters, "actions": list_actions, "action_labels": action_labels, "filter_style": filter_style, }, }, { "id": f"{primary_entity}_create", "title": create_title, "route": f"/{primary_entity}/create", "type": "form_create", "entity": primary_entity, "layout": { "form_fields": [f["name"] for f in primary_fields if f["name"] not in _SYSTEM_FIELDS], "actions": ["create"], "action_labels": {k: action_labels.get(k, DEFAULT_ACTION_LABELS.get(k, k)) for k in ("create", "save", "cancel")}, }, }, ] if has_dashboard: dash_widgets = build_dashboard_widgets( primary_fields, primary_filters, primary_list_cols, primary_entity, item_label, prompt or "" ) if faithful: # 有截图时去掉顶栏 KPI，避免破坏截图主区结构 types = {w.get("type") for w in dash_widgets} if ("line_chart" in types or "bar_chart" in types) and ( "status_strip" in types or "table" in types ): dash_widgets = [w for w in dash_widgets if w.get("type") != "kpi"] main_pages.append( { "id": f"{primary_entity}_dashboard", "title": dash_title, "route": f"/{primary_entity}/dashboard", "type": "dashboard", "entity": primary_entity, "layout": { "preset": layout_preset, "widgets": dash_widgets, }, } ) pages = main_pages + pages draft = { "version": "1.0", "meta": { "name": app_name, "slug": slug, "description": (prompt or "")[:500], "locale": "zh-CN", "source": {"prompt": prompt or ""}, "confidence": round(confidence, 2), "ui_preset": layout_preset if faithful else (ui_hints.get("ui_preset") or "default"), "project_context": ui_hints.get("project_context") or "", "platform_title": (ui_hints.get("ui") or {}).get("platform_title") or "", "ui": ui_hints.get("ui") or {}, }, "storage": {"mode": storage_mode or "schema_per_app", "engine": "postgres"}, "entities": entities, "apis": {"base_path": f"/api/v1/apps/{slug}", "resources": resources}, "pages": pages, "security": { "visibility": "private", "roles": [ { "name": "owner", "permissions": [ "app.read", "app.write", "app.admin", "row.create", "row.read", "row.update", "row.delete", "row.export", "row.import", ], }, {"name": "viewer", "permissions": ["app.read", "row.read", "row.export"]}, ], "row_policies": [{"entity": e["name"], "rule": "tenant_isolated"} for e in entities], }, "seed": {"import_excel": bool(excel_meta), "max_rows": 5000}, "relations": relations, } return draft, warnings, round(confidence, 2) @app.get("/health") def health(): return {"ok": True} @app.get("/api/v1/demo/fixtures") def demo_fixtures(): """测试素材清单（控制台一键预填用）。""" return fixtures_manifest() @app.get("/api/v1/demo/prompt") def demo_prompt(): try: text = read_prompt_text() except Exception as e: # noqa: BLE001 return {"ok": False, "prompt": "", "error": str(e)} return {"ok": True, "prompt": text, "slug": fixtures_manifest().get("slug")} @app.get("/api/v1/demo/file/{file_path:path}") def demo_file(file_path: str): return file_response(file_path) @app.post("/api/v1/preview") async def preview_create(payload: dict[str, Any]): """创建草稿预览（无需登录/发布），返回 preview_id 供 /#/preview/{id} 打开。""" bp = payload.get("blueprint") if not isinstance(bp, dict): return {"ok": False, "error": "blueprint required"} rows = payload.get("rows") if isinstance(payload.get("rows"), list) else [] resource = str(payload.get("resource") or "") pid = create_preview(bp, rows=rows, resource=resource) return { "ok": True, "preview_id": pid, "path": f"/#/preview/{pid}", "url_hint": f"WEB_BASE/#/preview/{pid}", } @app.get("/api/v1/preview/{preview_id}") def preview_get(preview_id: str): pack = get_preview(preview_id) if not pack: return {"ok": False, "error": "preview not found or expired"} return { "ok": True, "id": pack.get("id"), "blueprint": pack.get("blueprint"), "rows": pack.get("rows") or [], "resource": pack.get("resource") or "records", } @app.get("/api/v1/llm/providers") def llm_providers(): cfg = load_ai_config() default = ( (os.getenv("LLM_PROVIDER") or "").strip() or str(cfg.get("default_provider") or "deepseek") ) return {"providers": list_providers(), "default": default} @app.post("/api/v1/apps/generate") @app.post("/api/v1/apps:generate") # 兼容旧路径，勿再新增第三套 async def generate( prompt: str = Form(""), storage_mode: str = Form("schema_per_app"), llm_provider: str = Form(""), llm_model: str = Form(""), excel: UploadFile | None = File(None), data_file: UploadFile | None = File(None), # 兼容：json/csv/xlsx images: list[UploadFile] | None = File(None), layout_files: list[UploadFile] | None = File(None), # HTML/MHTML（Ctrl+S 另存为） ): with GenerateTrace() as trace: excel_meta = None upload = data_file if (data_file is not None and data_file.filename) else excel if upload is not None and upload.filename: content = await upload.read() if content: try: excel_meta = parse_tabular_upload(upload.filename, content) sheets = (excel_meta or {}).get("sheets") or [] trace.stage( "parse_data", upload.filename, sheets=len(sheets), rows=(sheets[0].get("row_count") if sheets else None), ) except Exception as e: # noqa: BLE001 trace.error(f"数据文件解析失败: {e}") return { "draft": None, "warnings": [f"数据文件解析失败: {e}"], "confidence": 0.0, "require_confirm": True, "generate_log": trace.as_payload(), } image_blobs: list[tuple[str, bytes, str]] = [] for img in images or []: if not img.filename: continue raw = await img.read() if raw: ctype = img.content_type or "image/png" image_blobs.append((img.filename, raw, ctype)) layout_parsed_list: list[dict] = [] layout_texts: list[str] = [] layout_names: list[str] = [] for lf in layout_files or []: if not lf.filename: continue raw = await lf.read() if not raw: continue layout_names.append(lf.filename) parsed = parse_layout_file(lf.filename, raw) layout_parsed_list.append(parsed) layout_texts.append(layout_summary_text(parsed)) image_count = len(image_blobs) html_count = len(layout_names) model_arg = llm_model.strip() or None user_prompt = prompt or "" trace.stage( "inputs", prompt_chars=len(user_prompt), images=image_count, layouts=html_count, llm_provider=llm_provider, llm_model=model_arg or "", image_names=[n for n, _, _ in image_blobs], layout_names=layout_names, data_file=(upload.filename if upload and upload.filename else ""), ) # HTML 按钮行并入用户需求，便于 parse_action_bar html_btn_extra = "" for parsed in layout_parsed_list: if parsed.get("ok") and parsed.get("buttons"): html_btn_extra += "\n操作：" + "、".join(parsed["buttons"][:20]) prompt_for_hints = user_prompt + html_btn_extra trace.stage("vision_extract", "开始理解截图") vision_text, vision_notes = understand_images(user_prompt, image_blobs, llm_provider, None) trace.stage( "vision_extract", "完成", chars=len(vision_text or ""), has_vision=bool(vision_text), notes=len(vision_notes or []), ) html_layout_text = "\n\n".join(layout_texts) effective_prompt = compose_effective_prompt( prompt_for_hints, image_count, vision_text, html_layout_text, html_count, ) faithful = wants_screenshot_layout(user_prompt, image_count, html_count) trace.stage( "compose_prompt", chars=len(effective_prompt or ""), screenshot_faithful=faithful, ) trace.stage("build_blueprint", "启发式草案") draft, warnings, confidence = build_blueprint( effective_prompt, excel_meta, storage_mode, image_count, html_count ) # url.config 抓包：导航页签 / 空接口告警 if excel_meta and isinstance(excel_meta, dict): for w in excel_meta.get("warnings") or []: warnings.append(str(w)) cap_ui = excel_meta.get("ui_hints") or {} if cap_ui.get("nav_items"): meta = draft.setdefault("meta", {}) ui = meta.setdefault("ui", {}) ui.setdefault("nav_items", cap_ui["nav_items"]) cap = excel_meta.get("capture") or {} if cap.get("pairs"): warnings.append( "已解析 url.config 抓包：" + "、".join( f"{p.get('action') or '?'}({'空' if p.get('empty') else p.get('kind')})" for p in cap["pairs"][:6] ) ) # 用 HTML 再补一轮 ui_hints（名称/页签），写回 meta for parsed in layout_parsed_list: if not parsed.get("ok"): continue hints = parse_ui_hints(prompt_for_hints, image_count, html_count) merge_layout_into_ui_hints(hints, parsed) meta = draft.setdefault("meta", {}) if hints.get("app_name") and ( not meta.get("name") or "数据管理" in str(meta.get("name")) ): meta["name"] = hints["app_name"] if hints.get("ui"): meta["ui"] = {**(meta.get("ui") or {}), **hints["ui"]} if hints["ui"].get("platform_title"): meta["platform_title"] = hints["ui"]["platform_title"] if hints.get("_html_buttons_line"): ab = parse_action_bar(hints["_html_buttons_line"]) if ab: for p in draft.get("pages") or []: if isinstance(p, dict) and p.get("type") == "list": lay = p.setdefault("layout", {}) lay["actions"] = ab["actions"] labels = dict(lay.get("action_labels") or {}) labels.update(ab["action_labels"]) lay["action_labels"] = labels # 页标题 + widget 标题（来自 HTML label） for p in draft.get("pages") or []: if not isinstance(p, dict): continue if p.get("type") == "list" and hints.get("list_title"): p["title"] = hints["list_title"] if p.get("type") == "form_create" and hints.get("create_title"): p["title"] = hints["create_title"] if p.get("type") == "dashboard": if hints.get("dash_title"): p["title"] = hints["dash_title"] elif p.get("title") and "概览" in str(p.get("title")): p["title"] = str(p["title"]).replace("概览", "看板") lay = p.setdefault("layout", {}) for w in lay.get("widgets") or []: if not isinstance(w, dict): continue if w.get("type") in {"line_chart", "bar_chart"} and hints.get("chart_title"): w["title"] = hints["chart_title"] if w.get("type") == "status_strip" and hints.get("strip_title"): w["title"] = hints["strip_title"] if w.get("type") == "table" and ( hints.get("table_title") or (hints.get("ui") or {}).get("table_title") ): w["title"] = hints.get("table_title") or hints["ui"]["table_title"] baseline = json.loads(json.dumps(draft)) # deep copy warnings.extend(vision_notes) for parsed in layout_parsed_list: if parsed.get("ok"): warnings.append(f"已解析页面布局: {parsed.get('filename')}") else: warnings.append(f"页面布局解析失败: {parsed.get('filename')} ({parsed.get('error')})") trace.notes(warnings) slug0 = (draft.get("meta") or {}).get("slug") or "" trace.stage( "build_blueprint", "完成", slug=slug0, pages=len(draft.get("pages") or []), confidence=confidence, ) excel_summary = "" if excel_meta and excel_meta.get("sheets"): sh = excel_meta["sheets"][0] excel_summary = f"sheet={sh.get('name')} headers={sh.get('headers')}" if html_layout_text: excel_summary = (excel_summary + " | " if excel_summary else "") + "html_layout=yes" trace.stage("llm_enhance", "代码模型润色蓝图") draft, llm_notes = enhance_blueprint_with_llm( draft, effective_prompt, excel_summary, llm_provider, model_arg ) warnings.extend(llm_notes) trace.notes(llm_notes) draft = harden_draft(draft, baseline) draft.setdefault("meta", {}).setdefault("source", {}) draft["meta"]["source"]["prompt"] = user_prompt draft["meta"]["source"]["llm_provider"] = llm_provider if model_arg: draft["meta"]["source"]["llm_model"] = model_arg if upload is not None and upload.filename: draft["meta"]["source"]["excel_ref"] = f"upload://{upload.filename}" if excel_meta and excel_meta.get("sheets"): draft["meta"]["source"]["data_format"] = excel_meta["sheets"][0].get("source") or "excel" if excel_meta["sheets"][0].get("row_count"): draft["meta"]["source"]["row_count"] = excel_meta["sheets"][0]["row_count"] if image_count: draft["meta"]["source"]["image_refs"] = [n for n, _, _ in image_blobs] if vision_text: draft["meta"]["source"]["vision_summary"] = vision_text[:1000] if html_count: draft["meta"]["source"]["layout_refs"] = layout_names draft["meta"]["source"]["layout_summary"] = html_layout_text[:2000] if image_count or html_count: draft["meta"]["source"]["screenshot_faithful"] = faithful if faithful: draft["meta"]["ui_preset"] = "screenshot_faithful" for p in draft.get("pages") or []: if isinstance(p, dict) and p.get("type") == "dashboard": p.setdefault("layout", {})["preset"] = "screenshot_faithful" draft["meta"]["source"]["generate_run_id"] = trace.run_id trace.stage("llm_enhance", "完成", slug=(draft.get("meta") or {}).get("slug") or "") fidelity_report: dict = {"skipped": True, "reason": "not_applicable"} if image_count and faithful: # 视觉摘录已写入 effective_prompt；此处：预览/计划 ↔ 原图打分 → 代码模型改蓝图 → 循环至 ≥95% trace.stage("fidelity_loop", "开始还原度迭代") preview_rows: list[dict] = [] if excel_meta and isinstance(excel_meta, dict): for sh in excel_meta.get("sheets") or []: if isinstance(sh, dict) and sh.get("row_dicts"): preview_rows = list(sh["row_dicts"])[:500] break draft, fidelity_report, fid_notes = await asyncio.to_thread( run_fidelity_loop, draft, image_blobs, vision_text, llm_provider=llm_provider, llm_model=model_arg, vision_model=None, # 视觉通道只用 VISION_MODEL，勿混入文本模型名 harden_fn=harden_draft, baseline=baseline, preview_rows=preview_rows, ) warnings.extend(fid_notes) draft = harden_draft(draft, baseline) draft.setdefault("meta", {}).setdefault("source", {}) draft["meta"]["source"]["fidelity"] = { "final_score": fidelity_report.get("final_score"), "passed": fidelity_report.get("passed"), "rounds": len(fidelity_report.get("rounds") or []), "target": fidelity_report.get("target"), } if fidelity_report.get("passed"): draft["meta"]["confidence"] = max( float(draft["meta"].get("confidence") or confidence or 0), 0.95, ) trace.stage( "fidelity_loop", "结束", final_score=fidelity_report.get("final_score"), passed=fidelity_report.get("passed"), rounds=len(fidelity_report.get("rounds") or []), target=fidelity_report.get("target"), ) else: trace.stage( "fidelity_loop", "跳过", reason=("no_images" if not image_count else "not_screenshot_faithful"), ) conf = float(draft.get("meta", {}).get("confidence") or confidence) log_payload = trace.as_payload() draft.setdefault("meta", {}).setdefault("source", {}) draft["meta"]["source"]["generate_log_file"] = log_payload.get("log_file") or "" return { "draft": draft, "warnings": warnings, "confidence": conf, "require_confirm": True, "llm_provider": llm_provider, "llm_model": model_arg or "", "fidelity": fidelity_report, "generate_log": log_payload, }