shanxia2.0/backend/scripts/backfill_crm_customer_license_urls.py

#!/usr/bin/env python3
"""用目标 sys_oss 将 sx 客户证照 OSS ID 回填为门户相对 URL。

场景：客户表已迁、当时 sys_oss 未齐，导致 business_license_img 等为空。

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

import argparse
import csv
import os
import sys
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]
MIGRATION_DIR = BACKEND_ROOT / "scripts" / "migration"
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))
if str(MIGRATION_DIR) not in sys.path:
    sys.path.insert(0, str(MIGRATION_DIR))

from minio_url import resolve_oss_id_csv  # noqa: E402

DEFAULT_TENANT_NO = "shanxia2.0"

FIELD_MAP = (
    # (sx_col, portal_col)
    ("business_license_images", "business_license_img"),
    ("business_permit_images", "business_permit_images"),
    ("other_cert_images", "other_cert_images"),
)


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_oss_map(target_engine: Engine, tenant_no: str) -> Dict[str, str]:
    with target_engine.connect() as conn:
        rows = conn.execute(
            text("SELECT id, url FROM sys_oss WHERE tenant_no=:t AND del_flag=0"),
            {"t": tenant_no},
        ).mappings().all()
    return {str(r["id"]): (r["url"] or "").strip() for r in rows if r["id"]}


def fetch_source_customers(source_engine: Engine) -> List[dict]:
    sql = """
    SELECT id, business_license_images, business_permit_images, other_cert_images
    FROM crm_customer
    WHERE del_flag = 0
    ORDER BY id ASC
    """
    with source_engine.connect() as conn:
        return [dict(r._mapping) for r in conn.execute(text(sql))]


def build_updates(
    source_rows: List[dict],
    oss_map: Dict[str, str],
) -> Tuple[List[dict], List[str]]:
    updates: List[dict] = []
    warnings: List[str] = []
    for row in source_rows:
        cid = legacy_id_to_str(row["id"])
        if not cid:
            continue
        if not any(str(row.get(s) or "").strip() for s, _ in FIELD_MAP):
            continue
        payload = {"id": cid}
        for src_col, tgt_col in FIELD_MAP:
            raw = row.get(src_col)
            if raw is None or str(raw).strip() == "":
                payload[tgt_col] = None
                continue
            resolved = resolve_oss_id_csv(raw, oss_map) or None
            if not resolved:
                warnings.append(f"customer#{cid} {src_col}={raw!r} 未能解析为 URL")
            payload[tgt_col] = resolved
        updates.append(payload)
    return updates, warnings


UPDATE_SQL = text(
    """
    UPDATE crm_customer SET
        business_license_img = :business_license_img,
        business_permit_images = :business_permit_images,
        other_cert_images = :other_cert_images,
        update_time = :update_time
    WHERE id = :id
    """
)


def write_report_csv(rows: List[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(
            f,
            fieldnames=[
                "customer_id",
                "business_license_img",
                "business_permit_images",
                "other_cert_images",
            ],
        )
        w.writeheader()
        for r in rows:
            w.writerow(
                {
                    "customer_id": r["id"],
                    "business_license_img": r.get("business_license_img") or "",
                    "business_permit_images": r.get("business_permit_images") or "",
                    "other_cert_images": r.get("other_cert_images") or "",
                }
            )


def main() -> int:
    parser = argparse.ArgumentParser(description="Backfill CRM customer license URLs from sys_oss")
    parser.add_argument("--source-url", default="")
    parser.add_argument("--target-url", default="")
    parser.add_argument("--tenant-no", default=DEFAULT_TENANT_NO)
    parser.add_argument("--env-file", default="")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument(
        "--report-out",
        default=str(
            BACKEND_ROOT.parent / "docs/data-migration/mappings/crm_customer_license_url_backfill.csv"
        ),
    )
    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 derive_source_url(target_url, "sx")

    source_engine = create_engine(source_url, pool_pre_ping=True)
    target_engine = create_engine(target_url, pool_pre_ping=True)

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

    oss_map = load_oss_map(target_engine, args.tenant_no)
    print(f"sys_oss 映射条数: {len(oss_map)}")

    source_rows = fetch_source_customers(source_engine)
    updates, warnings = build_updates(source_rows, oss_map)
    with_url = sum(
        1
        for u in updates
        if any(u.get(t) for _, t in FIELD_MAP)
    )
    print(f"源客户: {len(source_rows)}  待回填行: {len(updates)}  至少一证有 URL: {with_url}")

    if args.dry_run:
        print("\n预览（前 5 条）:")
        for u in updates[:5]:
            print(
                f"  id={u['id']} license={u.get('business_license_img')!r} "
                f"permit={u.get('business_permit_images')!r} other={u.get('other_cert_images')!r}"
            )
    else:
        now = datetime.now()
        applied = 0
        missing = 0
        with target_engine.begin() as conn:
            for u in updates:
                payload = {**u, "update_time": now}
                result = conn.execute(UPDATE_SQL, payload)
                if result.rowcount:
                    applied += 1
                else:
                    missing += 1
        print(f"已更新: {applied}  目标无此客户: {missing}")

    report_path = Path(args.report_out)
    write_report_csv(updates, report_path)
    print(f"回填明细: {report_path}")

    if warnings:
        print(f"警告 ({len(warnings)}):")
        for w in warnings[:20]:
            print(f"  - {w}")
        if len(warnings) > 20:
            print(f"  ... 另有 {len(warnings) - 20} 条")

    mode = "DRY-RUN" if args.dry_run else "COMMITTED"
    print(f"\n=== 客户证照 URL 回填 [{mode}] ===")
    return 0


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