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