120 lines
5.8 KiB
Python
120 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
"""全量 Visual Object Catalog 的原始 Style Layer 分类隔离审计，永不授权运行。"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
RESOLVED = frozenset({"LAYER_ONLY", "VISUAL_OBJECT_MATCH"})
|
|
|
|
|
|
def audit_isolation(style: dict[str, Any], mapping: dict[str, Any],
|
|
expected_count: int) -> dict[str, Any]:
|
|
layers = style.get("layers")
|
|
assignments = mapping.get("assignments")
|
|
if not isinstance(layers, list) or not isinstance(assignments, list) or not layers:
|
|
raise ValueError("Style/Mapping 输入结构无效")
|
|
if not isinstance(expected_count, int) or expected_count < 1:
|
|
raise ValueError("Full Catalog 预期对象数无效")
|
|
if len(assignments) != expected_count:
|
|
raise ValueError("Mapping 没有覆盖完整对象目录")
|
|
by_layer: dict[int, list[dict[str, Any]]] = defaultdict(list)
|
|
seen_objects: set[str] = set()
|
|
for row in assignments:
|
|
oid = row.get("visualObjectId")
|
|
if not isinstance(oid, str) or not oid or oid in seen_objects:
|
|
raise ValueError("Mapping Visual Object ID 重复/无效")
|
|
seen_objects.add(oid)
|
|
evidence = row.get("renderEvidence")
|
|
if not isinstance(evidence, dict) or not isinstance(evidence.get("components"), list) or not evidence["components"]:
|
|
raise ValueError(f"renderEvidence 缺失: {oid}")
|
|
observed_indices: set[int] = set()
|
|
for comp in evidence["components"]:
|
|
index = comp.get("styleIndex")
|
|
if type(index) is not int or index < 0 or index >= len(layers) or layers[index].get("id") != comp.get("styleLayerId"):
|
|
raise ValueError(f"Style index/layer identity mismatch: {oid}")
|
|
original = layers[index]
|
|
if original.get("source-layer") != evidence.get("sourceLayer"):
|
|
raise ValueError(f"Source Layer mismatch: {oid}")
|
|
observed_indices.add(index)
|
|
for index in observed_indices:
|
|
by_layer[index].append(row)
|
|
results: list[dict[str, Any]] = []
|
|
counts: Counter[str] = Counter()
|
|
for index, layer in enumerate(layers):
|
|
rows = by_layer.get(index, [])
|
|
statuses = Counter(row.get("status") for row in rows)
|
|
standards = Counter(row.get("standardLayer") for row in rows
|
|
if row.get("status") in RESOLVED and
|
|
isinstance(row.get("standardLayer"), str) and
|
|
row["standardLayer"])
|
|
unresolved = sum(row.get("status") not in RESOLVED or not isinstance(row.get("standardLayer"), str)
|
|
or not row["standardLayer"] for row in rows)
|
|
if not rows:
|
|
classification = "NO_OBSERVED_STYLE_FEATURES"
|
|
elif unresolved == len(rows):
|
|
classification = "ONLY_UNRESOLVED_OR_RESIDUAL"
|
|
elif unresolved > 0 or len(standards) > 1:
|
|
classification = "MIXED_OR_RESIDUAL"
|
|
elif len(standards) == 1:
|
|
classification = "ONE_STANDARD_IN_COMPLETE_CATALOG_CANDIDATE"
|
|
else:
|
|
classification = "NO_REVIEWED_STANDARD"
|
|
counts[classification] += 1
|
|
results.append({
|
|
"styleIndex": index, "styleLayerId": layer["id"],
|
|
"sourceId": layer.get("source"), "sourceLayer": layer.get("source-layer"),
|
|
"styleType": layer.get("type"), "observedVisualTypeCount": len(rows),
|
|
"assignedStandardCounts": dict(sorted(standards.items())),
|
|
"assignmentStatusCounts": dict(sorted(statuses.items())),
|
|
"unresolvedVisualTypeCount": unresolved,
|
|
"catalogClassification": classification,
|
|
"possibleWholeLayerVisibilityInFrozenSnapshot": classification == "ONE_STANDARD_IN_COMPLETE_CATALOG_CANDIDATE",
|
|
"productionWholeLayerVisibilityAuthorized": False,
|
|
"reasons": [
|
|
"原始 Style 可能含动态/未来数据；全量 Visual Object Catalog 并非完整 MapLibre GPU 结果",
|
|
"必须等待正式 standardLayer→displayGroup 审批及完整 Layer Selector 复核",
|
|
],
|
|
})
|
|
return {
|
|
"schemaVersion": "navsea-style-layer-isolation-audit/v1",
|
|
"scope": "FULL_RECORDED_VISUAL_TYPE_CATALOG_SNAPSHOT_NOT_RUNTIME_PROOF",
|
|
"expectedVisualTypeCount": expected_count,
|
|
"styleLayerCount": len(layers),
|
|
"layerCategoryCounts": dict(sorted(counts.items())),
|
|
"runtimeAuthorizationCount": 0,
|
|
"layers": results,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("--style", type=Path, required=True)
|
|
p.add_argument("--mapping", type=Path, required=True)
|
|
p.add_argument("--output", type=Path, required=True)
|
|
args = p.parse_args()
|
|
style_bytes = args.style.read_bytes()
|
|
mapping = json.loads(args.mapping.read_text(encoding="utf-8"))
|
|
source = mapping.get("renderEvidenceSource") or {}
|
|
if source.get("styleSha256") != hashlib.sha256(style_bytes).hexdigest():
|
|
raise ValueError("Mapping Style SHA-256 绑定不一致")
|
|
total = source.get("distinctVisualObjectCount")
|
|
out = audit_isolation(json.loads(style_bytes), mapping, total)
|
|
if args.output.exists():
|
|
raise ValueError("结果文件已存在：需要增加版本，不能覆盖历史审计")
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps({
|
|
"styleLayerCount": out["styleLayerCount"], "layerCategoryCounts": out["layerCategoryCounts"],
|
|
"runtimeAuthorizationCount": 0, "output": str(args.output),
|
|
}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|