pulse.py — Mine engagement signals from your Substack network. For each profile you follow, compute author-normalized engagement lift and flag outliers. Tags via Gemini.

pulse.py

"""
pulse.py — Mine engagement signals from your Substack network.

For each profile you follow:
  1. Pull recent posts (within WINDOW_DAYS)
  2. Compute the author's typical engagement baseline (median over the window)
  3. Flag posts that significantly outperformed their author's baseline
  4. Tag outliers by topic via Gemini
  5. Output a digest: which topics are over-performing in your network right now

Run:
  python substack_social/pulse.py

Outputs:
  substack_social/data/pulse_digest.md    — human-readable report
  substack_social/data/pulse_outliers.csv — flat table of outlier posts
  substack_social/data/pulse_raw.parquet  — full crawl, for re-analysis without re-fetching
"""

import os
import sys
import json
import time
import datetime as dt
from pathlib import Path
import requests
import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
SECRETS = ROOT / "secrets.env"

# Load secrets.env into os.environ if not already set
if SECRETS.exists():
    for line in SECRETS.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        os.environ.setdefault(k.strip(), v.strip())

SID = os.environ.get("SUBSTACK_SID", "")
GEMINI_KEY = os.environ.get("GEMINI_API_KEY", "")
if not SID:
    sys.exit("ERROR: SUBSTACK_SID not set")
if not GEMINI_KEY:
    sys.exit("ERROR: GEMINI_API_KEY not set (needed for topic tagging)")

PUB_HOSTNAME = os.environ.get("SUBSTACK_HOSTNAME", "")  # e.g. "yourpub.substack.com"
WINDOW_DAYS = 90
MIN_POSTS_FOR_BASELINE = 3
OUTLIER_THRESHOLD = 2.5
MIN_POST_AGE_DAYS = 3      # posts younger than this haven't matured engagement
MIN_ENGAGEMENT_FLOOR = 20  # ignore dead authors w/ trivial absolute engagement
GEMINI_MODEL = "gemini-2.5-flash"

session = requests.Session()
session.cookies.set("substack.sid", SID, domain=".substack.com")
H = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "application/json",
}

OUT_DIR = ROOT / "substack_social" / "data"
OUT_DIR.mkdir(parents=True, exist_ok=True)


def get_own_user_id() -> int:
    """Auto-detect the authenticated user's Substack user_id.
    Tries the user-setting hack first, then falls back to pulling the
    byline id off the most recent draft on SUBSTACK_HOSTNAME.
    """
    try:
        r = session.put(
            "https://substack.com/api/v1/user-setting",
            json={"type": "last_home_tab", "value_text": "inbox"},
            headers={**H, "Content-Type": "application/json"},
            timeout=10,
        )
        if r.status_code == 200:
            return int(r.json()["user_id"])
    except Exception:
        pass
    if PUB_HOSTNAME:
        try:
            r2 = session.get(
                f"https://{PUB_HOSTNAME}/api/v1/drafts?limit=1", headers=H, timeout=10
            )
            if r2.status_code == 200:
                posts = r2.json()
                if isinstance(posts, dict):
                    posts = posts.get("posts", [])
                if posts:
                    bylines = posts[0].get("publishedBylines", [])
                    if bylines:
                        return int(bylines[0]["id"])
        except Exception:
            pass
    raise RuntimeError(
        "Could not determine own user_id. Set SUBSTACK_HOSTNAME=yourpub.substack.com in env."
    )


def get_following(self_id: int) -> list[int]:
    r = session.get(
        "https://substack.com/api/v1/feed/following?limit=500", headers=H, timeout=15
    )
    r.raise_for_status()
    return [int(i) for i in r.json() if int(i) != self_id]


def fetch_author_posts(user_id: int) -> list[dict]:
    try:
        r = session.get(
            f"https://substack.com/api/v1/profile/posts?profile_user_id={user_id}&limit=50",
            headers=H,
            timeout=15,
        )
        if r.status_code != 200:
            return []
        return r.json().get("posts", [])
    except Exception:
        return []


def engagement_score(p: dict) -> float:
    reactions = p.get("reaction_count", 0) or 0
    comments = p.get("comment_count", 0) or 0
    restacks = p.get("restacks_count") or p.get("restacks") or 0
    return float(reactions + 3 * comments + 5 * restacks)


