feat: add segmented global CM95 pipeline

AGENTS.md

@@ -9,3 +9,9 @@
 7. 运行配置只使用 CM95_PG_* 与 CM95_SOURCE_DIR，不回退旧 MySQL 环境变量。
 8. 全国全量导入前必须先通过单 Cell、小样本、DB audit 和单 tile PBF 验证。
 9. 文档与日志不得记录数据库密码。
+10. Global CM95 使用独立 `cm95global` pipeline；输入 `/mnt/sda1/CM93_2014.zip`，数据库固定与日本基线分离为 `cm95_global_2014`。
+11. Global pipeline 稳定优先：每个 segment 独立进程、checkpoint/tag 后退出；systemd timer 启动下一段，不允许单 Python coordinator 连续驻留数天。
+12. X220 Global pipeline 只使用 CPU0/1、最多 2 PBF worker；低内存、高 load、低磁盘时必须 WAIT_RESOURCE，不领取新 segment。
+13. Global PBF 只使用 coverage/sparse 枚举，不允许 full-world aoi-complete z0-z12；日期变更线/异常超宽 coverage 必须 fail-closed。
+14. `/root/sourceserver/cm95` 与 NAS `cm95_chart` 日本基线继续保持只读参考，不得被 Global pipeline 修改。
+15. Global source quarantine 只能按精确 relative path + SHA256 + file size 绑定；必须在 shard 分配后过滤，并同时作用于 source scan / import / DB audit。禁止修改原始 ZIP、禁止为坏 cell 放宽全局 parser、禁止静默忽略 quarantine。

README.md

@@ -38,3 +38,9 @@ NAS 已预先建立数据库时，第 7 步不需要 --create-database。
 
 导入完成后先建立 FID manifest，再创建 production plan 并生成 PBF。
 mapping、tile policy、S-52、sprite/style 继续沿用参考工程已经验证的配置。
+
+## Global CM95 2014 segmented pipeline
+
+全球输入使用 `/mnt/sda1/CM93_2014.zip`，独立数据库 `cm95_global_2014`。运行器 `cm95global` 每次只执行一个 segment 后退出，由 systemd timer 启动下一段；默认固定 CPU0/1、低优先级、资源健康闸门、64 个 source/import/audit shard，以及 coverage/sparse z0-z12 PBF ledger。
+
+详见 `docs/Global_CM95_Pipeline_v1.md`。全球基础 PBF 与 bathymetry v5 保持两个独立长任务/产物。已确认损坏且无法可靠恢复的极少数 source cell 使用 path+SHA256+size 绑定的 quarantine 清单隔离；不修改原始 ZIP，也不放宽 CM93 parser。

config/global/cm95_global_2014_quarantine_v1.json

@@ -0,0 +1,24 @@
+{
+  "schema": "navsea.cm95.global-quarantine/v1",
+  "archive_sha256": "d78dc8aedbbce9686ae2fbef6fb84f860a9e4b324c9fc08201b9b40d48e48463",
+  "entries": [
+    {
+      "source_relative_path": "03900060/E/03940087.E",
+      "sha256": "6079fb710505aec80adbcbca43491198170f7b4b7cbd30b0909a2cb878bbfbac",
+      "file_size": 13692,
+      "scale_code": "E",
+      "region": "Turkey / Black Sea near northern Bosporus",
+      "reason": "feature 0 record length overruns file; no unique trustworthy suffix reconstruction"
+    },
+    {
+      "source_relative_path": "02700420/F/02870471.F",
+      "sha256": "6c4f83d8740074d0037a16ea364e9f11067e9f83c84e4413b726dc5545d2c88d",
+      "file_size": 8715,
+      "scale_code": "F",
+      "region": "FSM / Pohnpei State, Sapwuahfik (Ngatik) Atoll area",
+      "reason": "feature 41 INFORM string lacks NUL terminator and subsequent record boundary is not trustworthy",
+      "alternate_source_relative_path": "00300420/02700420/F/02870471.F",
+      "alternate_sha256": "72e1dbca14f20102a210ff0274e247ebd0423991e081460b4efd5e13d2f38a28"
+    }
+  ]
+}

config/global/cm95_global_2014_v1.json

@@ -0,0 +1,40 @@
+{
+  "schema": "navsea.cm95.global-pipeline/v1",
+  "source_archive": "/mnt/sda1/CM93_2014.zip",
+  "work_root": "/mnt/sda1/cm95-global",
+  "source_dir_name": "CM93_2014",
+  "database": "cm95_global_2014",
+  "shard_count": 64,
+  "cpu_affinity": [
+    0,
+    1
+  ],
+  "min_free_gib": 30,
+  "min_mem_available_mib": 900,
+  "max_load1": 6.0,
+  "max_database_gib": 80,
+  "segment_retry_limit": 3,
+  "pbf_workers": 2,
+  "pbf_segment_jobs": 100,
+  "pbf_max_attempts": 3,
+  "pbf_zooms": [
+    0,
+    1,
+    2,
+    3,
+    4,
+    5,
+    6,
+    7,
+    8,
+    9,
+    10,
+    11,
+    12
+  ],
+  "mapping": "config/mappings/cm95_to_navsea_v1.yaml",
+  "policy": "config/pbf/cm95_tile_policy_global_v1.yaml",
+  "s52_lookups": "resources/s52/lookup_rule_ir.json",
+  "s52_resource_contract": "resources/s52/resource_contract.json",
+  "quarantine": "config/global/cm95_global_2014_quarantine_v1.json"
+}

