"""N3 - resolver composition, over-merge indicators, specification curve, ties, June->October
reconciliation (cluster B). Read-only; _m/_keyed rebuilt as TEMP tables from resolve.py:17-32.
"""
from __future__ import annotations
import json, sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import *
sys.path.insert(0, str(Path(__file__).resolve().parent / "spid"))
import taxonomy as tx  # snapshot copy (hash-verified in N0)

EXAMPLE_PROJECT = "3e2705ba5b2dd27408b25f0145e34250"   # stats.json example_project (Cheniere Corpus Christi Stage 3)
STOP = ["project", "program", "development", "expansion", "acquisition", "phase", "initiative", "facility", "construction"]
THRESHOLDS = [0.45, 0.6, 0.7, 0.8, 0.9]

r = Run("resolver")
con = connect(SPID_DB)
tmin = con.execute("select min(extracted_at) from extracted_sections").fetchone()[0]
con.execute(f"""CREATE TEMP TABLE runmap AS SELECT section_id, CASE WHEN extracted_at = TIMESTAMP '{tmin}' THEN 'backfilled_june' ELSE 'october' END AS run FROM extracted_sections""")

# ---- exact rebuild of _m / _keyed (resolve.py:17-32) --------------------------------------------
con.execute("""
CREATE TEMP TABLE _m AS
SELECT *,
   lower(coalesce(try(locations[1]), '')) AS loc1,
   regexp_extract(lower(regexp_replace(project_name, '[^A-Za-z ]', '', 'g')), '([a-z]{4,})', 1) AS name_tok
FROM project_mentions""")
con.execute("""
CREATE TEMP TABLE _keyed AS
SELECT *,
   cik || '|' || project_type || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(nullif(name_tok,''),'general') END AS gkey,
   md5(cik || '|' || project_type || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(nullif(name_tok,''),'general') END) AS project_id
FROM _m""")
V = {}
V["rebuilt_projects"] = con.execute("select count(distinct project_id) from _keyed").fetchone()[0]
V["stored_projects"] = con.execute("select count(*) from projects").fetchone()[0]
V["project_id_set_equal"] = con.execute("select count(*) from (select project_id from _keyed except select project_id from projects) union all select count(*) from (select project_id from projects except select distinct project_id from _keyed)").fetchall()
V["mentions"] = con.execute("select count(*) from _keyed").fetchone()[0]

# ---- (a) key composition ----------------------------------------------------------------------
V["mentions_by_key_branch"] = con.execute("""
  select case when loc1<>'' then 'location' when nullif(name_tok,'') is not null then 'name_token' else 'general' end b, count(*) from _keyed group by 1""").fetchall()
V["projects_by_key_branch"] = con.execute("""
  select case when loc1<>'' then 'location' when nullif(name_tok,'') is not null then 'name_token' else 'general' end b, count(distinct project_id) from _keyed group by 1""").fetchall()
V["projects_keyed_general"] = con.execute("select count(distinct project_id) from _keyed where gkey like '%|general'").fetchone()[0]
V["mentions_keyed_general"] = con.execute("select count(*) from _keyed where gkey like '%|general'").fetchone()[0]
states = sorted(tx.US_STATES); countries = sorted(tx.COUNTRIES)
con.execute("CREATE TEMP TABLE geo AS SELECT unnest(?) AS loc, 'us_state' AS kind UNION ALL SELECT unnest(?), 'country'", [states, countries])
V["location_keyed_projects_by_geo_kind"] = con.execute("""
  with p as (select project_id, any_value(loc1) loc1 from _keyed where loc1<>'' group by 1)
  select coalesce(g.kind,'other_place') kind, count(*) from p left join geo g on p.loc1 = g.loc group by 1""").fetchall()
V["location_keyed_mentions_by_geo_kind"] = con.execute("""
  select coalesce(g.kind,'other_place') kind, count(*) from _keyed k left join geo g on k.loc1 = g.loc where k.loc1<>'' group by 1""").fetchall()
r.dump("top_name_tokens.csv", con, """
  with t as (select name_tok, count(*) mentions, count(distinct project_id) projects from _keyed where loc1='' and nullif(name_tok,'') is not null group by 1)
  select name_tok, mentions, projects, round(100.0*mentions/(select count(*) from _keyed where loc1=''),2) pct_of_tokenkeyed_mentions,
         round(100.0*projects/(select count(distinct project_id) from _keyed where loc1=''),2) pct_of_tokenkeyed_projects
  from t order by projects desc limit 50""")
