"""Run the parser task acceptance tests through an ISOLATED package loader.

The shared scraper/__init__.py currently imports the in-progress ingest layer
(sibling task t_f595542f) and is transiently broken, so a plain `import scraper`
fails. We therefore register a stub package `scrpkr` pointing at scraper/ and
load only the parser-core modules, leaving the sibling's modules untouched.
"""
import importlib.util, sys, types, logging, json, os, traceback

PKG = "scrpkr"
pkgmod = types.ModuleType(PKG)
pkgmod.__path__ = ["scraper"]
sys.modules[PKG] = pkgmod

def load(sub):
    name = f"{PKG}.{sub}"
    s = importlib.util.spec_from_file_location(name, f"scraper/{sub}.py")
    m = importlib.util.module_from_spec(s)
    sys.modules[name] = m
    s.loader.exec_module(m)
    return m

parsers = load("parsers")
io_mod = load("io")
schema = load("schema")
bill_id = load("bill_id")
fixtures = load("test_fixtures")
parse_cli = load("parse_cli")

# Capture warnings to assert "logged warning".
class CapHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.msgs = []
    def emit(self, rec):
        self.msgs.append(rec.getMessage())

results = []
def check(name, cond, detail=""):
    results.append((name, bool(cond), detail))
    print(("PASS" if cond else "FAIL"), name, "—", detail)

# --- Test 1: clean sample -------------------------------------------------
recs = parsers.parse("samples/assembly_dailyfile_latest.html")
types_ct = {}
for r in recs:
    types_ct[r.record_type] = types_ct.get(r.record_type, 0) + 1
check("clean: total records >= 156", len(recs) >= 156, f"got {len(recs)} by type {types_ct}")

bills = [r for r in recs if r.record_type == "bill"]
sample = bills[0]
have_fields = all(getattr(sample, f) is not None for f in
                   ["bill_number", "bill_id", "session", "short_title"])
check("clean: first bill record has bill_number/bill_id/session/short_title",
      have_fields, f"bill={getattr(sample,'bill_number',None)!r} id={getattr(sample,'bill_id',None)!r} session={getattr(sample,'session',None)!r}")
# The spec's worked example AB28 / item 32 / "Read third time" should be present.
ab28 = next((r for r in bills if r.bill_number == "AB 28"), None)
check("clean: AB 28 present with item_number 32 + action description",
      ab28 is not None and ab28.item_number == 32 and "third time" in (ab28.description or ""),
      f"found={'yes' if ab28 else 'no'}")

# --- Test 2: malformed row skip ------------------------------------------
cap = CapHandler()
cap.setLevel(logging.WARNING)
parsers.log.addHandler(cap)
cap_msgs_before = len(cap.msgs)
try:
    mrecs = parsers.parse_html(fixtures.malformed_document())
    aborted = False
except Exception:
    aborted = True
    mrecs = []
new_warns = len(cap.msgs) - cap_msgs_before
check("malformed: no unhandled exception aborts batch", not aborted, "")
check("malformed: bad rows logged as warnings", new_warns >= 2,
      f"{new_warns} warning(s): {cap.msgs[-3:]}")
check("malformed: valid records still produced (batch survived)",
      any(r.bill_number == "AB 100" for r in mrecs) and any(r.bill_number == "AB 300" for r in mrecs),
      f"produced bill_numbers={[r.bill_number for r in mrecs]}")

# --- Test 3: codec --------------------------------------------------------
dec = bill_id.parse_bill_id("202520260AB28")
check("codec: 202520260AB28 -> AB 28 / 2025-2026 / assembly",
      dec and dec["display"] == "AB 28" and dec["session"] == "2025-2026" and dec["house"] == "assembly",
      str(dec))

# --- Test 4: JSONL + SQLite roundtrip, zero unhandled exceptions ----------
out_dir = "out"
os.makedirs(out_dir, exist_ok=True)
jsonl = os.path.join(out_dir, "records.jsonl")
db = os.path.join(out_dir, "records.db")
ntests = 0
try:
    n_jsonl = io_mod.write_jsonl(recs, jsonl)
    n_db = io_mod.write_sqlite(recs, db)
except Exception:
    traceback.print_exc()
    n_jsonl = n_db = -1
check("io: JSONL wrote all records", n_jsonl == len(recs), f"jsonl={n_jsonl}/{len(recs)}")
check("io: SQLite wrote all records", n_db == len(recs), f"db_rows={n_db}/{len(recs)}")
# JSONL is one JSON object per line
ok_json = True
line_count = 0
with open(jsonl) as fh:
    for line in fh:
        line_count += 1
        try:
            json.loads(line)
        except Exception:
            ok_json = False
check("io: JSONL is one valid JSON object per line", ok_json and line_count == len(recs),
      f"lines={line_count}")

# --- Test 5: CLI end-to-end (parse_cli.handle_parse) ----------------------
import argparse
class _A: pass
cli_args = _A()
cli_args.input = "samples/assembly_dailyfile_latest.html"
cli_args.output = os.path.join(out_dir, "cli_records.jsonl")
cli_args.sqlite_path = None
cli_args.table = "daily_file_records"
cli_args.house = "assembly"
cli_args.source = schema.ASSEMBLY_DAILY_FILE
cli_args.no_hearings = False
cli_args.json_include_nulls = False
cli_args.parser = "lxml"
rc = parse_cli.handle_parse(cli_args)
cli_jsonl = os.path.join(out_dir, "cli_records.jsonl")
cli_lines = 0
if os.path.exists(cli_jsonl):
    with open(cli_jsonl) as fh:
        cli_lines = sum(1 for _ in fh)
check("cli: handle_parse returns 0 and wrote JSONL", rc == 0 and cli_lines == len(recs),
      f"rc={rc} lines={cli_lines}")

# --- verdict --------------------------------------------------------------
failed = [n for n, ok, _ in results if not ok]
print("\n==== SUMMARY ====")
print(f"passed {len(results)-len(failed)}/{len(results)}")
if failed:
    print("FAILED:", failed)
    sys.exit(1)
print("ALL PARSER ACCEPTANCE TESTS PASSED")