10_generate_pipeline_statistics.py

#!/usr/bin/env python3
"""
Generate summary tables for the automated literature-review pipeline.

Run:
    python 10_generate_pipeline_statistics.py

The script prints all tables to the terminal and writes CSV copies to:
    pipeline_statistics/
"""

from __future__ import annotations

import csv
import json
import re
from pathlib import Path
from typing import Iterable


BASE_DIR = Path(__file__).resolve().parent
DATABASE_DIR = BASE_DIR / "Database"
OUTPUT_DIR = BASE_DIR / "pipeline_statistics"

# Output labels used throughout the literature-review pipeline.
GENES = [
    "CTC1",
    "NAF1",
    "NHP2",
    "NOP10",
    "PARN",
    "POT1",
    "RTEL1",
    "TER",
    "TERT",
    "TINF2",
    "TPP1",
    "WRAP53",
    "ZCCHC8",
]

CLINVAR_FIELD_DESCRIPTIONS = {
    "Variation_ID": "ClinVar variation identifier",
    "Variant_Name": "ClinVar variant description",
    "Protein_Change": "Reported protein-level change",
    "Germline_Classification": "ClinVar germline classification",
    "Review_Status": "ClinVar review status",
    "Stars": "Review-star rating derived from review status",
    "Condition": "Associated disease or condition",
    "Citations_PubMed": "Linked PubMed identifiers",
}

AI_FIELD_DESCRIPTIONS = {
    "PMID": "PubMed identifier of the source article",
    "Article_Reference": "Article citation",
    "Article_DOI": "Article digital object identifier",
    "cDNA_Change": "Reported cDNA-level variant",
    "Protein_Change": "Reported protein-level variant",
    "Exon": "Reported exon",
    "Variant_Type": "Missense, nonsense, frameshift, splice, or other type",
    "Condition": "Associated clinical condition",
    "Zygosity": "Reported zygosity",
    "Compound_Het_Partner": "Compound-heterozygous partner variant",
    "Number_of_Patients": "Number of patients carrying the variant",
    "Inheritance": "Reported inheritance pattern",
    "Age_of_Onset": "Reported age of onset",
    "Classification_in_Paper": "Variant classification reported by the paper",
    "Functional_Evidence": "Whether functional evidence was reported",
    "Functional_Evidence_Detail": "Description of functional evidence",
    "Evidence_Sentence": "Exact supporting sentence from the article",
    "Table_or_Figure": "Supporting table or figure",
    "Page_Context": "Article section containing the evidence",
    "Variant_Previously_Reported": "Whether the variant was previously reported",
    "Cites_Which_Papers": "Earlier papers cited for the variant",
    "In_ClinVar": "ClinVar inclusion status",
    "Extraction_Confidence": "AI-reported extraction confidence",
    "Extraction_Notes": "Extraction caveats and notes",
}

AA3_TO_1 = {
    "ala": "A", "arg": "R", "asn": "N", "asp": "D", "cys": "C",
    "gln": "Q", "glu": "E", "gly": "G", "his": "H", "ile": "I",
    "leu": "L", "lys": "K", "met": "M", "phe": "F", "pro": "P",
    "ser": "S", "thr": "T", "trp": "W", "tyr": "Y", "val": "V",
    "ter": "*", "stop": "*", "x": "*",
}

EMPTY_VARIANT_TOKENS = {
    "", "na", "n/a", "none", "not stated", "not provided",
    "not specified", "unknown", "nan",
}


def read_csv_rows(path: Path) -> list[dict[str, str]]:
    if not path.exists():
        return []
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def read_csv_columns(path: Path) -> list[str]:
    if not path.exists():
        return []
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.reader(handle)
        return next(reader, [])


def count_nonempty(rows: Iterable[dict[str, str]], field: str) -> int:
    return sum(1 for row in rows if str(row.get(field, "")).strip())


