"""Cotejo de metadatos y UPM: publica solo agregados, nunca llaves individuales."""
from pathlib import Path
import argparse, hashlib, json
import pandas as pd
A=Path(__file__).resolve().parent
W=A.parents[1]/'dist/pulso-sanitario/tablero'
parser=argparse.ArgumentParser();parser.add_argument('--enaho',type=Path,required=True);parser.add_argument('--salida',type=Path,default=W/'datos',help='Directorio que contiene r_estimaciones.csv y enaho_2025_2026.csv; también recibe el cotejo');args=parser.parse_args()
D=args.salida; tables={};inventory=[]; samples=[]; comparisons=[]; modules=[]
num=lambda s:pd.to_numeric(s.str.replace(',','.',regex=False),errors='coerce')
for label,code,weight in [('2025','1031','FACTOR07'),('2026T1','1039','FACTOR'),('2026T2','1048','FACTOR')]:
 for module in ['01','04']:
  f=next((args.enaho/(code+'-Modulo'+module)).rglob('*'+('100' if module=='01' else '400')+'.csv'))
  d=pd.read_csv(f,sep=None,engine='python',encoding='latin1',dtype=str).fillna('');d.columns=d.columns.str.strip();d=d.apply(lambda s:s.str.strip())
  inventory.append(dict(periodo=label,modulo=module,archivo=str(f.relative_to(args.enaho)),bytes=f.stat().st_size,sha256=hashlib.file_digest(f.open('rb'),'sha256').hexdigest(),peso=weight,campos_marco_presentes=','.join(c for c in ['NCONGLOME','SUB_CONGLOME','PANEL','TIPENC'] if c in d)))
  d=d[d.RESULT.isin(['1','2'])].copy() if module=='01' else d[num(d[weight])>0].copy()
  d['trimestre']=(num(d.MES)-1)//3+1;d['peso_num']=num(d[weight]);tables[label,module]=d
  samples.append(dict(periodo=label,modulo=module,n_registros=len(d),upm=d.CONGLOME.nunique(),pesos_no_positivos=int((d.peso_num<=0).sum()),pesos_faltantes=int(d.peso_num.isna().sum()),**{'upm_multivalor_'+c:int((d.groupby('CONGLOME')[c].nunique()>1).sum()) for c in ['UBIGEO','DOMINIO','ESTRATO','MES']}))
 # Coherencia del factor entre módulos, no equivalencia de sus universos.
 h=tables[label,'01'];p=tables[label,'04'];keys=['MES','CONGLOME','VIVIENDA','HOGAR']
 assert not h.duplicated(keys).any()
 z=p.merge(h[keys+['peso_num']],on=keys,how='left',suffixes=('_400','_100'),validate='many_to_one')
 modules.append(dict(periodo=label,personas=len(z),sin_hogar_en_modulo100=int(z.peso_num_100.isna().sum()),pesos_distintos=int(((z.peso_num_400-z.peso_num_100).abs()>1e-8).sum())))
for module in ['01','04']:
 for q in [1,2]:
  a=tables['2025',module];b=tables['2026T'+str(q),module]
  x=a.groupby('CONGLOME');y=b.groupby('CONGLOME');shared=set(x.groups)&set(y.groups)
  r=dict(modulo=module,trimestre=q,upm2025=len(x),upm2026=len(y),upm_compartidas=len(shared),porcentaje_upm2026_compartidas=100*len(shared)/len(y))
  for c in ['UBIGEO','DOMINIO','ESTRATO','MES']:
   aa=x[c].agg(lambda s:tuple(sorted(s.unique())));bb=y[c].agg(lambda s:tuple(sorted(s.unique())))
   r['upm_cambio_'+c]=sum(aa[k]!=bb[k] for k in shared)
  comparisons.append(r)
qshared=len(set(tables['2026T1','01'].CONGLOME)&set(tables['2026T2','01'].CONGLOME))
assert qshared==0
pd.DataFrame(inventory).to_csv(D/'cotejo_inventario.csv',index=False)
pd.DataFrame(samples).to_csv(D/'cotejo_muestras.csv',index=False)
pd.DataFrame(comparisons).to_csv(D/'cotejo_upm.csv',index=False)
pd.DataFrame(modules).to_csv(D/'cotejo_modulos.csv',index=False)
# Identificación fila a fila de la familia interanual completa.
c=pd.read_csv(D/'enaho_2025_2026.csv');assert len(c)==100
for col in ['indicador','region','trimestre']: assert c[col].notna().all()
c=c[['trimestre','periodo','indicador','etiqueta','region']].copy();c.insert(0,'comparacion_id',['ENAHO-INTERANUAL-'+str(i).zfill(3) for i in range(1,101)])
c['estado']='REHECHO_REFERENCIA_ANUAL_2025';c['peso2025_usado']='FACTOR07 anual sin transformación; doce meses';c['peso2025_requerido']='Referencia anual 2025 aplicada; ver CONTRASTES_TEMPORALES.md';c['efecto']='Estimaciones 2025, diferencias, IC, p nominal y Holm';c['upm']='Reutilización documentada; enlace territorial con todo el anual 2025, con sensibilidad bajo independencia';c['varianza']='DOMINIO×ESTRATO no acreditado como estrato final de varianza';c.to_csv(D/'comparaciones_afectadas_cotejo.csv',index=False)
# Todas las estimaciones, no solo los 100 contrastes, reciben un estado acotado.
e=pd.read_csv(D/'r_estimaciones.csv'); e=e[['encuesta','periodo','indicador','nivel','grupo']].copy()
def state(r):
 if r.encuesta.startswith('ENDES'): return 'FUERA_DEL_COTEJO_ENAHO; se conserva la especificación previa'
 if r.encuesta=='ENAHO 2025' and r.periodo=='Año 2025':return 'PESO_ANUAL_DOCUMENTADO; estrato final de varianza pendiente'
 if r.encuesta=='ENAHO 2025':return 'RECALCULO_CON_PESO_TRIMESTRAL_2025'
 if r.periodo=='Semestre 2026':return 'AGREGADO_DE_TRIMESTRES; sin factor semestral acreditado'
 return 'FACTOR_TRIMESTRAL_DOCUMENTADO_POR_MODULO; estrato final pendiente'
e['estado_cotejo']=e.apply(state,axis=1);e.to_csv(D/'estimaciones_estado_cotejo.csv',index=False)
s=dict(fecha='2026-11-06',alcance='ENAHO 2025 anual y ENAHO 2026 T1/T2; extractos cotejados con los seis originales suministrados; no descarga oficial independiente',comparaciones_interanuales_recalcular=0,comparaciones_anual_trimestral_rehechas=100,estimaciones2025_trimestrales_recalcular=int(((e.encuesta=='ENAHO 2025')&(e.periodo!='Año 2025')).sum()),upm_compartidas_entre_trimestres2026=qshared,upm=comparisons,coherencia_modulos=modules,estados_estimaciones=e.estado_cotejo.value_counts().to_dict())
(D/'cotejo_documental.json').write_text(json.dumps(s,ensure_ascii=False,indent=2)+'\n')
print(json.dumps(s,ensure_ascii=False,indent=2))