|
#!/usr/bin/env python3 |
|
""" |
|
pull_substack_data.py — Fetch a public Substack creator's archive + notes. |
|
|
|
No auth required. Pulls everything that's publicly visible. |
|
|
|
Usage: |
|
python3 pull_substack_data.py <handle> [--posts N] [--notes M] [--bodies] |
|
|
|
Examples: |
|
python3 pull_substack_data.py claudiafaith |
|
python3 pull_substack_data.py mphinance --posts 20 --notes 30 --bodies |
|
|
|
Writes to ./<handle>_data.json. Feed the file to the prompt in |
|
research_substack_prompt.md. |
|
|
|
Requires: pip install requests |
|
""" |
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import re |
|
import sys |
|
from pathlib import Path |
|
|
|
import requests |
|
|
|
UA = "Mozilla/5.0 substack-research-dossier/1.0" |
|
TIMEOUT = 20 |
|
|
|
|
|
def get_publication_url(handle: str, session: requests.Session) -> str: |
|
"""Returns the user's primary publication URL (e.g. 'levelupwithai.substack.com').""" |
|
r = session.get(f"https://substack.com/api/v1/user/{handle}/public_profile", timeout=TIMEOUT) |
|
r.raise_for_status() |
|
profile = r.json() |
|
# primaryPublication is the canonical pub for that creator |
|
pub = profile.get("primaryPublication") or {} |
|
sub_url = pub.get("subdomain") or pub.get("custom_domain") |
|
if sub_url: |
|
return f"{sub_url}.substack.com" if "." not in sub_url else sub_url |
|
# Fall back: many handles ARE the subdomain |
|
return f"{handle}.substack.com" |
|
|
|
|
|
def get_profile(handle: str, session: requests.Session) -> dict: |
|
r = session.get(f"https://substack.com/api/v1/user/{handle}/public_profile", timeout=TIMEOUT) |
|
r.raise_for_status() |
|
profile = r.json() |
|
return { |
|
"handle": handle, |
|
"user_id": profile.get("id"), |
|
"name": profile.get("name"), |
|
"bio": profile.get("bio"), |
|
"photo_url": profile.get("photo_url"), |
|
"subscriber_count": profile.get("subscriberCount") or profile.get("subscriber_count"), |
|
"primary_publication": (profile.get("primaryPublication") or {}).get("name"), |
|
"primary_publication_url": (profile.get("primaryPublication") or {}).get("hostname"), |
|
"tld_url": (profile.get("primaryPublication") or {}).get("subdomain"), |
|
} |
|
|
|
|
|
def get_archive(pub_url: str, n: int, session: requests.Session) -> list[dict]: |
|
"""Pull the last N posts from the publication archive (public, no auth).""" |
|
out: list[dict] = [] |
|
offset = 0 |
|
while len(out) < n: |
|
limit = min(20, n - len(out)) |
|
url = f"https://{pub_url}/api/v1/archive?sort=new&offset={offset}&limit={limit}" |
|
r = session.get(url, timeout=TIMEOUT) |
|
if r.status_code != 200: |
|
break |
|
batch = r.json() |
|
if not batch: |
|
break |
|
out.extend(batch) |
|
offset += len(batch) |
|
if len(batch) < limit: |
|
break |
|
return out[:n] |
|
|
|
|
|
def get_post_body(pub_url: str, slug: str, session: requests.Session) -> dict | None: |
|
"""Try to fetch full post body. Returns None on failure.""" |
|
try: |
|
r = session.get(f"https://{pub_url}/api/v1/posts/{slug}", timeout=TIMEOUT) |
|
if r.status_code != 200: |
|
return None |
|
return r.json() |
|
except requests.RequestException: |
|
return None |
|
|
|
|
|
def strip_html(html: str) -> str: |
|
txt = re.sub(r"<[^>]+>", " ", html or "") |
|
txt = re.sub(r"\s+", " ", txt).strip() |
|
return txt |
|
|
|
|
|
def get_notes(user_id: int, n: int, session: requests.Session) -> list[dict]: |
|
"""Pull notes from public profile feed. No auth required for public notes.""" |
|
out: list[dict] = [] |
|
cursor: str | None = None |
|
while len(out) < n: |
|
url = f"https://substack.com/api/v1/reader/feed/profile/{user_id}?types=note" |
|
if cursor: |
|
url += f"&cursor={cursor}" |
|
r = session.get(url, timeout=TIMEOUT) |
|
if r.status_code != 200: |
|
break |
|
payload = r.json() |
|
items = payload.get("items", []) |
|
out.extend(items) |
|
next_cursor = payload.get("nextCursor") |
|
if not next_cursor or not items: |
|
break |
|
cursor = next_cursor |
|
# Filter to own notes (skip restacks of others) |
|
own: list[dict] = [] |
|
for item in out: |
|
users = (item.get("context", {}) or {}).get("users") or [] |
|
if any(int(u.get("id", -1)) == user_id for u in users): |
|
own.append(item) |
|
return own[:n] |
|
|
|
|
|
def extract_external_links(post_body_html: str) -> list[str]: |
|
"""Pull external (non-substack) hyperlinks from post HTML.""" |
|
if not post_body_html: |
|
return [] |
|
urls = re.findall(r'href="(https?://[^"]+)"', post_body_html) |
|
external = [u for u in urls if "substack.com" not in u and "substackcdn" not in u] |
|
return list(dict.fromkeys(external)) # dedupe, preserve order |
|
|
|
|
|
def main() -> int: |
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
|
ap.add_argument("handle", help="Substack handle (without @ or domain), e.g. 'claudiafaith'") |
|
ap.add_argument("--posts", type=int, default=10, help="Number of posts to pull (default 10)") |
|
ap.add_argument("--notes", type=int, default=20, help="Number of own-notes to pull (default 20)") |
|
ap.add_argument("--bodies", action="store_true", help="Fetch full post bodies (slower)") |
|
ap.add_argument("--out", help="Output file (default ./<handle>_data.json)") |
|
args = ap.parse_args() |
|
|
|
session = requests.Session() |
|
session.headers.update({"User-Agent": UA, "Accept": "application/json"}) |
|
|
|
print(f"Fetching profile for @{args.handle}...") |
|
profile = get_profile(args.handle, session) |
|
if not profile.get("user_id"): |
|
print(f"ERR: no user_id for @{args.handle}. Handle may not exist.", file=sys.stderr) |
|
return 1 |
|
pub_url = get_publication_url(args.handle, session) |
|
print(f" user_id={profile['user_id']}, pub={pub_url}") |
|
|
|
print(f"Fetching last {args.posts} posts...") |
|
raw_posts = get_archive(pub_url, args.posts, session) |
|
posts: list[dict] = [] |
|
for p in raw_posts: |
|
post = { |
|
"slug": p.get("slug"), |
|
"title": p.get("title"), |
|
"subtitle": p.get("subtitle"), |
|
"date": (p.get("post_date") or "")[:10], |
|
"url": p.get("canonical_url"), |
|
"audience": p.get("audience"), |
|
"hearts": (p.get("reactions") or {}).get("❤", 0), |
|
"comments": p.get("comment_count", 0), |
|
"type": p.get("type"), |
|
"external_links": [], |
|
} |
|
if args.bodies and p.get("audience") == "everyone": |
|
body = get_post_body(pub_url, post["slug"], session) |
|
if body: |
|
html = body.get("body_html") or "" |
|
post["body_text"] = strip_html(html) |
|
post["body_length"] = len(post["body_text"]) |
|
post["external_links"] = extract_external_links(html) |
|
else: |
|
post["body_text"] = p.get("truncated_body_text") or "" |
|
post["body_length"] = len(post["body_text"]) |
|
posts.append(post) |
|
|
|
print(f"Fetching last {args.notes} notes...") |
|
raw_notes = get_notes(profile["user_id"], args.notes, session) |
|
notes: list[dict] = [] |
|
for n in raw_notes: |
|
c = n.get("comment") or {} |
|
notes.append({ |
|
"note_id": c.get("id"), |
|
"posted_at": (n.get("context") or {}).get("timestamp"), |
|
"body": c.get("body") or "", |
|
"hearts": c.get("reaction_count", 0), |
|
"comments": c.get("children_count", 0), |
|
"restacks": c.get("restack_count") or c.get("restacks") or 0, |
|
}) |
|
|
|
output = { |
|
"fetched_with": "https://gist.github.com/mphinance/70b85606205398ce5a9fff81189c61a0 (substack-research-dossier)", |
|
"fetched_at_utc": None, # filled below |
|
"profile": profile, |
|
"posts": posts, |
|
"notes": notes, |
|
} |
|
from datetime import datetime, timezone |
|
output["fetched_at_utc"] = datetime.now(timezone.utc).isoformat() |
|
|
|
out_path = Path(args.out) if args.out else Path(f"{args.handle}_data.json") |
|
out_path.write_text(json.dumps(output, indent=2, ensure_ascii=False)) |
|
|
|
print(f"\nWrote {out_path}") |
|
print(f" Profile: {profile.get('name')} ({profile.get('subscriber_count')} subs)") |
|
print(f" Posts: {len(posts)} ({sum(1 for p in posts if p.get('audience') == 'everyone')} free)") |
|
print(f" Notes: {len(notes)}") |
|
print(f"\nNext step: open research_substack_prompt.md in Claude Code and run the dossier process on {out_path.name}") |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main()) |
