#!/usr/bin/env python3
from __future__ import annotations

import argparse
import datetime as dt
import json
import math
import time
import zipfile
import xml.etree.ElementTree as ET
from functools import wraps
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

import numpy as np
import pymysql
from shapely import get_parts, intersects_xy
from shapely.geometry import LineString, box
from shapely.ops import polygonize, unary_union
from shapely.prepared import prep


DB_NAME = "navsea_japan_coast_grid"
DB_USER = "root"
DB_PASSWORD = "2chi9ks2"
DB_HOST = "localhost"
DB_SOCKET = "/tmp/mysql.sock"

FISH_SRC = "coastline/C09-06.zip"
COAST_DIR = Path("coastline")
LAYER_NAME = "fish_port_20m"
CELL_SIZE_M = 20.0
COAST_BUFFER_M = 5.0
COAST_COARSE_LAYER = "coast_200m"
DEFAULT_COAST_COARSE_MARGIN_M = 0.0
DEFAULT_CHECKPOINT = "out/fish_port_20m_full_resume.checkpoint.json"
DEFAULT_CURRENT_PRC_FILE = "out/fish_port_20m_full_resume.current_prc.json"
RESUME_JOB_NAME = "fish_port_20m_full_mysql_resume"

RADIUS = 6378137.0
MAX_MERCATOR_LAT = 85.0511287798066


@dataclass(frozen=True)
class FishLine:
    prc: str
    fpc: str | None
    coords: list[tuple[float, float]]
    bbox: tuple[float, float, float, float]


class NoCoastlineCurvesError(RuntimeError):
    pass


def mercator_x(lon: float) -> float:
    return RADIUS * math.radians(lon)


def mercator_y(lat: float) -> float:
    lat = max(min(lat, MAX_MERCATOR_LAT), -MAX_MERCATOR_LAT)
    return RADIUS * math.log(math.tan(math.pi / 4.0 + math.radians(lat) / 2.0))


def lon_from_mercator(x: float) -> float:
    return math.degrees(x / RADIUS)


def lat_from_mercator(y: float) -> float:
    return math.degrees(2.0 * math.atan(math.exp(y / RADIUS)) - math.pi / 2.0)


def local_name(tag: str) -> str:
    return tag.split("}", 1)[-1]


def lonlat_from_text(text: str) -> tuple[float, float]:
    parts = [p for p in text.replace(",", " ").split() if p]
    if len(parts) < 2:
        raise ValueError(f"invalid coordinate text: {text!r}")
    lat = float(parts[0])
    lon = float(parts[1])
    return lon, lat


def log(message: str) -> None:
    print(f"[{dt.datetime.now().isoformat(timespec='seconds')}] {message}")


def summarize_value(value):
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, ET.Element):
        return f"<Element {local_name(value.tag)}>"
    if isinstance(value, dict):
        return f"dict(len={len(value)})"
    if isinstance(value, (list, tuple, set)):
        return f"{type(value).__name__}(len={len(value)})"
    if hasattr(value, "geom_type"):
        try:
            bounds = tuple(round(v, 3) for v in value.bounds)
        except Exception:
            bounds = None
        return f"{getattr(value, 'geom_type', type(value).__name__)}(bounds={bounds})"
    return repr(value)


