提交

backend/scripts/merge_duplicate_crm_customers.py

@@ -0,0 +1,725 @@
+#!/usr/bin/env python3
+"""合并 CRM 重复客户（测试/生产均可）。
+
+规则摘要：
+- 同名 del_flag=0 客户每组保留 create_time 最新的一条（新业务记录）
+- 旧客户软删，子表 customer_id 迁到保留客户
+- 联系人/开票：核心字段相同则去重；不同则迁移；旧联系人优先 is_primary
+- 开票默认以保留客户（新）侧为准；旧侧不同内容迁移保留
+- 修改前按 YYYYMMDD 备份涉及表
+
+用法：
+  cd backend
+  venv/bin/python scripts/merge_duplicate_crm_customers.py --dry-run
+  venv/bin/python scripts/merge_duplicate_crm_customers.py --execute
+"""
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from datetime import date, datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import Connection, Engine
+
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+if str(BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(BACKEND_ROOT))
+
+TABLES_TO_TOUCH = [
+    "crm_customer",
+    "crm_contact",
+    "crm_invoice_info",
+    "crm_customer_follow",
+    "crm_opportunity",
+    "crm_contract",
+    "crm_quote",
+    "crm_price_letter",
+    "crm_clue",
+    "crm_customer_pool",
+    "pl_plan",
+]
+
+CUSTOMER_FILL_FIELDS = [
+    "short_name",
+    "customer_type",
+    "mobile",
+    "telephone",
+    "email",
+    "website",
+    "level",
+    "source",
+    "industry",
+    "park_area",
+    "payment_terms",
+    "scale",
+    "province",
+    "city",
+    "district",
+    "address",
+    "owner_user_id",
+    "owner_user_name",
+    "business_license_img",
+    "business_permit_images",
+    "other_cert_images",
+    "clue_id",
+    "next_time",
+]
+
+INVOICE_COMPARE_FIELDS = [
+    "company_name",
+    "tax_no",
+    "bank_account",
+]
+
+
+def _norm(value: Any) -> str:
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def _contact_key(row: Dict[str, Any]) -> str:
+    mobile = _norm(row.get("mobile"))
+    telephone = _norm(row.get("telephone"))
+    phone = mobile or telephone
+    return f"{_norm(row.get('contact_name')).lower()}|{phone}"
+
+
+def _invoice_key(row: Dict[str, Any]) -> str:
+    return "|".join(_norm(row.get(f)).lower() for f in INVOICE_COMPARE_FIELDS)
+
+
+def _row_to_dict(row) -> Dict[str, Any]:
+    return dict(row._mapping)
+
+
+def load_database_url(env_file: Optional[str]) -> str:
+    if env_file:
+        path = Path(env_file)
+        if not path.is_absolute():
+            path = BACKEND_ROOT / path
+        for line in path.read_text(encoding="utf-8").splitlines():
+            line = line.strip()
+            if line.startswith("DATABASE_URL="):
+                return line.split("=", 1)[1].strip()
+    from app.core.config import settings
+
+    return settings.DATABASE_URL
+
+
+def discover_merge_pairs(conn: Connection) -> List[Dict[str, Any]]:
+    rows = conn.execute(
+        text(
+            """
+            SELECT id, customer_name, customer_no, create_time
+            FROM crm_customer
+            WHERE del_flag = 0 AND customer_name IS NOT NULL AND TRIM(customer_name) != ''
+            ORDER BY customer_name, create_time ASC, customer_no ASC
+            """
+        )
+    ).fetchall()
+    grouped: Dict[str, List[Dict[str, Any]]] = {}
+    for row in rows:
+        item = _row_to_dict(row)
+        grouped.setdefault(item["customer_name"], []).append(item)
+
+    pairs: List[Dict[str, Any]] = []
+    for name, items in grouped.items():
+        if len(items) != 2:
+            continue
+        source, keeper = items[0], items[1]
+        pairs.append(
+            {
+                "customer_name": name,
+                "from_id": source["id"],
+                "from_no": source["customer_no"],
+                "to_id": keeper["id"],
+                "to_no": keeper["customer_no"],
+            }
+        )
+    return pairs
+
+
+def backup_tables(conn: Connection, suffix: str, *, execute: bool) -> None:
+    for table in TABLES_TO_TOUCH:
+        backup_name = f"{table}_{suffix}"
+        sql = f"CREATE TABLE `{backup_name}` AS SELECT * FROM `{table}`"
+        print(f"[backup] {sql}")
+        if execute:
+            conn.execute(text(f"DROP TABLE IF EXISTS `{backup_name}`"))
+            conn.execute(text(sql))
+
+
+def fetch_customer(conn: Connection, customer_id: str) -> Optional[Dict[str, Any]]:
+    row = conn.execute(
+        text("SELECT * FROM crm_customer WHERE id = :id"),
+        {"id": customer_id},
+    ).fetchone()
+    return _row_to_dict(row) if row else None
+
+
+def merge_contacts(
+    conn: Connection,
+    pair: Dict[str, Any],
+    keeper_name: str,
+    *,
+    execute: bool,
+) -> Dict[str, int]:
+    stats = {"migrated": 0, "deduped": 0, "kept": 0}
+    from_rows = conn.execute(
+        text(
+            """
+            SELECT * FROM crm_contact
+            WHERE customer_id = :cid AND del_flag = 0
+            ORDER BY is_primary DESC, create_time ASC
+            """
+        ),
+        {"cid": pair["from_id"]},
+    ).fetchall()
+    to_rows = conn.execute(
+        text(
+            """
+            SELECT * FROM crm_contact
+            WHERE customer_id = :cid AND del_flag = 0
+            ORDER BY is_primary DESC, create_time ASC
+            """
+        ),
+        {"cid": pair["to_id"]},
+    ).fetchall()
+    keeper_keys = {_contact_key(_row_to_dict(r)) for r in to_rows}
+
+    for row in from_rows:
+        src = _row_to_dict(row)
+        key = _contact_key(src)
+        if key in keeper_keys:
+            stats["deduped"] += 1
+            print(
+                f"  [contact dedupe] {pair['customer_name']} duplicate key={key} "
+                f"soft-delete {src['id']}"
+            )
+            if execute:
+                conn.execute(
+                    text(
+                        """
+                        UPDATE crm_contact
+                        SET del_flag = 1, update_time = NOW(),
+                            remark = CONCAT(IFNULL(remark, ''), '[合并去重]')
+                        WHERE id = :id
+                        """
+                    ),
+                    {"id": src["id"]},
+                )
+            continue
+
+        stats["migrated"] += 1
+        print(
+            f"  [contact migrate] {src['id']} -> customer {pair['to_id']} "
+            f"set primary=1 (old contact)"
+        )
+        if execute:
+            conn.execute(
+                text(
+                    """
+                    UPDATE crm_contact
+                    SET is_primary = 0, update_time = NOW()
+                    WHERE customer_id = :cid AND del_flag = 0
+                    """
+                ),
+                {"cid": pair["to_id"]},
+            )
+            conn.execute(
+                text(
+                    """
+                    UPDATE crm_contact
+                    SET customer_id = :to_id,
+                        customer_name = :customer_name,
+                        is_primary = 1,
+                        update_time = NOW()
+                    WHERE id = :id
+                    """
+                ),
+                {
+                    "to_id": pair["to_id"],
+                    "customer_name": keeper_name,
+                    "id": src["id"],
+                },
+            )
+        keeper_keys.add(key)
+
+    stats["kept"] = len(to_rows)
+    return stats
+
+
+def normalize_invoice_defaults(
+    conn: Connection,
+    customer_id: str,
+    *,
+    execute: bool,
+    prefer_ids: Optional[List[str]] = None,
+) -> None:
+    """确保保留客户下活跃开票有且仅有一条 is_default=1；优先保留客户（新）侧原默认。"""
+    rows = conn.execute(
+        text(
+            """
+            SELECT id, is_default, create_time
+            FROM crm_invoice_info
+            WHERE customer_id = :cid AND del_flag = 0
+            ORDER BY is_default DESC, create_time ASC
+            """
+        ),
+        {"cid": customer_id},
+    ).fetchall()
+    if not rows:
+        return
+
+    active = [_row_to_dict(r) for r in rows]
+    prefer_set = set(prefer_ids or [])
+    default_id: Optional[str] = None
+
+    if len(active) == 1:
+        default_id = active[0]["id"]
+    else:
+        preferred = [item for item in active if item["id"] in prefer_set]
+        if preferred:
+            for item in preferred:
+                if str(item.get("is_default")) == "1":
+                    default_id = item["id"]
+                    break
+            if not default_id:
+                default_id = preferred[0]["id"]
+        if not default_id:
+            for item in active:
+                if str(item.get("is_default")) == "1":
+                    default_id = item["id"]
+                    break
+        if not default_id:
+            default_id = active[0]["id"]
+
+    print(f"  [invoice default] customer {customer_id} -> {default_id}")
+    if execute:
+        conn.execute(
+            text(
+                """
+                UPDATE crm_invoice_info
+                SET is_default = '0', update_time = NOW()
+                WHERE customer_id = :cid AND del_flag = 0
+                """
+            ),
+            {"cid": customer_id},
+        )
+        conn.execute(
+            text(
+                """
+                UPDATE crm_invoice_info
+                SET is_default = '1', update_time = NOW()
+                WHERE id = :id
+                """
+            ),
+            {"id": default_id},
+        )
+
+
+def merge_invoices(
+    conn: Connection,
+    pair: Dict[str, Any],
+    *,
+    execute: bool,
+) -> Dict[str, int]:
+    stats = {"migrated": 0, "deduped": 0}
+    from_rows = conn.execute(
+        text(
+            """
+            SELECT * FROM crm_invoice_info
+            WHERE customer_id = :cid AND del_flag = 0
+            ORDER BY is_default DESC, create_time ASC
+            """
+        ),
+        {"cid": pair["from_id"]},
+    ).fetchall()
+    to_rows = conn.execute(
+        text(
+            """
+            SELECT * FROM crm_invoice_info
+            WHERE customer_id = :cid AND del_flag = 0
+            ORDER BY is_default DESC, create_time ASC
+            """
+        ),
+        {"cid": pair["to_id"]},
+    ).fetchall()
+    keeper_original_ids = [_row_to_dict(r)["id"] for r in to_rows]
+    keeper_map = {_invoice_key(_row_to_dict(r)): _row_to_dict(r) for r in to_rows}
+
+    for row in from_rows:
+        src = _row_to_dict(row)
+        key = _invoice_key(src)
+        if key in keeper_map:
+            stats["deduped"] += 1
+            print(f"  [invoice dedupe] key={key} soft-delete {src['id']}")
+            if execute:
+                conn.execute(
+                    text(
+                        """
+                        UPDATE crm_invoice_info
+                        SET del_flag = 1, update_time = NOW(),
+                            remark = CONCAT(IFNULL(remark, ''), '[合并去重]')
+                        WHERE id = :id
+                        """
+                    ),
+                    {"id": src["id"]},
+                )
+            continue
+
+        stats["migrated"] += 1
+        print(f"  [invoice migrate] {src['id']} -> customer {pair['to_id']} is_default=0")
+        if execute:
+            conn.execute(
+                text(
+                    """
+                    UPDATE crm_invoice_info
+                    SET customer_id = :to_id,
+                        is_default = '0',
+                        update_time = NOW()
+                    WHERE id = :id
+                    """
+                ),
+                {"to_id": pair["to_id"], "id": src["id"]},
+            )
+        keeper_map[key] = src
+
+    normalize_invoice_defaults(
+        conn,
+        pair["to_id"],
+        execute=execute,
+        prefer_ids=keeper_original_ids,
+    )
+    return stats
+
+
+def migrate_customer_refs(
+    conn: Connection,
+    pair: Dict[str, Any],
+    keeper_name: str,
+    *,
+    execute: bool,
+) -> None:
+    updates = [
+        (
+            "crm_customer_follow",
+            "UPDATE crm_customer_follow SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_opportunity",
+            "UPDATE crm_opportunity SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_contract",
+            "UPDATE crm_contract SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_quote",
+            "UPDATE crm_quote SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_price_letter",
+            "UPDATE crm_price_letter SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_customer_pool",
+            "UPDATE crm_customer_pool SET customer_id=:to_id, customer_name=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "crm_clue",
+            "UPDATE crm_clue SET customer_id=:to_id, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+        (
+            "pl_plan",
+            "UPDATE pl_plan SET customer_id=:to_id, customer_name=:name, party_a=:name, update_time=NOW() "
+            "WHERE customer_id=:from_id AND del_flag=0",
+        ),
+    ]
+    params = {"from_id": pair["from_id"], "to_id": pair["to_id"], "name": keeper_name}
+    for table, sql in updates:
+        count = conn.execute(
+            text(f"SELECT COUNT(*) FROM `{table}` WHERE customer_id=:from_id AND del_flag=0"),
+            {"from_id": pair["from_id"]},
+        ).scalar()
+        if count:
+            print(f"  [{table}] migrate {count} rows")
+            if execute:
+                conn.execute(text(sql), params)
+
+
+def merge_customer_master(
+    conn: Connection,
+    pair: Dict[str, Any],
+    *,
+    execute: bool,
+) -> None:
+    keeper = fetch_customer(conn, pair["to_id"])
+    source = fetch_customer(conn, pair["from_id"])
+    if not keeper or not source:
+        return
+
+    sets = []
+    params: Dict[str, Any] = {"id": pair["to_id"]}
+    for field in CUSTOMER_FILL_FIELDS:
+        k_val = keeper.get(field)
+        s_val = source.get(field)
+        if _norm(k_val):
+            continue
+        if not _norm(s_val):
+            continue
+        sets.append(f"{field} = :{field}")
+        params[field] = s_val
+
+    deal_priority = {"dealed": 3, "lost": 2, "not_deal": 1}
+    if deal_priority.get(_norm(source.get("deal_status")), 0) > deal_priority.get(
+        _norm(keeper.get("deal_status")), 0
+    ):
+        sets.extend(
+            [
+                "deal_status = :deal_status",
+                "deal_time = COALESCE(:deal_time, deal_time)",
+                "deal_amount = GREATEST(IFNULL(deal_amount,0), IFNULL(:deal_amount,0))",
+            ]
+        )
+        params["deal_status"] = source.get("deal_status")
+        params["deal_time"] = source.get("deal_time")
+        params["deal_amount"] = source.get("deal_amount")
+
+    merge_note = (
+        f"[{date.today().isoformat()} 合并客户 {source.get('customer_no')}({source.get('id')})]"
+    )
+    old_remark = _norm(keeper.get("remark"))
+    if merge_note not in old_remark:
+        sets.append("remark = :remark")
+        params["remark"] = f"{old_remark}\n{merge_note}".strip()
+
+    if sets and execute:
+        sql = f"UPDATE crm_customer SET {', '.join(sets)}, update_time = NOW() WHERE id = :id"
+        conn.execute(text(sql), params)
+        print(f"  [customer master] filled fields: {', '.join(x.split(' ')[0] for x in sets)}")
+
+
+def recalc_customer_stats(conn: Connection, customer_id: str, *, execute: bool) -> None:
+    if not execute:
+        return
+    contact_count = conn.execute(
+        text("SELECT COUNT(*) FROM crm_contact WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    opportunity_count = conn.execute(
+        text("SELECT COUNT(*) FROM crm_opportunity WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    contract_count = conn.execute(
+        text("SELECT COUNT(*) FROM crm_contract WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    follow_count = conn.execute(
+        text("SELECT COUNT(*) FROM crm_customer_follow WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    deal_amount = conn.execute(
+        text("SELECT COALESCE(SUM(amount),0) FROM crm_contract WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    last_time = conn.execute(
+        text("SELECT MAX(follow_time) FROM crm_customer_follow WHERE customer_id=:id AND del_flag=0"),
+        {"id": customer_id},
+    ).scalar()
+    next_time = conn.execute(
+        text(
+            """
+            SELECT MAX(next_time) FROM (
+              SELECT next_time FROM crm_customer_follow WHERE customer_id=:id AND del_flag=0
+              UNION ALL
+              SELECT next_time FROM crm_customer WHERE id=:id
+            ) t
+            """
+        ),
+        {"id": customer_id},
+    ).scalar()
+    conn.execute(
+        text(
+            """
+            UPDATE crm_customer
+            SET contact_count=:contact_count,
+                opportunity_count=:opportunity_count,
+                contract_count=:contract_count,
+                follow_count=:follow_count,
+                deal_amount=:deal_amount,
+                last_time=COALESCE(:last_time, last_time),
+                next_time=COALESCE(:next_time, next_time),
+                update_time=NOW()
+            WHERE id=:id
+            """
+        ),
+        {
+            "id": customer_id,
+            "contact_count": contact_count,
+            "opportunity_count": opportunity_count,
+            "contract_count": contract_count,
+            "follow_count": follow_count,
+            "deal_amount": deal_amount,
+            "last_time": last_time,
+            "next_time": next_time,
+        },
+    )
+
+
+def soft_delete_source_customer(
+    conn: Connection,
+    pair: Dict[str, Any],
+    *,
+    execute: bool,
+) -> None:
+    note = f"[{date.today().isoformat()} 已合并至 {pair['to_no']}({pair['to_id']})]"
+    print(f"  [customer soft-delete] {pair['from_no']} ({pair['from_id']})")
+    if execute:
+        conn.execute(
+            text(
+                """
+                UPDATE crm_customer
+                SET del_flag = 1,
+                    remark = CONCAT(IFNULL(remark, ''), :note),
+                    update_time = NOW()
+                WHERE id = :id
+                """
+            ),
+            {"id": pair["from_id"], "note": f"\n{note}"},
+        )
+
+
+def verify(conn: Connection, merged_from_ids: List[str]) -> None:
+    dup_groups = conn.execute(
+        text(
+            """
+            SELECT COUNT(*) FROM (
+              SELECT customer_name FROM crm_customer
+              WHERE del_flag=0 AND customer_name IS NOT NULL AND TRIM(customer_name)!=''
+              GROUP BY customer_name HAVING COUNT(*)>1
+            ) t
+            """
+        )
+    ).scalar()
+
+    merge_orphans = 0
+    if merged_from_ids:
+        placeholders = ", ".join(f":id{i}" for i in range(len(merged_from_ids)))
+        params = {f"id{i}": cid for i, cid in enumerate(merged_from_ids)}
+        merge_orphans = conn.execute(
+            text(
+                f"""
+                SELECT COUNT(*) FROM (
+                  SELECT customer_id FROM crm_contact WHERE del_flag=0
+                  UNION ALL SELECT customer_id FROM crm_invoice_info WHERE del_flag=0
+                  UNION ALL SELECT customer_id FROM crm_contract WHERE del_flag=0
+                  UNION ALL SELECT customer_id FROM crm_quote WHERE del_flag=0
+                  UNION ALL SELECT customer_id FROM pl_plan WHERE del_flag=0
+                ) r
+                WHERE r.customer_id IN ({placeholders})
+                """
+            ),
+            params,
+        ).scalar()
+
+    global_orphans = {}
+    for table in [
+        "crm_contact",
+        "crm_invoice_info",
+        "crm_contract",
+        "crm_quote",
+        "pl_plan",
+    ]:
+        cnt = conn.execute(
+            text(
+                f"""
+                SELECT COUNT(*) FROM `{table}` r
+                LEFT JOIN crm_customer c ON c.id = r.customer_id AND c.del_flag = 0
+                WHERE r.del_flag = 0 AND r.customer_id IS NOT NULL AND c.id IS NULL
+                """
+            )
+        ).scalar()
+        global_orphans[table] = cnt
+
+    print("\n=== verification ===")
+    print(f"duplicate active customer name groups: {dup_groups}")
+    print(f"active refs to merged-deleted customer ids: {merge_orphans}")
+    for table, cnt in global_orphans.items():
+        if cnt:
+            print(f"pre-existing orphan refs in {table}: {cnt}")
+    if dup_groups != 0:
+        raise RuntimeError("duplicate customer names still exist")
+    if merge_orphans != 0:
+        raise RuntimeError("active child rows still reference merged-deleted customers")
+
+
+def run(engine: Engine, *, execute: bool, suffix: str) -> None:
+    pairs = []
+    with engine.connect() as conn:
+        pairs = discover_merge_pairs(conn)
+        print(f"found {len(pairs)} merge pairs")
+        for pair in pairs:
+            print(
+                f"  {pair['customer_name']}: {pair['from_no']} -> {pair['to_no']}"
+            )
+
+    if not pairs:
+        print("nothing to merge")
+        return
+
+    if execute:
+        with engine.begin() as conn:
+            backup_tables(conn, suffix, execute=True)
+
+    with engine.begin() as conn:
+        for pair in pairs:
+            keeper = fetch_customer(conn, pair["to_id"])
+            keeper_name = keeper["customer_name"] if keeper else pair["customer_name"]
+            print(f"\n>>> merge {pair['customer_name']}")
+            merge_contacts(conn, pair, keeper_name, execute=execute)
+            merge_invoices(conn, pair, execute=execute)
+            migrate_customer_refs(conn, pair, keeper_name, execute=execute)
+            merge_customer_master(conn, pair, execute=execute)
+            recalc_customer_stats(conn, pair["to_id"], execute=execute)
+            soft_delete_source_customer(conn, pair, execute=execute)
+
+    if execute:
+        with engine.connect() as conn:
+            verify(conn, [p["from_id"] for p in pairs])
+        print("\nmerge completed successfully")
+    else:
+        print("\ndry-run finished (no changes applied)")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Merge duplicate CRM customers")
+    parser.add_argument("--env-file", default=".env", help="env file with DATABASE_URL")
+    parser.add_argument("--suffix", default=date.today().strftime("%Y%m%d"))
+    parser.add_argument("--dry-run", action="store_true", help="preview only")
+    parser.add_argument("--execute", action="store_true", help="apply changes")
+    args = parser.parse_args()
+    if not args.dry_run and not args.execute:
+        parser.error("specify --dry-run or --execute")
+
+    db_url = load_database_url(args.env_file)
+    print(f"database: {re.sub(r':[^:@/]+@', ':***@', db_url)}")
+    print(f"backup suffix: {args.suffix}")
+    engine = create_engine(db_url, pool_pre_ping=True)
+    run(engine, execute=args.execute, suffix=args.suffix)
+
+
+if __name__ == "__main__":
+    main()