r.dump("top_location_keys.csv", con, """
  select loc1, count(*) mentions, count(distinct project_id) projects from _keyed where loc1<>'' group by 1 order by 2 desc limit 50""")
r.dump("key_branch_by_type.csv", con, """
  select project_type, count(distinct project_id) projects,
         count(distinct project_id) filter (where loc1<>'') loc_keyed,
         count(distinct project_id) filter (where loc1='' and nullif(name_tok,'') is not null) token_keyed,
         count(distinct project_id) filter (where loc1='' and nullif(name_tok,'') is null) general_keyed
  from _keyed group by 1 order by 2 desc""")

# ---- (b) over-merge indicators ----------------------------------------------------------------
con.execute("""
CREATE TEMP TABLE pstat AS
SELECT project_id, any_value(project_type) project_type, count(*) n_mentions,
       count(DISTINCT lower(trim(project_name))) n_names,
       count(DISTINCT lower(trim(coalesce(try(partners[1]), '')))) FILTER (WHERE len(partners)>0) n_partners,
       count(DISTINCT accession_number) n_filings, count(DISTINCT cik) n_ciks,
       count(DISTINCT lower(description)) n_desc
FROM _keyed GROUP BY 1""")
bucket = "case when n_mentions=1 then '1' when n_mentions=2 then '2' when n_mentions<=5 then '3-5' when n_mentions<=10 then '6-10' else '>10' end"
r.dump("overmerge_by_type.csv", con, f"""
  select project_type, count(*) projects, sum(n_mentions) mentions,
         round(100.0*count(*) filter (where n_names>=3)/count(*),2) pct_ge3_names,
         round(100.0*count(*) filter (where n_names>=2)/count(*),2) pct_ge2_names,
         round(100.0*count(*) filter (where n_partners>=3)/nullif(count(*) filter (where n_partners>=1),0),2) pct_ge3_partners_among_with_partner,
         count(*) filter (where n_partners>=3) n_ge3_partners,
         round(100.0*count(*) filter (where n_mentions>1 and n_names = n_mentions)/nullif(count(*) filter (where n_mentions>1),0),2) pct_multi_all_names_distinct
  from pstat group by 1 order by 2 desc""")
r.dump("overmerge_by_bucket.csv", con, f"""
  select {bucket} bucket, count(*) projects,
         round(100.0*count(*) filter (where n_names>=3)/count(*),2) pct_ge3_names,
         round(100.0*count(*) filter (where n_partners>=3)/nullif(count(*) filter (where n_partners>=1),0),2) pct_ge3_partners_among_with_partner,
         round(avg(n_names),2) mean_names
  from pstat group by 1 order by min(n_mentions)""")
r.dump("overmerge_ma_by_bucket.csv", con, f"""
  select {bucket} bucket, count(*) projects,
         count(*) filter (where n_partners>=1) with_partner,
         round(100.0*count(*) filter (where n_partners>=3)/nullif(count(*) filter (where n_partners>=1),0),2) pct_ge3_partners,
         round(100.0*count(*) filter (where n_partners>=2)/nullif(count(*) filter (where n_partners>=1),0),2) pct_ge2_partners,
         round(100.0*count(*) filter (where n_names>=3)/count(*),2) pct_ge3_names
  from pstat where project_type='ma_integration' group by 1 order by min(n_mentions)""")
# amount from a mention naming a different counterparty than the representative mention
V["amount_from_other_counterparty"] = con.execute("""
  with rep as (select project_id, arg_max(lower(trim(coalesce(try(partners[1]),''))), confidence) rep_partner from _keyed group by 1),
       amt as (select project_id, arg_max(lower(trim(coalesce(try(partners[1]),''))), amount_usd) amt_partner, max(amount_usd) mx from _keyed where amount_usd is not null group by 1)
  select count(*) filter (where rep_partner<>'' and amt_partner<>'' and rep_partner<>amt_partner) differing,
         count(*) filter (where rep_partner<>'' and amt_partner<>'') both_named, count(*) with_amount
  from amt join rep using(project_id)""").fetchone()
V["amount_from_other_counterparty_ma"] = con.execute("""
  with rep as (select project_id, any_value(project_type) t, arg_max(lower(trim(coalesce(try(partners[1]),''))), confidence) rep_partner from _keyed group by 1),
       amt as (select project_id, arg_max(lower(trim(coalesce(try(partners[1]),''))), amount_usd) amt_partner from _keyed where amount_usd is not null group by 1)
  select count(*) filter (where rep_partner<>'' and amt_partner<>'' and rep_partner<>amt_partner) differing,
         count(*) filter (where rep_partner<>'' and amt_partner<>'') both_named, count(*) with_amount
  from amt join rep using(project_id) where t='ma_integration'""").fetchone()