def trace_fn(label: str | None = None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            name = label or func.__name__
            arg_bits = [summarize_value(arg) for arg in args]
            kw_bits = [f"{key}={summarize_value(val)}" for key, val in kwargs.items()]
            joined = ", ".join(arg_bits + kw_bits)
            start = time.monotonic()
            log(f"[TRACE] {name} 开始 {joined}")
            try:
                result = func(*args, **kwargs)
            except Exception as exc:
                log(f"[TRACE] {name} 异常 {type(exc).__name__}: {exc}")
                raise
            elapsed = time.monotonic() - start
            log(f"[TRACE] {name} 完成 elapsed={elapsed:.2f}s")
            return result

        return wrapper

    return decorator


@trace_fn("load_xml")
def load_xml(zip_path: Path) -> ET.Element:
    with zipfile.ZipFile(zip_path) as zf:
        xml_name = next(name for name in zf.namelist() if name.endswith(".xml") and "META" not in name)
        return ET.fromstring(zf.read(xml_name))


def build_point_lookup(root: ET.Element) -> dict[str, tuple[float, float]]:
    lookup: dict[str, tuple[float, float]] = {}
    for point in root.iter():
        if local_name(point.tag) != "GM_Point":
            continue
        point_id = point.attrib.get("id")
        if not point_id:
            continue
        coord_text = None
        for el in point.iter():
            if local_name(el.tag).endswith("DirectPosition.coordinate") and (el.text or "").strip():
                coord_text = el.text.strip()
                break
        if coord_text is None:
            continue
        lookup[point_id] = lonlat_from_text(coord_text)
    return lookup


def extract_curve_coords(curve: ET.Element, point_lookup: dict[str, tuple[float, float]]) -> list[tuple[float, float]]:
    coords: list[tuple[float, float]] = []
    for pos in curve.iter():
        tag = local_name(pos.tag)
        if not tag.startswith("GM_Position"):
            continue
        direct = None
        ref = None
        for child in list(pos):
            child_tag = local_name(child.tag)
            if child_tag.endswith("DirectPosition.coordinate") and (child.text or "").strip():
                direct = child.text.strip()
                break
            if child_tag.endswith("PointRef.point"):
                ref = child.attrib.get("idref")
        if direct is not None:
            coords.append(lonlat_from_text(direct))
        elif ref and ref in point_lookup:
            coords.append(point_lookup[ref])
    return coords


def bbox_from_coords(coords: list[tuple[float, float]]) -> tuple[float, float, float, float]:
    xs = [p[0] for p in coords]
    ys = [p[1] for p in coords]
    return min(xs), min(ys), max(xs), max(ys)


def align_floor(value: float, step: float) -> float:
    return math.floor(value / step) * step


def align_ceil(value: float, step: float) -> float:
    return math.ceil(value / step) * step


def expand_bbox(bbox: tuple[float, float, float, float], margin_m: float) -> tuple[float, float, float, float]:
    return (bbox[0] - margin_m, bbox[1] - margin_m, bbox[2] + margin_m, bbox[3] + margin_m)


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() -> None:
    conn = mysql_connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}` "
                "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
            )
        conn.commit()
    finally:
        conn.close()


def ensure_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_source_cell (layer_name, source_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)
        cur.execute(
            """
            SELECT INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX
            FROM information_schema.STATISTICS
            WHERE TABLE_SCHEMA=%s AND TABLE_NAME='navsea_grid_cell' AND NON_UNIQUE=0
            ORDER BY INDEX_NAME, SEQ_IN_INDEX
            """,
            (DB_NAME,),
        )
        index_columns: dict[str, list[str]] = {}
        for index_name, column_name, _seq in cur.fetchall():
            index_columns.setdefault(index_name, []).append(column_name)
        desired_cols = ["layer_name", "source_name", "cell_id"]
        if index_columns.get("uniq_layer_source_cell") != desired_cols:
            if "uniq_layer_source_cell" in index_columns:
                cur.execute("ALTER TABLE navsea_grid_cell DROP INDEX uniq_layer_source_cell")
            if "uniq_layer_cell" in index_columns:
                cur.execute("ALTER TABLE navsea_grid_cell DROP INDEX uniq_layer_cell")
            cur.execute(
                "ALTER TABLE navsea_grid_cell ADD UNIQUE KEY uniq_layer_source_cell (layer_name, source_name, cell_id)"
            )
        elif "uniq_layer_cell" in index_columns:
            cur.execute("ALTER TABLE navsea_grid_cell DROP INDEX uniq_layer_cell")
    conn.commit()


def upsert_meta(
    cur,
    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:
    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,
        ),
    )


def insert_cells(cur, rows: list[tuple]) -> None:
    if not rows:
        return
    cur.executemany(
        """
        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)
        VALUES
          (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
          row_idx=VALUES(row_idx),
          col_idx=VALUES(col_idx),
          cell_size_m=VALUES(cell_size_m),
          state_name=VALUES(state_name),
          class_name=VALUES(class_name),
          source_name=VALUES(source_name),
          min_lon=VALUES(min_lon),
          min_lat=VALUES(min_lat),
          max_lon=VALUES(max_lon),
          max_lat=VALUES(max_lat)
        """,
        rows,
    )


def save_checkpoint(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp_path.replace(path)


def load_checkpoint(path: Path) -> dict | None:
    if not path.exists():
        return None
    return json.loads(path.read_text(encoding="utf-8"))


def normalize_prc_list(values: Iterable[str]) -> list[str]:
    unique = {normalize_prc_token(str(value)) for value in values}
    try:
        return sorted(unique, key=lambda item: int(item))
    except ValueError:
        return sorted(unique)


def normalize_prc_token(value: str) -> str:
    value = str(value).strip()
    if value.isdigit():
        return f"{int(value):02d}"
    return value


def normalize_fpc_token(value: str) -> str:
    return str(value).strip()


def load_resume_state_from_db(conn, job_name: str, layer_name: str) -> dict | None:
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT checkpoint_json
            FROM navsea_grid_import_state
            WHERE job_name=%s AND layer_name=%s
            """,
            (job_name, layer_name),
        )
        row = cur.fetchone()
        if not row or row[0] is None:
            return None
        state = json.loads(row[0])
        cur.execute(
            """
            SELECT prc
            FROM navsea_grid_import_progress
            WHERE job_name=%s AND layer_name=%s AND status_name='PRC_DONE'
            ORDER BY id
            """,
            (job_name, layer_name),
        )
        completed_prcs = [str(prc) for (prc,) in cur.fetchall()]
    completed_prcs = normalize_prc_list(completed_prcs)
    state["completed_prcs"] = completed_prcs
    state.setdefault("job_name", job_name)
    state.setdefault("layer_name", layer_name)
    state.setdefault("status", "running")
    return state


def load_resume_state(conn, _path: Path, job_name: str, layer_name: str) -> dict | None:
    state = load_resume_state_from_db(conn, job_name, layer_name)
    if state is not None:
        return state
    return None


def load_coarse_mask(conn, layer_name: str) -> tuple[object | None, tuple[float, float, float, float] | None, int]:
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT min_lon, min_lat, max_lon, max_lat
            FROM navsea_grid_cell
            WHERE layer_name=%s
            """,
            (layer_name,),
        )
        boxes = [box(float(min_lon), float(min_lat), float(max_lon), float(max_lat)) for min_lon, min_lat, max_lon, max_lat in cur]

    if not boxes:
        return None, None, 0

    geom = unary_union(boxes)
    return geom, geom.bounds, len(boxes)


def load_coarse_windows_for_bbox(
    conn,
    *,
    layer_name: str,
    bbox_mercator: tuple[float, float, float, float],
    margin_m: float,
) -> list[tuple[float, float, float, float]]:
    minx, miny, maxx, maxy = bbox_mercator
    query_minx = minx - margin_m
    query_miny = miny - margin_m
    query_maxx = maxx + margin_m
    query_maxy = maxy + margin_m

    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT min_lon, min_lat, max_lon, max_lat
            FROM navsea_grid_cell
            WHERE layer_name=%s
              AND NOT (
                max_lon < %s OR min_lon > %s OR
                max_lat < %s OR min_lat > %s
              )
            ORDER BY row_idx, col_idx
            """,
            (layer_name, query_minx, query_maxx, query_miny, query_maxy),
        )
        windows = [
            (
                float(row_min_lon),
                float(row_min_lat),
                float(row_max_lon),
                float(row_max_lat),
            )
            for row_min_lon, row_min_lat, row_max_lon, row_max_lat in cur
        ]
    log(
        f"[TRACE] load_coarse_windows_for_bbox 完成 layer={layer_name} "
        f"bbox={tuple(round(v, 3) for v in bbox_mercator)} margin={margin_m:.1f}m "
        f"windows={len(windows)}"
    )
    return windows


def delete_cells_by_ids(conn, layer_name: str, cell_ids: list[str]) -> None:
    if not cell_ids:
        return
    with conn.cursor() as cur:
        for offset in range(0, len(cell_ids), 1000):
            chunk = cell_ids[offset : offset + 1000]
            placeholders = ", ".join(["%s"] * len(chunk))
            cur.execute(
                f"DELETE FROM navsea_grid_cell WHERE layer_name=%s AND cell_id IN ({placeholders})",
                [layer_name, *chunk],
            )
    conn.commit()


def drop_existing_layer(conn, layer_name: str) -> None:
    with conn.cursor() as cur:
        cur.execute("DELETE FROM navsea_grid_cell WHERE layer_name=%s", (layer_name,))
        cur.execute("DELETE FROM navsea_grid_layer_meta WHERE layer_name=%s", (layer_name,))
        cur.execute("DELETE FROM navsea_grid_import_state WHERE job_name=%s", (RESUME_JOB_NAME,))
        cur.execute("DELETE FROM navsea_grid_import_progress WHERE job_name=%s", (RESUME_JOB_NAME,))
    conn.commit()


def drop_existing_prcs(conn, layer_name: str, prcs: set[str]) -> None:
    if not prcs:
        return
    source_patterns = [f"C09-06+coastline PRC={prc}%" for prc in sorted(normalize_prc_list(prcs))]
    source_where = " OR ".join(["source_name LIKE %s"] * len(source_patterns))
    progress_placeholders = ", ".join(["%s"] * len(prcs))
    with conn.cursor() as cur:
        cur.execute(
            f"DELETE FROM navsea_grid_cell WHERE layer_name=%s AND ({source_where})",
            [layer_name, *source_patterns],
        )
        cur.execute(
            f"DELETE FROM navsea_grid_import_progress WHERE job_name=%s AND prc IN ({progress_placeholders})",
            [RESUME_JOB_NAME, *sorted(prcs)],
        )
    conn.commit()


def drop_existing_fpc(conn, layer_name: str, prcs: set[str], fpc: str) -> None:
    if not prcs:
        return
    source_names = [f"C09-06+coastline PRC={prc} FPC={fpc}" for prc in sorted(normalize_prc_list(prcs))]
    placeholders = ", ".join(["%s"] * len(source_names))
    progress_placeholders = ", ".join(["%s"] * len(prcs))
    with conn.cursor() as cur:
        cur.execute(
            f"DELETE FROM navsea_grid_cell WHERE layer_name=%s AND source_name IN ({placeholders})",
            [layer_name, *source_names],
        )
        cur.execute(
            f"DELETE FROM navsea_grid_import_progress WHERE job_name=%s AND prc IN ({progress_placeholders})",
            [RESUME_JOB_NAME, *sorted(prcs)],
        )
    conn.commit()


def write_prc_status(path: Path, payload: dict) -> None:
    save_checkpoint(path, payload)


def insert_progress(
    cur,
    *,
    job_name: str,
    layer_name: str,
    prc: str,
    status_name: str,
    cluster_count: int,
    inserted_cells: int,
    land_cells: int,
    sea_cells: int,
    elapsed_sec: float,
) -> None:
    cur.execute(
        """
        INSERT INTO navsea_grid_import_progress
          (job_name, layer_name, prc, status_name, cluster_count, inserted_cells, land_cells, sea_cells, elapsed_sec)
        VALUES
          (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        """,
        (
            job_name,
            layer_name,
            prc,
            status_name,
            cluster_count,
            inserted_cells,
            land_cells,
            sea_cells,
            elapsed_sec,
        ),
    )


def rect_lonlat(min_lon: float, min_lat: float, max_lon: float, max_lat: float) -> dict:
    return {
        "type": "Polygon",
        "coordinates": [
            [
                [min_lon, min_lat],
                [max_lon, min_lat],
                [max_lon, max_lat],
                [min_lon, max_lat],
                [min_lon, min_lat],
            ]
        ],
    }


def bbox_intersects(a: tuple[float, float, float, float], b: tuple[float, float, float, float]) -> bool:
    return not (a[2] < b[0] or a[0] > b[2] or a[3] < b[1] or a[1] > b[3])


@trace_fn("discover_prcs")
def discover_prcs(root: ET.Element, selected_prc: set[str] | None) -> list[str]:
    obj = root.find(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}OBJ")
    if obj is None:
        raise SystemExit("OBJ block missing in C09-06.zip")

    prcs: set[str] = set()
    for feature in obj.findall(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}CB03"):
        prc_el = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}PRC")
        prc = normalize_prc_token(prc_el.text) if prc_el is not None and (prc_el.text or "").strip() else "__unknown__"
        if selected_prc is not None and prc not in selected_prc:
            continue
        prcs.add(prc)
    return sorted(prcs)


@trace_fn("discover_target_prcs")
def discover_target_prcs(
    root: ET.Element,
    selected_prc: set[str] | None,
    selected_fpc: str | None,
) -> list[str]:
    if selected_fpc is None:
        return discover_prcs(root, selected_prc)

    obj = root.find(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}OBJ")
    if obj is None:
        raise SystemExit("OBJ block missing in C09-06.zip")

    prcs: set[str] = set()
    for feature in obj.findall(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}CB03"):
        prc_el = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}PRC")
        fpc_el = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}FPC")
        prc = normalize_prc_token(prc_el.text) if prc_el is not None and (prc_el.text or "").strip() else "__unknown__"
        fpc = normalize_fpc_token(fpc_el.text) if fpc_el is not None and (fpc_el.text or "").strip() else None
        if selected_prc is not None and prc not in selected_prc:
            continue
        if fpc != selected_fpc:
            continue
        prcs.add(prc)
    return sorted(prcs)


@trace_fn("collect_fish_lines_for_prc")
def collect_fish_lines_for_prc(
    root: ET.Element,
    point_lookup: dict[str, tuple[float, float]],
    selected_prc: str,
    selected_fpc: str | None = None,
) -> list[FishLine]:
    obj = root.find(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}OBJ")
    if obj is None:
        raise SystemExit("OBJ block missing in C09-06.zip")

    fish_lines: list[FishLine] = []
    for feature in obj.findall(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}CB03"):
        prc_el = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}PRC")
        fpc_el = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}FPC")
        prc = normalize_prc_token(prc_el.text) if prc_el is not None and (prc_el.text or "").strip() else "__unknown__"
        if prc != selected_prc:
            continue
        fpc = normalize_fpc_token(fpc_el.text) if fpc_el is not None and (fpc_el.text or "").strip() else None
        if selected_fpc is not None and fpc != selected_fpc:
            continue
        loc = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}LOC")
        if loc is None:
            continue
        ref = loc.attrib.get("idref")
        if not ref:
            continue
        curve = obj.find(f".//*[@id='{ref}']")
        if curve is None:
            continue
        coords = extract_curve_coords(curve, point_lookup)
        if len(coords) < 2:
            continue
        bbox = bbox_from_coords(coords)
        fish_lines.append(FishLine(prc=prc, fpc=fpc, coords=coords, bbox=bbox))
    return fish_lines


def group_fish_lines_by_fpc(fish_lines: list[FishLine]) -> dict[str, list[FishLine]]:
    groups: dict[str, list[FishLine]] = {}
    for line in fish_lines:
        key = line.fpc or "__unknown__"
        groups.setdefault(key, []).append(line)
    return groups


@trace_fn("cluster_line_indices")
def cluster_line_indices(bboxes: list[tuple[float, float, float, float]], gap_m: float = 250.0) -> list[list[int]]:
    parent = list(range(len(bboxes)))

    def find(x: int) -> int:
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(a: int, b: int) -> None:
        ra = find(a)
        rb = find(b)
        if ra != rb:
            parent[rb] = ra

    expanded = [
        (bb[0] - gap_m, bb[1] - gap_m, bb[2] + gap_m, bb[3] + gap_m)
        for bb in bboxes
    ]
    for i in range(len(expanded)):
        for j in range(i + 1, len(expanded)):
            if bbox_intersects(expanded[i], expanded[j]):
                union(i, j)

    groups: dict[int, list[int]] = {}
    for idx in range(len(parent)):
        root = find(idx)
        groups.setdefault(root, []).append(idx)
    return list(groups.values())


@trace_fn("load_coastlines_for_prc")
def load_coastlines_for_prc(
    project_root: Path,
    prc: str,
    bbox: tuple[float, float, float, float] | None = None,
    search_margin_m: float = 250.0,
) -> tuple[list[LineString], tuple[float, float, float, float]]:
    bbox_geom = box(*expand_bbox(bbox, search_margin_m)) if bbox is not None else None
    lines: list[LineString] = []
    bbox = [float("inf"), float("inf"), float("-inf"), float("-inf")]
    coast_dir = project_root / COAST_DIR
    exact_path = coast_dir / f"C23-06_{prc}_GML.zip"
    candidate_paths = [exact_path] if exact_path.exists() else []
    fallback_paths = sorted(coast_dir.glob("C23-06_*_GML.zip"))

    if exact_path.exists():
        log(f"[TRACE] load_coastlines_for_prc 使用同号海岸线包 {exact_path.name}")
    elif fallback_paths:
        log(
            f"[TRACE] load_coastlines_for_prc 缺少同号海岸线包 {exact_path.name}，"
            f"将按 bbox 退回扫描 {len(fallback_paths)} 个现有海岸线包"
        )
        candidate_paths = fallback_paths
    else:
        raise FileNotFoundError(str(exact_path))

    tried_paths: list[Path] = []

    def scan_path(coast_path: Path) -> None:
        root = load_xml(coast_path)
        point_lookup = build_point_lookup(root)
        found_here = 0
        for curve in root.findall(".//{http://www.opengis.net/gml/3.2}Curve"):
            coords: list[tuple[float, float]] = []
            for pos_list in curve.findall(".//{http://www.opengis.net/gml/3.2}posList"):
                if not pos_list.text:
                    continue
                values = [float(part) for part in pos_list.text.split() if part]
                if len(values) < 4 or len(values) % 2 != 0:
                    continue
                for i in range(0, len(values), 2):
                    lat = values[i]
                    lon = values[i + 1]
                    coords.append((mercator_x(lon), mercator_y(lat)))
            if len(coords) < 2:
                coords = [(mercator_x(lon), mercator_y(lat)) for lon, lat in extract_curve_coords(curve, point_lookup)]
            if len(coords) < 2:
                continue
            line = LineString(coords)
            if bbox_geom is not None and not line.intersects(bbox_geom):
                continue
            lines.append(line)
            found_here += 1
            xs = [p[0] for p in coords]
            ys = [p[1] for p in coords]
            bbox[0] = min(bbox[0], min(xs))
            bbox[1] = min(bbox[1], min(ys))
            bbox[2] = max(bbox[2], max(xs))
            bbox[3] = max(bbox[3], max(ys))
        tried_paths.append(coast_path)
        if found_here > 0:
            log(
                f"[TRACE] load_coastlines_for_prc 海岸线命中 source={coast_path.name} "
                f"hits={found_here}"
            )

    for coast_path in candidate_paths:
        scan_path(coast_path)

    if not lines and exact_path.exists():
        other_paths = [p for p in fallback_paths if p != exact_path]
        if other_paths:
            log(
                f"[TRACE] load_coastlines_for_prc 同号包 {exact_path.name} 未命中 bbox，"
                f"改用其他 {len(other_paths)} 个海岸线包继续搜索"
            )
            for coast_path in other_paths:
                scan_path(coast_path)

    if not lines:
        raise NoCoastlineCurvesError(
            f"no coastline curves parsed for PRC={prc}; "
            f"tried={len(tried_paths)} bbox={tuple(round(v, 3) for v in bbox_geom.bounds) if bbox_geom is not None else None}"
        )
    return lines, (bbox[0], bbox[1], bbox[2], bbox[3])


@trace_fn("build_cells_for_geometry")
def build_cells_for_geometry(
    fish_geom,
    coast_prepared,
    *,
    prc: str,
    cell_size_m: float,
    scan_tile_m: float,
    source_name: str,
    max_scan_cells: int | None,
) -> tuple[list[tuple], int, int, tuple[float, float, float, float], dict[str, float | int]]:
    fish_prepared = prep(fish_geom)
    minx, miny, maxx, maxy = fish_geom.bounds
    start_x = align_floor(minx, cell_size_m)
    start_y = align_floor(miny, cell_size_m)
    end_x = align_ceil(maxx, cell_size_m)
    end_y = align_ceil(maxy, cell_size_m)
    estimated_cols = max(0, int(math.ceil((end_x - start_x) / cell_size_m)))
    estimated_rows = max(0, int(math.ceil((end_y - start_y) / cell_size_m)))
    estimated_cells = estimated_cols * estimated_rows
    if max_scan_cells is not None and estimated_cells > max_scan_cells:
        raise RuntimeError(
            "scan guard triggered: "
            f"estimated_cells={estimated_cells} max_scan_cells={max_scan_cells} "
            f"grid={estimated_cols}x{estimated_rows} bounds={tuple(round(v, 3) for v in fish_geom.bounds)} "
            f"source={source_name} prc={prc}"
        )
    log(
        f"[TRACE] build_cells_for_geometry 准备 "
        f"fish_bounds={tuple(round(v, 3) for v in fish_geom.bounds)} "
        f"scan_range=({start_x:.3f}, {start_y:.3f}, {end_x:.3f}, {end_y:.3f}) "
        f"grid={estimated_cols}x{estimated_rows} estimated_cells={estimated_cells} "
        f"cell={cell_size_m:.1f}m tile={scan_tile_m:.1f}m source={source_name}"
    )

    rows: list[tuple] = []
    land_count = 0
    sea_count = 0
    export_bbox = [float("inf"), float("inf"), float("-inf"), float("-inf")]
    profile = {
        "tile_checked": 0,
        "tile_hit": 0,
        "row_checked": 0,
        "row_hit": 0,
        "mask_row_hit": 0,
        "candidate_cells": 0,
        "precise_cell_checks": 0,
        "precise_hits": 0,
    }

    tile_start_x = align_floor(start_x, scan_tile_m)
    tile_start_y = align_floor(start_y, scan_tile_m)
    tile_end_x = align_ceil(end_x, scan_tile_m)
    tile_end_y = align_ceil(end_y, scan_tile_m)
    half = cell_size_m / 2.0
    log(
        f"[TRACE] build_cells_for_geometry tile_range="
        f"({tile_start_x:.3f}, {tile_start_y:.3f}, {tile_end_x:.3f}, {tile_end_y:.3f})"
    )

    # 对 200m 粗窗这类很小的输入，直接按 20m 网格线性扫描，
    # 避免再走一层 2000m tile 外循环。
    if (end_x - start_x) <= scan_tile_m and (end_y - start_y) <= scan_tile_m:
        profile["tile_checked"] = 1
        profile["tile_hit"] = 1
        log(f"[TRACE] build_cells_for_geometry 使用小窗直算模式")

        y = start_y
        while y < end_y:
            profile["row_checked"] += 1
            row_strip = box(start_x, y, end_x, y + cell_size_m)
            if not fish_prepared.intersects(row_strip):
                y += cell_size_m
                continue
            profile["row_hit"] += 1
            if profile["row_hit"] <= 20 or profile["row_hit"] % 50 == 0:
                log(
                    f"[TRACE] build_cells_for_geometry row_hit "
                    f"row_idx={profile['row_checked']} y={y:.3f} x_span=({start_x:.3f}, {end_x:.3f})"
                )

            ix_start = int(round(start_x / cell_size_m))
            ix_end = int(round(end_x / cell_size_m))
            ix_values = np.arange(ix_start, ix_end, dtype=np.int64)
            x_values = ix_values.astype(float) * cell_size_m
            x_center_values = x_values + half
            x_right_values = x_values + cell_size_m

            mask = (
                intersects_xy(fish_geom, x_center_values, y + half)
                | intersects_xy(fish_geom, x_values, y)
                | intersects_xy(fish_geom, x_right_values, y)
                | intersects_xy(fish_geom, x_values, y + cell_size_m)
                | intersects_xy(fish_geom, x_right_values, y + cell_size_m)
            )
            if not np.any(mask):
                y += cell_size_m
                continue
            profile["mask_row_hit"] += 1
            if profile["mask_row_hit"] <= 20 or profile["mask_row_hit"] % 50 == 0:
                log(
                    f"[TRACE] build_cells_for_geometry mask_hit "
                    f"row_idx={profile['row_checked']} candidate_mask={int(np.count_nonzero(mask))}"
                )

            mask = mask | np.r_[False, mask[:-1]] | np.r_[mask[1:], False]
            candidate_indices = np.flatnonzero(mask)
            profile["candidate_cells"] += int(candidate_indices.size)
            if candidate_indices.size and (profile["candidate_cells"] <= 50 or profile["candidate_cells"] % 200 == 0):
                log(
                    f"[TRACE] build_cells_for_geometry candidate_cells "
                    f"count={int(candidate_indices.size)} total={profile['candidate_cells']} "
                    f"row_y={y:.3f}"
                )
            for idx in candidate_indices:
                profile["precise_cell_checks"] += 1
                x = float(x_values[idx])
                cell_geom = box(x, y, x + cell_size_m, y + cell_size_m)
                if not fish_prepared.intersects(cell_geom):
                    continue
                profile["precise_hits"] += 1
                coast_hit = coast_prepared.intersects(cell_geom)
                if not coast_hit:
                    sea_count += 1
                    continue
                land_count += 1
                state_name = "COASTLINE"
                class_name = state_name
                min_lon = lon_from_mercator(x)
                min_lat = lat_from_mercator(y)
                max_lon = lon_from_mercator(x + cell_size_m)
                max_lat = lat_from_mercator(y + cell_size_m)
                export_bbox[0] = min(export_bbox[0], min_lon)
                export_bbox[1] = min(export_bbox[1], min_lat)
                export_bbox[2] = max(export_bbox[2], max_lon)
                export_bbox[3] = max(export_bbox[3], max_lat)
                ix = int(round(x / cell_size_m))
                iy = int(round(y / cell_size_m))
                cell_id = f"{ix}:{iy}"
                rows.append(
                    (
                        LAYER_NAME,
                        cell_id,
                        iy,
                        ix,
                        cell_size_m,
                        state_name,
                        class_name,
                        source_name,
                        min_lon,
                        min_lat,
                        max_lon,
                        max_lat,
                    )
                )
            y += cell_size_m

        if not rows:
            log(
                f"[TRACE] build_cells_for_geometry 结束 rows=0 coast=0 off_coast=0 "
                f"tile_hit={profile['tile_hit']}/{profile['tile_checked']} "
                f"row_hit={profile['row_hit']}/{profile['row_checked']} "
                f"candidate={profile['candidate_cells']} precise={profile['precise_hits']}/{profile['precise_cell_checks']}"
            )
            return [], 0, 0, (float("inf"), float("inf"), float("-inf"), float("-inf")), profile
        log(
            f"[TRACE] build_cells_for_geometry 结束 rows={len(rows)} coast={land_count} off_coast={sea_count} "
            f"tile_hit={profile['tile_hit']}/{profile['tile_checked']} "
            f"row_hit={profile['row_hit']}/{profile['row_checked']} "
            f"candidate={profile['candidate_cells']} precise={profile['precise_hits']}/{profile['precise_cell_checks']}"
        )
        return rows, land_count, sea_count, (export_bbox[0], export_bbox[1], export_bbox[2], export_bbox[3]), profile

    tile_y = tile_start_y
    while tile_y < tile_end_y:
        tile_x = tile_start_x
        tile_top = min(tile_y + scan_tile_m, end_y)
        while tile_x < tile_end_x:
            profile["tile_checked"] += 1
            tile_right = min(tile_x + scan_tile_m, end_x)
            tile_geom = box(tile_x, tile_y, tile_right, tile_top)
            if not fish_prepared.intersects(tile_geom):
                tile_x += scan_tile_m
                continue
            profile["tile_hit"] += 1
            log(
                f"[TRACE] build_cells_for_geometry tile_hit "
                f"idx={profile['tile_checked']} bbox=({tile_x:.3f}, {tile_y:.3f}, {tile_right:.3f}, {tile_top:.3f})"
            )

            local_start_x = max(start_x, tile_x)
            local_end_x = tile_right
            local_start_y = max(start_y, tile_y)
            local_end_y = tile_top

            ix_start = int(round(local_start_x / cell_size_m))
            ix_end = int(round(local_end_x / cell_size_m))
            ix_values = np.arange(ix_start, ix_end, dtype=np.int64)
            cell_count = int(ix_values.shape[0])
            if cell_count == 0:
                tile_x += scan_tile_m
                continue
            x_values = ix_values.astype(float) * cell_size_m
            x_center_values = x_values + half
            x_right_values = x_values + cell_size_m

            y = local_start_y
            while y < local_end_y:
                profile["row_checked"] += 1
                row_strip = box(local_start_x, y, local_end_x, y + cell_size_m)
                if not fish_prepared.intersects(row_strip):
                    y += cell_size_m
                    continue
                profile["row_hit"] += 1
                if profile["row_hit"] <= 20 or profile["row_hit"] % 50 == 0:
                    log(
                        f"[TRACE] build_cells_for_geometry row_hit "
                        f"row_idx={profile['row_checked']} y={y:.3f} x_span=({local_start_x:.3f}, {local_end_x:.3f})"
                    )

                mask = (
                    intersects_xy(fish_geom, x_center_values, y + half)
                    | intersects_xy(fish_geom, x_values, y)
                    | intersects_xy(fish_geom, x_right_values, y)
                    | intersects_xy(fish_geom, x_values, y + cell_size_m)
                    | intersects_xy(fish_geom, x_right_values, y + cell_size_m)
                )
                if not np.any(mask):
                    y += cell_size_m
                    continue
                profile["mask_row_hit"] += 1
                if profile["mask_row_hit"] <= 20 or profile["mask_row_hit"] % 50 == 0:
                    log(
                        f"[TRACE] build_cells_for_geometry mask_hit "
                        f"row_idx={profile['row_checked']} candidate_mask={int(np.count_nonzero(mask))}"
                    )

                mask = mask | np.r_[False, mask[:-1]] | np.r_[mask[1:], False]
                candidate_indices = np.flatnonzero(mask)
                profile["candidate_cells"] += int(candidate_indices.size)
                if candidate_indices.size and (profile["candidate_cells"] <= 50 or profile["candidate_cells"] % 200 == 0):
                    log(
                        f"[TRACE] build_cells_for_geometry candidate_cells "
                        f"count={int(candidate_indices.size)} total={profile['candidate_cells']} "
                        f"row_y={y:.3f}"
                    )
                for idx in candidate_indices:
                    profile["precise_cell_checks"] += 1
                    x = float(x_values[idx])
                    cell_geom = box(x, y, x + cell_size_m, y + cell_size_m)
                    if not fish_prepared.intersects(cell_geom):
                        continue
                    profile["precise_hits"] += 1
                    coast_hit = coast_prepared.intersects(cell_geom)
                    if not coast_hit:
                        sea_count += 1
                        continue
                    land_count += 1
                    state_name = "COASTLINE"
                    class_name = state_name
                    min_lon = lon_from_mercator(x)
                    min_lat = lat_from_mercator(y)
                    max_lon = lon_from_mercator(x + cell_size_m)
                    max_lat = lat_from_mercator(y + cell_size_m)
                    export_bbox[0] = min(export_bbox[0], min_lon)
                    export_bbox[1] = min(export_bbox[1], min_lat)
                    export_bbox[2] = max(export_bbox[2], max_lon)
                    export_bbox[3] = max(export_bbox[3], max_lat)
                    ix = int(round(x / cell_size_m))
                    iy = int(round(y / cell_size_m))
                    cell_id = f"{ix}:{iy}"
                    rows.append(
                        (
                            LAYER_NAME,
                            cell_id,
                            iy,
                            ix,
                            cell_size_m,
                            state_name,
                            class_name,
                            source_name,
                            min_lon,
                            min_lat,
                            max_lon,
                            max_lat,
                        )
                    )
                y += cell_size_m
            tile_x += scan_tile_m

    if not rows:
        log(
            f"[TRACE] build_cells_for_geometry 结束 rows=0 coast=0 off_coast=0 "
            f"tile_hit={profile['tile_hit']}/{profile['tile_checked']} "
            f"row_hit={profile['row_hit']}/{profile['row_checked']} "
            f"candidate={profile['candidate_cells']} precise={profile['precise_hits']}/{profile['precise_cell_checks']}"
        )
        return [], 0, 0, (float("inf"), float("inf"), float("-inf"), float("-inf")), profile
    log(
        f"[TRACE] build_cells_for_geometry 结束 rows={len(rows)} coast={land_count} off_coast={sea_count} "
        f"tile_hit={profile['tile_hit']}/{profile['tile_checked']} "
        f"row_hit={profile['row_hit']}/{profile['row_checked']} "
        f"candidate={profile['candidate_cells']} precise={profile['precise_hits']}/{profile['precise_cell_checks']}"
    )
    return rows, land_count, sea_count, (export_bbox[0], export_bbox[1], export_bbox[2], export_bbox[3]), profile


def process_port_group_no_resume(
    *,
    conn,
    cur,
    project_root: Path,
    prc: str,
    prc_lines: list[FishLine],
    args,
    selected_fpc: str | None,
    max_scan_cells: int | None,
) -> tuple[int, int, int, int, float]:
    prc_bboxes = [line.bbox for line in prc_lines]
    prc_bbox = (
        min(bb[0] for bb in prc_bboxes),
        min(bb[1] for bb in prc_bboxes),
        max(bb[2] for bb in prc_bboxes),
        max(bb[3] for bb in prc_bboxes),
    )
    cluster_groups = cluster_line_indices(prc_bboxes, gap_m=args.cluster_gap_m)
    if selected_fpc is not None:
        log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster 数={len(cluster_groups)}")
        log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 正在缓存整港海岸线并生成缓冲区 ...")
    else:
        log(f"[{LAYER_NAME}] PRC={prc} cluster 数={len(cluster_groups)}")
        log(f"[{LAYER_NAME}] PRC={prc} 正在缓存整县海岸线并生成缓冲区 ...")

    try:
        prc_coast_lines, prc_coast_bbox = load_coastlines_for_prc(
            project_root,
            prc,
            bbox=(
                mercator_x(prc_bbox[0]),
                mercator_y(prc_bbox[1]),
                mercator_x(prc_bbox[2]),
                mercator_y(prc_bbox[3]),
            ),
            search_margin_m=max(args.coast_buffer_m, args.coast_coarse_margin_m),
        )
    except NoCoastlineCurvesError as exc:
        if not args.skip_missing_coarse:
            raise
        status_name = "FPC_SKIPPED_NO_COASTLINE" if selected_fpc is not None else "PRC_SKIPPED_NO_COASTLINE"
        insert_progress(
            cur,
            job_name=RESUME_JOB_NAME,
            layer_name=LAYER_NAME,
            prc=prc,
            status_name=status_name,
            cluster_count=len(cluster_groups),
            inserted_cells=0,
            land_cells=0,
            sea_cells=0,
            elapsed_sec=0.0,
        )
        conn.commit()
        if selected_fpc is not None:
            log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 未找到海岸线；已按 --skip-missing-coarse 跳过。{exc}")
        else:
            log(f"[{LAYER_NAME}] PRC={prc} 未找到海岸线；已按 --skip-missing-coarse 跳过。{exc}")
        return 0, 0, 0, 0, 0.0
    if selected_fpc is not None:
        log(
            f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 海岸线缓存完成，"
            f"曲线数={len(prc_coast_lines)} bbox={prc_coast_bbox}"
        )
    else:
        log(
            f"[{LAYER_NAME}] PRC={prc} 海岸线缓存完成，"
            f"曲线数={len(prc_coast_lines)} bbox={prc_coast_bbox}"
        )

    prc_coast_geom = unary_union(prc_coast_lines)
    prc_coast_buffer = prc_coast_geom.buffer(args.coast_buffer_m, cap_style=2, join_style=2)
    prc_coast_prepared = prep(prc_coast_buffer)
    if selected_fpc is not None:
        log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 海岸缓冲缓存完成，开始处理各 part ...")
    else:
        log(f"[{LAYER_NAME}] PRC={prc} 海岸缓冲缓存完成，开始处理各 part ...")

    started_at = time.monotonic()
    prc_inserted_before = 0
    prc_land_before = 0
    prc_sea_before = 0
    inserted_total = 0
    land_total = 0
    sea_total = 0
    coarse_window_total = 0

    selected_port_rows: dict[str, tuple] | None = {} if selected_fpc is not None else None
    selected_port_coarse_windows: set[tuple[float, float, float, float]] | None = set() if selected_fpc is not None else None
    prc_profile = {
        "tile_checked": 0,
        "tile_hit": 0,
        "row_checked": 0,
        "row_hit": 0,
        "mask_row_hit": 0,
        "candidate_cells": 0,
        "precise_cell_checks": 0,
        "precise_hits": 0,
    }

    for cluster_idx, cluster in enumerate(cluster_groups):
        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} 开始，成员={len(cluster)}")
        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 正在合并渔港线 ...")

        cluster_lines = [
            LineString([(mercator_x(lon), mercator_y(lat)) for lon, lat in prc_lines[i].coords])
            for i in cluster
        ]
        merged = unary_union(cluster_lines)
        polygons = list(polygonize(merged))
        if polygons:
            fish_geom = unary_union(polygons)
            geom_mode = "polygonize"
        else:
            fish_geom = merged.buffer(args.cell_size_m, cap_style=2, join_style=2)
            geom_mode = "buffer"
        if fish_geom.is_empty:
            log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx} 为空，跳过")
            continue
        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 几何模式={geom_mode}，正在生成格网 ...")

        fish_parts = [part for part in get_parts(fish_geom) if not part.is_empty]
        if not fish_parts:
            log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 没有可拆分部件，跳过")
            continue

        for part_idx, fish_part in enumerate(fish_parts):
            fish_bbox = fish_part.bounds
            part_coarse_windows = load_coarse_windows_for_bbox(
                conn,
                layer_name=COAST_COARSE_LAYER,
                bbox_mercator=fish_bbox,
                margin_m=args.coast_coarse_margin_m,
            )
            if part_coarse_windows:
                log(
                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                    f"part={part_idx + 1}/{len(fish_parts)} 取回 coast_200m 格子={len(part_coarse_windows)}"
                )
            scan_windows = [box(*coarse_window) for coarse_window in part_coarse_windows]
            if selected_port_coarse_windows is not None:
                for coarse_window in part_coarse_windows:
                    selected_port_coarse_windows.add(tuple(float(v) for v in coarse_window))
            if not scan_windows:
                if args.allow_unbounded_fallback:
                    scan_windows = [fish_part]
                    log(
                        f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                        f"part={part_idx + 1}/{len(fish_parts)} 没有有效 coast_200m 交集，已允许回退原始范围"
                    )
                else:
                    message = (
                        f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                        f"part={part_idx + 1}/{len(fish_parts)} 没有有效 coast_200m 交集；"
                        f"fish_bbox={tuple(round(v, 3) for v in fish_bbox)} "
                        f"coarse_windows={len(part_coarse_windows)}。"
                    )
                    if args.skip_missing_coarse:
                        log(message + "已按 --skip-missing-coarse 跳过")
                        continue
                    raise RuntimeError(
                        message
                        + "按当前设计这属于异常：渔港 20m 必须以数据库 coast_200m 为底盘。"
                        + "如确需旧式整片扫描请显式加 --allow-unbounded-fallback；"
                        + "如确认允许漏算该 part，请显式加 --skip-missing-coarse。"
                    )

            log(
                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                f"part={part_idx + 1}/{len(fish_parts)} 实际扫描交集窗口={len(scan_windows)}"
            )
            if selected_fpc is not None:
                log(
                    f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                    f"part={part_idx + 1}/{len(fish_parts)} bbox={fish_bbox} 使用海岸线缓存 ..."
                )
            else:
                log(
                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                    f"part={part_idx + 1}/{len(fish_parts)} bbox={fish_bbox} 使用 part 级海岸线缓存 ..."
                )
            log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} part={part_idx + 1} 开始格网 ...")

            land_count = 0
            sea_count = 0
            bbox_lonlat = (float("inf"), float("inf"), float("-inf"), float("-inf"))
            profile = {key: 0 for key in prc_profile}
            part_inserted = 0
            batch: list[tuple] = []

            for window_idx, scan_geom in enumerate(scan_windows):
                if scan_geom.is_empty:
                    continue
                log(
                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                    f"part={part_idx + 1}/{len(fish_parts)} window={window_idx + 1}/{len(scan_windows)} "
                    f"正在生成格网 ..."
                )
                window_rows, window_land, window_sea, window_bbox, window_profile = build_cells_for_geometry(
                    scan_geom,
                    prc_coast_prepared,
                    prc=prc,
                    cell_size_m=args.cell_size_m,
                    scan_tile_m=args.cell_size_m * 10.0,
                    source_name=(
                        f"C09-06+coastline PRC={prc}"
                        + (f" FPC={selected_fpc}" if selected_fpc is not None else "")
                    ),
                    max_scan_cells=max_scan_cells,
                )
                land_count += window_land
                sea_count += window_sea
                bbox_lonlat = (
                    min(bbox_lonlat[0], window_bbox[0]),
                    min(bbox_lonlat[1], window_bbox[1]),
                    max(bbox_lonlat[2], window_bbox[2]),
                    max(bbox_lonlat[3], window_bbox[3]),
                )
                for key in prc_profile:
                    prc_profile[key] += window_profile[key]
                    profile[key] += window_profile[key]

                for row in window_rows:
                    if selected_port_rows is not None:
                        selected_port_rows[row[1]] = row
                    batch.append(row)
                    if len(batch) >= args.commit_every:
                        insert_cells(cur, batch)
                        conn.commit()
                        part_inserted += len(batch)
                        inserted_total += len(batch)
                        batch.clear()

            if batch:
                insert_cells(cur, batch)
                conn.commit()
                part_inserted += len(batch)
                inserted_total += len(batch)

            if part_inserted == 0:
                log(
                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} "
                    f"part={part_idx + 1} 没有生成 cell"
                )
                continue

            if selected_fpc is not None:
                log(
                    f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster={cluster_idx + 1} "
                    f"part={part_idx + 1} 格网完成，cell={part_inserted} coast={land_count} off_coast={sea_count} bbox={bbox_lonlat}"
                )
            else:
                log(
                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} "
                    f"part={part_idx + 1} 格网完成，cell={part_inserted} coast={land_count} off_coast={sea_count} bbox={bbox_lonlat}"
                )

            coarse_window_total += len(part_coarse_windows)
            land_total += land_count
            sea_total += sea_count

    prc_elapsed = time.monotonic() - started_at
    summary_inserted = len(selected_port_rows) if selected_port_rows is not None else inserted_total
    summary_coarse = len(selected_port_coarse_windows) if selected_port_coarse_windows is not None else coarse_window_total
    insert_progress(
        cur,
        job_name="fish_port_20m_full_mysql_resume",
        layer_name=LAYER_NAME,
        prc=prc,
        status_name="FPC_DONE" if selected_fpc is not None else "PRC_DONE",
        cluster_count=len(cluster_groups),
        inserted_cells=summary_inserted,
        land_cells=land_total,
        sea_cells=sea_total,
        elapsed_sec=prc_elapsed,
    )
    conn.commit()
    if selected_fpc is not None:
        log(
            f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 完成 cluster={len(cluster_groups)} "
            f"200x200={summary_coarse} final_20x20={summary_inserted} "
            f"coast={land_total} off_coast={sea_total} elapsed={prc_elapsed:.1f}s"
        )
    else:
        log(
            f"[{LAYER_NAME}] PRC={prc} 完成 cluster={len(cluster_groups)} "
            f"本段写入={inserted_total} coast={land_total} off_coast={sea_total} "
            f"elapsed={prc_elapsed:.1f}s"
        )
    return summary_inserted, land_total, sea_total, summary_coarse, prc_elapsed


@trace_fn("main")
def main() -> None:
    parser = argparse.ArgumentParser(description="Rebuild nationwide fish-port 20m grid into MySQL with resume support")
    parser.add_argument(
        "--fish-source",
        default=FISH_SRC,
        help="渔港原始包 C09-06.zip",
    )
    parser.add_argument(
        "--checkpoint-file",
        default=DEFAULT_CHECKPOINT,
        help="辅助断点文件（仅作记录）",
    )
    parser.add_argument(
        "--current-prc-file",
        default=DEFAULT_CURRENT_PRC_FILE,
        help="当前运行 PRC 的状态文件",
    )
    parser.add_argument(
        "--resume",
        action="store_true",
        default=True,
        help="按数据库里的断点继续，默认开启；与 --prc 同用时会被忽略",
    )
    parser.add_argument(
        "--no-resume",
        dest="resume",
        action="store_false",
        help="忽略数据库断点，从头规划但不清表",
    )
    parser.add_argument(
        "--reset-layer",
        action="store_true",
        help="先清空 fish_port_20m 的旧数据并删除断点",
    )
    parser.add_argument(
        "--prc",
        nargs="+",
        default=None,
        help="只处理指定 PRC；指定时会按该 PRC 完整重建并忽略 resume 断点",
    )
    parser.add_argument(
        "--fpc",
        default=None,
        help="只处理指定 FPC（单个渔港）；指定时会先反查所在 PRC，再只重算该港口",
    )
    parser.add_argument(
        "--cluster-gap-m",
        type=float,
        default=250.0,
        help="聚类连通判定的间隔（米）",
    )
    parser.add_argument(
        "--cell-size-m",
        type=float,
        default=CELL_SIZE_M,
        help="格网大小（米）",
    )
    parser.add_argument(
        "--coast-buffer-m",
        type=float,
        default=COAST_BUFFER_M,
        help="海岸缓冲距离（米），默认 5m，可按需要调大或调小",
    )
    parser.add_argument(
        "--fish-buffer-m",
        type=float,
        default=120.0,
        help="渔港几何外扩缓冲（米），保留兼容参数；当前默认扫描以 coast_200m 粗格为底盘",
    )
    parser.add_argument(
        "--commit-every",
        type=int,
        default=5000,
        help="每累计多少条 cell 写一次数据库并打印状态",
    )
    parser.add_argument(
        "--scan-tile-m",
        type=float,
        default=2000.0,
        help="格网扫描分块大小（米）",
    )
    parser.add_argument(
        "--coast-coarse-margin-m",
        type=float,
        default=DEFAULT_COAST_COARSE_MARGIN_M,
        help="coast_200m 粗筛底座外扩缓冲（米），默认 0m（精确按同区域粗格筛选）",
    )
    parser.add_argument(
        "--allow-unbounded-fallback",
        action="store_true",
        help="允许在没有 coast_200m 小窗时回退扫描原始渔港 part；默认关闭以避免超大范围拖僵系统",
    )
    parser.add_argument(
        "--skip-missing-coarse",
        action="store_true",
        help="遇到没有有效 coast_200m 交集的 part 时跳过；默认作为异常中止，避免静默漏算",
    )
    parser.add_argument(
        "--max-scan-cells",
        type=int,
        default=200000,
        help="单次 build_cells_for_geometry 允许估算扫描的最大 20m 格数，超过即中止；设为 0 表示不限制",
    )
    args = parser.parse_args()
    max_scan_cells = None if args.max_scan_cells <= 0 else args.max_scan_cells

    project_root = Path(__file__).resolve().parent.parent
    fish_source = Path(args.fish_source)
    if not fish_source.is_absolute():
        fish_source = project_root / fish_source
    if not fish_source.exists():
        raise SystemExit(f"missing input: {fish_source}")

    checkpoint_path = Path(args.checkpoint_file)
    if not checkpoint_path.is_absolute():
        checkpoint_path = project_root / checkpoint_path
    current_prc_path = Path(args.current_prc_file)
    if not current_prc_path.is_absolute():
        current_prc_path = project_root / current_prc_path

    ensure_database()
    conn = mysql_connect(DB_NAME)
    try:
        ensure_schema(conn)
        coast_coarse_geom, coast_coarse_bbox, coast_coarse_count = load_coarse_mask(conn, COAST_COARSE_LAYER)
        if coast_coarse_geom is not None and coast_coarse_count > 0:
            if args.coast_coarse_margin_m > 0:
                coast_coarse_geom = coast_coarse_geom.buffer(
                    args.coast_coarse_margin_m,
                    cap_style=2,
                    join_style=2,
                )
            coast_coarse_prepared = prep(coast_coarse_geom)
            log(
                f"[{LAYER_NAME}] coast_200m 粗筛底座就绪，cells={coast_coarse_count} "
                f"bbox={coast_coarse_bbox} margin={args.coast_coarse_margin_m:.1f}m"
            )
        else:
            coast_coarse_geom = None
            coast_coarse_prepared = None
            if args.allow_unbounded_fallback:
                log(f"[{LAYER_NAME}] coast_200m 粗筛底座为空，已允许回退按原始渔港范围扫描")
            elif args.skip_missing_coarse:
                log(f"[{LAYER_NAME}] coast_200m 粗筛底座为空，已允许跳过无粗筛窗口的 part")
            else:
                log(f"[{LAYER_NAME}] coast_200m 粗筛底座为空，后续遇到无粗筛窗口的 part 将作为异常中止")
        selected_prc = None if args.prc is None else {normalize_prc_token(str(x)) for x in args.prc}
        selected_fpc = normalize_fpc_token(args.fpc) if args.fpc else None
        if selected_fpc is not None and args.resume:
            log(f"[{LAYER_NAME}] 指定 FPC 模式，忽略 checkpoint 续跑，改为按指定渔港完整重建")
            args.resume = False
        if selected_prc is not None:
            if args.resume:
                log(f"[{LAYER_NAME}] 指定 PRC 模式，忽略 checkpoint 续跑，改为按指定 PRC 完整重建")
            args.resume = False
            log(f"[{LAYER_NAME}] 指定 PRC：{', '.join(sorted(selected_prc))}")
            drop_existing_prcs(conn, LAYER_NAME, selected_prc)
        if selected_fpc is not None:
            log(f"[{LAYER_NAME}] 指定 FPC：{selected_fpc}")
        if args.reset_layer:
            log(f"[{LAYER_NAME}] 清空旧数据并删除 checkpoint ...")
            drop_existing_layer(conn, LAYER_NAME)
            if checkpoint_path.exists():
                checkpoint_path.unlink()
            if current_prc_path.exists():
                current_prc_path.unlink()

        checkpoint = load_resume_state(conn, checkpoint_path, RESUME_JOB_NAME, LAYER_NAME) if args.resume else None
        completed_prcs = set(normalize_prc_list(checkpoint.get("completed_prcs", []))) if checkpoint else set()
        if checkpoint:
            log(
                f"[{LAYER_NAME}] 读取断点："
                f"last_prc={checkpoint.get('last_prc')} "
                f"completed={len(completed_prcs)} "
                f"current_prc={checkpoint.get('current_prc')}"
            )

        log(f"[{LAYER_NAME}] 开始解析渔港源：{fish_source}")
        root = load_xml(fish_source)
        point_lookup = build_point_lookup(root)

        prc_list = discover_target_prcs(root, selected_prc, selected_fpc)
        if not prc_list:
            if selected_fpc is not None:
                raise SystemExit(f"no fish-port line features selected for FPC={selected_fpc}")
            raise SystemExit("no fish-port line features selected")
        if selected_fpc is not None:
            log(f"[{LAYER_NAME}] 选中 PRC 数量：{len(prc_list)}（由 FPC 反查得到）")
            drop_existing_fpc(conn, LAYER_NAME, set(prc_list), selected_fpc)
        else:
            log(f"[{LAYER_NAME}] 选中 PRC 数量：{len(prc_list)}")

        if selected_prc is not None or selected_fpc is not None:
            with conn.cursor() as cur:
                for prc in prc_list:
                    if selected_fpc is not None:
                        prc_lines = collect_fish_lines_for_prc(root, point_lookup, prc, selected_fpc=selected_fpc)
                        if not prc_lines:
                            log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 没有渔港线要素，跳过")
                            continue
                        log(
                            f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 开始，"
                            f"线要素={len(prc_lines)}"
                        )
                        process_port_group_no_resume(
                            conn=conn,
                            cur=cur,
                            project_root=project_root,
                            prc=prc,
                            prc_lines=prc_lines,
                            args=args,
                            selected_fpc=selected_fpc,
                            max_scan_cells=max_scan_cells,
                        )
                    else:
                        prc_lines = collect_fish_lines_for_prc(root, point_lookup, prc, selected_fpc=None)
                        if not prc_lines:
                            log(f"[{LAYER_NAME}] PRC={prc} 没有渔港线要素，跳过")
                            continue
                        grouped_lines = group_fish_lines_by_fpc(prc_lines)
                        log(f"[{LAYER_NAME}] PRC={prc} 开始，FPC 组数={len(grouped_lines)}")
                        for group_fpc in sorted(grouped_lines.keys()):
                            group_lines = grouped_lines[group_fpc]
                            log(
                                f"[{LAYER_NAME}] PRC={prc} / FPC={group_fpc} 开始，"
                                f"线要素={len(group_lines)}"
                            )
                            process_port_group_no_resume(
                                conn=conn,
                                cur=cur,
                                project_root=project_root,
                                prc=prc,
                                prc_lines=group_lines,
                                args=args,
                                selected_fpc=group_fpc if group_fpc != "__unknown__" else None,
                                max_scan_cells=max_scan_cells,
                            )
            return

        total_inserted = 0
        total_land = 0
        total_sea = 0
        global_cluster_index = 0
        started_at = time.monotonic()
        checkpoint_state = checkpoint or {
            "job_name": RESUME_JOB_NAME,
            "layer_name": LAYER_NAME,
            "source_name": str(fish_source),
            "status": "running",
            "last_prc": None,
            "last_cluster_index": -1,
            "last_part_index": -1,
            "processed_clusters": 0,
            "inserted_cells": 0,
            "land_cells": 0,
            "sea_cells": 0,
            "current_prc": None,
            "current_prc_started_at": None,
            "completed_prcs": [],
            "updated_at": None,
        }

        resume_prc = checkpoint.get("current_prc") if checkpoint else None
        if resume_prc in completed_prcs:
            resume_prc = None
        resume_prc_index = prc_list.index(resume_prc) if resume_prc in prc_list else None
        if args.resume and selected_prc is None:
            pending_prcs = [
                prc
                for prc in prc_list
                if prc not in completed_prcs or prc == resume_prc
            ]
            if not pending_prcs:
                log(f"[{LAYER_NAME}] 所有 PRC 都已完成，resume 不需要继续")
                return
            prc_list = pending_prcs
            resume_prc_index = prc_list.index(resume_prc) if resume_prc in prc_list else None

        with conn.cursor() as cur:
            for prc_index, prc in enumerate(prc_list):
                if selected_prc is not None and prc not in selected_prc:
                    continue
                if args.resume and selected_prc is None and prc in completed_prcs and prc != resume_prc:
                    continue
                if resume_prc_index is not None and prc_index < resume_prc_index:
                    continue
                if resume_prc is not None and prc == resume_prc and checkpoint and checkpoint.get("last_cluster_index", -1) >= 0:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} 从断点恢复，"
                        f"last_cluster_index={checkpoint.get('last_cluster_index')}, "
                        f"last_part_index={checkpoint.get('last_part_index', -1)}"
                    )
                prc_lines = collect_fish_lines_for_prc(root, point_lookup, prc, selected_fpc=selected_fpc)
                if not prc_lines:
                    if selected_fpc is not None:
                        log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 没有渔港线要素，跳过")
                    else:
                        log(f"[{LAYER_NAME}] PRC={prc} 没有渔港线要素，跳过")
                    continue
                if selected_fpc is not None:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 开始，"
                        f"线要素={len(prc_lines)}，已累计={checkpoint_state['inserted_cells']}"
                    )
                else:
                    log(f"[{LAYER_NAME}] PRC={prc} 开始，线要素={len(prc_lines)}，已累计={checkpoint_state['inserted_cells']}")
                checkpoint_state.update(
                    {
                        "status": "running",
                        "current_prc": prc,
                        "current_prc_started_at": dt.datetime.now().isoformat(timespec="seconds"),
                        "completed_prcs": normalize_prc_list(completed_prcs),
                        "updated_at": dt.datetime.now().isoformat(timespec="seconds"),
                    }
                )
                save_checkpoint(checkpoint_path, checkpoint_state)
                write_prc_status(
                    current_prc_path,
                    {
                        "job_name": RESUME_JOB_NAME,
                        "layer_name": LAYER_NAME,
                        "status": "running",
                        "current_prc": prc,
                        "current_prc_started_at": checkpoint_state["current_prc_started_at"],
                        "last_prc": checkpoint_state.get("last_prc"),
                        "completed_prcs": normalize_prc_list(completed_prcs),
                        "updated_at": checkpoint_state["updated_at"],
                        **({"fpc": selected_fpc} if selected_fpc is not None else {}),
                    },
                )
                if selected_fpc is not None:
                    log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 准备按单港 bbox 裁切海岸线并做缓冲 ...")
                else:
                    log(f"[{LAYER_NAME}] PRC={prc} 准备按单港 bbox 裁切海岸线并做缓冲 ...")

                prc_bboxes = [line.bbox for line in prc_lines]
                prc_bbox = (
                    min(bb[0] for bb in prc_bboxes),
                    min(bb[1] for bb in prc_bboxes),
                    max(bb[2] for bb in prc_bboxes),
                    max(bb[3] for bb in prc_bboxes),
                )
                cluster_groups = cluster_line_indices(prc_bboxes, gap_m=args.cluster_gap_m)
                if selected_fpc is not None:
                    log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster 数={len(cluster_groups)}")
                else:
                    log(f"[{LAYER_NAME}] PRC={prc} cluster 数={len(cluster_groups)}")

                if selected_fpc is not None:
                    log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 正在缓存整港海岸线并生成缓冲区 ...")
                else:
                    log(f"[{LAYER_NAME}] PRC={prc} 正在缓存整县海岸线并生成缓冲区 ...")
                try:
                    prc_coast_lines, prc_coast_bbox = load_coastlines_for_prc(
                        project_root,
                        prc,
                        bbox=(
                            mercator_x(prc_bbox[0]),
                            mercator_y(prc_bbox[1]),
                            mercator_x(prc_bbox[2]),
                            mercator_y(prc_bbox[3]),
                        ),
                        search_margin_m=max(args.coast_buffer_m, args.coast_coarse_margin_m),
                    )
                except NoCoastlineCurvesError as exc:
                    if not args.skip_missing_coarse:
                        raise
                    status_name = "FPC_SKIPPED_NO_COASTLINE" if selected_fpc is not None else "PRC_SKIPPED_NO_COASTLINE"
                    insert_progress(
                        cur,
                        job_name=RESUME_JOB_NAME,
                        layer_name=LAYER_NAME,
                        prc=prc,
                        status_name=status_name,
                        cluster_count=len(cluster_groups),
                        inserted_cells=0,
                        land_cells=0,
                        sea_cells=0,
                        elapsed_sec=0.0,
                    )
                    conn.commit()
                    if selected_fpc is not None:
                        log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 未找到海岸线；已按 --skip-missing-coarse 跳过。{exc}")
                    else:
                        log(f"[{LAYER_NAME}] PRC={prc} 未找到海岸线；已按 --skip-missing-coarse 跳过。{exc}")
                    completed_prcs.add(prc)
                    continue
                if selected_fpc is not None:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 海岸线缓存完成，"
                        f"曲线数={len(prc_coast_lines)} bbox={prc_coast_bbox}"
                    )
                else:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} 海岸线缓存完成，"
                        f"曲线数={len(prc_coast_lines)} bbox={prc_coast_bbox}"
                    )
                prc_coast_geom = unary_union(prc_coast_lines)
                prc_coast_buffer = prc_coast_geom.buffer(args.coast_buffer_m, cap_style=2, join_style=2)
                prc_coast_prepared = prep(prc_coast_buffer)
                if selected_fpc is not None:
                    log(f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 海岸缓冲缓存完成，开始处理各 part ...")
                else:
                    log(f"[{LAYER_NAME}] PRC={prc} 海岸缓冲缓存完成，开始处理各 part ...")

                resume_cluster = 0
                if checkpoint and checkpoint.get("last_prc") == prc:
                    resume_cluster = int(checkpoint.get("last_cluster_index", -1)) + 1

                prc_started_at = time.monotonic()
                prc_inserted_before = checkpoint_state["inserted_cells"]
                prc_land_before = checkpoint_state["land_cells"]
                prc_sea_before = checkpoint_state["sea_cells"]
                selected_port_rows: dict[str, tuple] | None = {} if selected_fpc is not None else None
                selected_port_coarse_windows: set[tuple[float, float, float, float]] | None = set() if selected_fpc is not None else None
                prc_profile = {
                    "tile_checked": 0,
                    "tile_hit": 0,
                    "row_checked": 0,
                    "row_hit": 0,
                    "mask_row_hit": 0,
                    "candidate_cells": 0,
                    "precise_cell_checks": 0,
                    "precise_hits": 0,
                }

                for cluster_idx, cluster in enumerate(cluster_groups):
                    global_cluster_index += 1
                    if checkpoint and checkpoint.get("last_prc") == prc and cluster_idx < resume_cluster:
                        continue
                    log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} 开始，成员={len(cluster)}")
                    log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 正在合并渔港线 ...")

                    cluster_lines = [
                        LineString([(mercator_x(lon), mercator_y(lat)) for lon, lat in prc_lines[i].coords])
                        for i in cluster
                    ]
                    merged = unary_union(cluster_lines)
                    polygons = list(polygonize(merged))
                    if polygons:
                        fish_geom = unary_union(polygons)
                        geom_mode = "polygonize"
                    else:
                        fish_geom = merged.buffer(args.cell_size_m, cap_style=2, join_style=2)
                        geom_mode = "buffer"
                    if fish_geom.is_empty:
                        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx} 为空，跳过")
                        continue
                    log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 几何模式={geom_mode}，正在生成格网 ...")

                    fish_parts = [part for part in get_parts(fish_geom) if not part.is_empty]
                    if not fish_parts:
                        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} 没有可拆分部件，跳过")
                        continue

                    resume_part = 0
                    if checkpoint and checkpoint.get("last_prc") == prc and checkpoint.get("last_cluster_index") == cluster_idx:
                        resume_part = int(checkpoint.get("last_part_index", -1)) + 1

                    for part_idx, fish_part in enumerate(fish_parts):
                        if checkpoint and checkpoint.get("last_prc") == prc and checkpoint.get("last_cluster_index") == cluster_idx and part_idx < resume_part:
                            continue

                        fish_bbox = fish_part.bounds
                        scan_windows: list[object] = []
                        part_coarse_windows = load_coarse_windows_for_bbox(
                            conn,
                            layer_name=COAST_COARSE_LAYER,
                            bbox_mercator=fish_bbox,
                            margin_m=args.coast_coarse_margin_m,
                        )
                        if part_coarse_windows:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                f"part={part_idx + 1}/{len(fish_parts)} 取回 coast_200m 格子={len(part_coarse_windows)}"
                            )
                            for coarse_window in part_coarse_windows:
                                window_geom = box(*coarse_window)
                                scan_windows.append(window_geom)
                                if selected_port_coarse_windows is not None:
                                    selected_port_coarse_windows.add(tuple(float(v) for v in coarse_window))
                        if not scan_windows:
                            if args.allow_unbounded_fallback:
                                scan_windows = [fish_part]
                                log(
                                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                    f"part={part_idx + 1}/{len(fish_parts)} 没有有效 coast_200m 交集，已允许回退原始范围"
                                )
                            else:
                                message = (
                                    f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                    f"part={part_idx + 1}/{len(fish_parts)} 没有有效 coast_200m 交集；"
                                    f"fish_bbox={tuple(round(v, 3) for v in fish_bbox)} "
                                    f"coarse_windows={len(part_coarse_windows)}。"
                                )
                                if args.skip_missing_coarse:
                                    log(message + "已按 --skip-missing-coarse 跳过")
                                    continue
                                raise RuntimeError(
                                    message
                                    + "按当前设计这属于异常：渔港 20m 必须以数据库 coast_200m 为底盘。"
                                    + "如确需旧式整片扫描请显式加 --allow-unbounded-fallback；"
                                    + "如确认允许漏算该 part，请显式加 --skip-missing-coarse。"
                                )

                        log(
                            f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                            f"part={part_idx + 1}/{len(fish_parts)} 实际扫描交集窗口={len(scan_windows)}"
                        )
                        if selected_fpc is not None:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                f"part={part_idx + 1}/{len(fish_parts)} bbox={fish_bbox} 使用海岸线缓存 ..."
                            )
                        else:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                f"part={part_idx + 1}/{len(fish_parts)} bbox={fish_bbox} 使用 part 级海岸线缓存 ..."
                            )
                        coast_prepared = prc_coast_prepared
                        log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} part={part_idx + 1} 开始格网 ...")
                        land_count = 0
                        sea_count = 0
                        bbox_lonlat = (float("inf"), float("inf"), float("-inf"), float("-inf"))
                        profile = {key: 0 for key in prc_profile}
                        part_inserted = 0
                        batch: list[tuple] = []

                        for window_idx, scan_geom in enumerate(scan_windows):
                            if scan_geom.is_empty:
                                continue
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                f"part={part_idx + 1}/{len(fish_parts)} window={window_idx + 1}/{len(scan_windows)} "
                                f"正在生成格网 ..."
                            )
                            window_rows, window_land, window_sea, window_bbox, window_profile = build_cells_for_geometry(
                                scan_geom,
                                coast_prepared,
                                prc=prc,
                                cell_size_m=args.cell_size_m,
                                scan_tile_m=args.cell_size_m * 10.0,
                                source_name=(
                                    f"C09-06+coastline PRC={prc}"
                                    + (f" FPC={selected_fpc}" if selected_fpc is not None else "")
                                ),
                                max_scan_cells=max_scan_cells,
                            )
                            land_count += window_land
                            sea_count += window_sea
                            bbox_lonlat = (
                                min(bbox_lonlat[0], window_bbox[0]),
                                min(bbox_lonlat[1], window_bbox[1]),
                                max(bbox_lonlat[2], window_bbox[2]),
                                max(bbox_lonlat[3], window_bbox[3]),
                            )
                            for key in prc_profile:
                                prc_profile[key] += window_profile[key]
                                profile[key] += window_profile[key]

                            for row in window_rows:
                                if selected_port_rows is not None:
                                    selected_port_rows[row[1]] = row
                                batch.append(row)
                                if len(batch) >= args.commit_every:
                                    if selected_fpc is not None:
                                        delete_cells_by_ids(conn, LAYER_NAME, [item[1] for item in batch])
                                    insert_cells(cur, batch)
                                    conn.commit()
                                    part_inserted += len(batch)
                                    total_inserted += len(batch)
                                    checkpoint_state["inserted_cells"] += len(batch)
                                    checkpoint_state["updated_at"] = dt.datetime.now().isoformat(timespec="seconds")
                                    log(
                                        f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                        f"part={part_idx + 1}/{len(fish_parts)} "
                                        f"提交={part_inserted} 累计={checkpoint_state['inserted_cells']} "
                                        f"coast={checkpoint_state['land_cells']} off_coast={checkpoint_state['sea_cells']}"
                                    )
                                    batch.clear()

                        if batch:
                            if selected_fpc is not None:
                                delete_cells_by_ids(conn, LAYER_NAME, [item[1] for item in batch])
                            insert_cells(cur, batch)
                            conn.commit()
                            part_inserted += len(batch)
                            total_inserted += len(batch)
                            checkpoint_state["inserted_cells"] += len(batch)
                            checkpoint_state["updated_at"] = dt.datetime.now().isoformat(timespec="seconds")
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1}/{len(cluster_groups)} "
                                f"part={part_idx + 1}/{len(fish_parts)} "
                                f"提交={part_inserted} 累计={checkpoint_state['inserted_cells']} "
                                f"coast={checkpoint_state['land_cells']} off_coast={checkpoint_state['sea_cells']}"
                            )

                        if part_inserted == 0:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} "
                                f"part={part_idx + 1} 没有生成 cell"
                            )
                            checkpoint_state.update(
                                {
                                    "last_prc": prc,
                                    "last_cluster_index": cluster_idx,
                                    "last_part_index": part_idx,
                                    "processed_clusters": checkpoint_state["processed_clusters"] + 1,
                                    "updated_at": dt.datetime.now().isoformat(timespec="seconds"),
                                }
                            )
                            save_checkpoint(checkpoint_path, checkpoint_state)
                            continue
                        if selected_fpc is not None:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster={cluster_idx + 1} "
                                f"part={part_idx + 1} 格网完成，cell={part_inserted} coast={land_count} off_coast={sea_count} bbox={bbox_lonlat}"
                            )
                            log(
                                f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} cluster={cluster_idx + 1} "
                                f"part={part_idx + 1} 写库完成"
                            )
                        else:
                            log(
                                f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} "
                                f"part={part_idx + 1} 格网完成，cell={part_inserted} coast={land_count} off_coast={sea_count} bbox={bbox_lonlat}"
                            )
                            log(f"[{LAYER_NAME}] PRC={prc} cluster={cluster_idx + 1} part={part_idx + 1} 写库完成")
                        checkpoint_state.update(
                            {
                                "last_prc": prc,
                                "last_cluster_index": cluster_idx,
                                "last_part_index": part_idx,
                                "processed_clusters": checkpoint_state["processed_clusters"] + 1,
                                "land_cells": checkpoint_state["land_cells"] + land_count,
                                "sea_cells": checkpoint_state["sea_cells"] + sea_count,
                                "current_prc": prc,
                                "completed_prcs": normalize_prc_list(completed_prcs | {prc}),
                                "updated_at": dt.datetime.now().isoformat(timespec="seconds"),
                            }
                        )
                        save_checkpoint(checkpoint_path, checkpoint_state)
                        write_prc_status(
                            current_prc_path,
                            {
                                "job_name": RESUME_JOB_NAME,
                                "layer_name": LAYER_NAME,
                                "status": "running",
                                "current_prc": prc,
                                "current_prc_started_at": checkpoint_state.get("current_prc_started_at"),
                                "last_prc": prc,
                                "last_cluster_index": cluster_idx,
                                "last_part_index": part_idx,
                                "completed_prcs": normalize_prc_list(completed_prcs | {prc}),
                                "inserted_cells": checkpoint_state["inserted_cells"],
                                "land_cells": checkpoint_state["land_cells"],
                                "sea_cells": checkpoint_state["sea_cells"],
                                "updated_at": checkpoint_state["updated_at"],
                                **({"fpc": selected_fpc} if selected_fpc is not None else {}),
                            },
                        )
                        total_land += land_count
                        total_sea += sea_count

                prc_elapsed = time.monotonic() - prc_started_at
                prc_inserted_now = checkpoint_state["inserted_cells"] - prc_inserted_before
                prc_land_now = checkpoint_state["land_cells"] - prc_land_before
                prc_sea_now = checkpoint_state["sea_cells"] - prc_sea_before
                summary_inserted = len(selected_port_rows) if selected_port_rows is not None else prc_inserted_now
                summary_coarse = len(selected_port_coarse_windows) if selected_port_coarse_windows is not None else 0
                insert_progress(
                    cur,
                    job_name="fish_port_20m_full_mysql_resume",
                    layer_name=LAYER_NAME,
                    prc=prc,
                    status_name="FPC_DONE" if selected_fpc is not None else "PRC_DONE",
                    cluster_count=len(cluster_groups),
                    inserted_cells=summary_inserted,
                    land_cells=prc_land_now,
                    sea_cells=prc_sea_now,
                    elapsed_sec=prc_elapsed,
                )
                conn.commit()
                cur.execute(
                    """
                    INSERT INTO navsea_grid_import_state
                      (job_name, layer_name, source_name, checkpoint_json)
                    VALUES
                      (%s, %s, %s, %s)
                    ON DUPLICATE KEY UPDATE
                      layer_name=VALUES(layer_name),
                      source_name=VALUES(source_name),
                      checkpoint_json=VALUES(checkpoint_json)
                    """,
                    (
                        RESUME_JOB_NAME,
                        LAYER_NAME,
                        str(fish_source),
                        json.dumps(checkpoint_state, ensure_ascii=False),
                    ),
                )
                conn.commit()
                if selected_fpc is not None:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} / FPC={selected_fpc} 完成 cluster={len(cluster_groups)} "
                        f"200x200={summary_coarse} final_20x20={summary_inserted} "
                        f"coast={prc_land_now} off_coast={prc_sea_now} elapsed={prc_elapsed:.1f}s"
                    )
                else:
                    log(
                        f"[{LAYER_NAME}] PRC={prc} 完成 cluster={len(cluster_groups)} "
                        f"本段写入={prc_inserted_now} coast={prc_land_now} off_coast={prc_sea_now} "
                        f"elapsed={prc_elapsed:.1f}s"
                    )
                log(
                    f"[{LAYER_NAME}] PRC={prc} profiling "
                    f"tile={prc_profile['tile_hit']}/{prc_profile['tile_checked']} "
                    f"row={prc_profile['row_hit']}/{prc_profile['row_checked']} "
                    f"mask_row={prc_profile['mask_row_hit']} "
                    f"candidate={prc_profile['candidate_cells']} "
                    f"precise={prc_profile['precise_hits']}/{prc_profile['precise_cell_checks']}"
                )

                completed_prcs.add(prc)
                checkpoint_state["completed_prcs"] = normalize_prc_list(completed_prcs)
                checkpoint_state["status"] = "idle"
                checkpoint_state["current_prc"] = prc
                checkpoint_state["current_prc_started_at"] = checkpoint_state.get("current_prc_started_at")
                checkpoint_state["updated_at"] = dt.datetime.now().isoformat(timespec="seconds")
                save_checkpoint(checkpoint_path, checkpoint_state)
                write_prc_status(
                    current_prc_path,
                    {
                        "job_name": RESUME_JOB_NAME,
                        "layer_name": LAYER_NAME,
                        "status": "done",
                        "current_prc": prc,
                        "current_prc_started_at": checkpoint_state.get("current_prc_started_at"),
                        "completed_prcs": normalize_prc_list(completed_prcs),
                        "finished_at": checkpoint_state["updated_at"],
                        "updated_at": checkpoint_state["updated_at"],
                        **({"fpc": selected_fpc, "coarse_200m_intersections": summary_coarse, "final_20x20_cells": summary_inserted} if selected_fpc is not None else {}),
                    },
                )

                del prc_lines

            cur.execute("SELECT COUNT(*) FROM navsea_grid_cell WHERE layer_name=%s", (LAYER_NAME,))
            final_count = int(cur.fetchone()[0])
            cur.execute(
                """
                SELECT MIN(min_lon), MIN(min_lat), MAX(max_lon), MAX(max_lat)
                FROM navsea_grid_cell
                WHERE layer_name=%s
                """,
                (LAYER_NAME,),
            )
            bbox_row = cur.fetchone()
            bbox = (
                float(bbox_row[0]) if bbox_row and bbox_row[0] is not None else float("inf"),
                float(bbox_row[1]) if bbox_row and bbox_row[1] is not None else float("inf"),
                float(bbox_row[2]) if bbox_row and bbox_row[2] is not None else float("-inf"),
                float(bbox_row[3]) if bbox_row and bbox_row[3] is not None else float("-inf"),
            )
            upsert_meta(
                cur,
                LAYER_NAME,
                "全国渔港 20m 可航/不可航格",
                "coastline/C09-06.zip + navsea_japan_coast_grid.coast_200m",
                args.cell_size_m,
                final_count,
                bbox,
                None,
            )
            cur.execute(
                """
                INSERT INTO navsea_grid_import_state
                  (job_name, layer_name, source_name, checkpoint_json)
                VALUES
                  (%s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                  layer_name=VALUES(layer_name),
                  source_name=VALUES(source_name),
                  checkpoint_json=VALUES(checkpoint_json)
                """,
                (
                    RESUME_JOB_NAME,
                    LAYER_NAME,
                    str(fish_source),
                    json.dumps(checkpoint_state, ensure_ascii=False),
                ),
            )
            conn.commit()

        log(f"[{LAYER_NAME}] 完成，checkpoint={checkpoint_path}")
    finally:
        conn.close()


if __name__ == "__main__":
    main()