import json,gzip,hashlib,collections,os,time,re,sys
from pathlib import Path
from dm_sql_contract import ident,literal,target_type,default_sql,value_sql
r=Path(__file__).resolve().parents[1];b=json.loads((r/'private/mysql-upgraded-baseline.json').read_text());m=json.loads((r/'private/upgraded-export-manifest.json').read_text())
source_info=json.loads((r/'private/source-verification.json').read_text());source_sha256=source_info['sha256']
if source_sha256!='fa3a82a129a8366535c3303c3eb92eda097b4f7685e77a2adf5fc83a4b188d3a' or source_info.get('gzip_ok') is not True:raise SystemExit('Verified second source backup required')
from mysql_encoding import validate_encoding
validate_encoding(b.get('encoding')); validate_encoding(m.get('encoding'))
schema_only='--schema-only' in sys.argv
out=r/'private/dm-package';out.mkdir(mode=0o700,exist_ok=schema_only);(out/'sql').mkdir(exist_ok=schema_only);(out/'lob').mkdir(exist_ok=schema_only)
cols=collections.defaultdict(list);indexes=collections.defaultdict(list)
for c in b['columns']:cols[c['TABLE_NAME']].append(c)
for ix in b['indexes']:indexes[(ix['TABLE_NAME'],ix['INDEX_NAME'])].append(ix)
tables={t['TABLE_NAME']:t for t in b['tables']};mapping=[];schema=[];post=[];comments=[]
for name,cs in cols.items():
 defs=[]
 for c in cs:
  dt=target_type(c);auto='auto_increment' in c['EXTRA'];definition=ident(c['COLUMN_NAME'])+' '+dt
  if auto:definition+=' IDENTITY(%d,1)'%(int(tables[name]['AUTO_INCREMENT'] or 1))
  definition+=default_sql(c)
  if 'on update CURRENT_TIMESTAMP' in c['EXTRA']:definition+=' ON UPDATE SYSDATE'
  if c['IS_NULLABLE']=='NO':definition+=' NOT NULL'
  if 'unsigned' in c['COLUMN_TYPE']:
   upper={'tinyint':255,'int':4294967295,'bigint':18446744073709551615}.get(c['DATA_TYPE'])
   if upper is not None:
    if not auto:definition+=' CHECK('+ident(c['COLUMN_NAME'])+' BETWEEN 0 AND '+str(upper)+')'
    else:
     tr=ident('U_'+hashlib.sha256((name+'.'+c['COLUMN_NAME']).encode()).hexdigest()[:24])
     v=':NEW.'+ident(c['COLUMN_NAME'])
     post.append('CREATE TRIGGER '+tr+' BEFORE INSERT ON '+ident(name)+' FOR EACH ROW\nBEGIN\n IF '+v+' < 0 OR '+v+' > '+str(upper)+" THEN RAISE_APPLICATION_ERROR(-20001,'unsigned identity out of range'); END IF;\nEND;\n/")
  if c['DATA_TYPE']=='enum':
   if c['COLUMN_TYPE']!="enum('success','failed','denied','partial')":raise ValueError('Unreviewed enum declaration')
   definition+=' CHECK('+ident(c['COLUMN_NAME'])+" IN ('success','failed','denied','partial'))"
  defs.append(definition)
  mapping.append({**c,'TARGET_TYPE':dt,'TARGET_IDENTITY_SEED':int(tables[name]['AUTO_INCREMENT'] or 1)if auto else None,'IDENTITY_RUNTIME_LIMIT':'signed64; this snapshot safe'if auto and c['DATA_TYPE']=='bigint' and 'unsigned'in c['COLUMN_TYPE'] else None})
  if c['COLUMN_COMMENT']:
   comment=c['COLUMN_COMMENT'].replace('\r',' ').replace('\n',' ')
   comments.append('COMMENT ON COLUMN '+ident(name)+'.'+ident(c['COLUMN_NAME'])+' IS '+literal(comment)+';')
 pk=indexes.get((name,'PRIMARY'),[])
 if pk:defs.append('PRIMARY KEY ('+','.join(ident(i['COLUMN_NAME'])for i in pk)+')')
 schema.append('CREATE TABLE '+ident(name)+' (\n  '+',\n  '.join(defs)+'\n);')