config/pbf/cm95_tile_policy_global_v1.yaml

@@ -0,0 +1,102 @@
+schema: navsea.cm95.tile-policy/v1
+version: v1.4-global-coverage-sparse
+status: PRODUCTION_CANDIDATE
+xyz: true
+nominal_tile_pixels: 256
+buffer_tile_pixels: 8
+default_extent: 4096
+extent_by_zoom:
+  '10': 4096
+  '11': 4096
+  '12': 1048576
+enumeration_scales:
+  '0':
+  - 20000000
+  '1':
+  - 20000000
+  '2':
+  - 20000000
+  '3':
+  - 20000000
+  '4':
+  - 20000000
+  '5':
+  - 3000000
+  '6':
+  - 3000000
+  '7':
+  - 3000000
+  '8':
+  - 1000000
+  '9':
+  - 1000000
+  '10':
+  - 200000
+  '11':
+  - 50000
+  - 100000
+  '12':
+  - 7500
+  - 20000
+  - 50000
+scale_ladders:
+  '0':
+  - 20000000
+  - 3000000
+  '1':
+  - 20000000
+  - 3000000
+  '2':
+  - 20000000
+  - 3000000
+  '3':
+  - 20000000
+  - 3000000
+  '4':
+  - 20000000
+  - 3000000
+  '5':
+  - 3000000
+  - 20000000
+  '6':
+  - 3000000
+  - 20000000
+  '7':
+  - 3000000
+  - 20000000
+  '8':
+  - 1000000
+  - 3000000
+  - 20000000
+  '9':
+  - 1000000
+  - 3000000
+  - 20000000
+  '10':
+  - 200000
+  - 1000000
+  - 3000000
+  - 20000000
+  '11':
+  - 50000
+  - 100000
+  - 200000
+  - 1000000
+  - 3000000
+  - 20000000
+  '12':
+  - 7500
+  - 20000
+  - 50000
+  - 100000
+  - 200000
+  - 1000000
+  - 3000000
+  - 20000000
+notes:
+- Global coverage-sparse: z0-z4 使用 1:20M 物化范围；scale ladder 仍允许 1:3M 回退。
+- Global production 使用 coverage/sparse z0-z12；不物化 full-world 空 tile，不新增 z13/z14。
+- z12 enumeration 加入 1:7,500（G），避免 G 级细节被生产策略直接忽略。
+- z12 scale ladder 优先 7,500，再回退 20k/50k/100k/...。
+- z0-z11 extent=4096；z12 extent=1048576，保持已验证 NavSea/MapLibre 契约。
+- enumeration_scales 仅决定物化范围；scale_ladders 仅决定 tile 内选择/回退。

deploy/systemd/cm95-global.service

@@ -0,0 +1,19 @@
+[Unit]
+Description=CM95 Global segmented pipeline - one segment
+Wants=network-online.target
+After=network-online.target
+ConditionPathExists=/mnt/sda1/CM93_2014.zip
+
+[Service]
+Type=oneshot
+User=root
+WorkingDirectory=/root/sourceserver/cm95-pg
+ExecStart=/root/sourceserver/cm95-pg/.venv/bin/cm95global --config config/global/cm95_global_2014_v1.json run-next
+CPUAffinity=0 1
+Nice=10
+IOSchedulingClass=best-effort
+IOSchedulingPriority=7
+MemoryHigh=3G
+MemoryMax=4G
+TimeoutStartSec=4h
+KillMode=control-group

deploy/systemd/cm95-global.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=CM95 Global segmented pipeline scheduler
+
+[Timer]
+OnBootSec=5min
+OnUnitInactiveSec=2min
+AccuracySec=15s
+Persistent=true
+Unit=cm95-global.service
+
+[Install]
+WantedBy=timers.target

docs/Global_CM95_Pipeline_v1.md

