#!/usr/bin/env python3
"""
WARNING: For use on mortals only -- WIll break on tptacek and other mutant laureates
"""
import argparse
import csv
import json
import sys
import time
from pathlib import Path
import requests
API = "https://hn.algolia.com/api/v1/search_by_date"
HITS_PER_PAGE = 1000
ALGOLIA_DEEP_PAGE_LIMIT = 10000 # page * hitsPerPage must remain <= this
def fetch_page(username, page, created_after=None, created_before=None):
params = {
"tags": f"comment,author_{username}",
"hitsPerPage": HITS_PER_PAGE,
"page": page,
}
filters = []
if created_after is not None:
filters.append(f"created_at_i>{created_after}")
if created_before is not None:
filters.append(f"created_at_i<{created_before}")
if filters:
params["numericFilters"] = ",".join(filters)
resp = requests.get(API, params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def get_total_in_range(username, created_after=None, created_before=None):
data = fetch_page(username, 0, created_after, created_before)
return data["nbHits"], data
def dump_range(username, created_after, created_before, seen_ids, all_records, depth=0):
"""Recursively bisect time ranges to stay under Algolia's 10k page wall."""
total, first_page = get_total_in_range(username, created_after, created_before)
indent = " " * depth
lo = created_after if created_after is not None else "start"
hi = created_before if created_before is not None else "now"
print(f"{indent}Range [{lo} .. {hi}]: {total} comments")
if total == 0:
return
max_reachable = ALGOLIA_DEEP_PAGE_LIMIT # page*hitsPerPage cap
if total <= max_reachable:
# safe to page through directly
page = 0
while True:
data = first_page if page == 0 else fetch_page(username, page, created_after, created_before)
hits = data["hits"]
if not hits:
break
new = 0
for h in hits:
if h["objectID"] not in seen_ids:
seen_ids.add(h["objectID"])
all_records.append({
"id": h["objectID"],
"author": h.get("author"),
"created_at": h.get("created_at"),
"created_at_i": h.get("created_at_i"),
"comment_text": h.get("comment_text"),
"story_id": h.get("story_id"),
"story_title": h.get("story_title"),
"story_url": h.get("story_url"),
"parent_id": h.get("parent_id"),
"points": h.get("points"),
})
new += 1
print(f"{indent} page {page}: +{new} (running total {len(all_records)})")
if (page + 1) * HITS_PER_PAGE >= data["nbHits"]:
break
page += 1
time.sleep(0.5) # polite pacing, Be a nice human
else:
# bisect on time: find midpoint of range and recurse both halves
lo_i = created_after if created_after is not None else 0
# use "now" as a safe upper bound if open-ended
hi_i = created_before if created_before is not None else int(time.time())
mid = (lo_i + hi_i) // 2
print(f"{indent} -> exceeds {max_reachable}, bisecting at {mid}")
dump_range(username, lo_i, mid, seen_ids, all_records, depth + 1)
dump_range(username, mid, hi_i, seen_ids, all_records, depth + 1)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("username")
args = ap.parse_args()
username = args.username
seen_ids = set()
all_records = []
print(f"Fetching full comment history for '{username}'...")
dump_range(username, None, None, seen_ids, all_records)
all_records.sort(key=lambda r: r.get("created_at_i") or 0)
json_out = f"hn_comments_{username}.json"
csv_out = f"hn_comments_{username}.csv"
with open(json_out, "w", encoding="utf-8") as f:
json.dump(all_records, f, indent=2, ensure_ascii=False)
if all_records:
fieldnames = list(all_records[0].keys())
with open(csv_out, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(all_records)
print(f"\nDone. {len(all_records)} total comments.")
print(f" -> {json_out}")
print(f" -> {csv_out}")
if __name__ == "__main__":
main()