#!/usr/bin/env python3
"""Convert the heterogeneous MACStats workbooks into auditable tidy tables.

The extractor deliberately keeps source values unchanged and adds normalized
numeric values only in a separate field.  It also preserves workbook, sheet,
cell, report-edition, table-title, and period provenance for every observation.
"""

from __future__ import annotations

import argparse
import csv
import difflib
import hashlib
import json
import math
import re
import unicodedata
import xml.etree.ElementTree as ET
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable

import openpyxl
import pandas as pd


NOTE_PREFIXES = ("note", "notes", "source", "sources", "definition", "definitions", "methodology")
MISSING_MARKERS = {"", "-", "–", "—", "―", "..", "...", "n/a", "#n/a", "na", "not available"}
STATE_NAMES = [
    "Alabama", "Alaska", "Arizona", "Arkansas", "California", "Colorado", "Connecticut", "Delaware",
    "District of Columbia", "Florida", "Georgia", "Hawaii", "Idaho", "Illinois", "Indiana", "Iowa",
    "Kansas", "Kentucky", "Louisiana", "Maine", "Maryland", "Massachusetts", "Michigan", "Minnesota",
    "Mississippi", "Missouri", "Montana", "Nebraska", "Nevada", "New Hampshire", "New Jersey",
    "New Mexico", "New York", "North Carolina", "North Dakota", "Ohio", "Oklahoma", "Oregon",
    "Pennsylvania", "Rhode Island", "South Carolina", "South Dakota", "Tennessee", "Texas", "Utah",
    "Vermont", "Virginia", "Washington", "West Virginia", "Wisconsin", "Wyoming", "Puerto Rico",
    "American Samoa", "Guam", "Northern Mariana Islands", "U.S. Virgin Islands", "Virgin Islands",
]


def repair_mojibake(text: str) -> str:
    if any(marker in text for marker in ("ΓÇ", "â€", "Ã")):
        for encoding in ("cp437", "latin1"):
            try:
                repaired = text.encode(encoding).decode("utf-8")
                if repaired.count("�") == 0:
                    return repaired
            except (UnicodeEncodeError, UnicodeDecodeError):
                pass
    return text


def clean_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, float) and math.isnan(value):
        return ""
    text = repair_mojibake(str(value)).replace("\xa0", " ").replace("\r", " ").replace("\n", " ")
    return " ".join(text.split())


def slugify(value: str, max_length: int = 90) -> str:
    value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii").lower()
    value = re.sub(r"[^a-z0-9]+", "_", value).strip("_")
    return value[:max_length].rstrip("_") or "untitled"


def normalize_title(title: str) -> str:
    title = clean_text(title).lower()
    title = re.sub(r"\bage\s*1964\b", "age 19 64", title)
    title = re.sub(r"\bage\s*0?18\b", "age 0 18", title)
    title = title.replace("national health interview survey", "nhis")
    title = title.replace("medical expenditures panel survey", "meps")
    title = title.replace("medical expenditure panel survey", "meps")
    title = title.replace("source of health insurance", "source of health coverage")
    title = re.sub(r"children?\s+under\s+age\s+19", "age 0 18", title)
    title = re.sub(r"\be[- ]?fmaps?\b", "enhanced federal medical assistance percentage", title)
    title = re.sub(r"\bfmaps?\b", "federal medical assistance percentage", title)
    title = title.replace("t-msis", "msis").replace("tmsis", "msis")
    title = title.replace("full year equivalent", "fye").replace("full-year equivalent", "fye")
    title = title.replace("long term services and supports", "ltss").replace("long-term services and supports", "ltss")
    title = title.replace("federal poverty level", "fpl")
    title = title.replace("persons served", "beneficiaries")
    title = title.replace("non-institutionalized", "noninstitutionalized")
    title = re.sub(r"^(?:exhibit|table|figure)\s*\d+[a-z]?\s*[.:-]?\s*", "", title)
    title = re.sub(r"\b(?:sfys?|fys?|cys?)\s*\d{4}(?:\s*[-–—]\s*\d{2,4})?\b", " ", title)
    title = re.sub(r"\b(?:january|february|march|april|may|june|july|august|september|october|november|december)\s+\d{1,2}?,?\s*\d{4}\b", " ", title)
    title = re.sub(r"\b20\d{2}\b", " ", title)
    title = re.sub(r"\b19\d{2}\b", " ", title)
    title = re.sub(r"\b(?:millions|thousands|billions)\b", " ", title)
    title = re.sub(r"\b(?:updated|data|book|web)\b", " ", title)
    title = re.sub(r"\bwith standard errors?\b", " ", title)
    title = re.sub(r"[^a-z0-9]+", " ", title)
    return " ".join(title.split())


def title_similarity(left: str, right: str) -> float:
    left_norm, right_norm = normalize_title(left), normalize_title(right)
    left_tokens, right_tokens = set(left_norm.split()), set(right_norm.split())
    union = left_tokens | right_tokens
    jaccard = len(left_tokens & right_tokens) / len(union) if union else 0.0
    sequence = difflib.SequenceMatcher(None, left_norm, right_norm).ratio()
    containment = max(
        len(left_tokens & right_tokens) / max(1, len(left_tokens)),
        len(left_tokens & right_tokens) / max(1, len(right_tokens)),
    )
    score = 0.45 * jaccard + 0.30 * sequence + 0.25 * containment
    # Penalize pairs that share most generic words but disagree on a defining
    # analytic concept.  This prevents, for example, CHIP enrollment from
    # being crosswalked to CHIP spending merely because both are "by state."
    for first, second in (("enrollment", "spending"), ("nhis", "meps")):
        left_first, left_second = first in left_tokens, second in left_tokens
        right_first, right_second = first in right_tokens, second in right_tokens
        if (left_first != right_first or left_second != right_second) and (left_first or left_second) and (right_first or right_second):
            score *= 0.45
    if ("fye" in left_tokens) != ("fye" in right_tokens):
        score *= 0.70
    left_child = {"age", "0", "18"}.issubset(left_tokens)
    right_child = {"age", "0", "18"}.issubset(right_tokens)
    left_adult = {"age", "19", "64"}.issubset(left_tokens)
    right_adult = {"age", "19", "64"}.issubset(right_tokens)
    if (left_child and right_adult) or (left_adult and right_child):
        score *= 0.35
    return score


