#!/usr/bin/env python3 from __future__ import annotations import argparse import json from dataclasses import dataclass from pathlib import Path import pymysql DB_USER = "root" DB_PASSWORD = "2chi9ks2" DB_HOST = "localhost" DB_SOCKET = "/tmp/mysql.sock" DEFAULT_TARGET_DB = "navsea_japan_coast_grid" DEFAULT_SOURCE_DB = "navsea_fukuoka_saga_grid" DEFAULT_KEEP_LAYERS = ("fish_port_20m", "hazard_50m") DEFAULT_JOB_NAME = "fish_port_20m_full_mysql_resume" @dataclass(frozen=True) class LayerSummary: layer_name: str count: int bbox: tuple[float, float, float, float] | None source_desc: str | None export_file: str | None def mysql_connect(database: str | None = None): kwargs = { "host": DB_HOST, "user": DB_USER, "password": DB_PASSWORD, "charset": "utf8mb4", "autocommit": False, "cursorclass": pymysql.cursors.Cursor, } if DB_SOCKET: kwargs["unix_socket"] = DB_SOCKET if database: kwargs["database"] = database return pymysql.connect(**kwargs) def ensure_database(name: str) -> None: conn = mysql_connect() try: with conn.cursor() as cur: cur.execute( f"CREATE DATABASE IF NOT EXISTS `{name}` " "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" ) conn.commit() finally: conn.close() def ensure_target_schema(conn) -> None: stmts = [ """ CREATE TABLE IF NOT EXISTS navsea_grid_import_state ( job_name VARCHAR(64) NOT NULL PRIMARY KEY, layer_name VARCHAR(32) NOT NULL, source_name VARCHAR(191) NOT NULL, checkpoint_json LONGTEXT NOT NULL, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_grid_import_progress ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, job_name VARCHAR(64) NOT NULL, layer_name VARCHAR(32) NOT NULL, prc VARCHAR(8) NOT NULL, status_name VARCHAR(32) NOT NULL, cluster_count INT NOT NULL, inserted_cells BIGINT NOT NULL, land_cells BIGINT NOT NULL, sea_cells BIGINT NOT NULL, elapsed_sec DOUBLE NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_job_layer_prc (job_name, layer_name, prc), KEY idx_job_created (job_name, created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_grid_layer_meta ( layer_name VARCHAR(32) NOT NULL PRIMARY KEY, description VARCHAR(255) NOT NULL, source_desc TEXT NOT NULL, cell_size_m DOUBLE NOT NULL, feature_count BIGINT NOT NULL, bbox_min_lon DOUBLE NOT NULL, bbox_min_lat DOUBLE NOT NULL, bbox_max_lon DOUBLE NOT NULL, bbox_max_lat DOUBLE NOT NULL, export_file VARCHAR(255) DEFAULT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, """ CREATE TABLE IF NOT EXISTS navsea_grid_cell ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, layer_name VARCHAR(32) NOT NULL, cell_id VARCHAR(64) NOT NULL, row_idx INT NOT NULL, col_idx INT NOT NULL, cell_size_m DOUBLE NOT NULL, state_name VARCHAR(32) NOT NULL, class_name VARCHAR(32) NOT NULL, source_name VARCHAR(191) NOT NULL, min_lon DOUBLE NOT NULL, min_lat DOUBLE NOT NULL, max_lon DOUBLE NOT NULL, max_lat DOUBLE NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uniq_layer_cell (layer_name, cell_id), KEY idx_layer_state (layer_name, state_name), KEY idx_layer_rowcol (layer_name, row_idx, col_idx) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """, ] with conn.cursor() as cur: for stmt in stmts: cur.execute(stmt) conn.commit() def table_count(conn, table_name: str) -> int: with conn.cursor() as cur: cur.execute(f"SELECT COUNT(*) FROM `{table_name}`") return int(cur.fetchone()[0] or 0) def fetch_layer_summary(conn, layer_name: str) -> LayerSummary: with conn.cursor() as cur: cur.execute( """ SELECT COUNT(*), MIN(min_lon), MIN(min_lat), MAX(max_lon), MAX(max_lat) FROM navsea_grid_cell WHERE layer_name=%s """, (layer_name,), ) row = cur.fetchone() count = int(row[0] or 0) bbox = None if count == 0 else (float(row[1]), float(row[2]), float(row[3]), float(row[4])) cur.execute( """ SELECT description, source_desc, export_file FROM navsea_grid_layer_meta WHERE layer_name=%s """, (layer_name,), ) meta_row = cur.fetchone() if meta_row is None: source_desc = None export_file = None else: source_desc = str(meta_row[1]) export_file = meta_row[2] return LayerSummary(layer_name=layer_name, count=count, bbox=bbox, source_desc=source_desc, export_file=export_file) def clear_target_layers(conn, layers: tuple[str, ...], job_name: str) -> None: with conn.cursor() as cur: placeholders = ", ".join(["%s"] * len(layers)) cur.execute(f"DELETE FROM navsea_grid_cell WHERE layer_name IN ({placeholders})", layers) cur.execute(f"DELETE FROM navsea_grid_layer_meta WHERE layer_name IN ({placeholders})", layers) cur.execute( f"DELETE FROM navsea_grid_import_progress WHERE job_name=%s AND layer_name IN ({placeholders})", (job_name, *layers), ) cur.execute("DELETE FROM navsea_grid_import_state WHERE job_name=%s", (job_name,)) conn.commit() def clear_target_rows_by_job(conn, job_name: str) -> None: with conn.cursor() as cur: cur.execute("DELETE FROM navsea_grid_import_progress WHERE job_name=%s", (job_name,)) cur.execute("DELETE FROM navsea_grid_import_state WHERE job_name=%s", (job_name,)) conn.commit() def copy_layer_cells(source_db: str, target_conn, layer_name: str) -> int: with target_conn.cursor() as cur: cur.execute("DELETE FROM navsea_grid_cell WHERE layer_name=%s", (layer_name,)) cur.execute( """ INSERT INTO navsea_grid_cell (layer_name, cell_id, row_idx, col_idx, cell_size_m, state_name, class_name, source_name, min_lon, min_lat, max_lon, max_lat) SELECT layer_name, cell_id, row_idx, col_idx, cell_size_m, state_name, class_name, source_name, min_lon, min_lat, max_lon, max_lat FROM `{source_db}`.navsea_grid_cell WHERE layer_name=%s """.format(source_db=source_db), (layer_name,), ) target_conn.commit() cur.execute("SELECT COUNT(*) FROM navsea_grid_cell WHERE layer_name=%s", (layer_name,)) return int(cur.fetchone()[0] or 0) def copy_progress_tables(source_db: str, target_conn, job_name: str) -> None: with target_conn.cursor() as cur: cur.execute("DELETE FROM navsea_grid_import_progress WHERE job_name=%s", (job_name,)) cur.execute("DELETE FROM navsea_grid_import_state WHERE job_name=%s", (job_name,)) cur.execute( f""" INSERT INTO navsea_grid_import_state (job_name, layer_name, source_name, checkpoint_json, updated_at) SELECT job_name, layer_name, source_name, checkpoint_json, updated_at FROM `{source_db}`.navsea_grid_import_state WHERE job_name=%s """, (job_name,), ) cur.execute( f""" INSERT INTO navsea_grid_import_progress (job_name, layer_name, prc, status_name, cluster_count, inserted_cells, land_cells, sea_cells, elapsed_sec, created_at) SELECT job_name, layer_name, prc, status_name, cluster_count, inserted_cells, land_cells, sea_cells, elapsed_sec, created_at FROM `{source_db}`.navsea_grid_import_progress WHERE job_name=%s ORDER BY id """, (job_name,), ) target_conn.commit() def upsert_layer_meta( target_conn, *, layer_name: str, description: str, source_desc: str, cell_size_m: float, feature_count: int, bbox: tuple[float, float, float, float], export_file: str | None, ) -> None: with target_conn.cursor() as cur: cur.execute( """ INSERT INTO navsea_grid_layer_meta (layer_name, description, source_desc, cell_size_m, feature_count, bbox_min_lon, bbox_min_lat, bbox_max_lon, bbox_max_lat, export_file) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE description=VALUES(description), source_desc=VALUES(source_desc), cell_size_m=VALUES(cell_size_m), feature_count=VALUES(feature_count), bbox_min_lon=VALUES(bbox_min_lon), bbox_min_lat=VALUES(bbox_min_lat), bbox_max_lon=VALUES(bbox_max_lon), bbox_max_lat=VALUES(bbox_max_lat), export_file=VALUES(export_file) """, ( layer_name, description, source_desc, cell_size_m, feature_count, bbox[0], bbox[1], bbox[2], bbox[3], export_file, ), ) target_conn.commit() def drop_database(name: str) -> None: conn = mysql_connect() try: with conn.cursor() as cur: cur.execute(f"DROP DATABASE IF EXISTS `{name}`") conn.commit() finally: conn.close() def main() -> None: parser = argparse.ArgumentParser(description="把 navsea_fukuoka_saga_grid 合并进 navsea_japan_coast_grid。") parser.add_argument("--target-db", default=DEFAULT_TARGET_DB) parser.add_argument("--source-db", default=DEFAULT_SOURCE_DB) parser.add_argument("--keep-layers", nargs="+", default=list(DEFAULT_KEEP_LAYERS), help="默认搬运的 layer") parser.add_argument("--drop-source-db", action="store_true", help="合并完成后删除源数据库") parser.add_argument("--dry-run", action="store_true", help="只打印将要执行的动作，不真正写入") args = parser.parse_args() keep_layers = tuple(args.keep_layers) ensure_database(args.target_db) source_conn = mysql_connect(args.source_db) target_conn = mysql_connect(args.target_db) try: ensure_target_schema(target_conn) print(f"目标库: {args.target_db}") print(f"源库: {args.source_db}") print(f"搬运图层: {', '.join(keep_layers)}") if args.dry_run: for layer in keep_layers: summary = fetch_layer_summary(source_conn, layer) print(f"DRY-RUN {layer}: source_count={summary.count} bbox={summary.bbox}") return clear_target_layers(target_conn, keep_layers, DEFAULT_JOB_NAME) clear_target_rows_by_job(target_conn, DEFAULT_JOB_NAME) # 先把需要保留的图层从源库搬到全国库。 copied_counts: dict[str, int] = {} for layer in keep_layers: with source_conn.cursor() as cur: cur.execute("SELECT COUNT(*) FROM navsea_grid_cell WHERE layer_name=%s", (layer,)) source_count = int(cur.fetchone()[0] or 0) if source_count == 0: print(f"警告: 源库里没有 layer={layer}，跳过") continue copied = copy_layer_cells(args.source_db, target_conn, layer) copied_counts[layer] = copied print(f"已搬运 {layer}: {copied} 条") # 迁移状态表，方便后续继续 resume / 审计。 if table_count(source_conn, "navsea_grid_import_state") > 0: copy_progress_tables(args.source_db, target_conn, DEFAULT_JOB_NAME) print("已搬运 import_state / import_progress") # 重新写目标库里的图层元信息。 if "hazard_50m" in copied_counts: hazard_summary = fetch_layer_summary(target_conn, "hazard_50m") upsert_layer_meta( target_conn, layer_name="hazard_50m", description="福冈/佐贺 PBF 海上障碍 50m 黄格", source_desc="pbf-delivery-kyushu-reencoded z12", cell_size_m=50.0, feature_count=hazard_summary.count, bbox=hazard_summary.bbox or (0.0, 0.0, 0.0, 0.0), export_file="src/pbf/coastline-mysql/fukuoka_saga/hazard_50m_grid.geojson", ) if "fish_port_20m" in copied_counts: fish_summary = fetch_layer_summary(target_conn, "fish_port_20m") upsert_layer_meta( target_conn, layer_name="fish_port_20m", description="全国渔港 20m 可航/不可航格", source_desc="coastline/C09-06.zip + navsea_japan_coast_grid.coast_200m", cell_size_m=20.0, feature_count=fish_summary.count, bbox=fish_summary.bbox or (0.0, 0.0, 0.0, 0.0), export_file="src/pbf/coastline-mysql/fukuoka_saga/fish_port_20m_grid.geojson", ) manifest = { "target_db": args.target_db, "source_db": args.source_db, "kept_layers": list(keep_layers), "layer_counts": {layer: copied_counts.get(layer, 0) for layer in keep_layers}, "completed": True, } manifest_path = Path("out/navsea_merge_navgrid_mysql.manifest.json") manifest_path.parent.mkdir(parents=True, exist_ok=True) manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") print(f"已写出清单: {manifest_path}") finally: target_conn.close() source_conn.close() if args.drop_source_db: drop_database(args.source_db) print(f"已删除源数据库: {args.source_db}") if __name__ == "__main__": main()