def count_unique_nonempty(rows: Iterable[dict[str, str]], field: str) -> int:
    return len(
        {
            str(row.get(field, "")).strip()
            for row in rows
            if str(row.get(field, "")).strip()
        }
    )


def paper_folders(gene: str) -> list[Path]:
    gene_dir = DATABASE_DIR / gene.lower()
    if not gene_dir.exists():
        return []
    return sorted(path for path in gene_dir.iterdir() if path.is_dir())


def folder_has_pdf(folder: Path) -> bool:
    return any(path.is_file() and path.suffix.lower() == ".pdf" for path in folder.iterdir())


def count_files(gene: str, filename: str) -> int:
    gene_dir = DATABASE_DIR / gene.lower()
    if not gene_dir.exists():
        return 0
    return sum(1 for _ in gene_dir.rglob(filename))


def count_unique_metadata_dois(gene: str) -> int:
    """Count unique DOI values recorded in downloaded-article metadata files."""
    gene_dir = DATABASE_DIR / gene.lower()
    if not gene_dir.exists():
        return 0

    dois: set[str] = set()
    for path in gene_dir.rglob("metadata.txt"):
        for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
            match = re.match(r"\s*DOI\s{2,}(.+?)\s*$", line)
            if match:
                doi = match.group(1).strip().lower()
                if doi:
                    dois.add(doi)
                break
    return len(dois)


def normalize_variant_value(value: str) -> str:
    value = str(value or "").strip().lower()
    if value in {"", "not stated", "not provided", "unknown", "none", "n/a", "nan"}:
        return ""
    return re.sub(r"[\s()]", "", value)


def clean_variant_text(value: str) -> str:
    text = str(value or "").strip()
    return "" if text.lower() in EMPTY_VARIANT_TOKENS else text


def canonical_cdna(value: str) -> str:
    """Use the same cDNA normalization rules as the evaluation script."""
    text = clean_variant_text(value)
    if not text:
        return ""
    match = re.search(r"c\.\s*([A-Za-z0-9_+\-*>.?]+)", text)
    token = match.group(1) if match else text
    token = token.replace(" ", "").replace("_", "").replace(";", "").replace(",", "")
    token = re.sub(r"[^a-z0-9+\-*>.?]", "", token.lower())
    return f"c.{token}" if token else ""


def canonical_protein(value: str) -> str:
    """Use the same protein normalization rules as the evaluation script."""
    token = clean_variant_text(value)
    if not token:
        return ""
    if token.lower().startswith("p."):
        token = token[2:]
    token = token.replace("(", "").replace(")", "").replace(" ", "")
    for aa3, aa1 in AA3_TO_1.items():
        token = re.sub(aa3, aa1, token, flags=re.IGNORECASE)
    token = token.upper().replace("TER", "*").replace("STOP", "*")
    token = re.sub(r"[^A-Z0-9*?_]", "", token)
    return f"p.{token}" if token else ""


def variant_identifier(cdna: str, protein: str) -> tuple[str, str]:
    if cdna:
        return cdna, "cDNA"
    if protein:
        return protein, "Protein_fallback"
    return "", ""


def join_values(values: Iterable[str]) -> str:
    cleaned = {str(value).strip() for value in values if str(value).strip()}
    return "; ".join(sorted(cleaned))


def unique_ai_variants(rows: Iterable[dict[str, str]]) -> int:
    identities: set[str] = set()
    for row in rows:
        cdna = normalize_variant_value(row.get("cDNA_Change", ""))
        protein = normalize_variant_value(row.get("Protein_Change", ""))
        identity = cdna or protein
        if identity:
            identities.add(identity)
    return len(identities)


