"""Atomic migration of QC configuration, existing line standards and reviewed reports."""
import sys,json,re,hashlib,collections,copy
from pathlib import Path
from decimal import Decimal as D
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
audit=json.loads((P/'audit.json').read_text(encoding='utf8'));plan=json.loads((P/'plan.json').read_text(encoding='utf8'));reports=json.loads((P/'report-output.json').read_text(encoding='utf8'))
upgrade=json.loads((P/'report-template-upgrade.json').read_text(encoding='utf8'))
assert len(reports)==12
def canon(v):return json.loads(json.dumps(v,ensure_ascii=False,default=str))
def fingerprint(v):return hashlib.sha256(json.dumps(v,ensure_ascii=False,sort_keys=True,default=str).encode()).hexdigest()
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
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)
allowed={'mes_qc_indicator','mes_qc_template_indicator','mes_qc_iqc_line','mes_qc_ipqc_line','mes_qc_oqc_line','mes_qc_rqc_line','mes_qc_indicator_result_detail','qc_report_instance','qc_report_template','system_storage_attachment'}
merged={};new_units=[];sql_log=[];before={};unitids={}
for ch in plan['changes']:
 assert ch['table'] in allowed
 merged.setdefault((ch['table'],ch['id']),{}).update(ch['after'])
with c.cursor() as q:
 try:
  q.execute('SELECT GET_LOCK(%s,10)',('qc-indicator-standard-optimize',));assert list(q.fetchone().values())==[1]
  for t in allowed|{'mes_md_unit_measure','qc_report_template_version'}:
   q.execute('SELECT * FROM `'+t+'` ORDER BY id FOR UPDATE');before[t]=q.fetchall()
  # Refuse to apply a stale plan; inactive rows are preserved too.
  for t in allowed-{'system_storage_attachment'}:
   assert canon(before[t])==audit[t]['rows'],('Data changed after audit; regenerate plan',t)
  q.execute('SHOW FULL COLUMNS FROM system_storage_attachment');attachment_fields={x['Field'] for x in q.fetchall()}
  for name in plan['units']:
   found=[x for x in before['mes_md_unit_measure'] if x['name']==name and x['deleted'] in (0,b'\x00') and x['status']==0]
   if found:unitids[name]=found[0]['id'];continue
   code='QC_UNIT_'+hashlib.sha256(name.encode()).hexdigest()[:12].upper()
   assert not any(x['code']==code for x in before['mes_md_unit_measure'])
   sql='INSERT INTO mes_md_unit_measure (code,name,primary_flag,primary_id,change_rate,status,remark,creator,updater) VALUES (%s,%s,1,NULL,1,0,%s,%s,%s)'
   args=(code,name,'质量检测专用计量单位；依据现有方案标准及检验记录单位规范化。','1','1')
   q.execute(sql,args);unitids[name]=q.lastrowid;new_units.append({'id':q.lastrowid,'name':name,'code':code});sql_log.append(q.mogrify(sql,args)+';')
  for assignment in plan['unit_assignments']:
   merged.setdefault((assignment['table'],assignment['id']),{})['unit_measure_id']=unitids[assignment['unit_name']]
  existing_versions=before['qc_report_template_version']
  assert not any(x['template_id']==upgrade['template_id'] and x['version']==upgrade['version'] for x in existing_versions)
  old_version=next(x for x in existing_versions if x['template_id']==upgrade['template_id'] and x['version']==upgrade['source_version'] and x['deleted'] in (0,b'\x00'))
  sql='INSERT INTO qc_report_template_version (template_id,version,schema_json,status,description,creator,updater) VALUES (%s,%s,%s,%s,%s,%s,%s)'
  args=(upgrade['template_id'],upgrade['version'],json.dumps(upgrade['schema'],ensure_ascii=False,separators=(',',':')),old_version['status'],'质检数据规范化：展示完整标准要求、允许区间和检测单位；保留原模板版本。','1','1')
  q.execute(sql,args);new_version_id=q.lastrowid;sql_log.append(q.mogrify(sql,args)+';')
  merged[('qc_report_template',upgrade['template_id'])]={'current_version':upgrade['version']}
  for report in reports:
   assert report['html'].strip() and '{{' not in report['html']
   merged[('qc_report_instance',report['id'])]={'data_snapshot':json.dumps(report['snapshot'],ensure_ascii=False,separators=(',',':')),'render_html':report['html'],'template_version':report['template_version']}
  # Invalidate only stale PDF attachment bindings, retaining physical blobs and the previous report in the backup.
  reportids={x['id'] for x in reports};old_attachments=[]
  for row in before['system_storage_attachment']:
   if row.get('record_type')=='qc_report_instance' and str(row.get('record_id')) in {str(x) for x in reportids} and row['deleted'] in (0,b'\x00'):
    merged[('system_storage_attachment',row['id'])]={'deleted':1};old_attachments.append(row['id'])
  protected_tables=['mes_qc_iqc','mes_qc_ipqc','mes_qc_oqc','mes_qc_rqc','mes_qc_indicator_result','mes_qc_template','mes_qc_template_item','erp_sale_order','erp_sale_order_items','crm_contract','crm_sale_quotation','crm_receivable_plan','mes_pro_mps','mes_pro_work_order','mes_pro_task','mes_pro_feedback','mes_wm_batch','mes_wm_material_stock','mes_wm_transaction','mes_wm_product_sales','mes_wm_product_sales_line']
  protected={}
  for t in protected_tables:q.execute('SELECT * FROM `'+t+'` ORDER BY id');protected[t]=fingerprint(q.fetchall())
  dump('before-'+('apply' if apply else 'preview')+'.json',{'tables':before,'protected_hashes':protected})
  schema={t:{x['Field']:x for x in audit[t]['columns']} for t in allowed if t in audit}
  q.execute('SHOW FULL COLUMNS FROM system_storage_attachment');schema['system_storage_attachment']={x['Field']:x for x in q.fetchall()}
  byid={t:{x['id']:x for x in rr} for t,rr in before.items()};changes=[];batches=collections.defaultdict(list)
  for (t,rid),v in merged.items():
   row=byid[t][rid];v={k:z for k,z in v.items() if row[k]!=z}
   if not v:continue
   assert set(v)<=set(schema[t])
   for k,z in v.items():
    if z is None:assert schema[t][k]['Null']=='YES'
    if isinstance(z,str):
     m=re.match(r'(?:var)?char\((\d+)\)',schema[t][k]['Type']);assert not m or len(z)<=int(m[1]),(t,rid,k,len(z))
    if schema[t][k]['Type'].startswith('decimal') and z is not None:v[k]=D(z)
   v={k:z for k,z in v.items() if row[k]!=z}
   if not v:continue
   changes.append({'table':t,'id':rid,'before':{k:row[k] for k in v},'after':v})
   batches[(t,tuple(v.items()))].append(rid)
  for (t,values),ids in batches.items():
   for start in range(0,len(ids),400):
    chunk=ids[start:start+400]
    sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k,z in values)+',update_time=update_time WHERE id IN ('+','.join(['%s']*len(chunk))+')'
    args=tuple(z for k,z in values)+tuple(chunk);q.execute(sql,args);assert q.rowcount==len(chunk),(t,q.rowcount,len(chunk));sql_log.append(q.mogrify(sql,args)+';')
  expected={(x['table'],x['id']):x['after'] for x in changes}
  for t,rr in before.items():
   if t in ['mes_md_unit_measure','qc_report_template_version']:continue
   q.execute('SELECT * FROM `'+t+'` ORDER BY id');after=q.fetchall();assert len(after)==len(rr)
   for old,new in zip(rr,after):
    exp={**old,**expected.get((t,old['id']),{})}
    for k,z in new.items():
     if isinstance(z,D):assert z==D(exp[k]),(t,old['id'],k)
     elif schema[t].get(k,{}).get('Type','').startswith('bit') and k in expected.get((t,old['id']),{}):assert z==bytes([exp[k]])
     else:assert z==exp[k],(t,old['id'],k)
  for t,h in protected.items():q.execute('SELECT * FROM `'+t+'` ORDER BY id');assert fingerprint(q.fetchall())==h,t
  q.execute('SELECT id,parent_id,item_type,result_type FROM mes_qc_indicator WHERE deleted=0');ind={x['id']:x for x in q.fetchall()}
  for row in ind.values():
   if row['item_type']==2:assert row['result_type'] is None and row['parent_id']==0
   else:assert row['result_type'] in [1,2,3] and row['parent_id'] in ind and ind[row['parent_id']]['item_type']==2
  for t in ['mes_qc_template_indicator','mes_qc_iqc_line','mes_qc_ipqc_line','mes_qc_oqc_line','mes_qc_indicator_result_detail']:
   q.execute('SELECT indicator_id FROM `'+t+'` WHERE deleted=0');assert all(x['indicator_id'] in ind for x in q.fetchall()),t
  q.execute('SELECT * FROM qc_report_template_version ORDER BY id');versions=q.fetchall();assert len(versions)==len(existing_versions)+1
  assert versions[:-1]==existing_versions
  result={'applied':apply,'changes':changes,'counts':dict(collections.Counter(x['table'] for x in changes)),'new_units':new_units,'unit_ids':unitids,'report_template_version':{'id':new_version_id,'template_id':upgrade['template_id'],'version':upgrade['version']},'reports_updated':len(reports),'invalidated_pdf_attachments':old_attachments,'protected_hashes':protected,'verified_before_commit':True}
  dump('result-'+('applied' if apply else 'preview')+'.json',result)
  (P/('applied.sql' if apply else 'preview.sql')).write_text('START TRANSACTION;\n'+'\n'.join(sql_log)+'\nCOMMIT;\n',encoding='utf8')
  if apply:c.commit()
  else:c.rollback()
  print(json.dumps({k:result[k] for k in ['applied','counts','new_units','reports_updated','invalidated_pdf_attachments','verified_before_commit']},ensure_ascii=False))
 except:c.rollback();raise
 finally:q.execute('SELECT RELEASE_LOCK(%s)',('qc-indicator-standard-optimize',))
c.close()