1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
| import sys,json,csv,argparse,uuid
| from pathlib import Path
| from datetime import datetime,timedelta
| from collections import defaultdict
| sys.stdout.reconfigure(encoding='utf-8')
| ROOT=Path(__file__).resolve().parent
| sys.path.insert(0,str(ROOT.parent/'iqc-import-20261012/deps'))
| import pymysql
|
| def dump(path,value):
| path.write_text(json.dumps(value,ensure_ascii=False,indent=2,default=lambda v: v.hex() if isinstance(v,bytes) else str(v)),encoding='utf-8')
|
| def text(value):
| return str(value or '').strip().replace('管理员','管理人员')
|
| def make_content(row,biz_type,dicts,contacts):
| kind='客户' if biz_type==2 else '线索'
| starts=['资料跟进','档案梳理','信息整理','跟进资料核对']
| parts=[f'{starts[row["id"]%4]}：已梳理{kind}【{text(row["name"])[:100]}】的现有业务档案。']
| facts=[]
| for field,dt,label in [('industry_id','crm_customer_industry','行业'),('level','crm_customer_level','级别'),('source','crm_customer_source','来源')]:
| value=dicts.get((dt,str(row.get(field))))
| if value: facts.append(label+'：'+text(value)[:30])
| if facts: parts.append('；'.join(facts)+'。')
| available=[label for field,label in [('mobile','手机号'),('telephone','电话'),('email','邮箱')] if text(row.get(field))]
| parts.append('档案中已登记'+ '、'.join(available)+'。' if available else '档案中尚未登记手机号、电话和邮箱，后续需补充有效联系方式。')
| if biz_type==2:
| selected=[c for c in contacts.get(row['id'],[]) if c['master']==1]
| if selected:
| c=min(selected,key=lambda c:c['id'])
| parts.append('关键决策人：'+text(c['name'])[:30]+('（'+text(c['post'])[:30]+'）' if text(c['post']) else '')+'。')
| parts.append('系统登记成交状态：'+('已成交' if row['deal_status']==1 else '未成交')+'；后续可结合实际业务核实需求及合作进展。')
| else:
| parts.append('线索尚未转化，后续可核实需求、采购职责及合作条件，具备条件时再推进客户转化。')
| parts.append('本次完成资料梳理并记录待办处理结果。')
| content=''.join(parts)
| assert '管理员' not in content and len(content)<=500
| return content
|
| def is_target(row,biz_type,tomorrow):
| if row['deleted']!=0 or row['owner_user_id'] is None: return False
| if biz_type==1: return row['follow_up_status']==0 and row['transform_status']==0
| return row['follow_up_status']==0 or (row['contact_next_time'] is not None and row['contact_next_time']<tomorrow)
|
| def main():
| parser=argparse.ArgumentParser();parser.add_argument('--apply',action='store_true');opt=parser.parse_args()
| args=dict(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,autocommit=False,connect_timeout=15,read_timeout=120)
| conn=pymysql.connect(**args)
| batch='crm-backlog-'+uuid.uuid4().hex[:24]
| try:
| with conn.cursor() as q:
| q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();assert state['db']=='mom-hm';now=state['now'];tomorrow=now.replace(hour=0,minute=0,second=0,microsecond=0)+timedelta(days=1)
| lock=' FOR UPDATE' if opt.apply else ''
| before={}
| for table in ['crm_customer','crm_clue','crm_contact','crm_contract','crm_receivable','crm_receivable_plan','system_dict_data']:
| q.execute('SELECT * FROM '+table+' ORDER BY id'+lock);before[table]=q.fetchall()
| dictionaries={(r['dict_type'],r['value']):r['label'] for r in before['system_dict_data'] if r['deleted'] in (0,b'\x00') and r['status']==0}
| contacts=defaultdict(list)
| for row in before['crm_contact']:
| if row['deleted']==0: contacts[row['customer_id']].append(row)
| plan=[]
| for table,biz_type in [('crm_customer',2),('crm_clue',1)]:
| for row in before[table]:
| if not is_target(row,biz_type,tomorrow): continue
| next_time=row['contact_next_time'] if row['contact_next_time'] is not None and row['contact_next_time']>=tomorrow else None
| plan.append(dict(table=table,biz_type=biz_type,id=row['id'],name=row['name'],content=make_content(row,biz_type,dictionaries,contacts),next_time=next_time,old=row))
| summary=dict(database=state['db'],customers=sum(p['biz_type']==2 for p in plan),clues=sum(p['biz_type']==1 for p in plan),records=len(plan),cleared_schedules=sum(p['old']['contact_next_time'] is not None and p['next_time'] is None for p in plan),preserved_future_schedules=sum(p['next_time'] is not None for p in plan))
| if not opt.apply:
| dump(ROOT/'backlog-preview.json',{'summary':summary,'plan':plan});conn.rollback()
| print(json.dumps({'summary':summary,'examples':[{'type':p['biz_type'],'content':p['content']} for p in plan[:2]]},ensure_ascii=False));return
| q.execute('SELECT MAX(id) max_id,COUNT(*) count FROM crm_follow_up_record');follow_before=q.fetchone()
| q.execute('SELECT MAX(id) max_id,COUNT(*) count FROM system_operate_log');log_before=q.fetchone()
| backup=ROOT/('backlog-backup-'+now.strftime('%Y%m%d-%H%M%S')+'.json')
| dump(backup,{'database':state['db'],'batch':batch,'before':before,'plan':plan,'follow_before':follow_before,'log_before':log_before})
| # Use a truthful record type instead of recording unperformed phone calls or visits.
| existing=[r for r in before['system_dict_data'] if r['dict_type']=='crm_follow_up_type' and r['label']=='资料梳理' and r['deleted'] in (0,b'\x00') and r['status']==0]
| if existing: follow_type=int(existing[0]['value']);dict_id=None
| else:
| used={int(r['value']) for r in before['system_dict_data'] if r['dict_type']=='crm_follow_up_type' and str(r['value']).isdigit()}
| follow_type=max(used or {0})+1
| q.execute('INSERT INTO system_dict_data (sort,label,value,dict_type,status,color_type,remark,creator,updater,create_time,update_time) VALUES (%s,%s,%s,%s,0,%s,%s,%s,%s,%s,%s)',(follow_type,'资料梳理',str(follow_type),'crm_follow_up_type','info','基于现有档案的资料整理及营销待办处理','0','0',now,now));dict_id=q.lastrowid
| inserted=[]
| for p in plan:
| q.execute('INSERT INTO crm_follow_up_record (biz_type,biz_id,type,content,next_time,pic_urls,business_ids,contact_ids,creator,updater,create_time,update_time,deleted) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,0)',(p['biz_type'],p['id'],follow_type,p['content'],p['next_time'],'','','','0','0',now,now));follow_id=q.lastrowid
| q.execute('UPDATE '+p['table']+' SET follow_up_status=1,contact_last_time=%s,contact_last_content=%s,contact_next_time=%s,updater=%s,update_time=%s WHERE id=%s AND deleted=0',(now,p['content'],p['next_time'],'0',now,p['id']));assert q.rowcount==1
| kind='客户' if p['biz_type']==2 else '线索'
| action=kind+'跟进【'+text(p['name'])+'】：批量完成资料梳理，新增跟进记录并更新待办状态。'
| extra=json.dumps(dict(batch=batch,execution='database_batch',source='user_request',follow_record_id=follow_id,content=p['content'],old_follow_up_status=p['old']['follow_up_status'],old_next_time=str(p['old']['contact_next_time']),next_time=str(p['next_time'])),ensure_ascii=False)
| assert '管理员' not in action+extra and len(extra)<=2000
| # user_id=0 / creator=0 explicitly denotes batch execution, without impersonating a salesperson.
| q.execute('INSERT INTO system_operate_log (trace_id,user_id,user_type,type,sub_type,biz_id,action,success,extra,request_method,request_url,user_ip,user_agent,creator,updater,create_time,update_time,deleted,tenant_id) VALUES (%s,0,2,%s,%s,%s,%s,1,%s,%s,%s,NULL,%s,%s,%s,%s,%s,0,0)',(batch,'CRM '+kind,kind+'跟进',p['id'],action,extra,'BATCH','batch://crm/backlog/follow-up','Codex database batch','0','0',now,now));inserted.append(dict(biz_type=p['biz_type'],biz_id=p['id'],follow_id=follow_id,log_id=q.lastrowid))
| expected={(p['table'],p['id']):p for p in plan};allowed={'follow_up_status','contact_last_time','contact_last_content','contact_next_time','updater','update_time'}
| for table in ['crm_customer','crm_clue','crm_contact','crm_contract','crm_receivable','crm_receivable_plan']:
| q.execute('SELECT * FROM '+table+' ORDER BY id');after=q.fetchall();assert len(after)==len(before[table])
| for old,current in zip(before[table],after):
| p=expected.get((table,old['id']));assert all(current[k]==v for k,v in old.items() if not p or k not in allowed)
| if p: assert current['follow_up_status']==1 and current['contact_last_content']==p['content'] and current['contact_next_time']==p['next_time']
| for item in inserted:
| q.execute('SELECT biz_type,biz_id,content FROM crm_follow_up_record WHERE id=%s',(item['follow_id'],));r=q.fetchone();p=expected[('crm_customer' if item['biz_type']==2 else 'crm_clue',item['biz_id'])];assert r==dict(biz_type=p['biz_type'],biz_id=p['id'],content=p['content'])
| q.execute('SELECT COUNT(*) count FROM system_operate_log WHERE trace_id=%s',(batch,));assert q.fetchone()['count']==len(plan)
| dump(ROOT/'backlog-inserted-ids.json',dict(batch=batch,dict_id=dict_id,records=inserted))
| conn.commit()
| with pymysql.connect(**args) as verify:
| with verify.cursor() as v:
| v.execute('SELECT COUNT(*) count FROM system_operate_log WHERE trace_id=%s',(batch,));assert v.fetchone()['count']==len(plan)
| for table,biz_type in [('crm_customer',2),('crm_clue',1)]:
| v.execute('SELECT * FROM '+table+' ORDER BY id');rows=v.fetchall();assert not any(is_target(r,biz_type,tomorrow) for r in rows)
| summary.update(committed=True,verified_after_commit=True,batch=batch,backup=str(backup),follow_type=follow_type,follow_type_label='资料梳理',operation_logs=len(inserted),executed_at=str(now))
| dump(ROOT/'backlog-result.json',summary)
| with (ROOT/'待办跟进明细.csv').open('w',encoding='utf-8-sig',newline='') as f:
| w=csv.writer(f);w.writerow(['类型','编号','名称','跟进内容','下次联系时间','跟进记录ID','操作日志ID'])
| for p,ids in zip(plan,inserted):w.writerow(['客户' if p['biz_type']==2 else '线索',p['id'],p['name'],p['content'],p['next_time'],ids['follow_id'],ids['log_id']])
| print(json.dumps(summary,ensure_ascii=False))
| except Exception:
| conn.rollback();raise
| finally:conn.close()
|
| if __name__=='__main__':main()
|
|