seen_indexes={};index_mapping=[]
for (name,ixname),items in indexes.items():
 if ixname=='PRIMARY':
  sig=(name,tuple((i['COLUMN_NAME'],i['SUB_PART'])for i in items));seen_indexes[sig]={'target':'PRIMARY','unique':True}
  index_mapping.append({'table':name,'source_index':ixname,'target':'PRIMARY','source_unique':True,'equivalent_reuse':False})
for (name,ixname),items in sorted(indexes.items(),key=lambda kv:(kv[1][0]['NON_UNIQUE'],kv[0])):
 if ixname=='PRIMARY':continue
 sig=(name,tuple((i['COLUMN_NAME'],i['SUB_PART'])for i in items))
 if sig in seen_indexes:
  found=seen_indexes[sig]
  if not items[0]['NON_UNIQUE'] and not found['unique']:raise ValueError('Unique index would be weakened')
  index_mapping.append({'table':name,'source_index':ixname,'target':found['target'],'source_unique':not bool(items[0]['NON_UNIQUE']),'equivalent_reuse':True});continue
 ixid='I_'+hashlib.sha256((name+'.'+ixname).encode()).hexdigest()[:24].upper()
 seen_indexes[sig]={'target':ixid,'unique':not bool(items[0]['NON_UNIQUE'])}
 index_mapping.append({'table':name,'source_index':ixname,'target':ixid,'source_unique':not bool(items[0]['NON_UNIQUE']),'equivalent_reuse':False})
 fields=[]
 for i in items:
  if i['EXPRESSION'] or i['INDEX_TYPE']!='BTREE':raise ValueError('Unreviewed index')
  v=ident(i['COLUMN_NAME'])
  if i['SUB_PART']:v='CAST(SUBSTR('+v+',1,'+str(i['SUB_PART'])+') AS VARCHAR('+str(i['SUB_PART'])+' CHAR))'
  fields.append(v)
 post.append('CREATE '+('UNIQUE 'if not items[0]['NON_UNIQUE']else'')+'INDEX '+ident(ixid)+' ON '+ident(name)+'('+','.join(fields)+');')
fks=collections.defaultdict(list)
for fk in b['foreign_keys']:fks[(fk['TABLE_NAME'],fk['CONSTRAINT_NAME'])].append(fk)
for (name,fkname),items in fks.items():
 f=items[0];s='ALTER TABLE '+ident(name)+' ADD CONSTRAINT '+ident('F_'+hashlib.sha256((name+'.'+fkname).encode()).hexdigest()[:24])+' FOREIGN KEY('+','.join(ident(i['COLUMN_NAME'])for i in items)+') REFERENCES '+ident(f['REFERENCED_TABLE_NAME'])+'('+','.join(ident(i['REFERENCED_COLUMN_NAME'])for i in items)+')'
 for verb in ('DELETE','UPDATE'):
  rule=f[verb+'_RULE']
  if rule not in ('NO ACTION','RESTRICT'):s+=' ON '+verb+' '+rule
 post.append(s+';')
checks=json.loads((r/'private/mysql-upgraded-checks.json').read_text())
for c in checks:
 if c!={'name':'chk_companion_rating','table':'companion_meals','clause':'((`rating` is null) or ((`rating` >= 1) and (`rating` <= 5)))','enforced':'YES'}:raise ValueError('Unreviewed source CHECK constraint')
 post.append('ALTER TABLE "COMPANION_MEALS" ADD CONSTRAINT "CHK_COMPANION_RATING" CHECK("RATING" IS NULL OR ("RATING">=1 AND "RATING"<=5));')