@dataclass
class SeriesMatch:
    series_id: str
    canonical_title: str
    score: float
    match_method: str
    manual_review: int


class SeriesMatcher:
    def __init__(self, source_manifest: pd.DataFrame, overrides_path: Path) -> None:
        current = source_manifest[source_manifest["source_kind"].eq("current_exhibit")].copy()
        self.anchors = [
            (f"macstats_exhibit_{int(float(row.exhibit_number)):02d}", clean_text(row.exhibit_title))
            for row in current.itertuples()
        ]
        self.anchor_by_number = {series_id.rsplit("_", 1)[-1].lstrip("0") or "0": (series_id, title) for series_id, title in self.anchors}
        self.legacy_anchors: list[tuple[str, str]] = []
        self.overrides: dict[str, tuple[str, str]] = {}
        if overrides_path.exists():
            overrides = pd.read_csv(overrides_path).fillna("")
            for row in overrides.itertuples():
                self.overrides[clean_text(row.normalized_title)] = (clean_text(row.series_id), clean_text(row.canonical_title))

    def match(self, title: str, source_number: str = "", report_edition: str = "") -> SeriesMatch:
        normalized = normalize_title(title)
        if normalized in self.overrides:
            series_id, canonical_title = self.overrides[normalized]
            return SeriesMatch(series_id, canonical_title, 1.0, "manual_override", 0)
        report_year = int(report_edition[:4]) if re.match(r"^20\d{2}", report_edition) else 0
        number = re.sub(r"[^0-9].*$", "", clean_text(source_number)).lstrip("0")
        if report_year >= 2019 and number in self.anchor_by_number:
            direct_id, direct_title = self.anchor_by_number[number]
            direct_score = title_similarity(title, direct_title)
            if direct_score >= 0.58:
                return SeriesMatch(direct_id, direct_title, round(direct_score, 4), "stable_exhibit_number_and_title", 0)
        scored = sorted(
            ((title_similarity(title, anchor_title), series_id, anchor_title) for series_id, anchor_title in self.anchors),
            reverse=True,
        )
        best_score, best_id, best_title = scored[0]
        second_score = scored[1][0]
        if best_score >= 0.50:
            review = int(best_score < 0.64 or best_score - second_score < 0.025)
            return SeriesMatch(best_id, best_title, round(best_score, 4), "current_title_similarity", review)
        if self.legacy_anchors:
            legacy_scores = sorted(
                ((title_similarity(title, legacy_title), legacy_id, legacy_title) for legacy_id, legacy_title in self.legacy_anchors),
                reverse=True,
            )
            legacy_score, legacy_id, legacy_title = legacy_scores[0]
            if legacy_score >= 0.66:
                return SeriesMatch(legacy_id, legacy_title, round(legacy_score, 4), "legacy_title_similarity", 0)
        digest = hashlib.sha1(normalized.encode("utf-8")).hexdigest()[:6]
        legacy_id = f"legacy_{slugify(normalized, 55)}_{digest}"
        canonical = clean_text(title)
        self.legacy_anchors.append((legacy_id, canonical))
        return SeriesMatch(legacy_id, canonical, round(best_score, 4), "new_legacy_title", 1)


def parse_number(value: Any) -> float | None:
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        if isinstance(value, float) and math.isnan(value):
            return None
        return float(value)
    text = clean_text(value)
    if text.lower() in MISSING_MARKERS or text.lower() in {"nr", "ds", "suppressed", "not reported"}:
        return None
    negative = text.startswith("(") and text.endswith(")")
    text = text.strip("() ").replace(",", "").replace("$", "").replace("%", "")
    text = re.sub(r"(?<=\d)[*†‡a-z]+$", "", text, flags=re.I)
    if not re.fullmatch(r"[-+]?\d+(?:\.\d+)?", text):
        return None
    number = float(text)
    return -number if negative else number


def value_status(value: Any) -> str:
    text = clean_text(value).lower()
    if text in MISSING_MARKERS:
        return "not_available"
    if text in {"nr", "not reported"}:
        return "not_reported"
    if text in {"ds", "suppressed"}:
        return "suppressed"
    return "reported"


def first_nonempty(row: list[Any]) -> str:
    return next((clean_text(value) for value in row if clean_text(value)), "")


def is_note_row(row: list[Any]) -> bool:
    text = first_nonempty(row).lower().rstrip(":")
    return any(text == prefix or text.startswith(prefix + ":") for prefix in NOTE_PREFIXES)


def is_title_row(row: list[Any]) -> bool:
    text = first_nonempty(row).lower()
    return bool(re.match(r"^(?:exhibit|table|figure)\s*\d+", text))


