instaarchive-viewer/scripts/instagram-chrome-scrape.py

#!/usr/bin/env python3
"""
Find a profile's reels (or posts) 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.

`--tab posts` does the same for the main profile grid (`/<user>/`, both `/p/` and
`/reel/` links), for when gallery-dl's posts listing breaks -- on 2026-10-22,
1.32.14 returned "No results" and 1.32.17 "Requested user could not be found"
(upstream gh#9829), while single-post fetches kept working. Since the grid
runs newest-first, it stops scrolling once rounds turn up nothing but
already-archived posts (`--max-known-rounds`) instead of walking a whole
profile's history.

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/instagram-chrome-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_POST_HREF = re.compile(r"/(p|reel)/([^/?#]+)")

# tab -> (page path under the profile, link kinds to collect)
TABS = {
    "reels": ("reels/", ("reel",)),
    "posts": ("", ("p", "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)
# Posts tab only: stop after this many consecutive scroll rounds whose new
# shortcodes were all already archived. The first round is exempt -- it holds
# the pinned posts, which are old by nature and sit above the new ones.
DEFAULT_MAX_KNOWN_ROUNDS = 2


def scrape_js(kinds: tuple[str, ...]) -> str:
    selector = ", ".join(f'a[href*="/{k}/"]' for k in kinds)
    return (
        "(() => {"
        "window.scrollTo(0, document.body.scrollHeight);"
        f"return Array.from(document.querySelectorAll({json.dumps(selector)}))"
        ".map(a => a.getAttribute('href'));"
        "})()"
    )


def extract_shortcodes(hrefs: list[str],
                       kinds: tuple[str, ...] = ("reel",)) -> list[str]:
    codes = []
    for href in hrefs:
        if (m := RE_POST_HREF.search(href)) and m.group(1) in kinds:
            codes.append(m.group(2))
    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 should_stop_known(round_no: int, new: list[str], have: set[str],
                      streak: int, max_known_rounds: int) -> tuple[bool, int]:
    """
    Early-stop bookkeeping for a newest-first grid: returns (stop, streak).
    A round counts toward the streak when it found new shortcodes and every
    one is already archived; a round with anything unarchived resets it. An
    empty round leaves it alone (that is what max_idle_rounds is for), and
    round 1 never counts -- pinned posts come first.
    """
    if max_known_rounds <= 0 or round_no == 1 or not new:
        return False, streak
    if all(c in have for c in new):
        streak += 1
    else:
        streak = 0
    return streak >= max_known_rounds, streak


def scrape_grid(user: str, tab: str = "reels", 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,
                have: set[str] | None = None,
                max_known_rounds: int = 0,
                log=lambda msg: None) -> list[str]:
    """
    Open the profile's reels tab (or main grid) in a NEW tab of the
    already-signed-in Chrome, scroll it to the bottom repeatedly, and collect
    every unique shortcode that appears. Closes the tab when done either way.
    """
    path, kinds = TABS[tab]
    js = scrape_js(kinds)
    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}/{path}"})
        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
        known_streak = 0
        for i in range(max_scrolls):
            result = browser.send(
                "Runtime.evaluate",
                {"expression": js, "returnByValue": True},
                session_id=session_id)
            hrefs = result.get("result", {}).get("value") or []
            codes = extract_shortcodes(hrefs, kinds)
            new = list(dict.fromkeys(c for c in codes if c not in seen))
            seen.update(new)
            log(f"  scroll {i + 1}: {len(seen)} unique {tab} so far (+{len(new)})")

            stop, known_streak = should_stop_known(
                i + 1, new, have or set(), known_streak, max_known_rounds)
            if stop:
                log(f"  {known_streak} round(s) of already-archived {tab} in a row; "
                    f"stopping")
                break

            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")
    ap.add_argument("--tab", choices=sorted(TABS), default="reels",
                    help="reels tab, or the main posts grid (default: reels)")
    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("--max-known-rounds", type=int, default=None,
                    help="stop after this many consecutive scroll rounds of "
                         "only already-archived shortcodes; 0 disables "
                         f"(default: {DEFAULT_MAX_KNOWN_ROUNDS} for posts, "
                         "0 for reels)")
    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)

    max_known = args.max_known_rounds
    if max_known is None:
        max_known = DEFAULT_MAX_KNOWN_ROUNDS if args.tab == "posts" else 0

    path, _ = TABS[args.tab]
    print(f"scraping https://www.instagram.com/{args.profile}/{path} ...",
          file=sys.stderr)
    codes = scrape_grid(args.profile, tab=args.tab, cdp_port=args.cdp_port,
                        scroll_pause=tuple(args.scroll_pause),
                        max_idle_rounds=args.max_idle_rounds,
                        max_scrolls=args.max_scrolls,
                        have=have, max_known_rounds=max_known,
                        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)} {args.tab} on the page, {len(codes) - len(new_codes)} "
          f"already archived, {len(new_codes)} new", file=sys.stderr)

    # /p/<code>/ resolves reels too, so the posts grid needs no per-kind URL.
    if args.tab == "posts":
        urls = [f"https://www.instagram.com/p/{c}/" for c in new_codes]
    else:
        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())