#!/usr/bin/env python3 """Parse the offline 2026 high-attention Reader Picks workbook. This script reads attention_corpus_30.xlsx locally, leaves the original workbook unchanged, and writes article-level, comment-level, and parse-quality CSVs. It does not perform classification, scoring, embeddings, clustering, or interpretation. """ from __future__ import annotations import csv import re import sys from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from typing import Any from openpyxl import load_workbook WORKBOOK_PATH = Path("attention_corpus_30.xlsx") ARTICLE_OUTPUT = Path("data/attention_articles_2026.csv") COMMENT_OUTPUT = Path("data/attention_reader_picks_2026.csv") QUALITY_OUTPUT = Path("data/attention_parse_quality_2026.csv") CAPTURE_BATCH = "2026-08-25" EXPECTED_READER_PICKS = 10 ARTICLE_COLUMNS = [ "article_id", "headline", "publication_date", "section", "article_type", "author_and_qualification", "displayed_comment_total", "article_url", "capture_batch", ] COMMENT_COLUMNS = [ "article_id", "reader_pick_rank", "comment_text", "recommended_count", "reply_count", "commenter_location", "comment_date_text", "is_reply", "reply_target_if_explicit", "source", ] QUALITY_COLUMNS = [ "article_id", "parsed_reader_picks", "expected_reader_picks", "parse_status", "missing_recommendation_counts", "missing_reply_counts", "notes", ] class AttentionParseError(RuntimeError): """Expected parser failure with a user-facing message.""" @dataclass class ParsedComment: comment_text: str recommended_count: int | None reply_count: int | None commenter_location: str comment_date_text: str is_reply: bool reply_target_if_explicit: str def clean_cell(value: Any) -> str: if value is None: return "" return str(value).strip() def clean_metadata(value: Any) -> str: return " ".join(clean_cell(value).split()) def clean_header(value: Any) -> str: return clean_metadata(value).lower() def format_date(value: Any) -> str: if isinstance(value, datetime): return value.date().isoformat() if isinstance(value, date): return value.isoformat() return clean_metadata(value) def parse_int(value: Any) -> int | None: if value is None: return None text = str(value).replace(",", "").strip() if not text: return None if re.fullmatch(r"\d+", text): return int(text) return None def normalize_lines(raw_text: Any) -> list[str]: return [line.strip() for line in str(raw_text or "").splitlines() if line.strip()] def recommend_marker(lines: list[str], index: int) -> tuple[int, int] | None: line = lines[index] match = re.fullmatch(r"([\d,]+)\s*Recommend", line) if match: return int(match.group(1).replace(",", "")), index if ( re.fullmatch(r"[\d,]+", line) and index + 1 < len(lines) and lines[index + 1] == "Recommend" ): return int(line.replace(",", "")), index + 1 return None def reply_marker(lines: list[str], index: int) -> tuple[int, int] | None: if index >= len(lines): return None line = lines[index] match = re.fullmatch(r"([\d,]+)\s*Repl(?:y|ies)", line) if match: return int(match.group(1).replace(",", "")), index if ( re.fullmatch(r"[\d,]+", line) and index + 1 < len(lines) and lines[index + 1] in {"Reply", "Replies"} ): return int(line.replace(",", "")), index + 1 return None def dateish(value: str) -> bool: value = value.strip() return bool( re.search(r"\b(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Sept|Oct|Nov|Dec)\.", value) or re.search(r"\b\d+h ago\b", value) or re.search(r"\b\d+\s+hours?\s+ago\b", value) ) def reply_header(value: str) -> bool: return "In reply thread" in value def remove_reply_header_text(value: str) -> str: return value.replace("In reply thread", "").strip() def extract_reply_target(comment_text: str) -> str: if not comment_text.startswith("@"): return "" first_line = comment_text.splitlines()[0].strip() target_text = first_line[1:].strip() if not target_text: return "" # Standalone @target lines can include spaces, e.g. "@John Virgone". if "\n" in comment_text and len(target_text) <= 80 and not re.search( r"[.!?;]", target_text ): return target_text return target_text.split()[0].rstrip(":,.;") def parse_comment_block(block_lines: list[str]) -> tuple[str, str, str, bool]: in_reply_thread = any(reply_header(line) for line in block_lines) cleaned_lines = [remove_reply_header_text(line) for line in block_lines] cleaned_lines = [line for line in cleaned_lines if line] commenter_location = "" comment_date_text = "" comment_start = 0 for index, line in enumerate(cleaned_lines): if "·" not in line: continue before, after = line.split("·", 1) commenter_location = before.strip() comment_date_text = after.strip() comment_start = index + 1 if not comment_date_text and index + 1 < len(cleaned_lines) and dateish( cleaned_lines[index + 1] ): comment_date_text = cleaned_lines[index + 1].strip() comment_start = index + 2 break if not commenter_location and not comment_date_text: while comment_start < len(cleaned_lines) and re.search( r"\bcommented\b", cleaned_lines[comment_start] ): comment_start += 1 comment_lines = cleaned_lines[comment_start:] comment_text = "\n".join(comment_lines).strip() reply_target = extract_reply_target(comment_text) is_reply = in_reply_thread or bool(reply_target) return comment_text, commenter_location, comment_date_text, is_reply def parse_reader_picks(raw_text: Any) -> list[ParsedComment]: lines = normalize_lines(raw_text) comments: list[ParsedComment] = [] position = 0 while position < len(lines): marker: tuple[int, int, int] | None = None for index in range(position, len(lines)): recommendation = recommend_marker(lines, index) if recommendation: marker = (index, recommendation[0], recommendation[1]) break if marker is None: break recommend_start, recommended_count, recommend_end = marker block_lines = lines[position:recommend_start] reply_count = 0 next_position = recommend_end + 1 reply = reply_marker(lines, next_position) if reply: reply_count = reply[0] next_position = reply[1] + 1 comment_text, location, comment_date, is_reply = parse_comment_block(block_lines) reply_target = extract_reply_target(comment_text) if comment_text: comments.append( ParsedComment( comment_text=comment_text, recommended_count=recommended_count, reply_count=reply_count, commenter_location=location, comment_date_text=comment_date, is_reply=is_reply or bool(reply_target), reply_target_if_explicit=reply_target, ) ) position = next_position return comments def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def workbook_rows() -> list[dict[str, Any]]: if not WORKBOOK_PATH.exists(): raise AttentionParseError(f"Missing workbook: {WORKBOOK_PATH}") workbook = load_workbook(WORKBOOK_PATH, read_only=True, data_only=True) worksheet = workbook.active headers = [worksheet.cell(row=1, column=col).value for col in range(1, worksheet.max_column + 1)] header_map = {clean_header(header): index + 1 for index, header in enumerate(headers)} required = [ "article id", "headline", "date", "section", "article type", "author and qualification", "displayed comments", "article url", "top 10 comments from readers' picks", ] missing = [header for header in required if header not in header_map] if missing: raise AttentionParseError(f"Workbook missing required columns: {missing}") rows: list[dict[str, Any]] = [] for row_index in range(2, worksheet.max_row + 1): row = { key: worksheet.cell(row=row_index, column=column).value for key, column in header_map.items() } if clean_metadata(row.get("article id")): rows.append(row) return rows def build_outputs() -> dict[str, Any]: article_rows: list[dict[str, Any]] = [] comment_rows: list[dict[str, Any]] = [] quality_rows: list[dict[str, Any]] = [] duplicate_text_by_article: dict[str, list[str]] = {} for row in workbook_rows(): article_id = clean_metadata(row.get("article id")) headline = clean_metadata(row.get("headline")) raw_comments = row.get("top 10 comments from readers' picks") parsed_comments = parse_reader_picks(raw_comments) article_rows.append( { "article_id": article_id, "headline": headline, "publication_date": format_date(row.get("date")), "section": clean_metadata(row.get("section")), "article_type": clean_metadata(row.get("article type")), "author_and_qualification": clean_metadata( row.get("author and qualification") ), "displayed_comment_total": parse_int(row.get("displayed comments")) or "", "article_url": clean_cell(row.get("article url")), "capture_batch": CAPTURE_BATCH, } ) seen_text: dict[str, int] = {} for rank, comment in enumerate(parsed_comments, start=1): seen_text[comment.comment_text] = seen_text.get(comment.comment_text, 0) + 1 comment_rows.append( { "article_id": article_id, "reader_pick_rank": rank, "comment_text": comment.comment_text, "recommended_count": ( "" if comment.recommended_count is None else comment.recommended_count ), "reply_count": "" if comment.reply_count is None else comment.reply_count, "commenter_location": comment.commenter_location, "comment_date_text": comment.comment_date_text, "is_reply": "TRUE" if comment.is_reply else "FALSE", "reply_target_if_explicit": comment.reply_target_if_explicit, "source": "READER_PICKS", } ) duplicates = sorted(text for text, count in seen_text.items() if count > 1) if duplicates: duplicate_text_by_article[article_id] = duplicates missing_recommendations = sum( 1 for comment in parsed_comments if comment.recommended_count is None ) missing_replies = sum(1 for comment in parsed_comments if comment.reply_count is None) parse_status = ( "OK" if len(parsed_comments) == EXPECTED_READER_PICKS and not missing_recommendations and not missing_replies else "REVIEW" ) notes: list[str] = [] if len(parsed_comments) != EXPECTED_READER_PICKS: notes.append( f"parsed {len(parsed_comments)} surfaced items; expected {EXPECTED_READER_PICKS}" ) if missing_recommendations: notes.append("missing recommendation count in one or more parsed items") if missing_replies: notes.append("missing reply count in one or more parsed items") if duplicates: notes.append("duplicated comment text within article") quality_rows.append( { "article_id": article_id, "parsed_reader_picks": len(parsed_comments), "expected_reader_picks": EXPECTED_READER_PICKS, "parse_status": parse_status, "missing_recommendation_counts": missing_recommendations, "missing_reply_counts": missing_replies, "notes": "; ".join(notes), } ) write_csv(ARTICLE_OUTPUT, ARTICLE_COLUMNS, article_rows) write_csv(COMMENT_OUTPUT, COMMENT_COLUMNS, comment_rows) write_csv(QUALITY_OUTPUT, QUALITY_COLUMNS, quality_rows) article_count = len(article_rows) total_reader_picks = len(comment_rows) exactly_10 = sum(1 for row in quality_rows if row["parsed_reader_picks"] == 10) fewer_than_10 = sum(1 for row in quality_rows if row["parsed_reader_picks"] < 10) more_than_10 = sum(1 for row in quality_rows if row["parsed_reader_picks"] > 10) total_reply = sum(1 for row in comment_rows if row["is_reply"] == "TRUE") return { "articles_parsed": article_count, "total_reader_picks": total_reader_picks, "articles_exactly_10": exactly_10, "articles_fewer_than_10": fewer_than_10, "articles_more_than_10": more_than_10, "total_reply_reader_picks": total_reply, "total_top_level_reader_picks": total_reader_picks - total_reply, "duplicated_comment_text_within_article": { key: len(value) for key, value in duplicate_text_by_article.items() }, "missing_recommendation_counts": sum( int(row["missing_recommendation_counts"]) for row in quality_rows ), "review_articles": [ row["article_id"] for row in quality_rows if row["parse_status"] != "OK" ], } def main() -> int: try: report = build_outputs() except AttentionParseError as exc: print(f"Stopped: {exc}", file=sys.stderr) return 1 print("Attention corpus parsing report") print(f"articles parsed: {report['articles_parsed']}") print(f"total Reader Picks parsed: {report['total_reader_picks']}") print(f"articles with exactly 10: {report['articles_exactly_10']}") print(f"articles with fewer than 10: {report['articles_fewer_than_10']}") print(f"articles with more than 10: {report['articles_more_than_10']}") print(f"total top-level Reader Picks: {report['total_top_level_reader_picks']}") print(f"total reply Reader Picks: {report['total_reply_reader_picks']}") print( "duplicated comment text within same article: " f"{report['duplicated_comment_text_within_article'] or 'none'}" ) print(f"missing recommendation counts: {report['missing_recommendation_counts']}") print(f"articles flagged for review: {report['review_articles'] or 'none'}") print(f"wrote {ARTICLE_OUTPUT}") print(f"wrote {COMMENT_OUTPUT}") print(f"wrote {QUALITY_OUTPUT}") return 0 if __name__ == "__main__": raise SystemExit(main())