From edacc1477737b5ee6cf7f51b06d4b1db9b26846a Mon Sep 17 00:00:00 2001 From: kevinjylin Date: Tue, 13 Oct 2026 14:47:27 -0700 Subject: [PATCH] Stop rebilling completed Instagram recheck batches --- pipeline/README.md | 5 + pipeline/apify_posts.py | 5 +- pipeline/post_recheck.py | 78 ++++++---- pipeline/tests/test_post_recheck.py | 217 ++++++++++++++++++++++++++++ 4 files changed, 276 insertions(+), 29 deletions(-) diff --git a/pipeline/README.md b/pipeline/README.md index f024f579..5cdab8d2 100644 --- a/pipeline/README.md +++ b/pipeline/README.md @@ -413,3 +413,8 @@ over half of a batch reports deleted is treated as an outage and changes nothing Withdrawn sources carry `assessment.deleted`, which publication skips so the archived copy is not republished. Progress is kept in `data/post_rechecks.json`. A post is checked once, so a deletion after the check is not seen. +Each batch saves its check results before marking the paid run consumed. +An interrupted batch reuses its original run and post list; completed batches +use the next due list. Replaying a run cannot count as a second deletion +confirmation. Empty or unrelated output stops further batches and retains the +paid run for investigation rather than silently buying the same lookup again. diff --git a/pipeline/apify_posts.py b/pipeline/apify_posts.py index ea611d4d..841242f2 100644 --- a/pipeline/apify_posts.py +++ b/pipeline/apify_posts.py @@ -154,7 +154,10 @@ def details(self, codes: list[str], *, max_charge: float, timeout: int) -> tuple # This intent stays unconsumed even for FAILED runs: replacing it within # a batch could spend its reserved detail ceiling twice. A fresh plan's # cycle-specific filename permits a new attempt for incomplete profiles. - codes = saved.get("usernames", sorted(codes)) + if saved and not saved.get("consumed"): + codes = saved.get("usernames", sorted(codes)) + else: + codes = sorted(codes) client = ApifyClient(self.session.headers["Authorization"].removeprefix("Bearer "), path) run = client._run(hpix_input(posts=[f"https://www.instagram.com/p/{code}/" for code in codes]), {"usernames": codes, "actor_id": HPIX_ACTOR_ID, diff --git a/pipeline/post_recheck.py b/pipeline/post_recheck.py index bd2f393e..79c5dd83 100644 --- a/pipeline/post_recheck.py +++ b/pipeline/post_recheck.py @@ -120,7 +120,12 @@ def main() -> None: registry = assessed_events.load_registry() state = read_json(STATE_FILE) if STATE_FILE.exists() else {} due = due_sources(events, registry, state, now) - if not due: + details_file = RUN_FILE.with_name(RUN_FILE.stem + "-details.json") + saved = read_json(details_file) if details_file.exists() else {} + # Finish already-paid work before buying a new batch, even if its events + # have expired or its check state was saved just before an interruption. + recovery = saved.get("usernames", []) if saved and not saved.get("consumed") else [] + if not due and not recovery: log.info("post recheck: nothing due") return media_ids = [key.rsplit(":", 1)[-1] for key in due] @@ -131,12 +136,12 @@ def main() -> None: token = os.environ.get("APIFY_TOKEN", "") api = ApifyClient(token, RUN_FILE) results: dict[str, str] = {} - pending = sorted(by_code) - for i in range(0, len(pending), BATCH_SIZE): - batch = pending[i:i + BATCH_SIZE] + withdrawn = 0 + pending = sorted(set(by_code) - set(recovery)) + batches = ([recovery] if recovery else []) + [ + pending[i:i + BATCH_SIZE] for i in range(0, len(pending), BATCH_SIZE)] + for batch in batches: run, rows = api.details(batch, max_charge=MAX_CHARGE_USD, timeout=600) - details_file = RUN_FILE.with_name(RUN_FILE.stem + "-details.json") - write_json(details_file, {**read_json(details_file), "consumed": True}) # Count against the lookup's own ceiling; a charge-limited run may # have stopped before reaching every post, so nothing can be concluded. if run.get("status") != "SUCCEEDED": @@ -144,33 +149,50 @@ def main() -> None: reason = charge_limit_reason(run, MAX_CHARGE_USD, len(rows), details=True) if reason: raise CollectionDeferred(f"Post recheck: {reason}") - batch_result = classify(rows, set(pending)) + if not rows: + raise RuntimeError(f"Post recheck run {run['id']}: empty output; paid batch retained") + for row in rows: + node = row.get("data") if isinstance(row, dict) else None + code = (node.get("code") or node.get("shortcode")) if isinstance(node, dict) else None + match = POST_URL.search(str(row.get("input") or "")) if isinstance(row, dict) else None + identities = {value for value in (code, match.group(1) if match else None) if value} + if not identities or not identities.issubset(batch): + raise RuntimeError(f"Post recheck run {run['id']}: unrelated or unidentifiable output; paid batch retained") + batch_result = classify(rows, set(batch)) gone = sum(1 for v in batch_result.values() if v == "missing") if len(batch) >= 4 and gone * 2 > len(batch): # Posts this old almost never vanish together; the lookup is broken. raise RuntimeError(f"Post recheck: {gone} of {len(batch)} posts reported deleted; " "treating the response as an outage and changing nothing") results.update(batch_result) - - updates = [] - for code, outcome in results.items(): - media_id = by_code[code] - key = f"instagram:post:{media_id}" - entry = state.get(media_id) or {} - if outcome == "present": - state[media_id] = {"status": "present", "checked_at": iso(now)} - continue - misses = entry.get("misses", 0) + 1 - state[media_id] = {"status": "missing", "misses": misses, "checked_at": iso(now)} - log.info("post %s (%s): reported deleted (%d/%d)", media_id, code, misses, CONFIRMATIONS) - if misses >= CONFIRMATIONS: - updates.append(withdrawal(key, registry[key], now)) - state[media_id]["status"] = "withdrawn" - write_json(STATE_FILE, state) - if updates: - log.info("post recheck: withdrawing %d deleted post(s): %s", len(updates), - ", ".join(u["source_key"] for u in updates)) - log.info("Assessed publication: %s", assessed_events.publish(updates)) + updates = [] + for code, outcome in batch_result.items(): + # Recovery may contain sources that are no longer due. Its paid + # dataset is still consumed, but expired listings are left alone. + if code not in by_code: + continue + media_id = by_code[code] + key = f"instagram:post:{media_id}" + entry = state.get(media_id) or {} + if entry.get("run_id") == run["id"]: + continue + state[media_id] = {"status": outcome, "checked_at": iso(now), "run_id": run["id"]} + if outcome == "missing": + misses = entry.get("misses", 0) + 1 + state[media_id]["misses"] = misses + log.info("post %s (%s): reported deleted (%d/%d)", media_id, code, misses, CONFIRMATIONS) + if misses >= CONFIRMATIONS: + updates.append(withdrawal(key, registry[key], now)) + state[media_id]["status"] = "withdrawn" + if updates: + log.info("post recheck: withdrawing %d deleted post(s): %s", len(updates), + ", ".join(u["source_key"] for u in updates)) + log.info("Assessed publication: %s", assessed_events.publish(updates)) + withdrawn += len(updates) + # Persist each batch before marking its paid run consumed. If either + # write fails, replay retains the run ID and cannot double-count misses. + write_json(STATE_FILE, state) + write_json(details_file, {**read_json(details_file), "consumed": True}) log.info("post recheck: %d looked up, %d present, %d reported deleted, %d withdrawn", len(results), sum(v == "present" for v in results.values()), - sum(v == "missing" for v in results.values()), len(updates)) + sum(v == "missing" for v in results.values()), withdrawn) diff --git a/pipeline/tests/test_post_recheck.py b/pipeline/tests/test_post_recheck.py index d076c1e8..e58aa76d 100644 --- a/pipeline/tests/test_post_recheck.py +++ b/pipeline/tests/test_post_recheck.py @@ -2,15 +2,18 @@ from __future__ import annotations import sys +import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path +from unittest.mock import MagicMock, patch PIPELINE_ROOT = Path(__file__).resolve().parents[1] if str(PIPELINE_ROOT) not in sys.path: sys.path.insert(0, str(PIPELINE_ROOT)) import post_recheck as pr # noqa: E402 +import apify_posts as apify # noqa: E402 NOW = datetime(2026, 10, 9, 12, tzinfo=timezone.utc) @@ -62,5 +65,219 @@ def test_publishes_nothing_and_marks_the_source_deleted(self): self.assertEqual("complete", update["assessment"]["status"]) +class RecheckIntegration(unittest.TestCase): + """Exercise real batching, paid-run recovery and durable check state.""" + + def setUp(self): + directory = self.enterContext(tempfile.TemporaryDirectory()) + self.state_file = Path(directory) / "state.json" + self.run_file = Path(directory) / "recheck.json" + self.details_file = Path(directory) / "recheck-details.json" + self.enterContext(patch.object(pr, "STATE_FILE", self.state_file)) + self.enterContext(patch.object(pr, "RUN_FILE", self.run_file)) + self.enterContext(patch.dict(pr.os.environ, {"APIFY_TOKEN": "test"})) + self.clock = self.enterContext(patch.object(pr, "datetime", wraps=datetime)) + self.clock.now.return_value = NOW + self.events = [] + self.registry = {} + self.codes = {} + self.database = MagicMock() + query = self.database.table.return_value.select.return_value.eq.return_value + query.gte.return_value.lte.return_value.execute.return_value.data = self.events + self.enterContext(patch("db.client", return_value=self.database)) + self.enterContext(patch("assessed_events.load_registry", return_value=self.registry)) + self.publish = self.enterContext(patch("assessed_events.publish")) + self.enterContext(patch.object(pr, "shortcodes", side_effect=lambda ids: { + key: self.codes[key] for key in ids if key in self.codes})) + self.purchases = [] + self.runs = {} + self.datasets = {} + self.requests = [] + self.response_rows = None + self.fail_purchase = None + self.fail_dataset = None + self.run_updates = {} + self.enterContext(patch.object(apify.ApifyClient, "request", new=self.request)) + + def add_posts(self, count, offset=0): + for i in range(offset, offset + count): + key = str(i) + self.codes[key] = f"POST_{i:03d}" + self.events.append(event(key, 2)) + self.registry[f"instagram:post:{key}"] = source(key) + + def request(self, method, path, **kwargs): + self.requests.append((method, path)) + if method == "POST": + if len(self.purchases) == self.fail_purchase: + raise TimeoutError("ambiguous paid start") + codes = [url.rsplit("/", 2)[1] for url in kwargs["json"]["posts"]] + run_id = str(len(self.purchases)) + self.purchases.append(codes) + self.datasets[run_id] = (self.response_rows if self.response_rows is not None else + [{"kind": "post", "data": {"code": code}} for code in codes]) + run = {"id": run_id, "actId": apify.HPIX_ACTOR_ID, "status": "SUCCEEDED", + "startedAt": NOW.isoformat(), "defaultDatasetId": run_id, + "options": {"maxTotalChargeUsd": kwargs["params"]["maxTotalChargeUsd"]}, + "chargedEventCounts": {"individual_post_scraped": len(codes), "apify-actor-start": 1}, + "pricingInfo": {"pricingPerEvent": {"actorChargeEvents": { + "individual_post_scraped": {"eventPriceUsd": .00149}, + "apify-actor-start": {"eventPriceUsd": .00005}}}}} + run.update(self.run_updates) + self.runs[run_id] = run + return {"data": run} + if path.startswith("actor-runs/"): + return {"data": self.runs[path.split("/")[1]]} + if path.startswith("datasets/"): + if path.split("/")[1] == self.fail_dataset: + raise RuntimeError("dataset temporarily unavailable") + return self.datasets[path.split("/")[1]] + raise AssertionError((method, path)) + + def test_distinct_batches_and_later_cycle_do_not_recheck_present_posts(self): + self.add_posts(60) + pr.main() + self.assertEqual([50, 10], [len(batch) for batch in self.purchases]) + self.assertEqual(60, len(set(code for batch in self.purchases for code in batch))) + self.assertEqual(60, len(pr.read_json(self.state_file))) + pr.main() + self.assertEqual(2, len(self.purchases)) + self.add_posts(1, offset=60) + pr.main() + self.assertEqual(["POST_060"], self.purchases[-1]) + + def test_completed_legacy_list_does_not_replace_new_work(self): + pr.write_json(self.details_file, {"consumed": True, "usernames": ["EXPIRED"]}) + self.add_posts(1) + pr.main() + self.assertEqual([["POST_000"]], self.purchases) + + def test_state_write_failure_replays_paid_run_without_another_purchase(self): + self.add_posts(1) + write = pr.write_json + def fail_state(path, value): + if path == self.state_file: + raise OSError("disk unavailable") + write(path, value) + with patch.object(pr, "write_json", side_effect=fail_state): + with self.assertRaises(OSError): + pr.main() + self.assertFalse(pr.read_json(self.details_file)["consumed"]) + pr.main() + self.assertEqual(1, len(self.purchases)) + self.assertEqual("present", pr.read_json(self.state_file)["0"]["status"]) + + def test_later_batch_failure_keeps_completed_checks(self): + self.add_posts(60) + self.fail_purchase = 1 + with self.assertRaises(TimeoutError): + pr.main() + self.assertEqual(50, len(pr.read_json(self.state_file))) + with self.assertRaisesRegex(RuntimeError, "outcome unknown"): + pr.main() + self.assertEqual(1, len(self.purchases)) + + def test_pending_paid_run_is_consumed_even_when_its_events_expired(self): + self.add_posts(1) + with patch.object(pr, "write_json", side_effect=OSError("disk unavailable")): + with self.assertRaises(OSError): + pr.main() + self.events.clear() + pr.main() + self.assertEqual(1, len(self.purchases)) + self.assertTrue(pr.read_json(self.details_file)["consumed"]) + + def test_second_batch_resumes_original_list_after_due_work_changes(self): + self.add_posts(60) + self.fail_dataset = "1" + with self.assertRaisesRegex(RuntimeError, "dataset temporarily unavailable"): + pr.main() + self.assertEqual(50, len(pr.read_json(self.state_file))) + self.add_posts(1, offset=60) + self.fail_dataset = None + pr.main() + self.assertEqual([50, 10, 1], [len(batch) for batch in self.purchases]) + self.assertEqual(["POST_060"], self.purchases[-1]) + self.assertEqual(61, len(pr.read_json(self.state_file))) + + def test_failed_or_charge_limited_run_is_not_consumed_or_rebought(self): + self.add_posts(60) + for changes in ({"status": "FAILED"}, {"options": {"maxTotalChargeUsd": .05}}): + with self.subTest(changes=changes): + self.run_updates = changes + if self.details_file.exists(): + self.details_file.unlink() + before = len(self.purchases) + with self.assertRaises(RuntimeError): + pr.main() + self.assertFalse(pr.read_json(self.details_file)["consumed"]) + with self.assertRaises(RuntimeError): + pr.main() + self.assertEqual(before + 1, len(self.purchases)) + self.assertFalse(self.state_file.exists()) + + def test_withdrawal_failure_replays_paid_results_before_marking_withdrawn(self): + self.add_posts(1) + pr.write_json(self.state_file, {"0": {"status": "missing", "misses": 1, + "checked_at": (NOW - timedelta(hours=8)).isoformat(), "run_id": "old"}}) + self.response_rows = [{"kind": "post", "input": "https://www.instagram.com/p/POST_000/", + "error": "Post deleted"}] + self.publish.side_effect = RuntimeError("database unavailable") + with self.assertRaisesRegex(RuntimeError, "database unavailable"): + pr.main() + self.assertEqual("missing", pr.read_json(self.state_file)["0"]["status"]) + self.assertFalse(pr.read_json(self.details_file)["consumed"]) + self.publish.side_effect = None + pr.main() + self.assertEqual(1, len(self.purchases)) + self.assertEqual("withdrawn", pr.read_json(self.state_file)["0"]["status"]) + + def test_matching_rate_limit_stays_unknown_without_withdrawing(self): + self.add_posts(1) + self.response_rows = [{"input": "https://www.instagram.com/p/POST_000/", "error": "Rate limited"}] + pr.main() + self.assertEqual({}, pr.read_json(self.state_file)) + self.assertTrue(pr.read_json(self.details_file)["consumed"]) + self.publish.assert_not_called() + + def test_unrelated_or_empty_output_stops_without_buying_more_batches(self): + self.add_posts(60) + for rows in ([{"kind": "post", "data": {"code": "UNRELATED"}}], []): + with self.subTest(rows=rows): + self.response_rows = rows + if self.details_file.exists(): + self.details_file.unlink() + before = len(self.purchases) + with self.assertRaisesRegex(RuntimeError, "output"): + pr.main() + self.assertEqual(before + 1, len(self.purchases)) + with self.assertRaisesRegex(RuntimeError, "output"): + pr.main() + self.assertEqual(before + 1, len(self.purchases)) + self.assertFalse(self.state_file.exists()) + + def test_missing_confirmation_is_not_counted_twice_after_consumption_write_failure(self): + self.add_posts(1) + self.response_rows = [{"kind": "post", "input": "https://www.instagram.com/p/POST_000/", + "error": "Post deleted"}] + write = pr.write_json + def fail_consumption(path, value): + if path == self.details_file and value.get("consumed"): + raise OSError("disk unavailable") + write(path, value) + with patch.object(pr, "write_json", side_effect=fail_consumption): + with self.assertRaises(OSError): + pr.main() + self.clock.now.return_value = NOW + timedelta(hours=8) + pr.main() + self.assertEqual(1, len(self.purchases)) + self.assertEqual(1, pr.read_json(self.state_file)["0"]["misses"]) + self.publish.assert_not_called() + pr.main() + self.assertEqual(2, len(self.purchases)) + self.assertEqual("withdrawn", pr.read_json(self.state_file)["0"]["status"]) + self.publish.assert_called_once() + + if __name__ == "__main__": unittest.main()