shanxia2.0/backend/scripts/migrate_crm_payment_invoice.py

#!/usr/bin/env python3
"""将 sx 库 CRM 回款/开票相关表迁入 shanxia2.0。

门户侧当前仅有 crm_invoice_info 表；源库回款/发票主表及明细均为空，跳过。
实际迁移：crm_invoice_info（客户开票信息子表）。

用法：
  cd backend
  venv/bin/python scripts/migrate_crm_payment_invoice.py --dry-run --env-file .env.production
  venv/bin/python scripts/migrate_crm_payment_invoice.py --env-file .env.production
"""
from __future__ import annotations

import argparse
import csv
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse, urlunparse

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"
CRM_ADMIN_SOURCE_USER_ID = 1

# 源库存在但门户暂无对应业务表，或源库无数据
SKIPPED_TABLES = (
    "crm_payment",
    "crm_payment_plan",
    "crm_payment_collection",
    "crm_invoice",
    "crm_invoice_item",
)


@dataclass
class StepStats:
    source_count: int = 0
    inserted: int = 0
    skipped: int = 0


@dataclass
class MigrationStats:
    steps: Dict[str, StepStats] = field(default_factory=dict)
    warnings: List[str] = field(default_factory=list)

    def step(self, name: str) -> StepStats:
        if name not in self.steps:
            self.steps[name] = StepStats()
        return self.steps[name]


def legacy_id_to_str(legacy_id: Any) -> Optional[str]:
    if legacy_id is None or str(legacy_id).strip() == "":
        return None
    try:
        return str(int(legacy_id))
    except (TypeError, ValueError):
        return str(legacy_id).strip()


def derive_source_url(target_url: str, source_db: str = "sx") -> str:
    return urlunparse(urlparse(target_url)._replace(path=f"/{source_db}"))


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 load_runtime_settings():
    from app.core.config import Settings

    return Settings()


def load_user_mapping(path: Path) -> Tuple[Dict[int, str], Optional[str]]:
    crm_map: Dict[int, str] = {}
    admin_id: Optional[str] = None
    with path.open(encoding="utf-8-sig", newline="") as f:
        for row in csv.DictReader(f):
            new_id = (row.get("new_user_id") or "").strip()
            user_no = (row.get("user_no") or "").strip()
            crm_raw = (row.get("crm_user_id") or "").strip()
            if user_no == "admin" and new_id:
                admin_id = new_id
            if crm_raw and new_id:
                try:
                    crm_map[int(crm_raw)] = new_id
                except ValueError:
                    pass
    return crm_map, admin_id


def fetch_portal_admin_id(target_engine: Engine) -> Optional[str]:
    with target_engine.connect() as conn:
        row = conn.execute(
            text("SELECT id FROM sys_user WHERE user_no='admin' AND del_flag=0 LIMIT 1")
        ).first()
        return str(row[0]) if row and row[0] else None


def resolve_portal_user_id(
    source_user_id: Any,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
    label: str,
) -> Optional[str]:
    if source_user_id is None or str(source_user_id).strip() == "":
        return None
    try:
        uid = int(source_user_id)
    except (TypeError, ValueError):
        stats.warnings.append(f"{label}: 非法用户 id {source_user_id!r}")
        return None
    if uid == CRM_ADMIN_SOURCE_USER_ID:
        return admin_id
    mapped = crm_user_map.get(uid)
    if not mapped:
        stats.warnings.append(f"{label}: crm_user_id={uid} 无映射")
    return mapped


def _str_or_none(value: Any, max_len: int = 0) -> Optional[str]:
    if value is None:
        return None
    s = str(value).strip()
    if not s:
        return None
    return s[:max_len] if max_len else s


def fetch_source_rows(engine: Engine, table: str, include_deleted: bool) -> List[dict]:
    where = "" if include_deleted else "WHERE del_flag = 0"
    sql = f"SELECT * FROM `{table}` {where} ORDER BY id ASC"
    with engine.connect() as conn:
        return [dict(r._mapping) for r in conn.execute(text(sql))]


