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