def collect_clinvar_variant_records(gene: str) -> dict[str, list[dict[str, str]]]:
    records: dict[str, list[dict[str, str]]] = {}
    for row in read_csv_rows(BASE_DIR / f"clinvar_{gene.lower()}_missense_all.csv"):
        cdna = canonical_cdna(row.get("Variant_Name", ""))
        protein = canonical_protein(row.get("Protein_Change", ""))
        identifier, basis = variant_identifier(cdna, protein)
        if not identifier:
            continue
        enriched = dict(row)
        enriched["_cdna"] = cdna
        enriched["_protein"] = protein
        enriched["_basis"] = basis
        records.setdefault(identifier, []).append(enriched)
    return records


def collect_ai_variant_records(gene: str) -> dict[str, list[dict[str, str]]]:
    records: dict[str, list[dict[str, str]]] = {}
    for row in read_csv_rows(DATABASE_DIR / gene.lower() / "ALL_VARIANTS_MASTER.csv"):
        cdna = canonical_cdna(row.get("cDNA_Change", ""))
        protein = canonical_protein(row.get("Protein_Change", ""))
        identifier, basis = variant_identifier(cdna, protein)
        if not identifier:
            continue
        enriched = dict(row)
        enriched["_cdna"] = cdna
        enriched["_protein"] = protein
        enriched["_basis"] = basis
        records.setdefault(identifier, []).append(enriched)
    return records


def get_clinvar_files() -> list[Path]:
    return [
        BASE_DIR / f"clinvar_{gene.lower()}_missense_all.csv"
        for gene in GENES
        if (BASE_DIR / f"clinvar_{gene.lower()}_missense_all.csv").exists()
    ]


def print_total_clinvar_genes() -> None:
    """Print the total number and names of genes collected from ClinVar."""
    files = get_clinvar_files()
    genes = [path.name.removeprefix("clinvar_").removesuffix("_missense_all.csv").upper() for path in files]
    print("\nCLINVAR GENE COLLECTION")
    print(f"Total genes extracted from ClinVar: {len(genes)}")
    print(f"Genes: {', '.join(genes)}")


def build_clinvar_gene_table() -> list[dict[str, object]]:
    table = []
    for gene in GENES:
        path = BASE_DIR / f"clinvar_{gene.lower()}_missense_all.csv"
        rows = read_csv_rows(path)
        table.append(
            {
                "Gene": gene,
                "ClinVar_Variant_Records": len(rows),
                "ClinVar_Linked_Unique_PMIDs": count_unique_pmids_in_clinvar(rows),
                "Stored_Columns": len(read_csv_columns(path)),
            }
        )
    return table


def count_unique_pmids_in_clinvar(rows: Iterable[dict[str, str]]) -> int:
    pmids: set[str] = set()
    for row in rows:
        for pmid in str(row.get("Citations_PubMed", "")).split(";"):
            pmid = pmid.strip()
            if pmid.isdigit():
                pmids.add(pmid)
    return len(pmids)


def build_clinvar_fields_table() -> list[dict[str, object]]:
    files = get_clinvar_files()
    all_rows = [row for path in files for row in read_csv_rows(path)]
    columns = read_csv_columns(files[0]) if files else []
    return [
        {
            "Column_Number": index,
            "Stored_Column": column,
            "Description": CLINVAR_FIELD_DESCRIPTIONS.get(column, ""),
            "Nonempty_Records": count_nonempty(all_rows, column),
            "Total_Records": len(all_rows),
        }
        for index, column in enumerate(columns, start=1)
    ]


def build_article_sources_table() -> list[dict[str, object]]:
    table = []
    for gene in GENES:
        clinvar_articles = read_csv_rows(BASE_DIR / f"clinvar_{gene.lower()}_pmids.csv")
        telomerase_articles = read_csv_rows(BASE_DIR / f"telomerase_db_{gene.lower()}_papers.csv")
        merged_articles = read_csv_rows(BASE_DIR / f"{gene.lower()}_all_papers_merged.csv")
        table.append(
            {
                "Gene": gene,
                "ClinVar_Articles": len(clinvar_articles),
                "TelomeraseDB_Articles": len(telomerase_articles),
                "Merged_Unique_Article_Records": len(merged_articles),
                "Article_Records_With_Identified_DOI": count_unique_metadata_dois(gene),
            }
        )
    return table


