rebuild hotspot tiles for full semantic audit

navsea_rebuild_hotspot_tiles.py

@@ -0,0 +1,277 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import shutil
+import subprocess
+from pathlib import Path
+
+import mercantile
+
+from build_semantic_delivery_assets import build_mapping, load_audit_rows, remap_tile
+
+
+REPO_ROOT = Path(__file__).resolve().parent
+SOURCE_TILE_ROOT = Path(
+    "/home/wwwroot/newpec/exported_auto/"
+    "tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles"
+)
+FULL_ROOT = Path("/home/wwwroot/pbf-delivery-full-20260415")
+SEMANTIC_ROOT = Path("/home/wwwroot/pbf-delivery-full-semantic-20260416")
+DEFAULT_AOI_JSON = REPO_ROOT / "tasks/pbf/NavSea_Full_AOI_Visual_Audit_2026-04-18.json"
+DEFAULT_OUTPUT_DIR = REPO_ROOT / "report/full_hotspot_rebuild_2026-04-18"
+DEFAULT_TMP_ROOT = Path("/tmp/navsea_hotspot_rebuild")
+FID_KEY = "thisMyWorld@2026"
+BUNDLE_ID = "navsea-core"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="重建全国固定 AOI 热点 tile，修复旧坏几何。")
+    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("--tmp-root", type=Path, default=DEFAULT_TMP_ROOT)
+    parser.add_argument("--zoom", action="append", type=int, help="只修指定 zoom，可重复传入")
+    return parser.parse_args()
+
+
+def aoi_bbox(center: list[float], radius_nm: float) -> tuple[float, float, float, float]:
+    lng, lat = center
+    radius_km = radius_nm * 1.852
+    lat_delta = radius_km / 111.32
+    lon_delta = radius_km / (111.32 * math.cos(math.radians(lat)))
+    return (
+        lng - lon_delta,
+        lat - lat_delta,
+        lng + lon_delta,
+        lat + lat_delta,
+    )
+
+
+def hardlink_or_copy(src: Path, dst: Path) -> None:
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    if dst.exists():
+        dst.unlink()
+    try:
+        dst.hardlink_to(src)
+    except OSError:
+        shutil.copy2(src, dst)
+
+
+def collect_tiles(aoi_json: Path, zoom_filter: set[int] | None) -> tuple[list[dict], list[str]]:
+    aois = json.loads(aoi_json.read_text(encoding="utf-8"))
+    tile_map: dict[str, dict] = {}
+    labels: list[str] = []
+    for aoi in aois:
+        labels.append(aoi["label"])
+        west, south, east, north = aoi_bbox(aoi["center"], aoi["radius_nm"])
+        for zoom in aoi["zoom_levels"]:
+            if zoom_filter and zoom not in zoom_filter:
+                continue
+            for tile in mercantile.tiles(west, south, east, north, [zoom]):
+                rel = Path(str(tile.z)) / str(tile.x) / f"{tile.y}.pbf"
+                tile_map[str(rel)] = {
+                    "z": tile.z,
+                    "x": tile.x,
+                    "y": tile.y,
+                    "relative_path": str(rel),
+                }
+    tiles = sorted(tile_map.values(), key=lambda item: (item["z"], item["x"], item["y"]))
+    return tiles, labels
+
+
+def prepare_reference_root(reference_root: Path, tiles: list[dict]) -> int:
+    if reference_root.exists():
+        shutil.rmtree(reference_root)
+    prepared = 0
+    for tile in tiles:
+        rel = Path(tile["relative_path"])
+        src = SOURCE_TILE_ROOT / rel
+        if not src.exists():
+            continue
+        dst = reference_root / rel
+        hardlink_or_copy(src, dst)
+        prepared += 1
+    return prepared
+
+
+def backup_existing_tiles(backup_root: Path, target_root: Path, tiles: list[dict]) -> int:
+    if backup_root.exists():
+        shutil.rmtree(backup_root)
+    backed_up = 0
+    for tile in tiles:
+        rel = Path(tile["relative_path"])
+        src = target_root / rel
+        if not src.exists():
+            continue
+        dst = backup_root / rel
+        hardlink_or_copy(src, dst)
+        backed_up += 1
+    return backed_up
+
+
+def run_full_rebuild(reference_root: Path, rebuild_root: Path, zoom_filter: set[int] | None) -> None:
+    if rebuild_root.exists():
+        shutil.rmtree(rebuild_root)
+    cmd = [
+        str(REPO_ROOT / ".venv/bin/python"),
+        str(REPO_ROOT / "navsea_tile_builder.py"),
+        "--reference-tile-root",
+        str(reference_root),
+        "--output",
+        str(rebuild_root),
+        "--workers",
+        "1",
+        "--strip-legacy-japanese-delivery",
+        "--release-minimal",
+        "--fid-key",
+        FID_KEY,
+        "--bundle-id",
+        BUNDLE_ID,
+    ]
+    if zoom_filter:
+        cmd.extend(["--zmin", str(min(zoom_filter)), "--zmax", str(max(zoom_filter))])
+    subprocess.run(cmd, check=True, cwd=REPO_ROOT)
+
+
+def deploy_full_tiles(rebuild_root: Path, target_root: Path, tiles: list[dict]) -> int:
+    deployed = 0
+    for tile in tiles:
+        rel = Path(tile["relative_path"])
+        src = rebuild_root / rel
+        if not src.exists():
+            continue
+        hardlink_or_copy(src, target_root / rel)
+        deployed += 1
+    return deployed
+
+
+def deploy_semantic_tiles(rebuild_root: Path, target_root: Path, tiles: list[dict]) -> dict[str, int]:
+    rows = load_audit_rows()
+    mapping = build_mapping(rows)
+    value_map = {
+        row.old_sprite_key: mapping[row.old_sprite_key]
+        for row in rows
+        if row.used_in_pbf > 0 and row.old_sprite_key in mapping
+    }
+    needles = [old.encode("utf-8") for old in value_map]
+
+    stats = {
+        "tiles_total": 0,
+        "tiles_written": 0,
+        "tiles_semantic_rewritten": 0,
+        "features_semantic_changed": 0,
+    }
+    for tile in tiles:
+        rel = Path(tile["relative_path"])
+        src = rebuild_root / rel
+        if not src.exists():
+            continue
+        dst = target_root / rel
+        changed, feature_count = remap_tile(src, dst, value_map, needles)
+        stats["tiles_total"] += 1
+        stats["tiles_written"] += 1
+        if changed:
+            stats["tiles_semantic_rewritten"] += 1
+            stats["features_semantic_changed"] += feature_count
+    return stats
+
+
+def write_report(output_dir: Path, summary: dict) -> tuple[Path, Path]:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    report_json = output_dir / "full_hotspot_rebuild.json"
+    report_md = output_dir / "full_hotspot_rebuild.md"
+    report_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+    lines = [
+        "# 全国热点 AOI 旧坏 Tile 重建修复",
+        "",
+        "## 范围",
+        "",
+        f"- AOI 配置：`{summary['aoi_json']}`",
+        f"- AOI 标签：`{', '.join(summary['aoi_labels'])}`",
+        f"- zoom：`{summary['zoom_filter'] or '按 AOI 原配置'}`",
+        f"- 参考源：`{summary['source_root']}`",
+        f"- full 目标：`{summary['full_root']}`",
+        f"- semantic 目标：`{summary['semantic_root']}`",
+        "",
+        "## 数量",
+        "",
+        f"- 热点 tile 数：`{summary['tile_count']}`",
+        f"- 参考 tile 准备：`{summary['prepared_reference_tiles']}`",
+        f"- full 备份：`{summary['full_backup_tiles']}`",
+        f"- semantic 备份：`{summary['semantic_backup_tiles']}`",
+        f"- full 部署：`{summary['full_deployed_tiles']}`",
+        f"- semantic 部署：`{summary['semantic_stats']['tiles_written']}`",
+        f"- semantic 重写：`{summary['semantic_stats']['tiles_semantic_rewritten']}`",
+        f"- semantic 改动 feature：`{summary['semantic_stats']['features_semantic_changed']}`",
+        "",
+        "## 备份目录",
+        "",
+        f"- full backup: `{summary['full_backup_root']}`",
+        f"- semantic backup: `{summary['semantic_backup_root']}`",
+        "",
+        "## 说明",
+        "",
+        "- 这轮不是单纯回写 extent，而是用当前 builder 对固定热点 tile 重新生成 full 包。",
+        "- semantic 包再以重建后的 full tile 为底稿，按当前语义 sprite 映射规则重写相关属性。",
+        "- 目标是先消除全国审计热点里的旧坏几何与大片空白问题。",
+        "",
+    ]
+    report_md.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return report_md, report_json
+
+
+def main() -> None:
+    args = parse_args()
+    zoom_filter = set(args.zoom or [])
+    tiles, aoi_labels = collect_tiles(args.aoi_json, zoom_filter or None)
+
+    tmp_root = args.tmp_root
+    reference_root = tmp_root / "reference_tiles"
+    rebuild_root = tmp_root / "rebuild_full"
+    full_backup_root = tmp_root / "backup_full"
+    semantic_backup_root = tmp_root / "backup_semantic"
+
+    prepared_reference_tiles = prepare_reference_root(reference_root, tiles)
+    full_backup_tiles = backup_existing_tiles(full_backup_root, FULL_ROOT, tiles)
+    semantic_backup_tiles = backup_existing_tiles(semantic_backup_root, SEMANTIC_ROOT, tiles)
+    run_full_rebuild(reference_root, rebuild_root, zoom_filter or None)
+    full_deployed_tiles = deploy_full_tiles(rebuild_root, FULL_ROOT, tiles)
+    semantic_stats = deploy_semantic_tiles(rebuild_root, SEMANTIC_ROOT, tiles)
+
+    summary = {
+        "aoi_json": str(args.aoi_json),
+        "aoi_labels": aoi_labels,
+        "zoom_filter": sorted(zoom_filter),
+        "source_root": str(SOURCE_TILE_ROOT),
+        "full_root": str(FULL_ROOT),
+        "semantic_root": str(SEMANTIC_ROOT),
+        "tile_count": len(tiles),
+        "prepared_reference_tiles": prepared_reference_tiles,
+        "full_backup_root": str(full_backup_root),
+        "semantic_backup_root": str(semantic_backup_root),
+        "full_backup_tiles": full_backup_tiles,
+        "semantic_backup_tiles": semantic_backup_tiles,
+        "full_deployed_tiles": full_deployed_tiles,
+        "semantic_stats": semantic_stats,
+    }
+    report_md, report_json = write_report(args.output_dir, summary)
+    print(
+        json.dumps(
+            {
+                "report_md": str(report_md),
+                "report_json": str(report_json),
+                "tile_count": len(tiles),
+                "full_deployed_tiles": full_deployed_tiles,
+                "semantic_stats": semantic_stats,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()