#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any from mapbox_vector_tile.Mapbox import vector_tile_pb2 DEFAULT_SOURCE_ROOT = Path( "/home/wwwroot/newpec/exported_auto/" "tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles" ) @dataclass class LayerGeometryStats: point_count: int = 0 min_x: int = 0 max_x: int = 0 min_y: int = 0 max_y: int = 0 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="审计并按需修复 NavSea PBF 的几何 / extent / native 风险。") parser.add_argument("--target-root", type=Path, required=True) parser.add_argument("--label", required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT) parser.add_argument("--apply-fix", action="store_true", help="若检测到 extent 与源瓦片不一致，则按源瓦片 extent 原地修复。") parser.add_argument("--sample-limit", type=int, default=30) parser.add_argument("--zoom", action="append", type=int, help="只审指定 zoom，可重复传入多次。") return parser.parse_args() def zigzag_decode(value: int) -> int: return (value >> 1) ^ (-(value & 1)) def decode_feature_geometry(geometry: list[int]) -> LayerGeometryStats: stats = LayerGeometryStats() cursor_x = 0 cursor_y = 0 initialized = False idx = 0 while idx < len(geometry): command = geometry[idx] idx += 1 command_id = command & 0x7 command_count = command >> 3 if command_id in (1, 2): # MoveTo, LineTo for _ in range(command_count): if idx + 1 >= len(geometry): raise ValueError("geometry command truncated") cursor_x += zigzag_decode(geometry[idx]) cursor_y += zigzag_decode(geometry[idx + 1]) idx += 2 if not initialized: stats.min_x = stats.max_x = cursor_x stats.min_y = stats.max_y = cursor_y initialized = True else: stats.min_x = min(stats.min_x, cursor_x) stats.max_x = max(stats.max_x, cursor_x) stats.min_y = min(stats.min_y, cursor_y) stats.max_y = max(stats.max_y, cursor_y) stats.point_count += 1 elif command_id == 7: # ClosePath continue else: raise ValueError(f"unsupported geometry command id={command_id}") return stats def merge_layer_stats(existing: LayerGeometryStats | None, incoming: LayerGeometryStats) -> LayerGeometryStats: if existing is None or existing.point_count == 0: return incoming if incoming.point_count == 0: return existing return LayerGeometryStats( point_count=existing.point_count + incoming.point_count, min_x=min(existing.min_x, incoming.min_x), max_x=max(existing.max_x, incoming.max_x), min_y=min(existing.min_y, incoming.min_y), max_y=max(existing.max_y, incoming.max_y), ) def analyze_tile(tile_path: Path) -> dict[str, Any]: tile = vector_tile_pb2.tile() tile.ParseFromString(tile_path.read_bytes()) layer_summaries: list[dict[str, Any]] = [] extent_values: set[int] = set() tile_outside_extent = False tile_max_overflow = 0 for layer in tile.layers: extent = int(layer.extent or 4096) extent_values.add(extent) aggregate: LayerGeometryStats | None = None for feature in layer.features: geom_stats = decode_feature_geometry(list(feature.geometry)) aggregate = merge_layer_stats(aggregate, geom_stats) if aggregate is None: aggregate = LayerGeometryStats() outside_extent = False overflow = 0 if aggregate.point_count: overflow = max( 0, -aggregate.min_x, -aggregate.min_y, aggregate.max_x - extent, aggregate.max_y - extent, ) outside_extent = overflow > 0 tile_outside_extent = tile_outside_extent or outside_extent tile_max_overflow = max(tile_max_overflow, overflow) layer_summaries.append( { "name": layer.name, "extent": extent, "point_count": aggregate.point_count, "min_x": aggregate.min_x, "max_x": aggregate.max_x, "min_y": aggregate.min_y, "max_y": aggregate.max_y, "outside_extent": outside_extent, "overflow": overflow, } ) return { "extent_values": sorted(extent_values), "outside_extent": tile_outside_extent, "max_overflow": tile_max_overflow, "layers": layer_summaries, } def desired_extent_from_source(source_path: Path) -> int | None: if not source_path.exists(): return None tile = vector_tile_pb2.tile() tile.ParseFromString(source_path.read_bytes()) extents = sorted({int(layer.extent or 4096) for layer in tile.layers}) if not extents: return None if len(extents) == 1: return extents[0] return max(extents) def repair_tile_extents(tile_path: Path, desired_extent: int) -> bool: tile = vector_tile_pb2.tile() raw = tile_path.read_bytes() tile.ParseFromString(raw) changed = False for layer in tile.layers: current = int(layer.extent or 4096) if current != desired_extent: layer.extent = desired_extent changed = True if changed: tile_path.write_bytes(tile.SerializeToString()) return changed def hardlink_copy_tree(src_root: Path, dst_root: Path) -> None: if dst_root.exists(): raise SystemExit(f"backup root already exists: {dst_root}") dst_root.mkdir(parents=True, exist_ok=True) os.system(f"cp -al {src_root}/. {dst_root}/") def write_report( output_dir: Path, label: str, target_root: Path, source_root: Path, summary: dict[str, Any], ) -> tuple[Path, Path]: output_dir.mkdir(parents=True, exist_ok=True) report_json = output_dir / "geometry_native_audit.json" report_md = output_dir / "geometry_native_audit.md" report_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") lines = [ f"# {label} 几何 / Extent / Native 风险审计", "", "## 范围", "", f"- target root: `{target_root}`", f"- source root: `{source_root}`", f"- tiles: `{summary['tile_count']}`", f"- repaired tiles: `{summary['repaired_tiles']}`", "", "## 总体结论", "", f"- 坐标超当前 extent 的瓦片数：`{summary['outside_extent_tiles']}`", f"- 与源瓦片 extent 不一致的瓦片数：`{summary['source_extent_mismatch_tiles']}`", f"- 可直接按源 extent 修复的瓦片数：`{summary['fixable_from_source_tiles']}`", "", "## Zoom 汇总", "", ] for zoom in sorted(summary["zoom_summary"], key=int): item = summary["zoom_summary"][zoom] lines.extend( [ f"### z{zoom}", "", f"- tiles: `{item['tile_count']}`", f"- outside extent: `{item['outside_extent_tiles']}`", f"- source mismatch: `{item['source_extent_mismatch_tiles']}`", f"- extent counts: `{item['extent_counts']}`", "", ] ) lines.extend(["## 代表问题样本", ""]) for item in summary["sample_tiles"]: lines.extend( [ f"### `{item['tile']}`", "", f"- extent values: `{item['extent_values']}`", f"- source extent: `{item['source_extent']}`", f"- outside extent: `{item['outside_extent']}`", f"- max overflow: `{item['max_overflow']}`", f"- fixable from source: `{item['fixable_from_source']}`", "", ] ) report_md.write_text("\n".join(lines) + "\n", encoding="utf-8") return report_md, report_json def main() -> None: args = parse_args() target_root = args.target_root source_root = args.source_root output_dir = args.output_dir if not target_root.exists(): raise SystemExit(f"target root not found: {target_root}") if not source_root.exists(): raise SystemExit(f"source root not found: {source_root}") zoom_summary: dict[str, dict[str, Any]] = defaultdict( lambda: { "tile_count": 0, "outside_extent_tiles": 0, "source_extent_mismatch_tiles": 0, "extent_counts": Counter(), } ) sample_tiles: list[dict[str, Any]] = [] repaired_tiles = 0 outside_extent_tiles = 0 source_extent_mismatch_tiles = 0 fixable_from_source_tiles = 0 tile_paths = sorted(target_root.glob("*/*/*.pbf")) if args.zoom: zoom_filter = {str(item) for item in args.zoom} tile_paths = [path for path in tile_paths if path.parent.parent.name in zoom_filter] for tile_path in tile_paths: rel = tile_path.relative_to(target_root) zoom = rel.parts[0] analysis = analyze_tile(tile_path) source_extent = desired_extent_from_source(source_root / rel) current_extent = max(analysis["extent_values"]) if analysis["extent_values"] else None mismatch = source_extent is not None and current_extent is not None and source_extent != current_extent fixable = bool( mismatch and source_extent is not None and analysis["outside_extent"] and analysis["max_overflow"] <= max(0, source_extent - (current_extent or 0)) ) zoom_item = zoom_summary[zoom] zoom_item["tile_count"] += 1 for extent in analysis["extent_values"]: zoom_item["extent_counts"][extent] += 1 if analysis["outside_extent"]: outside_extent_tiles += 1 zoom_item["outside_extent_tiles"] += 1 if mismatch: source_extent_mismatch_tiles += 1 zoom_item["source_extent_mismatch_tiles"] += 1 if fixable: fixable_from_source_tiles += 1 if args.apply_fix and fixable and source_extent is not None: if repair_tile_extents(tile_path, source_extent): repaired_tiles += 1 analysis = analyze_tile(tile_path) if len(sample_tiles) < args.sample_limit and (analysis["outside_extent"] or mismatch): sample_tiles.append( { "tile": str(rel), "extent_values": analysis["extent_values"], "source_extent": source_extent, "outside_extent": analysis["outside_extent"], "max_overflow": analysis["max_overflow"], "fixable_from_source": fixable, } ) normalized_zoom_summary = { zoom: { "tile_count": item["tile_count"], "outside_extent_tiles": item["outside_extent_tiles"], "source_extent_mismatch_tiles": item["source_extent_mismatch_tiles"], "extent_counts": dict(sorted(item["extent_counts"].items())), } for zoom, item in sorted(zoom_summary.items(), key=lambda kv: int(kv[0])) } summary = { "label": args.label, "target_root": str(target_root), "source_root": str(source_root), "tile_count": len(tile_paths), "outside_extent_tiles": outside_extent_tiles, "source_extent_mismatch_tiles": source_extent_mismatch_tiles, "fixable_from_source_tiles": fixable_from_source_tiles, "repaired_tiles": repaired_tiles, "zoom_summary": normalized_zoom_summary, "sample_tiles": sample_tiles, } report_md, report_json = write_report(output_dir, args.label, target_root, source_root, summary) print( json.dumps( { "report_md": str(report_md), "report_json": str(report_json), "outside_extent_tiles": outside_extent_tiles, "source_extent_mismatch_tiles": source_extent_mismatch_tiles, "fixable_from_source_tiles": fixable_from_source_tiles, "repaired_tiles": repaired_tiles, }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()