#!/usr/bin/env python3 """Fetch a fixed monthly English Wikipedia pageview snapshot for the cohort.""" from __future__ import annotations import argparse import csv import datetime as dt import gzip import hashlib import io import json import os import pathlib import time import urllib.error import urllib.parse import urllib.request from collections import defaultdict API_ROOT = "https://wikimedia.org/api/rest_v1/metrics/pageviews/per-article" DEFAULT_USER_AGENT = "TrendiumoResearch/1.0 (https://trendiumo.com/contact)" def sha256_file(file: pathlib.Path) -> str: digest = hashlib.sha256() with file.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def read_cohort(file: pathlib.Path) -> list[dict[str, str]]: with file.open("r", encoding="utf-8", newline="") as stream: rows = list(csv.DictReader(stream)) required = {"item_id", "en_title", "en_page_id"} if not rows or not required.issubset(rows[0]): raise SystemExit(f"Cohort is empty or lacks columns: {sorted(required)}") return rows def endpoint( title: str, start: str, end: str, granularity: str = "daily", use_underscores: bool = True, ) -> str: # AQS stores the article path in MediaWiki URL form. Underscores are more # reliable for parenthesized and mixed-case titles than percent-encoded spaces. article = title.replace(" ", "_") if use_underscores else title encoded = urllib.parse.quote(article, safe="") return ( f"{API_ROOT}/en.wikipedia.org/all-access/all-agents/" f"{encoded}/{granularity}/{start}00/{end}00" ) def fetch_json(url: str, user_agent: str, attempts: int = 5) -> tuple[dict, int]: for attempt in range(1, attempts + 1): request = urllib.request.Request( url, headers={"User-Agent": user_agent, "Accept": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=45) as response: return json.load(response), response.status except urllib.error.HTTPError as error: if error.code == 404: # AQS occasionally returns a transient 404 for a valid page/date # combination. Retry briefly before recording an absent series. if attempt < min(attempts, 3): time.sleep(0.8 * attempt) continue return {"items": []}, 404 if error.code not in {429, 500, 502, 503, 504} or attempt == attempts: raise retry_after = error.headers.get("Retry-After") delay = float(retry_after) if retry_after and retry_after.isdigit() else min(20, 2**attempt) except (urllib.error.URLError, TimeoutError): if attempt == attempts: raise delay = min(20, 2**attempt) time.sleep(delay) raise RuntimeError("unreachable") def write_jsonl_gzip(file: pathlib.Path, rows: list[dict]) -> None: file.parent.mkdir(parents=True, exist_ok=True) with file.open("wb") as raw: with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed: with io.TextIOWrapper(compressed, encoding="utf-8", newline="\n") as text: for row in rows: text.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--cohort", default="evidence/cohort.csv") parser.add_argument( "--output", default="evidence/data/enwiki-pageviews-2025-07_2026-06.jsonl.gz", ) parser.add_argument("--metadata", default="evidence/pageview-fetch-metadata.json") parser.add_argument("--start", default="20250701") parser.add_argument("--end", default="20260630") parser.add_argument("--delay", type=float, default=0.08) parser.add_argument( "--user-agent", default=os.environ.get("WIKIMEDIA_USER_AGENT", DEFAULT_USER_AGENT), ) args = parser.parse_args() if len(args.start) != 8 or len(args.end) != 8 or not args.start.isdigit() or not args.end.isdigit(): raise SystemExit("--start and --end must be YYYYMMDD") cohort_file = pathlib.Path(args.cohort) output_file = pathlib.Path(args.output) metadata_file = pathlib.Path(args.metadata) cohort = read_cohort(cohort_file) start_date = dt.datetime.strptime(args.start, "%Y%m%d").date() end_date = dt.datetime.strptime(args.end, "%Y%m%d").date() if start_date > end_date: raise SystemExit("--start must not be after --end") broad_starts = [ dt.date(start_date.year, 1, 1).strftime("%Y%m%d"), dt.date(start_date.year - 1, 1, 1).strftime("%Y%m%d"), ] by_page: dict[str, dict] = {} output_rows = [] request_count = 0 for index, row in enumerate(cohort, start=1): page_key = row["en_page_id"] or row["en_title"] if page_key not in by_page: by_month: dict[str, int] = defaultdict(int) payload = {"items": []} api_status = 404 url = "" for broad_start in broad_starts: for granularity, use_underscores in ( ("daily", True), ("daily", False), ("monthly", True), ("monthly", False), ): url = endpoint( row["en_title"], broad_start, args.end, granularity, use_underscores, ) request_count += 1 payload, api_status = fetch_json(url, args.user_agent) if api_status == 200: break if api_status == 200: break if api_status != 200: raise SystemExit(f"Incomplete pageview series for {row['en_title']}: status={api_status}") items = payload.get("items") if not isinstance(items, list): raise SystemExit(f"Unexpected response for {row['en_title']}: items missing") for item in items: item_day = str(item["timestamp"])[:8] if args.start <= item_day <= args.end: by_month[item_day[:6]] += int(item["views"]) monthly = [ {"timestamp": month, "views": by_month[month]} for month in sorted(by_month) ] by_page[page_key] = { "requestedTitle": row["en_title"], "endpoint": url, "apiStatus": 200, "monthly": monthly, "views": sum(item["views"] for item in monthly), } if index < len(cohort): time.sleep(max(0.0, args.delay)) pageviews = by_page[page_key] output_rows.append({ "itemId": row["item_id"], "enPageId": int(row["en_page_id"]) if row["en_page_id"] else None, "enTitle": row["en_title"], "periodStart": args.start, "periodEnd": args.end, "apiStatus": pageviews["apiStatus"], "views": pageviews["views"], "monthly": pageviews["monthly"], }) output_rows.sort(key=lambda row: row["itemId"]) write_jsonl_gzip(output_file, output_rows) metadata = { "schemaVersion": 1, "source": API_ROOT, "project": "en.wikipedia.org", "access": "all-access", "agent": "all-agents", "granularity": "daily_aggregated_to_monthly", "periodStart": args.start, "periodEnd": args.end, "executedAt": dt.datetime.now(dt.timezone.utc).isoformat(), "rowCount": len(output_rows), "uniquePageCount": len(by_page), "requestCount": request_count, "queryPeriodStarts": broad_starts, "api404PageCount": sum(row["apiStatus"] == 404 for row in by_page.values()), "zeroViewPageCount": sum(row["views"] == 0 for row in by_page.values()), "sha256": sha256_file(output_file), } metadata_file.parent.mkdir(parents=True, exist_ok=True) metadata_file.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({ "rows": len(output_rows), "uniquePages": len(by_page), "api404Pages": metadata["api404PageCount"], "totalViews": sum(row["views"] for row in output_rows), "sha256": metadata["sha256"], }, ensure_ascii=False)) if __name__ == "__main__": main()