"""Independently reconcile Stata datasets to every supplied Excel source cell."""
from pathlib import Path
import json
import sys
import math
import hashlib
import xlrd
import numpy as np
import pandas as pd

sys.stdout.reconfigure(encoding="utf-8")
ROOT = Path(__file__).resolve().parent
results = {"method":"xlrd reads original source cells; pandas independently reads Stata dta", "checks":[], "passed":True}

def check(name, passed, **details):
    results["checks"].append({"name":name,"passed":bool(passed),**details})
    if not passed:
        results["passed"] = False
        print("FAIL", name, details)

def compare_frames(name, expected, actual, require_exact_numeric=False):
    check(name+" row_count",len(expected)==len(actual), expected=len(expected), actual=len(actual))
    check(name+" key_alignment",expected.index.equals(actual.index))
    if not expected.index.equals(actual.index):
        actual = actual.reindex(expected.index)
    for col in expected.columns:
        if col not in actual:
            check(name+" column "+col,False,reason="missing column")
            continue
        exp, got = expected[col], actual[col]
        if pd.api.types.is_datetime64_any_dtype(exp.dtype):
            ok = ((exp == got) | (exp.isna() & got.isna()))
            detail = {"kind":"date"}
        elif pd.api.types.is_numeric_dtype(got.dtype):
            exp = pd.to_numeric(exp.mask(exp.eq(""), np.nan),errors="raise").astype(float)
            got = got.astype(float)
            exact = (exp == got) | (exp.isna() & got.isna())
            ok = exact if require_exact_numeric else np.isclose(exp.to_numpy(),got.to_numpy(),rtol=1e-13,atol=1e-13,equal_nan=True)
            delta = np.abs(exp-got)
            detail = {"kind":"numeric","exact_equal_cells":int(exact.sum()),"missing_source":int(exp.isna().sum()),"missing_output":int(got.isna().sum()),"max_absolute_difference":float(delta.max()) if delta.notna().any() else 0}
        else:
            exp = exp.fillna("").astype(str)
            got = got.fillna("").astype(str)
            ok = exp == got
            detail = {"kind":"text","max_source_characters":int(exp.str.len().max()),"max_output_characters":int(got.str.len().max())}
        ok = np.asarray(ok)
        detail["cells_compared"] = int(len(exp))
        detail["mismatches"] = int((~ok).sum())
        if not ok.all():
            bad = np.flatnonzero(~ok)[:5]
            detail["examples"] = [{"key":str(exp.index[i]),"source":str(exp.iloc[i]),"output":str(got.iloc[i])} for i in bad]
        check(name+" column "+col,ok.all(),**detail)

financial_file = next((ROOT/"data").glob("FIN*.xls"))
w=xlrd.open_workbook(financial_file)
s=w.sheet_by_index(0)
financial_cols=["Comcd","Lcomnm","A_Stkcd","B_Stkcd","H_Stkcd","Lstflg","Enddt","Reporttype","ROA","ROE"]
financial=pd.DataFrame([s.row_values(i) for i in range(1,s.nrows)],columns=financial_cols)
financial["Enddt"]=pd.to_datetime(financial.Enddt.map(lambda x:xlrd.xldate_as_datetime(x,w.datemode)))
financial["resset_excel_row"]=np.arange(2,s.nrows+1)
financial=financial.set_index("resset_excel_row").sort_index()
fin_dta=pd.read_stata(ROOT/"resset_financial_all_years.dta",convert_categoricals=False)
compare_frames("financial_source_vs_clean",financial,fin_dta.set_index("resset_excel_row").sort_index())

govs=[]
source_hashes={financial_file.name:hashlib.sha256(financial_file.read_bytes()).hexdigest()}
for f in sorted((ROOT/"data").glob("CG*.xls")):
    w=xlrd.open_workbook(f)
    s=w.sheet_by_index(0)
    df=pd.DataFrame([s.row_values(i) for i in range(3,s.nrows)],columns=s.row_values(0))
    df=df.rename(columns={"Stkcd":"stkcd"})
    df["csmar_source"]=f.name
    df["csmar_excel_row"]=np.arange(4,s.nrows+1)
    govs.append(df)
    source_hashes[f.name]=hashlib.sha256(f.read_bytes()).hexdigest()
governance=pd.concat(govs,ignore_index=True).set_index(["csmar_source","csmar_excel_row"]).sort_index()
gov_dta=pd.read_stata(ROOT/"csmar_governance_1999_2014.dta",convert_categoricals=False)
compare_frames("governance_source_vs_clean",governance,gov_dta.set_index(["csmar_source","csmar_excel_row"]).sort_index())

merged=pd.read_stata(ROOT/"financial_governance_all_years.dta",convert_categoricals=False)
check("merged_key_unique",not merged.duplicated(["stkcd","year"]).any())
compare_frames("financial_source_vs_merged",financial,merged.loc[merged.in_resset==1].set_index("resset_excel_row").sort_index())
compare_frames("governance_source_vs_merged",governance,merged.loc[merged.in_csmar==1].set_index(["csmar_source","csmar_excel_row"]).sort_index())