def crawl() -> pd.DataFrame:
    cutoff = dt.datetime.utcnow() - dt.timedelta(days=WINDOW_DAYS)
    age_cutoff = dt.datetime.utcnow() - dt.timedelta(days=MIN_POST_AGE_DAYS)
    self_id = get_own_user_id()
    follows = get_following(self_id)
    print(f"Crawling {len(follows)} profiles (self user_id={self_id})...")
    rows = []
    for i, uid in enumerate(follows):
        if i and i % 20 == 0:
            print(f"  {i}/{len(follows)}  ({len(rows)} posts captured)")
        posts = fetch_author_posts(uid)
        for p in posts:
            pd_str = p.get("post_date")
            if not pd_str:
                continue
            try:
                pdate = dt.datetime.fromisoformat(
                    pd_str.replace("Z", "+00:00")
                ).replace(tzinfo=None)
            except Exception:
                continue
            if pdate < cutoff or pdate > age_cutoff:
                continue
            bylines = p.get("publishedBylines") or []
            author_name = handle = pub_name = ""
            if bylines:
                author_name = bylines[0].get("name", "") or ""
                handle = bylines[0].get("handle", "") or ""
                pubs = bylines[0].get("publicationUsers") or []
                if pubs:
                    pub_name = pubs[0].get("publication", {}).get("name", "") or ""
            rows.append(
                {
                    "author_id": uid,
                    "author_name": author_name,
                    "handle": handle,
                    "publication": pub_name,
                    "post_id": p.get("id"),
                    "title": p.get("title") or "Untitled",
                    "subtitle": p.get("subtitle") or "",
                    "snippet": p.get("truncated_body_text") or "",
                    "post_date": pdate,
                    "url": p.get("canonical_url") or "",
                    "reactions": p.get("reaction_count", 0) or 0,
                    "comments": p.get("comment_count", 0) or 0,
                    "restacks": p.get("restacks_count") or p.get("restacks") or 0,
                    "wordcount": p.get("wordcount") or 0,
                    "engagement": engagement_score(p),
                }
            )
        time.sleep(0.25)
    return pd.DataFrame(rows)


def add_baselines(df: pd.DataFrame) -> pd.DataFrame:
    baselines = df.groupby("author_id")["engagement"].median().to_dict()
    counts = df.groupby("author_id").size().to_dict()
    df["author_median"] = df["author_id"].map(baselines)
    df["author_post_count"] = df["author_id"].map(counts)
    df["lift"] = df["engagement"] / df["author_median"].replace(0, 1)
    return df


def find_outliers(df: pd.DataFrame) -> pd.DataFrame:
    out = df[
        (df["author_post_count"] >= MIN_POSTS_FOR_BASELINE)
        & (df["lift"] >= OUTLIER_THRESHOLD)
        & (df["engagement"] >= MIN_ENGAGEMENT_FLOOR)
    ].copy()
    return out.sort_values("lift", ascending=False)


def tag_topics_gemini(df: pd.DataFrame, batch_size: int = 15) -> pd.DataFrame:
    items = [
        {
            "id": int(row["post_id"]),
            "title": row["title"],
            "snippet": (row["snippet"] or "")[:240],
        }
        for _, row in df.iterrows()
    ]
    topics_by_id: dict[int, list[str]] = {}
    print(f"Tagging {len(items)} outlier posts via Gemini ({GEMINI_MODEL})...")
    for i in range(0, len(items), batch_size):
        batch = items[i : i + batch_size]
        prompt = (
            "You are tagging Substack posts by topic. For each post return 1-3 short topic tags. "
            "Tags should be:\n"
            "  - lowercase, hyphenated, 2-4 words each\n"
            "  - SPECIFIC, not generic (e.g. 'options-wheel-strategy' not 'finance'; "
            "'ai-agent-building' not 'tech'; 'trading-psychology' not 'mindset')\n"
            "Return ONLY valid JSON: an array of objects, one per post: "
            '[{"id": 123, "tags": ["tag1","tag2"]}, ...]\n\n'
            f"Posts:\n{json.dumps(batch, ensure_ascii=False)}"
        )
        url = (
            f"https://generativelanguage.googleapis.com/v1beta/models/"
            f"{GEMINI_MODEL}:generateContent?key={GEMINI_KEY}"
        )
        try:
            r = requests.post(
                url,
                json={
                    "contents": [{"parts": [{"text": prompt}]}],
                    "generationConfig": {"responseMimeType": "application/json"},
                },
                timeout=60,
            )
            if r.status_code != 200:
                print(f"  batch {i // batch_size}: HTTP {r.status_code} — {r.text[:200]}")
                continue
            text = r.json()["candidates"][0]["content"]["parts"][0]["text"]
            parsed = json.loads(text)
            for item in parsed:
                topics_by_id[int(item["id"])] = item.get("tags", []) or []
        except Exception as e:
            print(f"  batch {i // batch_size}: {e}")
        time.sleep(0.4)
    df["tags"] = df["post_id"].apply(
        lambda x: topics_by_id.get(int(x), [])
    )
    return df