@@ -0,0 +1,64 @@
+# Global CM95 Pipeline v1
+
+## 目标
+
+把 `/mnt/sda1/CM93_2014.zip` 的全球 CM93/CM95 数据稳定导入独立 PostGIS 数据库，并生成 coverage/sparse 的 z0-z12 基础 PBF。
+
+稳定跑完优先于速度。流水线禁止依赖一个长寿命 Python 进程连续运行数天。
+
+## 固定资源边界
+
+- X220：2 物理核 / 4 逻辑 CPU。
+- systemd service 固定 `CPUAffinity=0 1`，完整保留 CPU2/3 所在物理核给其他工作。
+- `Nice=10`，I/O priority=7。
+- 每个 segment 为独立 oneshot 进程，默认 4 小时 hard timeout。
+- MemoryHigh=3 GiB，MemoryMax=4 GiB。
+- PBF 最多 2 worker；单次最多 100 tile jobs。
+- `/mnt/sda1` 空闲低于 30 GiB 时不领取新 segment。
+- MemAvailable 低于 900 MiB 或 load1 高于 6 时进入 WAIT_RESOURCE，不破坏 checkpoint。
+
+## 阶段
+
+1. `PRECHECK`：ZIP CRC、SHA256、cell 数、代码 fingerprint。
+2. `EXTRACT`：原子解压到 `/mnt/sda1/cm95-global/source/CM93_2014`。
+3. `INVENTORY`：冻结 source manifest 与 64-shard 分配。
+4. `QUARANTINE_VALIDATE`：验证已知坏源的 relative path / SHA256 / size；不改写原始 source。
+5. `SOURCE_SCAN_000..063`：每 shard 独立 source-only parse audit；shard 分配完成后再过滤 quarantine，避免重排。
+6. `DB_INIT`：创建独立 `cm95_global_2014` + PostGIS schema。
+7. `POSTGIS_IMPORT_000..063`：每 shard 一个全新 importer 进程，并使用同一 quarantine。
+8. `DB_AUDIT_000..063`：每 shard 独立 source↔DB audit，并使用同一 quarantine。
+9. `FID_MANIFEST`：检查 int32 FID 容量后生成全球 deterministic manifest。
+10. `COVERAGE_PLAN_Z00..Z12`：每 zoom 独立生成 sparse tile ledger。
+11. `BASE_PBF_SEGMENT_xxxxxx`：每次领取一小批 tile；结束后释放全部内存/连接。
+12. `FINAL_AUDIT`：数据库 fingerprint、PBF ledger、实际文件数与冻结资源 hash。
+13. `PIPELINE_DONE`：基础 PostGIS + sparse PBF 完成。
+
+Bathymetry v5 为第二套独立 PBF，不与基础 PBF 放在同一长任务故障域中。
+
+## Source quarantine
+
+Global source 保持原始字节不变；parser 继续严格 fail-closed。若极少数 CM93 cell 经二进制检查确认无法可靠恢复，可进入 `config/global/cm95_global_2014_quarantine_v1.json`。每条 quarantine 必须绑定 source relative path、SHA256、file size 与原因；运行时任一项不匹配立即停止，禁止按文件名模糊跳过。
+
+Quarantine 在既有 shard 取模分配完成后过滤，因此不会改变其他 cell 的 shard 归属。SOURCE_SCAN、POSTGIS_IMPORT、DB_AUDIT 三阶段必须消费同一清单；最终 release/report 必须显式记录 quarantine 数量与字节数，不得把隔离源伪装成成功解析。
+
+## Checkpoint / Tag
+
+运行状态位于 `/mnt/sda1/cm95-global/state/`。每段 PASS 写 `tags/<SEGMENT>.PASS.json`；失败尝试写 `tags/<SEGMENT>.attemptN.FAIL.json`。
+
+Tag 固定记录开始/结束机器状态、elapsed、peak RSS、磁盘、swap、DB 大小和本段业务统计。
+
+同一 segment 最多自动尝试 3 次。source/import/audit shard 连续失败 3 次后继续其他 shard，到阶段边界 fail-closed；PBF 单 tile 最多尝试 3 次。
+
+## 代码冻结
+
+`PRECHECK` 会对 parser、PostGIS/PBF/global pipeline、config、S-52 resource 与 schema 计算代码 fingerprint。后续任一 segment 发现 fingerprint 改变，自动进入 `CODE_VERSION_CHANGED / PAUSED`，禁止一套 release 静默混用两个算法版本。
+
+## 调度
+
+安装并启动：`scripts/cm95_global_install_service.sh --enable`。
+
+查看：`.venv/bin/cm95global status`、`systemctl status cm95-global.timer --no-pager`、`journalctl -u cm95-global.service`。
+
+暂停/恢复：`.venv/bin/cm95global pause` / `.venv/bin/cm95global resume`。
+
+timer 只触发 `run-next`。每次只执行一个 segment，然后整个 Python 进程退出；下一次 timer 再启动全新进程。

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
 [project.scripts]
 cm95pg = "cm93db.cli:main"
 cm95pbf = "cm95pbf.cli:main"
+cm95global = "cm95global.cli:main"
 
 [tool.setuptools]
 package-dir = {"" = "src"}

scripts/cm95_global_install_service.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$PROJECT_ROOT"
+
+install -m 0644 deploy/systemd/cm95-global.service /etc/systemd/system/cm95-global.service
+install -m 0644 deploy/systemd/cm95-global.timer /etc/systemd/system/cm95-global.timer
+mkdir -p /mnt/sda1/cm95-global/{state/tags,state/locks,logs,reports,manifests,pbf}
+
+systemctl daemon-reload
+
+if [[ "${1:-}" == "--enable" ]]; then
+  systemctl enable --now cm95-global.timer
+fi
+
+echo "Installed: /etc/systemd/system/cm95-global.service"
+echo "Installed: /etc/systemd/system/cm95-global.timer"
+echo "Enable:    systemctl enable --now cm95-global.timer"
+echo "Status:    systemctl status cm95-global.timer --no-pager"
+echo "Pipeline:  .venv/bin/cm95global status"

src/cm93db/audit.py

@@ -18,7 +18,7 @@ from .database import (
 )
 from .dictionaries import load_object_dictionary, load_attribute_dictionary
 from .geometry import build_coverages, build_feature
-from .scanner import scan_cm93, CellFile
+from .scanner import scan_cm93, CellFile, normalize_relative_path
 
 
 @dataclass(frozen=True)
@@ -52,6 +52,13 @@ class PrecisionSample:
 class AuditResult:
     source_cells: int = 0
     source_parse_failed: int = 0