def detect_table(matrix: list[list[Any]]) -> tuple[list[int], int, int] | None:
    stats: list[dict[str, Any]] = []
    for row in matrix:
        values = [value for value in row if clean_text(value)]
        numeric = sum(parse_number(value) is not None for value in values)
        first = first_nonempty(row)
        long_prose = len(first) > 220
        numeric_candidate = not is_note_row(row) and not is_title_row(row) and not long_prose and numeric >= 2
        candidate = not is_note_row(row) and not is_title_row(row) and not long_prose and (numeric >= 2 or len(values) >= 3)
        stats.append({"nonempty": len(values), "numeric": numeric, "numeric_candidate": numeric_candidate, "candidate": candidate, "first": first})

    start: int | None = None
    for index in range(1, len(stats)):
        window = stats[index : index + 5]
        if sum(item["numeric_candidate"] for item in window) >= 3:
            start = index + next(i for i, item in enumerate(window) if item["numeric_candidate"])
            break
    if start is None:
        # Text-only crosswalks (for example, service-category mappings) may
        # contain few or no numeric cells.  Treat a multi-column label row
        # followed by a sustained run of nonempty rows as a table.
        for index in range(1, len(stats) - 3):
            if stats[index]["nonempty"] >= 2 and len(stats[index]["first"]) < 180 and not is_title_row(matrix[index]) and not is_note_row(matrix[index]):
                following = stats[index + 1 : index + 7]
                if sum(item["nonempty"] >= 1 for item in following) >= 4:
                    start = index + 1
                    end = start
                    empty_run = 0
                    for body_index in range(start, len(stats)):
                        if is_note_row(matrix[body_index]):
                            break
                        if stats[body_index]["nonempty"] == 0:
                            empty_run += 1
                            if empty_run >= 2:
                                break
                            continue
                        empty_run = 0
                        end = body_index
                    return [index], start, end
        return None

    end = start
    empty_run = 0
    for index in range(start, len(stats)):
        row = matrix[index]
        if is_note_row(row):
            break
        if stats[index]["nonempty"] == 0:
            empty_run += 1
            if empty_run >= 2:
                break
            continue
        empty_run = 0
        first = stats[index]["first"]
        if len(first) > 260 and stats[index]["numeric"] < 2:
            break
        if stats[index]["candidate"] or stats[index]["numeric"] >= 1:
            end = index

    headers: list[int] = []
    for index in range(start - 1, max(-1, start - 8), -1):
        row = matrix[index]
        if is_title_row(row) or len(first_nonempty(row)) > 220:
            break
        if not any(clean_text(value) for value in row):
            if headers:
                break
            continue
        headers.append(index)
    headers.reverse()
    if not headers and start > 0:
        headers = [start - 1]
    return headers, start, end


def merged_lookup(ws: openpyxl.worksheet.worksheet.Worksheet) -> dict[tuple[int, int], Any]:
    lookup: dict[tuple[int, int], Any] = {}
    for merged in ws.merged_cells.ranges:
        value = ws.cell(merged.min_row, merged.min_col).value
        for row in range(merged.min_row, merged.max_row + 1):
            for col in range(merged.min_col, merged.max_col + 1):
                lookup[(row, col)] = value
    return lookup


def header_path(
    ws: openpyxl.worksheet.worksheet.Worksheet,
    merged: dict[tuple[int, int], Any],
    header_rows: list[int],
    column: int,
) -> list[str]:
    parts: list[str] = []
    for row0 in header_rows:
        row = row0 + 1
        value = merged.get((row, column), ws.cell(row, column).value)
        text = clean_text(value)
        if not text:
            for previous_column in range(column - 1, 0, -1):
                previous_value = merged.get((row, previous_column), ws.cell(row, previous_column).value)
                previous_text = clean_text(previous_value)
                if previous_text:
                    if previous_column > 1 or previous_text.lower() not in {"state", "year", "fiscal year", "calendar year"}:
                        text = previous_text
                    break
        if text and (not parts or text != parts[-1]):
            parts.append(text)
    return parts


def strip_state_footnotes(label: str) -> tuple[str, str]:
    cleaned = clean_text(label)
    cleaned = re.sub(r'^N\.\s*Mariana Islands', 'Northern Mariana Islands', cleaned, flags=re.I)
    for state in sorted(STATE_NAMES, key=len, reverse=True):
        if re.fullmatch(re.escape(state) + r"[\s,;:*†‡\dA-Za-z.-]*", cleaned, flags=re.I):
            marker = cleaned[len(state) :].strip(" ,;:.-")
            return state, marker
    return "", ""


def infer_period(title: str, column_path_text: str, row_label: str) -> tuple[str, str, str]:
    candidates = [row_label, column_path_text, title]
    for candidate in candidates:
        if not re.search(r"\b(?:19|20)\d{2}\b", candidate):
            continue
        years = re.findall(r"\b(?:19|20)\d{2}\b", candidate)
        year = years[0] if len(years) == 1 else years[-1]
        lowered = candidate.lower()
        if re.search(r"\bsfy", lowered):
            kind = "state_fiscal_year"
        elif re.search(r"\bfy", lowered) or (row_label.strip() == year and re.search(r"\bfy", title.lower())):
            kind = "fiscal_year"
        elif re.search(r"\bcy", lowered) or 'national health expenditures' in title.lower():
            kind = "calendar_year"
        elif re.search(r"january|february|march|april|may|june|july|august|september|october|november|december", lowered):
            kind = "point_in_time"
        else:
            kind = "year"
        return year, kind, candidate
    return "", "", ""


