612 lines
21 KiB
Python
612 lines
21 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
r"""
|
||
CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1
|
||
|
||
目的
|
||
====
|
||
在已经完成 production 首轮生成后，对 Dataset 全量 chart_feature 与当前
|
||
cm95_to_navsea_v1.yaml 做只读对照，找出“当前 delivery 不会输出”的真实
|
||
object_class + geometry_type 组合。
|
||
|
||
这一步专门用于 production mapping 收口，不修改数据库、不生成 PBF。
|
||
|
||
默认输入：
|
||
dataset_id = 1
|
||
mapping = config/mappings/cm95_to_navsea_v1.yaml
|
||
|
||
默认输出：
|
||
build/production/cm95_global_mapping_gap_audit_v1.json
|
||
build/production/cm95_global_mapping_gap_variants_v1.csv
|
||
build/production/cm95_global_mapping_gap_samples_v1.jsonl
|
||
build/production/cm95_global_mapping_gap_summary_v1.txt
|
||
|
||
审计内容：
|
||
- 全 Dataset object_class + geometry_type + native_scale 数量；
|
||
- 当前 YAML 是否存在 exact rule；
|
||
- 同 object_class 的其它 geometry rule；
|
||
- 已知 hard_gate 是否出现未覆盖 geometry variant；
|
||
- 显式 ENGINEERING_ONLY / cm95_unmapped；
|
||
- 每个 gap variant 的属性键频率、样本值、relation20 覆盖；
|
||
- 每个 gap variant 的代表 Feature 样本。
|
||
|
||
数据库连接直接读取项目 .env：
|
||
CM93_DB_HOST
|
||
CM93_DB_PORT
|
||
CM93_DB_NAME
|
||
CM93_DB_USER
|
||
CM93_DB_PASSWORD
|
||
|
||
执行：
|
||
& ".\.venv\Scripts\python.exe" .\tools\cm95_global_mapping_gap_audit_v1.py `
|
||
--dataset-id 1 `
|
||
--mapping .\config\mappings\cm95_to_navsea_v1.yaml
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pymysql
|
||
import yaml
|
||
|
||
from cm93db.config import load_config
|
||
|
||
|
||
DELIVERY_MAIN_STATUSES = {
|
||
"MAPPED",
|
||
"MAPPED_CONDITIONAL",
|
||
"COMPOSITE",
|
||
"MAPPED_DERIVED_POINTS",
|
||
}
|
||
DELIVERY_DERIVED_STATUSES = {
|
||
"MAPPED",
|
||
"MAPPED_CONDITIONAL",
|
||
"DERIVED_LABEL_ONLY",
|
||
}
|
||
|
||
|
||
def parse_args():
|
||
p = argparse.ArgumentParser(
|
||
description="CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1"
|
||
)
|
||
p.add_argument("--dataset-id", type=int, default=1)
|
||
p.add_argument(
|
||
"--mapping",
|
||
default=r"config\mappings\cm95_to_navsea_v1.yaml",
|
||
)
|
||
p.add_argument(
|
||
"--output",
|
||
default=r"build\production\cm95_global_mapping_gap_audit_v1.json",
|
||
)
|
||
p.add_argument("--sample-per-variant", type=int, default=6)
|
||
p.add_argument("--attribute-value-limit", type=int, default=12)
|
||
return p.parse_args()
|
||
|
||
|
||
def json_dict(value: Any) -> dict[str, Any]:
|
||
if value is None:
|
||
return {}
|
||
if isinstance(value, dict):
|
||
return value
|
||
if isinstance(value, (bytes, bytearray, memoryview)):
|
||
value = bytes(value).decode("utf-8", errors="replace")
|
||
try:
|
||
obj = json.loads(str(value))
|
||
return obj if isinstance(obj, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def scalar_text(value: Any) -> str:
|
||
if isinstance(value, (dict, list)):
|
||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
return str(value)
|
||
|
||
|
||
def load_rules(path: Path):
|
||
obj = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
rules = obj.get("rules") or obj.get("mappings") or []
|
||
if not isinstance(rules, list):
|
||
raise RuntimeError("mapping YAML 中找不到 rules/mappings list")
|
||
|
||
exact: dict[tuple[str, str], dict[str, Any]] = {}
|
||
by_class: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||
|
||
for rule in rules:
|
||
if not isinstance(rule, dict):
|
||
continue
|
||
cls = str(rule.get("cm95_object_class") or "").strip()
|
||
geom = str(rule.get("geometry_type") or "").strip().upper()
|
||
if not cls or not geom:
|
||
continue
|
||
exact[(cls, geom)] = rule
|
||
by_class[cls].append(rule)
|
||
|
||
return obj, exact, by_class
|
||
|
||
|
||
def has_delivery_output(rule: dict[str, Any]) -> bool:
|
||
status = str(rule.get("mapping_status") or "")
|
||
target = str(rule.get("target_source_layer") or "")
|
||
if target and target != "cm95_unmapped" and status in DELIVERY_MAIN_STATUSES:
|
||
return True
|
||
|
||
derived = rule.get("derived_outputs") or []
|
||
if derived and status in DELIVERY_DERIVED_STATUSES:
|
||
for item in derived:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
layer = str(item.get("source_layer") or "")
|
||
if layer and layer != "cm95_unmapped":
|
||
return True
|
||
return False
|
||
|
||
|
||
def gap_reason(
|
||
cls: str,
|
||
geom: str,
|
||
exact_rules: dict[tuple[str, str], dict[str, Any]],
|
||
by_class: dict[str, list[dict[str, Any]]],
|
||
):
|
||
rule = exact_rules.get((cls, geom))
|
||
class_rules = by_class.get(cls, [])
|
||
|
||
if rule is None:
|
||
inherited_hard_gate = any(
|
||
str(r.get("safety_class") or "") == "hard_gate"
|
||
for r in class_rules
|
||
)
|
||
return {
|
||
"is_gap": True,
|
||
"reason": (
|
||
"NO_EXACT_RULE_KNOWN_HARD_GATE_VARIANT"
|
||
if inherited_hard_gate
|
||
else "NO_EXACT_RULE"
|
||
),
|
||
"rule": None,
|
||
"inherited_hard_gate": inherited_hard_gate,
|
||
}
|
||
|
||
if has_delivery_output(rule):
|
||
return {
|
||
"is_gap": False,
|
||
"reason": "DELIVERED",
|
||
"rule": rule,
|
||
"inherited_hard_gate": False,
|
||
}
|
||
|
||
target = str(rule.get("target_source_layer") or "")
|
||
status = str(rule.get("mapping_status") or "")
|
||
if target == "cm95_unmapped":
|
||
reason = "EXPLICIT_ENGINEERING_ONLY"
|
||
else:
|
||
reason = f"NOT_DELIVERED_STATUS_{status or 'EMPTY'}"
|
||
|
||
return {
|
||
"is_gap": True,
|
||
"reason": reason,
|
||
"rule": rule,
|
||
"inherited_hard_gate": (
|
||
str(rule.get("safety_class") or "") == "hard_gate"
|
||
),
|
||
}
|
||
|
||
|
||
def connect():
|
||
cfg = load_config()
|
||
return pymysql.connect(
|
||
host=cfg.host,
|
||
port=cfg.port,
|
||
user=cfg.user,
|
||
password=cfg.password,
|
||
database=cfg.database,
|
||
charset=cfg.charset,
|
||
cursorclass=pymysql.cursors.DictCursor,
|
||
autocommit=False,
|
||
connect_timeout=20,
|
||
read_timeout=900,
|
||
write_timeout=30,
|
||
)
|
||
|
||
|
||
def main():
|
||
a = parse_args()
|
||
mapping_path = Path(a.mapping).resolve()
|
||
output_path = Path(a.output).resolve()
|
||
outdir = output_path.parent
|
||
outdir.mkdir(parents=True, exist_ok=True)
|
||
|
||
_mapping_obj, exact_rules, by_class = load_rules(mapping_path)
|
||
|
||
with connect() as conn:
|
||
try:
|
||
conn.cursor().execute("SET SESSION TRANSACTION READ ONLY")
|
||
except Exception:
|
||
pass
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
f.object_class_id,
|
||
f.object_class,
|
||
f.geometry_type,
|
||
f.native_scale,
|
||
COUNT(*) AS feature_count,
|
||
COUNT(DISTINCT f.cell_id) AS cell_count
|
||
FROM chart_feature f
|
||
WHERE f.dataset_id=%s
|
||
AND f.geometry_type IN ('POINT','LINE','AREA')
|
||
GROUP BY
|
||
f.object_class_id,
|
||
f.object_class,
|
||
f.geometry_type,
|
||
f.native_scale
|
||
ORDER BY
|
||
f.object_class,
|
||
f.geometry_type,
|
||
f.native_scale
|
||
""",
|
||
(a.dataset_id,),
|
||
)
|
||
grouped = list(cur.fetchall())
|
||
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
object_class_id,
|
||
object_class,
|
||
geometry_code,
|
||
display_code,
|
||
description
|
||
FROM chart_dictionary_object
|
||
WHERE dataset_id=%s
|
||
""",
|
||
(a.dataset_id,),
|
||
)
|
||
dictionary_rows = list(cur.fetchall())
|
||
|
||
conn.rollback()
|
||
|
||
dictionary = {
|
||
int(r["object_class_id"]): r
|
||
for r in dictionary_rows
|
||
if r.get("object_class_id") is not None
|
||
}
|
||
|
||
variants: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for row in grouped:
|
||
cls = str(row["object_class"])
|
||
geom = str(row["geometry_type"]).upper()
|
||
key = (cls, geom)
|
||
v = variants.setdefault(
|
||
key,
|
||
{
|
||
"object_class_id": (
|
||
int(row["object_class_id"])
|
||
if row.get("object_class_id") is not None
|
||
else None
|
||
),
|
||
"object_class": cls,
|
||
"geometry_type": geom,
|
||
"feature_count": 0,
|
||
"cell_count_max": 0,
|
||
"scale_counts": Counter(),
|
||
},
|
||
)
|
||
cnt = int(row["feature_count"])
|
||
v["feature_count"] += cnt
|
||
v["cell_count_max"] = max(v["cell_count_max"], int(row["cell_count"]))
|
||
v["scale_counts"][str(int(row["native_scale"]))] += cnt
|
||
|
||
gap_variants = []
|
||
for key, v in sorted(
|
||
variants.items(),
|
||
key=lambda kv: (-kv[1]["feature_count"], kv[0][0], kv[0][1]),
|
||
):
|
||
cls, geom = key
|
||
state = gap_reason(cls, geom, exact_rules, by_class)
|
||
if not state["is_gap"]:
|
||
continue
|
||
|
||
oid = v["object_class_id"]
|
||
d = dictionary.get(oid) if oid is not None else None
|
||
rule = state["rule"]
|
||
class_rules = by_class.get(cls, [])
|
||
|
||
item = {
|
||
"object_class_id": oid,
|
||
"object_class": cls,
|
||
"geometry_type": geom,
|
||
"feature_count": int(v["feature_count"]),
|
||
"cell_count": int(v["cell_count_max"]),
|
||
"scale_counts": dict(sorted(v["scale_counts"].items(), key=lambda x: int(x[0]))),
|
||
"gap_reason": state["reason"],
|
||
"dictionary_description": d.get("description") if d else None,
|
||
"dictionary_geometry_code": d.get("geometry_code") if d else None,
|
||
"dictionary_display_code": d.get("display_code") if d else None,
|
||
"exact_rule_id": str(rule.get("rule_id")) if rule else None,
|
||
"exact_rule_status": str(rule.get("mapping_status")) if rule else None,
|
||
"exact_rule_target": str(rule.get("target_source_layer")) if rule else None,
|
||
"exact_rule_safety": str(rule.get("safety_class")) if rule else None,
|
||
"class_known_geometry_rules": [
|
||
{
|
||
"geometry_type": str(r.get("geometry_type") or ""),
|
||
"rule_id": str(r.get("rule_id") or ""),
|
||
"target_source_layer": str(r.get("target_source_layer") or ""),
|
||
"mapping_status": str(r.get("mapping_status") or ""),
|
||
"safety_class": str(r.get("safety_class") or ""),
|
||
"canonical_object_type": str(r.get("canonical_object_type") or ""),
|
||
}
|
||
for r in class_rules
|
||
],
|
||
"inherited_hard_gate": bool(state["inherited_hard_gate"]),
|
||
"attribute_key_counts": {},
|
||
"attribute_value_samples": {},
|
||
"relation20_present": 0,
|
||
"relation20_missing": 0,
|
||
"sample_features": [],
|
||
}
|
||
gap_variants.append(item)
|
||
|
||
gap_keys = {(x["object_class"], x["geometry_type"]) for x in gap_variants}
|
||
gap_classes = sorted({x[0] for x in gap_keys})
|
||
|
||
# Gap rows are normally a small subset of Dataset. Read them once and
|
||
# build attribute/sample evidence in Python, avoiding dozens of SQL round trips.
|
||
if gap_classes:
|
||
placeholders = ",".join(["%s"] * len(gap_classes))
|
||
with connect() as conn:
|
||
try:
|
||
conn.cursor().execute("SET SESSION TRANSACTION READ ONLY")
|
||
except Exception:
|
||
pass
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
f"""
|
||
SELECT
|
||
f.id AS feature_id,
|
||
f.cell_id,
|
||
c.source_relative_path,
|
||
f.source_record_index,
|
||
f.object_class_id,
|
||
f.object_class,
|
||
f.geometry_type,
|
||
f.native_scale,
|
||
f.raw_geotype,
|
||
f.coverage_id,
|
||
f.bbox_min_lon,
|
||
f.bbox_min_lat,
|
||
f.bbox_max_lon,
|
||
f.bbox_max_lat,
|
||
f.attributes_json,
|
||
f.geometry_repaired,
|
||
f.geometry_issue
|
||
FROM chart_feature f
|
||
JOIN chart_cell c ON c.id=f.cell_id
|
||
WHERE f.dataset_id=%s
|
||
AND f.object_class IN ({placeholders})
|
||
AND f.geometry_type IN ('POINT','LINE','AREA')
|
||
ORDER BY
|
||
f.object_class,
|
||
f.geometry_type,
|
||
f.native_scale,
|
||
c.source_relative_path,
|
||
f.source_record_index
|
||
""",
|
||
(a.dataset_id, *gap_classes),
|
||
)
|
||
source_rows = list(cur.fetchall())
|
||
conn.rollback()
|
||
else:
|
||
source_rows = []
|
||
|
||
item_by_key = {
|
||
(x["object_class"], x["geometry_type"]): x
|
||
for x in gap_variants
|
||
}
|
||
key_counters: dict[tuple[str, str], Counter] = defaultdict(Counter)
|
||
value_counters: dict[tuple[str, str, str], Counter] = defaultdict(Counter)
|
||
sample_counts: Counter = Counter()
|
||
|
||
sample_jsonl_path = outdir / "cm95_global_mapping_gap_samples_v1.jsonl"
|
||
with sample_jsonl_path.open("w", encoding="utf-8") as sample_file:
|
||
for row in source_rows:
|
||
key = (
|
||
str(row["object_class"]),
|
||
str(row["geometry_type"]).upper(),
|
||
)
|
||
item = item_by_key.get(key)
|
||
if item is None:
|
||
continue
|
||
|
||
attrs = json_dict(row.get("attributes_json"))
|
||
for attr_key, attr_value in attrs.items():
|
||
key_counters[key][str(attr_key)] += 1
|
||
value_counters[(key[0], key[1], str(attr_key))][scalar_text(attr_value)] += 1
|
||
|
||
rel = attrs.get("_cm93_relation20_raw")
|
||
if rel is None or rel == "":
|
||
item["relation20_missing"] += 1
|
||
else:
|
||
item["relation20_present"] += 1
|
||
|
||
if sample_counts[key] < max(1, int(a.sample_per_variant)):
|
||
sample_counts[key] += 1
|
||
sample = {
|
||
"feature_id": int(row["feature_id"]),
|
||
"cell_id": int(row["cell_id"]),
|
||
"source_relative_path": str(row["source_relative_path"]),
|
||
"source_record_index": int(row["source_record_index"]),
|
||
"object_class_id": (
|
||
int(row["object_class_id"])
|
||
if row.get("object_class_id") is not None
|
||
else None
|
||
),
|
||
"object_class": key[0],
|
||
"geometry_type": key[1],
|
||
"native_scale": int(row["native_scale"]),
|
||
"raw_geotype": row.get("raw_geotype"),
|
||
"coverage_id": row.get("coverage_id"),
|
||
"bbox": [
|
||
row.get("bbox_min_lon"),
|
||
row.get("bbox_min_lat"),
|
||
row.get("bbox_max_lon"),
|
||
row.get("bbox_max_lat"),
|
||
],
|
||
"attributes": attrs,
|
||
"geometry_repaired": bool(row.get("geometry_repaired")),
|
||
"geometry_issue": row.get("geometry_issue"),
|
||
}
|
||
item["sample_features"].append(sample)
|
||
sample_file.write(
|
||
json.dumps(sample, ensure_ascii=False, separators=(",", ":"), default=str)
|
||
+ "\n"
|
||
)
|
||
|
||
for item in gap_variants:
|
||
key = (item["object_class"], item["geometry_type"])
|
||
item["attribute_key_counts"] = dict(
|
||
key_counters[key].most_common()
|
||
)
|
||
values = {}
|
||
for attr_key, _count in key_counters[key].most_common():
|
||
values[attr_key] = [
|
||
{"value": value, "count": count}
|
||
for value, count in value_counters[
|
||
(key[0], key[1], attr_key)
|
||
].most_common(max(1, int(a.attribute_value_limit)))
|
||
]
|
||
item["attribute_value_samples"] = values
|
||
|
||
reason_counts = Counter()
|
||
gap_feature_total = 0
|
||
no_exact_total = 0
|
||
explicit_total = 0
|
||
inherited_hard_gate_total = 0
|
||
for item in gap_variants:
|
||
reason_counts[item["gap_reason"]] += 1
|
||
gap_feature_total += int(item["feature_count"])
|
||
if item["gap_reason"].startswith("NO_EXACT_RULE"):
|
||
no_exact_total += int(item["feature_count"])
|
||
else:
|
||
explicit_total += int(item["feature_count"])
|
||
if item["inherited_hard_gate"]:
|
||
inherited_hard_gate_total += int(item["feature_count"])
|
||
|
||
report = {
|
||
"schema": "navsea.cm95.global-mapping-gap-audit/v1",
|
||
"dataset_id": int(a.dataset_id),
|
||
"mapping_path": str(mapping_path),
|
||
"source_variant_count": len(variants),
|
||
"gap_variant_count": len(gap_variants),
|
||
"gap_object_class_count": len({x["object_class"] for x in gap_variants}),
|
||
"gap_feature_total": gap_feature_total,
|
||
"no_exact_rule_feature_total": no_exact_total,
|
||
"explicit_engineering_or_status_gap_feature_total": explicit_total,
|
||
"inherited_hard_gate_feature_total": inherited_hard_gate_total,
|
||
"gap_reason_counts": dict(reason_counts),
|
||
"gap_variants": gap_variants,
|
||
}
|
||
|
||
output_path.write_text(
|
||
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
csv_path = outdir / "cm95_global_mapping_gap_variants_v1.csv"
|
||
with csv_path.open("w", encoding="utf-8-sig", newline="") as f:
|
||
fields = [
|
||
"object_class_id",
|
||
"object_class",
|
||
"geometry_type",
|
||
"feature_count",
|
||
"cell_count",
|
||
"gap_reason",
|
||
"inherited_hard_gate",
|
||
"dictionary_description",
|
||
"dictionary_geometry_code",
|
||
"dictionary_display_code",
|
||
"exact_rule_id",
|
||
"exact_rule_status",
|
||
"exact_rule_target",
|
||
"exact_rule_safety",
|
||
"scale_counts",
|
||
"known_geometry_rules",
|
||
"attribute_keys",
|
||
"relation20_present",
|
||
"relation20_missing",
|
||
]
|
||
w = csv.DictWriter(f, fieldnames=fields)
|
||
w.writeheader()
|
||
for item in gap_variants:
|
||
w.writerow(
|
||
{
|
||
"object_class_id": item["object_class_id"],
|
||
"object_class": item["object_class"],
|
||
"geometry_type": item["geometry_type"],
|
||
"feature_count": item["feature_count"],
|
||
"cell_count": item["cell_count"],
|
||
"gap_reason": item["gap_reason"],
|
||
"inherited_hard_gate": item["inherited_hard_gate"],
|
||
"dictionary_description": item["dictionary_description"],
|
||
"dictionary_geometry_code": item["dictionary_geometry_code"],
|
||
"dictionary_display_code": item["dictionary_display_code"],
|
||
"exact_rule_id": item["exact_rule_id"],
|
||
"exact_rule_status": item["exact_rule_status"],
|
||
"exact_rule_target": item["exact_rule_target"],
|
||
"exact_rule_safety": item["exact_rule_safety"],
|
||
"scale_counts": json.dumps(item["scale_counts"], ensure_ascii=False, separators=(",", ":")),
|
||
"known_geometry_rules": json.dumps(item["class_known_geometry_rules"], ensure_ascii=False, separators=(",", ":")),
|
||
"attribute_keys": json.dumps(item["attribute_key_counts"], ensure_ascii=False, separators=(",", ":")),
|
||
"relation20_present": item["relation20_present"],
|
||
"relation20_missing": item["relation20_missing"],
|
||
}
|
||
)
|
||
|
||
txt_path = outdir / "cm95_global_mapping_gap_summary_v1.txt"
|
||
lines = [
|
||
"CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1",
|
||
"=" * 76,
|
||
f"Dataset ID : {a.dataset_id}",
|
||
f"Source variants : {len(variants)}",
|
||
f"Gap variants : {len(gap_variants)}",
|
||
f"Gap object classes : {report['gap_object_class_count']}",
|
||
f"Gap feature rows : {gap_feature_total}",
|
||
f"No exact rule rows : {no_exact_total}",
|
||
f"Explicit/status gap rows: {explicit_total}",
|
||
f"Inherited hard-gate rows: {inherited_hard_gate_total}",
|
||
"",
|
||
"Top gap variants:",
|
||
]
|
||
for item in gap_variants[:80]:
|
||
scales = ",".join(
|
||
f"{k}:{v}" for k, v in item["scale_counts"].items()
|
||
)
|
||
lines.append(
|
||
f" {item['object_class']:<10} {item['geometry_type']:<5} "
|
||
f"rows={item['feature_count']:<6} "
|
||
f"reason={item['gap_reason']:<34} "
|
||
f"hard={str(item['inherited_hard_gate']):<5} "
|
||
f"scales=[{scales}]"
|
||
)
|
||
lines += [
|
||
"",
|
||
f"JSON : {output_path}",
|
||
f"CSV : {csv_path}",
|
||
f"SAMPLES: {sample_jsonl_path}",
|
||
]
|
||
txt_path.write_text("\n".join(lines), encoding="utf-8")
|
||
|
||
print("\n".join(lines))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|