shanxia2.0/backend/scripts/renumber_crm_product_no.py

#!/usr/bin/env python3
"""将门户 crm_product.product_no 统一为 SXWL + 6 位流水，并写出编号映射 CSV。

设计：docs/superpowers/specs/2026-07-12-crm-product-no-renumber-design.md

用法：
  cd backend
  venv/bin/python scripts/renumber_crm_product_no.py --dry-run --env-file .env.production

  venv/bin/python scripts/renumber_crm_product_no.py --apply --env-file .env.production \\
    --mapping-out ../docs/data-migration/mappings/crm_product_no_mapping.csv
"""
from __future__ import annotations

import argparse
import csv
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine

BACKEND_ROOT = Path(__file__).resolve().parents[1]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))

DEFAULT_TENANT_NO = "shanxia2.0"
PRODUCT_NO_PREFIX = "SXWL"
PRODUCT_NO_SEQ_LEN = 6
NEW_NO_PATTERN = re.compile(rf"^{PRODUCT_NO_PREFIX}[0-9]{{{PRODUCT_NO_SEQ_LEN}}}$")
TEMP_PREFIX = "__REN_"


@dataclass
class RenumberStats:
    total: int = 0
    to_update: int = 0
    skipped: int = 0
    updated: int = 0
    warnings: List[str] = field(default_factory=list)


def load_settings_env(env_file: Optional[str]) -> None:
    if env_file:
        os.environ["ENV_FILE"] = env_file
    from dotenv import load_dotenv

    candidates: List[Path] = []
    if env_file:
        p = Path(env_file)
        candidates.append(p if p.is_absolute() else BACKEND_ROOT / p)
    candidates.extend([BACKEND_ROOT / ".env.production", BACKEND_ROOT / ".env"])
    for path in candidates:
        if path.is_file():
            load_dotenv(path, override=True)
            break


def product_id_sort_key(product_id: str) -> int:
    try:
        return int(product_id)
    except (TypeError, ValueError):
        return 0


def format_new_product_no(seq: int) -> str:
    if seq < 1 or seq > 10**PRODUCT_NO_SEQ_LEN - 1:
        raise ValueError(f"序号超出范围: {seq}")
    return f"{PRODUCT_NO_PREFIX}{seq:0{PRODUCT_NO_SEQ_LEN}d}"


def fetch_products(engine: Engine, tenant_no: str) -> List[dict]:
    sql = text(
        """
        SELECT id, product_no, product_name, category_id
        FROM crm_product
        WHERE tenant_no = :tenant_no AND del_flag = 0
        """
    )
    with engine.connect() as conn:
        rows = [dict(r._mapping) for r in conn.execute(sql, {"tenant_no": tenant_no})]
    rows.sort(key=lambda r: product_id_sort_key(str(r["id"])))
    return rows


def build_mapping(rows: List[dict], renumbered_at: Optional[str] = None) -> List[dict]:
    mapping: List[dict] = []
    for seq, row in enumerate(rows, start=1):
        pid = str(row["id"])
        old_no = (row.get("product_no") or "").strip()
        new_no = format_new_product_no(seq)
        mapping.append(
            {
                "product_id": pid,
                "old_product_no": old_no,
                "new_product_no": new_no,
                "product_name": (row.get("product_name") or "")[:200],
                "category_id": str(row.get("category_id") or ""),
                "seq": seq,
                "renumbered_at": renumbered_at or "",
            }
        )
    return mapping


def validate_mapping(mapping: List[dict], stats: RenumberStats) -> None:
    new_nos = [m["new_product_no"] for m in mapping]
    if len(new_nos) != len(set(new_nos)):
        stats.warnings.append("新编号存在重复")
    old_nos = [m["old_product_no"] for m in mapping if m["old_product_no"]]
    if len(old_nos) != len(set(old_nos)):
        stats.warnings.append("旧编号存在重复（仍将按 product_id 映射）")
    for m in mapping:
        if not NEW_NO_PATTERN.match(m["new_product_no"]):
            stats.warnings.append(f"非法新编号: {m['new_product_no']} (id={m['product_id']})")