+    source_features: int = 0
+    source_attributes: int = 0
+    source_points: int = 0
+    source_lines: int = 0
+    source_areas: int = 0
+    source_coverages: int = 0
+    source_sounding_points: int = 0
     db_missing_cells: int = 0
     db_not_done_cells: int = 0
     count_mismatches: int = 0
@@ -74,6 +81,13 @@ class AuditResult:
             "passed": self.passed,
             "source_cells": self.source_cells,
             "source_parse_failed": self.source_parse_failed,
+            "source_features": self.source_features,
+            "source_attributes": self.source_attributes,
+            "source_points": self.source_points,
+            "source_lines": self.source_lines,
+            "source_areas": self.source_areas,
+            "source_coverages": self.source_coverages,
+            "source_sounding_points": self.source_sounding_points,
             "db_missing_cells": self.db_missing_cells,
             "db_not_done_cells": self.db_not_done_cells,
             "count_mismatches": self.count_mismatches,
@@ -93,13 +107,23 @@ def _rel(cf: CellFile) -> str:
     return str(cf.relative_path).replace("/", "\\")
 
 
-def _select_cells(scan, limit: int | None, scales: set[str] | None, cell_name: str | None):
+def _select_cells(scan, limit: int | None, scales: set[str] | None, cell_name: str | None, shard_count: int | None = None, shard_index: int | None = None, exclude_relative_paths: set[str] | None = None):
     cells=list(scan.cells)
     if scales:
         cells=[c for c in cells if c.scale_code in scales]
     if cell_name:
         t=cell_name.lower()
         cells=[c for c in cells if c.path.name.lower()==t or c.path.stem.lower()==t]
+    if shard_count is not None:
+        if shard_count < 1:
+            raise ValueError("shard_count must be >= 1")
+        index = 0 if shard_index is None else shard_index
+        if index < 0 or index >= shard_count:
+            raise ValueError("shard_index must satisfy 0 <= index < shard_count")
+        cells=[c for i,c in enumerate(cells) if i % shard_count == index]
+    if exclude_relative_paths:
+        excluded={normalize_relative_path(x) for x in exclude_relative_paths}
+        cells=[c for c in cells if normalize_relative_path(c.relative_path) not in excluded]
     if limit is not None:
         cells=cells[:limit]
     return cells
@@ -126,13 +150,15 @@ def _source_stat(cell, cf: CellFile) -> SourceCellStat:
 
 
 def collect_source(root: Path, limit: int | None = None, scales: set[str] | None = None, cell_name: str | None = None,
-                   precision_samples: int = 0, validate_geometry: bool = False):
+                   precision_samples: int = 0, validate_geometry: bool = False,
+                   shard_count: int | None = None, shard_index: int | None = None,
+                   exclude_relative_paths: set[str] | None = None):
     scan=scan_cm93(root)
     if not scan.object_dictionary or not scan.attribute_dictionary:
         raise RuntimeError("CM93 字典未找到")
     objs={x.object_id:x for x in load_object_dictionary(scan.object_dictionary)}
     attrs={x.attribute_id:x for x in load_attribute_dictionary(scan.attribute_dictionary)}
-    cells=_select_cells(scan,limit,scales,cell_name)
+    cells=_select_cells(scan,limit,scales,cell_name,shard_count,shard_index,exclude_relative_paths)
     stats: dict[str,SourceCellStat]={}
     failures: list[str]=[]
     samples: list[PrecisionSample]=[]
@@ -193,17 +219,26 @@ def _precision_db_geom(conn, cell_id: int, sample: PrecisionSample):
 
 
 def run_audit(cfg: DbConfig, root: Path, limit: int | None = None, scales: set[str] | None = None, cell_name: str | None = None,
-              precision_samples: int = 128, tolerance_deg: float = 1e-12, validate_geometry: bool = False, source_only: bool = False, require_raw: bool = True) -> AuditResult:
+              precision_samples: int = 128, tolerance_deg: float = 1e-12, validate_geometry: bool = False, source_only: bool = False, require_raw: bool = True,
+              shard_count: int | None = None, shard_index: int | None = None, exclude_relative_paths: set[str] | None = None) -> AuditResult:
     result=AuditResult()
-    stats,failures,samples,geom_issues=collect_source(root,limit,scales,cell_name,precision_samples if not source_only else 0,validate_geometry)
+    stats,failures,samples,geom_issues=collect_source(root,limit,scales,cell_name,precision_samples if not source_only else 0,validate_geometry,shard_count,shard_index,exclude_relative_paths)
     result.source_cells=len(stats)+len(failures)
     result.source_parse_failed=len(failures)
     result.messages.extend(f"SOURCE_PARSE_FAILED {m}" for m in failures[:100])
+    result.source_features=sum(s.features for s in stats.values())
+    result.source_attributes=sum(s.attributes for s in stats.values())
+    result.source_points=sum(s.points for s in stats.values())
+    result.source_lines=sum(s.lines for s in stats.values())
+    result.source_areas=sum(s.areas for s in stats.values())
+    result.source_coverages=sum(s.coverages for s in stats.values())
+    result.source_sounding_points=sum(s.sounding_points for s in stats.values())
     result.unknown_objects=sum(s.unknown_objects for s in stats.values())
     result.unknown_attributes=sum(s.unknown_attributes for s in stats.values())
     print(f"Source cells parsed: {len(stats)}, failed: {len(failures)}")