def infer_unit(title: str, column_path_text: str, number_format: str) -> str:
    text = f"{column_path_text} {title}".lower()
    column = column_path_text.lower()
    # A mixed exhibit's title must not turn its count/dollar columns into rates.
    if '%' in number_format:
        return 'percent'
    if not re.search(r'percent|percentage|%|share|growth', column):
        if re.search(r'total.*enrollees|number of|enrollment.*(?:thousand|million)|fye enrollment', column):
            return next((u for u in ('billions', 'millions', 'thousands') if u[:-1] in text), 'source_units')
        if '$' in number_format or 'income as a percentage of the fpl for various family sizes' in title.lower():
            return next((u for u in ('billions', 'millions', 'thousands') if u[:-1] in text), 'dollars')
    if "%" in number_format or "percent" in text or "percentage" in text or "fmap" in text:
        return "percent"
    if "billion" in text:
        return "billions"
    if "million" in text:
        return "millions"
    if "thousand" in text:
        return "thousands"
    if "$" in number_format or "spending per" in text or "dollar" in text:
        return "dollars"
    return "source_units"


def infer_data_source(title: str, matrix: list[list[Any]]) -> str:
    cells = [clean_text(value) for row in matrix for value in row if clean_text(value)]
    source_notes = [re.split(r'\bSources?:', value, flags=re.I)[-1] for value in cells if re.search(r'\bSources?:', value, re.I)]
    # Methodological comparisons can mention sources not used for the exhibit.
    text = ' '.join([title] + (source_notes or cells)).lower()
    sources: list[str] = []
    if "national health interview survey" in text or re.search(r"\bnhis\b", text):
        sources.append("NHIS")
    if "medical expenditure panel survey" in text or "medical expenditures panel survey" in text or re.search(r"\bmeps\b", text):
        sources.append("MEPS")
    if "national survey of children with special health care needs" in text or "ns-cshcn" in text:
        sources.append("NS-CSHCN")
    if "national survey of children’s health" in text or "national survey of children's health" in text or re.search(r"\bnsch\b", text):
        sources.append("NSCH")
    if ("t-msis" in text or "tmsis" in text or re.search(r"\bmsis\b", text)) and "cms-64" in text:
        sources.append("T-MSIS/CMS-64")
    elif "cms-64" in text:
        sources.append("CMS-64")
    elif 't-msis' in text or 'tmsis' in text:
        sources.append('T-MSIS')
    elif re.search(r'\bmsis\b', text):
        sources.append('MSIS')
    if "national health expenditure" in text:
        sources.append("National Health Expenditure Accounts")
    if "american community survey" in text or re.search(r"\bacs\b", text):
        sources.append("ACS")
    return "; ".join(dict.fromkeys(sources))


def normalize_numeric(value: float | None, unit: str, number_format: str) -> tuple[float | None, str]:
    if value is None:
        return None, ""
    if unit == "percent":
        return (value * 100 if "%" in number_format else value), "percentage_points"
    if unit == "billions":
        return value * 1_000_000_000, "base_units"
    if unit == "millions":
        return value * 1_000_000, "base_units"
    if unit == "thousands":
        return value * 1_000, "base_units"
    return value, unit


def dimension_columns(matrix: list[list[Any]], start: int, end: int) -> int:
    body = matrix[start : end + 1]
    max_cols = max((len(row) for row in body), default=1)
    for column0 in range(1, max_cols):
        values = [row[column0] for row in body if column0 < len(row) and clean_text(row[column0])]
        numeric_count = sum(parse_number(value) is not None for value in values)
        if numeric_count >= 2 and numeric_count / max(1, len(values)) >= 0.25:
            return column0
    return 1


def table_title(profile: dict[str, Any], workbook_sheet_count: int) -> str:
    file_title = clean_text(profile.get("file_title", ""))
    sheet_title = clean_text(profile.get("sheet_title", ""))
    file_no = clean_text(profile.get("file_exhibit_number", ""))
    sheet_no = clean_text(profile.get("sheet_exhibit_number", ""))
    if sheet_title and len(sheet_title.split()) >= 4:
        return sheet_title
    return file_title or sheet_title


def sheet_number(name: str) -> str:
    match = re.search(r"(?:EXHIBIT|TABLE|EX)\s*[.-]?\s*(\d+[A-Za-z]?)", name, flags=re.I)
    return match.group(1) if match else ""


def workbook_exclusions(profiles: list[dict[str, Any]]) -> dict[str, str]:
    exclusions: dict[str, str] = {}
    names = [clean_text(profile["sheet_name"]) for profile in profiles]
    has_web = any("web" in name.lower() for name in names)
    bundle = len(profiles) > 10
    file_number = clean_text(profiles[0].get("file_exhibit_number", "")) if profiles else ""
    for profile in profiles:
        name = clean_text(profile["sheet_name"])
        lowered = name.lower()
        if re.search(r"\bcalc(?:ulation)?\b|underlying|raw data", lowered):
            exclusions[name] = "calculation_or_underlying_sheet"
        elif re.search(r"\bcontents?\b|tab guide", lowered):
            exclusions[name] = "contents_or_guide_sheet"
        elif "book" in lowered and has_web:
            exclusions[name] = "book_layout_duplicate_prefer_web"
        elif bundle and file_number and sheet_number(name) and sheet_number(name) != file_number:
            exclusions[name] = "duplicate_sheet_in_compiled_workbook"
    return exclusions