(out/'sql/01_schema.sql').write_text('\n'.join(schema)+'\n',encoding='utf8');(out/'sql/03_indexes_constraints.sql').write_text('\n'.join(post+comments)+'\n',encoding='utf8');(out/'column-mapping.json').write_text(json.dumps(mapping,ensure_ascii=False,indent=2));(out/'index-mapping.json').write_text(json.dumps(index_mapping,indent=2))
if schema_only:
 manifest=json.loads((out/'manifest.json').read_text())
 for rel in ('sql/01_schema.sql','sql/03_indexes_constraints.sql','column-mapping.json','index-mapping.json'):
  p=out/rel;manifest['files'][rel]={'bytes':p.stat().st_size,'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}
 (out/'manifest.json').write_text(json.dumps(manifest,indent=2));print('Reviewed schema and CHECK refreshed; data files untouched');sys.exit(0)
plan=[];total=0;lobn=0;t0=time.time();last=t0
for name,cs in cols.items():
 files=[];f=None;size=0;part=0;count=0;auto=any('auto_increment'in c['EXTRA']for c in cs)
 def openpart():
  global part,f,size
  part+=1;rel='sql/02_'+name+'_%03d.sql'%part;files.append(rel);f=(out/rel).open('w',encoding='utf8');size=0
  if auto:f.write('SET IDENTITY_INSERT '+ident(name)+' ON;\n')
 def closepart():
  if auto:f.write('SET IDENTITY_INSERT '+ident(name)+' OFF;\n')
  f.write('COMMIT;\n');f.close()
 src=r/'private/upgraded-export'/(name+'.jsonl.gz')
 if src.exists():
  with gzip.open(src,'rt',encoding='utf8')as rows:
   for line in rows:
    row=json.loads(line);vals=[];lob=False
    if len(row)!=len(cs):raise ValueError('Source row width mismatch '+name)
    for idx,(c,v)in enumerate(zip(cs,row)):
     if v is not None and (c['DATA_TYPE'].endswith('text')or c['DATA_TYPE']=='json') and len(v.encode('utf8'))>=16000:
      path='lob/'+name+'_%07d_%02d.txt'%(count+1,idx);(out/path).write_bytes(v.encode('utf8'));vals.append("@'"+path+"'");lob=True;lobn+=1
     else:vals.append(value_sql(c,v))
    s=('@insert'if lob else'INSERT')+' INTO '+ident(name)+'('+','.join(ident(c['COLUMN_NAME'])for c in cs)+') VALUES('+','.join(vals)+');\n'
    if f is None:openpart()
    elif size+len(s.encode('utf8'))>48*1024*1024:closepart();openpart()
    f.write(s);size+=len(s.encode('utf8'));count+=1;total+=1
    if count%1000==0:f.write('COMMIT;\n')
    if time.time()-last>20:print(json.dumps({'phase':'build_sql','table':name,'rows':total,'lobs':lobn}),flush=True);last=time.time()
 if f:closepart()
 if count!=int(m['counts'][name]):raise ValueError('Export count mismatch '+name)
 plan.append({'table':name,'rows':count,'data_files':files,'identity':auto,'identity_seed':int(tables[name]['AUTO_INCREMENT'] or 1)if auto else None})
manifest={'source_sha256':source_sha256,'source_encoding':m['encoding'],'code_commit':'c40700715f8c23eb33eb855beec44b28dd19ec5b','tables':plan,'total_rows':total,'lob_files':lobn,'max_source_field_utf8_bytes':max((int(n) for fields in m['max_utf8_lengths'].values() for n in fields.values()),default=0),'target_tables':len(plan),'target_columns':len(mapping),'files':{},'elapsed_s':round(time.time()-t0)}
for p in sorted(out.rglob('*')):
 if p.is_file():
  h=hashlib.sha256()
  with p.open('rb')as f:
   for chunk in iter(lambda:f.read(1024*1024),b''):h.update(chunk)
  manifest['files'][str(p.relative_to(out))]={'bytes':p.stat().st_size,'sha256':h.hexdigest()}
(out/'manifest.json').write_text(json.dumps(manifest,indent=2));print(json.dumps({'build_complete':True,'rows':total,'lobs':lobn,'files':len(manifest['files']),'elapsed_s':manifest['elapsed_s']}),flush=True)