import sys, json, csv, argparse
|
from collections import defaultdict, Counter
|
from pathlib import Path
|
from datetime import datetime
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
ROOT = Path(__file__).resolve().parent
|
sys.path.insert(0, str(ROOT.parent / 'iqc-import-20261012/deps'))
|
import pymysql
|
|
FIELDS = ('mobile', 'telephone', 'email')
|
POSITION_RANK = {
|
'总经理': 50,
|
'采购经理': 40, '供应链经理': 40, '商品部经理': 40,
|
'采购主管': 35, '商务主管': 35,
|
'店长': 30, '门店经理': 30, '药房负责人': 30,
|
'办公室主任': 25, '财务负责人': 25, '质量负责人': 25, '库房主管': 25,
|
'采购员': 10,
|
}
|
RULE = '关键决策人 > 职位层级 > 同客户直属下属数量 > 联系方式完整度 > 最近跟进时间 > 最近更新时间 > 联系人ID升序；三字段取同一联系人，空值保留客户原值'
|
|
def clean(value):
|
return value.strip() if isinstance(value, str) else value
|
|
def rank(contact, children):
|
return (int(contact.get('master') == 1), POSITION_RANK.get(clean(contact.get('post')), 0),
|
children[contact['id']], sum(bool(clean(contact.get(field))) for field in FIELDS),
|
contact.get('contact_last_time') or datetime.min,
|
contact.get('update_time') or datetime.min, -contact['id'])
|
|
def build_plan(customers, contacts):
|
grouped = defaultdict(list)
|
for contact in contacts:
|
grouped[contact['customer_id']].append(contact)
|
result = []
|
for customer in customers:
|
candidates = grouped.get(customer['id'], [])
|
if not candidates:
|
continue
|
candidate_ids = {c['id'] for c in candidates}
|
children = Counter(c['parent_id'] for c in candidates if c['parent_id'] in candidate_ids)
|
chosen = max(candidates, key=lambda c: rank(c, children))
|
values = {field: clean(chosen[field]) if clean(chosen.get(field)) else customer[field] for field in FIELDS}
|
changed = [field for field in FIELDS if values[field] != customer[field]]
|
result.append({'customer_id': customer['id'], 'customer_name': customer['name'],
|
'contact_id': chosen['id'], 'contact_name': chosen['name'],
|
'master': chosen['master'], 'post': chosen['post'], 'candidate_count': len(candidates),
|
'old': {field: customer[field] for field in FIELDS}, 'new': values, 'changed': changed})
|
return result
|
|
def write_json(path, value):
|
path.write_text(json.dumps(value, ensure_ascii=False, default=str, indent=2), encoding='utf-8')
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument('--apply', action='store_true')
|
options = parser.parse_args()
|
password = sys.stdin.readline().strip()
|
args = dict(host='36.213.150.237', port=10000, user='root', password=password,
|
database='mom-hm', charset='utf8mb4', connect_timeout=15, read_timeout=60,
|
cursorclass=pymysql.cursors.DictCursor, autocommit=False)
|
connection = pymysql.connect(**args)
|
try:
|
with connection.cursor() as cursor:
|
cursor.execute('SELECT DATABASE() AS db, NOW() AS now')
|
state = cursor.fetchone()
|
assert state['db'] == 'mom-hm'
|
lock = ' FOR UPDATE' if options.apply else ''
|
# Lock both source and target to keep selection and writes consistent.
|
cursor.execute('SELECT * FROM crm_customer ORDER BY id' + lock)
|
all_customers = cursor.fetchall()
|
cursor.execute('SELECT * FROM crm_contact ORDER BY id' + lock)
|
all_contacts = cursor.fetchall()
|
customers = [row for row in all_customers if row['deleted'] == 0]
|
customer_ids = {row['id'] for row in customers}
|
contacts = [row for row in all_contacts if row['deleted'] == 0 and row['customer_id'] in customer_ids]
|
plan = build_plan(customers, contacts)
|
key_counts = Counter(row['customer_id'] for row in contacts if row['master'] == 1)
|
changes = [row for row in plan if row['changed']]
|
summary = {'database': 'mom-hm', 'rule': RULE, 'active_customers': len(customers),
|
'active_contacts': sum(row['deleted'] == 0 for row in all_contacts),
|
'matched_customers': len(plan), 'multiple_contact_customers': sum(row['candidate_count'] > 1 for row in plan),
|
'updated_customers': len(changes),
|
'field_updates': {field: sum(field in row['changed'] for row in changes) for field in FIELDS},
|
'selected_positions': dict(Counter(row['post'] for row in plan)),
|
'selected_key_decision_makers': sum(row['master'] == 1 for row in plan)}
|
summary['customers_with_multiple_key_decision_makers'] = sum(count > 1 for count in key_counts.values())
|
if not options.apply:
|
write_json(ROOT / 'preview.json', {'summary': summary, 'plan': plan})
|
connection.rollback()
|
print(json.dumps(summary, ensure_ascii=False))
|
return
|
stamp = state['now'].strftime('%Y%m%d-%H%M%S')
|
backup = ROOT / ('backup-' + stamp + '.json')
|
write_json(backup, {'at': state['now'], 'database': state['db'], 'rule': RULE,
|
'customers': all_customers, 'contacts': all_contacts, 'plan': plan})
|
for row in changes:
|
cursor.execute('UPDATE crm_customer SET mobile=%s, telephone=%s, email=%s, updater=%s, update_time=%s WHERE id=%s AND deleted=0',
|
tuple(row['new'][field] for field in FIELDS) + ('crm-contact-sync-20261013', state['now'], row['customer_id']))
|
assert cursor.rowcount == 1
|
cursor.execute('SELECT * FROM crm_customer ORDER BY id')
|
after = {row['id']: row for row in cursor.fetchall()}
|
expected = {row['customer_id']: row for row in changes}
|
assert len(after) == len(all_customers)
|
for original in all_customers:
|
current = after[original['id']]
|
allowed = set(FIELDS) | {'updater', 'update_time'} if original['id'] in expected else set()
|
assert all(current[key] == value for key, value in original.items() if key not in allowed)
|
if original['id'] in expected:
|
assert all(current[field] == expected[original['id']]['new'][field] for field in FIELDS)
|
cursor.execute('SELECT * FROM crm_contact ORDER BY id')
|
assert cursor.fetchall() == all_contacts
|
connection.commit()
|
summary.update({'committed': True, 'executed_at': str(state['now']), 'backup': str(backup)})
|
# Reconnect to verify committed values, including unmatched and deleted customer rows.
|
with pymysql.connect(**args) as verify_connection:
|
with verify_connection.cursor() as cursor:
|
cursor.execute('SELECT * FROM crm_customer ORDER BY id')
|
assert {row['id']: row for row in cursor.fetchall()} == after
|
summary['verified_after_commit'] = True
|
write_json(ROOT / 'result.json', summary)
|
with (ROOT / '同步明细.csv').open('w', encoding='utf-8-sig', newline='') as file:
|
writer = csv.writer(file)
|
writer.writerow(['客户ID', '客户', '联系人ID', '选中联系人', '关键决策人', '职位', '候选人数',
|
'原手机号', '同步后手机号', '原电话', '同步后电话', '原邮箱', '同步后邮箱', '变更字段'])
|
for row in plan:
|
writer.writerow([row['customer_id'], row['customer_name'], row['contact_id'], row['contact_name'],
|
row['master'], row['post'], row['candidate_count'],
|
*[value for field in FIELDS for value in (row['old'][field], row['new'][field])],
|
','.join(row['changed'])])
|
print(json.dumps(summary, ensure_ascii=False))
|
except Exception:
|
connection.rollback()
|
raise
|
finally:
|
connection.close()
|
|
if __name__ == '__main__':
|
main()
|
