370 lines
16 KiB
Python
370 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the reusable preflight for an AI-design page restoration.
|
|
|
|
The preflight is intentionally useful before Runtime exists. It validates the
|
|
page package, clean-master geometry, engine-owned text slots, hit areas and
|
|
layer order. When captures are configured it also invokes the repository's
|
|
rendered-pixel fidelity checker for each viewport and emits one responsive
|
|
contact sheet plus a compact JSON report.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from PIL import Image, ImageDraw
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIDELITY_TOOL = ROOT / "tools" / "verify_visual_fidelity.py"
|
|
|
|
|
|
@dataclass
|
|
class Check:
|
|
check_id: str
|
|
passed: bool
|
|
detail: str
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {"id": self.check_id, "pass": self.passed, "detail": self.detail}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
parser.add_argument(
|
|
"--skip-fidelity",
|
|
action="store_true",
|
|
help="run package checks only; useful before Runtime captures exist",
|
|
)
|
|
parser.add_argument("--allow-fail", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def resolve(config_path: Path, value: str | Path) -> Path:
|
|
path = Path(value)
|
|
if not path.is_absolute():
|
|
path = config_path.parent / path
|
|
return path.resolve()
|
|
|
|
|
|
def load_json(path: Path) -> tuple[dict[str, Any] | list[Any] | None, str | None]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception as exc: # pragma: no cover - exact parser errors vary by Python
|
|
return None, f"{type(exc).__name__}: {exc}"
|
|
if not isinstance(value, (dict, list)):
|
|
return None, "top-level JSON value must be an object or array"
|
|
return value, None
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def normalized_rect(value: Any, format_name: str = "xywh") -> bool:
|
|
return (
|
|
isinstance(value, list)
|
|
and len(value) == 4
|
|
and all(isinstance(item, (int, float)) for item in value)
|
|
and all(0 <= float(item) <= 1 for item in value)
|
|
and (
|
|
(format_name == "xyxy" and value[2] > value[0] and value[3] > value[1])
|
|
or (format_name == "xywh" and value[2] > 0 and value[3] > 0 and value[0] + value[2] <= 1.000001 and value[1] + value[3] <= 1.000001)
|
|
)
|
|
)
|
|
|
|
|
|
def normalized_anchor(value: Any) -> bool:
|
|
"""Backward-compatible xywh validator used by callers/tests."""
|
|
return normalized_rect(value, "xywh")
|
|
|
|
|
|
def normalized_size(value: list[float], format_name: str) -> tuple[float, float]:
|
|
return (value[2] - value[0], value[3] - value[1]) if format_name == "xyxy" else (value[2], value[3])
|
|
|
|
|
|
def viewport_canvas(viewport: str, aspect: float) -> tuple[int, int]:
|
|
width_text, height_text = viewport.lower().split("x", 1)
|
|
width, height = int(width_text), int(height_text)
|
|
current = width / height
|
|
if current >= aspect:
|
|
return round(height * aspect), height
|
|
return width, round(width / aspect)
|
|
|
|
|
|
def check_required_files(
|
|
config_path: Path, config: dict[str, Any], checks: list[Check]
|
|
) -> dict[str, Path]:
|
|
resolved: dict[str, Path] = {}
|
|
paths: list[tuple[str, str]] = []
|
|
for key in (
|
|
"approved_master",
|
|
"clean_static_master",
|
|
"visual_fidelity_contract",
|
|
"manifest",
|
|
"layer_map",
|
|
"text_slots",
|
|
"hit_areas",
|
|
):
|
|
value = config.get(key)
|
|
if value:
|
|
paths.append((key, str(value)))
|
|
for item in config.get("required_json", []):
|
|
if isinstance(item, str):
|
|
paths.append((item, item))
|
|
elif isinstance(item, dict) and item.get("path"):
|
|
paths.append((str(item.get("id", item["path"])), str(item["path"])))
|
|
for key, value in paths:
|
|
path = resolve(config_path, value)
|
|
resolved[key] = path
|
|
exists = path.is_file()
|
|
checks.append(Check(f"file:{key}", exists, str(path)))
|
|
if exists and path.suffix.lower() == ".json":
|
|
_, error = load_json(path)
|
|
checks.append(Check(f"json:{key}", error is None, error or "valid JSON"))
|
|
return resolved
|
|
|
|
|
|
def check_images(
|
|
config_path: Path,
|
|
config: dict[str, Any],
|
|
resolved: dict[str, Path],
|
|
checks: list[Check],
|
|
) -> tuple[Image.Image | None, Image.Image | None]:
|
|
approved_path = resolved.get("approved_master")
|
|
static_path = resolved.get("clean_static_master")
|
|
if not approved_path or not static_path or not approved_path.is_file() or not static_path.is_file():
|
|
return None, None
|
|
try:
|
|
approved = Image.open(approved_path).convert("RGB")
|
|
static = Image.open(static_path).convert("RGB")
|
|
except Exception as exc:
|
|
checks.append(Check("images:open", False, str(exc)))
|
|
return None, None
|
|
expected_size = config.get("design_size")
|
|
if expected_size:
|
|
size_pass = list(approved.size) == list(expected_size) and list(static.size) == list(expected_size)
|
|
checks.append(Check("images:design_size", size_pass, f"approved={approved.size} static={static.size} expected={expected_size}"))
|
|
else:
|
|
checks.append(Check("images:matching_size", approved.size == static.size, f"approved={approved.size} static={static.size}"))
|
|
aspect = float(config.get("design_aspect", approved.width / approved.height))
|
|
actual_aspect = approved.width / approved.height
|
|
checks.append(Check("images:aspect", abs(actual_aspect - aspect) <= 0.01, f"actual={actual_aspect:.6f} expected={aspect:.6f}"))
|
|
fidelity_contract = resolved.get("visual_fidelity_contract")
|
|
if fidelity_contract and fidelity_contract.is_file():
|
|
contract, error = load_json(fidelity_contract)
|
|
if isinstance(contract, dict):
|
|
master_meta = contract.get("clean_static_master", {})
|
|
expected_hash = master_meta.get("sha256") if isinstance(master_meta, dict) else None
|
|
actual_hash = sha256(static_path)
|
|
checks.append(Check("images:clean_master_hash", bool(expected_hash) and actual_hash == expected_hash, f"actual={actual_hash} expected={expected_hash}"))
|
|
elif error:
|
|
checks.append(Check("images:fidelity_contract", False, error))
|
|
return approved, static
|
|
|
|
|
|
def check_text_slots(config_path: Path, path: Path | None, config: dict[str, Any], checks: list[Check]) -> None:
|
|
if not path or not path.is_file():
|
|
return
|
|
data, _ = load_json(path)
|
|
slots = data.get("slots") if isinstance(data, dict) else None
|
|
valid = isinstance(slots, list) and bool(slots)
|
|
errors: list[str] = []
|
|
anchor_format = str(config.get("text_anchor_format", "xywh"))
|
|
if valid:
|
|
for slot in slots:
|
|
if not isinstance(slot, dict) or slot.get("owner") != "engine":
|
|
errors.append(str(slot.get("id", "unknown")) if isinstance(slot, dict) else "invalid slot")
|
|
continue
|
|
if not normalized_rect(slot.get("anchor"), anchor_format):
|
|
errors.append(f"{slot.get('id', 'unknown')}:anchor")
|
|
required_scales = set(config.get("blocking_text_scales", [100, 130]))
|
|
if not required_scales.issubset(set(slot.get("scale", []))):
|
|
errors.append(f"{slot.get('id', 'unknown')}:scale")
|
|
if not {"zh_CN", "en"}.issubset(set(slot.get("locales", []))):
|
|
errors.append(f"{slot.get('id', 'unknown')}:locale")
|
|
checks.append(Check("text_slots:engine_owned", valid and not errors, "ok" if not errors else ", ".join(errors)))
|
|
if isinstance(data, dict):
|
|
policy = str(data.get("policy", ""))
|
|
checks.append(Check("text_slots:no_baked_copy", "generated PNG" in policy or "engine" in policy.lower(), policy or "missing policy"))
|
|
|
|
|
|
def check_hit_areas(config_path: Path, path: Path | None, config: dict[str, Any], checks: list[Check]) -> None:
|
|
if not path or not path.is_file():
|
|
return
|
|
data, _ = load_json(path)
|
|
areas = data.get("areas") if isinstance(data, dict) else None
|
|
minimum = int(config.get("minimum_target_px", data.get("minimum_target_px", 48) if isinstance(data, dict) else 48))
|
|
anchor_format = str(config.get("hit_area_format", "xywh"))
|
|
errors: list[str] = []
|
|
if not isinstance(areas, list) or not areas:
|
|
errors.append("missing areas")
|
|
else:
|
|
for area in areas:
|
|
if not isinstance(area, dict) or not normalized_rect(area.get("anchor"), anchor_format):
|
|
errors.append(str(area.get("id", "unknown")) if isinstance(area, dict) else "invalid")
|
|
continue
|
|
anchor = area["anchor"]
|
|
anchor_width, anchor_height = normalized_size(anchor, anchor_format)
|
|
if anchor_width <= 0 or anchor_height <= 0:
|
|
errors.append(str(area.get("id", "unknown")))
|
|
checks.append(Check("hit_areas:normalized", not errors, f"format={anchor_format}" if not errors else ", ".join(errors)))
|
|
evidence_value = config.get("input_verification")
|
|
if evidence_value:
|
|
evidence_path = resolve(config_path, str(evidence_value))
|
|
evidence, error = load_json(evidence_path) if evidence_path.is_file() else (None, "file missing")
|
|
evidence_pass = (
|
|
isinstance(evidence, dict)
|
|
and evidence.get("result") == "PASS"
|
|
and int(evidence.get("minimum_target_px", 0)) >= minimum
|
|
and bool(evidence.get("target_check"))
|
|
)
|
|
checks.append(Check("hit_areas:runtime_target_evidence", evidence_pass, str(evidence_path) if evidence_pass else (error or "runtime target evidence is incomplete")))
|
|
else:
|
|
checks.append(Check("hit_areas:runtime_target_evidence", True, "skipped; no Runtime input evidence configured"))
|
|
if isinstance(data, dict) and isinstance(data.get("focus_order"), list):
|
|
ids = [area.get("id") for area in data.get("areas", []) if isinstance(area, dict)]
|
|
focus = data["focus_order"]
|
|
checks.append(Check("hit_areas:focus_order", len(focus) == len(set(focus)) and set(focus).issubset(set(ids)), "ok" if len(focus) == len(set(focus)) and set(focus).issubset(set(ids)) else "unknown or duplicate focus target"))
|
|
|
|
|
|
def check_layers(path: Path | None, config: dict[str, Any], checks: list[Check]) -> None:
|
|
if not path or not path.is_file():
|
|
return
|
|
data, _ = load_json(path)
|
|
layers = data.get("layers") if isinstance(data, dict) else None
|
|
errors: list[str] = []
|
|
bounds_format = str(config.get("layer_bounds_format", "xywh"))
|
|
if not isinstance(layers, list) or not layers:
|
|
errors.append("missing layers")
|
|
else:
|
|
z_values = []
|
|
for layer in layers:
|
|
if not isinstance(layer, dict) or not normalized_rect(layer.get("bounds"), bounds_format):
|
|
errors.append(str(layer.get("id", "unknown")) if isinstance(layer, dict) else "invalid")
|
|
else:
|
|
z_values.append(layer.get("z"))
|
|
if z_values != sorted(z_values) or len(z_values) != len(set(z_values)):
|
|
errors.append("z-order must be strictly increasing")
|
|
checks.append(Check("layers:normalized_and_ordered", not errors, "ok" if not errors else ", ".join(errors)))
|
|
|
|
|
|
def make_capture_sheet(captures: list[tuple[str, Path]], output: Path) -> Path | None:
|
|
opened: list[tuple[str, Image.Image]] = []
|
|
for label, path in captures:
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
opened.append((label, Image.open(path).convert("RGB")))
|
|
except Exception:
|
|
continue
|
|
if not opened:
|
|
return None
|
|
thumb_width = 560
|
|
tile_height = 350
|
|
sheet = Image.new("RGB", (thumb_width * 2, tile_height * ((len(opened) + 1) // 2)), "#17252c")
|
|
draw = ImageDraw.Draw(sheet)
|
|
for index, (label, image) in enumerate(opened):
|
|
x = (index % 2) * thumb_width
|
|
y = (index // 2) * tile_height
|
|
image.thumbnail((thumb_width - 20, tile_height - 52), Image.Resampling.LANCZOS)
|
|
sheet.paste(image, (x + 10, y + 34))
|
|
draw.text((x + 12, y + 10), label, fill="#f5e6b8")
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
sheet.save(output)
|
|
return output
|
|
|
|
|
|
def run_fidelity(
|
|
config_path: Path,
|
|
config: dict[str, Any],
|
|
resolved: dict[str, Path],
|
|
output: Path,
|
|
checks: list[Check],
|
|
) -> list[dict[str, Any]]:
|
|
captures = config.get("captures", [])
|
|
if not captures:
|
|
checks.append(Check("fidelity:captures", True, "skipped; no captures configured"))
|
|
return []
|
|
if not resolved.get("approved_master") or not resolved.get("visual_fidelity_contract"):
|
|
checks.append(Check("fidelity:configuration", False, "approved_master and visual_fidelity_contract are required"))
|
|
return []
|
|
reports: list[dict[str, Any]] = []
|
|
sheet_inputs: list[tuple[str, Path]] = []
|
|
for capture in captures:
|
|
if not isinstance(capture, dict) or not capture.get("path"):
|
|
checks.append(Check("fidelity:capture_config", False, str(capture)))
|
|
continue
|
|
label = str(capture.get("label", Path(str(capture["path"])).stem))
|
|
runtime = resolve(config_path, str(capture["path"]))
|
|
sheet_inputs.append((label, runtime))
|
|
capture_output = output / "visual-fidelity" / label
|
|
command = [
|
|
sys.executable,
|
|
str(FIDELITY_TOOL),
|
|
"--approved", str(resolved["approved_master"]),
|
|
"--runtime", str(runtime),
|
|
"--contract", str(resolved["visual_fidelity_contract"]),
|
|
"--output", str(capture_output),
|
|
"--allow-fail",
|
|
]
|
|
completed = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
report_path = capture_output / "report.json"
|
|
report, error = load_json(report_path) if report_path.is_file() else (None, "report missing")
|
|
passed = completed.returncode == 0 and isinstance(report, dict) and report.get("result") == "PASS"
|
|
checks.append(Check(f"fidelity:{label}", passed, str(report.get("result") if isinstance(report, dict) else error)))
|
|
if isinstance(report, dict):
|
|
reports.append({"label": label, "report": str(report_path), "result": report.get("result"), "checks": report.get("checks", [])})
|
|
sheet = make_capture_sheet(sheet_inputs, output / "responsive-side-by-side.png")
|
|
checks.append(Check("fidelity:responsive_sheet", sheet is not None, str(sheet) if sheet else "no readable captures"))
|
|
return reports
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
config_path = args.config.resolve()
|
|
output = args.output.resolve()
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
config, error = load_json(config_path)
|
|
if not isinstance(config, dict):
|
|
raise SystemExit(f"invalid config: {error or 'top-level object required'}")
|
|
checks: list[Check] = []
|
|
resolved = check_required_files(config_path, config, checks)
|
|
check_images(config_path, config, resolved, checks)
|
|
check_text_slots(config_path, resolved.get("text_slots"), config, checks)
|
|
check_hit_areas(config_path, resolved.get("hit_areas"), config, checks)
|
|
check_layers(resolved.get("layer_map"), config, checks)
|
|
fidelity_reports = [] if args.skip_fidelity else run_fidelity(config_path, config, resolved, output, checks)
|
|
passed = all(check.passed for check in checks)
|
|
report = {
|
|
"schema_version": 1,
|
|
"page_id": config.get("page_id"),
|
|
"config": str(config_path),
|
|
"result": "PASS" if passed else "FAIL_PAGE_PREFLIGHT",
|
|
"checks": [check.as_dict() for check in checks],
|
|
"fidelity_reports": fidelity_reports,
|
|
"artifacts": ["report.json"] + (["responsive-side-by-side.png"] if (output / "responsive-side-by-side.png").is_file() else []),
|
|
}
|
|
(output / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
passed_count = sum(1 for check in checks if check.passed)
|
|
marker = "ABG_PAGE_PREFLIGHT_OK" if passed else "ABG_PAGE_PREFLIGHT_FAILED"
|
|
print(f"{marker} page={config.get('page_id')} checks={passed_count}/{len(checks)} report={output / 'report.json'}")
|
|
return 0 if passed or args.allow_fail else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|