From 7f8b379ae88da42fd313759ec045056a1dddd802 Mon Sep 17 00:00:00 2001 From: kevinjylin <146326281+kevinjylin@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:35:42 -0700 Subject: [PATCH] Add resolve_ids.py to fill Instagram user IDs in accounts.json Automate CommentPicker-style username to ID lookup as an offline pipeline step after discover.py, so scrape.py can keep using cached instagram_user_id values without manual copy/paste. --- pipeline/resolve_ids.py | 309 +++++++++++++++++++++++++++++ pipeline/tests/test_resolve_ids.py | 171 ++++++++++++++++ 2 files changed, 480 insertions(+) create mode 100644 pipeline/resolve_ids.py create mode 100644 pipeline/tests/test_resolve_ids.py diff --git a/pipeline/resolve_ids.py b/pipeline/resolve_ids.py new file mode 100644 index 0000000..3a93006 --- /dev/null +++ b/pipeline/resolve_ids.py @@ -0,0 +1,309 @@ +"""Fill missing Instagram user IDs in accounts.json. + +Offline step after discover.py: given handles, resolve numeric instagram_user_id +the same way tools like CommentPicker do (Instagram web_profile_info), then write +IDs back into accounts.json so scrape.py can use cached IDs. + +Usage: + python resolve_ids.py + python resolve_ids.py --dry-run + python resolve_ids.py --force + python resolve_ids.py --only acm_ucr,cyber_ucr + python resolve_ids.py --strict +""" +from __future__ import annotations + +import argparse +import json +import logging +import os +import random +import sys +import time +from pathlib import Path +from typing import Any, Callable + +import requests +from dotenv import load_dotenv + +load_dotenv(Path(__file__).resolve().parent.parent / ".env.local") +load_dotenv(Path(__file__).resolve().parent / ".env") + +from config import ACCOUNTS_FILE # noqa: E402 + +log = logging.getLogger("pipeline.resolve_ids") + +WEB_PROFILE_URL = "https://i.instagram.com/api/v1/users/web_profile_info/" +IG_APP_ID = "936619743392459" +TIMEOUT_S = 20 +JITTER_RANGE = (2.0, 5.0) +BACKOFF_S = 45.0 +MAX_ATTEMPTS = 3 + +DEFAULT_HEADERS = { + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ), + "Accept": "*/*", + "Accept-Language": "en-US,en;q=0.9", + "Origin": "https://www.instagram.com", + "Referer": "https://www.instagram.com/", + "x-ig-app-id": IG_APP_ID, +} + + +class ResolveError(Exception): + """Failed to resolve a single handle.""" + + +def _jitter() -> None: + time.sleep(random.uniform(*JITTER_RANGE)) + + +def _normalize_handle(raw: str | None) -> str | None: + if not raw: + return None + s = str(raw).strip().lstrip("@").rstrip("/").lower() + s = s.split("?", 1)[0].split("/", 1)[0] + if not s: + return None + return s + + +def _needs_id(account: dict[str, Any], *, force: bool) -> bool: + if force: + return True + return account.get("instagram_user_id") is None + + +def _parse_user_id(payload: dict[str, Any], handle: str) -> int: + user = (payload.get("data") or {}).get("user") or {} + username = str(user.get("username") or "").strip().lower() + if username and username != handle.lower(): + raise ResolveError( + f"username mismatch: requested {handle!r}, got {username!r}" + ) + raw_id = user.get("id") + if raw_id is None: + raise ResolveError("response missing data.user.id") + try: + return int(raw_id) + except (TypeError, ValueError) as e: + raise ResolveError(f"invalid user id: {raw_id!r}") from e + + +def fetch_user_id( + session: requests.Session, + handle: str, + *, + sleep_fn: Callable[[], None] | None = None, +) -> int: + """Resolve handle to numeric Instagram user id via web_profile_info.""" + last_error: Exception | None = None + for attempt in range(1, MAX_ATTEMPTS + 1): + try: + resp = session.get( + WEB_PROFILE_URL, + params={"username": handle}, + timeout=TIMEOUT_S, + ) + except (requests.ConnectionError, requests.Timeout) as e: + last_error = e + log.warning("%s: network error (%s), attempt %d/%d", handle, e, attempt, MAX_ATTEMPTS) + if attempt < MAX_ATTEMPTS: + time.sleep(BACKOFF_S if attempt > 1 else 5.0) + continue + + if resp.status_code in (429, 401, 403): + last_error = ResolveError(f"HTTP {resp.status_code}") + log.warning( + "%s: HTTP %d, backing off %ss (attempt %d/%d)", + handle, + resp.status_code, + BACKOFF_S, + attempt, + MAX_ATTEMPTS, + ) + if attempt < MAX_ATTEMPTS: + time.sleep(BACKOFF_S) + continue + + if resp.status_code == 404: + raise ResolveError("profile not found (404)") + + if resp.status_code != 200: + raise ResolveError(f"HTTP {resp.status_code}: {resp.text[:200]}") + + try: + payload = resp.json() + except ValueError as e: + raise ResolveError("response was not JSON") from e + + user_id = _parse_user_id(payload, handle) + if sleep_fn is not None: + sleep_fn() + return user_id + + raise ResolveError(f"gave up after {MAX_ATTEMPTS} attempts: {last_error}") + + +def load_accounts(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + with path.open(encoding="utf-8") as f: + return json.load(f).get("accounts", []) + + +def write_accounts(path: Path, accounts: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + payload = {"accounts": accounts} + with tmp.open("w", encoding="utf-8") as f: + json.dump(payload, f, indent=2) + f.write("\n") + os.replace(tmp, path) + + +def resolve_accounts( + accounts: list[dict[str, Any]], + *, + session: requests.Session, + force: bool = False, + only: set[str] | None = None, + dry_run: bool = False, + jitter: bool = True, + fetch_fn: Callable[[requests.Session, str], int] | None = None, +) -> tuple[list[dict[str, Any]], dict[str, int]]: + """Return (updated_accounts, stats). + + stats keys: scanned, skipped, filled, failed, unchanged + """ + fetch = fetch_fn or (lambda s, h: fetch_user_id(s, h)) + stats = {"scanned": 0, "skipped": 0, "filled": 0, "failed": 0, "unchanged": 0} + updated: list[dict[str, Any]] = [] + + for account in accounts: + entry = dict(account) + handle = _normalize_handle(entry.get("handle")) + if not handle: + updated.append(entry) + continue + + entry["handle"] = handle + stats["scanned"] += 1 + + if only is not None and handle not in only: + stats["skipped"] += 1 + updated.append(entry) + continue + + if not _needs_id(entry, force=force): + stats["unchanged"] += 1 + updated.append(entry) + continue + + try: + user_id = fetch(session, handle) + except Exception as e: # noqa: BLE001 — per-account isolation + stats["failed"] += 1 + log.error("%s: resolve failed: %s", handle, e) + updated.append(entry) + if jitter: + _jitter() + continue + + prev = entry.get("instagram_user_id") + entry["instagram_user_id"] = user_id + stats["filled"] += 1 + if prev is None: + log.info("%s: set instagram_user_id=%s", handle, user_id) + else: + log.info("%s: updated instagram_user_id %s to %s", handle, prev, user_id) + + updated.append(entry) + if jitter: + _jitter() + + if dry_run: + log.info("dry-run: not writing accounts.json (%s)", stats) + return updated, stats + + +def _parse_only(raw: str | None) -> set[str] | None: + if not raw: + return None + handles = set() + for part in raw.split(","): + h = _normalize_handle(part) + if h: + handles.add(h) + return handles or None + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Fill missing instagram_user_id values in accounts.json" + ) + parser.add_argument( + "--force", + action="store_true", + help="Re-resolve IDs even when instagram_user_id is already set", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Resolve and log, but do not write accounts.json", + ) + parser.add_argument( + "--only", + help="Comma-separated handles to resolve (default: all missing)", + ) + parser.add_argument( + "--strict", + action="store_true", + help="Exit non-zero if any targeted handle failed to resolve", + ) + parser.add_argument( + "--no-jitter", + action="store_true", + help="Skip sleep between requests (tests / local debugging only)", + ) + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + accounts = load_accounts(ACCOUNTS_FILE) + if not accounts: + log.warning("No accounts found in %s", ACCOUNTS_FILE) + return 0 + + session = requests.Session() + session.headers.update(DEFAULT_HEADERS) + + updated, stats = resolve_accounts( + accounts, + session=session, + force=args.force, + only=_parse_only(args.only), + dry_run=args.dry_run, + jitter=not args.no_jitter, + ) + + if not args.dry_run and stats["filled"] > 0: + write_accounts(ACCOUNTS_FILE, updated) + log.info("Wrote %s (%s)", ACCOUNTS_FILE, stats) + else: + log.info("Done (%s)", stats) + + if args.strict and stats["failed"] > 0: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pipeline/tests/test_resolve_ids.py b/pipeline/tests/test_resolve_ids.py new file mode 100644 index 0000000..42fbd0b --- /dev/null +++ b/pipeline/tests/test_resolve_ids.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import importlib +import json +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest import mock + + +PIPELINE_ROOT = Path(__file__).resolve().parents[1] + +if str(PIPELINE_ROOT) not in sys.path: + sys.path.insert(0, str(PIPELINE_ROOT)) + + +class ResolveIdsTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None # type: ignore[misc] + cls._tmpdir = tempfile.TemporaryDirectory() + accounts_path = Path(cls._tmpdir.name) / "accounts.json" + fake_config = types.SimpleNamespace(ACCOUNTS_FILE=accounts_path) + cls._saved_modules = { + name: sys.modules.get(name) for name in ("config", "resolve_ids", "dotenv") + } + sys.modules["dotenv"] = fake_dotenv + sys.modules["config"] = fake_config + sys.modules.pop("resolve_ids", None) + cls.resolve_ids = importlib.import_module("resolve_ids") + cls.accounts_path = accounts_path + + @classmethod + def tearDownClass(cls) -> None: + cls._tmpdir.cleanup() + for name, mod in cls._saved_modules.items(): + if mod is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = mod + + def setUp(self) -> None: + self.accounts_path.write_text( + json.dumps( + { + "accounts": [ + { + "handle": "acm_ucr", + "label": "ACM @ UCR", + "category": "club", + }, + { + "handle": "cyber_ucr", + "label": "Cyber @ UCR", + "category": "club", + "instagram_user_id": 38460809748, + }, + ] + } + ) + + "\n", + encoding="utf-8", + ) + + def test_parse_user_id_happy_path(self) -> None: + payload = {"data": {"user": {"username": "acm_ucr", "id": "10839758322"}}} + self.assertEqual(self.resolve_ids._parse_user_id(payload, "acm_ucr"), 10839758322) + + def test_parse_user_id_mismatch(self) -> None: + payload = {"data": {"user": {"username": "other", "id": "1"}}} + with self.assertRaises(self.resolve_ids.ResolveError): + self.resolve_ids._parse_user_id(payload, "acm_ucr") + + def test_fills_missing_ids(self) -> None: + accounts = self.resolve_ids.load_accounts(self.accounts_path) + + def fake_fetch(_session, handle: str) -> int: + return {"acm_ucr": 10839758322, "cyber_ucr": 999}[handle] + + updated, stats = self.resolve_ids.resolve_accounts( + accounts, + session=mock.Mock(), + force=False, + dry_run=False, + jitter=False, + fetch_fn=fake_fetch, + ) + by_handle = {a["handle"]: a for a in updated} + self.assertEqual(by_handle["acm_ucr"]["instagram_user_id"], 10839758322) + self.assertEqual(by_handle["cyber_ucr"]["instagram_user_id"], 38460809748) + self.assertEqual(stats["filled"], 1) + self.assertEqual(stats["unchanged"], 1) + + def test_skips_existing_unless_force(self) -> None: + accounts = self.resolve_ids.load_accounts(self.accounts_path) + calls: list[str] = [] + + def fake_fetch(_session, handle: str) -> int: + calls.append(handle) + return 1 + + self.resolve_ids.resolve_accounts( + accounts, + session=mock.Mock(), + force=False, + jitter=False, + fetch_fn=fake_fetch, + ) + self.assertEqual(calls, ["acm_ucr"]) + + calls.clear() + self.resolve_ids.resolve_accounts( + accounts, + session=mock.Mock(), + force=True, + jitter=False, + fetch_fn=fake_fetch, + ) + self.assertEqual(sorted(calls), ["acm_ucr", "cyber_ucr"]) + + def test_dry_run_does_not_write(self) -> None: + before = self.accounts_path.read_text(encoding="utf-8") + accounts = self.resolve_ids.load_accounts(self.accounts_path) + + updated, stats = self.resolve_ids.resolve_accounts( + accounts, + session=mock.Mock(), + dry_run=True, + jitter=False, + fetch_fn=lambda _s, _h: 10839758322, + ) + self.assertEqual(stats["filled"], 1) + self.assertEqual(updated[0]["instagram_user_id"], 10839758322) + self.assertEqual(self.accounts_path.read_text(encoding="utf-8"), before) + + def test_failure_preserves_file_contents_on_write_path(self) -> None: + accounts = self.resolve_ids.load_accounts(self.accounts_path) + + def boom(_session, handle: str) -> int: + raise self.resolve_ids.ResolveError("profile not found (404)") + + updated, stats = self.resolve_ids.resolve_accounts( + accounts, + session=mock.Mock(), + jitter=False, + fetch_fn=boom, + ) + self.assertEqual(stats["failed"], 1) + self.assertNotIn("instagram_user_id", updated[0]) + # Existing ID untouched + self.assertEqual(updated[1]["instagram_user_id"], 38460809748) + + def test_write_accounts_atomic_roundtrip(self) -> None: + accounts = [ + { + "handle": "acm_ucr", + "label": "ACM @ UCR", + "category": "club", + "instagram_user_id": 10839758322, + } + ] + self.resolve_ids.write_accounts(self.accounts_path, accounts) + loaded = self.resolve_ids.load_accounts(self.accounts_path) + self.assertEqual(loaded, accounts) + + +if __name__ == "__main__": + unittest.main()