zkxw/backend/scripts/run_sop_visible_deploy.py

#!/usr/bin/env python3
"""在测试/本地库执行 SOP 精简展示部署 SQL。"""
from __future__ import annotations

import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "backend"))

from sqlalchemy import create_engine, text  # noqa: E402

try:
    from app.core.config import settings  # noqa: E402
except Exception as exc:  # pragma: no cover
    raise SystemExit(f"无法加载 backend 配置: {exc}") from exc

SQL_DIR = ROOT / "docs" / "sql"

SOP_CODES = [
    "SOP-CC-001-E01-01",
    "SOP-CC-002-E01-01",
    "SOP-CC-003-E01-01",
    "SOP-CC-006-E01-01",
    "SOP-SC-003-E01-01",
    "SOP-SC-004-E01-01",
    "SOP-SC-005-E01-01",
    "SOP-SC-006-E01-01",
    "SOP-SC-007-E01-01",
    "SOP-SC-008-E01-01",
    "SOP-SC-009-E01-01",
    "SOP-SC-010-E01-01",
    "SOP-SC-011-E01-01",
    "SOP-SC-012-E01-01",
    "SOP-SC-013-E01-01",
    "SOP-SC-014-E01-01",
]

UPDATE_STMT_RE = re.compile(r"\bUPDATE\b.+?;", re.IGNORECASE | re.DOTALL)


def _strip_sql_comments(sql: str) -> str:
    lines = []
    for line in sql.splitlines():
        if line.strip().startswith("--"):
            continue
        lines.append(line)
    return "\n".join(lines)


def _run_seed_file(conn, path: Path) -> int:
    sql = path.read_text(encoding="utf-8").strip()
    if not sql:
        return 0
    result = conn.execute(text(sql))
    return result.rowcount or 0


def _run_sync_file(conn, path: Path) -> int:
    raw = _strip_sql_comments(path.read_text(encoding="utf-8"))
    total = 0
    for stmt in UPDATE_STMT_RE.findall(raw):
        result = conn.execute(text(stmt.strip()))
        total += result.rowcount or 0
    return total


def main() -> None:
    engine = create_engine(settings.DATABASE_URL)
    total = 0
    with engine.begin() as conn:
        for code in SOP_CODES:
            short = code.replace("SOP-", "").replace("-E01-01", "").lower()
            seed = SQL_DIR / f"2026-08-25-seed-sop-{short}-content.sql"
            sync = SQL_DIR / f"2026-08-25-sync-sop-{short}-content-to-routes-tasks.sql"
            n1 = _run_seed_file(conn, seed)
            n2 = _run_sync_file(conn, sync) if sync.exists() else 0
            total += n1 + n2
            print(f"{code}: process={n1}, sync={n2}")
    print(f"done, affected rows (approx): {total}")


if __name__ == "__main__":
    main()