278 lines
9.8 KiB
Python
278 lines
9.8 KiB
Python
#!/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()
|