def extract_sheet(
    ws: openpyxl.worksheet.worksheet.Worksheet,
    profile: dict[str, Any],
    matcher: SeriesMatcher,
    workbook_sheet_count: int,
) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, str]]]:
    max_row = int(profile.get("last_nonempty_row") or ws.max_row)
    max_col = int(profile.get("last_nonempty_column") or ws.max_column)
    matrix = [[ws.cell(row, col).value for col in range(1, max_col + 1)] for row in range(1, max_row + 1)]
    detection = detect_table(matrix)
    title = table_title(profile, workbook_sheet_count)
    source_number = clean_text(profile.get("sheet_exhibit_number") or profile.get("file_exhibit_number"))
    data_source = infer_data_source(title, matrix)
    mapping_title = f"{title} {data_source}" if data_source in {"NHIS", "MEPS"} else title
    series = matcher.match(mapping_title, source_number, profile["report_edition"])
    table_key = hashlib.sha1(
        f"{profile['report_edition']}|{profile['workbook_path']}|{profile['sheet_name']}".encode("utf-8")
    ).hexdigest()[:16]
    issues: list[dict[str, str]] = []
    catalog = {
        "table_key": table_key,
        "series_id": series.series_id,
        "canonical_title": series.canonical_title,
        "source_table_title": title,
        "source_exhibit_number": source_number,
        "report_edition": profile["report_edition"],
        "source_kind": profile["source_kind"],
        "workbook_file_name": profile["workbook_file_name"],
        "workbook_path": profile["workbook_path"],
        "sheet_name": profile["sheet_name"],
        "estimate_type": "standard_error" if re.search(r"\bSEs?\b", profile["sheet_name"], flags=re.I) else "estimate",
        "data_source": data_source,
        "series_match_score": series.score,
        "series_match_method": series.match_method,
        "manual_review": series.manual_review,
        "header_rows": "",
        "data_start_row": "",
        "data_end_row": "",
        "dimension_column_count": "",
        "observation_count": 0,
        "table_fingerprint": "",
    }
    if detection is None:
        chart_observations = extract_chart_caches(
            Path(profile.get("_resolved_workbook_path", profile["workbook_path"])), profile, title, series, data_source, table_key
        )
        if chart_observations:
            fingerprint_payload = [
                (row["row_label"], row["column_path"], row["value_raw"], row["estimate_type"])
                for row in chart_observations
            ]
            catalog.update(
                {
                    "header_rows": "chart_cache",
                    "data_start_row": "chart_cache",
                    "data_end_row": "chart_cache",
                    "dimension_column_count": 1,
                    "observation_count": len(chart_observations),
                    "table_fingerprint": hashlib.sha256(
                        json.dumps(fingerprint_payload, ensure_ascii=False).encode("utf-8")
                    ).hexdigest(),
                }
            )
            return catalog, chart_observations, issues
        issues.append({"table_key": table_key, "issue_type": "no_tabular_block_detected", "detail": title})
        return catalog, [], issues

    header_rows, start, end = detection
    first_measure0 = dimension_columns(matrix, start, end)
    merged = merged_lookup(ws)
    observations: list[dict[str, Any]] = []
    for row0 in range(start, end + 1):
        row_values = matrix[row0]
        row_dimensions: list[str] = []
        for col0 in range(first_measure0):
            value = merged.get((row0 + 1, col0 + 1), row_values[col0] if col0 < len(row_values) else None)
            row_dimensions.append(clean_text(value))
        row_label = row_dimensions[0] if row_dimensions else ""
        if not row_label and not any(clean_text(value) for value in row_values[first_measure0:]):
            continue
        state, footnote_marker = strip_state_footnotes(row_label)
        total_label = row_label.lower().strip()
        is_total = int(bool(re.match(r"^(?:total|totals|subtotal|all states|united states|u\.s\. total|national total|percent(?:age)? of total)\b", total_label)))
        for col0 in range(first_measure0, max_col):
            raw = row_values[col0] if col0 < len(row_values) else None
            if not clean_text(raw):
                continue
            headers = header_path(ws, merged, header_rows, col0 + 1)
            column_path_text = " | ".join(headers) or f"column_{col0 + 1}"
            number = parse_number(raw)
            status = value_status(raw)
            number_format = clean_text(ws.cell(row0 + 1, col0 + 1).number_format)
            unit = infer_unit(title, column_path_text, number_format)
            normalized, normalized_unit = normalize_numeric(number, unit, number_format)
            period_year, period_type, period_label = infer_period(title, column_path_text, row_label)
            observation = {
                "series_id": series.series_id,
                "canonical_title": series.canonical_title,
                "table_key": table_key,
                "report_edition": profile["report_edition"],
                "data_source": data_source,
                "source_exhibit_number": catalog["source_exhibit_number"],
                "source_table_title": title,
                "estimate_type": catalog["estimate_type"],
                "period_year": period_year,
                "period_type": period_type,
                "period_label": period_label,
                "state": state,
                "row_label": row_label,
                "row_footnote_marker": footnote_marker,
                "row_dimension_1": row_dimensions[0] if len(row_dimensions) > 0 else "",
                "row_dimension_2": row_dimensions[1] if len(row_dimensions) > 1 else "",
                "row_dimension_3": row_dimensions[2] if len(row_dimensions) > 2 else "",
                "row_dimension_4": row_dimensions[3] if len(row_dimensions) > 3 else "",
                "column_path": column_path_text,
                "column_dimension_1": headers[0] if len(headers) > 0 else "",
                "column_dimension_2": headers[1] if len(headers) > 1 else "",
                "column_dimension_3": headers[2] if len(headers) > 2 else "",
                "column_dimension_4": headers[3] if len(headers) > 3 else "",
                "metric": headers[-1] if headers else f"column_{col0 + 1}",
                "value_raw": clean_text(raw),
                "value_numeric": number,
                "value_status": status,
                "source_unit": unit,
                "value_normalized": normalized,
                "normalized_unit": normalized_unit,
                "is_total_row": is_total,
                "is_note_row": 0,
                "include_in_reporting": int(not is_total),
                "source_workbook": profile["workbook_file_name"],
                "source_sheet": profile["sheet_name"],
                "source_cell": ws.cell(row0 + 1, col0 + 1).coordinate,
                "source_path": profile["workbook_path"],
            }
            observations.append(observation)

    fingerprint_payload = [
        (row["row_label"], row["column_path"], row["value_raw"], row["estimate_type"]) for row in observations
    ]
    fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, ensure_ascii=False).encode("utf-8")).hexdigest()
    catalog.update(
        {
            "header_rows": ";".join(str(row + 1) for row in header_rows),
            "data_start_row": start + 1,
            "data_end_row": end + 1,
            "dimension_column_count": first_measure0,
            "observation_count": len(observations),
            "table_fingerprint": fingerprint,
        }
    )
    if not observations:
        issues.append({"table_key": table_key, "issue_type": "zero_observations", "detail": title})
    return catalog, observations, issues


