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

feat: give the sync a memory, so it stops paying for the same listing twice

Browse Source 
Nothing in this tool had any memory: every invocation started from zero
and would happily re-enumerate a profile it had listed minutes earlier.
That is what suspended the account -- the listing passes, not the
downloads -- and an aborted run re-enumerating five profiles on restart
was a large part of the bill.

Three changes, in order of how much they save:

- Seeding is now a one-time bootstrap per source. After the first
  successful sync the archive DB records everything gallery-dl has seen,
  so the source is never probed again. A second full sync costs roughly
  half what the first did.
- Stories never seed at all. A story cannot be in the archive before it
  is fetched, so there is nothing to seed from, and probing would double
  the cost of the cheapest surface we have.
- A source fetched within --min-interval (20h) is refused, and listing
  results are cached for --probe-ttl (24h), so a restart mid-run is free
  rather than a repeat. --force overrides both.

--only replaces --no-stories and takes any subset of the surfaces, which
is what makes a daily stories-only run possible: one source per profile,
no seeding, a handful of requests. Everything else stays monthly.

Tested with stdlib unittest -- no new dependencies, and it runs anywhere
the sync does. The cases include the aborted-restart scenario, which now
plans zero work.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ergosteur
2026-08-17 13:53:18 -04:00
co-authored by Claude Opus 5
parent 7a085d2272
commit 4ce5cf048e
5 changed files with 476 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/__pycache__/gdl-sync.cpython-314.pyc
BIN
View File
Binary file not shown.
scripts/__pycache__/test_gdl_sync.cpython-314.pyc
BIN
View File
Binary file not shown.
scripts/gdl-sync.py
+203 -26
View File
@@ -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.
scripts/test_gdl_sync.py
+231
View File
@@ -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)