Files
pbf/navsea_extent_shift_audit.py
2026-04-18 14:26:54 +08:00

83 lines
2.5 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
from navsea_tile_reencode_guard import validate_reencoded_tile
DEFAULT_SOURCE_ROOT = Path(
"/home/wwwroot/newpec/exported_auto/"
"tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles"
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="审计 PBF 是否存在高 extent 被 4096 口径重编码的几何错移。")
parser.add_argument("--target-root", type=Path, required=True)
parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT)
parser.add_argument("--output-json", type=Path)
parser.add_argument("--zoom", action="append", type=int)
parser.add_argument("--sample-limit", type=int, default=50)
return parser.parse_args()
def should_include(rel: Path, zoom_filter: set[int] | None) -> bool:
if not zoom_filter:
return True
try:
return int(rel.parts[0]) in zoom_filter
except Exception:
return False
def main() -> None:
args = parse_args()
zoom_filter = set(args.zoom or [])
suspicious_tiles: list[dict[str, Any]] = []
checked = 0
for tile_path in sorted(args.target_root.glob("*/*/*.pbf")):
rel = tile_path.relative_to(args.target_root)
if not should_include(rel, zoom_filter or None):
continue
source_tile = args.source_root / rel
if not source_tile.exists():
continue
checked += 1
verdict = validate_reencoded_tile(source_tile, tile_path)
if verdict["suspicious"]:
suspicious_tiles.append(
{
"tile": str(rel),
**verdict,
}
)
suspicious_tiles.sort(
key=lambda item: (
item["output_max_overflow"] - item["source_max_overflow"],
item["tile"],
),
reverse=True,
)
summary = {
"target_root": str(args.target_root),
"source_root": str(args.source_root),
"zoom_filter": sorted(zoom_filter),
"tiles_checked": checked,
"suspicious_tile_count": len(suspicious_tiles),
"samples": suspicious_tiles[: args.sample_limit],
}
if args.output_json:
args.output_json.parent.mkdir(parents=True, exist_ok=True)
args.output_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps(summary, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()