#!/usr/bin/env python3 """Prepare the offline Kaggle NYT Comments historical validation corpus. Inputs are read-only CSV files under historical/. Outputs are normalized historical tables for later MRI work. This script does not scrape, make network requests, classify text, run sentiment/topic/embedding models, or modify source files. """ from __future__ import annotations import csv import math import re import sys from pathlib import Path from typing import Any import pandas as pd SOURCE_ROOT = Path("historical") INVENTORY_PATH = SOURCE_ROOT / "historical_inventory.csv" COMMENTS_PARQUET = SOURCE_ROOT / "comments_2017_2018_master.parquet" ARTICLES_PARQUET = SOURCE_ROOT / "articles_2017_2018_master.parquet" JOINED_PARQUET = SOURCE_ROOT / "comments_with_articles_2017_2018.parquet" DISCUSSIONS_CSV = SOURCE_ROOT / "discussions_2017_2018.csv" NOTES_PATH = SOURCE_ROOT / "HISTORICAL_DATA_NOTES.md" MONTHS = { "Jan": "January", "Feb": "February", "March": "March", "April": "April", "May": "May", } MONTH_NUMBERS = { "January": 1, "February": 2, "March": 3, "April": 4, "May": 5, } COMMENT_COLUMNS_TO_KEEP = [ "commentID", "articleID", "commentBody", "createDate", "approveDate", "recommendations", "replyCount", "editorsSelection", "parentID", "commentType", "sectionName", "userLocation", "depth", "newDesk", "articleWordCount", "printPage", "typeOfMaterial", ] ARTICLE_COLUMNS_TO_KEEP = [ "articleID", "headline", "snippet", "abstract", "pubDate", "sectionName", "newDesk", "documentType", "typeOfMaterial", "articleWordCount", "webURL", "keywords", "source", "printPage", "byline", "multimedia", ] class HistoricalPrepError(RuntimeError): """Expected preparation failure with a user-facing message.""" def parse_source_filename(path: Path) -> tuple[str, int, str, int]: match = re.fullmatch(r"(Articles|Comments)(Jan|Feb|March|April|May)(20\d{2})\.csv", path.name) if not match: raise HistoricalPrepError(f"Unrecognized source filename: {path}") prefix, month_token, year_text = match.groups() month = MONTHS[month_token] return ( "ARTICLE" if prefix == "Articles" else "COMMENT", int(year_text), month, MONTH_NUMBERS[month], ) def source_files() -> list[Path]: files = sorted(SOURCE_ROOT.glob("**/*.csv")) relevant = [] for path in files: if path.name.startswith(("Articles", "Comments")): parse_source_filename(path) relevant.append(path) if not relevant: raise HistoricalPrepError("No Kaggle article/comment CSV files found under historical/.") return relevant def read_header(path: Path) -> list[str]: with path.open(newline="", encoding="utf-8", errors="replace") as handle: return next(csv.reader(handle)) def read_csv(path: Path, *, usecols: list[str] | None = None) -> pd.DataFrame: return pd.read_csv( path, usecols=usecols, dtype=str, encoding="utf-8", low_memory=False, keep_default_na=False, ) def build_inventory(files: list[Path]) -> pd.DataFrame: rows = [] for path in files: file_type, year, month, _month_number = parse_source_filename(path) header = read_header(path) with path.open("r", encoding="utf-8", errors="replace", newline="") as handle: row_count = sum(1 for _ in csv.reader(handle)) - 1 rows.append( { "filename": str(path), "year": year, "month": month, "file_type": file_type, "row_count": row_count, "column_names": "|".join(header), "file_size": path.stat().st_size, } ) return pd.DataFrame(rows) def normalize_id(value: Any) -> str: text = "" if value is None else str(value).strip() if not text: return "" if re.fullmatch(r"\d+\.0", text): return text[:-2] return text def numeric_series(series: pd.Series) -> pd.Series: return pd.to_numeric(series.replace("", pd.NA), errors="coerce") def int_series(series: pd.Series) -> pd.Series: return numeric_series(series).round().astype("Int64") def bool_series(series: pd.Series) -> pd.Series: normalized = series.astype(str).str.strip().str.lower() return normalized.isin(["true", "1", "yes"]) def datetime_from_epoch(series: pd.Series) -> pd.Series: return pd.to_datetime(numeric_series(series), unit="s", utc=True, errors="coerce") def normalize_comment_file(path: Path) -> pd.DataFrame: file_type, year, month, month_number = parse_source_filename(path) if file_type != "COMMENT": raise HistoricalPrepError(f"Not a comment file: {path}") header = read_header(path) usecols = [column for column in COMMENT_COLUMNS_TO_KEEP if column in header] df = read_csv(path, usecols=usecols) for column in COMMENT_COLUMNS_TO_KEEP: if column not in df.columns: df[column] = pd.NA df = df[COMMENT_COLUMNS_TO_KEEP] for column in ["commentID", "articleID", "parentID"]: df[column] = df[column].map(normalize_id) for column in ["createDate", "approveDate", "recommendations", "replyCount", "depth", "articleWordCount", "printPage"]: df[column] = int_series(df[column]) df["editorsSelection"] = bool_series(df["editorsSelection"]) df["create_datetime"] = datetime_from_epoch(df["createDate"].astype("string")) df["approve_datetime"] = datetime_from_epoch(df["approveDate"].astype("string")) df["source_file"] = str(path) df["source_year"] = year df["source_month"] = month df["source_month_number"] = month_number return df def normalize_article_file(path: Path) -> pd.DataFrame: file_type, year, month, month_number = parse_source_filename(path) if file_type != "ARTICLE": raise HistoricalPrepError(f"Not an article file: {path}") header = read_header(path) usecols = [column for column in ARTICLE_COLUMNS_TO_KEEP if column in header] df = read_csv(path, usecols=usecols) for column in ARTICLE_COLUMNS_TO_KEEP: if column not in df.columns: df[column] = pd.NA df = df[ARTICLE_COLUMNS_TO_KEEP] df["articleID"] = df["articleID"].map(normalize_id) for column in ["articleWordCount", "printPage"]: df[column] = int_series(df[column]) df["publication_date"] = pd.to_datetime(df["pubDate"].replace("", pd.NA), utc=True, errors="coerce") df["article_url"] = df["webURL"] df["source_file"] = str(path) df["source_year"] = year df["source_month"] = month df["source_month_number"] = month_number return df def top_level_mask(comments: pd.DataFrame) -> pd.Series: parent_zero = comments["parentID"].fillna("").astype(str).isin(["", "0", "0.0"]) depth_one = comments["depth"].fillna(0).astype("Int64").eq(1) return parent_zero | depth_one def add_temporal_and_recognition_fields(comments: pd.DataFrame) -> pd.DataFrame: comments = comments.copy() comments["is_top_level"] = top_level_mask(comments) comments["comment_order"] = pd.Series(pd.NA, index=comments.index, dtype="Int64") comments["comment_order_pct"] = pd.Series(pd.NA, index=comments.index, dtype="Float64") comments["EARLY_25"] = False comments["EARLY_50"] = False comments["LATE_25"] = False comments["LATE_50"] = False comments["recommendation_rank"] = pd.Series(pd.NA, index=comments.index, dtype="Int64") comments["recommendation_percentile"] = pd.Series(pd.NA, index=comments.index, dtype="Float64") comments["TOP_10_RECOGNITION"] = False comments["TOP_25_RECOGNITION"] = False comments["TOP_DECILE_RECOGNITION"] = False top = comments[comments["is_top_level"]].copy() if top.empty: return comments sort_columns = ["articleID", "create_datetime", "commentID"] top = top.sort_values(sort_columns, kind="mergesort") top["comment_order"] = top.groupby("articleID").cumcount() + 1 top_counts = top.groupby("articleID")["commentID"].transform("size") top["comment_order_pct"] = top["comment_order"] / top_counts top["EARLY_25"] = top["comment_order_pct"].le(0.25) top["EARLY_50"] = top["comment_order_pct"].le(0.50) top["LATE_25"] = top["comment_order_pct"].gt(0.75) top["LATE_50"] = top["comment_order_pct"].gt(0.50) rec_values = top["recommendations"].fillna(0).astype("Int64") top["recommendation_rank"] = ( rec_values.groupby(top["articleID"]).rank(method="min", ascending=False).astype("Int64") ) top["recommendation_percentile"] = ( rec_values.groupby(top["articleID"]).rank(method="max", ascending=True, pct=True).astype("Float64") ) top["TOP_10_RECOGNITION"] = top["recommendation_rank"].le(10) top["TOP_25_RECOGNITION"] = top["recommendation_rank"].le(25) decile_cutoff = top_counts.map(lambda count: max(1, math.ceil(count * 0.10))) top["TOP_DECILE_RECOGNITION"] = top["recommendation_rank"].le(decile_cutoff) update_columns = [ "comment_order", "comment_order_pct", "EARLY_25", "EARLY_50", "LATE_25", "LATE_50", "recommendation_rank", "recommendation_percentile", "TOP_10_RECOGNITION", "TOP_25_RECOGNITION", "TOP_DECILE_RECOGNITION", ] comments.loc[top.index, update_columns] = top[update_columns] return comments def deduplicate_comments(comments: pd.DataFrame) -> tuple[pd.DataFrame, int, int]: duplicate_mask = comments["commentID"].duplicated(keep=False) duplicate_id_count = int(comments.loc[duplicate_mask, "commentID"].nunique()) duplicate_rows_removed = int(comments["commentID"].duplicated().sum()) comments = comments.copy() comments["source_files"] = comments["source_file"] if duplicate_id_count: duplicate_source_files = ( comments.loc[duplicate_mask] .groupby("commentID")["source_file"] .agg(lambda values: "|".join(sorted(set(values)))) ) comments.loc[duplicate_mask, "source_files"] = comments.loc[ duplicate_mask, "commentID" ].map(duplicate_source_files) comments = comments.sort_values( ["source_year", "source_month_number", "source_file", "commentID"], kind="mergesort", ) comments = comments.drop_duplicates(subset=["commentID"], keep="first") comments = comments.sort_index(kind="mergesort").reset_index(drop=True) return comments, duplicate_id_count, duplicate_rows_removed def article_lookup(articles: pd.DataFrame) -> pd.DataFrame: lookup = articles.sort_values(["articleID", "publication_date", "source_file"], kind="mergesort") lookup = lookup.drop_duplicates(subset=["articleID"], keep="first") rename_map = { "headline": "article_headline", "snippet": "article_snippet", "abstract": "article_abstract", "publication_date": "article_publication_date", "pubDate": "article_pubDate", "sectionName": "article_sectionName", "newDesk": "article_newDesk", "documentType": "article_documentType", "typeOfMaterial": "article_typeOfMaterial", "articleWordCount": "article_wordCount", "webURL": "article_webURL", "article_url": "article_url", "keywords": "article_keywords", } keep = ["articleID", *rename_map.keys()] lookup = lookup[[column for column in keep if column in lookup.columns]].rename(columns=rename_map) return lookup def most_common_nonempty(series: pd.Series) -> str: values = series.dropna().astype(str) values = values[values.str.len() > 0] if values.empty: return "" return values.mode().iloc[0] def build_discussions(comments: pd.DataFrame, articles: pd.DataFrame) -> pd.DataFrame: lookup = article_lookup(articles) article_ids_with_metadata = set(lookup["articleID"].dropna().astype(str)) grouped = comments.groupby("articleID", dropna=False) discussions = grouped.agg( total_comments=("commentID", "size"), top_level_comments=("is_top_level", "sum"), total_recommendations=("recommendations", "sum"), max_recommendations=("recommendations", "max"), median_recommendations=("recommendations", "median"), editors_picks=("editorsSelection", "sum"), first_comment_time=("create_datetime", "min"), last_comment_time=("create_datetime", "max"), section_from_comments=("sectionName", most_common_nonempty), ).reset_index() discussions["replies"] = discussions["total_comments"] - discussions["top_level_comments"] duration = discussions["last_comment_time"] - discussions["first_comment_time"] discussions["discussion_duration"] = duration.dt.total_seconds().astype("Float64") discussions = discussions.merge(lookup, on="articleID", how="left") discussions["article_metadata_available"] = discussions["articleID"].isin(article_ids_with_metadata).map( {True: "AVAILABLE", False: "NOT_AVAILABLE"} ) discussions["publication_date"] = discussions["article_publication_date"] discussions["headline"] = discussions["article_headline"] discussions["section"] = discussions["article_sectionName"].fillna("") discussions.loc[discussions["section"].eq(""), "section"] = discussions.loc[ discussions["section"].eq(""), "section_from_comments" ] discussions["eligible_25_plus"] = discussions["top_level_comments"].ge(25) discussions["eligible_50_plus"] = discussions["top_level_comments"].ge(50) discussions["eligible_100_plus"] = discussions["top_level_comments"].ge(100) top_decile_threshold = discussions["total_comments"].quantile(0.90) discussions["high_engagement"] = discussions["total_comments"].ge(top_decile_threshold) output_columns = [ "articleID", "publication_date", "headline", "section", "total_comments", "top_level_comments", "replies", "total_recommendations", "max_recommendations", "median_recommendations", "editors_picks", "first_comment_time", "last_comment_time", "discussion_duration", "article_metadata_available", "eligible_25_plus", "eligible_50_plus", "eligible_100_plus", "high_engagement", ] return discussions[output_columns].sort_values(["publication_date", "articleID"], kind="mergesort") def write_notes(report: dict[str, Any], warnings: list[str]) -> None: warning_text = "\n".join(f"- {warning}" for warning in warnings) if warnings else "- No major data-quality warnings." NOTES_PATH.write_text( f"""# Historical Data Notes ## Source Local downloaded Kaggle dataset: **New York Times Comments** by Aashita Kesarwani. Source files were found under `historical/archive-3/` and were left unchanged. No network requests, scraping, classification, sentiment analysis, topic modeling, embeddings, clustering, LLM labeling, or MRI scoring were performed. ## Date Range The normalized corpus covers source files for January-May 2017 and January-April 2018. Parsed comment timestamp range: - First comment: `{report['date_range_start']}` - Last comment: `{report['date_range_end']}` ## Outputs - `historical/historical_inventory.csv` - `historical/comments_2017_2018_master.parquet` - `historical/articles_2017_2018_master.parquet` - `historical/comments_with_articles_2017_2018.parquet` - `historical/discussions_2017_2018.csv` ## Comment Fields Used The normalized comment table preserves non-identity fields available in the source: `commentID`, `articleID`, `commentBody`, `createDate`, `approveDate`, `recommendations`, `replyCount`, `editorsSelection`, `parentID`, `commentType`, `sectionName`, `userLocation`, `depth`, `newDesk`, `articleWordCount`, `printPage`, and `typeOfMaterial`. The raw source includes commenter display/user fields such as `userDisplayName`, `parentUserDisplayName`, `userID`, profile URL/title, picture URL, and permanent IDs. These are not preserved in the normalized analytical parquet files because they are not required for deduplication or the planned temporal/recognition validation. ## Article Fields Used The normalized article table preserves available fields including `articleID`, `headline`, `snippet`, `abstract`, `pubDate`, parsed `publication_date`, `sectionName`, `newDesk`, `documentType`, `typeOfMaterial`, `articleWordCount`, `webURL`, standardized `article_url`, `keywords`, `source`, `printPage`, `byline`, and `multimedia`. ## Transformations - Source provenance columns were added: `source_file`, `source_year`, `source_month`, `source_month_number`, and `source_files`. - ID-like values were normalized to strings without trailing `.0` artifacts. - Repeated source rows with duplicate `commentID` values were deduplicated in the normalized analytical comment table. The retained row is the earliest source-month/source-file occurrence, and `source_files` preserves every source file where that comment appeared. - Epoch timestamp fields were preserved as integer fields where parseable and converted into UTC `create_datetime` / `approve_datetime` helper fields. - Top-level comments were identified using `parentID` equal to zero/blank or `depth` equal to 1. - Chronological position fields were computed only for top-level comments within each article: `comment_order`, `comment_order_pct`, `EARLY_25`, `EARLY_50`, `LATE_25`, and `LATE_50`. - Recognition fields were computed only for top-level comments within each article: `recommendation_rank`, `recommendation_percentile`, `TOP_10_RECOGNITION`, `TOP_25_RECOGNITION`, and `TOP_DECILE_RECOGNITION`. - Article metadata was joined onto comments by `articleID` where available. - Discussion-level rows are built from articles appearing in the comment files. ## Eligibility Flags The discussion table includes only basic, non-interpretive eligibility flags: - `eligible_25_plus`: at least 25 top-level comments - `eligible_50_plus`: at least 50 top-level comments - `eligible_100_plus`: at least 100 top-level comments - `high_engagement`: total comments at or above the historical 90th percentile ## Limitations {warning_text} """, encoding="utf-8", ) def main() -> int: try: files = source_files() inventory = build_inventory(files) inventory.to_csv(INVENTORY_PATH, index=False) article_frames = [normalize_article_file(path) for path in files if path.name.startswith("Articles")] comment_frames = [normalize_comment_file(path) for path in files if path.name.startswith("Comments")] articles = pd.concat(article_frames, ignore_index=True) comments = pd.concat(comment_frames, ignore_index=True) source_comment_rows = len(comments) comments, duplicate_comment_ids, duplicate_rows_removed = deduplicate_comments(comments) comments = add_temporal_and_recognition_fields(comments) articles.to_parquet(ARTICLES_PARQUET, index=False, engine="pyarrow") comments.to_parquet(COMMENTS_PARQUET, index=False, engine="pyarrow") lookup = article_lookup(articles) joined = comments.merge(lookup, on="articleID", how="left") joined["article_metadata_available"] = joined["article_headline"].notna() joined.to_parquet(JOINED_PARQUET, index=False, engine="pyarrow") discussions = build_discussions(comments, articles) discussions.to_csv(DISCUSSIONS_CSV, index=False) normalized_comments = len(comments) article_count = int(articles["articleID"].nunique(dropna=True)) comments_linked = int(joined["article_metadata_available"].sum()) top_level_comments = int(comments["is_top_level"].sum()) replies = int((~comments["is_top_level"]).sum()) missing_article_matches = int((~joined["article_metadata_available"]).sum()) missing_timestamps = int(comments["create_datetime"].isna().sum()) negative_recommendations = int((comments["recommendations"].fillna(0) < 0).sum()) articles_25 = int(discussions["eligible_25_plus"].sum()) articles_50 = int(discussions["eligible_50_plus"].sum()) articles_100 = int(discussions["eligible_100_plus"].sum()) high_engagement = int(discussions["high_engagement"].sum()) date_start = comments["create_datetime"].min() date_end = comments["create_datetime"].max() monthly_counts = ( comments.groupby(["source_year", "source_month_number", "source_month"]) .size() .reset_index(name="comments") .sort_values(["source_year", "source_month_number"]) ) warnings = [] if duplicate_rows_removed: warnings.append( f"{duplicate_rows_removed} repeated source rows across {duplicate_comment_ids} duplicate commentID values were deduplicated." ) if missing_article_matches: warnings.append(f"{missing_article_matches} comments did not match article metadata by articleID.") if missing_timestamps: warnings.append(f"{missing_timestamps} comments have missing/unparseable createDate timestamps.") if negative_recommendations: warnings.append(f"{negative_recommendations} comments have negative recommendation values.") if source_comment_rows != normalized_comments: warnings.append( f"Source comment row count ({source_comment_rows}) differs from normalized comments ({normalized_comments})." ) report = { "source_comment_rows": source_comment_rows, "normalized_comments": normalized_comments, "article_count": article_count, "comments_linked": comments_linked, "top_level_comments": top_level_comments, "replies": replies, "articles_25": articles_25, "articles_50": articles_50, "articles_100": articles_100, "high_engagement": high_engagement, "date_range_start": "" if pd.isna(date_start) else date_start.isoformat(), "date_range_end": "" if pd.isna(date_end) else date_end.isoformat(), "missing_article_matches": missing_article_matches, "missing_timestamps": missing_timestamps, "negative_recommendations": negative_recommendations, "duplicate_comment_ids": duplicate_comment_ids, "duplicate_rows_removed": duplicate_rows_removed, "monthly_counts": monthly_counts, } write_notes(report, warnings) print("Historical preparation report") print(f"total source comments: {source_comment_rows}") print(f"normalized comments: {normalized_comments}") print(f"deduplicated source duplicate rows: {duplicate_rows_removed}") print(f"article count: {article_count}") print(f"comments linked to articles: {comments_linked}") print(f"top-level comments: {top_level_comments}") print(f"replies: {replies}") print(f"articles with 25+ top-level comments: {articles_25}") print(f"articles with 50+ top-level comments: {articles_50}") print(f"articles with 100+ top-level comments: {articles_100}") print(f"high-engagement discussions: {high_engagement}") print(f"date range: {report['date_range_start']} to {report['date_range_end']}") print("monthly comment counts:") for row in monthly_counts.to_dict("records"): print(f" {row['source_year']}-{int(row['source_month_number']):02d}: {int(row['comments'])}") print("major data-quality warnings:") if warnings: for warning in warnings: print(f" - {warning}") else: print(" none") print(f"wrote {INVENTORY_PATH}") print(f"wrote {COMMENTS_PARQUET}") print(f"wrote {ARTICLES_PARQUET}") print(f"wrote {JOINED_PARQUET}") print(f"wrote {DISCUSSIONS_CSV}") print(f"wrote {NOTES_PATH}") except HistoricalPrepError as exc: print(f"Stopped: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())