from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Any import pymysql import yaml ROOT = Path(__file__).resolve().parent MAPPINGS_DIR = ROOT / "tasks" / "pbf" / "mappings" @dataclass(frozen=True) class DbConfig: host: str = "localhost" port: int = 3306 user: str = "root" password: str = "2chi9ks2" database: str = "pbf_analysis" unix_socket: str | None = "/tmp/mysql.sock" DDL_STATEMENTS = [ "DROP VIEW IF EXISTS navsea_new_to_legacy_mapping_vw", "DROP VIEW IF EXISTS navsea_legacy_to_new_mapping_vw", "DROP VIEW IF EXISTS navsea_mapping_registry_vw", "DROP TABLE IF EXISTS navsea_render_rules", "DROP TABLE IF EXISTS navsea_taxonomy_rules", "DROP TABLE IF EXISTS navsea_field_value_rules", "DROP TABLE IF EXISTS navsea_field_name_rules", "DROP TABLE IF EXISTS navsea_source_layer_rules", "DROP TABLE IF EXISTS navsea_rule_bundle", """ CREATE TABLE IF NOT EXISTS navsea_rule_bundle ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, status VARCHAR(20) NOT NULL, effective_date DATE DEFAULT NULL, taxonomy_ruleset VARCHAR(64) DEFAULT NULL, render_ruleset VARCHAR(64) DEFAULT NULL, source_layer_ruleset VARCHAR(64) DEFAULT NULL, field_name_ruleset VARCHAR(64) DEFAULT NULL, field_value_ruleset VARCHAR(64) DEFAULT NULL, source_of_truth VARCHAR(32) NOT NULL DEFAULT 'yaml+sql', notes TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_source_layer_rules ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, source_layer_jp VARCHAR(100) NOT NULL, source_layer_std VARCHAR(100) NOT NULL, semantic_granularity VARCHAR(50) DEFAULT NULL, canonical_family VARCHAR(100) DEFAULT NULL, render_strategy VARCHAR(50) DEFAULT NULL, preserve_source_layer TINYINT(1) NOT NULL DEFAULT 1, style_bound TINYINT(1) NOT NULL DEFAULT 1, style_types_json LONGTEXT, notes TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version, source_layer_jp), KEY idx_source_layer_std (source_layer_std), KEY idx_source_layer_family (canonical_family) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_field_name_rules ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, field_name_jp VARCHAR(100) NOT NULL, field_name_std VARCHAR(100) NOT NULL, field_group_name VARCHAR(50) NOT NULL, keep_in_delivery TINYINT(1) NOT NULL DEFAULT 1, keep_in_engineering TINYINT(1) NOT NULL DEFAULT 1, normalization_class VARCHAR(64) NOT NULL, notes TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version, field_name_jp), UNIQUE KEY uniq_field_name_std (bundle_id, bundle_version, field_name_std) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_field_value_rules ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, rule_id VARCHAR(64) NOT NULL, rule_revision INT NOT NULL, field_name_jp VARCHAR(100) NOT NULL, legacy_value VARCHAR(191) NOT NULL, standardized_field VARCHAR(100) NOT NULL, standardized_value VARCHAR(191) NOT NULL, geom_scope VARCHAR(100) DEFAULT NULL, source_layer_scope VARCHAR(200) DEFAULT NULL, canonical_object_scope VARCHAR(191) DEFAULT NULL, rule_priority INT NOT NULL DEFAULT 100, notes TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version, rule_id, rule_revision), KEY idx_field_value_lookup (field_name_jp, legacy_value), KEY idx_standardized_lookup (standardized_field, standardized_value) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_taxonomy_rules ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, rule_id VARCHAR(64) NOT NULL, rule_revision INT NOT NULL, priority INT NOT NULL, enabled TINYINT(1) NOT NULL DEFAULT 1, match_scope VARCHAR(32) NOT NULL, match_expr_json LONGTEXT NOT NULL, output_json LONGTEXT NOT NULL, canonical_family VARCHAR(100) NOT NULL, canonical_object_type VARCHAR(191) NOT NULL, detection_key_template VARCHAR(191) DEFAULT NULL, rule_reason TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version, rule_id, rule_revision), KEY idx_taxonomy_object (canonical_object_type), KEY idx_taxonomy_family (canonical_family) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_render_rules ( bundle_id VARCHAR(64) NOT NULL, bundle_version VARCHAR(32) NOT NULL, rule_id VARCHAR(64) NOT NULL, rule_revision INT NOT NULL, priority INT NOT NULL, enabled TINYINT(1) NOT NULL DEFAULT 1, match_expr_json LONGTEXT NOT NULL, output_json LONGTEXT NOT NULL, chart_render_type VARCHAR(32) DEFAULT NULL, chart_symbol_family VARCHAR(64) DEFAULT NULL, chart_symbol_code VARCHAR(64) DEFAULT NULL, chart_line_style VARCHAR(64) DEFAULT NULL, chart_fill_style VARCHAR(64) DEFAULT NULL, chart_text_style VARCHAR(64) DEFAULT NULL, chart_priority INT DEFAULT NULL, chart_visibility_min INT DEFAULT NULL, chart_visibility_max INT DEFAULT NULL, area_usage_class VARCHAR(64) DEFAULT NULL, hazard_class VARCHAR(64) DEFAULT NULL, hazard_severity VARCHAR(64) DEFAULT NULL, rule_reason TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NULL DEFAULT NULL, PRIMARY KEY (bundle_id, bundle_version, rule_id, rule_revision), KEY idx_render_symbol (chart_symbol_code), KEY idx_render_fill (chart_fill_style), KEY idx_render_text (chart_text_style) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE OR REPLACE VIEW navsea_mapping_registry_vw AS SELECT 'source_layer' AS mapping_type, bundle_id, bundle_version, source_layer_jp AS legacy_key, source_layer_jp AS legacy_value, 'source_layer_std' AS new_key, source_layer_std AS new_value, NULL AS rule_id, notes FROM navsea_source_layer_rules UNION ALL SELECT 'field_name' AS mapping_type, bundle_id, bundle_version, field_name_jp AS legacy_key, field_name_jp AS legacy_value, 'field_name_std' AS new_key, field_name_std AS new_value, NULL AS rule_id, notes FROM navsea_field_name_rules UNION ALL SELECT 'field_value' AS mapping_type, bundle_id, bundle_version, field_name_jp AS legacy_key, legacy_value AS legacy_value, standardized_field AS new_key, standardized_value AS new_value, rule_id AS rule_id, notes FROM navsea_field_value_rules UNION ALL SELECT 'taxonomy' AS mapping_type, bundle_id, bundle_version, 'match_expr' AS legacy_key, CAST(match_expr_json AS CHAR(4096)) AS legacy_value, 'canonical_object_type' AS new_key, canonical_object_type AS new_value, rule_id AS rule_id, rule_reason AS notes FROM navsea_taxonomy_rules UNION ALL SELECT 'render' AS mapping_type, bundle_id, bundle_version, 'match_expr' AS legacy_key, CAST(match_expr_json AS CHAR(4096)) AS legacy_value, 'render_output' AS new_key, CAST(output_json AS CHAR(4096)) AS new_value, rule_id AS rule_id, rule_reason AS notes FROM navsea_render_rules """, """ CREATE OR REPLACE VIEW navsea_legacy_to_new_mapping_vw AS SELECT mapping_type, bundle_id, bundle_version, legacy_key, legacy_value, new_key, new_value, rule_id, notes FROM navsea_mapping_registry_vw """, """ CREATE OR REPLACE VIEW navsea_new_to_legacy_mapping_vw AS SELECT mapping_type, bundle_id, bundle_version, new_key, new_value, legacy_key, legacy_value, rule_id, notes FROM navsea_mapping_registry_vw """, ] def load_yaml(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as f: return yaml.safe_load(f) or {} def json_or_none(value: Any) -> str | None: if value in (None, "", [], {}): return None return json.dumps(value, ensure_ascii=False, sort_keys=True) def connect(db: DbConfig) -> pymysql.Connection: kwargs: dict[str, Any] = { "host": db.host, "port": db.port, "user": db.user, "password": db.password, "database": db.database, "charset": "utf8mb4", "autocommit": False, "cursorclass": pymysql.cursors.DictCursor, } if db.unix_socket: kwargs["unix_socket"] = db.unix_socket return pymysql.connect(**kwargs) def purge_bundle(cur: pymysql.cursors.Cursor, bundle_id: str, bundle_version: str) -> None: tables = ( "navsea_source_layer_rules", "navsea_field_name_rules", "navsea_field_value_rules", "navsea_taxonomy_rules", "navsea_render_rules", "navsea_rule_bundle", ) for table in tables: cur.execute( f"DELETE FROM {table} WHERE bundle_id=%s AND bundle_version=%s", (bundle_id, bundle_version), ) def load_canonical_layer_rules(cur: pymysql.cursors.Cursor) -> dict[str, dict[str, Any]]: cur.execute( """ SELECT source_layer, semantic_granularity, canonical_family, render_strategy, preserve_source_layer, style_bound, style_types, notes FROM canonical_layer_rules """ ) return {row["source_layer"]: row for row in cur.fetchall()} def sync() -> dict[str, int]: bundle = load_yaml(MAPPINGS_DIR / "navsea_rule_bundle_v1.yaml") source_layer_rules = load_yaml(MAPPINGS_DIR / "navsea_source_layer_rules_v1.yaml") field_name_rules = load_yaml(MAPPINGS_DIR / "navsea_field_name_rules_v1.yaml") field_value_rules = load_yaml(MAPPINGS_DIR / "navsea_field_value_rules_v1.yaml") taxonomy_rules = load_yaml(MAPPINGS_DIR / "navsea_taxonomy_rules_v1.yaml") render_rules = load_yaml(MAPPINGS_DIR / "navsea_render_rules_v1.yaml") bundle_id = bundle["bundle_id"] bundle_version = bundle["bundle_version"] counts = { "source_layer_rules": 0, "field_name_rules": 0, "field_value_rules": 0, "taxonomy_rules": 0, "render_rules": 0, } with connect(DbConfig()) as conn: with conn.cursor() as cur: for ddl in DDL_STATEMENTS: cur.execute(ddl) purge_bundle(cur, bundle_id, bundle_version) cur.execute( """ INSERT INTO navsea_rule_bundle ( bundle_id, bundle_version, status, effective_date, taxonomy_ruleset, render_ruleset, source_layer_ruleset, field_name_ruleset, field_value_ruleset, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, bundle["status"], bundle.get("effective_date"), bundle.get("taxonomy_ruleset"), bundle.get("render_ruleset"), bundle.get("source_layer_ruleset"), bundle.get("field_name_ruleset"), bundle.get("field_value_ruleset"), bundle.get("notes"), ), ) canonical_layers = load_canonical_layer_rules(cur) for rule in source_layer_rules.get("rules", []): layer_meta = canonical_layers.get(rule["source_layer_jp"], {}) style_types = layer_meta.get("style_types") style_types_json = None if style_types: style_types_json = json.dumps( [part.strip() for part in str(style_types).split(",") if part.strip()], ensure_ascii=False, ) cur.execute( """ INSERT INTO navsea_source_layer_rules ( bundle_id, bundle_version, source_layer_jp, source_layer_std, semantic_granularity, canonical_family, render_strategy, preserve_source_layer, style_bound, style_types_json, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, rule["source_layer_jp"], rule["source_layer_std"], layer_meta.get("semantic_granularity"), layer_meta.get("canonical_family"), layer_meta.get("render_strategy"), int(layer_meta.get("preserve_source_layer", 1)), int(layer_meta.get("style_bound", 1)), style_types_json, rule.get("notes") or layer_meta.get("notes"), ), ) counts["source_layer_rules"] += 1 for rule in field_name_rules.get("rules", []): cur.execute( """ INSERT INTO navsea_field_name_rules ( bundle_id, bundle_version, field_name_jp, field_name_std, field_group_name, keep_in_delivery, keep_in_engineering, normalization_class, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, rule["field_name_jp"], rule["field_name_std"], rule["field_group"], int(bool(rule["keep_in_delivery"])), int(bool(rule["keep_in_engineering"])), rule["normalization_class"], rule.get("notes"), ), ) counts["field_name_rules"] += 1 for rule in field_value_rules.get("rules", []): cur.execute( """ INSERT INTO navsea_field_value_rules ( bundle_id, bundle_version, rule_id, rule_revision, field_name_jp, legacy_value, standardized_field, standardized_value, geom_scope, source_layer_scope, canonical_object_scope, rule_priority, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, rule["rule_id"], int(rule["rule_revision"]), rule["field_name_jp"], str(rule["legacy_value"]), rule["standardized_field"], rule["standardized_value"], rule.get("geom_scope"), rule.get("source_layer_scope"), rule.get("canonical_object_scope"), int(rule.get("rule_priority", 100)), rule.get("notes"), ), ) counts["field_value_rules"] += 1 for rule in taxonomy_rules.get("rules", []): output = rule["output"] cur.execute( """ INSERT INTO navsea_taxonomy_rules ( bundle_id, bundle_version, rule_id, rule_revision, priority, enabled, match_scope, match_expr_json, output_json, canonical_family, canonical_object_type, detection_key_template, rule_reason ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, rule["rule_id"], int(rule["rule_revision"]), int(rule["priority"]), int(bool(rule["enabled"])), rule["match_scope"], json_or_none(rule["match_expr"]), json_or_none(output), output["canonical_family"], output["canonical_object_type"], output.get("detection_key_template"), rule.get("reason"), ), ) counts["taxonomy_rules"] += 1 for rule in render_rules.get("rules", []): output = rule["output"] cur.execute( """ INSERT INTO navsea_render_rules ( bundle_id, bundle_version, rule_id, rule_revision, priority, enabled, match_expr_json, output_json, chart_render_type, chart_symbol_family, chart_symbol_code, chart_line_style, chart_fill_style, chart_text_style, chart_priority, chart_visibility_min, chart_visibility_max, area_usage_class, hazard_class, hazard_severity, rule_reason ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bundle_id, bundle_version, rule["rule_id"], int(rule["rule_revision"]), int(rule["priority"]), int(bool(rule["enabled"])), json_or_none(rule["match_expr"]), json_or_none(output), output.get("chart_render_type"), output.get("chart_symbol_family"), output.get("chart_symbol_code"), output.get("chart_line_style"), output.get("chart_fill_style"), output.get("chart_text_style"), output.get("chart_priority"), output.get("chart_visibility_min"), output.get("chart_visibility_max"), output.get("area_usage_class"), output.get("hazard_class"), output.get("hazard_severity"), rule.get("reason"), ), ) counts["render_rules"] += 1 conn.commit() return counts if __name__ == "__main__": result = sync() print(json.dumps(result, ensure_ascii=False, indent=2))