ergosteur/instaarchive-viewer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

tooling: instagram-chrome-scrape.py (was reels-scrape.py) gains --tab posts with early stop; chrome-sync.sh + reels/posts-sync.sh wrappers; MANUAL-RUN.md; atomic-deploy note

Browse Source
This commit is contained in:
ergosteur
2026-10-22 23:53:35 -04:00
parent 30c9fbe611
commit 1b67765d8b
10 changed files with 457 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/chrome-sync.sh
Executable
+116
View File
@@ -0,0 +1,116 @@
#!/bin/bash
# The full scrape-then-fetch pipeline for one profile: scroll the real page in
# the signed-in Chrome (instagram-chrome-scrape.py), then fetch and publish
# whatever's new as single posts. See TOOLING.md ("Reels: the API is blocked,
# scrape by scrolling instead", and 2026-10-22 for the posts grid) for why this
# exists at all -- it drives the actual signed-in browser session instead of
# the listing API, and is slower by design.
#
# Usage: GDL_TAB=reels|posts ./chrome-sync.sh <username-or-profile-url>
# Normally called through reels-sync.sh / posts-sync.sh, which set GDL_TAB.
#
# Deliberately NOT wired into gdl-cron.sh or the timers -- see TOOLING.md.
# Exits non-zero if either step does. Everything is logged.
set -eu
RAW="${1:?usage: GDL_TAB=reels|posts chrome-sync.sh <username-or-profile-url>}"
TAB="${GDL_TAB:?set GDL_TAB=reels or posts (or use reels-sync.sh / posts-sync.sh)}"
GDL_HOME="${GDL_HOME:-$HOME/gdl}"
GDL_PYTHON="${GDL_PYTHON:-$HOME/.local/share/pipx/venvs/gallery-dl/bin/python3}"
INDEX="${GDL_INDEX:-https://instaarchive.ergosteur.com}"
PUBLISH="${GDL_PUBLISH:-agentapi@10.20.28.200:/volume1/rslsync/sync/Instagram-archive/archives/}"
# Same hand-paced pacing gdl-cron.sh uses -- see its comment for why. Override
# per-run with GDL_SLEEP_REQUEST etc. if you ever need to, but raise them
# rather than lower them.
SLEEP_REQUEST="${GDL_SLEEP_REQUEST:-12 20}"
SLEEP="${GDL_SLEEP:-5 10}"
RATE="${GDL_RATE:-500K}"
SCROLL_PAUSE="${GDL_SCROLL_PAUSE:-2.0 3.5}"
MAX_IDLE_ROUNDS="${GDL_MAX_IDLE_ROUNDS:-3}"
PATH="$HOME/.local/bin:$PATH"; export PATH
if [ ! -x "$GDL_PYTHON" ]; then
echo "instagram-chrome-scrape.py needs gallery-dl's own pipx venv python (websocket-client" >&2
echo "was injected there, not into the system python): $GDL_PYTHON not found" >&2
exit 2
fi
# Same username-from-URL parsing gdl-sync.py already does for --urls-file,
# reused rather than re-implemented so the two never drift apart.
PROFILE=$("$GDL_PYTHON" -c "
import re, sys, importlib.util
from pathlib import Path
spec = importlib.util.spec_from_file_location('gdl_sync', Path('$GDL_HOME/gdl-sync.py'))
gdl = importlib.util.module_from_spec(spec)
sys.modules['gdl_sync'] = gdl
spec.loader.exec_module(gdl)
raw = '$RAW'
m = gdl.RE_PROFILE_URL.match(raw)
user = m.group('user') if m else raw.strip('/')
if not user or '/' in user or ' ' in user:
print(f'cannot read a username from {raw!r}', file=sys.stderr)
sys.exit(1)
print(user)
")
mkdir -p "$GDL_HOME/logs"
LOG="$GDL_HOME/logs/$TAB-$PROFILE-$(date +%Y%m%d-%H%M%S)-$$.log"
URLS_FILE="$GDL_HOME/$PROFILE-$TAB.txt"
STAGING="$GDL_HOME/staging-$TAB-$PROFILE"
echo "=== $TAB-sync $PROFILE $(date -Is) ===" | tee -a "$LOG"
# The exit status has to survive the pipe into tee -- see gdl-cron.sh's
# comment on PIPESTATUS for why this needs to be bash, not sh.
set +e
"$GDL_PYTHON" "$GDL_HOME/instagram-chrome-scrape.py" \
--profile "$PROFILE" \
--tab "$TAB" \
--index "$INDEX" \
--scroll-pause $SCROLL_PAUSE \
--max-idle-rounds "$MAX_IDLE_ROUNDS" \
--out "$URLS_FILE" 2>&1 | tee -a "$LOG"
scrape_status=${PIPESTATUS[0]}
set -e
if [ "$scrape_status" -ne 0 ]; then
echo "=== exit $scrape_status (scrape failed) at $(date -Is) ===" | tee -a "$LOG"
exit "$scrape_status"
fi
if [ ! -s "$URLS_FILE" ]; then
echo "no new $TAB for $PROFILE; nothing to fetch" | tee -a "$LOG"
echo "=== exit 0 at $(date -Is) ===" | tee -a "$LOG"
exit 0
fi
# Staging is wiped every run on purpose -- same reasoning as gdl-cron.sh: what
# we already hold is decided by the archive dedupe in instagram-chrome-scrape.py, not by
# what happens to be sitting in staging.
rm -rf "$STAGING"
set +e
# shellcheck disable=SC2086
"$GDL_HOME/gdl-sync.py" \
--publish "$PUBLISH" \
--staging "$STAGING" \
--post-urls-file "$URLS_FILE" \
--sleep-request $SLEEP_REQUEST \
--sleep $SLEEP \
--rate "$RATE" \
--execute 2>&1 | tee -a "$LOG"
status=${PIPESTATUS[0]}
set -e
echo "=== exit $status at $(date -Is) ===" | tee -a "$LOG"
# Keep the log directory from growing without bound -- scoped to this
# script's own logs so it never touches gdl-cron.sh's rotation.
ls -1t "$GDL_HOME/logs" | grep "^$TAB-[^0-9]" | tail -n +30 | while read -r old; do
rm -f "$GDL_HOME/logs/$old"
done
exit "$status"
scripts/gdl-sync.py
+2 -2
View File
@@ -225,7 +225,7 @@ def read_urls_file(path: Path) -> list[str]:
def read_post_urls_file(path: Path) -> list[str]:
"""
Read individual post/reel URLs from a file, one per line -- the output
format `reels-scrape.py` writes. Blank lines and `#` comments are skipped;
format `instagram-chrome-scrape.py` writes. Blank lines and `#` comments are skipped;
order is kept and duplicates dropped, same conventions as read_urls_file.
"""
urls: list[str] = []
@@ -929,7 +929,7 @@ def main() -> int:
help="file of post/reel URLs, one per line; # comments and "
"blank lines allowed. Same handling as --post-url, "
"paced with --sleep-request between items. This is "
"the format reels-scrape.py writes")
"the format instagram-chrome-scrape.py writes")
ap.add_argument("--cookies", default="chrome:/home/matt/.config/google-chrome-devtools",
help="gallery-dl --cookies-from-browser value")
ap.add_argument("--archive-db", type=Path, default=None,
scripts/reels-scrape.py → scripts/instagram-chrome-scrape.py
Regular → Executable
+102 -33
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Find a profile's reels by scrolling the real page, not calling the API.
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
@@ -11,11 +11,19 @@ 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/reels-scrape.py --profile someuser \\
./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 \\
@@ -38,7 +46,13 @@ import time
from pathlib import Path
from urllib.request import urlopen
RE_REEL_HREF = re.compile(r"/reel/([^/?#]+)")
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
@@ -46,21 +60,29 @@ RE_REEL_HREF = re.compile(r"/reel/([^/?#]+)")
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'));"
"})()"
)
# 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 extract_shortcodes(hrefs: list[str]) -> list[str]:
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_REEL_HREF.search(href):
codes.append(m.group(1))
if (m := RE_POST_HREF.search(href)) and m.group(1) in kinds:
codes.append(m.group(2))
return codes
@@ -100,23 +122,45 @@ class CDP:
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]:
def should_stop_known(round_no: int, new: list[str], have: set[str],
streak: int, max_known_rounds: int) -> tuple[bool, int]:
"""
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.
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}/reels/"})
"url": f"https://www.instagram.com/{user}/{path}"})
target_id = target["targetId"]
attach = browser.send("Target.attachToTarget", {
"targetId": target_id, "flatten": True})
@@ -128,16 +172,24 @@ def scrape_reels(user: str, cdp_port: int = 9222,
seen: set[str] = set()
idle = 0
known_streak = 0
for i in range(max_scrolls):
result = browser.send(
"Runtime.evaluate",
{"expression": SCRAPE_JS, "returnByValue": True},
{"expression": 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]
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 reels so far (+{len(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
@@ -162,7 +214,9 @@ 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")
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. "
@@ -179,6 +233,11 @@ def main() -> int:
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)")
@@ -208,20 +267,30 @@ def main() -> int:
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/ ...",
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_reels(args.profile, cdp_port=args.cdp_port,
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)} reel(s) on the page, {len(codes) - len(new_codes)} "
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)
urls = [f"https://www.instagram.com/{args.profile}/reel/{c}/"
for c in new_codes]
# /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 ""))
scripts/posts-sync.sh
Executable
+11
View File
@@ -0,0 +1,11 @@
#!/bin/bash
# chrome-sync.sh for the main profile grid: scroll /<user>/ in the signed-in
# Chrome, collect /p/ and /reel/ links, fetch only the new ones as single
# posts. A by-hand fallback for when gallery-dl's posts listing is broken
# (2026-10-22: upstream gh#9829) while single-post fetches still work -- see
# TOOLING.md. Stops scrolling once it reaches already-archived posts.
#
# Usage: ./posts-sync.sh <username-or-profile-url>
set -eu
export GDL_TAB=posts
exec "$(dirname "$0")/chrome-sync.sh" "$@"
scripts/reels-sync.sh
+4 -110
View File
@@ -1,114 +1,8 @@
#!/bin/bash
# The full reels pipeline for one profile: scrape by scrolling the real page,
# then fetch and publish whatever's new. See TOOLING.md ("Reels: the API is
# blocked, scrape by scrolling instead") for why this exists at all -- the
# dedicated reels API is blocked, this drives the actual signed-in browser
# session instead, and is slower by design.
# chrome-sync.sh for a profile's reels tab. See TOOLING.md ("Reels: the API is
# blocked, scrape by scrolling instead").
#
# Usage: ./reels-sync.sh <username-or-profile-url>
# ./reels-sync.sh zindoriyam
# ./reels-sync.sh https://www.instagram.com/zindoriyam/
#
# Deliberately NOT wired into gdl-cron.sh or the timers -- see TOOLING.md.
# Exits non-zero if either step does. Everything is logged.
set -eu
RAW="${1:?usage: reels-sync.sh <username-or-profile-url>}"
GDL_HOME="${GDL_HOME:-$HOME/gdl}"
GDL_PYTHON="${GDL_PYTHON:-$HOME/.local/share/pipx/venvs/gallery-dl/bin/python3}"
INDEX="${GDL_INDEX:-https://instaarchive.ergosteur.com}"
PUBLISH="${GDL_PUBLISH:-agentapi@10.20.28.200:/volume1/rslsync/sync/Instagram-archive/archives/}"
# Same hand-paced pacing gdl-cron.sh uses -- see its comment for why. Override
# per-run with GDL_SLEEP_REQUEST etc. if you ever need to, but raise them
# rather than lower them.
SLEEP_REQUEST="${GDL_SLEEP_REQUEST:-12 20}"
SLEEP="${GDL_SLEEP:-5 10}"
RATE="${GDL_RATE:-500K}"
SCROLL_PAUSE="${GDL_SCROLL_PAUSE:-2.0 3.5}"
MAX_IDLE_ROUNDS="${GDL_MAX_IDLE_ROUNDS:-3}"
PATH="$HOME/.local/bin:$PATH"; export PATH
if [ ! -x "$GDL_PYTHON" ]; then
echo "reels-scrape.py needs gallery-dl's own pipx venv python (websocket-client" >&2
echo "was injected there, not into the system python): $GDL_PYTHON not found" >&2
exit 2
fi
# Same username-from-URL parsing gdl-sync.py already does for --urls-file,
# reused rather than re-implemented so the two never drift apart.
PROFILE=$("$GDL_PYTHON" -c "
import re, sys, importlib.util
from pathlib import Path
spec = importlib.util.spec_from_file_location('gdl_sync', Path('$GDL_HOME/gdl-sync.py'))
gdl = importlib.util.module_from_spec(spec)
sys.modules['gdl_sync'] = gdl
spec.loader.exec_module(gdl)
raw = '$RAW'
m = gdl.RE_PROFILE_URL.match(raw)
user = m.group('user') if m else raw.strip('/')
if not user or '/' in user or ' ' in user:
print(f'cannot read a username from {raw!r}', file=sys.stderr)
sys.exit(1)
print(user)
")
mkdir -p "$GDL_HOME/logs"
LOG="$GDL_HOME/logs/reels-$PROFILE-$(date +%Y%m%d-%H%M%S)-$$.log"
URLS_FILE="$GDL_HOME/$PROFILE-reels.txt"
STAGING="$GDL_HOME/staging-reels-$PROFILE"
echo "=== reels-sync $PROFILE $(date -Is) ===" | tee -a "$LOG"
# The exit status has to survive the pipe into tee -- see gdl-cron.sh's
# comment on PIPESTATUS for why this needs to be bash, not sh.
set +e
"$GDL_PYTHON" "$GDL_HOME/reels-scrape.py" \
--profile "$PROFILE" \
--index "$INDEX" \
--scroll-pause $SCROLL_PAUSE \
--max-idle-rounds "$MAX_IDLE_ROUNDS" \
--out "$URLS_FILE" 2>&1 | tee -a "$LOG"
scrape_status=${PIPESTATUS[0]}
set -e
if [ "$scrape_status" -ne 0 ]; then
echo "=== exit $scrape_status (scrape failed) at $(date -Is) ===" | tee -a "$LOG"
exit "$scrape_status"
fi
if [ ! -s "$URLS_FILE" ]; then
echo "no new reels for $PROFILE; nothing to fetch" | tee -a "$LOG"
echo "=== exit 0 at $(date -Is) ===" | tee -a "$LOG"
exit 0
fi
# Staging is wiped every run on purpose -- same reasoning as gdl-cron.sh: what
# we already hold is decided by the archive dedupe in reels-scrape.py, not by
# what happens to be sitting in staging.
rm -rf "$STAGING"
set +e
# shellcheck disable=SC2086
"$GDL_HOME/gdl-sync.py" \
--publish "$PUBLISH" \
--staging "$STAGING" \
--post-urls-file "$URLS_FILE" \
--sleep-request $SLEEP_REQUEST \
--sleep $SLEEP \
--rate "$RATE" \
--execute 2>&1 | tee -a "$LOG"
status=${PIPESTATUS[0]}
set -e
echo "=== exit $status at $(date -Is) ===" | tee -a "$LOG"
# Keep the log directory from growing without bound -- scoped to this
# script's own logs so it never touches gdl-cron.sh's rotation.
ls -1t "$GDL_HOME/logs" | grep '^reels-' | tail -n +30 | while read -r old; do
rm -f "$GDL_HOME/logs/$old"
done
exit "$status"
export GDL_TAB=reels
exec "$(dirname "$0")/chrome-sync.sh" "$@"
scripts/test_gdl_sync.py
+1 -1
View File
@@ -326,7 +326,7 @@ class UrlsFile(unittest.TestCase):
class PostUrlsFile(unittest.TestCase):
"""The format reels-scrape.py writes: whole URLs, not usernames."""
"""The format instagram-chrome-scrape.py writes: whole URLs, not usernames."""
def test_skips_comments_blanks_and_duplicates(self):
with tempfile.TemporaryDirectory() as d:
scripts/test_instagram_chrome_scrape.py
+71
View File
@@ -0,0 +1,71 @@
#!/usr/bin/env python3
"""
Tests for the pure parts of instagram-chrome-scrape.py -- everything except the actual
CDP session, which needs a live signed-in Chrome and is exercised by hand.
python3 -m unittest discover -s scripts -p 'test_*.py'
"""
import importlib.util
import sys
import unittest
from pathlib import Path
_spec = importlib.util.spec_from_file_location(
"instagram_chrome_scrape", Path(__file__).with_name("instagram-chrome-scrape.py"))
scrape = importlib.util.module_from_spec(_spec)
sys.modules["instagram_chrome_scrape"] = scrape
_spec.loader.exec_module(scrape)
class ExtractShortcodes(unittest.TestCase):
def test_reads_relative_and_absolute_hrefs(self):
hrefs = [
"/someuser/reel/ABC123/",
"https://www.instagram.com/someuser/reel/DEF456/",
"/reel/GHI789/?img_index=1",
]
self.assertEqual(scrape.extract_shortcodes(hrefs),
["ABC123", "DEF456", "GHI789"])
def test_ignores_non_reel_links(self):
hrefs = ["/someuser/", "/someuser/p/ABC123/", "/explore/tags/foo/"]
self.assertEqual(scrape.extract_shortcodes(hrefs), [])
def test_empty_input(self):
self.assertEqual(scrape.extract_shortcodes([]), [])
def test_posts_kinds_read_both_post_and_reel_links(self):
hrefs = ["/someuser/p/ABC123/", "/p/DEF456/?img_index=2",
"/someuser/reel/GHI789/", "/explore/tags/foo/"]
self.assertEqual(scrape.extract_shortcodes(hrefs, ("p", "reel")),
["ABC123", "DEF456", "GHI789"])
class ShouldStopKnown(unittest.TestCase):
have = {"OLD1", "OLD2", "OLD3", "OLD4"}
def step(self, round_no, new, streak=0, max_known=2):
return scrape.should_stop_known(round_no, new, self.have, streak, max_known)
def test_first_round_never_counts(self):
# pinned posts: old, but on top of the grid
self.assertEqual(self.step(1, ["OLD1", "OLD2"]), (False, 0))
def test_stops_after_consecutive_known_rounds(self):
stop, streak = self.step(2, ["OLD1"])
self.assertEqual((stop, streak), (False, 1))
self.assertEqual(self.step(3, ["OLD2"], streak), (True, 2))
def test_unarchived_resets_streak(self):
self.assertEqual(self.step(3, ["OLD3", "NEW1"], streak=1), (False, 0))
def test_empty_round_leaves_streak(self):
self.assertEqual(self.step(3, [], streak=1), (False, 1))
def test_disabled(self):
self.assertEqual(self.step(5, ["OLD1"], streak=9, max_known=0), (False, 9))
if __name__ == "__main__":
unittest.main(verbosity=2)
scripts/test_reels_scrape.py
-40
View File
@@ -1,40 +0,0 @@
#!/usr/bin/env python3
"""
Tests for the pure parts of reels-scrape.py -- everything except the actual
CDP session, which needs a live signed-in Chrome and is exercised by hand.
python3 -m unittest discover -s scripts -p 'test_*.py'
"""
import importlib.util
import sys
import unittest
from pathlib import Path
_spec = importlib.util.spec_from_file_location(
"reels_scrape", Path(__file__).with_name("reels-scrape.py"))
scrape = importlib.util.module_from_spec(_spec)
sys.modules["reels_scrape"] = scrape
_spec.loader.exec_module(scrape)
class ExtractShortcodes(unittest.TestCase):
def test_reads_relative_and_absolute_hrefs(self):
hrefs = [
"/someuser/reel/ABC123/",
"https://www.instagram.com/someuser/reel/DEF456/",
"/reel/GHI789/?img_index=1",
]
self.assertEqual(scrape.extract_shortcodes(hrefs),
["ABC123", "DEF456", "GHI789"])
def test_ignores_non_reel_links(self):
hrefs = ["/someuser/", "/someuser/p/ABC123/", "/explore/tags/foo/"]
self.assertEqual(scrape.extract_shortcodes(hrefs), [])
def test_empty_input(self):
self.assertEqual(scrape.extract_shortcodes([]), [])
if __name__ == "__main__":
unittest.main(verbosity=2)