-    print(f"Source features: {sum(s.features for s in stats.values())}")
-    print(f"Source attributes: {sum(s.attributes for s in stats.values())}")
+    print(f"Source features: {result.source_features}")
+    print(f"Source attributes: {result.source_attributes}")
+    print(f"Source sounding points: {result.source_sounding_points}")
     result.geometry_validity_issues=geom_issues
     if validate_geometry:
         print(f"Source geometry issues (Shapely validity/structure): {geom_issues}")
@@ -218,19 +253,26 @@ def run_audit(cfg: DbConfig, root: Path, limit: int | None = None, scales: set[s
             result.db_missing_cells=len(stats)
             result.messages.append("数据库中找不到该 source_root 对应 dataset")
             return result
+        rels=list(stats)
         with conn.cursor() as cur:
-            cur.execute("SELECT id,source_relative_path,import_status,source_sha256,feature_count,attribute_count FROM chart_cell WHERE dataset_id=%s", (did,))
+            cur.execute("SELECT id,source_relative_path,import_status,source_sha256,feature_count,attribute_count FROM chart_cell WHERE dataset_id=%s AND source_relative_path = ANY(%s)", (did,rels))
             rows=cur.fetchall()
         db_by_rel={str(r["source_relative_path"]):r for r in rows}
         cellids=[int(r["id"]) for r in rows]
-        fcnt=_group_counts(conn,f"SELECT cell_id,COUNT(*) n FROM chart_feature WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id")
-        acnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_feature_attribute a JOIN chart_feature f ON f.id=a.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id")
-        pcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_point_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id")
-        lcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_line_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id")
-        arcnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_area_geometry g JOIN chart_feature f ON f.id=g.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id")
-        ccnt=_group_counts(conn,f"SELECT cell_id,COUNT(*) n FROM chart_coverage WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id")
-        scnt=_group_counts(conn,f"SELECT f.cell_id,COUNT(*) n FROM chart_sounding_point s JOIN chart_feature f ON f.id=s.feature_id JOIN chart_cell c ON c.id=f.cell_id WHERE c.dataset_id={did} GROUP BY f.cell_id")
-        rawbytes=_group_counts(conn,f"SELECT cell_id,SUM(OCTET_LENGTH(raw_blob)) n FROM chart_raw_record WHERE cell_id IN (SELECT id FROM chart_cell WHERE dataset_id={did}) GROUP BY cell_id")
+        def grouped(sql: str) -> dict[int,int]:
+            if not cellids:
+                return {}
+            with conn.cursor() as cur:
+                cur.execute(sql,(cellids,))
+                return {int(r["cell_id"]):int(r["n"]) for r in cur.fetchall()}
+        fcnt=grouped("SELECT cell_id,COUNT(*) n FROM chart_feature WHERE cell_id = ANY(%s) GROUP BY cell_id")
+        acnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_feature_attribute a JOIN chart_feature f ON f.id=a.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id")
+        pcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_point_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id")
+        lcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_line_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id")
+        arcnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_area_geometry g JOIN chart_feature f ON f.id=g.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id")
+        ccnt=grouped("SELECT cell_id,COUNT(*) n FROM chart_coverage WHERE cell_id = ANY(%s) GROUP BY cell_id")
+        scnt=grouped("SELECT f.cell_id,COUNT(*) n FROM chart_sounding_point s JOIN chart_feature f ON f.id=s.feature_id WHERE f.cell_id = ANY(%s) GROUP BY f.cell_id")
+        rawbytes=grouped("SELECT cell_id,SUM(OCTET_LENGTH(raw_blob)) n FROM chart_raw_record WHERE cell_id = ANY(%s) GROUP BY cell_id")
 
         for rel,s in stats.items():
             r=db_by_rel.get(rel)
@@ -282,6 +324,9 @@ def print_result(r: AuditResult) -> None:
     print("\n=== CM93 AUDIT ===")
     print(f"source_cells          : {r.source_cells}")
     print(f"source_parse_failed   : {r.source_parse_failed}")
+    print(f"source_features       : {r.source_features}")
+    print(f"source_attributes     : {r.source_attributes}")
+    print(f"source_sounding_points: {r.source_sounding_points}")
     print(f"db_missing_cells      : {r.db_missing_cells}")
     print(f"db_not_done_cells     : {r.db_not_done_cells}")
     print(f"count_mismatches      : {r.count_mismatches}")

src/cm93db/cli.py

@@ -129,7 +129,7 @@ def cmd_import(args):
     opt=ImportOptions(root=root,limit=args.limit,scales=_scales(args.scale),cell_name=args.cell,
                       reimport=args.reimport,retry_failed=not args.skip_failed,fail_fast=args.fail_fast,
                       store_raw_tables=not args.no_raw_tables,shard_count=args.shard_count,
-                      shard_index=args.shard_index)
+                      shard_index=args.shard_index,exclude_relative_paths=set(args.exclude_relative_path or []))
     s=run_import(cfg,opt)
     print("\n=== IMPORT SUMMARY ===")
     for k in ("cells_discovered","cells_done","cells_skipped","cells_failed","features_source","features_imported","attributes_source","attributes_imported","unknown_object_count","unknown_attribute_count","geometry_issue_count","sounding_points"):
@@ -142,7 +142,8 @@ def cmd_audit(args):
     cfg=load_config() if not args.source_only else load_config()
     r=run_audit(cfg,_root(args.root),limit=args.limit,scales=_scales(args.scale),cell_name=args.cell,
                 precision_samples=args.precision_samples,tolerance_deg=args.tolerance,
-                validate_geometry=args.validate_geometry,source_only=args.source_only,require_raw=not args.allow_missing_raw)
+                validate_geometry=args.validate_geometry,source_only=args.source_only,require_raw=not args.allow_missing_raw,
+                shard_count=args.shard_count,shard_index=args.shard_index,exclude_relative_paths=set(args.exclude_relative_path or []))
     print_result(r)
     report = Path(args.report) if args.report else (project_root() / "logs" / f"cm93_audit_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
     report.parent.mkdir(parents=True,exist_ok=True)
@@ -179,6 +180,7 @@ def build_parser():
     s.add_argument("--no-raw-tables", action="store_true", help="不保存 decoded 原始 table；不建议正式母库使用")
     s.add_argument("--shard-count", type=int, help="可选并行分片总数；默认不分片")
     s.add_argument("--shard-index", type=int, help="当前分片编号，从 0 开始；需与 --shard-count 配合")
+    s.add_argument("--exclude-relative-path", action="append", default=[], help="按 source relative path 显式排除 cell；可重复，仅用于已审计 quarantine")
     s.set_defaults(func=cmd_import)
 
     s = sub.add_parser("audit", help="重新解析源 CM93 并与数据库逐 cell 审计")
@@ -191,6 +193,9 @@ def build_parser():
     s.add_argument("--validate-geometry", action="store_true", help="额外执行 Shapely validity 全量检查，较慢")
     s.add_argument("--source-only", action="store_true", help="只审计源文件解析，不查询数据库")
     s.add_argument("--allow-missing-raw", action="store_true", help="不把 raw table 未保存视为审计失败")
+    s.add_argument("--shard-count", type=int, help="可选稳定分片总数；与 import 使用相同排序/取模规则")
+    s.add_argument("--shard-index", type=int, help="当前分片编号，从 0 开始；需与 --shard-count 配合")
+    s.add_argument("--exclude-relative-path", action="append", default=[], help="按 source relative path 显式排除 cell；可重复，仅用于已审计 quarantine")
     s.add_argument("--report", help="JSON 审计报告路径；默认写入项目 logs/")
     s.set_defaults(func=cmd_audit)
     return p

src/cm93db/importer.py

@@ -21,7 +21,7 @@ from .database import (
 from psycopg.types.json import Jsonb
 from .dictionaries import load_object_dictionary, load_attribute_dictionary, ObjectDef, AttributeDef
 from .geometry import build_coverages, build_feature, BuiltFeature
-from .scanner import scan_cm93, CellFile
+from .scanner import scan_cm93, CellFile, normalize_relative_path
 
 RAW_CHUNK = 512 * 1024
 
@@ -50,6 +50,7 @@ class ImportOptions:
     store_raw_tables: bool = True
     shard_count: int | None = None
     shard_index: int | None = None
+    exclude_relative_paths: set[str] | None = None
 
 
 @dataclass
@@ -861,6 +862,13 @@ def _selected_cells(
             raise ValueError("shard_index must satisfy 0 <= index < shard_count")
         cells = [c for i, c in enumerate(cells) if i % opt.shard_count == shard_index]
 
+    if opt.exclude_relative_paths:
+        excluded = {normalize_relative_path(x) for x in opt.exclude_relative_paths}
+        cells = [
+            c for c in cells
+            if normalize_relative_path(c.relative_path) not in excluded
+        ]
+
     if opt.limit is not None:
         cells = cells[:opt.limit]
 

src/cm93db/scanner.py

@@ -28,6 +28,11 @@ class ScanResult:
     object_dictionary: Path | None
     attribute_dictionary: Path | None
 
+def normalize_relative_path(value: str | Path) -> str:
+    """Canonical source-relative path used only for selection/exclusion matching."""
+    return str(value).replace("\\", "/").lstrip("./").casefold()
+
+
 def _find_case_insensitive(root: Path, names: set[str]) -> Path | None:
     lowers = {n.lower() for n in names}
     for p in root.rglob("*"):

src/cm95global/__init__.py

@@ -0,0 +1,2 @@
+__version__ = "1.0.0"
+PIPELINE_CONTRACT_VERSION = "cm95-global-pipeline/1.0.0"

src/cm95global/cli.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from .config import load_global_config
+from .pipeline import GlobalPipeline
+from .state import PipelineLock
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="cm95global",
+        description="Segmented, restart-safe Global CM95 -> PostGIS -> sparse PBF pipeline",
+    )
+    p.add_argument(
+        "--config",
+        default="config/global/cm95_global_2014_v1.json",
+        help="global pipeline JSON config",
+    )
+    sub = p.add_subparsers(dest="command", required=True)
+    sub.add_parser("run-next", help="run exactly one segment, write checkpoint, then exit")
+    sub.add_parser("status", help="show current persistent pipeline status")
+    sub.add_parser("pause", help="pause future timer-driven segments")
+    r = sub.add_parser("resume", help="resume future timer-driven segments")
+    r.add_argument(
+        "--retry-exhausted",
+        action="store_true",
+        help="clear exhausted segment retry markers",
+    )
+    return p
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    cfg = load_global_config(Path(args.config))
+    pipe = GlobalPipeline(cfg)
+
+    if args.command == "status":
+        print(json.dumps(pipe.status(), ensure_ascii=False, indent=2))
+        return
+    if args.command == "pause":
+        pipe.pause("manual pause")
+        print(json.dumps(pipe.status(), ensure_ascii=False, indent=2))
+        return
+    if args.command == "resume":
+        pipe.resume(retry_exhausted=bool(args.retry_exhausted))
+        print(json.dumps(pipe.status(), ensure_ascii=False, indent=2))
+        return
+
+    lock = cfg.state_root / "locks" / "pipeline.lock"
+    with PipelineLock(lock):
+        result = pipe.run_next()
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

src/cm95global/config.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class GlobalConfig:
+    project_root: Path
+    config_path: Path
+    source_archive: Path
+    work_root: Path
+    source_dir_name: str
+    database: str
+    shard_count: int
+    cpu_affinity: tuple[int, ...]
+    min_free_gib: float
+    min_mem_available_mib: int
+    max_load1: float
+    max_database_gib: float
+    segment_retry_limit: int
+    pbf_workers: int
+    pbf_segment_jobs: int
+    pbf_max_attempts: int
+    pbf_zooms: tuple[int, ...]
+    mapping: Path
+    policy: Path
+    s52_lookups: Path
+    s52_resource_contract: Path
+    quarantine: Path
+
+    @property
+    def source_root(self) -> Path:
+        return self.work_root / "source" / self.source_dir_name
+
+    @property
+    def state_root(self) -> Path:
+        return self.work_root / "state"
+
+    @property
+    def tag_root(self) -> Path:
+        return self.state_root / "tags"
+
+    @property
+    def log_root(self) -> Path:
+        return self.work_root / "logs"
+
+    @property
+    def report_root(self) -> Path:
+        return self.work_root / "reports"
+
+    @property
+    def manifest_root(self) -> Path:
+        return self.work_root / "manifests"
+
+    @property
+    def fid_manifest(self) -> Path:
+        return self.manifest_root / "cm95_global_fid_manifest_v1.sqlite"
+
+    @property
+    def pbf_root(self) -> Path:
+        return self.work_root / "pbf"
+
+    @property
+    def pbf_ledger(self) -> Path:
+        return self.state_root / "global_pbf_ledger_v1.sqlite"
+
+    @property
+    def source_manifest(self) -> Path:
+        return self.state_root / "source_manifest_v1.sqlite"
+
+
+def _resolve(project_root: Path, raw: str) -> Path:
+    p = Path(raw)
+    return p if p.is_absolute() else project_root / p
+
+
+def load_global_config(path: Path) -> GlobalConfig:
+    path = Path(path).resolve()
+    project_root = path.parents[2]
+    obj = json.loads(path.read_text(encoding="utf-8"))
+    if obj.get("schema") != "navsea.cm95.global-pipeline/v1":
+        raise RuntimeError(f"unsupported global config schema: {obj.get('schema')}")
+    return GlobalConfig(
+        project_root=project_root,
+        config_path=path,
+        source_archive=Path(obj["source_archive"]),
+        work_root=Path(obj["work_root"]),
+        source_dir_name=str(obj["source_dir_name"]),
+        database=str(obj["database"]),
+        shard_count=int(obj["shard_count"]),
+        cpu_affinity=tuple(int(x) for x in obj["cpu_affinity"]),
+        min_free_gib=float(obj["min_free_gib"]),
+        min_mem_available_mib=int(obj["min_mem_available_mib"]),
+        max_load1=float(obj["max_load1"]),
+        max_database_gib=float(obj["max_database_gib"]),
+        segment_retry_limit=int(obj["segment_retry_limit"]),
+        pbf_workers=int(obj["pbf_workers"]),
+        pbf_segment_jobs=int(obj["pbf_segment_jobs"]),
+        pbf_max_attempts=int(obj["pbf_max_attempts"]),
+        pbf_zooms=tuple(int(x) for x in obj["pbf_zooms"]),
+        mapping=_resolve(project_root, obj["mapping"]),
+        policy=_resolve(project_root, obj["policy"]),
+        s52_lookups=_resolve(project_root, obj["s52_lookups"]),
+        s52_resource_contract=_resolve(project_root, obj["s52_resource_contract"]),
+        quarantine=_resolve(project_root, obj["quarantine"]),
+    )

src/cm95global/state.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import resource
+import shutil
+from typing import Any
+
+
+STATE_SCHEMA = "navsea.cm95.global-state/v1"
+TAG_SCHEMA = "navsea.cm95.global-segment-tag/v1"
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def atomic_json(path: Path, obj: dict[str, Any]) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
+    os.replace(tmp, path)
+
+
+def load_json(path: Path, default: dict[str, Any] | None = None) -> dict[str, Any]:
+    path = Path(path)
+    if not path.exists():
+        return dict(default or {})
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def sha256_file(path: Path, block: int = 4 * 1024 * 1024) -> str:
+    h = hashlib.sha256()
+    with Path(path).open("rb") as f:
+        for chunk in iter(lambda: f.read(block), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def code_fingerprint(project_root: Path) -> str:
+    root = Path(project_root)
+    items: list[Path] = []
+    for rel in ("src/cm93db", "src/cm95pbf", "src/cm95global", "config", "resources/s52", "sql"):
+        base = root / rel
+        if base.is_dir():
+            items.extend(p for p in base.rglob("*") if p.is_file() and "__pycache__" not in p.parts)
+        elif base.is_file():
+            items.append(base)
+    for rel in ("pyproject.toml", "requirements.txt"):
+        p = root / rel
+        if p.is_file():
+            items.append(p)
+    h = hashlib.sha256()
+    for p in sorted(set(items), key=lambda x: str(x.relative_to(root))):
+        rel = str(p.relative_to(root)).replace(os.sep, "/")
+        h.update(rel.encode())
+        h.update(b"\0")
+        h.update(str(p.stat().st_size).encode())
+        h.update(b"\0")
+        with p.open("rb") as f:
+            for chunk in iter(lambda: f.read(1024 * 1024), b""):
+                h.update(chunk)
+    return h.hexdigest()
+
+
+def _meminfo_kib() -> dict[str, int]:
+    out: dict[str, int] = {}
+    for line in Path("/proc/meminfo").read_text().splitlines():
+        key, rest = line.split(":", 1)
+        value = rest.strip().split()[0]
+        out[key] = int(value)
+    return out
+
+
+def machine_snapshot(work_root: Path) -> dict[str, Any]:
+    mem = _meminfo_kib()
+    target = Path(work_root)
+    probe = target if target.exists() else target.parent
+    probe.mkdir(parents=True, exist_ok=True)
+    disk = shutil.disk_usage(probe)
+    load1, load5, load15 = os.getloadavg()
+    return {
+        "at": utc_now(),
+        "load": [load1, load5, load15],
+        "mem_available_mib": round(mem.get("MemAvailable", 0) / 1024, 1),
+        "swap_used_mib": round((mem.get("SwapTotal", 0) - mem.get("SwapFree", 0)) / 1024, 1),
+        "disk_total_gib": round(disk.total / 1024**3, 2),
+        "disk_used_gib": round(disk.used / 1024**3, 2),
+        "disk_free_gib": round(disk.free / 1024**3, 2),
+    }
+
+
+def peak_rss_mib() -> float:
+    self_kib = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    child_kib = resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss
+    return round(max(self_kib, child_kib) / 1024, 1)
+
+
+class PipelineLock:
+    def __init__(self, path: Path):
+        self.path = Path(path)
+        self.fd = None
+
+    def __enter__(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.fd = self.path.open("w")
+        try:
+            fcntl.flock(self.fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            self.fd.close()
+            raise RuntimeError("global pipeline already running") from exc
+        self.fd.write(f"{os.getpid()}\n")
+        self.fd.flush()
+        return self
+
+
+    def __exit__(self, exc_type, exc, tb):
+        if self.fd is not None:
+            try:
+                fcntl.flock(self.fd.fileno(), fcntl.LOCK_UN)
+            finally:
+                self.fd.close()
+
+
+def default_state() -> dict[str, Any]:
+    return {
+        "schema": STATE_SCHEMA,
+        "created_at": utc_now(),
+        "updated_at": utc_now(),
+        "paused": False,
+        "pause_reason": None,
+        "baseline_code_fingerprint": None,
+        "archive_sha256": None,
+        "attempts": {},
+        "exhausted": [],
+        "pbf_segment_seq": 0,
+        "last_action": None,
+        "last_result": None,
+    }
+
+
+def write_tag(tag_root: Path, key: str, result: str, payload: dict[str, Any]) -> Path:
+    safe = key.replace("/", "_").replace(" ", "_")
+    path = Path(tag_root) / f"{safe}.{result}.json"
+    atomic_json(path, {
+        "schema": TAG_SCHEMA,
+        "key": key,
+        "result": result,
+        "written_at": utc_now(),
+        **payload,
+    })
+    return path

src/cm95pbf/production.py

@@ -195,10 +195,9 @@ def _connect(db: DbConfig):
 
 def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]):
     coverage_wkb_sql = as_wkb_sql("geom")
-    placeholders = ",".join(["%s"] * len(scales))
     with conn.cursor() as cur:
         cur.execute(
-            f"""
+            """
             SELECT
                 id,
                 native_scale,
@@ -208,10 +207,10 @@ def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]):
                 bbox_max_lat
             FROM chart_cell
             WHERE dataset_id=%s
-              AND native_scale IN ({placeholders})
+              AND native_scale = ANY(%s)
             ORDER BY native_scale, id
             """,
-            (dataset_id, *scales),
+            (dataset_id, list(scales)),
         )
         cells = list(cur.fetchall())
 
@@ -219,17 +218,16 @@ def _fetch_plan_geometries(conn, dataset_id: int, scales: list[int]):
         return []
 
     cell_ids = [int(c["id"]) for c in cells]
-    ph = ",".join(["%s"] * len(cell_ids))
     with conn.cursor() as cur:
         cur.execute(
             f"""
             SELECT cell_id, {coverage_wkb_sql} AS geom_wkb
             FROM chart_coverage
             WHERE dataset_id=%s
-              AND cell_id IN ({ph})
+              AND cell_id = ANY(%s)
             ORDER BY cell_id, id
             """,
-            (dataset_id, *cell_ids),
+            (dataset_id, cell_ids),
         )
         coverage_rows = list(cur.fetchall())