def count_source_rows(engine: Engine, table: str) -> int:
    with engine.connect() as conn:
        try:
            return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}` WHERE del_flag=0")).scalar_one())
        except Exception:
            return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}`")).scalar_one())


def target_table_exists(engine: Engine, table: str) -> bool:
    with engine.connect() as conn:
        row = conn.execute(text("SHOW TABLES LIKE :t"), {"t": table}).first()
        return row is not None


def target_row_exists(conn, table: str, row_id: str) -> bool:
    return int(conn.execute(text(f"SELECT COUNT(*) FROM `{table}` WHERE id=:id"), {"id": row_id}).scalar_one()) > 0


def transform_invoice_info(
    row: dict,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
) -> Optional[dict]:
    rid = legacy_id_to_str(row["id"])
    label = f"invoice_info#{rid}"
    customer_id = legacy_id_to_str(row.get("customer_id"))
    if not customer_id:
        stats.warnings.append(f"{label}: 缺少 customer_id，已跳过")
        return None

    company_name = _str_or_none(row.get("company_name"), 200)
    if not company_name:
        stats.warnings.append(f"{label}: company_name 为空，已跳过")
        return None

    def map_user(field_val: Any) -> Optional[str]:
        return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)

    is_default = (str(row.get("is_default") or "0").strip() or "0")[:1]
    if is_default not in ("0", "1"):
        is_default = "1" if is_default in ("true", "True", "Y", "y") else "0"

    return {
        "id": rid,
        "customer_id": customer_id,
        "company_name": company_name,
        "tax_no": None,
        "company_address": _str_or_none(row.get("company_address"), 500),
        "legal_representative": _str_or_none(row.get("legal_representative"), 100),
        "agent": _str_or_none(row.get("agent"), 100),
        "phone_fax": _str_or_none(row.get("phone_fax"), 100),
        "bank_name": _str_or_none(row.get("bank_name"), 200),
        "bank_account": _str_or_none(row.get("bank_account"), 100),
        "is_default": is_default,
        "tenant_no": tenant_no,
        "order_num": 0,
        "remark": _str_or_none(row.get("remark"), 512),
        "create_by": map_user(row.get("create_by")),
        "create_dept_id": "",
        "create_time": row.get("create_time") or datetime.now(),
        "update_by": map_user(row.get("update_by")),
        "update_time": row.get("update_time") or row.get("create_time") or datetime.now(),
        "del_flag": int(row.get("del_flag") or 0),
    }


INSERT_INVOICE_INFO_SQL = text(
    """
    INSERT INTO crm_invoice_info (
        id, customer_id, company_name, tax_no, company_address, legal_representative,
        agent, phone_fax, bank_name, bank_account, is_default,
        tenant_no, order_num, remark, create_by, create_dept_id,
        create_time, update_by, update_time, del_flag
    ) VALUES (
        :id, :customer_id, :company_name, :tax_no, :company_address, :legal_representative,
        :agent, :phone_fax, :bank_name, :bank_account, :is_default,
        :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
        :create_time, :update_by, :update_time, :del_flag
    )
    """
)


