#!/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//` 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) have = {code for code, _ in gdl.index_existing(index.listing(args.profile))} print(f"archive already holds {len(have)} shortcode(s) for {args.profile}", 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())