instaarchive-viewer/scripts/reels-scrape.py

#!/usr/bin/env python3
"""
Find a profile's reels by scrolling the real page, not calling the API.

gallery-dl's dedicated reels extractor POSTs to /api/v1/clips/user/, which
Instagram now 302-redirects to the home page for this account -- confirmed on
2026-08-26/27 across multiple profiles, hours apart, with a freshly-warmed
session and a correct X-IG-WWW-Claim header. The reels tab itself loads fine
in a real browser, so this drives the SAME logged-in Chrome (already exposed
on loopback CDP for MCP automation -- see TOOLING.md) via the DevTools
protocol, scrolls it like a person would, and scrapes `/reel/<code>/` links
out of the rendered page instead.

This only finds shortcodes; it never downloads anything itself. What comes
out the other end is deduped against the archive (via the same --index used
elsewhere) and printed as plain post URLs -- feed them to gdl-sync.py:

    ./scripts/reels-scrape.py --profile someuser \\
        --index https://instaarchive.ergosteur.com > /tmp/someuser-reels.txt

    ./scripts/gdl-sync.py --publish user@host:/path --staging /var/tmp/gdl \\
        --post-urls-file /tmp/someuser-reels.txt \\
        --sleep-request 12 20 --sleep 5 10 --rate 500K --execute

Requires the `websocket-client` package (only imported inside scrape_reels,
so everything else here stays importable -- and testable -- without it).
"""

from __future__ import annotations

import argparse
import itertools
import json
import random
import re
import sys
import time
from pathlib import Path
from urllib.request import urlopen

RE_REEL_HREF = re.compile(r"/reel/([^/?#]+)")

# The reels tab has finished loading once a scroll round finds no NEW
# shortcodes this many times in a row -- lazy-loaded feeds sometimes stall
# for a round or two before producing more, so one dry round is not enough.
DEFAULT_MAX_IDLE_ROUNDS = 3
DEFAULT_MAX_SCROLLS = 200
DEFAULT_SCROLL_PAUSE = (2.0, 3.5)

SCRAPE_JS = (
    "(() => {"
    "window.scrollTo(0, document.body.scrollHeight);"
    "return Array.from(document.querySelectorAll('a[href*=\"/reel/\"]'))"
    ".map(a => a.getAttribute('href'));"
    "})()"
)


def extract_shortcodes(hrefs: list[str]) -> list[str]:
    codes = []
    for href in hrefs:
        if m := RE_REEL_HREF.search(href):
            codes.append(m.group(1))
    return codes


class CDPError(RuntimeError):
    pass


class CDP:
    """
    A deliberately minimal synchronous DevTools Protocol client: one request
    in flight at a time, which is all a linear scroll-and-scrape loop needs.
    Anything fancier (concurrent requests, event subscriptions) is scope this
    script has no reason to carry.
    """

    def __init__(self, ws_url: str, timeout: float = 30.0):
        import websocket  # local: keep this importable without the package
        self.ws = websocket.create_connection(ws_url, timeout=timeout)
        self._ids = itertools.count(1)

    def send(self, method: str, params: dict | None = None,
             session_id: str | None = None) -> dict:
        msg_id = next(self._ids)
        payload = {"id": msg_id, "method": method, "params": params or {}}
        if session_id:
            payload["sessionId"] = session_id
        self.ws.send(json.dumps(payload))
        while True:
            msg = json.loads(self.ws.recv())
            if msg.get("id") != msg_id:
                continue  # an event notification, not our reply -- ignore
            if "error" in msg:
                raise CDPError(f"{method}: {msg['error']}")
            return msg.get("result", {})

    def close(self) -> None:
        self.ws.close()