def write_mapping_csv(path: Path, mapping: List[dict], renumbered_at: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    for row in mapping:
        if not row.get("renumbered_at"):
            row["renumbered_at"] = renumbered_at
    fieldnames = [
        "product_id",
        "old_product_no",
        "new_product_no",
        "product_name",
        "category_id",
        "seq",
        "renumbered_at",
    ]
    with path.open("w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(mapping)


def apply_renumber(engine: Engine, mapping: List[dict], tenant_no: str, stats: RenumberStats) -> None:
    """两阶段 UPDATE，避免 uk_product_no 中间态冲突。"""
    phase1_sql = text(
        """
        UPDATE crm_product
        SET product_no = :temp_no, update_time = NOW()
        WHERE id = :id AND tenant_no = :tenant_no AND del_flag = 0
        """
    )
    phase2_sql = text(
        """
        UPDATE crm_product
        SET product_no = :new_no, update_time = NOW()
        WHERE id = :id AND tenant_no = :tenant_no AND del_flag = 0
        """
    )

    with engine.begin() as conn:
        for m in mapping:
            old_no = m["old_product_no"]
            new_no = m["new_product_no"]
            if old_no == new_no:
                stats.skipped += 1
                continue
            stats.to_update += 1
            temp_no = f"{TEMP_PREFIX}{m['product_id']}"[:50]
            conn.execute(
                phase1_sql,
                {"temp_no": temp_no, "id": m["product_id"], "tenant_no": tenant_no},
            )
        for m in mapping:
            old_no = m["old_product_no"]
            new_no = m["new_product_no"]
            if old_no == new_no:
                continue
            result = conn.execute(
                phase2_sql,
                {"new_no": new_no, "id": m["product_id"], "tenant_no": tenant_no},
            )
            stats.updated += int(result.rowcount or 0)


def main() -> int:
    parser = argparse.ArgumentParser(description="CRM 产品编号统一为 SXWL + 6 位流水")
    parser.add_argument("--dry-run", action="store_true", help="仅预览，不写库")
    parser.add_argument("--apply", action="store_true", help="执行 UPDATE 并写映射")
    parser.add_argument("--env-file", default=None, help="如 .env.production")
    parser.add_argument(
        "--mapping-out",
        default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/crm_product_no_mapping.csv"),
        help="编号映射 CSV 输出路径",
    )
    parser.add_argument("--tenant-no", default=DEFAULT_TENANT_NO)
    args = parser.parse_args()

    if args.apply and args.dry_run:
        print("请只指定 --dry-run 或 --apply 之一")
        return 1
    if not args.apply and not args.dry_run:
        args.dry_run = True

    load_settings_env(args.env_file)
    from app.core.config import Settings

    settings = Settings()
    engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True)

    stats = RenumberStats()
    rows = fetch_products(engine, args.tenant_no)
    stats.total = len(rows)
    if stats.total == 0:
        print("目标库无有效产品，退出")
        return 1

    now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    mapping = build_mapping(rows, renumbered_at=now_str if args.apply else "")
    validate_mapping(mapping, stats)

    for m in mapping:
        if m["old_product_no"] != m["new_product_no"]:
            stats.to_update += 1
        else:
            stats.skipped += 1

    mode = "DRY-RUN" if args.dry_run else "APPLY"
    print(f"租户: {args.tenant_no}")
    print(f"有效产品: {stats.total}")
    print(f"需更新编号: {stats.to_update}  已符合跳过: {stats.skipped}")
    print("\n抽样映射（前 5 / 后 3）:")
    for m in mapping[:5]:
        print(f"  id={m['product_id']} {m['old_product_no']!r} -> {m['new_product_no']}")
    if len(mapping) > 8:
        print("  ...")
    for m in mapping[-3:]:
        print(f"  id={m['product_id']} {m['old_product_no']!r} -> {m['new_product_no']}")
    next_seq = stats.total + 1
    print(f"\n下一自动编号应为: {format_new_product_no(next_seq)}")

    if stats.warnings:
        print("\n警告:")
        for w in stats.warnings:
            print(f"  - {w}")

    if args.dry_run:
        print("\n=== 产品编号统一 [DRY-RUN] ===")
        print("未写库。确认后加 --apply --mapping-out ...")
        return 0

    mapping_path = Path(args.mapping_out)
    if not mapping_path.is_absolute():
        mapping_path = BACKEND_ROOT / mapping_path

    apply_stats = RenumberStats(total=stats.total)
    apply_renumber(engine, mapping, args.tenant_no, apply_stats)

    write_mapping_csv(mapping_path, mapping, now_str)

    print(f"\n映射已写入: {mapping_path}")
    print("\n=== 产品编号统一 [COMMITTED] ===")
    print(f"更新行数: {apply_stats.updated}")
    print(f"跳过（已符合）: {apply_stats.skipped}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())