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()