V["amount_from_other_name"] = con.execute("""
  with rep as (select project_id, arg_max(lower(trim(project_name)), confidence) rep_name from _keyed group by 1),
       amt as (select project_id, arg_max(lower(trim(project_name)), amount_usd) amt_name from _keyed where amount_usd is not null group by 1)
  select count(*) filter (where rep_name<>amt_name) differing, count(*) with_amount from amt join rep using(project_id)""").fetchone()
r.dump("example_provenance_keys.csv", con, """
  select project_id, cik, project_type, gkey, count(*) n, count(distinct lower(project_name)) n_names, list(distinct left(project_name,60)) names
  from _keyed where project_id in (select project_id from projects where project_id in (
     select project_id from project_timeline group by 1 having count(*) between 4 and 8) and total_amount_usd > 5e8 and project_type in ('plant_construction','data_center'))
  group by 1,2,3,4 order by n desc limit 30""")

# ---- (c) specification curve ----------------------------------------------------------------------
stop_sql = ",".join("'" + s + "'" for s in STOP)
con.execute(f"""
CREATE TEMP TABLE km AS
SELECT mention_id, cik, project_type, confidence, loc1, name_tok, gkey,
  regexp_replace(trim(regexp_replace(lower(project_name), '[^a-z0-9 ]', '', 'g')), ' +', ' ', 'g') AS name_norm,
  list_filter(string_split(regexp_replace(lower(project_name), '[^a-z ]', '', 'g'), ' '), lambda x: length(x) >= 4) AS toks,
  list_filter(string_split(regexp_replace(lower(project_name), '[^a-z ]', '', 'g'), ' '), lambda x: length(x) >= 4 AND x NOT IN ({stop_sql})) AS toks_sw,
  trim(regexp_replace(regexp_replace(regexp_replace(loc1, '\\b(u\\.s\\.|usa|u\\.s\\.a\\.|united states of america)\\b', 'united states', 'g'), '[^a-z, ]', '', 'g'), ' +', ' ', 'g')) AS loc_norm
FROM _keyed""")
con.execute("""
CREATE TEMP TABLE keys AS
SELECT mention_id, confidence, project_type,
  cik || '|' || project_type AS k_cik_type,
  gkey AS k_current,
  cik || '|' || project_type || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(toks_sw[1], 'general') END AS k_current_stopwords,
  cik || '|' || project_type || '|' || CASE WHEN loc_norm <> '' THEN trim(string_split(loc_norm, ',')[1]) ELSE coalesce(toks_sw[1], 'general') END AS k_city_normalised,
  cik || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(nullif(name_tok,''), 'general') END AS k_type_dropped,
  cik || '|' || project_type || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(toks_sw[1], 'general') || '|' || coalesce(toks_sw[2], '') END AS k_two_tokens,
  cik || '|' || project_type || '|' || coalesce(toks_sw[1], 'general') AS k_name_only,
  cik || '|' || project_type || '|' || name_norm || '|' || loc1 AS k_strict,
  mention_id AS k_mention_level
FROM km""")
KEYS = ["k_cik_type", "k_type_dropped", "k_current", "k_current_stopwords", "k_city_normalised", "k_name_only", "k_two_tokens", "k_strict", "k_mention_level"]
ex_mention = con.execute("select arg_max(mention_id, confidence) from _keyed where project_id = ?", [EXAMPLE_PROJECT]).fetchone()[0]
rows = []; bytype = []
for k in KEYS:
    for t in THRESHOLDS:
        q = f"""
        with g as (select {k} key, count(*) n, max(mention_id = '{ex_mention}')::int has_ex from keys where confidence >= {t} group by 1)
        select count(*) projects, sum(n) mentions, round(100.0*count(*) filter (where n=1)/count(*),2) single_pct,
               round(100.0*count(*) filter (where n>10)/count(*),2) gt10_pct, round(avg(n),3) mean_mentions,
               max(n) max_cluster, coalesce(max(n) filter (where has_ex=1),0) example_cluster_size
        from g"""
        row = con.execute(q).fetchone()
        rows.append((k, t) + tuple(row))
        bt = con.execute(f"""
        with g as (select {k} key, any_value(project_type) project_type, count(*) n from keys where confidence >= {t} group by 1)
        select project_type, count(*) n from g group by 1""").fetchall()
        bytype += [(k, t, pt, n) for pt, n in bt]
        print(k, t, row, flush=True)