def build_processing_table() -> list[dict[str, object]]:
    table = []
    for gene in GENES:
        merged_count = len(read_csv_rows(BASE_DIR / f"{gene.lower()}_all_papers_merged.csv"))
        folders = paper_folders(gene)
        pdf_count = sum(1 for folder in folders if folder_has_pdf(folder))
        extraction_rows = read_csv_rows(DATABASE_DIR / gene.lower() / "extraction_summary.csv")
        text_done = sum(1 for row in extraction_rows if row.get("status", "").strip().lower() == "done")
        ai_processed = count_files(gene, "variants.csv")
        ai_positive = count_files(gene, "gemini_response.json")
        ai_errors = count_files(gene, "gemini_error.txt")
        table.append(
            {
                "Gene": gene,
                "Merged_Article_Records": merged_count,
                "Articles_With_PDF": pdf_count,
                "Articles_Missing_PDF": max(merged_count - pdf_count, 0),
                "Articles_With_Text_Extracted": text_done,
                "Articles_With_AI_Response": ai_processed,
                "AI_Responses_With_At_Least_One_Variant": ai_positive,
                "AI_Error_Files": ai_errors,
            }
        )
    return table


def build_ai_variant_table() -> list[dict[str, object]]:
    table = []
    for gene in GENES:
        rows = read_csv_rows(DATABASE_DIR / gene.lower() / "ALL_VARIANTS_MASTER.csv")
        normalized_records = collect_ai_variant_records(gene)
        table.append(
            {
                "Gene": gene,
                "AI_Extracted_Variant_Rows": len(rows),
                "AI_Extracted_Unique_Variants": len(normalized_records),
                "AI_Output_Fields": len(AI_FIELD_DESCRIPTIONS),
                "Articles_Represented_By_Variant_Rows": count_unique_nonempty(rows, "PMID"),
            }
        )
    return table


def build_ai_fields_table() -> list[dict[str, object]]:
    all_rows = [
        row
        for gene in GENES
        for row in read_csv_rows(DATABASE_DIR / gene.lower() / "ALL_VARIANTS_MASTER.csv")
    ]
    return [
        {
            "Field_Number": index,
            "Requested_AI_Field": field,
            "Description": description,
            "Nonempty_Extracted_Rows": count_nonempty(all_rows, field),
            "Total_AI_Variant_Rows": len(all_rows),
        }
        for index, (field, description) in enumerate(AI_FIELD_DESCRIPTIONS.items(), start=1)
    ]


def build_variant_comparison_table() -> list[dict[str, object]]:
    """Load the normalized ClinVar-versus-paper comparison produced by step 08."""
    summary_path = DATABASE_DIR / "verification" / "variant_overlap_summary.csv"
    rows = read_csv_rows(summary_path)
    if not rows:
        return []

    return [
        {
            "Gene": row.get("Gene", ""),
            "ClinVar_Unique_Variants": row.get("ClinVar_Variants", "0"),
            "AI_Extracted_Unique_Variants": row.get("Paper_Variants", "0"),
            "Common_Variants": row.get("Common_Variants", "0"),
            "ClinVar_Only_Variants": row.get("ClinVar_Only", "0"),
            "AI_Extracted_Only_Variants": row.get("Papers_Only", "0"),
            "Precision": row.get("Precision", "0"),
            "ClinVar_Retrieval_Rate": row.get("Recall_RetrievalRate", "0"),
            "F1": row.get("F1", "0"),
            "Jaccard": row.get("Jaccard", "0"),
        }
        for row in rows
    ]


