"""Partial rehearsal payload: unresolved zero-date tables deferred, never relabel as final."""
import json,gzip,sys,re,hashlib
from pathlib import Path
from mysql_baseline import q,ROOT,ident as mi
from build_front_schema import ident
sys.path.insert(0,str(ROOT.parent/'foodsafe2-mysql-dm-migration-20261023/scripts'))
from dm_sql_contract import literal,numeric
NUM={'tinyint','smallint','mediumint','int','bigint','decimal','float','double'}
for db in ('zhct','zhct2'):
    base=json.loads((ROOT/'private'/('final-'+db)/'baseline.json').read_text())
    metadata=json.loads((ROOT/'private'/('metadata-'+db+'_stage.json')).read_text())
    pk={i['TABLE_NAME']:i['COLUMN_NAME'] for i in metadata['indexes'] if i['INDEX_NAME']=='PRIMARY'}
    out=ROOT/'private'/('final-data-'+db);out.mkdir(mode=0o700,exist_ok=True)
    (out/'sql').mkdir(exist_ok=True);(out/'lob').mkdir(exist_ok=True)
    floats={};floatcount=0
    for table,cs in base['metadata'].items():
        fc=[c['name'] for c in cs if c['data_type'] in ('float','double')]
        if not fc or not base['counts'][table]:continue
        expr=','.join('IF('+mi(c)+" IS NULL,'N',CONCAT('V',"+mi(c)+"+0e0))" for c in fc)
        lines=q('SELECT '+mi(pk[table])+','+expr+' FROM '+mi(db+'_stage')+'.'+mi(table)+';')
        floats[table]={}
        for line in lines:
            a=line.split('\t');assert len(a)==len(fc)+1
            floats[table][a[0]]=dict(zip(fc,[None if x=='N' else x[1:] for x in a[1:]]));floatcount+=len(fc)
    (out/'float-source.json').write_text(json.dumps(floats))
    manifest={'partial_rehearsal_only':False,'source_db':db,'deferred_tables':[],'tables':[],'rows':0,'float_values':floatcount,'files':{}}
    for table,cs in base['metadata'].items():
        src=ROOT/'private'/('final-'+db)/(table+'.hex.jsonl.gz')
        rows=[];zero=[]
        with gzip.open(src,'rt') as f:
            for lineno,line in enumerate(f,1):
                values=[None if v is None else bytes.fromhex(v).decode('utf8') for v in json.loads(line)]
                for c,v in zip(cs,values):
                    if v is not None and c['data_type'] in ('date','datetime','timestamp') and (v[:4]=='0000' or v[5:7]=='00' or v[8:10]=='00'):zero.append({'row':lineno,'field':c['name']})
                rows.append(values)
        if zero:raise ValueError('Unresolved zero date in final dataset')
        if zero:
            manifest['deferred_tables'].append({'table':table,'rows':len(rows),'zero_date_cells':len(zero)})
            continue
        if not rows:
            manifest['tables'].append({'table':table,'rows':0,'data_file':None});continue
        path='sql/'+table+'.sql';auto=any('auto_increment' in c['extra'] for c in cs)
        pki=next(i for i,c in enumerate(cs) if c['name']==pk[table])
        with (out/path).open('w') as f:
            if auto:f.write('SET IDENTITY_INSERT '+ident(table)+' ON;\n')
            for n,row in enumerate(rows,1):
                vals=[];lob=False
                for ci,(c,v) in enumerate(zip(cs,row)):
                    if c['data_type'] in ('float','double'):v=floats[table][row[pki]][c['name']]
                    if v is None:vals.append('NULL')
                    elif c['data_type'] in NUM:vals.append(numeric(v))
                    elif c['data_type'].endswith('text') and len(v.encode())>=16000:
                        rel='lob/'+table+'_'+str(n)+'_'+str(ci)+'.txt';(out/rel).write_bytes(v.encode());vals.append("@'"+rel+"'");lob=True
                    else:vals.append(literal(v))
                f.write(('@insert' if lob else 'INSERT')+' INTO '+ident(table)+'('+','.join(ident(c['name']) for c in cs)+') VALUES('+','.join(vals)+');\n')
                if n%1000==0:f.write('COMMIT;\n')
            if auto:f.write('SET IDENTITY_INSERT '+ident(table)+' OFF;\n')
            f.write('COMMIT;\n')
        manifest['tables'].append({'table':table,'rows':len(rows),'data_file':path});manifest['rows']+=len(rows)
    for p in out.rglob('*'):
        if p.is_file() and p.name!='manifest.json':manifest['files'][str(p.relative_to(out))]=hashlib.sha256(p.read_bytes()).hexdigest()
    (out/'manifest.json').write_text(json.dumps(manifest,indent=2))
    print(json.dumps({k:manifest[k] for k in ['source_db','partial_rehearsal_only','deferred_tables','rows','float_values']}))