r.write_csv("spec_curve.csv", ["key", "threshold", "projects", "mentions", "single_mention_pct", "gt10_mention_pct", "mean_mentions", "max_cluster", "example_cluster_size_cheniere"], rows)
r.write_csv("spec_curve_by_type.csv", ["key", "threshold", "project_type", "projects"], bytype)
cur = [x for x in rows if x[0] == "k_current" and x[1] == 0.45][0]
V["project_count_current"] = cur[2]
V["project_count_range_045"] = {k: [x[2] for x in rows if x[0] == k and x[1] == 0.45][0] for k in KEYS}
V["project_count_strict_045"] = V["project_count_range_045"]["k_strict"]
V["project_count_loose_045_cik_type"] = V["project_count_range_045"]["k_cik_type"]
V["example_mention"] = ex_mention

# ---- (d) ties and idempotence ----------------------------------------------------------------------
D = {}
D["projects_with_tie_at_max_confidence"] = con.execute("""
  with mx as (select project_id, max(confidence) mc from _keyed group by 1)
  select count(*) from (select k.project_id from _keyed k join mx using(project_id) where k.confidence = mx.mc group by 1 having count(*)>1)""").fetchone()[0]
D["projects_with_tie_and_distinct_names"] = con.execute("""
  with mx as (select project_id, max(confidence) mc from _keyed group by 1)
  select count(*) from (select k.project_id from _keyed k join mx using(project_id) where k.confidence = mx.mc group by 1 having count(distinct project_name)>1)""").fetchone()[0]
D["projects_with_tie_and_distinct_descriptions"] = con.execute("""
  with mx as (select project_id, max(confidence) mc from _keyed group by 1)
  select count(*) from (select k.project_id from _keyed k join mx using(project_id) where k.confidence = mx.mc group by 1 having count(distinct description)>1)""").fetchone()[0]
D["projects_with_status_tie_on_last_seen"] = con.execute("""
  with mx as (select project_id, max(filing_date) md from _keyed group by 1)
  select count(*) from (select k.project_id from _keyed k join mx using(project_id) where k.filing_date = mx.md group by 1 having count(distinct status)>1)""").fetchone()[0]
D["mentions_at_confidence_1"] = con.execute("select count(*) from project_mentions where confidence = 1.0").fetchone()[0]
# stored projects vs rebuilt with the original (tie-ambiguous) SQL: how many representative fields differ
con.execute("""
CREATE TEMP TABLE rebuilt AS
SELECT project_id, arg_max(project_name, confidence) project_name, arg_max(status, filing_date) status, arg_max(description, confidence) description,
       max(amount_usd) total_amount_usd, count(*) n_mentions
FROM _keyed GROUP BY 1""")
D["rebuilt_vs_stored_diff"] = con.execute("""
  select count(*) filter (where r.project_name is distinct from p.project_name) name_diff,
         count(*) filter (where r.status is distinct from p.status) status_diff,
         count(*) filter (where r.description is distinct from p.description) desc_diff,
         count(*) filter (where r.total_amount_usd is distinct from p.total_amount_usd) amount_diff,
         count(*) filter (where r.n_mentions is distinct from p.n_mentions) n_diff, count(*) n
  from rebuilt r join projects p using(project_id)""").fetchone()
con.execute("""
CREATE TEMP TABLE rebuilt2 AS
SELECT project_id, arg_max(project_name, confidence) project_name, arg_max(status, filing_date) status, arg_max(description, confidence) description
FROM _keyed GROUP BY 1""")
D["original_sql_run_twice_diff"] = con.execute("""
  select count(*) filter (where a.project_name is distinct from b.project_name), count(*) filter (where a.status is distinct from b.status), count(*) filter (where a.description is distinct from b.description)
  from rebuilt a join rebuilt2 b using(project_id)""").fetchone()
det = """
SELECT project_id, project_name, status, description FROM (
  SELECT project_id,
         first_value(project_name) OVER (PARTITION BY project_id ORDER BY confidence DESC, mention_id) project_name,
         first_value(description) OVER (PARTITION BY project_id ORDER BY confidence DESC, mention_id) description,
         first_value(status) OVER (PARTITION BY project_id ORDER BY filing_date DESC, mention_id) status,
         row_number() OVER (PARTITION BY project_id ORDER BY mention_id) rn
  FROM _keyed) WHERE rn = 1"""
