931 lines
37 KiB
Python
931 lines
37 KiB
Python
"""截图还原迭代闭环（通用，不绑定行业）。
|
||
|
||
流程：
|
||
1. 视觉模型从参考截图生成结构化提示（上游已完成）
|
||
2. 代码模型生成/润色蓝图（上游已完成）
|
||
3. 将「当前蓝图将渲染的 UI 计划」与参考截图交给视觉模型打分给差异
|
||
4. 代码模型按差异改蓝图
|
||
5. 重复直至 score>=目标 或达到最大轮次
|
||
|
||
可选：若环境有 playwright，用蓝图 meta 渲染简易预览 HTML 截图再对比（更接近真截图）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from llm import (
|
||
_api_key,
|
||
enhance_blueprint_with_llm,
|
||
resolve_vision_provider,
|
||
)
|
||
from generate_log import glog
|
||
|
||
|
||
def fidelity_target() -> int:
|
||
try:
|
||
return max(50, min(100, int(os.getenv("FIDELITY_TARGET", "95"))))
|
||
except ValueError:
|
||
return 95
|
||
|
||
|
||
def fidelity_max_rounds() -> int:
|
||
try:
|
||
return max(1, min(8, int(os.getenv("FIDELITY_MAX_ROUNDS", "4"))))
|
||
except ValueError:
|
||
return 4
|
||
|
||
|
||
def fidelity_enabled() -> bool:
|
||
v = (os.getenv("FIDELITY_LOOP") or "1").strip().lower()
|
||
return v not in {"0", "false", "off", "no"}
|
||
|
||
|
||
def ui_plan_from_draft(draft: dict[str, Any]) -> str:
|
||
"""把蓝图里会影响展示的字段压成「将渲染的界面计划」，供视觉对比。"""
|
||
meta = draft.get("meta") or {}
|
||
ui = meta.get("ui") or {}
|
||
pages = draft.get("pages") or []
|
||
dash = next((p for p in pages if isinstance(p, dict) and p.get("type") == "dashboard"), None)
|
||
widgets = ((dash or {}).get("layout") or {}).get("widgets") or []
|
||
|
||
lines = [
|
||
"【当前蓝图将渲染的界面计划 · 非像素截图，但须与参考图文案/结构一致】",
|
||
f"ui_preset: {meta.get('ui_preset') or ''}",
|
||
f"平台抬头 platform_title: {meta.get('platform_title') or ui.get('platform_title') or ''}",
|
||
f"平台英文 platform_subtitle: {ui.get('platform_subtitle') or ''}",
|
||
f"系统名 name: {meta.get('name') or ''}",
|
||
f"壳链 shell_links: {'、'.join(ui.get('shell_links') or [])}",
|
||
f"浮条 float_actions: {'、'.join(ui.get('float_actions') or [])}",
|
||
f"导航 nav_items: {'、'.join(ui.get('nav_items') or [])}",
|
||
f"工程/业务上下文 project_context: {meta.get('project_context') or ui.get('project_context') or ''}",
|
||
f"统计标签: {ui.get('stats_left_label') or ''} / {ui.get('stats_right_label') or ''}",
|
||
f"单选 filter_radios: {'、'.join(ui.get('filter_radios') or ui.get('section_options') or [])}",
|
||
f"单选字段 radio_field: {ui.get('radio_field') or ''}",
|
||
f"下拉 select_label/select_field: {ui.get('select_label') or ''} / {ui.get('select_field') or ''}",
|
||
f"提示 filter_hint: {ui.get('filter_hint') or ''}",
|
||
f"侧标 chart/strip/table: {ui.get('chart_side_label') or ''} | {ui.get('strip_side_label') or ''} | {ui.get('table_side_label') or ''}",
|
||
f"表标题 table_title: {ui.get('table_title') or ''}",
|
||
f"表头 table_headers: {'、'.join(ui.get('table_headers') or [])}",
|
||
f"chart_style: {ui.get('chart_style') or ''}",
|
||
f"invert_y: {ui.get('invert_y')}",
|
||
f"y_unit / y_axis_label: {ui.get('y_unit') or ''} / {ui.get('y_axis_label') or ''}",
|
||
]
|
||
legs = ui.get("legend_items") or []
|
||
if legs:
|
||
lines.append(
|
||
"图例 legend_items: "
|
||
+ "、".join(
|
||
f"{x.get('label')}({x.get('color')})" if isinstance(x, dict) else str(x) for x in legs
|
||
)
|
||
)
|
||
lines.append(f"看板页标题: {(dash or {}).get('title') or ''}")
|
||
for w in widgets:
|
||
if not isinstance(w, dict):
|
||
continue
|
||
lines.append(
|
||
"widget: "
|
||
+ json.dumps(
|
||
{
|
||
"type": w.get("type"),
|
||
"title": w.get("title"),
|
||
"x_field": w.get("x_field"),
|
||
"metrics": w.get("metrics"),
|
||
"metric": w.get("metric"),
|
||
"variant": w.get("variant"),
|
||
"value_field": w.get("value_field"),
|
||
"secondary_field": w.get("secondary_field"),
|
||
"cycle_days": w.get("cycle_days"),
|
||
"columns": w.get("columns"),
|
||
"filter_field": w.get("filter_field"),
|
||
"y_unit": w.get("y_unit"),
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _platform_base() -> str:
|
||
return (os.getenv("PLATFORM_BASE") or os.getenv("GATEWAY_BASE") or "http://127.0.0.1:8180").rstrip("/")
|
||
|
||
|
||
def _web_base() -> str:
|
||
return (os.getenv("WEB_BASE") or "http://127.0.0.1:5173").rstrip("/")
|
||
|
||
|
||
def real_screen_enabled() -> bool:
|
||
v = (os.getenv("FIDELITY_REAL_SCREEN") or "1").strip().lower()
|
||
return v not in {"0", "false", "off", "no"}
|
||
|
||
|
||
def _login_platform() -> str:
|
||
user = os.getenv("FIDELITY_USER") or "demo"
|
||
password = os.getenv("FIDELITY_PASSWORD") or "demo123"
|
||
with httpx.Client(timeout=30.0) as client:
|
||
r = client.post(
|
||
f"{_platform_base()}/api/v1/auth/login",
|
||
json={"username": user, "password": password},
|
||
)
|
||
r.raise_for_status()
|
||
return r.json()["access_token"]
|
||
|
||
|
||
def _publish_draft(draft: dict[str, Any], token: str) -> str:
|
||
"""发布蓝图并返回 slug。用独立 slug 避免误伤正式应用时可设 FIDELITY_SLUG_SUFFIX=_fidcheck。"""
|
||
meta = draft.setdefault("meta", {})
|
||
base_slug = str(meta.get("slug") or "app").strip() or "app"
|
||
suffix = (os.getenv("FIDELITY_SLUG_SUFFIX") or "").strip()
|
||
slug = f"{base_slug}{suffix}" if suffix else base_slug
|
||
meta["slug"] = slug
|
||
# 保证 apis.base_path 一致
|
||
apis = draft.setdefault("apis", {})
|
||
if isinstance(apis, dict):
|
||
apis["base_path"] = f"/api/v1/apps/{slug}"
|
||
with httpx.Client(timeout=120.0) as client:
|
||
r = client.post(
|
||
f"{_platform_base()}/api/v1/apps/{slug}/publish",
|
||
headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
|
||
json={"blueprint": draft},
|
||
)
|
||
r.raise_for_status()
|
||
return slug
|
||
|
||
|
||
def _capture_real_app_screenshot_impl(
|
||
draft: dict[str, Any],
|
||
*,
|
||
preview_rows: list[dict[str, Any]] | None = None,
|
||
) -> tuple[bytes | None, str, str | None]:
|
||
"""优先走草稿预览 /#/preview/{id}（无需登录/不必等发布）；失败再回退公开应用页。"""
|
||
try:
|
||
from playwright.sync_api import sync_playwright # type: ignore
|
||
except Exception as e: # noqa: BLE001
|
||
return None, f"无 playwright，无法截真页面: {e}", None
|
||
|
||
from preview_store import create_preview, get_preview
|
||
|
||
web = _web_base()
|
||
out_dir = Path(os.getenv("FIDELITY_SHOT_DIR") or "").resolve() if os.getenv("FIDELITY_SHOT_DIR") else None
|
||
goto_timeout = int(os.getenv("FIDELITY_GOTO_TIMEOUT_MS") or "45000")
|
||
slug = str(((draft.get("meta") or {}).get("slug")) or "preview")
|
||
preview_id = create_preview(
|
||
json.loads(json.dumps(draft)),
|
||
rows=preview_rows or [],
|
||
)
|
||
target = f"{web}/#/preview/{preview_id}"
|
||
pack = get_preview(preview_id) or {
|
||
"id": preview_id,
|
||
"ok": True,
|
||
"blueprint": draft,
|
||
"rows": preview_rows or [],
|
||
"resource": "records",
|
||
}
|
||
# 浏览器侧可能访问不到 AI；把预览包注入页面 + 拦截 /preview API
|
||
inject_payload = {
|
||
"ok": True,
|
||
"id": preview_id,
|
||
"blueprint": pack.get("blueprint") or draft,
|
||
"rows": pack.get("rows") or preview_rows or [],
|
||
"resource": pack.get("resource") or "records",
|
||
}
|
||
# 清掉本机过期登录态，避免 React 的 session-expired 把 /#/preview 踢回控制台
|
||
inject_js = (
|
||
"try{localStorage.removeItem('ajz_session');}catch(e){}"
|
||
"window.__AJZ_PREVIEW__ = "
|
||
+ json.dumps(inject_payload, ensure_ascii=False)
|
||
+ ";"
|
||
)
|
||
|
||
def _fmt_err(err: BaseException) -> str:
|
||
msg = str(err).strip() or repr(err)
|
||
return msg.replace("\n", " ")[:500]
|
||
|
||
try:
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True)
|
||
try:
|
||
page = browser.new_page(viewport={"width": 1600, "height": 1100})
|
||
page.add_init_script(inject_js)
|
||
|
||
def _fulfill_preview(route): # type: ignore
|
||
route.fulfill(
|
||
status=200,
|
||
content_type="application/json; charset=utf-8",
|
||
body=json.dumps(inject_payload, ensure_ascii=False).encode("utf-8"),
|
||
)
|
||
|
||
page.route("**/api/v1/preview/**", _fulfill_preview)
|
||
page.route("**/ai/api/v1/preview/**", _fulfill_preview)
|
||
page.goto(target, wait_until="domcontentloaded", timeout=goto_timeout)
|
||
# 再写一次，防止极端时序下 init 被覆盖
|
||
page.evaluate(inject_js)
|
||
try:
|
||
page.wait_for_selector(
|
||
".sf-shell-bar, .sf-root, .gen-app-ops, .gen-app, .sf-line-wrap",
|
||
timeout=45000,
|
||
)
|
||
except Exception:
|
||
snip = ""
|
||
try:
|
||
snip = page.locator("body").inner_text(timeout=2000)[:240]
|
||
except Exception:
|
||
snip = page.url
|
||
raise RuntimeError(
|
||
f"预览壳未出现（请确认 Web 已含 /#/preview 前端）。page={page.url} text={snip!r}"
|
||
) from None
|
||
# 仅当仍在登录台（无预览壳）时判定失败
|
||
if page.locator("button").filter(has_text="登录").count() and not page.locator(
|
||
".sf-shell-bar, .sf-root, .gen-app"
|
||
).count():
|
||
raise RuntimeError("预览页落到登录界面")
|
||
page.wait_for_timeout(2500)
|
||
if page.locator(".gen-app").count():
|
||
png = page.locator(".gen-app").first.screenshot(type="png")
|
||
else:
|
||
png = page.screenshot(type="png", full_page=True)
|
||
finally:
|
||
browser.close()
|
||
if out_dir:
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
(out_dir / f"preview_{preview_id}.png").write_bytes(png)
|
||
return (
|
||
png,
|
||
f"已截取草稿预览（/#/preview/{preview_id}，注入数据无需再拉 AI）参与对比",
|
||
slug,
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
# 回退：尝试发布后的公开应用页（仍不登录）
|
||
try:
|
||
token = _login_platform()
|
||
pub_slug = _publish_draft(json.loads(json.dumps(draft)), token)
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True)
|
||
try:
|
||
page = browser.new_page(viewport={"width": 1600, "height": 1100})
|
||
page.add_init_script(
|
||
"try{localStorage.removeItem('ajz_session');}catch(e){}"
|
||
)
|
||
page.goto(
|
||
f"{web}/#/app/{pub_slug}",
|
||
wait_until="domcontentloaded",
|
||
timeout=goto_timeout,
|
||
)
|
||
page.wait_for_selector(
|
||
".sf-shell-bar, .sf-root, .gen-app-ops, .gen-app",
|
||
timeout=45000,
|
||
)
|
||
if page.locator("button").filter(has_text="登录").count() and not page.locator(
|
||
".sf-shell-bar, .sf-root, .gen-app"
|
||
).count():
|
||
raise RuntimeError("公开应用页仍显示登录")
|
||
page.wait_for_timeout(2500)
|
||
if page.locator(".gen-app").count():
|
||
png = page.locator(".gen-app").first.screenshot(type="png")
|
||
else:
|
||
png = page.screenshot(type="png", full_page=True)
|
||
finally:
|
||
browser.close()
|
||
if out_dir:
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
(out_dir / f"real_{pub_slug}.png").write_bytes(png)
|
||
return png, f"预览失败后回退公开应用页（slug={pub_slug}）：{_fmt_err(e)}", pub_slug
|
||
except Exception as e2: # noqa: BLE001
|
||
return (
|
||
None,
|
||
f"真页面截图失败: preview={_fmt_err(e)}; fallback={_fmt_err(e2)}（目标 {target}）",
|
||
slug,
|
||
)
|
||
|
||
|
||
def capture_real_app_screenshot(
|
||
draft: dict[str, Any],
|
||
*,
|
||
preview_rows: list[dict[str, Any]] | None = None,
|
||
) -> tuple[bytes | None, str, str | None]:
|
||
"""
|
||
草稿预览截图。Windows+uvicorn 下禁止同进程 Playwright，改为独立 shot_worker.py。
|
||
"""
|
||
if not real_screen_enabled():
|
||
return None, "FIDELITY_REAL_SCREEN=0，跳过真页面截图", None
|
||
|
||
if (os.getenv("FIDELITY_SHOT_INPROCESS") or "").strip() in {"1", "true", "yes"}:
|
||
return _capture_real_app_screenshot_impl(draft, preview_rows=preview_rows)
|
||
|
||
import subprocess
|
||
import tempfile
|
||
|
||
from preview_store import create_preview
|
||
|
||
ai_dir = str(Path(__file__).resolve().parent)
|
||
worker = Path(ai_dir) / "shot_worker.py"
|
||
preview_id = create_preview(json.loads(json.dumps(draft)), rows=preview_rows or [])
|
||
resource = "records"
|
||
try:
|
||
apis = draft.get("apis") or {}
|
||
res0 = (apis.get("resources") or [None])[0]
|
||
if isinstance(res0, dict) and res0.get("path"):
|
||
resource = str(res0["path"]).lstrip("/")
|
||
elif (draft.get("resources") or [{}])[0].get("name"):
|
||
resource = str(draft["resources"][0]["name"])
|
||
except Exception:
|
||
pass
|
||
|
||
with tempfile.TemporaryDirectory(prefix="ajz_shot_") as td:
|
||
inp = Path(td) / "in.json"
|
||
outp = Path(td) / "out.json"
|
||
inp.write_text(
|
||
json.dumps(
|
||
{
|
||
"draft": draft,
|
||
"rows": preview_rows or [],
|
||
"preview_id": preview_id,
|
||
"resource": resource,
|
||
},
|
||
ensure_ascii=False,
|
||
default=str,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
env = os.environ.copy()
|
||
env["FIDELITY_SHOT_INPUT"] = str(inp)
|
||
env["FIDELITY_SHOT_OUTPUT"] = str(outp)
|
||
env.setdefault("WEB_BASE", _web_base())
|
||
env.pop("PYTHONASYNCIODEBUG", None)
|
||
creationflags = 0
|
||
if sys.platform == "win32":
|
||
# Detach from uvicorn's console/asyncio job so Playwright can spawn Chromium
|
||
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(
|
||
subprocess, "CREATE_NEW_PROCESS_GROUP", 0
|
||
)
|
||
try:
|
||
proc = subprocess.run(
|
||
[sys.executable, str(worker)],
|
||
cwd=ai_dir,
|
||
env=env,
|
||
timeout=190,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
creationflags=creationflags,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return None, "真页面截图子进程超时（190s）", None
|
||
err_tail = ((proc.stderr or "") + "\n" + (proc.stdout or "")).strip()[-600:]
|
||
if not outp.is_file():
|
||
return (
|
||
None,
|
||
f"真页面截图子进程无输出(code={proc.returncode}): {err_tail or 'no stderr'}",
|
||
None,
|
||
)
|
||
try:
|
||
data = json.loads(outp.read_text(encoding="utf-8"))
|
||
except Exception as e: # noqa: BLE001
|
||
return None, f"真页面截图结果无法解析: {e}; stderr={err_tail}", None
|
||
png_b64 = data.get("png_b64")
|
||
png = base64.b64decode(png_b64) if png_b64 else None
|
||
note = str(data.get("note") or "")
|
||
if not note:
|
||
note = f"shot_worker code={proc.returncode} stderr={err_tail or 'empty'}"
|
||
elif not png:
|
||
note = f"{note} | via=shot_worker code={proc.returncode}"
|
||
if err_tail:
|
||
note = f"{note} stderr={err_tail[:400]}"
|
||
else:
|
||
if "via=shot_worker" not in note:
|
||
note = f"{note} | via=shot_worker"
|
||
return png, note, data.get("slug")
|
||
|
||
|
||
def _try_preview_screenshot(
|
||
draft: dict[str, Any],
|
||
*,
|
||
preview_rows: list[dict[str, Any]] | None = None,
|
||
) -> tuple[bytes | None, str, bool]:
|
||
"""截取预览/真页面。返回 (png, note, is_real_screen)。"""
|
||
real, note, _slug = capture_real_app_screenshot(draft, preview_rows=preview_rows)
|
||
if real:
|
||
return real, note, True
|
||
return None, note, False
|
||
|
||
|
||
_COMPARE_PROMPT = """你是严格的 UI 还原验收员。必须「逐区对照真页面截图与参考图」，禁止因文案大致相似就给高分。
|
||
|
||
输入图顺序：
|
||
- 图1起：参考原系统截图（可能多张，为同页不同展开态，合并理解）
|
||
- 最后一张（若有）：【系统真页面截图】——只以这张判定还原度，文字计划仅作辅助
|
||
|
||
重要：文案计划一致 ≠ 画面合格。真页面缺主图、主图被监督条替代、分区错位，必须大幅扣分。
|
||
|
||
## 必检分区（zone_checklist，每项 true/false；任一项 false → 对应 fails 必写）
|
||
1) shell_header：顶栏平台名/系统名/右侧链接是否与原图一致
|
||
2) nav：次级导航文案与高亮是否大体一致
|
||
3) filter_bar：工程上下文、统计、单选、工点下拉、红色提示是否齐全且文案接近
|
||
4) main_chart：是否存在「纵断面/主曲线图」——带坐标轴的折线或柱线混合大图（不是数字格、不是表格、不是监督条）
|
||
5) chart_legend：主图上方/旁侧图例是否存在且系列名接近
|
||
6) status_strip：按时测量监督是否为上灰下绿双段竖条（不是单色数字格矩阵、不是把主图区填成监督条）
|
||
7) exceed_table：底部超限表标题与列是否接近
|
||
8) side_labels：左侧竖标是否与三区（主图/监督/表）一一对应且正向可读
|
||
9) float_dock：右侧浮条是否存在
|
||
|
||
## 硬性否决（任一条成立 → fails 必写，且 score 上限）
|
||
1) 文字倒置/镜像 → score≤35
|
||
2) 文字严重倾斜、重叠、裁切变形 → score≤45
|
||
3) 主图未横向铺满（右侧大片空白）→ score≤55
|
||
4) 顶栏/导航/筛选/侧标/表头/图例文案缺失或不一致 → score≤70
|
||
5) 分区结构不对（缺侧标、缺状态条、缺浮条、侧标与内容错位）→ score≤60
|
||
6) 图表形态不符，或真页面用监督条/数字格顶替主曲线图 → score≤50
|
||
7) zone_checklist.main_chart=false → score≤50
|
||
8) zone_checklist.status_strip=false 且原图有监督条 → score≤55
|
||
|
||
若没有「系统真页面截图」：
|
||
- 不得判定 pass=true；score 最高 75
|
||
- fails 必须含：「缺少真页面截图，无法验收字形方向/铺满/畸形」
|
||
|
||
## 禁止虚高
|
||
- 不得因为「顶栏和表头大致对上」就把 score 抬到 ≥90
|
||
- 缺主图或主图形态错误时，即使其它区都对，score 也不得超过 50
|
||
- fails 为空仅当 zone_checklist 全部为 true 且肉眼无明显布局差异
|
||
- 上一轮差异（若有）必须再次核对，未修复的必须继续写进 fails
|
||
|
||
上一轮未修复差异：
|
||
{prev_fails}
|
||
|
||
输出【仅 JSON 对象】：
|
||
{{
|
||
"score": 0到100的整数,
|
||
"pass": true或false,
|
||
"fails": ["区名: 现象；应如何改正"],
|
||
"passes": ["已对齐项"],
|
||
"zone_checklist": {{
|
||
"shell_header": true或false,
|
||
"nav": true或false,
|
||
"filter_bar": true或false,
|
||
"main_chart": true或false,
|
||
"chart_legend": true或false,
|
||
"status_strip": true或false,
|
||
"exceed_table": true或false,
|
||
"side_labels": true或false,
|
||
"float_dock": true或false
|
||
}},
|
||
"visual_checks": {{
|
||
"text_upright": true或false,
|
||
"chart_fills_width": true或false,
|
||
"no_malformation": true或false
|
||
}},
|
||
"patch_hints": {{
|
||
"meta.ui": {{}}
|
||
}}
|
||
}}
|
||
pass 仅当：score>={target} 且 fails 为空 且 zone_checklist 全 true 且 visual_checks 三项均为 true 且已提供真页面截图。
|
||
|
||
【截图视觉摘录】
|
||
{vision}
|
||
|
||
【当前蓝图界面计划】
|
||
{plan}
|
||
|
||
【是否有系统真页面截图】
|
||
{has_preview}
|
||
"""
|
||
|
||
|
||
_CRITICAL_ZONES = (
|
||
("main_chart", "主图区: 真页面缺少与原图对应的折线/柱状主曲线图（或被监督条/数字格顶替）", 50),
|
||
("status_strip", "监督条: 真页面缺少上灰下绿双段竖条监督区，或形态不符", 55),
|
||
("side_labels", "侧标: 左侧竖标缺失、错位或与内容区不对应", 60),
|
||
("exceed_table", "超限表: 底部超限测点表缺失或列/标题不符", 65),
|
||
("filter_bar", "筛选条: 工程上下文/统计/单选/提示缺失或不一致", 70),
|
||
("shell_header", "顶栏: 平台抬头/系统名/右侧链接缺失或不一致", 70),
|
||
("nav", "导航: 次级导航文案或高亮与原图不一致", 70),
|
||
("chart_legend", "图例: 主图图例缺失或系列名不符", 65),
|
||
("float_dock", "浮条: 右侧浮条缺失", 75),
|
||
)
|
||
|
||
|
||
def _blueprint_structure_fails(draft: dict[str, Any]) -> list[tuple[str, int]]:
|
||
"""蓝图结构硬校验：缺主图/监督条/表时不允许高分达标。"""
|
||
pages = draft.get("pages") or []
|
||
dash = next((p for p in pages if isinstance(p, dict) and p.get("type") == "dashboard"), None)
|
||
widgets = ((dash or {}).get("layout") or {}).get("widgets") or []
|
||
types = {str(w.get("type")) for w in widgets if isinstance(w, dict)}
|
||
ui = ((draft.get("meta") or {}).get("ui") or {}) if isinstance(draft.get("meta"), dict) else {}
|
||
out: list[tuple[str, int]] = []
|
||
if not (types & {"line_chart", "bar_chart", "area_chart"}):
|
||
out.append(("蓝图结构: dashboard 缺少折线/柱状主图 widget", 45))
|
||
if "status_strip" not in types:
|
||
out.append(("蓝图结构: dashboard 缺少 status_strip 监督条", 50))
|
||
if "table" not in types:
|
||
out.append(("蓝图结构: dashboard 缺少超限 table", 55))
|
||
if not (ui.get("chart_side_label") and ui.get("strip_side_label") and ui.get("table_side_label")):
|
||
out.append(("蓝图结构: meta.ui 三区侧标不全（chart/strip/table_side_label）", 60))
|
||
if not (ui.get("legend_items") or ui.get("chart_legend")):
|
||
# legend_items 更常见
|
||
if not ui.get("legend_items"):
|
||
out.append(("蓝图结构: meta.ui.legend_items 为空，主图图例难对齐", 65))
|
||
return out
|
||
|
||
|
||
def _enforce_score_caps(
|
||
score: int,
|
||
fails: list[str],
|
||
checks: dict[str, Any],
|
||
zones: dict[str, Any],
|
||
*,
|
||
has_real: bool,
|
||
draft: dict[str, Any] | None,
|
||
prev_fails: list[str] | None,
|
||
) -> tuple[int, list[str], bool]:
|
||
"""服务端强制扣分/否决，防止视觉模型虚高。"""
|
||
fails = list(fails)
|
||
if not has_real:
|
||
score = min(score, 75)
|
||
msg = "缺少真页面截图，无法验收字形方向/铺满/畸形"
|
||
if msg not in fails:
|
||
fails.append(msg)
|
||
|
||
for key, label, cap in (
|
||
("text_upright", "竖排/正文存在倒置或不可正向阅读", 35),
|
||
("chart_fills_width", "主图未横向铺满（右侧大片空白）", 55),
|
||
("no_malformation", "存在明显畸形/变形", 45),
|
||
):
|
||
if checks.get(key) is False:
|
||
score = min(score, cap)
|
||
if not any(label[:6] in f for f in fails):
|
||
fails.append(label)
|
||
|
||
for key, label, cap in _CRITICAL_ZONES:
|
||
if zones.get(key) is False:
|
||
score = min(score, cap)
|
||
if not any(key in f or label[:4] in f for f in fails):
|
||
fails.append(label)
|
||
|
||
# zone_checklist 未返回时，不默认信任满分
|
||
if has_real and not zones:
|
||
score = min(score, 80)
|
||
msg = "视觉打分未返回 zone_checklist，按不完整验收处理"
|
||
if msg not in fails:
|
||
fails.append(msg)
|
||
|
||
if draft is not None:
|
||
for label, cap in _blueprint_structure_fails(draft):
|
||
score = min(score, cap)
|
||
if label not in fails:
|
||
fails.append(label)
|
||
|
||
# 上一轮硬伤若本轮 fails 空但分数暴涨，仍保守封顶（模型常漏检）
|
||
if prev_fails and score >= 90:
|
||
severe = [f for f in prev_fails if any(k in f for k in ("主图", "折线", "分区", "侧标", "监督", "形态"))]
|
||
if severe and not fails:
|
||
score = min(score, 70)
|
||
fails.append(
|
||
"上一轮结构性差异疑似未复核："
|
||
+ "；".join(severe[:3])
|
||
+ "。请对照真页面复查，勿虚高满分"
|
||
)
|
||
|
||
ok = score >= fidelity_target() and not fails and has_real
|
||
if zones:
|
||
ok = ok and all(zones.get(k) is True for k, _, _ in _CRITICAL_ZONES)
|
||
for key in ("text_upright", "chart_fills_width", "no_malformation"):
|
||
if checks.get(key) is False:
|
||
ok = False
|
||
return score, fails, ok
|
||
|
||
|
||
def _parse_json_obj(text: str) -> dict[str, Any] | None:
|
||
t = (text or "").strip()
|
||
if t.startswith("```"):
|
||
t = t.strip("`")
|
||
if t.startswith("json"):
|
||
t = t[4:].lstrip()
|
||
try:
|
||
obj = json.loads(t)
|
||
return obj if isinstance(obj, dict) else None
|
||
except json.JSONDecodeError:
|
||
m = re.search(r"\{[\s\S]*\}", t)
|
||
if not m:
|
||
return None
|
||
try:
|
||
obj = json.loads(m.group(0))
|
||
return obj if isinstance(obj, dict) else None
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
def vision_score_fidelity(
|
||
ref_images: list[tuple[str, bytes, str]],
|
||
plan: str,
|
||
vision_extract: str,
|
||
preview_png: bytes | None,
|
||
provider_id: str | None = None,
|
||
model: str | None = None,
|
||
*,
|
||
is_real_screen: bool = False,
|
||
draft: dict[str, Any] | None = None,
|
||
prev_fails: list[str] | None = None,
|
||
) -> tuple[dict[str, Any], list[str]]:
|
||
notes: list[str] = []
|
||
target = fidelity_target()
|
||
pid, meta, model_name = resolve_vision_provider(provider_id, model)
|
||
if not meta.get("supports_vision"):
|
||
return {"score": 0, "pass": False, "fails": ["视觉模型不可用"], "passes": []}, [
|
||
f"{meta['label']} 无视觉能力，跳过打分"
|
||
]
|
||
api_key = _api_key(meta)
|
||
if not api_key:
|
||
return {"score": 0, "pass": False, "fails": ["未配置视觉 API Key"], "passes": []}, [
|
||
f"未配置 {meta.get('api_key_env')}"
|
||
]
|
||
|
||
has_real = bool(preview_png) and is_real_screen
|
||
prev_txt = "；".join((prev_fails or [])[:8]) if prev_fails else "（无，首轮）"
|
||
content: list[dict[str, Any]] = [
|
||
{
|
||
"type": "text",
|
||
"text": _COMPARE_PROMPT.format(
|
||
target=target,
|
||
vision=vision_extract or "（无）",
|
||
plan=plan,
|
||
prev_fails=prev_txt,
|
||
has_preview="有真页面截图（请逐区检查；缺主图不得高分）"
|
||
if has_real
|
||
else "无真页面截图（禁止判定达标）",
|
||
),
|
||
}
|
||
]
|
||
for name, raw, ctype in ref_images[:3]:
|
||
if not ctype.startswith("image/"):
|
||
ctype = "image/png"
|
||
b64 = base64.b64encode(raw).decode("ascii")
|
||
content.append({"type": "image_url", "image_url": {"url": f"data:{ctype};base64,{b64}"}})
|
||
if preview_png and is_real_screen:
|
||
b64 = base64.b64encode(preview_png).decode("ascii")
|
||
content.append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}})
|
||
|
||
base = (meta.get("base_url") or "").rstrip("/")
|
||
try:
|
||
with httpx.Client(timeout=150.0) as client:
|
||
def _post(mname: str):
|
||
return client.post(
|
||
f"{base}/chat/completions",
|
||
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
|
||
json={
|
||
"model": mname,
|
||
"temperature": 0.0,
|
||
"messages": [{"role": "user", "content": content}],
|
||
},
|
||
)
|
||
|
||
resp = _post(model_name)
|
||
if resp.status_code >= 400:
|
||
alt = "qwen-vl-plus"
|
||
err = resp.text[:300]
|
||
if model_name != alt:
|
||
notes.append(f"{model_name} 打分 {resp.status_code}，尝试 {alt}: {err}")
|
||
resp = _post(alt)
|
||
model_name = alt
|
||
resp.raise_for_status()
|
||
text = resp.json()["choices"][0]["message"]["content"]
|
||
if isinstance(text, list):
|
||
text = "".join((x.get("text") if isinstance(x, dict) else str(x)) for x in text)
|
||
obj = _parse_json_obj(str(text))
|
||
if not obj:
|
||
notes.append("视觉打分返回非 JSON，本轮记 0 分")
|
||
return {
|
||
"score": 0,
|
||
"pass": False,
|
||
"fails": ["打分解析失败"],
|
||
"passes": [],
|
||
"raw": str(text)[:800],
|
||
}, notes
|
||
raw_score = int(obj.get("score") or 0)
|
||
fails = [str(x) for x in (obj.get("fails") or []) if x]
|
||
passes = [str(x) for x in (obj.get("passes") or []) if x]
|
||
checks = obj.get("visual_checks") if isinstance(obj.get("visual_checks"), dict) else {}
|
||
zones = obj.get("zone_checklist") if isinstance(obj.get("zone_checklist"), dict) else {}
|
||
|
||
score, fails, ok = _enforce_score_caps(
|
||
raw_score,
|
||
fails,
|
||
checks,
|
||
zones,
|
||
has_real=has_real,
|
||
draft=draft,
|
||
prev_fails=prev_fails,
|
||
)
|
||
# 模型自称 pass 但服务端否决时记一条
|
||
if bool(obj.get("pass")) and not ok:
|
||
notes.append(
|
||
f"视觉模型自评达标被否决：原始 {raw_score} → 校正 {score}（fails={len(fails)}）"
|
||
)
|
||
notes.append(
|
||
f"视觉打分 {meta['label']}/{model_name}: {score}/100"
|
||
f"（原始 {raw_score}，真页面={'是' if has_real else '否'}）"
|
||
)
|
||
return {
|
||
"score": score,
|
||
"pass": ok,
|
||
"fails": fails,
|
||
"passes": passes,
|
||
"visual_checks": checks,
|
||
"zone_checklist": zones,
|
||
"raw_score": raw_score,
|
||
"patch_hints": obj.get("patch_hints") if isinstance(obj.get("patch_hints"), dict) else {},
|
||
}, notes
|
||
except Exception as e: # noqa: BLE001
|
||
return {"score": 0, "pass": False, "fails": [f"打分调用失败: {e}"], "passes": []}, [str(e)]
|
||
|
||
|
||
def _deep_merge_ui(dst: dict[str, Any], src: dict[str, Any]) -> dict[str, Any]:
|
||
out = dict(dst)
|
||
for k, v in src.items():
|
||
if v is None or v == "":
|
||
continue
|
||
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
||
out[k] = {**out[k], **v}
|
||
else:
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def apply_patch_hints(draft: dict[str, Any], hints: dict[str, Any]) -> dict[str, Any]:
|
||
"""把视觉 patch_hints 机械合并进 draft（再交给代码模型细修）。"""
|
||
if not hints:
|
||
return draft
|
||
meta = draft.setdefault("meta", {})
|
||
for key in ("name", "platform_title", "project_context", "ui_preset"):
|
||
if hints.get(f"meta.{key}"):
|
||
meta[key] = hints[f"meta.{key}"]
|
||
elif hints.get(key) and key in {"name", "platform_title", "project_context"}:
|
||
meta[key] = hints[key]
|
||
ui_hint = hints.get("meta.ui") or hints.get("ui")
|
||
if isinstance(ui_hint, dict):
|
||
meta["ui"] = _deep_merge_ui(meta.get("ui") or {}, ui_hint)
|
||
return draft
|
||
|
||
|
||
def refine_blueprint_against_fails(
|
||
draft: dict[str, Any],
|
||
fails: list[str],
|
||
vision_extract: str,
|
||
plan: str,
|
||
provider_id: str | None = None,
|
||
model: str | None = None,
|
||
) -> tuple[dict[str, Any], list[str]]:
|
||
"""代码模型按差异清单改蓝图。"""
|
||
instruction = (
|
||
"根据视觉验收 fails 修改 draft，目标还原参考截图。"
|
||
"只改 meta / meta.ui / pages 标题与 dashboard widgets 配置；"
|
||
"禁止改成无关行业模板；禁止删除 entities。"
|
||
"必须把 fails 中的原文写进对应字段。"
|
||
"返回完整 draft JSON。"
|
||
)
|
||
prompt = (
|
||
f"{instruction}\n\n[视觉摘录]\n{vision_extract}\n\n[界面计划]\n{plan}\n\n"
|
||
f"[fails]\n" + "\n".join(f"- {x}" for x in fails)
|
||
)
|
||
return enhance_blueprint_with_llm(draft, prompt, "fidelity_refine", provider_id, model)
|
||
|
||
|
||
def run_fidelity_loop(
|
||
draft: dict[str, Any],
|
||
ref_images: list[tuple[str, bytes, str]],
|
||
vision_extract: str,
|
||
*,
|
||
llm_provider: str | None = None,
|
||
llm_model: str | None = None,
|
||
vision_model: str | None = None,
|
||
harden_fn=None,
|
||
baseline: dict[str, Any] | None = None,
|
||
preview_rows: list[dict[str, Any]] | None = None,
|
||
) -> tuple[dict[str, Any], dict[str, Any], list[str]]:
|
||
"""
|
||
返回 (draft, report, notes)。
|
||
report: {target, rounds:[{score,fails,passes}], final_score, passed}
|
||
"""
|
||
notes: list[str] = []
|
||
log = glog()
|
||
if not fidelity_enabled():
|
||
if log:
|
||
log.stage("fidelity_loop", "已关闭 FIDELITY_LOOP=0")
|
||
return draft, {"skipped": True, "reason": "FIDELITY_LOOP=0"}, ["还原度迭代已关闭"]
|
||
if not ref_images:
|
||
if log:
|
||
log.stage("fidelity_loop", "无参考截图")
|
||
return draft, {"skipped": True, "reason": "no_images"}, ["无参考截图，跳过还原度迭代"]
|
||
|
||
target = fidelity_target()
|
||
max_rounds = fidelity_max_rounds()
|
||
rounds: list[dict[str, Any]] = []
|
||
current = draft
|
||
notes.append(f"开始还原度迭代：目标>={target}%，最多 {max_rounds} 轮")
|
||
if log:
|
||
log.notes(notes[-1:])
|
||
log.info(
|
||
"fidelity.begin",
|
||
target=target,
|
||
max_rounds=max_rounds,
|
||
ref_images=len(ref_images),
|
||
preview_rows=len(preview_rows or []),
|
||
)
|
||
|
||
for i in range(1, max_rounds + 1):
|
||
plan = ui_plan_from_draft(current)
|
||
preview, preview_note, is_real = _try_preview_screenshot(
|
||
current, preview_rows=preview_rows
|
||
)
|
||
notes.append(f"第{i}轮: {preview_note}")
|
||
if log:
|
||
log.stage(
|
||
f"fidelity_round_{i}",
|
||
preview_note,
|
||
round=i,
|
||
real_screen=is_real,
|
||
)
|
||
prev_fails = list(rounds[-1].get("fails") or []) if rounds else []
|
||
score_obj, score_notes = vision_score_fidelity(
|
||
ref_images,
|
||
plan,
|
||
vision_extract,
|
||
preview,
|
||
llm_provider,
|
||
vision_model, # 只用视觉模型，不用 deepseek-chat 等文本模型名
|
||
is_real_screen=is_real,
|
||
draft=current,
|
||
prev_fails=prev_fails,
|
||
)
|
||
notes.extend(score_notes)
|
||
if log:
|
||
log.notes(score_notes)
|
||
log.info(
|
||
f"fidelity.round_{i}.score",
|
||
round=i,
|
||
score=score_obj.get("score"),
|
||
raw_score=score_obj.get("raw_score"),
|
||
passed=bool(score_obj.get("pass")),
|
||
fails_n=len(score_obj.get("fails") or []),
|
||
)
|
||
rounds.append(
|
||
{
|
||
"round": i,
|
||
"score": score_obj.get("score"),
|
||
"raw_score": score_obj.get("raw_score"),
|
||
"pass": score_obj.get("pass"),
|
||
"fails": score_obj.get("fails") or [],
|
||
"passes": score_obj.get("passes") or [],
|
||
"zone_checklist": score_obj.get("zone_checklist") or {},
|
||
"real_screen": is_real,
|
||
}
|
||
)
|
||
if score_obj.get("pass"):
|
||
notes.append(f"第{i}轮达标：{score_obj.get('score')}% ≥ {target}%（已对照真页面）")
|
||
if log:
|
||
log.note(notes[-1])
|
||
break
|
||
|
||
fails = list(score_obj.get("fails") or [])
|
||
if not fails:
|
||
fails = [f"总分仅 {score_obj.get('score')}，未达 {target}，请按视觉摘录补全 meta.ui"]
|
||
if log:
|
||
log.info(
|
||
f"fidelity.round_{i}.patch",
|
||
round=i,
|
||
fails="; ".join(fails[:8]),
|
||
)
|
||
current = apply_patch_hints(current, score_obj.get("patch_hints") or {})
|
||
current, refine_notes = refine_blueprint_against_fails(
|
||
current, fails, vision_extract, plan, llm_provider, llm_model
|
||
)
|
||
notes.extend(refine_notes)
|
||
if log:
|
||
log.notes(refine_notes)
|
||
if harden_fn and baseline is not None:
|
||
current = harden_fn(current, baseline)
|
||
|
||
final = rounds[-1]["score"] if rounds else 0
|
||
report = {
|
||
"target": target,
|
||
"max_rounds": max_rounds,
|
||
"rounds": rounds,
|
||
"final_score": final,
|
||
"passed": bool(rounds and rounds[-1].get("pass")),
|
||
}
|
||
if not report["passed"]:
|
||
notes.append(
|
||
f"还原度迭代结束未达标：最终 {final}%（目标 {target}%），请根据 fails 继续生成或手工改蓝图"
|
||
)
|
||
if log:
|
||
log.warning(notes[-1], final_score=final, target=target)
|
||
elif log:
|
||
log.info("fidelity.passed", final_score=final, target=target)
|
||
return current, report, notes
|