726 lines
23 KiB
Python
726 lines
23 KiB
Python
#!/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()
|