Author Strategy

#!/usr/bin/env python3
"""
Analyze publication patterns from a PubMed CSV and generate visualizations + strategy report.

Usage:
    python analyze_patterns.py /path/to/publications.csv [--output-dir /path/to/report/]
"""

import argparse
from collections import Counter
from pathlib import Path

import matplotlib
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns

matplotlib.rcParams["font.family"] = "Apple SD Gothic Neo"
matplotlib.rcParams["axes.unicode_minus"] = False

COLORS = {
    "GBD": "#e74c3c",
    "NHIS/Claims": "#3498db",
    "SR/MA": "#2ecc71",
    "Cross-national": "#9b59b6",
    "AI/ML": "#f39c12",
    "National survey": "#1abc9c",
    "Biobank": "#e67e22",
    "Clinical trial": "#34495e",
    "Letter/Commentary": "#95a5a6",
    "Case report": "#bdc3c7",
    "Other": "#7f8c8d",
}


def load_data(csv_path: str) -> pd.DataFrame:
    df = pd.read_csv(csv_path)
    df["year"] = pd.to_numeric(df["year"], errors="coerce")
    return df


def plot_yearly_stacked(df: pd.DataFrame, report_dir: Path, author_name: str):
    fig, ax = plt.subplots(figsize=(14, 7))
    min_year = max(int(df["year"].min()), df["year"].max() - 8) if len(df) > 50 else int(df["year"].min())
    df_yr = df[df["year"] >= min_year].copy()
    pivot = df_yr.groupby(["year", "study_type"]).size().unstack(fill_value=0)
    col_order = pivot.sum().sort_values(ascending=False).index
    pivot = pivot[col_order]
    colors = [COLORS.get(c, "#cccccc") for c in pivot.columns]
    pivot.plot(kind="bar", stacked=True, ax=ax, color=colors, width=0.8)
    ax.set_xlabel("Year", fontsize=12)
    ax.set_ylabel("Publications", fontsize=12)
    ax.set_title(f"{author_name}: Yearly Publication Count by Study Type (N={len(df)})", fontsize=14, fontweight="bold")
    ax.legend(title="Study Type", bbox_to_anchor=(1.02, 1), loc="upper left", fontsize=9)
    plt.tight_layout()
    fig.savefig(report_dir / "01_yearly_stacked.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_study_type_pie(df: pd.DataFrame, report_dir: Path):
    fig, ax = plt.subplots(figsize=(10, 8))
    counts = df["study_type"].value_counts()
    colors = [COLORS.get(c, "#cccccc") for c in counts.index]
    ax.pie(counts, labels=counts.index, autopct="%1.1f%%", colors=colors, pctdistance=0.85, startangle=90)
    ax.set_title(f"Study Type Distribution (N={len(df)})", fontsize=14, fontweight="bold")
    plt.tight_layout()
    fig.savefig(report_dir / "02_study_type_pie.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_author_position(df: pd.DataFrame, report_dir: Path):
    fig, axes = plt.subplots(1, 2, figsize=(14, 6))
    # Positional heuristic only (not leadership metadata): first / middle / last / unknown.
    pos_order = ["first", "middle", "last", "unknown"]
    pos_counts = df["author_position"].value_counts().reindex(pos_order, fill_value=0)
    colors_pos = ["#e74c3c", "#95a5a6", "#2ecc71", "#bdc3c7"]
    bars = axes[0].bar(pos_counts.index, pos_counts.values, color=colors_pos)
    axes[0].set_title("Author Position Overall", fontsize=12, fontweight="bold")
    axes[0].set_ylabel("Count")
    for bar, val in zip(bars, pos_counts.values):
        axes[0].text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 2,
                     f"{val}\n({val / len(df) * 100:.1f}%)", ha="center", fontsize=9)

    key_types = [t for t in ["GBD", "NHIS/Claims", "SR/MA", "Cross-national", "National survey"]
                 if t in df["study_type"].values]
    if key_types:
        pos_by_type = df[df["study_type"].isin(key_types)].groupby(
            ["study_type", "author_position"]).size().unstack(fill_value=0)
        pos_by_type = pos_by_type.reindex(columns=pos_order, fill_value=0)
        pos_by_type.plot(kind="bar", ax=axes[1], color=colors_pos, width=0.8)
        axes[1].set_title("Author Position by Study Type", fontsize=12, fontweight="bold")
        axes[1].set_ylabel("Count")
        axes[1].legend(title="Position", fontsize=8)
        axes[1].tick_params(axis="x", rotation=0)

    plt.tight_layout()
    fig.savefig(report_dir / "03_author_position.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_journal_tier_heatmap(df: pd.DataFrame, report_dir: Path):
    fig, ax = plt.subplots(figsize=(12, 8))
    pivot = df.groupby(["study_type", "journal_tier"]).size().unstack(fill_value=0)
    tier_order = ["Lancet family", "Nature family", "NEJM/BMJ/JAMA", "IF>=10", "Other"]
    pivot = pivot.reindex(columns=[c for c in tier_order if c in pivot.columns], fill_value=0)
    pivot = pivot.loc[pivot.sum(axis=1).sort_values(ascending=False).index]
    sns.heatmap(pivot, annot=True, fmt="d", cmap="YlOrRd", ax=ax, linewidths=0.5)
    ax.set_title("Study Type x Journal Tier (count)", fontsize=14, fontweight="bold")
    plt.tight_layout()
    fig.savefig(report_dir / "04_journal_tier_heatmap.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_topic_distribution(df: pd.DataFrame, report_dir: Path):
    fig, ax = plt.subplots(figsize=(10, 7))
    topic_counts = df["topic"].value_counts()
    colors_topic = sns.color_palette("husl", len(topic_counts))
    bars = ax.barh(topic_counts.index[::-1], topic_counts.values[::-1], color=colors_topic[::-1])
    ax.set_xlabel("Publications")
    ax.set_title(f"Topic Distribution (N={len(df)})", fontsize=14, fontweight="bold")
    for bar, val in zip(bars, topic_counts.values[::-1]):
        ax.text(bar.get_width() + 1, bar.get_y() + bar.get_height() / 2,
                f"{val} ({val / len(df) * 100:.1f}%)", va="center", fontsize=9)
    plt.tight_layout()
    fig.savefig(report_dir / "05_topic_distribution.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_growth_curve(df: pd.DataFrame, report_dir: Path):
    fig, ax = plt.subplots(figsize=(12, 6))
    min_year = max(int(df["year"].min()), df["year"].max() - 10)
    df_yr = df[df["year"] >= min_year].copy()
    yearly = df_yr.groupby("year").size().sort_index()
    cumulative = yearly.cumsum()
    ax.plot(cumulative.index, cumulative.values, "o-", color="#e74c3c", linewidth=2, markersize=8)
    ax2 = ax.twinx()
    ax2.bar(yearly.index, yearly.values, alpha=0.3, color="#3498db")
    ax.set_xlabel("Year")
    ax.set_ylabel("Cumulative", color="#e74c3c")
    ax2.set_ylabel("Yearly", color="#3498db")
    ax.set_title("Publication Growth Curve", fontsize=14, fontweight="bold")
    for y in cumulative.index:
        ax.annotate(f"{cumulative[y]}", (y, cumulative[y]),
                    textcoords="offset points", xytext=(0, 10), ha="center", fontsize=9)
    plt.tight_layout()
    fig.savefig(report_dir / "06_growth_curve.png", dpi=150, bbox_inches="tight")
    plt.close()


def plot_strategy_roi(df: pd.DataFrame, report_dir: Path):
    fig, ax = plt.subplots(figsize=(12, 8))
    data = []
    for st in df["study_type"].value_counts().index:
        subset = df[df["study_type"] == st]
        n = len(subset)
        high_tier = subset["journal_tier"].isin(
            ["Lancet family", "Nature family", "NEJM/BMJ/JAMA", "IF>=10"]).mean() * 100
        first_last = subset["author_position"].isin(["first", "last"]).mean() * 100
        data.append({"type": st, "count": n, "high_tier_pct": high_tier, "first_last_pct": first_last})
    plot_df = pd.DataFrame(data)
    ax.scatter(
        plot_df["high_tier_pct"], plot_df["first_last_pct"],
        s=plot_df["count"] * 3,
        c=[COLORS.get(t, "#cccccc") for t in plot_df["type"]],
        alpha=0.7, edgecolors="black", linewidths=0.5
    )
    for _, row in plot_df.iterrows():
        ax.annotate(f'{row["type"]}\n(n={row["count"]})',
                    (row["high_tier_pct"], row["first_last_pct"]),
                    fontsize=8, ha="center", va="bottom")
    ax.set_xlabel("% High-Tier Journal (IF>=10)", fontsize=12)
    ax.set_ylabel("% First or Last Author (positional)", fontsize=12)
    ax.set_title("Strategy ROI: Journal Quality vs Author Position vs Volume", fontsize=14, fontweight="bold")
    ax.axhline(y=50, color="gray", linestyle="--", alpha=0.3)
    ax.axvline(x=20, color="gray", linestyle="--", alpha=0.3)
    plt.tight_layout()
    fig.savefig(report_dir / "07_strategy_roi.png", dpi=150, bbox_inches="tight")
    plt.close()


def generate_report(df: pd.DataFrame, report_dir: Path, author_name: str):
    total = len(df)
    types = df["study_type"].value_counts()
    positions = df["author_position"].value_counts()
    high_tier = len(df[df["journal_tier"].isin(
        ["Lancet family", "Nature family", "NEJM/BMJ/JAMA", "IF>=10"])])
    first_last = len(df[df["author_position"].isin(["first", "last"])])

    # Top 3 study types
    top_types = types.head(3)
    type_rows = "\n".join(
        f"| {st} | {n} | {n / total * 100:.1f}% |"
        for st, n in top_types.items()
    )

    # Top 3 topics
    top_topics = df["topic"].value_counts().head(5)
    topic_rows = "\n".join(
        f"| {tp} | {n} | {n / total * 100:.1f}% |"
        for tp, n in top_topics.items()
    )

    # Year range
    years = df["year"].dropna().astype(int)
    year_range = f"{years.min()}-{years.max()}"
    recent_year = years.max()
    recent_count = len(df[df["year"] == recent_year])

    report = f"""# {author_name} — Publication Strategy Analysis

## Summary

| Metric | Value |
|--------|-------|
| Total PubMed publications | {total} |
| Year range | {year_range} |
| {recent_year} publications | {recent_count} |
| High-tier journals (Lancet/Nature/NEJM/BMJ/JAMA/IF>=10) | {high_tier} ({high_tier / total * 100:.1f}%) |
| First or last author (positional heuristic) | {first_last} ({first_last / total * 100:.1f}%) |

## Study Type Breakdown

| Type | Count | % |
|------|-------|---|
{type_rows}
| Other | {total - top_types.sum()} | {(total - top_types.sum()) / total * 100:.1f}% |

## Topic Clusters (Top 5)

| Topic | Count | % |
|-------|-------|---|
{topic_rows}

## Author Position (positional heuristic — not leadership metadata)

| Position | Count | % |
|----------|-------|---|
| First author | {positions.get("first", 0)} | {positions.get("first", 0) / total * 100:.1f}% |
| Last author | {positions.get("last", 0)} | {positions.get("last", 0) / total * 100:.1f}% |
| Middle | {positions.get("middle", 0)} | {positions.get("middle", 0) / total * 100:.1f}% |
| Unknown | {positions.get("unknown", 0)} | {positions.get("unknown", 0) / total * 100:.1f}% |

## Key Observations

1. **Primary strategy**: {types.index[0]} ({types.iloc[0]} papers, {types.iloc[0] / total * 100:.1f}%)
2. **Secondary strategy**: {types.index[1] if len(types) > 1 else "N/A"} ({types.iloc[1] if len(types) > 1 else 0} papers)
3. **High-tier placement rate**: {high_tier / total * 100:.1f}%
4. **First/last positional rate** (positional heuristic, not leadership): {first_last / total * 100:.1f}%

## Visualizations

- `01_yearly_stacked.png` — yearly publication count by study type
- `02_study_type_pie.png` — study type distribution
- `03_author_position.png` — author position overall and by study type
- `04_journal_tier_heatmap.png` — study type x journal tier
- `05_topic_distribution.png` — topic clusters
- `06_growth_curve.png` — cumulative publication growth
- `07_strategy_roi.png` — journal quality vs author position vs volume

---
Generated: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}
Data source: PubMed, {total} records
"""

    with open(report_dir / "analysis_report.md", "w") as f:
        f.write(report)
    print(f"Saved: analysis_report.md")


def main():
    parser = argparse.ArgumentParser(description="Analyze author publication patterns")
    parser.add_argument("csv_path", help="Path to publications CSV")
    parser.add_argument("--output-dir", "-o", help="Output directory for report", default=None)
    parser.add_argument("--author-name", help="Author name for report title", default="Author")
    args = parser.parse_args()

    csv_path = Path(args.csv_path)
    report_dir = Path(args.output_dir) if args.output_dir else csv_path.parent / "report"
    report_dir.mkdir(parents=True, exist_ok=True)

    df = load_data(str(csv_path))
    print(f"Loaded {len(df)} records\n")

    plot_yearly_stacked(df, report_dir, args.author_name)
    print("Saved: 01_yearly_stacked.png")
    plot_study_type_pie(df, report_dir)
    print("Saved: 02_study_type_pie.png")
    plot_author_position(df, report_dir)
    print("Saved: 03_author_position.png")
    plot_journal_tier_heatmap(df, report_dir)
    print("Saved: 04_journal_tier_heatmap.png")
    plot_topic_distribution(df, report_dir)
    print("Saved: 05_topic_distribution.png")
    plot_growth_curve(df, report_dir)
    print("Saved: 06_growth_curve.png")
    plot_strategy_roi(df, report_dir)
    print("Saved: 07_strategy_roi.png")
    generate_report(df, report_dir, args.author_name)

    print(f"\nAll outputs in: {report_dir}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Classify a queried author's publication trajectory into archetypes A1-A6 (+ composite).

EXPLAINABLE MULTI-LABEL HEURISTIC, NOT AN OBJECTIVE VERDICT. Each surfaced label carries
a score, a confidence band, and evidence drawn from the queried author's OWN PMIDs.

The rubric (signals, weights, thresholds, provenance) is the canonical YAML
references/trajectory_archetypes.yaml. The score formula, confidence bands, negative
rule, and evidence shapes are documented there and implemented here verbatim.

A corpus_manifest.json with review_status == "approved" whose hashes match the CSV is
REQUIRED — surname-alone resolution is forbidden, so an unreviewed corpus cannot be
classified (see SKILL.md Step 6/7).

Stdlib + PyYAML only (no pandas/Biopython).

Usage:
  python3 classify_archetypes.py publications.csv --manifest corpus_manifest.json \
      --rubric references/trajectory_archetypes.yaml -o report_dir/
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import sys
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover
    sys.stderr.write("ERROR: PyYAML is required (pip install pyyaml).\n")
    sys.exit(2)

HERE = Path(__file__).resolve().parent
DEFAULT_RUBRIC = HERE / "references" / "trajectory_archetypes.yaml"
COMPUTABLE = {"source-derived", "rule-derived"}
DISCLAIMER = (
    "These archetype labels are EXPLAINABLE HEURISTICS, NOT objective classifications. "
    "They are computed from PubMed metadata and title/abstract term matches only. Author "
    "position is a positional heuristic (first/middle/last/unknown), not authoritative "
    "leadership or corresponding-author metadata. h-index, citation counts, and "
    "venue-impact tiers are unavailable in this PubMed-only analysis and are never inferred."
)
_CONF_ORDER = {"low": 1, "med": 2, "high": 3}


# ---------------------------------------------------------------------------
# Manifest gate (Fix 1 + 1b): manifest must be approved AND bound to this CSV.
# ---------------------------------------------------------------------------


def _csv_pmids(rows: list[dict]) -> list[str]:
    return [str(r.get("pmid", "")).strip() for r in rows if str(r.get("pmid", "")).strip()]


def _pmid_set_hash(pmids) -> str:
    joined = "\n".join(sorted(set(str(p) for p in pmids)))
    return hashlib.sha256(joined.encode("utf-8")).hexdigest()


def require_approved_manifest(manifest_path: Path, csv_path: Path, rows: list[dict]) -> dict:
    """Halt unless the manifest is approved and its hashes match the CSV."""
    if not manifest_path.exists():
        raise SystemExit(
            f"GATE: no corpus_manifest.json at {manifest_path}. Run fetch_pubmed.py and "
            "approve the corpus first (surname-alone classification is forbidden)."
        )
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))

    if manifest.get("review_status") != "approved":
        raise SystemExit(
            f"GATE: manifest review_status is '{manifest.get('review_status')}', not 'approved'. "
            "Review candidates.json and re-run fetch_pubmed.py with --approve."
        )

    actual_csv_sha = hashlib.sha256(csv_path.read_bytes()).hexdigest()
    if manifest.get("csv_sha256") != actual_csv_sha:
        raise SystemExit(
            "GATE: csv_sha256 mismatch — this manifest was not generated for this CSV. "
            "Re-run fetch_pubmed.py to regenerate a bound manifest."
        )

    actual_pmid_hash = _pmid_set_hash(_csv_pmids(rows))
    if manifest.get("pmid_set_hash") != actual_pmid_hash:
        raise SystemExit("GATE: pmid_set_hash mismatch — CSV PMID set differs from the manifest.")

    if manifest.get("record_count") != len(rows):
        raise SystemExit(
            f"GATE: record_count mismatch (manifest {manifest.get('record_count')} vs CSV {len(rows)})."
        )
    return manifest


# ---------------------------------------------------------------------------
# Signal metrics
# ---------------------------------------------------------------------------


def _text(r: dict, fields: str = "title") -> str:
    t = (r.get("title", "") or "").lower()
    if fields == "title_abstract":
        t += " " + (r.get("abstract", "") or "").lower()
    return t


def _is_ai(r: dict, terms: list[str]) -> bool:
    text = (r.get("title", "") or "").lower() + " " + (r.get("abstract", "") or "").lower()
    return r.get("study_type") == "AI/ML" or any(t in text for t in terms)


def _year(r: dict):
    y = str(r.get("year", "")).strip()
    return int(y) if y.isdigit() else None


def m_title_term_count(records, p):
    terms = [t.lower() for t in p["any_terms"]]
    fields = p.get("fields", "title")
    matched = [r["pmid"] for r in records if any(t in _text(r, fields) for t in terms)]
    return len(matched) >= p["min_matching_papers"], {"pmids": matched}


def m_study_type_fraction(records, p):
    types = set(p["types"])
    n = len(records)
    hit = [r["pmid"] for r in records if r.get("study_type") in types]
    frac = len(hit) / n if n else 0.0
    return frac >= p["fraction_min"], {
        "summary": {"value": round(frac, 3), "computed_from_n": n, "representative_pmids": hit[:5]}
    }


def m_distinct_topics_within_type(records, p):
    types = set(p["types"])
    subset = [r for r in records if r.get("study_type") in types]
    topics = {r.get("topic") for r in subset if r.get("topic")}
    return len(topics) >= p["min_distinct_topics"], {
        "summary": {"value": len(topics), "computed_from_n": len(subset),
                    "representative_pmids": [r["pmid"] for r in subset[:5]]}
    }


def m_ai_term_fraction(records, p):
    terms = [t.lower() for t in p["terms"]]
    n = len(records)
    hit = [r["pmid"] for r in records if _is_ai(r, terms)]
    frac = len(hit) / n if n else 0.0
    direction = p["direction"]
    if direction == "ge":
        fired = frac >= p["fraction"]
    elif direction == "le":
        fired = frac <= p["fraction"]
    elif direction == "between":
        fired = p["lo"] <= frac <= p["hi"]
    else:
        fired = False
    return fired, {"summary": {"value": round(frac, 3), "computed_from_n": n, "representative_pmids": hit[:5]}}


def m_author_count_papers(records, p):
    amin = p["author_count_min"]
    hit = [r["pmid"] for r in records if str(r.get("n_authors", "")).isdigit() and int(r["n_authors"]) >= amin]
    return len(hit) >= p["min_papers"], {"pmids": hit}


def m_author_count_share(records, p):
    amin = p["author_count_min"]
    n = len(records)
    hit = [r["pmid"] for r in records if str(r.get("n_authors", "")).isdigit() and int(r["n_authors"]) >= amin]
    share = len(hit) / n if n else 0.0
    return share >= p["share_min"], {
        "summary": {"value": round(share, 3), "computed_from_n": n, "representative_pmids": hit[:5]}
    }


def m_venue_drift_ai(records, p):
    terms = [t.lower() for t in p["terms"]]
    yrs = sorted(y for y in (_year(r) for r in records) if y is not None)
    if len(yrs) < 3:
        return False, {"pmids": []}
    ymin, ymax = yrs[0], yrs[-1]
    third = (ymax - ymin) / 3 if ymax > ymin else 0
    early_cut, late_cut = ymin + third, ymax - third
    early_ai, late_ai = 0, []
    for r in records:
        y = _year(r)
        if y is None or not _is_ai(r, terms):
            continue
        if y <= early_cut:
            early_ai += 1
        if y >= late_cut:
            late_ai.append(r["pmid"])
    return (early_ai == 0 and len(late_ai) >= p["min_late"]), {"pmids": late_ai}


def m_dual_genre_copresence(records, p):
    a_terms = [t.lower() for t in p["genre_a_terms"]]
    a_types = set(p.get("genre_a_study_types", []))
    b_terms = [t.lower() for t in p["genre_b_terms"]]
    a_hits, b_hits = [], []
    for r in records:
        text = (r.get("title", "") or "").lower() + " " + (r.get("abstract", "") or "").lower()
        if r.get("study_type") in a_types or any(t in text for t in a_terms):
            a_hits.append(r["pmid"])
        if any(t in text for t in b_terms):
            b_hits.append(r["pmid"])
    fired = len(a_hits) >= p["min_each"] and len(b_hits) >= p["min_each"]
    return fired, {"pmids": list(dict.fromkeys(a_hits[:3] + b_hits[:3]))}


def m_genre_pair_temporal(records, p):
    first_terms = [t.lower() for t in p["first_terms"]]
    second_terms = [t.lower() for t in p["second_terms"]]
    gap = p["max_year_gap"]
    firsts, seconds = [], []
    for r in records:
        y = _year(r)
        if y is None:
            continue
        title = (r.get("title", "") or "").lower()
        if any(t in title for t in first_terms):
            firsts.append((y, r["pmid"]))
        if any(t in title for t in second_terms):
            seconds.append((y, r["pmid"]))
    for yf, pf in firsts:
        for ys, ps in seconds:
            if 0 <= ys - yf <= gap and pf != ps:
                return True, {"pmids": [pf, ps]}
    return False, {"pmids": []}


def m_senior_on_ai_papers(records, p):
    terms = [t.lower() for t in p["terms"]]
    ai = [r for r in records if _is_ai(r, terms)]
    if not ai:
        return False, {"pmids": []}
    senior = [r["pmid"] for r in ai if r.get("author_position") == p["position"]]
    return (len(senior) / len(ai)) >= p["fraction_min"], {"pmids": senior}


def m_pure_ai_no_clinical_floor(records, p):
    terms = [t.lower() for t in p["terms"]]
    n = len(records)
    if n == 0:
        return False, {"pmids": []}
    ai = sum(1 for r in records if _is_ai(r, terms))
    return (ai / n) >= 0.999, {"pmids": []}


METRICS = {
    "title_term_count": (m_title_term_count, "per_paper"),
    "study_type_fraction": (m_study_type_fraction, "corpus_level"),
    "distinct_topics_within_type": (m_distinct_topics_within_type, "corpus_level"),
    "ai_term_fraction": (m_ai_term_fraction, "corpus_level"),
    "author_count_papers": (m_author_count_papers, "per_paper"),
    "author_count_share": (m_author_count_share, "corpus_level"),
    "venue_drift_ai": (m_venue_drift_ai, "per_paper"),
    "dual_genre_copresence": (m_dual_genre_copresence, "per_paper"),
    "genre_pair_temporal": (m_genre_pair_temporal, "per_paper"),
    "senior_on_ai_papers": (m_senior_on_ai_papers, "per_paper"),
    "pure_ai_no_clinical_floor": (m_pure_ai_no_clinical_floor, "per_paper"),
}


# ---------------------------------------------------------------------------
# Scoring (pure function)
# ---------------------------------------------------------------------------


def _present_columns(records: list[dict]) -> set:
    cols = set()
    for r in records:
        cols |= set(r.keys())
    return cols


def _cap_confidence(conf: str, ceiling: str) -> str:
    if conf is None:
        return None
    if _CONF_ORDER.get(conf, 0) > _CONF_ORDER.get(ceiling, 3):
        return ceiling
    return conf


def _score_one(key: str, arc: dict, records: list[dict], present: set) -> dict:
    n = len(records)
    denom = 0.0
    numer = 0.0
    fired_ids: list[str] = []
    evidence_pmids: list[str] = []
    evidence_summary: list[dict] = []
    verify_signals: list[dict] = []

    for sig in arc.get("signals", []):
        prov = sig.get("provenance")
        if prov == "unavailable":
            verify_signals.append({"id": sig["id"], "note": sig.get("verify_note", "")})
            continue
        if prov not in COMPUTABLE:
            continue
        metric_name = sig.get("metric")
        if metric_name not in METRICS:
            continue
        required = set(sig.get("required_columns", []))
        if not required.issubset(present):
            continue  # not computable for this dataset -> excluded from denominator
        fn, kind = METRICS[metric_name]
        weight = float(sig.get("weight", 0.0))
        denom += weight
        fired, ev = fn(records, sig.get("params", {}))
        if fired:
            numer += weight
            fired_ids.append(sig["id"])
            if kind == "per_paper":
                evidence_pmids.extend(ev.get("pmids", []))
            else:
                summary = dict(ev.get("summary", {}))
                summary["signal_id"] = sig["id"]
                evidence_summary.append(summary)

    # Negatives suppress the label regardless of score.
    negatives_fired = []
    for neg in arc.get("negatives", []) or []:
        if set(neg.get("required_columns", [])).issubset(present):
            fn, _ = METRICS.get(neg.get("metric"), (None, None))
            if fn is not None:
                fired, _ev = fn(records, neg.get("params", {}))
                if fired:
                    negatives_fired.append(neg["id"])

    score = round(numer / denom, 3) if denom > 0 else 0.0
    distinct_fired = len(fired_ids)
    min_sample = arc["min_sample"]
    threshold = arc["score_threshold"]

    suppressed = bool(negatives_fired)
    surfaced = (not suppressed) and (n >= min_sample) and (score >= threshold) and (distinct_fired >= 1)

    if distinct_fired >= 3 and n >= min_sample:
        confidence = "high"
    elif distinct_fired >= 2:
        confidence = "med"
    elif distinct_fired >= 1:
        confidence = "low"
    else:
        confidence = None
    confidence = _cap_confidence(confidence, arc.get("max_confidence_mvp", "high"))

    reason = None
    if not surfaced:
        if suppressed:
            reason = f"suppressed by negative rule(s): {', '.join(negatives_fired)}"
        elif n < min_sample:
            reason = f"sample n={n} below min_sample={min_sample}"
        elif distinct_fired == 0:
            reason = "no computable signal fired"
        else:
            reason = f"score {score} below threshold {threshold}"

    result = {
        "label": key,
        "name": arc["name"],
        "score": score,
        "confidence": confidence if surfaced else None,
        "surfaced": surfaced,
        "insufficient_evidence": not surfaced,
        "reason": reason,
        "fired_signals": fired_ids,
        "denominator": round(denom, 3),
        "evidence_pmids": list(dict.fromkeys(evidence_pmids)),
        "evidence_summary": evidence_summary,
        "verify_signals": verify_signals,
        "negatives_fired": negatives_fired,
        "max_confidence_mvp": arc.get("max_confidence_mvp", "high"),
    }
    if surfaced and arc.get("flag"):
        result["flag"] = arc["flag"]
    return result


def score_archetypes(records: list[dict], rubric: dict) -> dict:
    """Pure function: records (list of CSV row dicts) + rubric -> classification dict."""
    present = _present_columns(records)
    archetypes = {key: _score_one(key, arc, records, present)
                  for key, arc in rubric["archetypes"].items()}

    composites = {}
    for key, comp in (rubric.get("composites") or {}).items():
        requires = comp.get("requires_all", [])
        base_ok = all(archetypes.get(r, {}).get("surfaced") for r in requires)
        extra = comp.get("extra_condition", {})
        extra_fired, extra_ev = False, {"pmids": []}
        if base_ok and extra:
            if set(extra.get("required_columns", [])).issubset(present):
                fn, _ = METRICS.get(extra.get("metric"), (None, None))
                if fn is not None:
                    extra_fired, extra_ev = fn(records, extra.get("params", {}))
        surfaced = bool(base_ok and extra_fired)
        composites[key] = {
            "label": key,
            "name": comp["name"],
            "type": "composite",
            "requires_all": requires,
            "surfaced": surfaced,
            "evidence_pmids": extra_ev.get("pmids", []) if surfaced else [],
            "reason": None if surfaced else (
                f"requires all of {requires} to surface" if not base_ok else "extra condition not met"),
        }

    return {
        "disclaimer": DISCLAIMER,
        "sample_n": len(records),
        "archetypes": archetypes,
        "composites": composites,
    }


# ---------------------------------------------------------------------------
# Rubric + IO
# ---------------------------------------------------------------------------


def load_rubric(path: Path) -> tuple[dict, str]:
    raw = path.read_bytes()
    rubric = yaml.safe_load(raw.decode("utf-8"))
    return rubric, hashlib.sha256(raw).hexdigest()


def read_csv(path: Path) -> list[dict]:
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def _md_report(results: dict, meta: dict) -> str:
    out = ["# Trajectory-Archetype Classification", ""]
    out.append(f"> **{DISCLAIMER}**")
    out.append("")
    out.append("## Provenance and disambiguation")
    out.append("")
    out.append(f"- Rubric version: `{meta['rubric_version']}` (sha256 `{meta['rubric_sha256'][:12]}…`)")
    out.append(f"- Corpus: {results['sample_n']} records, manifest review_status `{meta['manifest_review_status']}`")
    basis = meta.get("disambiguation_basis", {})
    basis_str = ", ".join(f"{k}={v}" for k, v in basis.items() if v) or "surname only (review the candidate clusters)"
    out.append(f"- Disambiguation basis: {basis_str}")
    out.append("")
    out.append("## Surfaced archetypes (multi-label)")
    out.append("")
    surfaced = [a for a in results["archetypes"].values() if a["surfaced"]]
    if not surfaced:
        out.append("_No archetype surfaced above threshold — insufficient evidence for all (see below)._")
    for a in sorted(surfaced, key=lambda x: x["score"], reverse=True):
        out.append(f"### {a['label']} — {a['name']}")
        out.append("")
        out.append(f"- Score **{a['score']}** · confidence **{a['confidence']}** "
                   f"(max in MVP: {a['max_confidence_mvp']})")
        out.append(f"- Fired signals: {', '.join(a['fired_signals'])}")
        if a.get("flag"):
            out.append(f"- ⚠️ {a['flag']}")
        if a["evidence_pmids"]:
            out.append(f"- Evidence PMIDs: {', '.join(a['evidence_pmids'][:15])}")
        for s in a["evidence_summary"]:
            out.append(f"- Evidence ({s['signal_id']}): value={s['value']} over n={s['computed_from_n']}")
        if a["verify_signals"]:
            out.append(f"- [VERIFY] not computable in MVP: {', '.join(s['id'] for s in a['verify_signals'])}")
        out.append("")

    comp_surfaced = [c for c in results["composites"].values() if c["surfaced"]]
    if comp_surfaced:
        out.append("## Composite patterns")
        out.append("")
        for c in comp_surfaced:
            out.append(f"- **{c['label']} — {c['name']}** (requires {', '.join(c['requires_all'])}); "
                       f"evidence PMIDs: {', '.join(c['evidence_pmids'][:10])}")
        out.append("")

    out.append("## Insufficient evidence")
    out.append("")
    for a in results["archetypes"].values():
        if not a["surfaced"]:
            out.append(f"- {a['label']} ({a['name']}): {a['reason']}")
    out.append("")
    return "\n".join(out).rstrip() + "\n"


def main() -> int:
    ap = argparse.ArgumentParser(description="Classify an author's trajectory into archetypes (explainable heuristic)")
    ap.add_argument("csv_path", help="Finalized publications CSV from fetch_pubmed.py")
    ap.add_argument("--manifest", help="corpus_manifest.json (default: alongside CSV)", default=None)
    ap.add_argument("--rubric", help="Rubric YAML", default=str(DEFAULT_RUBRIC))
    ap.add_argument("--output-dir", "-o", help="Output directory", default=None)
    args = ap.parse_args()

    csv_path = Path(args.csv_path)
    manifest_path = Path(args.manifest) if args.manifest else csv_path.with_name("corpus_manifest.json")
    out_dir = Path(args.output_dir) if args.output_dir else csv_path.parent / "report"
    out_dir.mkdir(parents=True, exist_ok=True)

    rows = read_csv(csv_path)
    manifest = require_approved_manifest(manifest_path, csv_path, rows)

    rubric, rubric_sha = load_rubric(Path(args.rubric))
    results = score_archetypes(rows, rubric)

    meta = {
        "rubric_version": rubric.get("rubric_version", "unknown"),
        "rubric_sha256": rubric_sha,
        "manifest_review_status": manifest.get("review_status"),
        "disambiguation_basis": manifest.get("disambiguators", {}),
    }
    out_json = {**meta, **results}
    (out_dir / "archetype_results.json").write_text(
        json.dumps(out_json, indent=2, ensure_ascii=False), encoding="utf-8")
    (out_dir / "archetype_report.md").write_text(_md_report(results, meta), encoding="utf-8")
    print(f"Wrote {out_dir / 'archetype_results.json'} and archetype_report.md")
    surfaced = [a["label"] for a in results["archetypes"].values() if a["surfaced"]]
    print(f"Surfaced archetypes: {', '.join(surfaced) if surfaced else '(none — insufficient evidence)'}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Fetch PubMed publications for an author, with an explicit disambiguation gate.

Network fetch uses NCBI E-utilities via Biopython. All XML parsing and target-author
attribution lives in the stdlib-only pubmed_parse module (so that logic is CI-tested
without Biopython).

Disambiguation gate (see SKILL.md Step 6):
  - A surname alone does NOT resolve an author. Pass disambiguators
    (--initials / --orcid / --affiliation / --year-from / --year-to) and review the
    candidate clusters this script surfaces.
  - The finalized CSV is always written WITH a corpus_manifest.json that records the
    query, disambiguators, included/excluded PMIDs, a CSV sha256, a sorted-PMID-set
    hash, and review_status.
  - review_status is "approved" ONLY when --approve is passed, which is a HUMAN gate:
    the user passes it after reviewing the clusters. classify_archetypes.py refuses to
    run on a manifest that is not approved (or whose hashes do not match the CSV).

Usage:
    python fetch_pubmed.py "Author Name" --initials AB [--orcid 0000-...] \
        [--affiliation "Some Institution"] [--year-from 2015] [--year-to 2026] \
        [--disambiguate] [--include-pmids inc.txt] [--exclude-pmids exc.txt] [--approve] \
        [--output out.csv] [--email user@example.com]
"""

import argparse
import csv
import hashlib
import json
import re
import time
import xml.etree.ElementTree as ET
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

from Bio import Entrez

import pubmed_parse

BATCH_SIZE = 100


def build_query(author: str, affiliation: str = "", year_from: str = "", year_to: str = "") -> str:
    q = f'"{author}"[Author]'
    if affiliation:
        q += f' AND "{affiliation}"[Affiliation]'
    if year_from or year_to:
        lo = year_from or "1900"
        hi = year_to or "3000"
        q += f' AND ("{lo}"[dp] : "{hi}"[dp])'
    return q


def search_pubmed(query: str) -> list[str]:
    handle = Entrez.esearch(db="pubmed", term=query, retmax=0)
    record = Entrez.read(handle)
    handle.close()
    total = int(record["Count"])
    print(f"Total results: {total}")

    pmids: list[str] = []
    for start in range(0, total, BATCH_SIZE):
        handle = Entrez.esearch(db="pubmed", term=query, retstart=start, retmax=BATCH_SIZE)
        record = Entrez.read(handle)
        handle.close()
        pmids.extend(record["IdList"])
        time.sleep(0.4)
    return pmids


def fetch_details(pmids: list[str], target_last: str, target_initials: str, target_orcid: str) -> list[dict]:
    all_records: list[dict] = []
    for start in range(0, len(pmids), BATCH_SIZE):
        batch = pmids[start:start + BATCH_SIZE]
        print(f"Fetching details {start+1}-{start+len(batch)} of {len(pmids)}...")
        handle = Entrez.efetch(db="pubmed", id=batch, rettype="xml", retmode="xml")
        xml_data = handle.read()
        handle.close()
        all_records.extend(
            pubmed_parse.records_from_xml(xml_data, target_last, target_initials, target_orcid)
        )
        time.sleep(0.5)
    return all_records


def read_pmid_file(path: str) -> set[str]:
    text = Path(path).read_text(encoding="utf-8")
    return {tok for tok in re.split(r"[\s,]+", text) if tok.strip().isdigit()}


def _affiliation_key(aff: str) -> str:
    if not aff:
        return "(affiliation unknown)"
    # First comma-delimited segment, lowercased, trimmed — a coarse institution token.
    seg = aff.split(",")[0].strip().lower()
    return seg[:60] or "(affiliation unknown)"


def cluster_candidates(records: list[dict]) -> list[dict]:
    """Group records into candidate clusters by target affiliation token + year span."""
    buckets: dict[str, list[dict]] = {}
    for r in records:
        buckets.setdefault(_affiliation_key(r.get("target_affiliation", "")), []).append(r)
    clusters = []
    for key, rows in buckets.items():
        years = [int(r["year"]) for r in rows if str(r.get("year", "")).isdigit()]
        clusters.append({
            "affiliation_token": key,
            "n_papers": len(rows),
            "year_min": min(years) if years else None,
            "year_max": max(years) if years else None,
            "sample_titles": [r["title"][:90] for r in rows[:3]],
            "pmids": [r["pmid"] for r in rows],
        })
    clusters.sort(key=lambda c: c["n_papers"], reverse=True)
    return clusters


def sha256_file(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def pmid_set_hash(pmids) -> str:
    joined = "\n".join(sorted(set(str(p) for p in pmids)))
    return hashlib.sha256(joined.encode("utf-8")).hexdigest()


def write_manifest(manifest_path: Path, *, query, disambiguators, included, excluded,
                   approved, output_csv: Path, record_count: int, pmids) -> None:
    manifest = {
        "schema": "author-strategy/corpus_manifest@1",
        "query": query,
        "disambiguators": disambiguators,
        "included_pmids": sorted(set(str(p) for p in included)) if included else [],
        "excluded_pmids": sorted(set(str(p) for p in excluded)) if excluded else [],
        "review_status": "approved" if approved else "pending",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "record_count": record_count,
        "csv_path": output_csv.name,
        "csv_sha256": sha256_file(output_csv),
        "pmid_set_hash": pmid_set_hash(pmids),
    }
    manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8")


def main():
    parser = argparse.ArgumentParser(description="Fetch PubMed publications for an author (with disambiguation gate)")
    parser.add_argument("author", help="Author name for PubMed search (e.g., 'Kim DK')")
    parser.add_argument("--last-name", help="Last name for attribution (auto-detected if omitted)")
    parser.add_argument("--initials", default="", help="Target author initials for attribution (e.g., 'DK')")
    parser.add_argument("--orcid", default="", help="Target author ORCID (authoritative attribution)")
    parser.add_argument("--affiliation", default="", help="Affiliation filter to refine the PubMed search")
    parser.add_argument("--year-from", default="", help="Earliest publication year (refines search)")
    parser.add_argument("--year-to", default="", help="Latest publication year (refines search)")
    parser.add_argument("--output", "-o", help="Output CSV path", default=None)
    parser.add_argument("--manifest", help="Corpus manifest path (default: alongside CSV)", default=None)
    parser.add_argument("--include-pmids", help="File of PMIDs to keep (whitespace/comma separated)", default=None)
    parser.add_argument("--exclude-pmids", help="File of PMIDs to drop", default=None)
    parser.add_argument("--disambiguate", action="store_true",
                        help="Surface candidate clusters for manual review (does not auto-approve)")
    parser.add_argument("--approve", action="store_true",
                        help="HUMAN GATE: mark the corpus manifest review_status=approved after reviewing clusters")
    parser.add_argument("--email", help="Email for NCBI E-utilities", default="research@example.com")
    args = parser.parse_args()

    Entrez.email = args.email

    target_last = args.last_name or args.author.split()[-1]
    query = build_query(args.author, args.affiliation, args.year_from, args.year_to)

    if args.output:
        output_csv = Path(args.output)
    else:
        safe_name = args.author.replace(" ", "_").replace('"', "")
        output_csv = Path(f"{safe_name}_publications.csv")
    manifest_path = Path(args.manifest) if args.manifest else output_csv.with_name("corpus_manifest.json")

    has_disambiguator = bool(args.initials or args.orcid or args.affiliation or args.year_from or args.year_to)

    print(f"Searching PubMed for: {query}")
    print(f"Target last name for attribution: {target_last}")
    pmids = search_pubmed(query)
    print(f"Found {len(pmids)} PMIDs")
    if not pmids:
        print("No results found. Check the author name format (e.g., 'Yon DK' vs 'Yon D').")
        return

    records = fetch_details(pmids, target_last, args.initials, args.orcid)

    include = read_pmid_file(args.include_pmids) if args.include_pmids else None
    exclude = read_pmid_file(args.exclude_pmids) if args.exclude_pmids else set()
    if include is not None:
        records = [r for r in records if r["pmid"] in include]
    if exclude:
        records = [r for r in records if r["pmid"] not in exclude]
    print(f"Finalized corpus: {len(records)} records")

    output_csv.parent.mkdir(parents=True, exist_ok=True)
    with open(output_csv, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=pubmed_parse.CSV_FIELDNAMES)
        writer.writeheader()
        writer.writerows(records)
    print(f"Saved CSV: {output_csv}")

    # Candidate clusters for disambiguation review.
    clusters = cluster_candidates(records)
    candidates_path = output_csv.with_name("candidates.json")
    candidates_path.write_text(json.dumps(clusters, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"\n=== Candidate clusters (review before classifying) ===")
    for c in clusters:
        span = f"{c['year_min']}-{c['year_max']}" if c["year_min"] else "n/a"
        print(f"  [{c['n_papers']:>3} papers | {span}] {c['affiliation_token']}")
        for t in c["sample_titles"]:
            print(f"        - {t}")

    disambiguators = {
        "initials": args.initials, "orcid": args.orcid, "affiliation": args.affiliation,
        "year_from": args.year_from, "year_to": args.year_to,
    }
    write_manifest(
        manifest_path, query=query, disambiguators=disambiguators,
        included=include, excluded=exclude, approved=args.approve,
        output_csv=output_csv, record_count=len(records), pmids=[r["pmid"] for r in records],
    )
    print(f"\nSaved manifest: {manifest_path} (review_status={'approved' if args.approve else 'pending'})")

    if not args.approve:
        print("\n[GATE] Classification is BLOCKED until you approve the corpus.")
        if not has_disambiguator:
            print("       A surname alone does not resolve an author — pass --initials/--orcid/"
                  "--affiliation/--year-from/--year-to.")
        print("       Review candidates.json, then re-run with --include-pmids/--exclude-pmids "
              "and --approve to finalize.")

    # Quick summary.
    types = Counter(r["study_type"] for r in records)
    positions = Counter(r["author_position"] for r in records)
    print("\n=== Study Type Distribution ===")
    for k, v in types.most_common():
        print(f"  {k}: {v} ({v/len(records)*100:.1f}%)")
    print("\n=== Author Position (positional heuristic) ===")
    for k, v in positions.most_common():
        print(f"  {k}: {v} ({v/len(records)*100:.1f}%)")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Stdlib-only PubMed XML parsing + target-author attribution + record classifiers.

Split out of fetch_pubmed.py so the safety-critical logic — which author on a paper
is the *target* author, and what metadata is attributed to them — is unit-tested in CI
WITHOUT requiring Biopython (fetch_pubmed.py keeps the Bio.Entrez network dependency).

Design rules:
- Target-author attribution never borrows a co-author's ORCID/affiliation. When two
  same-surname authors appear on one paper and initials/ORCID cannot disambiguate, the
  record's target metadata (and position) is `unknown` — never guessed.
- Author position is a positional heuristic only: first / middle / last / unknown, plus
  the real `EqualContrib` flag when PubMed marks it. It is NOT leadership metadata.

Only the standard library is imported here.
"""

from __future__ import annotations

import re
import xml.etree.ElementTree as ET

# ---------------------------------------------------------------------------
# Target-author attribution
# ---------------------------------------------------------------------------


def _norm_initials(s: str) -> str:
    """Normalize initials to uppercase letters only (e.g., 'D. K.' -> 'DK')."""
    return re.sub(r"[^A-Za-z]", "", s or "").upper()


def _norm_orcid(s: str) -> str:
    """Normalize an ORCID to the 16 hex-ish chars (strip URL prefix, dashes)."""
    s = (s or "").strip()
    s = re.sub(r"^https?://orcid\.org/", "", s, flags=re.I)
    return re.sub(r"[^0-9Xx]", "", s).upper()


def match_target_author(
    authors: list[dict],
    target_last: str,
    target_initials: str = "",
    target_orcid: str = "",
) -> tuple[int | None, str]:
    """Return (index_of_target_author, match_basis).

    match_basis is one of: orcid, initials, surname-unique, ambiguous-initials,
    ambiguous-surname, no-surname-match. The two `ambiguous-*` bases mean the target
    could not be uniquely identified on this paper -> callers must NOT attribute
    co-author metadata.
    """
    tl = (target_last or "").lower()
    surname_cands = [i for i, a in enumerate(authors) if a.get("LastName", "").lower() == tl]
    if not surname_cands:
        # fallback: substring (handles compound surnames / encoding quirks)
        surname_cands = [i for i, a in enumerate(authors) if tl and tl in a.get("LastName", "").lower()]
    if not surname_cands:
        return None, "no-surname-match"

    # 1. ORCID is authoritative.
    if target_orcid:
        tgt = _norm_orcid(target_orcid)
        orcid_matches = [i for i in surname_cands if _norm_orcid(authors[i].get("ORCID", "")) == tgt and tgt]
        if len(orcid_matches) >= 1:
            return orcid_matches[0], "orcid"
        # provided ORCID matched no surname candidate -> fall through to initials/surname.

    # 2. Initials.
    if target_initials:
        ti = _norm_initials(target_initials)
        init_matches = [i for i in surname_cands if _norm_initials(authors[i].get("Initials", "")) == ti and ti]
        if len(init_matches) == 1:
            return init_matches[0], "initials"
        if len(init_matches) > 1:
            return init_matches[0], "ambiguous-initials"
        # no initials match -> fall through to surname.

    # 3. Surname only.
    if len(surname_cands) == 1:
        return surname_cands[0], "surname-unique"
    return surname_cands[0], "ambiguous-surname"


def classify_author_position(idx: int | None, n_authors: int) -> str:
    """Positional heuristic only: first / middle / last / unknown."""
    if idx is None or n_authors <= 0:
        return "unknown"
    if idx == 0:
        return "first"
    if idx == n_authors - 1:
        return "last"
    return "middle"


# ---------------------------------------------------------------------------
# Content classifiers (unchanged logic, stdlib-only)
# ---------------------------------------------------------------------------


def classify_study_type(title: str, abstract: str, mesh_terms: list[str], pub_types: list[str]) -> str:
    text = (title + " " + abstract).lower()
    pub_lower = " ".join(pub_types).lower()

    if "global burden" in text or "gbd" in text:
        return "GBD"
    if ("systematic review" in text or "meta-analysis" in text or
            "systematic review" in pub_lower or "meta-analysis" in pub_lower):
        return "SR/MA"
    if ("national health insurance" in text or "nhis" in text or
            "claims database" in text or "nationwide cohort" in text):
        return "NHIS/Claims"
    if ("cross-national" in text or "binational" in text or
            ("korea" in text and ("united states" in text or "japan" in text or
             "france" in text or "american" in text))):
        return "Cross-national"
    if ("knhanes" in text or "nhanes" in text or "national health and nutrition" in text or
            "kchs" in text or "national survey" in text):
        return "National survey"
    if "biobank" in text:
        return "Biobank"
    if ("machine learning" in text or "deep learning" in text or
            "artificial intelligence" in text or "neural network" in text):
        return "AI/ML"
    if "randomized" in text or "clinical trial" in pub_lower:
        return "Clinical trial"
    if "case report" in text or "case report" in pub_lower:
        return "Case report"
    if "letter" in pub_lower or "comment" in pub_lower or "editorial" in pub_lower:
        return "Letter/Commentary"
    return "Other"


def classify_topic(title: str, abstract: str, mesh_terms: list[str]) -> str:
    text = (title + " " + abstract).lower()
    topics = {
        "Allergy/Respiratory": ["allergy", "allergic", "asthma", "respiratory", "atopic",
                                 "rhinitis", "eczema", "copd", "pneumonia", "lung disease"],
        "Cardiovascular": ["cardiovascular", "coronary", "heart", "myocardial", "hypertension",
                            "stroke", "atherosclerosis", "arrhythmia"],
        "Mental health": ["depression", "anxiety", "mental health", "psychiatric", "suicide",
                          "adhd", "autism", "bipolar", "schizophrenia"],
        "Infectious": ["covid", "sars-cov", "infection", "vaccine", "vaccination", "herpes zoster",
                       "influenza", "hepatitis", "tuberculosis"],
        "Oncology": ["cancer", "tumor", "malignant", "neoplasm", "carcinoma", "leukemia",
                     "lymphoma"],
        "Metabolic": ["diabetes", "obesity", "metabolic", "lipid", "cholesterol", "fatty liver",
                      "bmi", "insulin"],
        "Nutrition/Lifestyle": ["diet", "nutrition", "physical activity", "exercise", "sleep",
                                "sedentary", "alcohol", "smoking"],
        "Musculoskeletal": ["osteoporosis", "fracture", "arthritis", "bone", "sarcopenia",
                            "musculoskeletal", "spine", "joint"],
        "Neurological": ["dementia", "alzheimer", "parkinson", "epilepsy", "migraine",
                         "cerebrovascular", "brain", "cognitive", "aneurysm"],
        "Radiology/Imaging": ["radiolog", "imaging", "ct ", "mri", "ultrasound", "x-ray",
                              "mammograph", "pet", "contrast"],
        "GI/Hepatology": ["gastro", "liver", "hepat", "pancrea", "colon", "bowel",
                          "endoscop", "cirrhosis"],
        "Ophthalmology": ["ophthalm", "vision", "macular", "retinal", "eye", "glaucoma"],
        "Pediatrics": ["child", "pediatric", "adolescent", "infant", "neonatal", "prenatal",
                       "offspring"],
    }
    scores = {}
    for topic, keywords in topics.items():
        score = sum(1 for kw in keywords if kw in text)
        if score > 0:
            scores[topic] = score
    if not scores:
        return "Other"
    return max(scores, key=scores.get)


def classify_journal_tier(journal: str) -> str:
    j = (journal or "").lower()
    if any(x in j for x in ["lancet"]):
        return "Lancet family"
    if any(x in j for x in ["nature", "nat med", "nat rev", "nat commun"]):
        return "Nature family"
    if any(x in j for x in ["n engl j med", "bmj", "jama"]):
        return "NEJM/BMJ/JAMA"
    high_if = ["circulation", "eur heart j", "allergy", "j allergy clin immunol",
               "ebiomedic", "sci adv", "cell", "ann oncol", "gut", "radiology",
               "eur radiol", "invest radiol"]
    if any(x in j for x in high_if):
        return "IF>=10"
    return "Other"


# ---------------------------------------------------------------------------
# Article parsing
# ---------------------------------------------------------------------------

CSV_FIELDNAMES = [
    "pmid", "title", "journal", "journal_abbrev", "year",
    "n_authors", "author_position", "equal_contrib", "match_basis",
    "target_initials", "target_orcid", "target_affiliation",
    "study_type", "topic", "journal_tier", "pub_types", "mesh_terms", "abstract",
]


def _extract_authors(art) -> list[dict]:
    authors: list[dict] = []
    author_list = art.find(".//AuthorList") if art is not None else None
    if author_list is None:
        return authors
    for auth_el in author_list.findall("Author"):
        last = auth_el.find("LastName")
        fore = auth_el.find("ForeName")
        initials = auth_el.find("Initials")
        collective = auth_el.find("CollectiveName")
        orcid = ""
        for ident in auth_el.findall("Identifier"):
            if (ident.get("Source") or "").upper() == "ORCID":
                orcid = (ident.text or "").strip()
                break
        affiliation = ""
        aff_el = auth_el.find(".//AffiliationInfo/Affiliation")
        if aff_el is not None and aff_el.text:
            affiliation = aff_el.text.strip()
        authors.append({
            "LastName": last.text if last is not None else "",
            "ForeName": fore.text if fore is not None else "",
            "Initials": initials.text if initials is not None else "",
            "CollectiveName": collective.text if collective is not None else "",
            "ORCID": orcid,
            "Affiliation": affiliation,
            "EqualContrib": (auth_el.get("EqualContrib") or "").upper(),
        })
    return authors


def parse_article(article, target_last: str, target_initials: str = "", target_orcid: str = "") -> dict:
    """Parse one <PubmedArticle> element into a flat record dict.

    Target-author metadata (ORCID/affiliation/initials/position/equal_contrib) is
    attributed ONLY when the target author is uniquely identified on the paper.
    """
    medline = article.find(".//MedlineCitation")
    art = medline.find(".//Article") if medline is not None else None

    pmid = ""
    pmid_el = medline.find(".//PMID") if medline is not None else None
    if pmid_el is not None:
        pmid = pmid_el.text or ""

    title = ""
    title_el = art.find(".//ArticleTitle") if art is not None else None
    if title_el is not None:
        title = "".join(title_el.itertext()).strip()

    journal = ""
    journal_el = art.find(".//Journal/Title") if art is not None else None
    if journal_el is not None:
        journal = journal_el.text or ""

    journal_abbrev = ""
    ja_el = art.find(".//Journal/ISOAbbreviation") if art is not None else None
    if ja_el is not None:
        journal_abbrev = ja_el.text or ""

    year = ""
    year_el = art.find(".//Journal/JournalIssue/PubDate/Year") if art is not None else None
    if year_el is not None:
        year = year_el.text or ""
    else:
        medline_date = art.find(".//Journal/JournalIssue/PubDate/MedlineDate") if art is not None else None
        if medline_date is not None and medline_date.text:
            match = re.search(r"(20\d{2}|19\d{2})", medline_date.text)
            if match:
                year = match.group(1)

    authors = _extract_authors(art)
    n_authors = len(authors)

    idx, match_basis = match_target_author(authors, target_last, target_initials, target_orcid)
    ambiguous = match_basis in ("ambiguous-initials", "ambiguous-surname")

    if idx is None or ambiguous:
        # Cannot uniquely attribute -> never borrow a co-author's metadata.
        author_position = "unknown"
        equal_contrib = ""
        t_initials = t_orcid = t_affiliation = ""
    else:
        author_position = classify_author_position(idx, n_authors)
        tgt = authors[idx]
        equal_contrib = "Y" if tgt.get("EqualContrib") == "Y" else ""
        t_initials = tgt.get("Initials", "")
        t_orcid = tgt.get("ORCID", "")
        t_affiliation = tgt.get("Affiliation", "")

    abstract = ""
    abstract_el = art.find(".//Abstract") if art is not None else None
    if abstract_el is not None:
        parts = ["".join(at.itertext()).strip() for at in abstract_el.findall("AbstractText")]
        abstract = " ".join(parts)

    mesh_terms = []
    mesh_list = medline.find(".//MeshHeadingList") if medline is not None else None
    if mesh_list is not None:
        for mh in mesh_list.findall("MeshHeading"):
            desc = mh.find("DescriptorName")
            if desc is not None:
                mesh_terms.append(desc.text or "")

    pub_types = []
    pt_list = art.find(".//PublicationTypeList") if art is not None else None
    if pt_list is not None:
        for pt in pt_list.findall("PublicationType"):
            pub_types.append(pt.text or "")

    study_type = classify_study_type(title, abstract, mesh_terms, pub_types)
    topic = classify_topic(title, abstract, mesh_terms)
    journal_tier = classify_journal_tier(journal_abbrev or journal)

    return {
        "pmid": pmid,
        "title": title,
        "journal": journal,
        "journal_abbrev": journal_abbrev,
        "year": year,
        "n_authors": n_authors,
        "author_position": author_position,
        "equal_contrib": equal_contrib,
        "match_basis": match_basis,
        "target_initials": t_initials,
        "target_orcid": t_orcid,
        "target_affiliation": t_affiliation,
        "study_type": study_type,
        "topic": topic,
        "journal_tier": journal_tier,
        "pub_types": "; ".join(pub_types),
        "mesh_terms": "; ".join(mesh_terms),
        "abstract": abstract[:500],
    }


def records_from_xml(xml_data, target_last: str, target_initials: str = "", target_orcid: str = "") -> list[dict]:
    """Parse a PubmedArticleSet XML (bytes or str) into records. Used by tests + fetch."""
    if isinstance(xml_data, bytes):
        root = ET.fromstring(xml_data)
    else:
        root = ET.fromstring(xml_data)
    return [parse_article(a, target_last, target_initials, target_orcid)
            for a in root.findall(".//PubmedArticle")]