def build_master_variant_mapping_table() -> list[dict[str, object]]:
    """
    Build one GitHub-ready union table containing every unique ClinVar and AI variant.

    Common variants with conflicting normalized protein changes are explicitly
    flagged for review instead of being treated as confidently concordant.
    """
    table: list[dict[str, object]] = []
    for gene in GENES:
        clinvar = collect_clinvar_variant_records(gene)
        ai = collect_ai_variant_records(gene)

        for identifier in sorted(set(clinvar) | set(ai)):
            clinvar_rows = clinvar.get(identifier, [])
            ai_rows = ai.get(identifier, [])

            clinvar_cdna = join_values(row["_cdna"] for row in clinvar_rows)
            clinvar_protein = join_values(row["_protein"] for row in clinvar_rows)
            ai_cdna = join_values(row["_cdna"] for row in ai_rows)
            ai_protein = join_values(row["_protein"] for row in ai_rows)

            clinvar_protein_set = {row["_protein"] for row in clinvar_rows if row["_protein"]}
            ai_protein_set = {row["_protein"] for row in ai_rows if row["_protein"]}
            protein_overlap = clinvar_protein_set & ai_protein_set

            if clinvar_rows and ai_rows:
                if clinvar_protein_set and ai_protein_set and not protein_overlap:
                    status = "Common_Protein_Conflict_Review"
                    confidence = "Needs manual review"
                    protein_concordance = "Conflict"
                    notes = (
                        "Normalized identifier occurs in both sources, but normalized "
                        "protein changes disagree. Check transcript/accession and source evidence."
                    )
                elif protein_overlap:
                    status = "Common_Concordant"
                    confidence = "High"
                    protein_concordance = "Concordant"
                    notes = "Normalized identifier and at least one normalized protein change agree."
                else:
                    status = "Common_Partial_Evidence"
                    confidence = "Moderate"
                    protein_concordance = "Protein missing in one or both sources"
                    notes = "Normalized identifier agrees, but protein-level confirmation is unavailable."
            elif clinvar_rows:
                status = "ClinVar_Only"
                confidence = "Not applicable"
                protein_concordance = ""
                notes = "Variant was not found in the AI-extracted literature variant set."
            else:
                status = "AI_Only"
                confidence = "Requires validation"
                protein_concordance = ""
                notes = (
                    "Variant was extracted from literature but was not found in the "
                    "missense-only ClinVar reference set."
                )

            basis = (
                clinvar_rows[0]["_basis"]
                if clinvar_rows
                else ai_rows[0]["_basis"]
            )
            table.append(
                {
                    "Gene": gene,
                    "Normalized_Variant_ID": identifier,
                    "Mapping_Status": status,
                    "Is_Common": "Yes" if clinvar_rows and ai_rows else "No",
                    "Is_Strict_Common": "Yes" if status == "Common_Concordant" else "No",
                    "Match_Basis": basis,
                    "Mapping_Confidence": confidence,
                    "Protein_Concordance": protein_concordance,
                    "ClinVar_Normalized_cDNA": clinvar_cdna,
                    "ClinVar_Normalized_Protein": clinvar_protein,
                    "ClinVar_Transcript_Accessions": join_values(
                        accession
                        for row in clinvar_rows
                        for accession in re.findall(
                            r"\b[A-Z]{2}_\d+(?:\.\d+)?\b",
                            row.get("Variant_Name", ""),
                        )
                    ),
                    "ClinVar_Raw_Variant_Names": join_values(
                        row.get("Variant_Name", "") for row in clinvar_rows
                    ),
                    "ClinVar_Raw_Protein_Changes": join_values(
                        row.get("Protein_Change", "") for row in clinvar_rows
                    ),
                    "ClinVar_Variation_IDs": join_values(
                        row.get("Variation_ID", "") for row in clinvar_rows
                    ),
                    "ClinVar_Record_Count": len(clinvar_rows),
                    "AI_Normalized_cDNA": ai_cdna,
                    "AI_Normalized_Protein": ai_protein,
                    "AI_Raw_cDNA_Changes": join_values(
                        row.get("cDNA_Change", "") for row in ai_rows
                    ),
                    "AI_Raw_Protein_Changes": join_values(
                        row.get("Protein_Change", "") for row in ai_rows
                    ),
                    "AI_Source_Article_IDs": join_values(
                        row.get("PMID", "") for row in ai_rows
                    ),
                    "AI_Record_Count": len(ai_rows),
                    "Mapping_Notes": notes,
                }
            )
    return table


