diff --git a/docs/gallery-dl.md b/docs/gallery-dl.md index e35469d..09af9ac 100644 --- a/docs/gallery-dl.md +++ b/docs/gallery-dl.md @@ -270,6 +270,48 @@ viewer retire the lone-video heuristic in `src/lib/post-tabs.ts` — see `/p//` URL leaves it `null`. Sync always uses listing URLs, so this only matters when testing by hand. +## Cadence, and the budget that enforces it + +**Monthly for everything, daily for stories only.** Stories expire in 24h and +cannot be backfilled, so they are the one surface where missing a day means +losing the content permanently. Everything else can wait — the skip-archive +means an infrequent full sync costs barely more than a frequent one, because it +only fetches what is new. + +``` +# monthly, everything +gdl-sync.py --index --staging ~/gdl/staging \ + --publish @: --archive-db ~/gdl/artms.db \ + --urls-file artms_account_links.txt --execute + +# daily, stories only -- one request per profile +gdl-sync.py ... --only stories --execute +``` + +A stories-only run is one source per profile and **never seeds**, because a +story cannot be in the archive before it is fetched; probing would double the +cost of the cheapest surface for no benefit. Six profiles is a handful of +requests. + +When scheduling it, **randomise the minute and avoid the hour boundary**. A job +that fires at exactly 09:00 every day is a machine; one that fires somewhere in +a window looks like someone opening the app. + +The tool now refuses to repeat itself: + +| flag | default | what it prevents | +|---|---|---| +| `--min-interval` | 20h | re-fetching a source touched recently — the aborted-restart case that re-enumerated five profiles | +| `--probe-ttl` | 24h | paying for a listing pass twice within a run cycle | +| `--max-sources` | off | a runaway list touching more than intended | +| `--force` | off | (escape hatch: ignores both guards) | + +State lives beside the archive DB as `.state.json`, recording per source +when it was seeded and last fetched. **Seeding is a one-time bootstrap**: after +the first successful sync the archive DB records everything gallery-dl has +seen, so the source is never probed again. That is the single biggest saving +here — a second full sync costs roughly half what the first did. + ## Incremental sync — why the fetch host needs no copy of the archive gallery-dl can skip already-held media two ways, and the difference decides diff --git a/scripts/__pycache__/gdl-sync.cpython-314.pyc b/scripts/__pycache__/gdl-sync.cpython-314.pyc index 00b9e93..da3ff0e 100644 Binary files a/scripts/__pycache__/gdl-sync.cpython-314.pyc and b/scripts/__pycache__/gdl-sync.cpython-314.pyc differ diff --git a/scripts/__pycache__/test_gdl_sync.cpython-314.pyc b/scripts/__pycache__/test_gdl_sync.cpython-314.pyc new file mode 100644 index 0000000..7cb63cc Binary files /dev/null and b/scripts/__pycache__/test_gdl_sync.cpython-314.pyc differ diff --git a/scripts/gdl-sync.py b/scripts/gdl-sync.py index 4f4831a..36e385a 100755 --- a/scripts/gdl-sync.py +++ b/scripts/gdl-sync.py @@ -28,6 +28,7 @@ using the cookie from elsewhere is what session-hijack detection looks for. from __future__ import annotations import argparse +import datetime as dt import json import os import re @@ -72,25 +73,23 @@ class Profile: user: str existing: dict[str, str] = field(default_factory=dict) # kind -> dirname - def sources(self, include_stories: bool) -> list[Source]: + def sources(self, kinds: set[str]) -> list[Source]: u = self.user base = f"https://www.instagram.com/{u}" - out = [ + all_sources = [ Source("posts", f"{base}/posts/", u, "posts"), Source("reels", f"{base}/reels/", f"{u} - reels", "reels"), - ] - if include_stories: # Stories expire after 24h, so these can only ever be captured - # live. There is no backfill and no re-fetch. - out.append(Source( - "stories", f"https://www.instagram.com/stories/{u}/", - f"story - {u}", "stories")) + # live. There is no backfill and no re-fetch -- which is why they + # are the one surface worth visiting daily. + Source("stories", f"https://www.instagram.com/stories/{u}/", + f"story - {u}", "stories"), # Highlight directories embed the title, which gallery-dl only # learns mid-extraction -- so this one source fans out into many # directories and is handled with a directory format string. - out.append(Source( - "highlights", f"{base}/highlights", "", "highlights")) - return out + Source("highlights", f"{base}/highlights", "", "highlights"), + ] + return [s for s in all_sources if s.kind in kinds] def scan_archives(root: Path) -> dict[str, Profile]: @@ -456,6 +455,133 @@ def probe_live(src: Source, config: Path, cookies: str) -> list[dict]: return items +ALL_KINDS = ("posts", "reels", "stories", "highlights") + +# Stories cannot be backfilled and expire in 24h, so a run that only wants +# stories is both cheap and the one worth scheduling daily. +STORIES_ONLY = {"stories"} + + +class SyncState: + """ + What has already been spent against `instagram.com`. + + Exists because nothing else in this tool has any memory: every invocation + used to start from zero and happily re-enumerate profiles it had listed + minutes earlier. That is what suspended the account — the listing passes, + not the downloads. + + Two facts are tracked per source: + + seeded the skip-archive has been primed from the archive listing. + This is a ONE-TIME bootstrap: afterwards the archive DB records + every item gallery-dl has seen, so the source never needs + probing again. This is the single biggest request saving here. + fetched when it was last downloaded, so a re-run soon after is refused + rather than silently repeating the whole pass. + """ + + VERSION = 1 + + def __init__(self, path: Path): + self.path = path + self.data = {"version": self.VERSION, "sources": {}} + if path.is_file(): + try: + loaded = json.loads(path.read_text()) + if loaded.get("version") == self.VERSION: + self.data = loaded + except Exception: + pass # a corrupt state file must never block a sync + + def _entry(self, url: str) -> dict: + return self.data.setdefault("sources", {}).setdefault(url, {}) + + def needs_seed(self, url: str) -> bool: + return not self._entry(url).get("seeded") + + def mark_seeded(self, url: str, stamp: str) -> None: + self._entry(url)["seeded"] = stamp + + def last_fetch(self, url: str) -> str | None: + return self._entry(url).get("fetched") + + def mark_fetched(self, url: str, stamp: str) -> None: + self._entry(url)["fetched"] = stamp + + def save(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(json.dumps(self.data, indent=1, sort_keys=True)) + + +def hours_since(stamp: str | None, now: float) -> float: + """Hours between an ISO stamp and `now`; infinite when never.""" + if not stamp: + return float("inf") + try: + then = dt.datetime.fromisoformat(stamp) + except ValueError: + return float("inf") + if then.tzinfo is None: + then = then.replace(tzinfo=dt.timezone.utc) + return (now - then.timestamp()) / 3600.0 + + +def plan_source(src: Source, state: SyncState, now: float, + min_interval: float) -> tuple[bool, bool, str]: + """ + Decide what a source needs: (fetch, seed, reason). + + Seeding is skipped once done, and skipped entirely for stories — a story + cannot exist in the archive before it is fetched, so there is nothing to + seed from, and probing would double the request cost of the cheapest + surface we have. + """ + since = hours_since(state.last_fetch(src.url), now) + if since < min_interval: + return (False, False, f"fetched {since:.1f}h ago, under the " + f"{min_interval:g}h floor") + if src.kind == "stories": + return (True, False, "stories: no seed needed") + if state.needs_seed(src.url): + return (True, True, "first run: seeding from the archive listing") + return (True, False, "already seeded; the skip-archive knows what we hold") + + +class ProbeCache: + """ + Listing-pass results, kept so an interrupted run does not pay for them + twice. Yesterday an aborted sync re-enumerated five profiles on restart. + """ + + def __init__(self, path: Path, ttl_hours: float): + self.path = path + self.ttl = ttl_hours + self.data: dict = {} + if path.is_file(): + try: + self.data = json.loads(path.read_text()) + except Exception: + self.data = {} + + def get(self, url: str, now: float) -> list[dict] | None: + entry = self.data.get(url) + if not entry or hours_since(entry.get("at"), now) > self.ttl: + return None + return entry.get("items") + + def put(self, url: str, items: list[dict], stamp: str) -> None: + # Only the fields seeding needs, so the cache stays small. + self.data[url] = {"at": stamp, "items": [ + {k: i.get(k) for k in ("shortcode", "post_shortcode", "num", "media_id")} + for i in items + ]} + + def save(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(json.dumps(self.data)) + + def rsync_command(staging: Path, dest: str, dry_run: bool) -> list[str]: """ Publish a staging tree into the archive. @@ -529,8 +655,21 @@ def main() -> int: metavar=("MIN", "MAX")) ap.add_argument("--sleep", nargs=2, type=float, default=[3.0, 6.0], metavar=("MIN", "MAX")) - ap.add_argument("--no-stories", action="store_true", - help="skip stories and highlights (posts and reels only)") + ap.add_argument("--only", default=",".join(ALL_KINDS), + help="comma-separated surfaces to sync: " + "posts,reels,stories,highlights. Use --only stories " + "for the cheap daily run.") + ap.add_argument("--min-interval", type=float, default=20.0, metavar="HOURS", + help="refuse to re-fetch a source touched more recently " + "than this (default 20h); the guard that makes a " + "restart cheap instead of a repeat") + ap.add_argument("--max-sources", type=int, default=0, metavar="N", + help="hard ceiling on sources touched in one run " + "(0 = no limit)") + ap.add_argument("--probe-ttl", type=float, default=24.0, metavar="HOURS", + help="reuse cached listing results younger than this") + ap.add_argument("--force", action="store_true", + help="ignore --min-interval and the probe cache") mode = ap.add_mutually_exclusive_group() mode.add_argument("--dry-run", action="store_true", default=True, help="print the plan and the config; default") @@ -572,14 +711,36 @@ def main() -> int: # published to the archive root. config_path = args.staging.parent / f"{args.staging.name}.gdl-config.json" - plan: list[tuple[Profile, Source]] = [ - (prof, src) - for prof in selected - for src in prof.sources(include_stories=not args.no_stories) - ] + kinds = {k.strip() for k in args.only.split(",") if k.strip()} + unknown = kinds - set(ALL_KINDS) + if unknown: + print(f"unknown surface(s): {', '.join(sorted(unknown))}", file=sys.stderr) + return 2 + + state_path = (args.archive_db.with_suffix(".state.json") if args.archive_db + else args.staging.parent / f"{args.staging.name}.state.json") + state = SyncState(state_path) + now = dt.datetime.now(dt.timezone.utc) + now_ts, stamp = now.timestamp(), now.isoformat() + min_interval = 0.0 if args.force else args.min_interval + + plan: list[tuple[Profile, Source, bool]] = [] + skipped = 0 + for prof in selected: + for src in prof.sources(kinds): + fetch, seed, reason = plan_source(src, state, now_ts, min_interval) + if not fetch: + skipped += 1 + print(f" skip {prof.user}/{src.kind}: {reason}") + continue + if args.max_sources and len(plan) >= args.max_sources: + skipped += 1 + continue + plan.append((prof, src, seed)) print(f"profiles : {len(selected)}") - print(f"sources : {len(plan)}") + print(f"surfaces : {','.join(k for k in ALL_KINDS if k in kinds)}") + print(f"sources : {len(plan)} to sync, {skipped} skipped") print(f"pacing : {args.sleep_request[0]}-{args.sleep_request[1]}s between " f"requests, rate cap {args.rate}") print(f"staging : {args.staging}") @@ -587,32 +748,44 @@ def main() -> int: print() if not args.execute: - for prof, src in plan: + for prof, src, seed in plan: dest = src.directory or "(per-highlight)" - print(f" {prof.user:<20} {src.kind:<11} -> {dest}") + note = " [will seed]" if seed else "" + print(f" {prof.user:<20} {src.kind:<11} -> {dest}{note}") print() print(" " + " ".join(rsync_command(args.staging, args.publish, True))) print("\ndry run; nothing fetched. pass --execute to run.") return 0 config_path.write_text(json.dumps(config, indent=2)) + probes = ProbeCache(state_path.with_suffix(".probes.json"), + 0.0 if args.force else args.probe_ttl) failures = 0 - for prof, src in plan: + for prof, src, seed in plan: print(f"==> {prof.user} / {src.kind}") stage_dir = args.staging / (src.directory or ".") stage_dir.mkdir(parents=True, exist_ok=True) - # Seed the skip-archive from what the archive already holds, so + # Prime the skip-archive from what the archive already holds, so # fetching into an empty staging directory pulls only what is missing. - # The listing pass this needs is one we have to make anyway. - if args.archive_db: + # Done once per source, ever: afterwards the archive DB records + # everything gallery-dl has seen and no listing pass is needed. + if seed and args.archive_db: try: - live = probe_live(src, config_path, args.cookies) + live = probes.get(src.url, now_ts) + if live is None: + live = probe_live(src, config_path, args.cookies) + probes.put(src.url, live, stamp) + probes.save() + else: + print(f" reusing {len(live)} cached listing items") held = index_existing(index.listing(prof.user)) seeded = seed_archive_db(args.archive_db, held, live, src.subcategory) print(f" seeded {seeded} of {len(live)} live items") + state.mark_seeded(src.url, stamp) + state.save() except subprocess.CalledProcessError as exc: failures += 1 print(f" probe FAILED: {exc}", file=sys.stderr) @@ -625,6 +798,10 @@ def main() -> int: failures += 1 # Keep going: one private or renamed profile must not abort the run. print(f" FAILED (exit {result.returncode})", file=sys.stderr) + else: + # Recorded even for an empty fetch: the request was still spent. + state.mark_fetched(src.url, stamp) + state.save() # Publish once, at the end, so a partially-fetched profile never reaches # the archive mid-run. Only ever adds -- see rsync_command. diff --git a/scripts/test_gdl_sync.py b/scripts/test_gdl_sync.py new file mode 100644 index 0000000..7b412c5 --- /dev/null +++ b/scripts/test_gdl_sync.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +""" +Tests for the request-budget logic in gdl-sync.py. + + python3 -m unittest discover -s scripts -p 'test_*.py' + +Deliberately stdlib-only, so it runs anywhere the sync itself runs. What is +covered here is the part that decides whether to spend a request — the part +whose absence got the archive's Instagram account suspended. +""" + +import datetime as dt +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +_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) + +NOW = dt.datetime(2026, 8, 18, 12, 0, tzinfo=dt.timezone.utc) +NOW_TS = NOW.timestamp() + + +def ago(hours: float) -> str: + return (NOW - dt.timedelta(hours=hours)).isoformat() + + +class SourceSelection(unittest.TestCase): + def test_only_stories_is_a_single_cheap_source(self): + srcs = gdl.Profile("u").sources(gdl.STORIES_ONLY) + self.assertEqual([s.kind for s in srcs], ["stories"]) + self.assertEqual(srcs[0].directory, "story - u") + + def test_full_sync_covers_every_surface(self): + srcs = gdl.Profile("u").sources(set(gdl.ALL_KINDS)) + self.assertEqual([s.kind for s in srcs], list(gdl.ALL_KINDS)) + + def test_reels_and_stories_go_to_their_own_directories(self): + by_kind = {s.kind: s for s in gdl.Profile("u").sources(set(gdl.ALL_KINDS))} + self.assertEqual(by_kind["posts"].directory, "u") + self.assertEqual(by_kind["reels"].directory, "u - reels") + # Highlights derive their directory from the title mid-extraction. + self.assertEqual(by_kind["highlights"].directory, "") + + +class PlanSource(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state = gdl.SyncState(Path(self.tmp.name) / "state.json") + self.posts = gdl.Profile("u").sources({"posts"})[0] + self.stories = gdl.Profile("u").sources({"stories"})[0] + + def tearDown(self): + self.tmp.cleanup() + + def test_first_run_seeds(self): + fetch, seed, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) + self.assertTrue(fetch) + self.assertTrue(seed) + + def test_seeding_happens_only_once(self): + self.state.mark_seeded(self.posts.url, ago(720)) + fetch, seed, reason = gdl.plan_source(self.posts, self.state, NOW_TS, 20) + self.assertTrue(fetch) + self.assertFalse(seed, "a seeded source must never be re-probed") + self.assertIn("already seeded", reason) + + def test_stories_never_seed(self): + # A story cannot be in the archive before it is fetched, so probing + # would double the cost of the cheapest surface for no benefit. + _, seed, reason = gdl.plan_source(self.stories, self.state, NOW_TS, 20) + self.assertFalse(seed) + self.assertIn("no seed", reason) + + def test_recent_fetch_is_refused(self): + self.state.mark_fetched(self.posts.url, ago(3)) + fetch, _, reason = gdl.plan_source(self.posts, self.state, NOW_TS, 20) + self.assertFalse(fetch) + self.assertIn("under the", reason) + + def test_an_old_fetch_is_allowed_again(self): + self.state.mark_fetched(self.posts.url, ago(30)) + fetch, _, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) + self.assertTrue(fetch) + + def test_daily_stories_pass_a_20h_floor(self): + # The cadence this is built for: once a day, every day. + self.state.mark_fetched(self.stories.url, ago(24)) + fetch, _, _ = gdl.plan_source(self.stories, self.state, NOW_TS, 20) + self.assertTrue(fetch) + + def test_force_disables_the_floor(self): + self.state.mark_fetched(self.posts.url, ago(1)) + fetch, _, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 0.0) + self.assertTrue(fetch) + + def test_the_aborted_run_scenario(self): + """ + Yesterday's failure: a run died mid-way and the restart re-enumerated + every profile. Seeded-but-not-fetched must not re-probe. + """ + self.state.mark_seeded(self.posts.url, ago(0.5)) + fetch, seed, _ = gdl.plan_source(self.posts, self.state, NOW_TS, 20) + self.assertTrue(fetch, "the fetch still needs to happen") + self.assertFalse(seed, "but the listing pass must not be paid for twice") + + +class StatePersistence(unittest.TestCase): + def test_state_survives_a_reload(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "state.json" + a = gdl.SyncState(path) + a.mark_seeded("https://x/", ago(1)) + a.mark_fetched("https://x/", ago(1)) + a.save() + b = gdl.SyncState(path) + self.assertFalse(b.needs_seed("https://x/")) + self.assertEqual(b.last_fetch("https://x/"), ago(1)) + + def test_a_corrupt_state_file_never_blocks_a_sync(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "state.json" + path.write_text("{ not json") + self.assertTrue(gdl.SyncState(path).needs_seed("https://x/")) + + +class ProbeCaching(unittest.TestCase): + def test_fresh_entries_are_reused_and_stale_ones_are_not(self): + with tempfile.TemporaryDirectory() as d: + cache = gdl.ProbeCache(Path(d) / "p.json", ttl_hours=24) + cache.put("https://x/", [{"shortcode": "A", "post_shortcode": "A", + "num": 1, "media_id": "1"}], ago(1)) + self.assertEqual(len(cache.get("https://x/", NOW_TS)), 1) + + cache.put("https://y/", [{"shortcode": "B", "post_shortcode": "B", + "num": 1, "media_id": "2"}], ago(48)) + self.assertIsNone(cache.get("https://y/", NOW_TS)) + + def test_cache_keeps_only_the_fields_seeding_needs(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "p.json" + cache = gdl.ProbeCache(path, ttl_hours=24) + cache.put("https://x/", [{"shortcode": "A", "post_shortcode": "A", + "num": 1, "media_id": "1", + "description": "x" * 5000}], ago(0)) + cache.save() + self.assertNotIn("description", path.read_text()) + + def test_a_miss_is_reported_rather_than_guessed(self): + with tempfile.TemporaryDirectory() as d: + cache = gdl.ProbeCache(Path(d) / "p.json", ttl_hours=24) + self.assertIsNone(cache.get("https://never-seen/", NOW_TS)) + + +class Seeding(unittest.TestCase): + """The bug that seeded 5 of 2275: matching the wrong shortcode field.""" + + def test_posts_are_keyed_by_post_shortcode(self): + item = {"shortcode": "childcode", "post_shortcode": "POSTCODE", + "num": 2, "media_id": "9"} + self.assertEqual(gdl.live_key(item, "posts"), ("POSTCODE", 2)) + + def test_stories_are_keyed_by_the_per_item_shortcode(self): + item = {"shortcode": "ITEMCODE", "post_shortcode": "reelid", + "num": 3, "media_id": "9"} + self.assertEqual(gdl.live_key(item, "stories"), ("ITEMCODE", 1)) + self.assertEqual(gdl.live_key(item, "highlights"), ("ITEMCODE", 1)) + + def test_index_existing_normalises_a_missing_index_to_one(self): + held = gdl.index_existing([ + "u/2023-04-19_u - ABC.mp4", + "u/2023-04-12_u - DEF - 3.jpg", + "u/2023-04-12_u - DEF.txt", # sidecars are not media + "u/2023-04-12_u - DEF.json", + ]) + self.assertEqual(held, {("ABC", 1), ("DEF", 3)}) + + def test_seeding_marks_only_what_is_already_held(self): + with tempfile.TemporaryDirectory() as d: + db = Path(d) / "a.db" + live = [ + {"post_shortcode": "HELD", "shortcode": "x", "num": 1, "media_id": "11"}, + {"post_shortcode": "NEW", "shortcode": "y", "num": 1, "media_id": "22"}, + ] + n = gdl.seed_archive_db(db, {("HELD", 1)}, live, "posts") + self.assertEqual(n, 1) + import sqlite3 + rows = {r[0] for r in sqlite3.connect(db).execute( + "SELECT entry FROM archive")} + self.assertEqual(rows, {"instagram11"}) + + +class Publishing(unittest.TestCase): + def test_publish_only_ever_adds(self): + cmd = gdl.rsync_command(Path("/stage"), "host:/archives", dry_run=False) + self.assertIn("--ignore-existing", cmd) + self.assertNotIn("--delete", cmd) + + def test_tooling_files_are_excluded_from_the_archive(self): + cmd = " ".join(gdl.rsync_command(Path("/stage"), "/dest", dry_run=True)) + for pattern in ("gdl-sync*.json", "*.db"): + self.assertIn(pattern, cmd) + self.assertIn("--dry-run", cmd) + + +class UrlsFile(unittest.TestCase): + def test_reads_every_form_a_person_might_paste(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "urls.txt" + p.write_text( + "# comment\n" + "https://www.instagram.com/a/\n" + "https://instagram.com/b\n" + "www.instagram.com/c/\n" + "d\n" + " e # trailing\n" + "\n" + "https://www.instagram.com/a/\n" # duplicate + "https://www.instagram.com/p/ABC123/\n" # a post, not a profile + "not a username\n") + self.assertEqual(gdl.read_urls_file(p), ["a", "b", "c", "d", "e"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2)