#!/usr/bin/env python3
"""Create leakage-free predictive windows for the accepted 120-discussion sample.

This script is fully offline. It reads the normalized historical master comments
and the accepted 120-discussion calibration sample, then writes new derived CSVs.
It does not modify master datasets or run any classification, scoring, NLP, or
interpretive analysis.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

import pandas as pd


ROOT = Path("historical")
COMMENTS_MASTER = ROOT / "comments_2017_2018_master.parquet"
CALIBRATION_DISCUSSIONS = ROOT / "mri_historical_calibration_120_discussions.csv"

PREDICTIVE_COMMENTS_CSV = ROOT / "mri_predictive_120_comments.csv"
PREDICTIVE_DISCUSSIONS_CSV = ROOT / "mri_predictive_120_discussions.csv"


class PredictivePacketError(RuntimeError):
    """Expected preparation or validation failure."""


def require_inputs() -> None:
    missing = [
        str(path)
        for path in [COMMENTS_MASTER, CALIBRATION_DISCUSSIONS]
        if not path.exists()
    ]
    if missing:
        raise PredictivePacketError(f"Missing required input file(s): {', '.join(missing)}")


def parse_bool(series: pd.Series) -> pd.Series:
    if series.dtype == bool:
        return series
    return series.astype(str).str.strip().str.lower().isin(["true", "1", "yes"])


def load_calibration_discussions() -> pd.DataFrame:
    discussions = pd.read_csv(CALIBRATION_DISCUSSIONS, dtype={"articleID": str})
    if len(discussions) != 120:
        raise PredictivePacketError(
            f"Expected 120 calibration discussions, found {len(discussions)}."
        )
    if discussions["articleID"].duplicated().any():
        duplicates = discussions.loc[discussions["articleID"].duplicated(), "articleID"].tolist()
        raise PredictivePacketError(
            "Calibration discussion file contains duplicate articleID values: "
            + ", ".join(duplicates[:10])
        )
    return discussions


def load_top_level_comments(article_ids: set[str]) -> pd.DataFrame:
    comments = pd.read_parquet(
        COMMENTS_MASTER,
        columns=[
            "commentID",
            "articleID",
            "commentBody",
            "createDate",
            "recommendations",
            "editorsSelection",
            "is_top_level",
        ],
    )
    comments["articleID"] = comments["articleID"].astype(str)
    comments = comments[comments["articleID"].isin(article_ids) & parse_bool(comments["is_top_level"])].copy()
    comments["commentID"] = comments["commentID"].astype(str)
    comments["createDate"] = pd.to_numeric(comments["createDate"], errors="coerce")
    comments["recommendations"] = pd.to_numeric(comments["recommendations"], errors="coerce")
    comments["editorsSelection"] = parse_bool(comments["editorsSelection"])
    return comments


def add_order_and_targets(comments: pd.DataFrame) -> pd.DataFrame:
    ordered = comments.sort_values(
        ["articleID", "createDate", "commentID"],
        ascending=[True, True, True],
        kind="mergesort",
    ).copy()
    ordered["comment_order"] = ordered.groupby("articleID").cumcount() + 1
    ordered["top_level_comment_total"] = ordered.groupby("articleID")["commentID"].transform("size")
    ordered["comment_order_pct"] = ordered["comment_order"] / ordered["top_level_comment_total"]
    ordered["is_early_25"] = ordered["comment_order"].le(25)
    ordered["is_late_25"] = ordered["comment_order"].gt(ordered["top_level_comment_total"] - 25)

    future_candidates = ordered[ordered["comment_order"].gt(25)].sort_values(
        ["articleID", "recommendations", "createDate", "commentID"],
        ascending=[True, False, True, True],
        kind="mergesort",
    ).copy()
    future_candidates["future_recognition_rank"] = (
        future_candidates.groupby("articleID").cumcount() + 1
    )

    ordered["future_recognition_rank"] = pd.Series(pd.NA, index=ordered.index, dtype="Int64")
    ordered.loc[future_candidates.index, "future_recognition_rank"] = future_candidates[
        "future_recognition_rank"
    ].astype("Int64")
    ordered["is_future_top_10_recognition"] = ordered["future_recognition_rank"].le(10).fillna(False)

    global_ranked = ordered.sort_values(
        ["articleID", "recommendations", "createDate", "commentID"],
        ascending=[True, False, True, True],
        kind="mergesort",
    ).copy()
    global_ranked["global_recognition_rank"] = global_ranked.groupby("articleID").cumcount() + 1
    ordered["global_recognition_rank"] = pd.Series(pd.NA, index=ordered.index, dtype="Int64")
    ordered.loc[global_ranked.index, "global_recognition_rank"] = global_ranked[
        "global_recognition_rank"
    ].astype("Int64")
    ordered["is_global_top_10_recognition"] = ordered["global_recognition_rank"].le(10).fillna(False)
    ordered["is_global_top_1_recognition"] = ordered["global_recognition_rank"].eq(1).fillna(False)

    return ordered


def validate_article_windows(flagged: pd.DataFrame, article_ids: set[str]) -> None:
    summary = flagged.groupby("articleID").agg(
        top_level_comments=("commentID", "size"),
        early_count=("is_early_25", "sum"),
        future_count=("is_future_top_10_recognition", "sum"),
        late_count=("is_late_25", "sum"),
        missing_text=("commentBody", lambda values: int((values.isna() | values.astype(str).str.strip().eq("")).sum())),
        missing_create_date=("createDate", lambda values: int(values.isna().sum())),
        missing_recommendations=("recommendations", lambda values: int(values.isna().sum())),
        duplicate_comment_ids=("commentID", lambda values: int(values.duplicated().sum())),
    )
    missing_articles = sorted(article_ids - set(summary.index.astype(str)))
    if missing_articles:
        raise PredictivePacketError(
            "No top-level comments found for calibration articleID(s): "
            + ", ".join(missing_articles[:10])
        )
    failures = summary[
        summary["early_count"].ne(25)
        | summary["future_count"].ne(10)
        | summary["late_count"].ne(25)
        | summary["missing_create_date"].gt(0)
        | summary["missing_recommendations"].gt(0)
        | summary["duplicate_comment_ids"].gt(0)
    ]
    if not failures.empty:
        raise PredictivePacketError(
            "Window validation failed for at least one discussion; inspect articleID(s): "
            + ", ".join(failures.index.astype(str).tolist()[:10])
        )


def build_predictive_comments(flagged: pd.DataFrame, discussions: pd.DataFrame) -> pd.DataFrame:
    article_meta = discussions[["articleID", "headline", "publication_date", "sectionName"]].copy()
    windows = flagged[
        flagged["is_early_25"] | flagged["is_future_top_10_recognition"] | flagged["is_late_25"]
    ].copy()
    windows = windows.merge(article_meta, on="articleID", how="left")
    output_columns = [
        "articleID",
        "headline",
        "publication_date",
        "sectionName",
        "commentID",
        "commentBody",
        "createDate",
        "recommendations",
        "comment_order",
        "comment_order_pct",
        "is_early_25",
        "is_future_top_10_recognition",
        "future_recognition_rank",
        "is_late_25",
        "editorsSelection",
    ]
    windows = windows[output_columns].sort_values(
        ["articleID", "comment_order", "future_recognition_rank"], kind="mergesort"
    )
    windows.to_csv(PREDICTIVE_COMMENTS_CSV, index=False)
    return windows


def iso_from_epoch_seconds(value: Any) -> str:
    if pd.isna(value):
        return ""
    return pd.to_datetime(value, unit="s", utc=True).isoformat()


def build_predictive_discussions(flagged: pd.DataFrame, discussions: pd.DataFrame) -> pd.DataFrame:
    top_summary = flagged.groupby("articleID").agg(
        top_level_comments=("commentID", "size"),
        first_comment_epoch=("createDate", "min"),
        last_comment_epoch=("createDate", "max"),
    ).reset_index()
    top_summary["first_comment_time"] = top_summary["first_comment_epoch"].map(iso_from_epoch_seconds)
    top_summary["last_comment_time"] = top_summary["last_comment_epoch"].map(iso_from_epoch_seconds)
    top_summary["discussion_duration"] = (
        top_summary["last_comment_epoch"] - top_summary["first_comment_epoch"]
    )
    top_summary = top_summary.drop(columns=["first_comment_epoch", "last_comment_epoch"])

    future = flagged[flagged["is_future_top_10_recognition"]].copy()
    future_summary = future.groupby("articleID").agg(
        future_top10_total_recommendations=("recommendations", "sum"),
        future_top1_recommendations=("recommendations", "max"),
        median_future_top10_comment_order_pct=("comment_order_pct", "median"),
    ).reset_index()

    base_columns = [
        "articleID",
        "publication_date",
        "headline",
        "snippet",
        "sectionName",
        "newsDesk",
        "typeOfMaterial",
        "total_comments",
        "total_recommendations",
        "max_recommendations",
        "median_recommendations",
        "editors_picks",
        "keywords",
        "publication_year",
        "sampling_section",
    ]
    base = discussions[[column for column in base_columns if column in discussions.columns]].copy()
    output = base.merge(top_summary, on="articleID", how="left").merge(
        future_summary, on="articleID", how="left"
    )
    early_effect_columns = [
        "articleID",
        "fraction_all_top_level_represented_by_early_25",
        "fraction_global_top10_recognition_from_early_25",
        "early_recognition_enrichment_ratio",
        "global_top1_recognition_in_early_25",
    ]
    output = output.merge(
        early_recognition_by_discussion(flagged)[early_effect_columns],
        on="articleID",
        how="left",
    )
    ordered_columns = [
        "articleID",
        "publication_date",
        "headline",
        "snippet",
        "sectionName",
        "newsDesk",
        "typeOfMaterial",
        "total_comments",
        "top_level_comments",
        "total_recommendations",
        "max_recommendations",
        "median_recommendations",
        "editors_picks",
        "first_comment_time",
        "last_comment_time",
        "discussion_duration",
        "future_top10_total_recommendations",
        "future_top1_recommendations",
        "median_future_top10_comment_order_pct",
        "fraction_all_top_level_represented_by_early_25",
        "fraction_global_top10_recognition_from_early_25",
        "early_recognition_enrichment_ratio",
        "global_top1_recognition_in_early_25",
        "keywords",
        "publication_year",
        "sampling_section",
    ]
    output = output[ordered_columns].sort_values(["publication_date", "articleID"], kind="mergesort")
    output.to_csv(PREDICTIVE_DISCUSSIONS_CSV, index=False)
    return output


def early_recognition_by_discussion(flagged: pd.DataFrame) -> pd.DataFrame:
    per_article = flagged.groupby("articleID").agg(
        top_level_comments=("commentID", "size"),
    ).reset_index()
    global_top10 = flagged[flagged["is_global_top_10_recognition"]].copy()
    global_top10_summary = global_top10.groupby("articleID").agg(
        global_top10_count=("commentID", "size"),
        global_top10_in_early_count=("is_early_25", "sum"),
    ).reset_index()
    global_top1 = flagged[flagged["is_global_top_1_recognition"]].copy()
    global_top1_summary = global_top1.groupby("articleID").agg(
        global_top1_in_early_count=("is_early_25", "sum")
    ).reset_index()

    per_article = per_article.merge(global_top10_summary, on="articleID", how="left").merge(
        global_top1_summary, on="articleID", how="left"
    )
    for column in ["global_top10_count", "global_top10_in_early_count", "global_top1_in_early_count"]:
        per_article[column] = per_article[column].fillna(0)
    per_article["fraction_all_top_level_represented_by_early_25"] = (
        25 / per_article["top_level_comments"]
    )
    per_article["fraction_global_top10_recognition_from_early_25"] = (
        per_article["global_top10_in_early_count"] / per_article["global_top10_count"]
    )
    per_article["early_recognition_enrichment_ratio"] = (
        per_article["fraction_global_top10_recognition_from_early_25"]
        / per_article["fraction_all_top_level_represented_by_early_25"]
    )
    per_article["global_top1_recognition_in_early_25"] = per_article[
        "global_top1_in_early_count"
    ].gt(0)
    return per_article[
        [
            "articleID",
            "top_level_comments",
            "global_top10_count",
            "global_top10_in_early_count",
            "global_top1_in_early_count",
            "fraction_all_top_level_represented_by_early_25",
            "fraction_global_top10_recognition_from_early_25",
            "early_recognition_enrichment_ratio",
            "global_top1_recognition_in_early_25",
        ]
    ]


def early_recognition_effect(flagged: pd.DataFrame) -> dict[str, float]:
    per_article = early_recognition_by_discussion(flagged)
    observed = per_article["global_top10_in_early_count"].sum() / per_article["global_top10_count"].sum()
    expected = per_article["fraction_all_top_level_represented_by_early_25"].mean()
    enrichment = observed / expected
    top1_early = per_article["global_top1_recognition_in_early_25"].mean()
    return {
        "observed_global_top10_early_pct": float(observed),
        "expected_chronological_share_pct": float(expected),
        "enrichment_ratio": float(enrichment),
        "global_top1_early_pct": float(top1_early),
    }


def structural_report(
    flagged: pd.DataFrame,
    windows: pd.DataFrame,
    discussions: pd.DataFrame,
) -> dict[str, Any]:
    counts = flagged.groupby("articleID").agg(
        early_count=("is_early_25", "sum"),
        future_count=("is_future_top_10_recognition", "sum"),
        late_count=("is_late_25", "sum"),
    )
    early_future_overlap = int(
        (flagged["is_early_25"] & flagged["is_future_top_10_recognition"]).sum()
    )
    future_late_overlap = int(
        (flagged["is_future_top_10_recognition"] & flagged["is_late_25"]).sum()
    )
    future_count_total = int(flagged["is_future_top_10_recognition"].sum())
    future = flagged[flagged["is_future_top_10_recognition"]]

    duplicate_window_ids = int(windows.duplicated(subset=["articleID", "commentID"]).sum())
    missing_window_text = int(
        (windows["commentBody"].isna() | windows["commentBody"].astype(str).str.strip().eq("")).sum()
    )
    missing_window_timestamps = int(windows["createDate"].isna().sum())
    missing_window_recommendations = int(windows["recommendations"].isna().sum())

    effect = early_recognition_effect(flagged)
    return {
        "discussion_count": len(discussions),
        "total_unique_window_comments": len(windows),
        "discussions_with_exactly_25_early": int(counts["early_count"].eq(25).sum()),
        "discussions_with_exactly_10_future": int(counts["future_count"].eq(10).sum()),
        "discussions_with_exactly_25_late": int(counts["late_count"].eq(25).sum()),
        "early_future_overlap": early_future_overlap,
        "future_late_overlap": future_late_overlap,
        "future_late_overlap_pct": future_late_overlap / future_count_total,
        "median_future_top10_comment_order_pct": float(future["comment_order_pct"].median()),
        "missing_window_text": missing_window_text,
        "missing_window_timestamps": missing_window_timestamps,
        "missing_window_recommendations": missing_window_recommendations,
        "duplicate_comment_ids_within_article": duplicate_window_ids,
        **effect,
    }


def print_report(report: dict[str, Any]) -> None:
    print("Historical predictive 120 packet")
    print(f"120 discussions: {report['discussion_count']}")
    print(f"total unique window comments: {report['total_unique_window_comments']}")
    print(f"discussions with exactly 25 EARLY comments: {report['discussions_with_exactly_25_early']}")
    print(
        "discussions with exactly 10 FUTURE_TOP_10 comments: "
        f"{report['discussions_with_exactly_10_future']}"
    )
    print(f"discussions with exactly 25 LATE comments: {report['discussions_with_exactly_25_late']}")
    print(f"EARLY_25 and FUTURE_TOP_10 overlap: {report['early_future_overlap']}")
    print(
        "FUTURE_TOP_10 and LATE_25 overlap: "
        f"{report['future_late_overlap']} ({report['future_late_overlap_pct']:.1%})"
    )
    print(
        "median chronological percentile of FUTURE_TOP_10: "
        f"{report['median_future_top10_comment_order_pct']:.3f}"
    )
    print(
        "missing text/timestamps/recommendations in predictive window rows: "
        f"{report['missing_window_text']}/{report['missing_window_timestamps']}/"
        f"{report['missing_window_recommendations']}"
    )
    print(f"duplicate comment IDs within article: {report['duplicate_comment_ids_within_article']}")
    print("early-recognition structural effect:")
    print(
        "  observed percentage of global top-10 recognition comments in EARLY_25: "
        f"{report['observed_global_top10_early_pct']:.1%}"
    )
    print(
        "  expected percentage based purely on chronological share: "
        f"{report['expected_chronological_share_pct']:.1%}"
    )
    print(f"  enrichment ratio: {report['enrichment_ratio']:.2f}")
    print(
        "  percentage of #1 global recognition comments occurring in EARLY_25: "
        f"{report['global_top1_early_pct']:.1%}"
    )
    print(f"wrote {PREDICTIVE_COMMENTS_CSV}")
    print(f"wrote {PREDICTIVE_DISCUSSIONS_CSV}")


def main() -> int:
    try:
        require_inputs()
        discussions = load_calibration_discussions()
        article_ids = set(discussions["articleID"].astype(str))
        top_comments = load_top_level_comments(article_ids)
        flagged = add_order_and_targets(top_comments)
        validate_article_windows(flagged, article_ids)
        predictive_comments = build_predictive_comments(flagged, discussions)
        build_predictive_discussions(flagged, discussions)
        report = structural_report(flagged, predictive_comments, discussions)
        print_report(report)
    except PredictivePacketError as exc:
        print(f"Stopped: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())