shanxia2.0/backend/scripts/backfill_crm_contract_attachment_urls.py

#!/usr/bin/env python3
"""将合同 attachment_files / contract_file 中的旧 CRM 数字 ossId 替换为 MinIO 相对路径。

门户约定：业务字段里的「ossId」实际存的是 fileUrl（相对 object key）。
迁入数据仍为雪花 ossId，会导致预签名路径变成 /ruoyi/<ossId> 而无法访问。

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

import argparse
import csv
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse

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"
OSS_ID_RE = re.compile(r"^\d{10,20}$")


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(engine: Engine, tenant_no: str) -> Dict[str, str]:
    with 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"] and r["url"]}


def resolve_token(raw: Any, oss_map: Dict[str, str]) -> Tuple[str, bool]:
    """返回 (新值, 是否发生了 id→url 替换)。"""
    token = str(raw or "").strip()
    if not token:
        return "", False
    if token in oss_map:
        return oss_map[token], True
    if OSS_ID_RE.match(token):
        return token, False  # 未命中映射，保持原样并记警告
    return token, False


def rewrite_attachment_files(raw: Optional[str], oss_map: Dict[str, str], warnings: List[str], cid: str) -> Tuple[Optional[str], int]:
    if raw is None or not str(raw).strip():
        return None, 0
    value = str(raw).strip()
    replaced = 0
    if value.startswith("["):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            warnings.append(f"contract#{cid}: attachment_files JSON 无效")
            return value, 0
        if not isinstance(parsed, list):
            return value, 0
        out: List[dict] = []
        for item in parsed:
            if isinstance(item, dict):
                oid = str(item.get("ossId") or item.get("oss_id") or "").strip()
                new_id, changed = resolve_token(oid, oss_map)
                if changed:
                    replaced += 1
                elif oid and OSS_ID_RE.match(oid):
                    warnings.append(f"contract#{cid}: ossId={oid} 无 sys_oss.url")
                out.append({"ossId": new_id or oid, "category": item.get("category") or "other"})
            elif isinstance(item, str):
                new_id, changed = resolve_token(item, oss_map)
                if changed:
                    replaced += 1
                out.append({"ossId": new_id or item, "category": "other"})
        return (json.dumps(out, ensure_ascii=False) if out else None), replaced

    # 逗号分隔
    parts: List[str] = []
    for token in value.split(","):
        token = token.strip()
        if not token:
            continue
        new_id, changed = resolve_token(token, oss_map)
        if changed:
            replaced += 1
        elif OSS_ID_RE.match(token):
            warnings.append(f"contract#{cid}: token={token} 无 sys_oss.url")
        parts.append(new_id or token)
    return (",".join(parts) if parts else None), replaced


def rewrite_contract_file(raw: Optional[str], oss_map: Dict[str, str], warnings: List[str], cid: str) -> Tuple[Optional[str], int]:
    if raw is None or not str(raw).strip():
        return None, 0
    value = str(raw).strip()
    # 可能是单个 id、逗号分隔或 JSON
    if value.startswith("["):
        return rewrite_attachment_files(value, oss_map, warnings, cid)
    new_id, changed = resolve_token(value, oss_map)
    if changed:
        return new_id, 1
    if OSS_ID_RE.match(value):
        warnings.append(f"contract#{cid}: contract_file={value} 无 sys_oss.url")
    return value, 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Backfill contract attachment ossId → MinIO relative URL")
    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_contract_attachment_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
    engine = create_engine(target_url, pool_pre_ping=True)

    print(f"目标库: {urlparse(target_url).path}")
    oss_map = load_oss_map(engine, args.tenant_no)
    print(f"sys_oss 映射条数: {len(oss_map)}")

    with engine.connect() as conn:
        rows = [
            dict(r._mapping)
            for r in conn.execute(
                text(
                    """
                    SELECT id, contract_no, attachment_files, contract_file
                    FROM crm_contract
                    WHERE del_flag = 0
                      AND (
                        (attachment_files IS NOT NULL AND attachment_files <> '')
                        OR (contract_file IS NOT NULL AND contract_file <> '')
                      )
                    ORDER BY id
                    """
                )
            )
        ]

    warnings: List[str] = []
    report: List[dict] = []
    updates: List[dict] = []
    total_replaced = 0

    for row in rows:
        cid = str(row["id"])
        new_att, n1 = rewrite_attachment_files(row.get("attachment_files"), oss_map, warnings, cid)
        new_cf, n2 = rewrite_contract_file(row.get("contract_file"), oss_map, warnings, cid)
        replaced = n1 + n2
        if replaced == 0 and new_att == row.get("attachment_files") and new_cf == row.get("contract_file"):
            continue
        if replaced == 0:
            continue
        total_replaced += replaced
        updates.append(
            {
                "id": cid,
                "attachment_files": new_att,
                "contract_file": new_cf,
            }
        )
        report.append(
            {
                "contract_id": cid,
                "contract_no": row.get("contract_no") or "",
                "replaced_count": replaced,
                "old_attachment_files": (row.get("attachment_files") or "")[:200],
                "new_attachment_files": (new_att or "")[:200],
                "old_contract_file": (row.get("contract_file") or "")[:200],
                "new_contract_file": (new_cf or "")[:200],
            }
        )

    print(f"含附件合同: {len(rows)}  需更新: {len(updates)}  id→url 替换次数: {total_replaced}")

    if args.dry_run:
        print("\n预览（前 3 条）:")
        for r in report[:3]:
            print(f"  {r['contract_no']}: {r['old_attachment_files']!r}")
            print(f"    -> {r['new_attachment_files']!r}")
    else:
        now = datetime.now()
        applied = 0
        with engine.begin() as conn:
            for u in updates:
                result = conn.execute(
                    text(
                        """
                        UPDATE crm_contract
                        SET attachment_files = :attachment_files,
                            contract_file = :contract_file,
                            update_time = :update_time
                        WHERE id = :id
                        """
                    ),
                    {**u, "update_time": now},
                )
                if result.rowcount:
                    applied += 1
        print(f"已更新合同: {applied}")

    report_path = Path(args.report_out)
    report_path.parent.mkdir(parents=True, exist_ok=True)
    with report_path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(
            f,
            fieldnames=[
                "contract_id",
                "contract_no",
                "replaced_count",
                "old_attachment_files",
                "new_attachment_files",
                "old_contract_file",
                "new_contract_file",
            ],
        )
        w.writeheader()
        w.writerows(report)
    print(f"明细: {report_path}")

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

    mode = "DRY-RUN" if args.dry_run else "COMMITTED"
    print(f"\n=== 合同附件 ossId→URL 回填 [{mode}] ===")
    return 0


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