#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import json import math import sqlite3 from dataclasses import asdict, dataclass from pathlib import Path from typing import Iterable import pymysql DB_USER = "root" DB_PASSWORD = "2chi9ks2" DB_HOST = "localhost" DB_SOCKET = "/tmp/mysql.sock" DEFAULT_OUTPUT_DB = Path("out/navsea_national_navigation_fusion.sqlite") DEFAULT_MANIFEST = Path("out/navsea_national_navigation_fusion.manifest.json") DEFAULT_SOURCES = ( { "source_id": "coast_200m", "layer_key": "coast_200m", "grid_level": 200, "role": "base_coast", "description": "全国海岸线 200m 硬阻塞格", "source_type": "mysql_grid", "database_name": "navsea_japan_coast_grid", "source_table": "navsea_grid_cell", "source_layer": "coast_200m", "source_path": "coastline/C23-06_*_GML.zip", "notes": "全国 200x200 海岸线底盘", }, { "source_id": "fish_port_20m", "layer_key": "fish_port_20m", "grid_level": 20, "role": "port_detail", "description": "全国渔港 20m 陆基/海面格", "source_type": "mysql_grid", "database_name": "navsea_fukuoka_saga_grid", "source_table": "navsea_grid_cell", "source_layer": "fish_port_20m", "source_path": "coastline/C09-06.zip", "notes": "当前仓库里的渔港 20m 生成结果", }, { "source_id": "hazard_50m", "layer_key": "hazard_50m", "grid_level": 50, "role": "overlay", "description": "海上障碍 50m 黄格", "source_type": "mysql_grid", "database_name": "navsea_fukuoka_saga_grid", "source_table": "navsea_grid_cell", "source_layer": "hazard_50m", "source_path": "pbf-delivery-kyushu-reencoded z12", "notes": "当前仓库里的障碍叠加层", }, ) @dataclass(frozen=True) class SourceSpec: source_id: str layer_key: str grid_level: int role: str description: str source_type: str database_name: str source_table: str source_layer: str source_path: str notes: str = "" def mysql_connect(database: str): kwargs = { "host": DB_HOST, "user": DB_USER, "password": DB_PASSWORD, "charset": "utf8mb4", "autocommit": False, "cursorclass": pymysql.cursors.SSCursor, } if DB_SOCKET: kwargs["unix_socket"] = DB_SOCKET kwargs["database"] = database return pymysql.connect(**kwargs) def open_output_db(path: Path, recreate: bool) -> sqlite3.Connection: if recreate and path.exists(): path.unlink() path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=OFF") conn.execute("PRAGMA temp_store=MEMORY") conn.execute("PRAGMA cache_size=200000") conn.execute("PRAGMA busy_timeout=5000") return conn def create_schema(conn: sqlite3.Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS source_registry ( source_id TEXT PRIMARY KEY, source_type TEXT NOT NULL, database_name TEXT NOT NULL, source_table TEXT NOT NULL, source_layer TEXT NOT NULL, source_path TEXT NOT NULL, status TEXT NOT NULL, notes TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS layer_registry ( layer_key TEXT PRIMARY KEY, grid_level INTEGER NOT NULL, role TEXT NOT NULL, description TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS layer_summary ( layer_key TEXT PRIMARY KEY, source_id TEXT NOT NULL, database_name TEXT NOT NULL, source_table TEXT NOT NULL, source_layer TEXT NOT NULL, grid_level INTEGER NOT NULL, cell_size_m REAL NOT NULL, cell_count INTEGER NOT NULL, bbox_min_lon REAL, bbox_min_lat REAL, bbox_max_lon REAL, bbox_max_lat REAL, generated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS fusion_cell ( layer_key TEXT NOT NULL, cell_id TEXT NOT NULL, source_id TEXT NOT NULL, source_type TEXT NOT NULL, database_name TEXT NOT NULL, source_table TEXT NOT NULL, source_layer TEXT NOT NULL, grid_level INTEGER NOT NULL, row_idx INTEGER NOT NULL, col_idx INTEGER NOT NULL, cell_size_m REAL NOT NULL, state_name TEXT NOT NULL, class_name TEXT NOT NULL, source_name TEXT NOT NULL, min_lon REAL NOT NULL, min_lat REAL NOT NULL, max_lon REAL NOT NULL, max_lat REAL NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (layer_key, cell_id) ); CREATE INDEX IF NOT EXISTS idx_fusion_cell_layer_grid ON fusion_cell(layer_key, grid_level); CREATE INDEX IF NOT EXISTS idx_fusion_cell_layer_state ON fusion_cell(layer_key, state_name); CREATE INDEX IF NOT EXISTS idx_fusion_cell_bbox ON fusion_cell(layer_key, min_lon, min_lat, max_lon, max_lat); """ ) conn.commit() def seed_registry(conn: sqlite3.Connection, sources: Iterable[SourceSpec]) -> None: conn.executemany( "INSERT OR REPLACE INTO source_registry VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [ ( src.source_id, src.source_type, src.database_name, src.source_table, src.source_layer, src.source_path, "ready", src.notes, ) for src in sources ], ) conn.executemany( "INSERT OR REPLACE INTO layer_registry VALUES (?, ?, ?, ?)", [ ( src.layer_key, src.grid_level, src.role, src.description, ) for src in sources ], ) conn.executemany( "INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)", [ ("fusion_version", "v1"), ("fusion_scope", "japan"), ("fusion_note", "全国 200m / 20m / 50m 障碍层统一汇聚"), ("source_count", str(len(tuple(sources)))), ("generated_at", dt.datetime.now().isoformat(timespec="seconds")), ("layer_keys", json.dumps([src.layer_key for src in sources], ensure_ascii=False)), ], ) conn.commit() def insert_batch(conn: sqlite3.Connection, rows: list[tuple]) -> None: conn.executemany( """ INSERT OR REPLACE INTO fusion_cell( layer_key, cell_id, source_id, source_type, database_name, source_table, source_layer, grid_level, row_idx, col_idx, cell_size_m, state_name, class_name, source_name, min_lon, min_lat, max_lon, max_lat ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, rows, ) def import_layer(conn: sqlite3.Connection, spec: SourceSpec, *, batch_size: int = 5000) -> dict[str, object]: mysql_conn = mysql_connect(spec.database_name) try: with mysql_conn.cursor() as cur: cur.execute( f""" SELECT cell_id, row_idx, col_idx, cell_size_m, state_name, class_name, source_name, min_lon, min_lat, max_lon, max_lat FROM {spec.source_table} WHERE layer_name=%s ORDER BY row_idx, col_idx """, (spec.source_layer,), ) count = 0 batch: list[tuple] = [] bbox_min_lon = bbox_min_lat = math.inf bbox_max_lon = bbox_max_lat = -math.inf for row in cur: ( cell_id, row_idx, col_idx, cell_size_m, state_name, class_name, source_name, min_lon, min_lat, max_lon, max_lat, ) = row batch.append( ( spec.layer_key, str(cell_id), spec.source_id, spec.source_type, spec.database_name, spec.source_table, spec.source_layer, int(spec.grid_level), int(row_idx), int(col_idx), float(cell_size_m), str(state_name), str(class_name), str(source_name), float(min_lon), float(min_lat), float(max_lon), float(max_lat), ) ) bbox_min_lon = min(bbox_min_lon, float(min_lon)) bbox_min_lat = min(bbox_min_lat, float(min_lat)) bbox_max_lon = max(bbox_max_lon, float(max_lon)) bbox_max_lat = max(bbox_max_lat, float(max_lat)) count += 1 if len(batch) >= batch_size: insert_batch(conn, batch) conn.commit() batch.clear() if batch: insert_batch(conn, batch) conn.commit() finally: mysql_conn.close() bbox: tuple[float, float, float, float] | None if count == 0: bbox = None else: bbox = (bbox_min_lon, bbox_min_lat, bbox_max_lon, bbox_max_lat) with conn.cursor() as cur: cur.execute( """ INSERT OR REPLACE INTO layer_summary( layer_key, source_id, database_name, source_table, source_layer, grid_level, cell_size_m, cell_count, bbox_min_lon, bbox_min_lat, bbox_max_lon, bbox_max_lat, generated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( spec.layer_key, spec.source_id, spec.database_name, spec.source_table, spec.source_layer, spec.grid_level, float(spec.grid_level), count, bbox[0] if bbox else None, bbox[1] if bbox else None, bbox[2] if bbox else None, bbox[3] if bbox else None, dt.datetime.now().isoformat(timespec="seconds"), ), ) conn.commit() return { "source_id": spec.source_id, "layer_key": spec.layer_key, "grid_level": spec.grid_level, "count": count, "bbox_lonlat": list(bbox) if bbox else None, "database_name": spec.database_name, "source_table": spec.source_table, "source_layer": spec.source_layer, "cell_size_m": float(spec.grid_level), "source_path": spec.source_path, } def build_manifest(conn: sqlite3.Connection, import_summaries: list[dict[str, object]]) -> dict[str, object]: meta = dict(conn.execute("SELECT key, value FROM meta").fetchall()) layer_rows = conn.execute( """ SELECT layer_key, grid_level, role, description FROM layer_registry ORDER BY grid_level, layer_key """ ).fetchall() return { "meta": meta, "layers": [ { "layer_key": row[0], "grid_level": row[1], "role": row[2], "description": row[3], } for row in layer_rows ], "imports": import_summaries, "cell_counts": [ { "layer_key": row[0], "count": row[1], } for row in conn.execute( "SELECT layer_key, COUNT(*) FROM fusion_cell GROUP BY layer_key ORDER BY layer_key" ).fetchall() ], } def write_template(path: Path) -> None: payload = { "fusion_version": "v1", "sources": [asdict(SourceSpec(**spec)) for spec in DEFAULT_SOURCES], "notes": "这个模板只是记录默认汇聚层，不会执行导入。", } path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser(description="把全国 200m / 20m / 50m 障碍层汇聚到一个 SQLite 数据库。") parser.add_argument("--output-db", type=Path, default=DEFAULT_OUTPUT_DB) parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST) parser.add_argument("--recreate", action="store_true", help="重建输出数据库") parser.add_argument("--emit-template", action="store_true", help="只输出默认模板 JSON，不执行汇聚") parser.add_argument("--batch-size", type=int, default=5000, help="每批写入多少条格网记录") args = parser.parse_args() if args.emit_template: write_template(args.manifest) print(f"已写出模板: {args.manifest}") return sources = [SourceSpec(**spec) for spec in DEFAULT_SOURCES] conn = open_output_db(args.output_db, recreate=args.recreate) try: create_schema(conn) seed_registry(conn, sources) import_summaries = [] for spec in sources: summary = import_layer(conn, spec, batch_size=args.batch_size) import_summaries.append(summary) print( f"已汇聚 {spec.layer_key}: count={summary['count']} " f"source={spec.database_name}.{spec.source_table}/{spec.source_layer}" ) if int(summary["count"]) == 0: print(f"警告: {spec.layer_key} 没有导入到任何记录，请检查源库或 source_layer") manifest = build_manifest(conn, import_summaries) args.manifest.parent.mkdir(parents=True, exist_ok=True) args.manifest.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") conn.commit() finally: conn.close() print(f"已生成融合库: {args.output_db}") print(f"已写出清单: {args.manifest}") if __name__ == "__main__": main()