def build_master_mapping_status_summary(
    master_rows: list[dict[str, object]],
) -> list[dict[str, object]]:
    counts: dict[str, int] = {}
    for row in master_rows:
        status = str(row.get("Mapping_Status", ""))
        counts[status] = counts.get(status, 0) + 1
    return [
        {"Mapping_Status": status, "Unique_Variants": count}
        for status, count in sorted(counts.items())
    ]


def build_gene_level_mapping_summary(
    master_rows: list[dict[str, object]],
) -> list[dict[str, object]]:
    """Summarize audited ClinVar-versus-AI variant mapping for each gene."""
    table: list[dict[str, object]] = []
    for gene in GENES:
        rows = [row for row in master_rows if row.get("Gene") == gene]

        clinvar_count = sum(1 for row in rows if int(row.get("ClinVar_Record_Count", 0) or 0) > 0)
        ai_count = sum(1 for row in rows if int(row.get("AI_Record_Count", 0) or 0) > 0)
        provisional_common = sum(1 for row in rows if row.get("Is_Common") == "Yes")
        strict_common = sum(1 for row in rows if row.get("Is_Strict_Common") == "Yes")
        partial = sum(1 for row in rows if row.get("Mapping_Status") == "Common_Partial_Evidence")
        conflicts = sum(
            1 for row in rows if row.get("Mapping_Status") == "Common_Protein_Conflict_Review"
        )
        clinvar_only = sum(1 for row in rows if row.get("Mapping_Status") == "ClinVar_Only")
        ai_only = sum(1 for row in rows if row.get("Mapping_Status") == "AI_Only")

        strict_precision = strict_common / ai_count if ai_count else 0.0
        strict_retrieval = strict_common / clinvar_count if clinvar_count else 0.0
        strict_f1 = (
            2 * strict_precision * strict_retrieval / (strict_precision + strict_retrieval)
            if strict_precision + strict_retrieval
            else 0.0
        )

        table.append(
            {
                "Gene": gene,
                "ClinVar_Unique_Variants": clinvar_count,
                "AI_Unique_Variants": ai_count,
                "Provisional_Common": provisional_common,
                "Strict_Common": strict_common,
                "Common_Partial_Evidence": partial,
                "Common_Protein_Conflicts": conflicts,
                "ClinVar_Only": clinvar_only,
                "AI_Only": ai_only,
                "Strict_Precision": round(strict_precision, 4),
                "Strict_ClinVar_Retrieval_Rate": round(strict_retrieval, 4),
                "Strict_F1": round(strict_f1, 4),
            }
        )

    total_clinvar = sum(int(row["ClinVar_Unique_Variants"]) for row in table)
    total_ai = sum(int(row["AI_Unique_Variants"]) for row in table)
    total_strict = sum(int(row["Strict_Common"]) for row in table)
    total_precision = total_strict / total_ai if total_ai else 0.0
    total_retrieval = total_strict / total_clinvar if total_clinvar else 0.0
    total_f1 = (
        2 * total_precision * total_retrieval / (total_precision + total_retrieval)
        if total_precision + total_retrieval
        else 0.0
    )
    table.append(
        {
            "Gene": "TOTAL",
            "ClinVar_Unique_Variants": total_clinvar,
            "AI_Unique_Variants": total_ai,
            "Provisional_Common": sum(int(row["Provisional_Common"]) for row in table),
            "Strict_Common": total_strict,
            "Common_Partial_Evidence": sum(int(row["Common_Partial_Evidence"]) for row in table),
            "Common_Protein_Conflicts": sum(int(row["Common_Protein_Conflicts"]) for row in table),
            "ClinVar_Only": sum(int(row["ClinVar_Only"]) for row in table),
            "AI_Only": sum(int(row["AI_Only"]) for row in table),
            "Strict_Precision": round(total_precision, 4),
            "Strict_ClinVar_Retrieval_Rate": round(total_retrieval, 4),
            "Strict_F1": round(total_f1, 4),
        }
    )
    return table


