#!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess from pathlib import Path from typing import Any from urllib.parse import urlencode from PIL import Image, ImageChops, ImageStat DEFAULT_COMPARE_URL = "http://192.168.200.184/newpec/navsea-compare-full-audit.html" DEFAULT_AOI_JSON = Path("/root/sourceserver/pbf/tasks/pbf/NavSea_Full_AOI_Visual_Audit_2026-04-18.json") DEFAULT_OUTPUT_DIR = Path("/root/sourceserver/pbf/report/full_aoi_visual_audit_2026-04-18") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="对全国 full / semantic 做 AOI 截图 + 图像 diff 审计。") parser.add_argument("--compare-url", default=DEFAULT_COMPARE_URL) parser.add_argument("--aoi-json", type=Path, default=DEFAULT_AOI_JSON) parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) parser.add_argument("--width", type=int, default=1280) parser.add_argument("--height", type=int, default=720) parser.add_argument("--toolbar-height", type=int, default=120) parser.add_argument("--bottom-trim", type=int, default=30) parser.add_argument("--timeout-sec", type=int, default=120) parser.add_argument("--virtual-time-budget-ms", type=int, default=30000) parser.add_argument("--variant", action="append", choices=("full", "semantic"), help="只跑指定 variant，可重复传入") return parser.parse_args() def run_browser_screenshot( compare_url: str, output_png: Path, width: int, height: int, timeout_sec: int, virtual_time_budget_ms: int, ) -> None: cmd = [ "timeout", f"{timeout_sec}s", "google-chrome", "--headless=new", "--disable-gpu", "--enable-unsafe-swiftshader", "--no-sandbox", f"--virtual-time-budget={virtual_time_budget_ms}", f"--window-size={width},{height}", f"--screenshot={output_png}", compare_url, ] subprocess.run(cmd, check=True, capture_output=True, text=True) def compute_visual_metrics(left_img: Image.Image, right_img: Image.Image) -> dict[str, Any]: diff = ImageChops.difference(left_img, right_img) diff_rgb = diff.convert("RGB") diff_gray = diff.convert("L") stat = ImageStat.Stat(diff_rgb) mean_abs = [round(v, 4) for v in stat.mean] rms = [round(v, 4) for v in stat.rms] bbox = diff_gray.getbbox() nonzero_pixels = 0 if bbox: mask = diff_gray.point(lambda value: 255 if value else 0) histogram = mask.histogram() nonzero_pixels = histogram[255] if len(histogram) > 255 else 0 total_pixels = left_img.size[0] * left_img.size[1] changed_ratio = round(nonzero_pixels / total_pixels, 6) if total_pixels else 0.0 return { "changed_pixels": nonzero_pixels, "total_pixels": total_pixels, "changed_ratio": changed_ratio, "mean_abs_rgb": mean_abs, "rms_rgb": rms, "diff_bbox": list(bbox) if bbox else None, "diff_image": diff, } def crop_compare_panes(full_png: Path, left_png: Path, right_png: Path, diff_png: Path, toolbar_height: int, bottom_trim: int) -> dict[str, Any]: image = Image.open(full_png).convert("RGBA") usable_top = toolbar_height usable_bottom = max(usable_top + 1, image.height - bottom_trim) mid_x = image.width // 2 left = image.crop((0, usable_top, mid_x, usable_bottom)) right = image.crop((mid_x, usable_top, image.width, usable_bottom)) left.save(left_png) right.save(right_png) metrics = compute_visual_metrics(left, right) diff_image = metrics.pop("diff_image") diff_image.convert("RGB").point(lambda value: min(255, value * 4)).save(diff_png) metrics["left_crop"] = [0, usable_top, mid_x, usable_bottom] metrics["right_crop"] = [mid_x, usable_top, image.width, usable_bottom] return metrics def build_case_url(compare_url: str, variant: str, center: list[float], zoom: int | float) -> str: query = urlencode( { "variant": variant, "audit": "1", "center": f"{center[0]},{center[1]}", "zoom": zoom, } ) separator = "&" if "?" in compare_url else "?" return f"{compare_url}{separator}{query}" def write_report(output_dir: Path, summary: dict[str, Any]) -> tuple[Path, Path]: report_json = output_dir / "full_aoi_visual_audit.json" report_md = output_dir / "full_aoi_visual_audit.md" report_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") lines = [ "# 全国 AOI 截图对比审计", "", "## 总览", "", f"- compare page: `{summary['compare_url']}`", f"- cases: `{len(summary['cases'])}`", "", "## 差异排序", "", ] for case in summary["cases"]: lines.extend( [ f"### {case['label']} · {case['variant']} · z{case['zoom']}", "", f"- changed ratio: `{case['changed_ratio']}`", f"- changed pixels: `{case['changed_pixels']}` / `{case['total_pixels']}`", f"- mean abs rgb: `{case['mean_abs_rgb']}`", f"- compare url: `{case['compare_url']}`", f"- left image: `{case['left_image']}`", f"- right image: `{case['right_image']}`", f"- diff image: `{case['diff_image']}`", "", ] ) report_md.write_text("\n".join(lines) + "\n", encoding="utf-8") return report_md, report_json def main() -> None: args = parse_args() output_dir = args.output_dir output_dir.mkdir(parents=True, exist_ok=True) aois = json.loads(args.aoi_json.read_text(encoding="utf-8")) cases: list[dict[str, Any]] = [] variants = args.variant or ["full", "semantic"] for variant in variants: for aoi in aois: for zoom in aoi["zoom_levels"]: case_id = f"{aoi['id']}_{variant}_z{zoom}" case_dir = output_dir / case_id case_dir.mkdir(parents=True, exist_ok=True) compare_url = build_case_url(args.compare_url, variant, aoi["center"], zoom) full_png = case_dir / "compare_full.png" left_png = case_dir / "compare_left.png" right_png = case_dir / "compare_right.png" diff_png = case_dir / "compare_diff.png" run_browser_screenshot( compare_url=compare_url, output_png=full_png, width=args.width, height=args.height, timeout_sec=args.timeout_sec, virtual_time_budget_ms=args.virtual_time_budget_ms, ) metrics = crop_compare_panes( full_png=full_png, left_png=left_png, right_png=right_png, diff_png=diff_png, toolbar_height=args.toolbar_height, bottom_trim=args.bottom_trim, ) cases.append( { "id": case_id, "label": aoi["label"], "variant": variant, "zoom": zoom, "center": aoi["center"], "radius_nm": aoi["radius_nm"], "compare_url": compare_url, "changed_pixels": metrics["changed_pixels"], "total_pixels": metrics["total_pixels"], "changed_ratio": metrics["changed_ratio"], "mean_abs_rgb": metrics["mean_abs_rgb"], "rms_rgb": metrics["rms_rgb"], "diff_bbox": metrics["diff_bbox"], "left_image": str(left_png), "right_image": str(right_png), "diff_image": str(diff_png), } ) cases.sort(key=lambda item: item["changed_ratio"], reverse=True) summary = { "compare_url": args.compare_url, "cases": cases, } report_md, report_json = write_report(output_dir, summary) print( json.dumps( { "report_md": str(report_md), "report_json": str(report_json), "top_cases": cases[:5], }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()