def build_digest(outliers: pd.DataFrame, all_df: pd.DataFrame) -> str:
    now = dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
    lines: list[str] = []
    lines.append("# Substack Pulse — what's resonating in your network")
    lines.append("")
    lines.append(f"*Generated {now}*")
    lines.append("")
    lines.append("## Scope")
    lines.append(
        f"- Crawled {all_df['author_id'].nunique()} authors over last {WINDOW_DAYS} days"
    )
    lines.append(
        f"- {len(all_df)} posts total, {len(outliers)} flagged outliers "
        f"(≥{OUTLIER_THRESHOLD}x author baseline, ≥{MIN_ENGAGEMENT_FLOOR} engagement)"
    )
    lines.append("")

    # Topic table
    tag_rows = []
    for _, row in outliers.iterrows():
        for t in (row.get("tags") or []):
            tag_rows.append(
                {
                    "tag": t,
                    "lift": row["lift"],
                    "engagement": row["engagement"],
                    "title": row["title"],
                    "author": row["author_name"],
                }
            )
    tag_df = pd.DataFrame(tag_rows)

    if not tag_df.empty:
        tag_summary = (
            tag_df.groupby("tag")
            .agg(
                count=("tag", "size"),
                median_lift=("lift", "median"),
                total_engagement=("engagement", "sum"),
            )
            .sort_values(["count", "median_lift"], ascending=False)
        )
        lines.append("## Topics over-performing")
        lines.append("")
        lines.append("| Topic | Outlier posts | Median lift | Total engagement |")
        lines.append("|---|---:|---:|---:|")
        for tag, row in tag_summary.head(30).iterrows():
            lines.append(
                f"| {tag} | {int(row['count'])} | {row['median_lift']:.1f}x | {int(row['total_engagement'])} |"
            )
        lines.append("")

    # Top outlier posts
    lines.append("## Top 25 outlier posts")
    lines.append("")
    for _, row in outliers.head(25).iterrows():
        tags = ", ".join((row.get("tags") or [])[:3]) or "_untagged_"
        lines.append(
            f"### [{row['title']}]({row['url']})"
        )
        lines.append(
            f"by **{row['author_name']}** ({row['publication']}) — "
            f"{row['lift']:.1f}x lift · {int(row['engagement'])} engagement "
            f"({int(row['reactions'])}❤ · {int(row['comments'])}💬 · {int(row['restacks'])}🔁)"
        )
        lines.append(f"_Tags: {tags}_")
        lines.append("")
        if row["snippet"]:
            snip = row["snippet"][:240].replace("\n", " ")
            lines.append(f"> {snip}…")
            lines.append("")

    return "\n".join(lines)


def main():
    raw = crawl()
    print(f"\nCaptured {len(raw)} posts across {raw['author_id'].nunique()} authors")
    if raw.empty:
        sys.exit("No posts captured — check SID validity")

    raw = add_baselines(raw)
    outliers = find_outliers(raw)
    print(f"Found {len(outliers)} outlier posts")

    if not outliers.empty:
        outliers = tag_topics_gemini(outliers)

    digest = build_digest(outliers, raw)
    (OUT_DIR / "pulse_digest.md").write_text(digest, encoding="utf-8")
    outliers.to_csv(OUT_DIR / "pulse_outliers.csv", index=False)
    raw.to_parquet(OUT_DIR / "pulse_raw.parquet")
    print(f"\nDigest:   {OUT_DIR / 'pulse_digest.md'}")
    print(f"Outliers: {OUT_DIR / 'pulse_outliers.csv'}")
    print(f"Raw:      {OUT_DIR / 'pulse_raw.parquet'}")


if __name__ == "__main__":
    main()