def print_variant_comparison_method() -> None:
    """Explain exactly how variants were normalized and classified as common."""
    print(f"\n{'=' * 100}")
    print("HOW COMMON VARIANTS WERE IDENTIFIED")
    print("=" * 100)
    print(
        "1. ClinVar cDNA changes were parsed from Variant_Name. AI-extracted cDNA changes "
        "were read from cDNA_Change."
    )
    print(
        "2. cDNA values were normalized by removing spaces, underscores, semicolons, "
        "commas, and unsupported characters, then converting them to lowercase."
    )
    print(
        "3. Protein changes were normalized by removing spaces and parentheses, converting "
        "three-letter amino-acid names to one-letter codes, and standardizing stop codons."
    )
    print(
        "4. The normalized cDNA change was used as the primary variant identifier. "
        "The normalized protein change was used only when no cDNA change was available."
    )
    print(
        "5. A variant was classified as COMMON when its normalized identifier occurred in "
        "both the ClinVar set and the AI-extracted paper set for the same gene."
    )
    print(
        "6. ClinVar-only variants occurred only in ClinVar. AI-extracted-only variants "
        "occurred only in the extracted literature set."
    )
    print(
        "Important: ClinVar collection was restricted to missense variants, whereas the AI "
        "extracted all reported variant types. AI-extracted-only variants are therefore not "
        "automatically false positives."
    )
    print(
        "For the master mapping table, common variants with conflicting protein changes are "
        "labeled Common_Protein_Conflict_Review and should not be treated as confirmed matches."
    )


def add_total_row(rows: list[dict[str, object]], sum_columns: list[str]) -> list[dict[str, object]]:
    if not rows:
        return rows
    total = {field: "" for field in rows[0]}
    total[next(iter(total))] = "TOTAL"
    for column in sum_columns:
        total[column] = sum(int(row.get(column, 0) or 0) for row in rows)
    return [*rows, total]


def print_table(title: str, rows: list[dict[str, object]]) -> None:
    print(f"\n{'=' * 100}\n{title}\n{'=' * 100}")
    if not rows:
        print("No data available.")
        return

    columns = list(rows[0])
    widths = {
        column: min(
            max(len(column), *(len(str(row.get(column, ""))) for row in rows)),
            45,
        )
        for column in columns
    }

    def render(row: dict[str, object]) -> str:
        return " | ".join(
            str(row.get(column, ""))[: widths[column]].ljust(widths[column])
            for column in columns
        )

    print(render({column: column for column in columns}))
    print("-+-".join("-" * widths[column] for column in columns))
    for row in rows:
        print(render(row))


def write_csv_table(filename: str, rows: list[dict[str, object]]) -> None:
    if not rows:
        return
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    path = OUTPUT_DIR / filename
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)


def write_json_summary(tables: dict[str, list[dict[str, object]]]) -> None:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    with (OUTPUT_DIR / "pipeline_statistics.json").open("w", encoding="utf-8") as handle:
        json.dump(tables, handle, indent=2)


