841 lines
35 KiB
Python
841 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""将 sx 库 CRM 合同相关表迁入 shanxia2.0。
|
|
|
|
顺序：crm_contract_template → crm_contract → crm_contract_product →
|
|
crm_contract_product_price_tier → crm_contract_approval → crm_contract_change
|
|
用户外键经 user_mapping；附件 OSS ID / URL 规范化；attachment_files JSON 保留 ossId。
|
|
|
|
用法：
|
|
cd backend
|
|
venv/bin/python scripts/migrate_crm_contract.py --dry-run --env-file .env.production
|
|
venv/bin/python scripts/migrate_crm_contract.py --env-file .env.production
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
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 extract_minio_object_key, join_relative_urls, resolve_oss_id_csv # noqa: E402
|
|
|
|
DEFAULT_TENANT_NO = "shanxia2.0"
|
|
CRM_ADMIN_SOURCE_USER_ID = 1
|
|
|
|
MIGRATION_STEPS = (
|
|
("crm_contract_template", "contract_template"),
|
|
("crm_contract", "contract"),
|
|
("crm_contract_product", "contract_product"),
|
|
("crm_contract_product_price_tier", "contract_product_price_tier"),
|
|
("crm_contract_approval", "contract_approval"),
|
|
("crm_contract_change", "contract_change"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class StepStats:
|
|
source_count: int = 0
|
|
inserted: int = 0
|
|
updated: 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,
|
|
*,
|
|
required: bool = False,
|
|
) -> Optional[str]:
|
|
if source_user_id is None or str(source_user_id).strip() == "":
|
|
return admin_id if required else None
|
|
try:
|
|
uid = int(source_user_id)
|
|
except (TypeError, ValueError):
|
|
stats.warnings.append(f"{label}: 非法用户 id {source_user_id!r}")
|
|
return admin_id if required else 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 admin_id if required else None
|
|
return mapped
|
|
|
|
|
|
def load_oss_id_to_url(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 normalize_attachment_csv(raw: Any, oss_map: Dict[str, str]) -> Optional[str]:
|
|
if raw is None or str(raw).strip() == "":
|
|
return None
|
|
oss_resolved = resolve_oss_id_csv(raw, oss_map)
|
|
if oss_resolved:
|
|
return oss_resolved
|
|
parts: List[str] = []
|
|
for token in str(raw).split(","):
|
|
token = token.strip()
|
|
if not token:
|
|
continue
|
|
if token.isdigit() and token in oss_map:
|
|
parts.append(oss_map[token])
|
|
else:
|
|
key = extract_minio_object_key(token)
|
|
if key:
|
|
parts.append(key)
|
|
joined = join_relative_urls(parts)
|
|
return joined or None
|
|
|
|
|
|
def normalize_contract_attachment_files(raw: Any, oss_map: Dict[str, str]) -> Optional[str]:
|
|
if raw is None or str(raw).strip() == "":
|
|
return None
|
|
value = str(raw).strip()
|
|
if value.startswith("["):
|
|
try:
|
|
parsed = json.loads(value)
|
|
if isinstance(parsed, list):
|
|
out: List[dict] = []
|
|
for item in parsed:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
oss_id = legacy_id_to_str(item.get("ossId") or item.get("oss_id"))
|
|
if not oss_id:
|
|
continue
|
|
out.append({"ossId": oss_id, "category": item.get("category") or "other"})
|
|
if out:
|
|
return json.dumps(out, ensure_ascii=False)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return normalize_attachment_csv(raw, oss_map)
|
|
|
|
|
|
def normalize_file_path(raw: Any, oss_map: Dict[str, str]) -> Optional[str]:
|
|
if raw is None or str(raw).strip() == "":
|
|
return None
|
|
value = str(raw).strip()
|
|
if value.isdigit() and value in oss_map:
|
|
return oss_map[value]
|
|
key = extract_minio_object_key(value)
|
|
return key or None
|
|
|
|
|
|
def _dec(value: Any, default: Optional[Decimal] = None) -> Optional[Decimal]:
|
|
if value is None or value == "":
|
|
return default
|
|
return Decimal(str(value))
|
|
|
|
|
|
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 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_contract_template(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract_template#{rid}"
|
|
|
|
def map_user(field_val: Any) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)
|
|
|
|
return {
|
|
"id": rid,
|
|
"template_no": (row.get("template_no") or "")[:50],
|
|
"template_name": (row.get("template_name") or "")[:200],
|
|
"contract_type": _str_or_none(row.get("contract_type"), 20),
|
|
"template_content": row.get("template_content"),
|
|
"payment_terms": row.get("payment_terms"),
|
|
"delivery_terms": row.get("delivery_terms"),
|
|
"contract_terms": row.get("contract_terms"),
|
|
"party_b": _str_or_none(row.get("party_b"), 200),
|
|
"party_b_address": _str_or_none(row.get("party_b_address"), 255),
|
|
"party_b_contact": _str_or_none(row.get("party_b_contact"), 50),
|
|
"party_b_mobile": _str_or_none(row.get("party_b_mobile"), 20),
|
|
"default_currency": (row.get("default_currency") or "CNY")[:10],
|
|
"default_tax_rate": _dec(row.get("default_tax_rate"), Decimal("6.00")),
|
|
"default_period": int(row.get("default_period") or 12),
|
|
"is_active": int(row.get("is_active") if row.get("is_active") is not None else 1),
|
|
"usage_count": int(row.get("usage_count") or 0),
|
|
"sort": int(row.get("sort") or 0),
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 500),
|
|
"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),
|
|
}
|
|
|
|
|
|
def transform_contract(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
oss_map: Dict[str, str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract#{rid}"
|
|
|
|
def map_user(field_val: Any, *, required: bool = False) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label, required=required)
|
|
|
|
owner_id = map_user(row.get("owner_user_id"), required=True)
|
|
|
|
return {
|
|
"id": rid,
|
|
"contract_no": (row.get("contract_no") or "")[:50],
|
|
"contract_name": (row.get("contract_name") or "")[:200],
|
|
"customer_id": legacy_id_to_str(row.get("customer_id")) or "",
|
|
"customer_name": _str_or_none(row.get("customer_name"), 200),
|
|
"opportunity_id": legacy_id_to_str(row.get("opportunity_id")),
|
|
"quote_id": legacy_id_to_str(row.get("quote_id")),
|
|
"parent_contract_id": legacy_id_to_str(row.get("parent_contract_id")),
|
|
"renewal_count": int(row.get("renewal_count") or 0),
|
|
"template_id": legacy_id_to_str(row.get("template_id")),
|
|
"contract_template": (row.get("contract_template") or "shanxia")[:50],
|
|
"contact_id": legacy_id_to_str(row.get("contact_id")),
|
|
"contact_name": _str_or_none(row.get("contact_name"), 50),
|
|
"contact_mobile": _str_or_none(row.get("contact_mobile"), 20),
|
|
"contract_type": (row.get("contract_type") or "sale")[:20],
|
|
"contract_nature": (row.get("contract_nature") or "new")[:20],
|
|
"amount": _dec(row.get("amount"), Decimal("0.00")),
|
|
"currency": (row.get("currency") or "CNY")[:10],
|
|
"tax_rate": _dec(row.get("tax_rate"), Decimal("6.00")),
|
|
"amount_with_tax": _dec(row.get("amount_with_tax"), Decimal("0.00")),
|
|
"status": (row.get("status") or "draft")[:20],
|
|
"performance_status": (row.get("performance_status") or "normal")[:20],
|
|
"performance_rate": _dec(row.get("performance_rate"), Decimal("0.00")),
|
|
"customer_satisfaction": int(row["customer_satisfaction"]) if row.get("customer_satisfaction") is not None else None,
|
|
"approval_status": _str_or_none(row.get("approval_status"), 20),
|
|
"approval_flow_id": legacy_id_to_str(row.get("approval_flow_id")),
|
|
"contract_file": normalize_file_path(row.get("contract_file"), oss_map),
|
|
"attachment_files": normalize_contract_attachment_files(row.get("attachment_files"), oss_map),
|
|
"sign_date": row.get("sign_date"),
|
|
"actual_sign_date": row.get("actual_sign_date"),
|
|
"start_date": row.get("start_date"),
|
|
"end_date": row.get("end_date"),
|
|
"contract_period": int(row["contract_period"]) if row.get("contract_period") is not None else None,
|
|
"owner_user_id": owner_id or "",
|
|
"owner_user_name": _str_or_none(row.get("owner_user_name"), 50),
|
|
"received_amount": _dec(row.get("received_amount"), Decimal("0.00")),
|
|
"unreceived_amount": _dec(row.get("unreceived_amount"), Decimal("0.00")),
|
|
"invoice_amount": _dec(row.get("invoice_amount"), Decimal("0.00")),
|
|
"uninvoice_amount": _dec(row.get("uninvoice_amount"), Decimal("0.00")),
|
|
"party_a": _str_or_none(row.get("party_a"), 200),
|
|
"party_a_address": _str_or_none(row.get("party_a_address"), 255),
|
|
"party_a_contact": _str_or_none(row.get("party_a_contact"), 50),
|
|
"party_a_mobile": _str_or_none(row.get("party_a_mobile"), 20),
|
|
"party_a_legal_person": _str_or_none(row.get("party_a_legal_person"), 50),
|
|
"party_a_agent": _str_or_none(row.get("party_a_agent"), 50),
|
|
"party_a_phone_fax": _str_or_none(row.get("party_a_phone_fax"), 50),
|
|
"party_a_bank": _str_or_none(row.get("party_a_bank"), 100),
|
|
"party_a_account": _str_or_none(row.get("party_a_account"), 50),
|
|
"party_b": _str_or_none(row.get("party_b"), 200),
|
|
"party_b_address": _str_or_none(row.get("party_b_address"), 255),
|
|
"party_b_contact": _str_or_none(row.get("party_b_contact"), 50),
|
|
"party_b_mobile": _str_or_none(row.get("party_b_mobile"), 20),
|
|
"party_b_legal_person": _str_or_none(row.get("party_b_legal_person"), 100),
|
|
"party_b_agent": _str_or_none(row.get("party_b_agent"), 100),
|
|
"party_b_phone_fax": _str_or_none(row.get("party_b_phone_fax"), 100),
|
|
"party_b_bank": _str_or_none(row.get("party_b_bank"), 200),
|
|
"party_b_account": _str_or_none(row.get("party_b_account"), 100),
|
|
"payment_terms": row.get("payment_terms"),
|
|
"delivery_terms": row.get("delivery_terms"),
|
|
"contract_terms": row.get("contract_terms"),
|
|
"settlement_method": _str_or_none(row.get("settlement_method"), 50),
|
|
"template_content": row.get("template_content"),
|
|
"settlement_clause_drafts": row.get("settlement_clause_drafts"),
|
|
"invalid_reason": _str_or_none(row.get("invalid_reason"), 500),
|
|
"invalid_time": row.get("invalid_time"),
|
|
"version_no": int(row.get("version_no") if row.get("version_no") is not None else row.get("version") or 0),
|
|
"supplier_id": legacy_id_to_str(row.get("supplier_id")),
|
|
"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),
|
|
}
|
|
|
|
|
|
def transform_contract_product(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract_product#{rid}"
|
|
|
|
def map_user(field_val: Any) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)
|
|
|
|
return {
|
|
"id": rid,
|
|
"contract_id": legacy_id_to_str(row.get("contract_id")) or "",
|
|
"product_id": legacy_id_to_str(row.get("product_id")) or "",
|
|
"product_name": (row.get("product_name") or "")[:200],
|
|
"product_no": _str_or_none(row.get("product_no"), 50),
|
|
"category_id": legacy_id_to_str(row.get("category_id")),
|
|
"category_name": _str_or_none(row.get("category_name"), 100),
|
|
"model": _str_or_none(row.get("model"), 100),
|
|
"unit": (row.get("unit") or "件")[:20],
|
|
"quantity": _dec(row.get("quantity"), Decimal("0")),
|
|
"price": _dec(row.get("price"), Decimal("0.00")),
|
|
"discount": _dec(row.get("discount"), Decimal("0.00")),
|
|
"tax_rate": _dec(row.get("tax_rate"), Decimal("6.00")),
|
|
"tax_amount": _dec(row.get("tax_amount"), Decimal("0.00")),
|
|
"subtotal": _dec(row.get("subtotal"), Decimal("0.00")),
|
|
"delivery_date": row.get("delivery_date"),
|
|
"delivery_status": (row.get("delivery_status") or "pending")[:20],
|
|
"delivered_quantity": _dec(row.get("delivered_quantity"), Decimal("0")),
|
|
"description": _str_or_none(row.get("description"), 1000),
|
|
"package_spec": _str_or_none(row.get("package_spec"), 200),
|
|
"delivery_method": _str_or_none(row.get("delivery_method"), 100),
|
|
"sort": int(row.get("sort") or 0),
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 500),
|
|
"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),
|
|
}
|
|
|
|
|
|
def transform_contract_price_tier(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract_price_tier#{rid}"
|
|
|
|
def map_user(field_val: Any) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)
|
|
|
|
letter_oss_ids = row.get("letter_oss_ids")
|
|
letter_oss_id = legacy_id_to_str(row.get("letter_oss_id"))
|
|
|
|
return {
|
|
"id": rid,
|
|
"contract_id": legacy_id_to_str(row.get("contract_id")) or "",
|
|
"contract_product_id": legacy_id_to_str(row.get("contract_product_id")) or "",
|
|
"start_date": row.get("start_date"),
|
|
"end_date": row.get("end_date"),
|
|
"price": _dec(row.get("price"), Decimal("0")),
|
|
"letter_oss_ids": letter_oss_ids if letter_oss_ids else None,
|
|
"letter_oss_id": letter_oss_id,
|
|
"sort": int(row.get("sort") or 0),
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 500),
|
|
"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),
|
|
}
|
|
|
|
|
|
def transform_contract_approval(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
oss_map: Dict[str, str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract_approval#{rid}"
|
|
|
|
def map_user(field_val: Any, *, required: bool = False) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label, required=required)
|
|
|
|
return {
|
|
"id": rid,
|
|
"contract_id": legacy_id_to_str(row.get("contract_id")) or "",
|
|
"flow_id": legacy_id_to_str(row.get("flow_id")),
|
|
"step": int(row.get("step") or 1),
|
|
"step_name": _str_or_none(row.get("step_name"), 100),
|
|
"approver_id": map_user(row.get("approver_id"), required=True) or "",
|
|
"approver_name": _str_or_none(row.get("approver_name"), 50),
|
|
"approver_dept": _str_or_none(row.get("approver_dept"), 100),
|
|
"approval_result": (row.get("approval_result") or "pending")[:20],
|
|
"approval_opinion": row.get("approval_opinion"),
|
|
"approval_time": row.get("approval_time"),
|
|
"approval_duration": int(row["approval_duration"]) if row.get("approval_duration") is not None else None,
|
|
"attachments": normalize_attachment_csv(row.get("attachments"), oss_map),
|
|
"is_current": int(row.get("is_current") if row.get("is_current") is not None else 1),
|
|
"sort": int(row.get("sort") or 0),
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 500),
|
|
"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),
|
|
}
|
|
|
|
|
|
def transform_contract_change(
|
|
row: dict,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
oss_map: Dict[str, str],
|
|
stats: MigrationStats,
|
|
) -> dict:
|
|
rid = legacy_id_to_str(row["id"])
|
|
label = f"contract_change#{rid}"
|
|
|
|
def map_user(field_val: Any) -> Optional[str]:
|
|
return resolve_portal_user_id(field_val, crm_user_map, admin_id, stats, label)
|
|
|
|
return {
|
|
"id": rid,
|
|
"contract_id": legacy_id_to_str(row.get("contract_id")) or "",
|
|
"contract_no": _str_or_none(row.get("contract_no"), 50),
|
|
"change_type": (row.get("change_type") or "")[:50],
|
|
"change_reason": (row.get("change_reason") or "")[:500],
|
|
"change_content": row.get("change_content"),
|
|
"old_value": row.get("old_value"),
|
|
"new_value": row.get("new_value"),
|
|
"change_amount": _dec(row.get("change_amount")),
|
|
"approval_status": (row.get("approval_status") or "pending")[:20],
|
|
"approver_id": map_user(row.get("approver_id")),
|
|
"approver_name": _str_or_none(row.get("approver_name"), 50),
|
|
"approval_time": row.get("approval_time"),
|
|
"approval_opinion": _str_or_none(row.get("approval_opinion"), 500),
|
|
"attachment_files": normalize_contract_attachment_files(row.get("attachment_files"), oss_map),
|
|
"tenant_no": tenant_no,
|
|
"order_num": 0,
|
|
"remark": _str_or_none(row.get("remark"), 500),
|
|
"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_CONTRACT_TEMPLATE_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract_template (
|
|
id, template_no, template_name, contract_type, template_content, payment_terms,
|
|
delivery_terms, contract_terms, party_b, party_b_address, party_b_contact, party_b_mobile,
|
|
default_currency, default_tax_rate, default_period, is_active, usage_count, sort,
|
|
tenant_no, order_num, remark, create_by, create_dept_id, create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :template_no, :template_name, :contract_type, :template_content, :payment_terms,
|
|
:delivery_terms, :contract_terms, :party_b, :party_b_address, :party_b_contact, :party_b_mobile,
|
|
:default_currency, :default_tax_rate, :default_period, :is_active, :usage_count, :sort,
|
|
:tenant_no, :order_num, :remark, :create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
INSERT_CONTRACT_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract (
|
|
id, contract_no, contract_name, customer_id, customer_name, opportunity_id, quote_id,
|
|
parent_contract_id, renewal_count, template_id, contract_template, contact_id, contact_name,
|
|
contact_mobile, contract_type, contract_nature, amount, currency, tax_rate, amount_with_tax,
|
|
status, performance_status, performance_rate, customer_satisfaction, approval_status,
|
|
approval_flow_id, contract_file, attachment_files, sign_date, actual_sign_date, start_date,
|
|
end_date, contract_period, owner_user_id, owner_user_name, received_amount, unreceived_amount,
|
|
invoice_amount, uninvoice_amount, party_a, party_a_address, party_a_contact, party_a_mobile,
|
|
party_a_legal_person, party_a_agent, party_a_phone_fax, party_a_bank, party_a_account,
|
|
party_b, party_b_address, party_b_contact, party_b_mobile, party_b_legal_person, party_b_agent,
|
|
party_b_phone_fax, party_b_bank, party_b_account, payment_terms, delivery_terms, contract_terms,
|
|
settlement_method, template_content, settlement_clause_drafts, invalid_reason, invalid_time,
|
|
version_no, supplier_id, tenant_no, order_num, remark, create_by, create_dept_id, create_time,
|
|
update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :contract_no, :contract_name, :customer_id, :customer_name, :opportunity_id, :quote_id,
|
|
:parent_contract_id, :renewal_count, :template_id, :contract_template, :contact_id, :contact_name,
|
|
:contact_mobile, :contract_type, :contract_nature, :amount, :currency, :tax_rate, :amount_with_tax,
|
|
:status, :performance_status, :performance_rate, :customer_satisfaction, :approval_status,
|
|
:approval_flow_id, :contract_file, :attachment_files, :sign_date, :actual_sign_date, :start_date,
|
|
:end_date, :contract_period, :owner_user_id, :owner_user_name, :received_amount, :unreceived_amount,
|
|
:invoice_amount, :uninvoice_amount, :party_a, :party_a_address, :party_a_contact, :party_a_mobile,
|
|
:party_a_legal_person, :party_a_agent, :party_a_phone_fax, :party_a_bank, :party_a_account,
|
|
:party_b, :party_b_address, :party_b_contact, :party_b_mobile, :party_b_legal_person, :party_b_agent,
|
|
:party_b_phone_fax, :party_b_bank, :party_b_account, :payment_terms, :delivery_terms, :contract_terms,
|
|
:settlement_method, :template_content, :settlement_clause_drafts, :invalid_reason, :invalid_time,
|
|
:version_no, :supplier_id, :tenant_no, :order_num, :remark, :create_by, :create_dept_id, :create_time,
|
|
:update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
INSERT_CONTRACT_PRODUCT_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract_product (
|
|
id, contract_id, product_id, product_name, product_no, category_id, category_name,
|
|
model, unit, quantity, price, discount, tax_rate, tax_amount, subtotal,
|
|
delivery_date, delivery_status, delivered_quantity, description, package_spec,
|
|
delivery_method, sort, tenant_no, order_num, remark, create_by, create_dept_id,
|
|
create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :contract_id, :product_id, :product_name, :product_no, :category_id, :category_name,
|
|
:model, :unit, :quantity, :price, :discount, :tax_rate, :tax_amount, :subtotal,
|
|
:delivery_date, :delivery_status, :delivered_quantity, :description, :package_spec,
|
|
:delivery_method, :sort, :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
|
|
:create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
INSERT_CONTRACT_PRICE_TIER_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract_product_price_tier (
|
|
id, contract_id, contract_product_id, start_date, end_date, price,
|
|
letter_oss_ids, letter_oss_id, sort, tenant_no, order_num, remark,
|
|
create_by, create_dept_id, create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :contract_id, :contract_product_id, :start_date, :end_date, :price,
|
|
:letter_oss_ids, :letter_oss_id, :sort, :tenant_no, :order_num, :remark,
|
|
:create_by, :create_dept_id, :create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
INSERT_CONTRACT_APPROVAL_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract_approval (
|
|
id, contract_id, flow_id, step, step_name, approver_id, approver_name, approver_dept,
|
|
approval_result, approval_opinion, approval_time, approval_duration, attachments,
|
|
is_current, sort, tenant_no, order_num, remark, create_by, create_dept_id,
|
|
create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :contract_id, :flow_id, :step, :step_name, :approver_id, :approver_name, :approver_dept,
|
|
:approval_result, :approval_opinion, :approval_time, :approval_duration, :attachments,
|
|
:is_current, :sort, :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
|
|
:create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
INSERT_CONTRACT_CHANGE_SQL = text(
|
|
"""
|
|
INSERT INTO crm_contract_change (
|
|
id, contract_id, contract_no, change_type, change_reason, change_content, old_value,
|
|
new_value, change_amount, approval_status, approver_id, approver_name, approval_time,
|
|
approval_opinion, attachment_files, tenant_no, order_num, remark, create_by, create_dept_id,
|
|
create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
:id, :contract_id, :contract_no, :change_type, :change_reason, :change_content, :old_value,
|
|
:new_value, :change_amount, :approval_status, :approver_id, :approver_name, :approval_time,
|
|
:approval_opinion, :attachment_files, :tenant_no, :order_num, :remark, :create_by, :create_dept_id,
|
|
:create_time, :update_by, :update_time, :del_flag
|
|
)
|
|
"""
|
|
)
|
|
|
|
STEP_INSERT_SQL = {
|
|
"crm_contract_template": INSERT_CONTRACT_TEMPLATE_SQL,
|
|
"crm_contract": INSERT_CONTRACT_SQL,
|
|
"crm_contract_product": INSERT_CONTRACT_PRODUCT_SQL,
|
|
"crm_contract_product_price_tier": INSERT_CONTRACT_PRICE_TIER_SQL,
|
|
"crm_contract_approval": INSERT_CONTRACT_APPROVAL_SQL,
|
|
"crm_contract_change": INSERT_CONTRACT_CHANGE_SQL,
|
|
}
|
|
|
|
|
|
def write_mapping_csv(source_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 source_rows:
|
|
sid = legacy_id_to_str(row["id"])
|
|
w.writerow({"source_id": sid, "new_id": sid, "migrated_at": migrated_at})
|
|
|
|
|
|
def migrate_table(
|
|
source_engine: Engine,
|
|
target_engine: Engine,
|
|
table: str,
|
|
*,
|
|
tenant_no: str,
|
|
crm_user_map: Dict[int, str],
|
|
admin_id: Optional[str],
|
|
oss_map: Dict[str, str],
|
|
stats: MigrationStats,
|
|
dry_run: bool,
|
|
update_existing: bool,
|
|
include_deleted: bool,
|
|
) -> List[dict]:
|
|
step = stats.step(table)
|
|
source_rows = fetch_source_rows(source_engine, table, include_deleted)
|
|
step.source_count = len(source_rows)
|
|
|
|
transformers = {
|
|
"crm_contract_template": lambda r: transform_contract_template(r, tenant_no, crm_user_map, admin_id, stats),
|
|
"crm_contract": lambda r: transform_contract(r, tenant_no, crm_user_map, admin_id, oss_map, stats),
|
|
"crm_contract_product": lambda r: transform_contract_product(r, tenant_no, crm_user_map, admin_id, stats),
|
|
"crm_contract_product_price_tier": lambda r: transform_contract_price_tier(
|
|
r, tenant_no, crm_user_map, admin_id, stats
|
|
),
|
|
"crm_contract_approval": lambda r: transform_contract_approval(
|
|
r, tenant_no, crm_user_map, admin_id, oss_map, stats
|
|
),
|
|
"crm_contract_change": lambda r: transform_contract_change(
|
|
r, tenant_no, crm_user_map, admin_id, oss_map, stats
|
|
),
|
|
}
|
|
payloads = [transformers[table](r) for r in source_rows]
|
|
|
|
if dry_run:
|
|
step.inserted = len(payloads)
|
|
return source_rows
|
|
|
|
insert_sql = STEP_INSERT_SQL[table]
|
|
with target_engine.begin() as conn:
|
|
for payload in payloads:
|
|
rid = payload["id"]
|
|
if target_row_exists(conn, table, rid):
|
|
if update_existing:
|
|
step.skipped += 1
|
|
else:
|
|
step.skipped += 1
|
|
else:
|
|
conn.execute(insert_sql, payload)
|
|
step.inserted += 1
|
|
return source_rows
|
|
|
|
|
|
def print_report(stats: MigrationStats, dry_run: bool) -> None:
|
|
mode = "DRY-RUN" if dry_run else "COMMITTED"
|
|
print(f"\n=== CRM 合同迁移 [{mode}] ===")
|
|
for table, _ in MIGRATION_STEPS:
|
|
s = stats.step(table)
|
|
print(f"{table}: 源={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,
|
|
update_existing: bool,
|
|
include_deleted: bool,
|
|
only_tables: Optional[List[str]],
|
|
) -> 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)
|
|
oss_map = load_oss_id_to_url(target_engine, tenant_no)
|
|
print(f"sys_oss 映射条数: {len(oss_map)}")
|
|
|
|
selected = [t for t, _ in MIGRATION_STEPS if not only_tables or t in only_tables]
|
|
for table in selected:
|
|
print(f"\n>> 迁移 {table} ...")
|
|
source_rows = migrate_table(
|
|
source_engine,
|
|
target_engine,
|
|
table,
|
|
tenant_no=tenant_no,
|
|
crm_user_map=crm_user_map,
|
|
admin_id=admin_id,
|
|
oss_map=oss_map,
|
|
stats=stats,
|
|
dry_run=dry_run,
|
|
update_existing=update_existing,
|
|
include_deleted=include_deleted,
|
|
)
|
|
if table == "crm_contract" and not dry_run:
|
|
write_mapping_csv(source_rows, mapping_dir / "crm_contract_id_mapping.csv")
|
|
|
|
print_report(stats, dry_run)
|
|
return stats
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Migrate CRM contract tables 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("--update", action="store_true")
|
|
parser.add_argument("--include-deleted", action="store_true")
|
|
parser.add_argument("--tables", default="", help="逗号分隔，默认全部六表")
|
|
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)
|
|
only_tables = [t.strip() for t in args.tables.split(",") if t.strip()] or None
|
|
|
|
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,
|
|
update_existing=args.update,
|
|
include_deleted=args.include_deleted,
|
|
only_tables=only_tables,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|