from __future__ import annotations import argparse from collections import Counter, defaultdict import json import os from pathlib import Path from typing import Any from dotenv import load_dotenv import pymysql from cm95pbf.s52 import ( lookup_condition_matches, normalize_cm93_attributes_for_s52, normalize_cm93_object_for_s52, ) IMPLEMENTED_CSP = frozenset({ "DEPARE01", "DEPARE02", "DEPCNT02", "LIGHTS05", "SOUNDG02", "OBSTRN04", "WRECKS02", "QUAPOS01", "SLCONS03", "RESTRN01", "RESARE02", "TOPMAR01", }) def _json_object(value: Any) -> dict[str, Any]: if value is None: return {} if isinstance(value, dict): return value try: parsed = json.loads(value) except (TypeError, json.JSONDecodeError): return {} return parsed if isinstance(parsed, dict) else {} def _geometry_key(geometry_type: str, point_table: str, boundary_table: str): return { "POINT": (point_table, "Point"), "SOUNDING": (point_table, "Point"), "LINE": ("Lines", "Line"), "AREA": (boundary_table, "Area"), }.get(str(geometry_type).upper()) def _prepare_lookups(rows: list[dict[str, Any]]): index: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) for row in rows: key = ( str(row.get("object_class") or ""), str(row.get("table_name") or ""), str(row.get("geometry_type") or ""), ) index[key].append(row) for key, candidates in index.items(): candidates.sort(key=lambda r: (-len(r.get("attribute_conditions_raw") or []), int(r.get("id") or 0))) return index def _resolve(candidates: list[dict[str, Any]], attrs: dict[str, Any]): normalized = normalize_cm93_attributes_for_s52(attrs) fallback = None for row in candidates: conditions = row.get("attribute_conditions_raw") or [] if not conditions: if fallback is None: fallback = row continue if all(lookup_condition_matches(c, normalized) for c in conditions): return row return fallback def main() -> int: parser = argparse.ArgumentParser(description="流式审计整个 CM95 Dataset 的 S-52 Lookup/CSP 覆盖率。") parser.add_argument("--lookups", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) parser.add_argument("--env-file", default=Path(".env"), type=Path) parser.add_argument("--dataset-id", default=1, type=int) parser.add_argument("--point-table", default="Paper") parser.add_argument("--boundary-table", default="Plain") args = parser.parse_args() lookup_rows = json.loads(args.lookups.read_text(encoding="utf-8")) lookup_index = _prepare_lookups(lookup_rows) load_dotenv(args.env_file) conn = pymysql.connect( host=os.environ["CM93_DB_HOST"], port=int(os.environ.get("CM93_DB_PORT", "3306")), user=os.environ["CM93_DB_USER"], password=os.environ["CM93_DB_PASSWORD"], database=os.environ["CM93_DB_NAME"], charset=os.environ.get("CM93_DB_CHARSET", "utf8mb4"), autocommit=True, cursorclass=pymysql.cursors.SSCursor, ) total = 0 matched = 0 no_lookup = 0 empty_candidates = 0 by_geometry = Counter() matched_by_geometry = Counter() no_lookup_by_key = Counter() no_candidate_by_key = Counter() condition_miss_by_key = Counter() matched_by_key = Counter() rcids = Counter() csp_counts = Counter() implemented_csp_counts = Counter() unsupported_csp_counts = Counter() opcode_counts = Counter() direct_symbol_counts = Counter() pattern_counts = Counter() complex_line_counts = Counter() area_color_counts = Counter() simple_line_counts = Counter() tx_rule_counts = Counter() te_rule_counts = Counter() substitution_counts = Counter() sql = ( "SELECT object_class, geometry_type, attributes_json " "FROM chart_feature WHERE dataset_id=%s ORDER BY id" ) with conn.cursor() as cur: cur.execute(sql, (args.dataset_id,)) for object_class, geometry_type, attrs_raw in cur: total += 1 object_class = str(object_class) geometry_type = str(geometry_type) attrs = _json_object(attrs_raw) lookup_class, lookup_attrs = normalize_cm93_object_for_s52( object_class, geometry_type, attrs ) if lookup_class != object_class: substitution_counts[(object_class, lookup_class, geometry_type)] += 1 by_geometry[geometry_type] += 1 tg = _geometry_key(geometry_type, args.point_table, args.boundary_table) if tg is None: no_lookup += 1 empty_candidates += 1 no_lookup_by_key[(object_class, geometry_type)] += 1 no_candidate_by_key[(object_class, geometry_type)] += 1 continue table_name, lookup_geometry = tg candidates = lookup_index.get((lookup_class, table_name, lookup_geometry), []) if not candidates: no_lookup += 1 empty_candidates += 1 no_lookup_by_key[(object_class, geometry_type)] += 1 no_candidate_by_key[(object_class, geometry_type)] += 1 continue hit = _resolve(candidates, lookup_attrs) if hit is None: no_lookup += 1 no_lookup_by_key[(object_class, geometry_type)] += 1 condition_miss_by_key[(object_class, geometry_type)] += 1 continue matched += 1 matched_by_geometry[geometry_type] += 1 matched_by_key[(object_class, geometry_type)] += 1 rcids[str(hit.get("rcid") or "")] += 1 rules = hit.get("rules") or [] for rule in rules: opcode = str(rule.get("opcode") or "") if opcode: opcode_counts[opcode] += 1 args_list = rule.get("args") or [] if opcode == "SY" and args_list: direct_symbol_counts[str(args_list[0]).strip("'\"")[:8]] += 1 elif opcode == "AP" and args_list: pattern_counts[str(args_list[0]).strip("'\"")[:8]] += 1 elif opcode == "LC" and args_list: complex_line_counts[str(args_list[0]).strip("'\"")[:8]] += 1 elif opcode == "AC" and args_list: area_color_counts[str(args_list[0]).strip("'\"")] += 1 elif opcode == "LS" and len(args_list) >= 3: simple_line_counts[tuple(str(x).strip("'\"") for x in args_list[:3])] += 1 elif opcode == "TX": tx_rule_counts[tuple(str(x) for x in args_list)] += 1 elif opcode == "TE": te_rule_counts[tuple(str(x) for x in args_list)] += 1 if opcode == "CS" and args_list: name = str(args_list[0])[:8] csp_counts[name] += 1 if name in IMPLEMENTED_CSP: implemented_csp_counts[name] += 1 else: unsupported_csp_counts[name] += 1 conn.close() top_no_lookup = [ {"object_class": k[0], "geometry_type": k[1], "count": v} for k, v in no_lookup_by_key.most_common() ] top_matched = [ {"object_class": k[0], "geometry_type": k[1], "count": v} for k, v in matched_by_key.most_common() ] report = { "schema": "navsea.cm95.s52-full-database-audit/v1", "dataset_id": args.dataset_id, "point_table": args.point_table, "boundary_table": args.boundary_table, "feature_count": total, "matched_lookup_count": matched, "no_lookup_count": no_lookup, "no_candidate_count": empty_candidates, "condition_no_match_count": no_lookup - empty_candidates, "match_rate": matched / total if total else None, "geometry_counts": dict(by_geometry), "matched_geometry_counts": dict(matched_by_geometry), "lookup_rcid_top100": rcids.most_common(100), "opcode_counts": dict(opcode_counts), "csp_counts": dict(csp_counts), "implemented_csp_counts": dict(implemented_csp_counts), "unsupported_csp_counts": dict(unsupported_csp_counts), "direct_symbol_top100": direct_symbol_counts.most_common(100), "symbol_resource_counts": direct_symbol_counts.most_common(), "pattern_resource_counts": pattern_counts.most_common(), "complex_line_resource_counts": complex_line_counts.most_common(), "area_color_counts": area_color_counts.most_common(), "simple_line_rule_counts": [[list(k), v] for k, v in simple_line_counts.most_common()], "tx_rule_counts": [[list(k), v] for k, v in tx_rule_counts.most_common()], "te_rule_counts": [[list(k), v] for k, v in te_rule_counts.most_common()], "class_substitution_counts": [ {"source_object_class": k[0], "s52_object_class": k[1], "geometry_type": k[2], "count": v} for k, v in substitution_counts.most_common() ], "no_lookup_by_object_geometry": top_no_lookup, "no_candidate_by_object_geometry": [ {"object_class": k[0], "geometry_type": k[1], "count": v} for k, v in no_candidate_by_key.most_common() ], "condition_no_match_by_object_geometry": [ {"object_class": k[0], "geometry_type": k[1], "count": v} for k, v in condition_miss_by_key.most_common() ], "matched_by_object_geometry": top_matched, "notes": { "context_dependent_csp": "本报告只统计 Lookup 实际触发量，不执行需要邻接几何、sounding depth、TOPMAR 平台关系等上下文的 CSP。", "implemented_csp": sorted(IMPLEMENTED_CSP), }, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({ "feature_count": total, "matched_lookup_count": matched, "no_lookup_count": no_lookup, "no_candidate_count": empty_candidates, "condition_no_match_count": no_lookup - empty_candidates, "match_rate": report["match_rate"], "csp_counts": report["csp_counts"], "unsupported_csp_counts": report["unsupported_csp_counts"], "class_substitution_counts": report["class_substitution_counts"], "top_no_lookup": top_no_lookup[:30], }, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())