"""Authorized CRM data cleanup, transaction preview by default."""
|
import sys,json,re,hashlib,collections
|
from pathlib import Path
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent;apply='--apply' in sys.argv
|
c=pymysql.connect(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,read_timeout=90)
|
mutable=['crm_business','crm_business_product','crm_contact_business','crm_follow_up_record','crm_permission','crm_sale_quotation','crm_contract']
|
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
def fingerprint(v):return hashlib.sha256(json.dumps(v,ensure_ascii=False,sort_keys=True,default=str).encode()).hexdigest()
|
def is_deleted(v):return v in (1,True,b'\x01')
|
changes=[];sql_log=[];before={};deleted=[];flow_updates=[]
|
with c.cursor() as q:
|
try:
|
q.execute('SELECT GET_LOCK(%s,10)',('crm-business-flow-cleanup',));assert list(q.fetchone().values())==[1]
|
q.execute('SELECT * FROM crm_business_status_type WHERE deleted=0');flows=q.fetchall();flowids={x['name']:x['id'] for x in flows};assert {'标准直销流程','电销流程'}<=set(flowids)
|
direct=flowids['标准直销流程'];tele=flowids['电销流程']
|
q.execute('SELECT * FROM crm_business_status WHERE deleted=0');stages={x['id']:x for x in q.fetchall()}
|
ordered={fid:sorted([x for x in stages.values() if x['type_id']==fid],key=lambda x:x['sort']) for fid in [direct,tele]}
|
stage_names={fid:{x['name']:x['id'] for x in ordered[fid]} for fid in ordered}
|
q.execute('SELECT id,name FROM crm_customer');customer_names={x['id']:x['name'] for x in q.fetchall()}
|
for t in mutable:q.execute('SELECT * FROM '+t+' ORDER BY id FOR UPDATE');before[t]=q.fetchall()
|
active=[x for x in before['crm_business'] if not x['deleted']]
|
deleted=[x['id'] for x in active if '北京同仁堂' in x['name'] or '北京同仁堂' in customer_names.get(x['customer_id'],'')];deleted_set=set(deleted)
|
assert len(deleted)==463,('Unexpected cleanup scope; audit again',len(deleted))
|
# Preserve all operational tables, customers, prices, products and receivable plans.
|
schema=json.loads(Path('scripts/october-chain-20261014/schema-snapshot.json').read_text(encoding='utf8'))
|
protected=sorted(set(t for t,s in schema.items() if isinstance(s,dict) and 'columns' in s)-set(mutable)|{'crm_customer','crm_contact','crm_business_status','crm_business_status_type'})
|
protected_hashes={}
|
for t in protected:q.execute('SELECT * FROM `'+t+'` ORDER BY id');protected_hashes[t]=fingerprint(q.fetchall())
|
dump('before-'+('apply' if apply else 'preview')+'.json',{'tables':before,'protected_hashes':protected_hashes,'delete_ids':deleted})
|
def update(t,row,v):
|
v={k:z for k,z in v.items() if row[k]!=z}
|
if not v:return
|
sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k in v)+',update_time=update_time WHERE id=%s'
|
args=tuple(v.values())+(row['id'],)
|
changes.append({'table':t,'id':row['id'],'before':{k:row[k] for k in v},'after':v});sql_log.append(q.mogrify(sql,args)+';')
|
# Unbind before deleting. business_id is optional on both document types.
|
for t in ['crm_sale_quotation','crm_contract']:
|
for row in before[t]:
|
if not is_deleted(row['deleted']) and row['business_id'] in deleted_set:update(t,row,{'business_id':None})
|
for t in ['crm_business_product','crm_contact_business']:
|
for row in before[t]:
|
if not row['deleted'] and row['business_id'] in deleted_set:update(t,row,{'deleted':1})
|
for row in before['crm_permission']:
|
if not row['deleted'] and row['biz_type']==4 and row['biz_id'] in deleted_set:update('crm_permission',row,{'deleted':1})
|
for row in before['crm_follow_up_record']:
|
if row['deleted']:continue
|
v={}
|
if row['biz_type']==4 and row['biz_id'] in deleted_set:v['deleted']=1
|
value=row['business_ids']
|
if value:
|
isjson=value.strip().startswith('[');ids=json.loads(value) if isjson else [int(x) for x in value.split(',') if x.strip()]
|
kept=[x for x in ids if int(x) not in deleted_set]
|
if kept!=ids:v['business_ids']=(json.dumps(kept,separators=(',',':')) if isjson else ','.join(map(str,kept))) if kept else None
|
update('crm_follow_up_record',row,v)
|
for row in active:
|
if row['id'] in deleted_set:update('crm_business',row,{'deleted':1});continue
|
text=row['name']+' '+customer_names.get(row['customer_id'],'');target=tele if re.search(r'电商|电子商务|399\s*零号店',text) else direct
|
current=stages.get(row['status_id']);stage=row['status_id']
|
if not current or current['type_id']!=target:
|
if row['end_status']==1:stage=stage_names[target].get('赢单',ordered[target][-1]['id'])
|
elif current and current['name'] in stage_names[target]:stage=stage_names[target][current['name']]
|
else:
|
name=current['name'] if current else ''
|
mapping={'初步接洽':'线索初触达','订单报价':'方案递送','赢单':'待签约'} if target==tele else {'线索初触达':'初步接洽','方案递送':'订单报价','待现场洽谈':'订单报价','待签约':'商务谈判'}
|
stage=stage_names[target].get(mapping.get(name,''),ordered[target][0]['id'])
|
if target!=row['status_type_id'] or stage!=row['status_id']:flow_updates.append({'id':row['id'],'name':row['name'],'flow_id':target,'stage_id':stage,'end_status':row['end_status']})
|
update('crm_business',row,{'status_type_id':target,'status_id':stage})
|
# Batch identical changes, reducing locks and network round trips for large imported datasets.
|
batches=collections.defaultdict(list)
|
for change in changes:batches[(change['table'],tuple(change['after'].items()))].append(change['id'])
|
for (t,values),ids in batches.items():
|
for start in range(0,len(ids),500):
|
chunk=ids[start:start+500]
|
sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k,v in values)+',update_time=update_time WHERE id IN ('+','.join(['%s']*len(chunk))+')'
|
q.execute(sql,tuple(v for k,v in values)+tuple(chunk));assert q.rowcount==len(chunk)
|
# Check exact expected rows and unchanged operational tables in the same transaction.
|
bykey={(x['table'],x['id']):x['after'] for x in changes}
|
for t,rr in before.items():
|
q.execute('SELECT * FROM '+t+' ORDER BY id');after=q.fetchall();assert len(after)==len(rr)
|
for old,new in zip(rr,after):assert new=={**old,**bykey.get((t,old['id']),{})},(t,old['id'])
|
for t,h in protected_hashes.items():q.execute('SELECT * FROM `'+t+'` ORDER BY id');assert fingerprint(q.fetchall())==h,(t,'Unexpected operational change')
|
q.execute('SELECT b.*,c.name customer_name,s.type_id FROM crm_business b LEFT JOIN crm_customer c ON c.id=b.customer_id LEFT JOIN crm_business_status s ON s.id=b.status_id AND s.deleted=0 WHERE b.deleted=0');remaining=q.fetchall()
|
counts=collections.Counter()
|
for row in remaining:
|
text=row['name']+' '+(row['customer_name'] or '');assert '北京同仁堂' not in text
|
target=tele if re.search(r'电商|电子商务|399\s*零号店',text) else direct
|
assert row['status_type_id']==row['type_id']==target;counts[target]+=1
|
for t in ['crm_sale_quotation','crm_contract','crm_business_product','crm_contact_business']:
|
q.execute('SELECT COUNT(*) n FROM '+t+' x JOIN crm_business b ON b.id=x.business_id WHERE x.deleted=0 AND b.deleted=1 AND b.id IN ('+','.join(['%s']*len(deleted))+')',tuple(deleted));assert q.fetchone()['n']==0,t
|
summary={'applied':apply,'deleted_businesses':len(deleted),'remaining_businesses':len(remaining),'tele_businesses':counts[tele],'direct_businesses':counts[direct],'flow_updates':flow_updates,'change_counts':dict(collections.Counter(x['table'] for x in changes)),'changes':changes,'protected_hashes':protected_hashes}
|
dump('result-'+('applied' if apply else 'preview')+'.json',summary)
|
(P/('applied.sql' if apply else 'preview.sql')).write_text('\n'.join(sql_log),encoding='utf8')
|
if apply:c.commit()
|
else:c.rollback()
|
print(json.dumps({k:v for k,v in summary.items() if k not in ['changes','protected_hashes','flow_updates']},ensure_ascii=False))
|
except:
|
c.rollback();raise
|
finally:q.execute('SELECT RELEASE_LOCK(%s)',('crm-business-flow-cleanup',))
|
c.close()
|