def main() -> None:
    print_total_clinvar_genes()

    clinvar_gene_table = add_total_row(
        build_clinvar_gene_table(),
        ["ClinVar_Variant_Records", "ClinVar_Linked_Unique_PMIDs"],
    )
    clinvar_fields_table = build_clinvar_fields_table()
    article_sources_table = add_total_row(
        build_article_sources_table(),
        [
            "ClinVar_Articles",
            "TelomeraseDB_Articles",
            "Merged_Unique_Article_Records",
            "Article_Records_With_Identified_DOI",
        ],
    )
    processing_table = add_total_row(
        build_processing_table(),
        [
            "Merged_Article_Records",
            "Articles_With_PDF",
            "Articles_Missing_PDF",
            "Articles_With_Text_Extracted",
            "Articles_With_AI_Response",
            "AI_Responses_With_At_Least_One_Variant",
            "AI_Error_Files",
        ],
    )
    ai_variant_table = add_total_row(
        build_ai_variant_table(),
        [
            "AI_Extracted_Variant_Rows",
            "AI_Extracted_Unique_Variants",
            "Articles_Represented_By_Variant_Rows",
        ],
    )
    ai_fields_table = build_ai_fields_table()
    variant_comparison_table = build_variant_comparison_table()
    master_variant_mapping_table = build_master_variant_mapping_table()
    master_mapping_status_summary = build_master_mapping_status_summary(
        master_variant_mapping_table
    )
    gene_level_mapping_summary = build_gene_level_mapping_summary(
        master_variant_mapping_table
    )

    tables = {
        "table_01_clinvar_gene_summary": clinvar_gene_table,
        "table_02_clinvar_stored_fields": clinvar_fields_table,
        "table_03_article_sources_by_gene": article_sources_table,
        "table_04_article_processing_by_gene": processing_table,
        "table_05_ai_variants_by_gene": ai_variant_table,
        "table_06_variant_comparison_by_gene": variant_comparison_table,
        "table_07_ai_requested_fields": ai_fields_table,
        "table_08_master_mapping_status_summary": master_mapping_status_summary,
        "table_09_gene_level_mapping_summary": gene_level_mapping_summary,
    }

    titles = {
        "table_01_clinvar_gene_summary": "TABLE 1. CLINVAR RECORDS COLLECTED BY GENE",
        "table_02_clinvar_stored_fields": "TABLE 2. COLUMNS RETRIEVED AND STORED FROM CLINVAR",
        "table_03_article_sources_by_gene": "TABLE 3. ARTICLES IDENTIFIED FROM CLINVAR AND THE TELOMERASE DATABASE",
        "table_04_article_processing_by_gene": "TABLE 4. ARTICLE DOWNLOAD, TEXT EXTRACTION, AND AI PROCESSING",
        "table_05_ai_variants_by_gene": "TABLE 5. VARIANTS EXTRACTED BY THE AI PIPELINE",
        "table_06_variant_comparison_by_gene": "TABLE 6. CLINVAR AND AI-EXTRACTED VARIANT COMPARISON",
        "table_07_ai_requested_fields": "TABLE 7. STRUCTURED FIELDS REQUESTED FROM THE AI MODEL",
        "table_08_master_mapping_status_summary": "TABLE 8. MASTER VARIANT MAPPING STATUS SUMMARY",
        "table_09_gene_level_mapping_summary": "TABLE 9. AUDITED GENE-LEVEL CLINVAR AND AI VARIANT MAPPING",
    }

    for name, rows in tables.items():
        print_table(titles[name], rows)
        write_csv_table(f"{name}.csv", rows)

    write_csv_table(
        "table_01_all_clinvar_ai_common_variants.csv",
        master_variant_mapping_table,
    )
    write_csv_table(
        "table_02_common_variants_mapping_audit.csv",
        [row for row in master_variant_mapping_table if row.get("Is_Common") == "Yes"],
    )
    print_variant_comparison_method()
    write_json_summary(tables)
    print(
        "GitHub-ready master mapping table saved to: "
        f"{OUTPUT_DIR / 'table_01_all_clinvar_ai_common_variants.csv'}"
    )
    print(f"\nStatistics tables saved to: {OUTPUT_DIR}")


if __name__ == "__main__":
    main()