def chart_points(container: ET.Element | None, namespace: dict[str, str]) -> dict[int, str]:
    if container is None:
        return {}
    points: dict[int, str] = {}
    for point in container.findall(".//c:pt", namespace):
        index = int(point.attrib.get("idx", len(points)))
        value = point.find("c:v", namespace)
        if value is not None and value.text is not None:
            points[index] = clean_text(value.text)
    return points


def extract_chart_caches(
    workbook_path: Path,
    profile: dict[str, Any],
    title: str,
    series: SeriesMatch,
    data_source: str,
    table_key: str,
) -> list[dict[str, Any]]:
    namespace = {"c": "http://schemas.openxmlformats.org/drawingml/2006/chart"}
    observations: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()
    try:
        with zipfile.ZipFile(workbook_path) as archive:
            chart_files = sorted(name for name in archive.namelist() if re.fullmatch(r"xl/charts/chart\d+\.xml", name))
            for chart_file in chart_files:
                root = ET.fromstring(archive.read(chart_file))
                for series_index, chart_series in enumerate(root.findall(".//c:ser", namespace), start=1):
                    series_name_node = chart_series.find(".//c:tx//c:v", namespace)
                    series_name = clean_text(series_name_node.text if series_name_node is not None else f"series_{series_index}")
                    categories = chart_points(chart_series.find("c:cat", namespace), namespace)
                    if not categories:
                        categories = chart_points(chart_series.find("c:xVal", namespace), namespace)
                    values_container = chart_series.find("c:val", namespace)
                    if values_container is None:
                        values_container = chart_series.find("c:yVal", namespace)
                    values = chart_points(values_container, namespace)
                    format_node = values_container.find(".//c:formatCode", namespace) if values_container is not None else None
                    number_format = clean_text(format_node.text if format_node is not None else "")
                    for point_index, raw in values.items():
                        row_label = categories.get(point_index, f"point_{point_index + 1}")
                        dedup_key = (series_name, row_label, raw)
                        if dedup_key in seen:
                            continue
                        seen.add(dedup_key)
                        number = parse_number(raw)
                        unit = infer_unit(title, series_name, number_format)
                        normalized, normalized_unit = normalize_numeric(number, unit, number_format)
                        period_year, period_type, period_label = infer_period(title, series_name, row_label)
                        observations.append(
                            {
                                "series_id": series.series_id,
                                "canonical_title": series.canonical_title,
                                "table_key": table_key,
                                "report_edition": profile["report_edition"],
                                "data_source": data_source,
                                "source_exhibit_number": clean_text(profile.get("sheet_exhibit_number") or profile.get("file_exhibit_number")),
                                "source_table_title": title,
                                "estimate_type": "estimate",
                                "period_year": period_year,
                                "period_type": period_type,
                                "period_label": period_label,
                                "state": "",
                                "row_label": row_label,
                                "row_footnote_marker": "",
                                "row_dimension_1": row_label,
                                "row_dimension_2": "",
                                "row_dimension_3": "",
                                "row_dimension_4": "",
                                "column_path": series_name,
                                "column_dimension_1": series_name,
                                "column_dimension_2": "",
                                "column_dimension_3": "",
                                "column_dimension_4": "",
                                "metric": series_name,
                                "value_raw": raw,
                                "value_numeric": number,
                                "value_status": value_status(raw),
                                "source_unit": unit,
                                "value_normalized": normalized,
                                "normalized_unit": normalized_unit,
                                "is_total_row": int(row_label.lower().startswith(("all enrollees", "total"))),
                                "is_note_row": 0,
                                "include_in_reporting": int(not row_label.lower().startswith(("all enrollees", "total"))),
                                "source_workbook": profile["workbook_file_name"],
                                "source_sheet": profile["sheet_name"],
                                "source_cell": f"{chart_file}#series-{series_index}-point-{point_index}",
                                "source_path": profile["workbook_path"],
                            }
                        )
    except (zipfile.BadZipFile, ET.ParseError, KeyError):
        return []
    return observations


def write_csv(path: Path, rows: Iterable[dict[str, Any]], columns: list[str] | None = None) -> None:
    rows = list(rows)
    path.parent.mkdir(parents=True, exist_ok=True)
    if columns is None:
        columns = list(rows[0]) if rows else []
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore")
        if columns:
            writer.writeheader()
            writer.writerows(rows)


