#!/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")]

#!/usr/bin/env python3
"""Render references/trajectory_archetypes.md from the canonical YAML rubric.

The YAML (references/trajectory_archetypes.yaml) is the single source of truth.
This script regenerates the human-readable Markdown narrative from it so the two
can never drift. CI / the skill test runs `--check` to assert they are in sync.

Usage:
  python3 render_archetype_doc.py            # write the .md
  python3 render_archetype_doc.py --check     # exit 1 if the .md is stale
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

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

HERE = Path(__file__).resolve().parent
YAML_PATH = HERE / "references" / "trajectory_archetypes.yaml"
MD_PATH = HERE / "references" / "trajectory_archetypes.md"

PROV_LABEL = {
    "source-derived": "`source-derived`",
    "rule-derived": "`rule-derived`",
    "unavailable": "`unavailable` [VERIFY]",
}


def _signal_line(sig: dict) -> str:
    prov = PROV_LABEL.get(sig.get("provenance", ""), sig.get("provenance", ""))
    weight = sig.get("weight", 0.0)
    narrative = sig.get("narrative", "").strip()
    line = f"- **{sig['id']}** ({prov}, weight {weight}) — {narrative}"
    note = sig.get("verify_note")
    if note:
        line += f" _{note.strip()}_"
    return line


def render(data: dict) -> str:
    out: list[str] = []
    out.append("<!-- GENERATED FILE — do not edit by hand.")
    out.append("     Source of truth: trajectory_archetypes.yaml")
    out.append("     Regenerate: python3 render_archetype_doc.py -->")
    out.append("")
    out.append("# Trajectory-Archetype Rubric")
    out.append("")
    out.append(f"Rubric version: **{data['rubric_version']}**")
    out.append("")
    out.append(
        "Classification output is an **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 fetched PMIDs. Below the "
        "minimum sample, below the score threshold, or with a negative rule firing, the "
        "archetype is reported as `insufficient evidence`."
    )
    out.append("")
    out.append("## Provenance tags")
    out.append("")
    out.append("- `source-derived` — a raw PubMed record field (n_authors, year, pub_types, study_type).")
    out.append("- `rule-derived` — a threshold / share / fraction / term-match computed over raw fields.")
    out.append(
        "- `unavailable` [VERIFY] — cannot be computed in the PubMed-only MVP "
        "(h-index, citation counts/half-life, venue-impact tier, repository/preprint "
        "links, cross-platform divergence). Weight 0; excluded from the score denominator; never fabricated."
    )
    out.append("")
    out.append("## Scoring")
    out.append("")
    out.append("- A signal is *computable for a dataset* iff its provenance is `source-derived` or `rule-derived` AND all its required columns are present.")
    out.append("- `score = (sum of weights of fired computable signals) / (sum of weights of all computable signals)`, clamped to [0, 1].")
    out.append("- `unavailable` signals never enter the numerator or denominator — they surface only as [VERIFY] notes.")
    out.append("- A **negative** rule firing suppresses the label (`insufficient evidence`) regardless of score.")
    out.append("- A label is surfaced iff: no negative fired, sample `n >= min_sample`, `score >= score_threshold`, and at least one computable signal fired.")
    out.append("")
    out.append("## Confidence (capped at each archetype's `max_confidence_mvp`)")
    out.append("")
    out.append("- **high** — >= 3 distinct computable signals fired AND `n >= min_sample`")
    out.append("- **med** — >= 2 distinct computable signals fired")
    out.append("- **low** — >= 1 computable signal fired")
    out.append("")
    out.append("## Author position caveat")
    out.append("")
    out.append(
        "Author position is a positional heuristic (`first` / `middle` / `last` / "
        "`unknown`, plus real `EqualContrib` metadata when PubMed marks it). It is NOT "
        "authoritative leadership or corresponding-author metadata, which are "
        "`unavailable` in this MVP."
    )
    out.append("")
    out.append("## Archetypes (multi-label; an author may score on several)")
    out.append("")

    for key in sorted(data["archetypes"].keys()):
        arc = data["archetypes"][key]
        out.append(f"### {key} — {arc['name']}")
        out.append("")
        out.append(arc["summary"].strip())
        out.append("")
        out.append(
            f"_min_sample: {arc['min_sample']} · score_threshold: {arc['score_threshold']} "
            f"· max_confidence_mvp: {arc['max_confidence_mvp']}_"
        )
        out.append("")
        if arc.get("flag"):
            out.append(f"> **Flag.** {arc['flag'].strip()}")
            out.append("")
        out.append("Signals:")
        out.append("")
        for sig in arc["signals"]:
            out.append(_signal_line(sig))
        out.append("")
        negs = arc.get("negatives") or []
        if negs:
            out.append("Negatives (rule the archetype OUT):")
            out.append("")
            for neg in negs:
                out.append(f"- **{neg['id']}** — {neg.get('narrative', '').strip()}")
            out.append("")

    composites = data.get("composites") or {}
    if composites:
        out.append("## Composite patterns (computed combinations, not independent labels)")
        out.append("")
        for key in sorted(composites.keys()):
            comp = composites[key]
            out.append(f"### {key} — {comp['name']}")
            out.append("")
            out.append(comp["summary"].strip())
            out.append("")
            out.append(f"- Requires all of: {', '.join(comp['requires_all'])}")
            extra = comp.get("extra_condition", {})
            if extra:
                out.append(f"- Extra condition — {extra.get('narrative', '').strip()}")
            out.append(f"- {comp.get('narrative', '').strip()}")
            out.append("")

    return "\n".join(out).rstrip() + "\n"


def main() -> int:
    ap = argparse.ArgumentParser(description="Render trajectory_archetypes.md from the YAML rubric")
    ap.add_argument("--check", action="store_true", help="verify the .md is in sync; exit 1 on drift")
    args = ap.parse_args()

    if not YAML_PATH.exists():
        sys.stderr.write(f"ERROR: missing {YAML_PATH}\n")
        return 2
    data = yaml.safe_load(YAML_PATH.read_text(encoding="utf-8"))
    rendered = render(data)

    if args.check:
        current = MD_PATH.read_text(encoding="utf-8") if MD_PATH.exists() else ""
        if current != rendered:
            sys.stderr.write(
                f"DRIFT: {MD_PATH.name} is out of sync with the YAML rubric. "
                "Run: python3 render_archetype_doc.py\n"
            )
            return 1
        print(f"OK: {MD_PATH.name} matches the rubric YAML.")
        return 0

    MD_PATH.write_text(rendered, encoding="utf-8")
    print(f"Wrote {MD_PATH}")
    return 0


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

---
name: author-strategy
description: PubMed author profile analysis. Author name → PubMed fetch → study-type classification → visualization → strategy report → optional trajectory-archetype classification.
triggers: author-strategy, 저자 분석, publication analysis, 다작 분석, 연구 전략 분석, author profile, reverse engineer strategy, trajectory archetype, career archetype
tools: Read, Write, Edit, Bash, Glob, Grep
model: inherit
---

# /author-strategy — PubMed Author Strategy Analysis

## Purpose

Analyze a researcher's PubMed publication portfolio to reverse-engineer their research strategy. Produces a CSV dataset, 7 visualizations, and a strategy report.

## Prerequisites

- Python 3.10+ with `biopython`, `pandas`, `matplotlib`, `seaborn`, and `pyyaml` (PyYAML is required by the archetype classifier and the rubric renderer)
- Scripts: `${CLAUDE_SKILL_DIR}/fetch_pubmed.py`, `${CLAUDE_SKILL_DIR}/analyze_patterns.py`, `${CLAUDE_SKILL_DIR}/pubmed_parse.py` (stdlib parser), `${CLAUDE_SKILL_DIR}/classify_archetypes.py`, `${CLAUDE_SKILL_DIR}/render_archetype_doc.py`
- Rubric: `${CLAUDE_SKILL_DIR}/references/trajectory_archetypes.yaml` (canonical) and `${CLAUDE_SKILL_DIR}/references/trajectory_archetypes.md` (generated)

## Workflow

### Step 1: Gather Input

Ask the user for:
1. **Author name** (PubMed format, e.g., "Kim DK" or "Lee KS")
2. **Last name** for position classification (auto-detected if ambiguous)
3. **Output directory** (default: `~/.local/cache/author-strategy/{AuthorName}/`)

### Step 2: Fetch PubMed Data

```bash
python "${CLAUDE_SKILL_DIR}/fetch_pubmed.py" "{Author Name}" \
  --last-name "{LastName}" \
  --output "{output_dir}/data/{name}_publications.csv" \
  --email "{user_email}"
```

Review the console summary (total count, study type distribution, author position).
If count is 0, suggest alternative name formats (e.g., "Yon DK" vs "Yon D" vs "Yon Dong Keon").

### Step 3: Generate Visualizations and Report

```bash
python "${CLAUDE_SKILL_DIR}/analyze_patterns.py" "{output_dir}/data/{name}_publications.csv" \
  --output-dir "{output_dir}/report/" \
  --author-name "{Author Name}"
```

This produces:
- 7 PNG charts (01-07)
- `analysis_report.md` with strategy breakdown

### Step 4: Interpret and Present

Read `analysis_report.md` and present to the user:

1. **Executive summary**: total publications, growth trajectory, high-tier rate
2. **Primary strategy**: what study type dominates and why
3. **Author position analysis**: first/last positional rate vs middle (positional heuristic only — not leadership or corresponding-author metadata, which are unavailable here)
4. **Topic clusters**: research focus areas
5. **ROI quadrant**: which strategies yield high-tier + leadership vs. volume only
6. **Replication opportunities**: which patterns are replicable with Claude Code + public databases

### Step 5: Optional — MA Gap Identification

If the user asks "what MA topics are feasible with this professor?":
- Cross-reference topic clusters with existing MA plans in memory
- Identify gaps where the professor has domain expertise but no MA published
- Output a prioritized list of MA proposals

## Optional: Trajectory-Archetype Classification

A second, opt-in capability that classifies the author's trajectory into abstract
career archetypes (A1–A6 + a composite) as an **explainable, multi-label,
confidence-scored heuristic — not an objective verdict**. The rubric is the canonical
`references/trajectory_archetypes.yaml`. This path is **gated**: a surname alone does not
resolve an author, so the corpus must pass an explicit disambiguation review before it
can be classified.

### Step 6: Disambiguation Gate (required before classification)

Pass disambiguators so the target author is uniquely attributed (a surname alone is never
sufficient):

```bash
python "${CLAUDE_SKILL_DIR}/fetch_pubmed.py" "{Author Name}" \
  --initials "{Initials}" --orcid "{ORCID}" \
  --affiliation "{Institution}" --year-from "{YYYY}" --year-to "{YYYY}" \
  --output "{output_dir}/data/{name}_publications.csv" --email "{user_email}"
```

This writes the CSV, a `candidates.json` of affiliation/year candidate clusters, and a
`corpus_manifest.json` with `review_status: pending`. **Present the candidate clusters to
the user for review.** The user decides include/exclude. Only after the user has reviewed
the clusters do you finalize and approve the corpus (the `--approve` flag is a human gate
— never set it without explicit user review/approval):

```bash
python "${CLAUDE_SKILL_DIR}/fetch_pubmed.py" "{Author Name}" \
  --initials "{Initials}" --affiliation "{Institution}" \
  --include-pmids "{included.txt}" --exclude-pmids "{excluded.txt}" --approve \
  --output "{output_dir}/data/{name}_publications.csv" --email "{user_email}"
```

The manifest is cryptographically bound to the CSV (`csv_sha256` + `pmid_set_hash`); the
classifier refuses to run on an unapproved or mismatched corpus.

### Step 7: Run the Classifier and Present

```bash
python "${CLAUDE_SKILL_DIR}/classify_archetypes.py" \
  "{output_dir}/data/{name}_publications.csv" \
  --manifest "{output_dir}/data/corpus_manifest.json" \
  --rubric "${CLAUDE_SKILL_DIR}/references/trajectory_archetypes.yaml" \
  --output-dir "{output_dir}/report/"
```

Read `archetype_report.md` and present it to the user, **stating up front that the labels
are explainable heuristics, not objective classifications**. For each surfaced archetype,
show the score, confidence band, and the author's own evidence PMIDs. Honor the `[VERIFY]`
markers (h-index/citation/venue-tier are unavailable) and the A5 participation flag. List
the `insufficient evidence` archetypes too.

To retune the rubric, edit only the YAML and regenerate the narrative doc:

```bash
python "${CLAUDE_SKILL_DIR}/render_archetype_doc.py"        # regenerate the .md
python "${CLAUDE_SKILL_DIR}/render_archetype_doc.py" --check # CI/test sync gate
```

## Study Type Classifier

The classifier is tuned for Korean epidemiology and public health researchers. Categories:

| Type | Detection Pattern |
|------|------------------|
| GBD | "global burden" or "gbd" in title/abstract |
| SR/MA | "systematic review" or "meta-analysis" |
| NHIS/Claims | "national health insurance", "nhis", "claims database", "nationwide cohort" |
| Cross-national | Country pairs or "cross-national"/"binational" |
| National survey | "knhanes", "nhanes", "kchs", "national survey" |
| Biobank | "biobank" |
| AI/ML | "machine learning", "deep learning", "artificial intelligence" |
| Clinical trial | "randomized" or publication type |
| Case report | "case report" |
| Letter/Commentary | Publication type = letter/comment/editorial |

**Known limitation**: The classifier may undercount NHIS studies when they appear in Cross-national or Other categories. The report notes this.

## Known Limitations

- The study type classifier is tuned for epidemiology and public health researchers. May undercount specialized study types for other fields.
- NHIS studies may be undercounted when they appear in cross-national or "other" categories.
- PubMed search requires an email for NCBI E-utilities (set via `--email` flag).

## Anti-Hallucination

- **Never fabricate publication counts, h-index, or journal metrics.** All numbers must come from PubMed API output.
- **Never invent study classifications.** If a paper cannot be classified, label it as "Other" rather than guessing.
- If PubMed returns 0 results, suggest alternative name formats rather than generating fake data.
- **Archetype labels are explainable heuristics, not objective classifications.** Every label must carry a score, a confidence band, and evidence (the queried author's own PMIDs). Below the minimum sample or with conflicting signals, report `insufficient evidence` — never force a label.
- **Metadata + stored abstract only.** Signals are computed from PubMed metadata and the title/abstract text already fetched. Do not retrieve full text, follow external links, or resolve preprints. Signals that need citations, citation half-life, venue-impact tier, repository/preprint links, or corresponding-author role are `unavailable` and surface as `[VERIFY]` — never inferred.
- **Author position is a positional heuristic** (first/middle/last/unknown + real EqualContrib). Never present it as authoritative leadership or corresponding-author metadata.
- **Never resolve an author by surname alone.** Classification requires an approved, CSV-bound `corpus_manifest.json`; present candidate clusters for the user to confirm.

## Output Structure

```
{output_dir}/
  data/
    {name}_publications.csv
    candidates.json          # disambiguation candidate clusters (Step 6)
    corpus_manifest.json     # review_status + csv_sha256 + pmid_set_hash (Step 6)
  report/
    analysis_report.md
    01_yearly_stacked.png
    02_study_type_pie.png
    03_author_position.png
    04_journal_tier_heatmap.png
    05_topic_distribution.png
    06_growth_curve.png
    07_strategy_roi.png
    archetype_report.md      # trajectory-archetype classification (Step 7)
    archetype_results.json   # machine-readable labels + scores + evidence
```

schema_version: 2
name: author-strategy
layer: D
owner_domain: author_profile_analysis
maturity: official

when_to_use: "Analyze a PubMed author profile (study-type mix, trends) and produce a strategy report; optionally classify the trajectory into explainable career archetypes (A1-A6) after an author-disambiguation gate."
when_NOT_to_use: "Finding meta-analysis topics (use ma-scout); literature search for a manuscript (use search-lit)."

inputs:
  - "author name / PubMed identifier"
  - "disambiguators for archetype mode (initials / ORCID / affiliation / year window)"
outputs:
  - "study-type classification"
  - "profile visualization"
  - "strategy report"
  - "trajectory-archetype multi-label classification (explainable heuristic)"
side_effects:
  - writes_report_artifacts
  - network_access_pubmed
downstream_consumers:
  - ma-scout
forbidden_actions:
  - fabricate_publication_records
  - infer_identity_without_disambiguation

# v2.1 quality card
purpose: "Summarize an author's PubMed publication profile and surface strategy options from the actual record."
safety_boundaries:
  - "Records are fetched from PubMed, not recalled from memory; author disambiguation is explicit."
  - "Reports describe the public record only; no private or speculative attribution."
known_limitations:
  - "Name collisions on PubMed can blur profiles; the archetype path requires an explicit, manifest-gated disambiguation review."
  - "Archetype labels are explainable heuristics, not objective classifications; citation/h-index/venue-tier signals are unavailable and marked [VERIFY]."
  - "No standalone demo; output is an advisory report."
validation_commands:
  - "manual review of the fetched record against PubMed"
  - "bash skills/author-strategy/tests/test_archetype_classifier.sh"
  - "python3 skills/author-strategy/render_archetype_doc.py --check"
evidence_surface: manual_workflow