def write_mapping_csv(rows: List[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    migrated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["source_id", "new_id", "migrated_at"])
        w.writeheader()
        for row in rows:
            sid = legacy_id_to_str(row["id"])
            w.writerow({"source_id": sid, "new_id": sid, "migrated_at": migrated_at})


def report_skipped_tables(source_engine: Engine, target_engine: Engine, stats: MigrationStats) -> None:
    print("\n>> 跳过表（门户无表或源库无数据）:")
    for table in SKIPPED_TABLES:
        source_count = count_source_rows(source_engine, table)
        target_exists = target_table_exists(target_engine, table)
        step = stats.step(table)
        step.source_count = source_count
        if source_count == 0:
            print(f"  {table}: 源库 0 行，跳过")
        elif not target_exists:
            print(f"  {table}: 源库 {source_count} 行，门户无对应表，跳过")
            stats.warnings.append(f"{table}: 源库有 {source_count} 行但门户无表，未迁移")
        else:
            print(f"  {table}: 源={source_count} 目标表存在（本脚本未实现，需单独处理）")


def migrate_invoice_info(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    crm_user_map: Dict[int, str],
    admin_id: Optional[str],
    stats: MigrationStats,
    dry_run: bool,
    mapping_dir: Path,
) -> None:
    table = "crm_invoice_info"
    step = stats.step(table)
    source_rows = fetch_source_rows(source_engine, table, include_deleted=False)
    step.source_count = len(source_rows)

    payloads: List[dict] = []
    for row in source_rows:
        payload = transform_invoice_info(row, tenant_no, crm_user_map, admin_id, stats)
        if payload:
            payloads.append(payload)
        else:
            step.skipped += 1

    if dry_run:
        step.inserted = len(payloads)
        print(f"{table}: 源={step.source_count} 可插入={step.inserted} 跳过={step.skipped}")
        return

    with target_engine.begin() as conn:
        for payload in payloads:
            rid = payload["id"]
            if target_row_exists(conn, table, rid):
                step.skipped += 1
            else:
                conn.execute(INSERT_INVOICE_INFO_SQL, payload)
                step.inserted += 1

    write_mapping_csv(source_rows, mapping_dir / "crm_invoice_info_id_mapping.csv")
    print(f"{table}: 源={step.source_count} 插入={step.inserted} 跳过={step.skipped}")


def print_report(stats: MigrationStats, dry_run: bool) -> None:
    mode = "DRY-RUN" if dry_run else "COMMITTED"
    print(f"\n=== CRM 回款/开票迁移 [{mode}] ===")
    for name in (*SKIPPED_TABLES, "crm_invoice_info"):
        s = stats.step(name)
        if s.source_count or s.inserted or s.skipped:
            print(f"{name}: 源={s.source_count} 插入={s.inserted} 跳过={s.skipped}")
    if stats.warnings:
        print(f"警告 ({len(stats.warnings)}):")
        for w in stats.warnings[:20]:
            print(f"  - {w}")
        if len(stats.warnings) > 20:
            print(f"  ... 另有 {len(stats.warnings) - 20} 条")


def run_migration(
    source_engine: Engine,
    target_engine: Engine,
    *,
    tenant_no: str,
    user_mapping_path: Path,
    mapping_dir: Path,
    dry_run: bool,
) -> MigrationStats:
    stats = MigrationStats()
    crm_user_map, admin_id = load_user_mapping(user_mapping_path)
    if not admin_id:
        admin_id = fetch_portal_admin_id(target_engine)

    report_skipped_tables(source_engine, target_engine, stats)
    print("\n>> 迁移 crm_invoice_info ...")
    migrate_invoice_info(
        source_engine,
        target_engine,
        tenant_no=tenant_no,
        crm_user_map=crm_user_map,
        admin_id=admin_id,
        stats=stats,
        dry_run=dry_run,
        mapping_dir=mapping_dir,
    )
    print_report(stats, dry_run)
    return stats


def main() -> int:
    parser = argparse.ArgumentParser(description="Migrate CRM payment/invoice bundle from sx to portal")
    parser.add_argument("--source-url", default="")
    parser.add_argument("--target-url", default="")
    parser.add_argument("--tenant-no", default=DEFAULT_TENANT_NO)
    parser.add_argument(
        "--user-mapping",
        default=str(BACKEND_ROOT.parent / "docs/data-migration/mappings/user_mapping_result.csv"),
    )
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--env-file", default=os.environ.get("ENV_FILE", ""))
    args = parser.parse_args()

    load_settings_env(args.env_file or None)
    settings = load_runtime_settings()
    target_url = args.target_url or settings.DATABASE_URL
    source_url = args.source_url or os.environ.get("SX_SOURCE_DATABASE_URL") or derive_source_url(target_url)

    print(f"源库: {urlparse(source_url).path}")
    print(f"目标库: {urlparse(target_url).path}")

    run_migration(
        create_engine(source_url, pool_pre_ping=True),
        create_engine(target_url, pool_pre_ping=True),
        tenant_no=args.tenant_no,
        user_mapping_path=Path(args.user_mapping),
        mapping_dir=BACKEND_ROOT.parent / "docs/data-migration/mappings",
        dry_run=args.dry_run,
    )
    return 0


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