3 ms·

I am

• eth0up 3mo ago
  #!/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 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()