"""Normalize only the isolated October supplement; retain source and measurement provenance."""
|
import sys,json,re,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;TAG='OCT0814-DEMO';NOTE='10月8日至14日业务链演示补充'
|
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)
|
schema=json.loads((P/'schema-snapshot.json').read_text(encoding='utf8'))
|
rules=json.loads((P/'number-rules.json').read_text(encoding='utf8'))
|
manifest=json.loads((P/'result-applied.json').read_text(encoding='utf8'))
|
def save(name,obj): (P/name).write_text(json.dumps(obj,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
def canon(obj):return json.loads(json.dumps(obj,default=str,ensure_ascii=False))
|
rows={};mapping={};reservations=[];changes=[];log=[]
|
rule_ids={'mes_wm_arrival_notice':13,'mes_wm_item_receipt':14,'mes_wm_batch':19,'mes_wm_product_sales':20,'mes_wm_sales_notice':28,'mes_pro_work_order':30,'mes_pro_task':31,'mes_pro_feedback':33,'mes_qc_iqc':40,'mes_qc_ipqc':41,'mes_qc_oqc':42,'mes_qc_indicator_result':45,'mes_wm_product_issue':64,'mes_pro_mps':65,'mes_qc_template':67}
|
q=c.cursor()
|
try:
|
q.execute('SELECT GET_LOCK(%s,10)',(TAG+'-normalize',));assert list(q.fetchone().values())==[1]
|
for t,s in schema.items():
|
if not isinstance(s,dict) or 'columns' not in s:continue
|
cols=[x['Field'] for x in s['columns'] if re.match(r'(?:var)?char|text',x['Type'])]
|
if not cols:continue
|
clause=' OR '.join('`'+k+'` LIKE %s' for k in cols)
|
q.execute('SELECT * FROM `'+t+'` WHERE '+clause,tuple('%'+TAG+'%' for _ in cols));found={x['id']:x for x in q.fetchall()}
|
if 'remark' in cols:
|
q.execute('SELECT * FROM `'+t+'` WHERE remark LIKE %s',(NOTE+'%',));found.update({x['id']:x for x in q.fetchall()})
|
if found:rows[t]=list(found.values())
|
assert len(rows['erp_sale_order'])==4 and len(rows['mes_wm_batch'])==42
|
save('before-normalize-'+('apply' if apply else 'dry-run')+'.json',rows)
|
q.execute('SELECT id,name FROM mdm_item');items={x['id']:x['name'] for x in q.fetchall()}
|
q.execute('SELECT id,name FROM mes_pro_process');processes={x['id']:x['name'] for x in q.fetchall()}
|
q.execute('SELECT id,name FROM crm_customer');customers={x['id']:x['name'] for x in q.fetchall()}
|
def occupied(t,col,prefix):
|
q.execute('SELECT `'+col+'` v FROM `'+t+'` WHERE `'+col+'` LIKE %s',(prefix+'%',));return [x['v'] for x in q.fetchall()]
|
def next_code(t,col,prefix,width,start=1,extra=()):
|
vals=occupied(t,col,prefix)+list(extra)+list(mapping.values())
|
nums=[int(v[len(prefix):]) for v in vals if v and v.startswith(prefix) and v[len(prefix):].isdigit()]
|
n=max(nums+[start-1])+1;assert n<10**width
|
return prefix+str(n).zfill(width),n
|
# Marketing records follow the numbering conventions already used by the business data.
|
for t,col,prefix,width in [('erp_sale_order','no','26',6),('crm_sale_quotation','no','BJ-202610-L',3),('crm_contract','no','HT-202610-L',3)]:
|
for row in rows[t]:
|
new,n=next_code(t,col,prefix,width);mapping[row[col]]=new
|
for t,rid in rule_ids.items():
|
rule=next(x for x in rules['mes_md_auto_code_rule'] if x['id']==rid)
|
parts=sorted([x for x in rules['mes_md_auto_code_part'] if x['rule_id']==rid],key=lambda x:x['sort'])
|
for row in rows[t]:
|
day=row['create_time'].strftime('%Y%m%d');prefix='';width=None;start=1
|
for part in parts:
|
if part['type']==3:prefix+=part['fix_character'][:part['length']]
|
elif part['type']==2:prefix+=day
|
elif part['type']==4:width=part['length'];start=part['serial_start_no']
|
else:raise ValueError((t,part))
|
q.execute('SELECT result FROM mes_md_auto_code_record WHERE rule_id=%s',(rid,));used=[x['result'] for x in q.fetchall()]
|
# IR_ is shared by receipts and indicator results: avoid cross-module string ambiguity.
|
if prefix.startswith('IR_'):used+=occupied('mes_wm_item_receipt','code',prefix)+occupied('mes_qc_indicator_result','code',prefix)
|
new,n=next_code(t,'code',prefix,width,start,used)
|
if len(new)<rule['max_length']:new=new.rjust(rule['max_length'],rule['padded_char'])
|
mapping[row['code']]=new
|
reservations.append({'rule_id':rid,'result':new,'serial_no':n,'input_char':None,'creator':row['creator'],'updater':row['updater'],'create_time':row['create_time'],'update_time':row['create_time'],'deleted':0})
|
for t,prefix,width in [('mes_wm_item_consume','LLD',4),('mes_wm_product_produce','RKD',4),('mes_wm_product_receipt','PR_',3)]:
|
for row in rows[t]:
|
new,n=next_code(t,'code',prefix+row['create_time'].strftime('%Y%m%d'),width);mapping[row['code']]=new
|
for row in rows.get('mes_pro_work_order_param_record',[]):mapping[row['record_no']]='TASK-'+str(row['task_id'])
|
for row in rows['mes_wm_product_sales']:mapping[row['shipping_number']]=mapping[row['code']]
|
batch_codes={x['id']:mapping[x['code']] for x in rows['mes_wm_batch']}
|
for t,rr in rows.items():
|
for row in rr:
|
v={}
|
for k,old in row.items():
|
if isinstance(old,str) and old in mapping:v[k]=mapping[old]
|
if 'lot_number' in row and isinstance(row['lot_number'],str) and TAG in row['lot_number']:
|
bid=row['id'] if t=='mes_wm_batch' else row.get('batch_id');v['lot_number']=batch_codes[bid]
|
if 'name' in row and isinstance(row['name'],str) and '演示' in row['name']:
|
item=items.get(row.get('item_id') or row.get('product_id'),'');proc=processes.get(row.get('process_id'),'')
|
if t.startswith('crm_'):
|
link=next(x for x in manifest['links'] if row['id']==x[{'crm_business':'business','crm_sale_quotation':'quotation_id','crm_contract':'contract_id'}[t]])
|
suffix={'crm_business':'销售商机','crm_sale_quotation':'销售报价','crm_contract':'销售合同'}[t];name=link['product']+suffix
|
elif t=='mes_qc_template':name={'OCT0814-DEMO-IQC':'来料外观及数量核验方案','OCT0814-DEMO-IPQC':'燕窝过程检验方案','OCT0814-DEMO-OQC':'燕窝出货检验方案'}[row['code']]
|
elif t.startswith('mes_qc_'):name=item+{'mes_qc_iqc':'来料检验','mes_qc_ipqc':'过程检验','mes_qc_oqc':'出货检验'}[t]
|
elif t=='mes_pro_task':name=item+'-'+proc
|
elif t=='mes_pro_work_order':name=item+'生产工单'
|
else:name=row['name'].replace('演示','').replace('工序投入消耗',proc+'工序投料').replace('工序产出',proc+'工序产出')
|
assert name and '演示' not in name;v['name']=name
|
if t=='mes_wm_product_sales':v['carrier']='自有车辆配送'
|
if not v:continue
|
# Only named fields and exact code references may change; quantities, money and audit times stay intact.
|
assert not set(v)&{'remark','content','value','param_value','sampling_location','check_method','status','quantity','total_price'}
|
sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k in v)+',update_time=update_time WHERE id=%s'
|
args=tuple(v.values())+(row['id'],);q.execute(sql,args);assert q.rowcount==1
|
log.append(q.mogrify(sql,args)+';');changes.append({'table':t,'id':row['id'],'before':{k:row[k] for k in v},'after':v})
|
for v in reservations:
|
cols=list(v);sql='INSERT INTO mes_md_auto_code_record (`'+'`,`'.join(cols)+'`) VALUES ('+','.join(['%s']*len(cols))+')'
|
q.execute(sql,tuple(v.values()));log.append(q.mogrify(sql,tuple(v.values()))+';')
|
# Re-read every scoped row and prove that no unrelated field changed.
|
bykey={(x['table'],x['id']):x['after'] for x in changes}
|
for t,rr in rows.items():
|
for row in rr:
|
q.execute('SELECT * FROM `'+t+'` WHERE id=%s',(row['id'],));actual=q.fetchone();expected={**row,**bykey.get((t,row['id']),{})};assert actual==expected,(t,row['id'])
|
for k,value in actual.items():
|
if isinstance(value,str) and (k in ['code','no','record_no','lot_number','shipping_number'] or k.endswith('_code') or k.endswith('_no')):assert TAG not in value,(t,k,value)
|
if isinstance(actual.get('name'),str):assert '演示' not in actual['name'],(t,actual['id'])
|
summary={'applied':apply,'changed_rows':len(changes),'code_count':len(mapping),'number_reservations':len(reservations),'changes':changes,'mapping':mapping,'scoped_ids':{t:[x['id'] for x in rr] for t,rr in rows.items()}}
|
save('normalize-'+('applied' if apply else 'preview')+'.json',summary)
|
(P/('normalize-applied.sql' if apply else 'normalize-preview.sql')).write_text('\n'.join(log),encoding='utf8')
|
if apply:c.commit()
|
else:c.rollback()
|
print(json.dumps({k:summary[k] for k in ['applied','changed_rows','code_count','number_reservations']},ensure_ascii=False))
|
except:
|
c.rollback();raise
|
finally:
|
q.execute('SELECT RELEASE_LOCK(%s)',(TAG+'-normalize',));c.close()
|
