#!/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())