def scrape_reels(user: str, cdp_port: int = 9222,
                 scroll_pause: tuple[float, float] = DEFAULT_SCROLL_PAUSE,
                 max_idle_rounds: int = DEFAULT_MAX_IDLE_ROUNDS,
                 max_scrolls: int = DEFAULT_MAX_SCROLLS,
                 log=lambda msg: None) -> list[str]:
    """
    Open the profile's reels tab in a NEW tab of the already-signed-in Chrome,
    scroll it to the bottom repeatedly, and collect every unique `/reel/`
    shortcode that appears. Closes the tab when done either way.
    """
    version = json.loads(urlopen(f"http://localhost:{cdp_port}/json/version",
                                 timeout=10).read())
    browser = CDP(version["webSocketDebuggerUrl"])
    target_id = None
    try:
        target = browser.send("Target.createTarget", {
            "url": f"https://www.instagram.com/{user}/reels/"})
        target_id = target["targetId"]
        attach = browser.send("Target.attachToTarget", {
            "targetId": target_id, "flatten": True})
        session_id = attach["sessionId"]
        browser.send("Page.enable", session_id=session_id)
        browser.send("Runtime.enable", session_id=session_id)

        time.sleep(4.0)  # initial page load, before the first scroll

        seen: set[str] = set()
        idle = 0
        for i in range(max_scrolls):
            result = browser.send(
                "Runtime.evaluate",
                {"expression": SCRAPE_JS, "returnByValue": True},
                session_id=session_id)
            hrefs = result.get("result", {}).get("value") or []
            codes = extract_shortcodes(hrefs)
            new = [c for c in codes if c not in seen]
            seen.update(new)
            log(f"  scroll {i + 1}: {len(seen)} unique reels so far (+{len(new)})")

            if new:
                idle = 0
            else:
                idle += 1
                if idle >= max_idle_rounds:
                    break

            time.sleep(random.uniform(*scroll_pause))

        return sorted(seen)
    finally:
        if target_id:
            try:
                browser.send("Target.closeTarget", {"targetId": target_id})
            except CDPError:
                pass  # best-effort cleanup; a leftover tab is harmless
        browser.close()


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--profile", required=True,
                    help="Instagram username to scrape reels for")
    ap.add_argument("--index", required=True,
                    help="existing archive listing, same as gdl-sync.py's "
                         "--index: a local root, or the viewer's base URL. "
                         "Used only to dedupe -- already-archived shortcodes "
                         "are dropped before anything is printed")
    ap.add_argument("--cdp-port", type=int, default=9222,
                    help="Chrome's loopback DevTools port (see TOOLING.md)")
    ap.add_argument("--scroll-pause", nargs=2, type=float,
                    default=list(DEFAULT_SCROLL_PAUSE), metavar=("MIN", "MAX"),
                    help="random pause between scrolls, seconds")
    ap.add_argument("--max-idle-rounds", type=int, default=DEFAULT_MAX_IDLE_ROUNDS,
                    help="stop after this many consecutive scrolls with no "
                         "new reels")
    ap.add_argument("--max-scrolls", type=int, default=DEFAULT_MAX_SCROLLS,
                    help="hard ceiling on scroll rounds, in case a page never "
                         "goes idle")
    ap.add_argument("--out", type=Path,
                    help="write new reel URLs here, one per line (default: "
                         "stdout)")
    args = ap.parse_args()

    import importlib.util
    spec = importlib.util.spec_from_file_location(
        "gdl_sync", Path(__file__).with_name("gdl-sync.py"))
    gdl = importlib.util.module_from_spec(spec)
    sys.modules["gdl_sync"] = gdl
    spec.loader.exec_module(gdl)

    index = gdl.ArchiveIndex(args.index)
    # Deduped against the WHOLE archive, not just this profile's own
    # directories: a shortcode is globally unique, and a reels tab commonly
    # surfaces reposts/collabs by OTHER tracked accounts. Missing that let a
    # 2026-08-27 run re-fetch 9 reels already held under their true owner's
    # directory -- gallery-dl filed them correctly there (by the post's real
    # `username`, not the scraped profile), so nothing was lost, but it spent
    # 9 avoidable Instagram requests to find that out. All of this is local
    # requests to our own viewer, never to instagram.com, so checking every
    # profile costs nothing on the budget that actually matters.
    profiles = index.profiles()
    have: set[str] = set()
    for profile in profiles:
        have.update(code for code, _ in gdl.index_existing(index.listing(profile)))
    print(f"archive already holds {len(have)} shortcode(s) across "
          f"{len(profiles)} profile(s)", file=sys.stderr)

    print(f"scraping https://www.instagram.com/{args.profile}/reels/ ...",
          file=sys.stderr)
    codes = scrape_reels(args.profile, cdp_port=args.cdp_port,
                        scroll_pause=tuple(args.scroll_pause),
                        max_idle_rounds=args.max_idle_rounds,
                        max_scrolls=args.max_scrolls,
                        log=lambda msg: print(msg, file=sys.stderr))

    new_codes = [c for c in codes if c not in have]
    print(f"found {len(codes)} reel(s) on the page, {len(codes) - len(new_codes)} "
          f"already archived, {len(new_codes)} new", file=sys.stderr)

    urls = [f"https://www.instagram.com/{args.profile}/reel/{c}/"
            for c in new_codes]
    text = "\n".join(urls)
    if args.out:
        args.out.write_text(text + ("\n" if text else ""))
        print(f"wrote {len(urls)} URL(s) to {args.out}", file=sys.stderr)
    else:
        if text:
            print(text)

    return 0


if __name__ == "__main__":
    sys.exit(main())