def data_dictionary_rows() -> list[dict[str, str]]:
    definitions = {
        "series_id": "Stable cross-report identifier based on the February 2026 exhibit title; legacy_* indicates no confident current-series match.",
        "canonical_title": "Current exhibit title used as the cross-year series label.",
        "table_key": "Stable hash of report edition, workbook path, and worksheet.",
        "report_edition": "MACStats report publication edition in YYYY-MM form.",
        "data_source": "Primary source inferred from the table title and source notes when identifiable (for example, NHIS, MEPS, ACS, CMS-64, or T-MSIS/CMS-64).",
        "source_exhibit_number": "Exhibit or table number printed in the source edition; not assumed stable across years.",
        "source_table_title": "Title printed in or derived from the source workbook.",
        "estimate_type": "estimate or standard_error, derived from the worksheet label.",
        "period_year": "Four-digit analytic year inferred from the row, column header, or table title, in that order.",
        "period_type": "fiscal_year, state_fiscal_year, calendar_year, point_in_time, or year.",
        "period_label": "Exact source text used to infer period_year and period_type.",
        "state": "Standard state or territory name when row_label can be safely resolved.",
        "row_label": "Primary row identifier exactly as published, normalized only for whitespace.",
        "row_footnote_marker": "Text appended to a recognized state name, usually a footnote marker.",
        "row_dimension_1": "First source row-dimension value.",
        "row_dimension_2": "Second source row-dimension value, when present.",
        "row_dimension_3": "Third source row-dimension value, when present.",
        "row_dimension_4": "Fourth source row-dimension value, when present.",
        "column_path": "Pipe-delimited hierarchy constructed from the source worksheet's multirow headers.",
        "column_dimension_1": "First level of the source column-header hierarchy.",
        "column_dimension_2": "Second level of the source column-header hierarchy.",
        "column_dimension_3": "Third level of the source column-header hierarchy.",
        "column_dimension_4": "Fourth level of the source column-header hierarchy.",
        "metric": "Lowest nonblank level in the source column-header hierarchy.",
        "value_raw": "Source cell value with whitespace normalized; symbols and status codes are retained.",
        "value_numeric": "Numeric interpretation of value_raw without changing the source scale.",
        "value_status": "reported, not_available, not_reported, or suppressed.",
        "source_unit": "Unit inferred from cell format, column header, and title: percent, billions, millions, thousands, dollars, or source_units.",
        "value_normalized": "Numeric value converted to percentage points or base units when the source unit is explicit.",
        "normalized_unit": "percentage_points, base_units, dollars, source_units, or blank for nonnumeric cells.",
        "is_total_row": "1 when row_label is a recognized total or national summary; otherwise 0.",
        "is_note_row": "1 for a note/source row; these rows are normally excluded before observation creation.",
        "include_in_reporting": "0 for recognized total/subtotal/composition rows; 1 otherwise. This is only a row filter: it does not identify additive columns, harmonize categories, or make all remaining cells safe to sum.",
        "source_workbook": "Original workbook filename.",
        "source_sheet": "Original worksheet name.",
        "source_cell": "Original Excel cell coordinate, or chart-cache series/point locator when a workbook stores values only in an embedded chart.",
        "source_path": "Local source-workbook path retained for audit and pipeline reruns.",
    }
    return [{"field": field, "type": "string" if field not in {"value_numeric", "value_normalized", "is_total_row", "is_note_row", "include_in_reporting"} else "numeric", "description": description} for field, description in definitions.items()]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--project-root", type=Path, default=Path(__file__).resolve().parents[1])
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    project = args.project_root
    source_manifest = pd.read_csv(project / "data" / "raw" / "source_manifest.csv")
    profiles = [json.loads(line) for line in (project / "data" / "interim" / "workbook_profile.jsonl").read_text(encoding="utf-8").splitlines()]
    matcher = SeriesMatcher(source_manifest, project / "config" / "series_overrides.csv")

    profile_groups: dict[str, list[dict[str, Any]]] = {}
    for profile in profiles:
        profile_groups.setdefault(profile["workbook_path"], []).append(profile)

    table_catalog: list[dict[str, Any]] = []
    observations_by_table: dict[str, list[dict[str, Any]]] = {}
    issues: list[dict[str, str]] = []
    exclusions: list[dict[str, str]] = []

    for workbook_index, (workbook_path, workbook_profiles) in enumerate(sorted(profile_groups.items()), start=1):
        resolved_workbook_path = Path(workbook_path)
        if not resolved_workbook_path.is_absolute():
            resolved_workbook_path = project / resolved_workbook_path
        workbook = openpyxl.load_workbook(resolved_workbook_path, read_only=False, data_only=True)
        excluded = workbook_exclusions(workbook_profiles)
        profile_by_name = {profile["sheet_name"]: profile for profile in workbook_profiles}
        for sheet_name in workbook.sheetnames:
            profile = profile_by_name.get(sheet_name)
            if profile is None:
                continue
            profile = dict(profile)
            profile["_resolved_workbook_path"] = str(resolved_workbook_path)
            ws = workbook[sheet_name]
            if ws.sheet_state != "visible":
                exclusions.append({"workbook_path": workbook_path, "sheet_name": sheet_name, "reason": "hidden_sheet"})
                continue
            if sheet_name in excluded:
                exclusions.append({"workbook_path": workbook_path, "sheet_name": sheet_name, "reason": excluded[sheet_name]})
                continue
            if table_title(profile, len(workbook_profiles)).lower().startswith("please see the macstats pdf"):
                exclusions.append({"workbook_path": workbook_path, "sheet_name": sheet_name, "reason": "pdf_reference_without_excel_table"})
                continue
            catalog, observations, sheet_issues = extract_sheet(ws, profile, matcher, len(workbook_profiles))
            table_catalog.append(catalog)
            observations_by_table[catalog["table_key"]] = observations
            issues.extend(sheet_issues)
        workbook.close()
        if workbook_index % 50 == 0:
            print(f"Processed {workbook_index}/{len(profile_groups)} workbooks", flush=True)

    # Drop exact within-edition duplicates while retaining a transparent audit row.
    catalog_df = pd.DataFrame(table_catalog)
    duplicate_rows: list[dict[str, Any]] = []
    keep_keys: set[str] = set()
    for _, group in catalog_df.groupby(["report_edition", "series_id", "estimate_type", "table_fingerprint"], dropna=False):
        group = group.sort_values(["observation_count", "workbook_file_name"], ascending=[False, True])
        keeper = group.iloc[0]
        keep_keys.add(keeper.table_key)
        for row in group.iloc[1:].itertuples():
            duplicate_rows.append(
                {
                    "kept_table_key": keeper.table_key,
                    "dropped_table_key": row.table_key,
                    "report_edition": row.report_edition,
                    "series_id": row.series_id,
                    "reason": "exact_table_fingerprint_duplicate",
                }
            )

    kept_catalog = catalog_df[catalog_df["table_key"].isin(keep_keys)].copy()
    all_observations = [row for key in keep_keys for row in observations_by_table.get(key, [])]
    observations_df = pd.DataFrame(all_observations)
    observation_columns = list(data_dictionary_rows()[0].keys())  # placeholder overwritten below
    observation_columns = [row["field"] for row in data_dictionary_rows()]
    if not observations_df.empty:
        observations_df = observations_df.reindex(columns=observation_columns)
        observations_df = observations_df.sort_values(
            ["series_id", "period_year", "report_edition", "estimate_type", "row_label", "column_path", "source_cell"],
            na_position="last",
        )

    processed = project / "data" / "processed"
    processed.mkdir(parents=True, exist_ok=True)
    observations_df.to_csv(processed / "macstats_tidy_long.csv", index=False)
    latest_keys = [
        "series_id", "estimate_type", "period_year", "period_type", "state",
        "row_label", "row_dimension_2", "row_dimension_3", "row_dimension_4",
        "column_path", "metric", "source_unit", "data_source",
    ]
    latest_vintage = (
        observations_df.sort_values("report_edition", ascending=False)
        .drop_duplicates(subset=latest_keys, keep="first")
        .sort_values(["series_id", "period_year", "estimate_type", "row_label", "column_path"], na_position="last")
    )
    latest_vintage.to_csv(processed / "macstats_latest_vintage.csv", index=False)
    kept_catalog.sort_values(["series_id", "report_edition", "source_exhibit_number"]).to_csv(processed / "table_catalog.csv", index=False)
    catalog_df[[
        "table_key", "report_edition", "source_exhibit_number", "source_table_title", "series_id",
        "canonical_title", "series_match_score", "series_match_method", "manual_review",
    ]].to_csv(processed / "series_crosswalk.csv", index=False)
    series_catalog = (
        kept_catalog.groupby("series_id", as_index=False)
        .agg(canonical_title=("canonical_title", "first"), first_report_edition=("report_edition", "min"), last_report_edition=("report_edition", "max"), table_count=("table_key", "count"), manual_review=("manual_review", "max"))
        .sort_values("series_id")
    )
    series_catalog.to_csv(processed / "series_catalog.csv", index=False)
    write_csv(processed / "data_dictionary.csv", data_dictionary_rows(), ["field", "type", "description"])

    tables_dir = processed / "tables"
    tables_dir.mkdir(parents=True, exist_ok=True)
    latest_tables_dir = processed / "tables_latest"
    latest_tables_dir.mkdir(parents=True, exist_ok=True)
    for generated_dir in (tables_dir, latest_tables_dir):
        for stale_csv in generated_dir.glob("*.csv"):
            stale_csv.unlink()
    if not observations_df.empty:
        for series_id, group in observations_df.groupby("series_id", sort=True):
            group.to_csv(tables_dir / f"{series_id}.csv", index=False)
        for series_id, group in latest_vintage.groupby("series_id", sort=True):
            group.to_csv(latest_tables_dir / f"{series_id}.csv", index=False)

    qa = project / "qa"
    qa.mkdir(parents=True, exist_ok=True)
    write_csv(qa / "extraction_issues.csv", issues, ["table_key", "issue_type", "detail"])
    write_csv(qa / "excluded_worksheets.csv", exclusions, ["workbook_path", "sheet_name", "reason"])
    write_csv(qa / "duplicate_tables.csv", duplicate_rows, ["kept_table_key", "dropped_table_key", "report_edition", "series_id", "reason"])
    coverage = (
        kept_catalog.groupby("report_edition", as_index=False)
        .agg(tables=("table_key", "count"), series=("series_id", "nunique"), observations=("observation_count", "sum"), tables_needing_review=("manual_review", "sum"))
    )
    coverage.to_csv(qa / "coverage_summary.csv", index=False)

    print(f"Kept {len(kept_catalog)} tables across {series_catalog.shape[0]} series")
    print(f"Wrote {len(observations_df):,} tidy observations")
    print(f"Wrote {len(latest_vintage):,} latest-vintage observations")
    print(f"Flagged {len(issues)} extraction issues and {int(kept_catalog.manual_review.sum())} table mappings for review")


if __name__ == "__main__":
    main()
