1808 lines
75 KiB
Python
1808 lines
75 KiB
Python
"""
|
||
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,
|
||
}
|