con.execute(f"CREATE TEMP TABLE det1 AS {det}")
con.execute(f"CREATE TEMP TABLE det2 AS {det}")
D["deterministic_tiebreak_run_twice_diff"] = con.execute("""
  select count(*) filter (where a.project_name is distinct from b.project_name), count(*) filter (where a.status is distinct from b.status), count(*) filter (where a.description is distinct from b.description)
  from det1 a join det2 b using(project_id)""").fetchone()
D["deterministic_vs_stored_diff"] = con.execute("""
  select count(*) filter (where a.project_name is distinct from p.project_name), count(*) filter (where a.status is distinct from p.status), count(*) filter (where a.description is distinct from p.description)
  from det1 a join projects p using(project_id)""").fetchone()
r.write_json("ties_idempotence.json", D)

# ---- (e) June -> October reconciliation (S&P 500 CIKs) -------------------------------------------
con.execute("""CREATE TEMP TABLE kr AS SELECT k.*, rm.run, co.in_sp500, co.in_russell3000 FROM _keyed k LEFT JOIN runmap rm USING (section_id) LEFT JOIN src_companies co ON k.cik = co.cik""")
R = {}
R["sp500_ciks"] = con.execute("select count(*) from src_companies where in_sp500").fetchone()[0]
R["mentions_in_backfilled_block"] = con.execute("select count(*) from kr where run='backfilled_june'").fetchone()[0]
R["projects_from_backfilled_block_only"] = con.execute("select count(distinct project_id) from kr where run='backfilled_june'").fetchone()[0]
R["expected_june"] = {"mentions": 39930, "projects": 19227, "timeline_rows": 39930}
R["sp500_mentions_total"] = con.execute("select count(*) from kr where in_sp500").fetchone()[0]
R["sp500_projects_total"] = con.execute("select count(distinct project_id) from kr where in_sp500").fetchone()[0]
R["paper_sp500_projects_october"] = 27766
R["sp500_projects_by_run_composition"] = con.execute("""
  with p as (select project_id, max(run='backfilled_june')::int has_june, max(run='october')::int has_oct from kr where in_sp500 group by 1)
  select case when has_june=1 and has_oct=1 then 'both runs' when has_june=1 then 'june only' else 'october only' end c, count(*) from p group by 1""").fetchall()
R["sp500_mentions_by_run_form"] = con.execute("select run, form_type, count(*), count(distinct accession_number) from kr where in_sp500 group by 1,2 order by 1,2").fetchall()
R["sp500_mentions_filing_date_le_20260525_by_run"] = con.execute("select run, count(*), count(distinct project_id) from kr where in_sp500 and filing_date <= DATE '2026-05-25' group by 1").fetchall()
R["sp500_projects_filing_date_le_20260525"] = con.execute("select count(distinct project_id) from kr where in_sp500 and filing_date <= DATE '2026-05-25'").fetchone()[0]
R["backfilled_block_mentions_not_sp500"] = con.execute("select count(*) from kr where run='backfilled_june' and not coalesce(in_sp500,false)").fetchone()[0]
R["june_10q_accessions_in_block_ledger"] = con.execute(f"""
  select count(distinct e.accession_number) from extracted_sections e join src_filings f using(accession_number) where e.extracted_at = TIMESTAMP '{tmin}' and f.form_type='10-Q'""").fetchone()[0]
R["ledger_block_accessions_by_form"] = con.execute(f"""
  select f.form_type, count(distinct e.accession_number), count(*) from extracted_sections e join src_filings f using(accession_number) where e.extracted_at = TIMESTAMP '{tmin}' group by 1 order by 1""").fetchall()
R["sp500_october_ledger_by_form"] = con.execute(f"""
  select f.form_type, count(distinct e.accession_number), count(*), sum(e.n_mentions) from extracted_sections e join src_filings f using(accession_number) join src_companies co on f.cik=co.cik
  where e.extracted_at <> TIMESTAMP '{tmin}' and co.in_sp500 group by 1 order by 1""").fetchall()
R["sp500_block_ledger_by_form"] = con.execute(f"""
  select f.form_type, count(distinct e.accession_number), count(*), sum(e.n_mentions) from extracted_sections e join src_filings f using(accession_number) join src_companies co on f.cik=co.cik
  where e.extracted_at = TIMESTAMP '{tmin}' and co.in_sp500 group by 1 order by 1""").fetchall()
r.write_json("june_october_reconciliation.json", R)
r.dump("june_october_reconciliation.csv", con, """
  select coalesce(in_sp500,false) in_sp500, run, form_type, count(*) mentions, count(distinct project_id) projects_touched, count(distinct accession_number) filings, count(distinct cik) issuers
  from kr group by 1,2,3 order by 1,2,3""")
r.write_json("composition.json", V)
r.finish()