expected_fin=financial.reset_index()[["A_Stkcd","Enddt"]].rename(columns={"A_Stkcd":"stkcd"})
expected_fin["year"]=expected_fin.Enddt.dt.year
expected_gov=governance.reset_index()[["stkcd","Reptdt"]]
expected_gov["year"]=expected_gov.Reptdt.str[:4].astype(int)
expected_merge=expected_fin.merge(expected_gov,on=["stkcd","year"],how="outer",validate="1:1",indicator=True)
status_map={"left_only":1,"right_only":2,"both":3}
expected_merge["merge_status"]=expected_merge._merge.astype(str).map(status_map)
expected_merge=expected_merge.set_index(["stkcd","year"]).sort_index()
actual_merge=merged.set_index(["stkcd","year"]).sort_index()
check("merge_keyset",expected_merge.index.equals(actual_merge.index))
check("merge_status_every_row",np.array_equal(expected_merge.merge_status.to_numpy(),actual_merge._merge.to_numpy()))
check("in_resset_flag",np.array_equal(merged.in_resset.to_numpy(),merged._merge.isin([1,3]).astype(int).to_numpy()))
check("in_csmar_flag",np.array_equal(merged.in_csmar.to_numpy(),merged._merge.isin([2,3]).astype(int).to_numpy()))
check("matched_flag",np.array_equal(merged.matched.to_numpy(),(merged._merge==3).astype(int).to_numpy()))
check("complete_roa_roe_flag",np.array_equal(merged.complete_roa_roe.to_numpy(),((merged._merge==3)&merged.ROA.notna()&merged.ROE.notna()).astype(int).to_numpy()))
check("unusual_code_retained",merged.loc[merged.code_format_exception==1,["stkcd","year"]].to_dict(orient="records")==[{"stkcd":"0991","year":1998},{"stkcd":"0991","year":1999}])

expected_annual=expected_merge.reset_index().groupby(["year","merge_status"]).size().unstack(fill_value=0)
actual_annual=merged.groupby(["year","_merge"]).size().unstack(fill_value=0)
check("annual_merge_counts",expected_annual.equals(actual_annual))

def verify_csv_against_dta(stem):
    dta=pd.read_stata(ROOT/(stem+".dta"),convert_categoricals=False)
    numeric=[col for col in dta if pd.api.types.is_numeric_dtype(dta[col].dtype)]
    dates=[col for col in dta if pd.api.types.is_datetime64_any_dtype(dta[col].dtype)]
    types={col:(float if col in numeric else str) for col in dta}
    csv=pd.read_csv(ROOT/(stem+".csv"),dtype=types,keep_default_na=False,
                    na_values={col:[""] for col in numeric},float_precision="round_trip",encoding="utf-8")
    check(stem+" CSV column_names_order",list(dta.columns)==list(csv.columns),column_count=len(csv.columns))
    for col in dates:
        date_strings=csv[col]
        valid=date_strings.eq("") | date_strings.str.fullmatch(r"\d{4}-\d{2}-\d{2}")
        check(stem+" CSV ISO date "+col,valid.all())
        csv[col]=pd.to_datetime(date_strings.mask(date_strings.eq("")),format="%Y-%m-%d",errors="raise")
    compare_frames(stem+" DTA_vs_CSV",dta,csv,require_exact_numeric=True)
    return {"rows":len(csv),"columns":len(csv.columns),"numeric_columns":len(numeric),
            "date_columns":dates,"cells_compared":int(csv.size),"numeric_rule":"17-digit CSV parsed with float_precision=round_trip; exact equality including missing values"}

success_marker="WEEK8_ANALYSIS_COMPLETED_SUCCESSFULLY"
check("Stata_success_log",success_marker in (ROOT/"week8_stata.log").read_text(encoding="utf-8"))
results["csv_validation"]={stem:verify_csv_against_dta(stem) for stem in ["financial_governance_all_years","df_example_metrics"]}
results["annual_counts"]={str(y):{str(k):int(v) for k,v in row.items()} for y,row in actual_annual.to_dict(orient="index").items()}
results["summary"]={"financial_rows":len(financial),"governance_rows":len(governance),"merged_rows":len(merged),"merge_counts":{str(k):int(v) for k,v in merged._merge.value_counts().sort_index().items()},"financial_cells_compared_per_dataset":int(financial.size),"governance_cells_compared_per_dataset":int(governance.size),"matched_complete_roa_roe":int(merged.complete_roa_roe.sum()),"matched_missing_roa":int(merged.loc[merged.matched==1,"ROA"].isna().sum()),"matched_missing_roe":int(merged.loc[merged.matched==1,"ROE"].isna().sum())}
results["source_sha256"]=source_hashes
results["check_count"]=len(results["checks"])
results["summary"]["all_cells_compared"]=sum(x.get("cells_compared",0) for x in results["checks"])
(ROOT/"validation_results.json").write_text(json.dumps(results,ensure_ascii=False,indent=2,allow_nan=False),encoding="utf-8")
print(json.dumps({"passed":results["passed"],"check_count":results["check_count"],"summary":results["summary"]},ensure_ascii=False,indent=2))
if not results["passed"]:
    raise SystemExit(1)