From c9ee4bde30906a327c0288724aa3705603e322d0 Mon Sep 17 00:00:00 2001 From: Antoine Vandeplanque Date: Mon, 12 Oct 2026 19:55:15 +0200 Subject: [PATCH] fetch details incrementally on listing changes or weekly refresh --- docs/SSH_RUN.md | 41 ++++++++++++++++++ ops/app/compose.runtime.yaml | 2 + ops/ssh/host_run.py | 18 ++++++-- src/car_tracker/collect.py | 21 +++++++++- src/car_tracker/detail_policy.py | 67 ++++++++++++++++++++++++++++++ src/car_tracker/pipeline.py | 3 ++ tests/test_collect.py | 14 +++++++ tests/test_database_integration.py | 31 ++++++++++++++ tests/test_detail_policy.py | 53 +++++++++++++++++++++++ tests/test_pipeline.py | 15 +++++++ tests/test_ssh_run.py | 16 +++++++ 11 files changed, 275 insertions(+), 6 deletions(-) create mode 100644 src/car_tracker/detail_policy.py create mode 100644 tests/test_detail_policy.py diff --git a/docs/SSH_RUN.md b/docs/SSH_RUN.md index d977800..bdf835a 100644 --- a/docs/SSH_RUN.md +++ b/docs/SSH_RUN.md @@ -183,6 +183,8 @@ COLLECT_FAMILY=saab-900 COLLECT_MAX_PAGES=1 COLLECT_MAX_DETAILS=all COLLECT_DELAY=5 +COLLECT_DETAIL_POLICY=incremental +COLLECT_DETAIL_REFRESH_DAYS=7 ``` The SSH wrapper reads resolved Compose settings and supplies a validated, fixed @@ -202,6 +204,45 @@ increase runtime; the Actions timeout remains 45 minutes. The pipeline and local state still retain exit 1 for partial coverage; the SSH wrapper converts verified limit-only outcomes into an Actions warning/success. No run is launched by editing app.env. +### Incremental details + +`COLLECT_DETAIL_POLICY=incremental` fetches detail pages only when: + +1. There is no successfully imported detail observation for the ad. +2. The search-visible listing data changed since the search accompanying its last + successful detail fetch (price, source update timestamp, title/description, + vehicle attributes, location, status/category and publication/expiration dates). +3. Its last imported detail is at least `COLLECT_DETAIL_REFRESH_DAYS` old (default 7). + +Counters, promotion/boost flags, attribute order and excluded personal attributes +do not trigger a refresh. Detail-only changes invisible in search results are +detected only on the age-based refresh. A weekly refresh is an eligibility rule +for the next manually requested run, not a weekly schedule. + +History comes from accepted observations in committed PostgreSQL imports. The +baseline search is the latest search at/before the last successful detail, not +the most recent search overall: failed enrichment cannot acknowledge a change. +Missing baselines conservatively trigger a request. Cache skips are recorded as +`not_attempted` with reason `fresh_unchanged`; they do not create a detail observation, +advance freshness, consume max-details, or claim a new detail was fetched. Such +intentional skips may leave raw coverage partial but produce the existing Actions +warning/success when there are no actual failures or rejected records. + +The pipeline loads a read-only history snapshot before collection. Failure to load +it prevents collection rather than silently falling back to fetching everything. +Standalone `car-tracker` collection rejects incremental mode; use the database-backed +`car-tracker-pipeline`. No schema migration is required, and existing import/replay +semantics are unchanged. For this homelab-sized dataset the snapshot contains the +latest accepted detail and preceding search per listing. + +**Activate only after a new image supporting these flags is published.** Install +the updated host_run.py and compose.runtime.yaml, then add the two policy variables +to host app.env and select that new release. For older images use +`COLLECT_DETAIL_POLICY=always`; defaults remain always for backward compatibility. +COLLECT_MAX_DETAILS=all still means all eligible ads, not all discovered ads. +Existing details older than seven days will be eligible on the first incremental +run, so this does not guarantee fewer requests immediately or eliminate 403s. + ### Execution outcomes Local outcomes: `~/containers/car-tracker/state/requests/.json`. diff --git a/ops/app/compose.runtime.yaml b/ops/app/compose.runtime.yaml index 6edc513..5fc1ddd 100644 --- a/ops/app/compose.runtime.yaml +++ b/ops/app/compose.runtime.yaml @@ -34,6 +34,8 @@ services: COLLECT_MAX_PAGES: "${COLLECT_MAX_PAGES:-1}" COLLECT_MAX_DETAILS: "${COLLECT_MAX_DETAILS:-1}" COLLECT_DELAY: "${COLLECT_DELAY:-5}" + COLLECT_DETAIL_POLICY: "${COLLECT_DETAIL_POLICY:-always}" + COLLECT_DETAIL_REFRESH_DAYS: "${COLLECT_DETAIL_REFRESH_DAYS:-7}" extra_hosts: - "host.docker.internal:host-gateway" volumes: diff --git a/ops/ssh/host_run.py b/ops/ssh/host_run.py index 8efbe7e..8b2c4b9 100644 --- a/ops/ssh/host_run.py +++ b/ops/ssh/host_run.py @@ -38,9 +38,9 @@ def limit_reasons(manifest, pipeline): for item in details.get("listings", {}).values(): if item.get("status") == "complete": continue - if item.get("status") != "not_attempted" or item.get("reason") != "detail_limit": + if item.get("status") != "not_attempted" or item.get("reason") not in ("detail_limit", "fresh_unchanged"): return [] - reasons.add("detail_limit") + reasons.add(item["reason"]) return sorted(reasons) @@ -59,12 +59,13 @@ def outcome_summary(manifest, pipeline): failed = sum(item.get("status") == "failed" for item in listings) skipped = sum(item.get("status") == "not_attempted" for item in listings) stopped = sum(item.get("reason") == "run_stopped" for item in listings) + cached = sum(item.get("reason") == "fresh_unchanged" for item in listings) lines = [ f"Collection: {status(manifest.get('status'))}. Import: {status(pipeline.get('import_status'))}.", f"Import records: accepted={count(summary.get('accepted'))}, rejected={count(summary.get('rejected'))}, " f"field issues={count(summary.get('field_issues'))}.", f"Details: completed={completed}, failed={failed}, not attempted={skipped} " - f"(collection stopped early={stopped}).", + f"(fresh unchanged details reused={cached}, collection stopped early={stopped}).", ] items = [*manifest.get("searches", []), *listings] http = sorted({item["http_status"] for item in items @@ -132,7 +133,8 @@ def actions_outcome(code, request, state, raw_root): print("::error::" + " ".join(lines) + " See private host logs/manifest; no automatic retry.", flush=True) return code explanations = {"page_limit": "the configured page limit left additional search pages unfetched", - "detail_limit": "the configured detail limit left some discovered ads without detail attempts"} + "detail_limit": "the configured detail limit left some discovered ads without detail attempts", + "fresh_unchanged": "unchanged ads with fresh previously imported details were intentionally not fetched again"} message = "; ".join(explanations[reason] for reason in reasons) print(f"::warning::Import succeeded with no rejected records, but coverage is intentionally partial: {message}. " "This is not full source coverage. Host state retains pipeline exit 1.", flush=True) @@ -147,6 +149,12 @@ def collection_command(settings): pages = settings.get("COLLECT_MAX_PAGES", "1") details = settings.get("COLLECT_MAX_DETAILS", "1") delay = settings.get("COLLECT_DELAY", "5") + policy = settings.get("COLLECT_DETAIL_POLICY", "always") + refresh = settings.get("COLLECT_DETAIL_REFRESH_DAYS", "7") + if policy not in ("always", "incremental"): + raise ValueError("Detail policy must be always or incremental") + if not isinstance(refresh, str) or not refresh.isdecimal() or int(refresh) < 1: + raise ValueError("Detail refresh days must be positive") if not isinstance(family, str) or not family.strip(): raise ValueError("A collection family or all is required") if not isinstance(pages, str) or not pages.isdecimal() or int(pages) < 1: @@ -158,6 +166,8 @@ def collection_command(settings): raise ValueError("Delay must be finite and at least one second") args = ["car-tracker-pipeline", "--headed", "--config", "/app/searches.toml", "--output", "/data", "--max-pages", pages, "--delay", delay] + if policy == "incremental": + args.extend(["--detail-policy", policy, "--detail-refresh-days", refresh]) if family != "all": args.extend(["--family", family]) if details != "all": diff --git a/src/car_tracker/collect.py b/src/car_tracker/collect.py index 790ac5a..73ec4c5 100644 --- a/src/car_tracker/collect.py +++ b/src/car_tracker/collect.py @@ -114,7 +114,7 @@ def write_records(path, records): temporary.replace(path) -def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, max_details=None): +def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, max_details=None, detail_policy=None): run_id = str(uuid.uuid4()) directory = output / datetime.now(timezone.utc).strftime("%Y-%m-%d") / run_id directory.mkdir(mode=0o700, parents=True, exist_ok=False) @@ -126,6 +126,8 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m "planned_searches": searches, "details": { "enabled": not search_only, "max_details": max_details, + "policy": "incremental" if detail_policy is not None else "always", + "refresh_days": getattr(detail_policy, "refresh_days", None), "status": "disabled" if search_only else "running", "listings": {}, }, } @@ -192,6 +194,12 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m item = details[identifier] if item["status"] != "pending": continue + if detail_policy is not None: + decision = detail_policy(ad) + item["detail_decision"] = decision + if decision == "fresh_unchanged": + item.update(status="not_attempted", reason="fresh_unchanged") + continue if max_details is not None and detail_attempts >= max_details: item.update(status="not_attempted", reason="detail_limit") continue @@ -280,11 +288,15 @@ def parse_collection_args(argv=None, description=None): parser.add_argument("--headed", action="store_true") parser.add_argument("--search-only", action="store_true", help="Skip individual listing pages") parser.add_argument("--max-details", type=int, help="Maximum unique detail requests per run") + parser.add_argument("--detail-policy", choices=("always", "incremental"), default="always") + parser.add_argument("--detail-refresh-days", type=int, default=7) args = parser.parse_args(argv) if args.max_pages < 1 or not math.isfinite(args.delay) or args.delay < 1: parser.error("max-pages must be positive and delay must be finite and >= 1 second") if args.max_details is not None and args.max_details < 1: parser.error("max-details must be positive") + if args.detail_refresh_days < 1: + parser.error("detail-refresh-days must be positive") with args.config.open("rb") as stream: searches = tomllib.load(stream).get("searches", []) if not isinstance(searches, list) or not searches or any( @@ -330,6 +342,7 @@ def collect_browser(args): result = collect( fetch, args.searches, args.output, args.max_pages, args.delay, args.search_only, args.max_details, + detail_policy=getattr(args, "detail_decider", None), ) finally: browser.close() @@ -350,6 +363,10 @@ def collect_browser(args): def main(): - path, status = collect_browser(parse_collection_args()) + args = parse_collection_args() + if args.detail_policy == "incremental": + print("Incremental details require car-tracker-pipeline and imported database history.", file=sys.stderr) + return 3 + path, status = collect_browser(args) print(f"{status}: {path}") return 0 if status == "complete" else 1 diff --git a/src/car_tracker/detail_policy.py b/src/car_tracker/detail_policy.py new file mode 100644 index 0000000..c1d144f --- /dev/null +++ b/src/car_tracker/detail_policy.py @@ -0,0 +1,67 @@ +"""Choose detail requests from successfully imported history, not failed attempts.""" + +from datetime import datetime, timedelta, timezone +import json + +from .collect import minimize + + +CHANGE_FIELDS = { + "subject", "body", "price", "price_cents", "brand", "status", "ad_type", + "category_id", "category_name", "attributes", "location", "index_date", + "first_publication_date", "expiration_date", +} + + +def signature(ad): + value = {key: item for key, item in minimize(ad).items() if key in CHANGE_FIELDS} + if "attributes" in value: + value["attributes"] = sorted(value["attributes"], key=lambda item: json.dumps(item, sort_keys=True)) + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +class DetailPolicy: + def __init__(self, history, refresh_days=7, clock=None): + self.history = history + self.refresh_days = refresh_days + self.clock = clock or (lambda: datetime.now(timezone.utc)) + + def __call__(self, ad): + previous = self.history.get(str(ad["list_id"])) + if previous is None: + return "never_detailed" + observed_at, search = previous + current = self.clock() + if observed_at > current: + return "invalid_history_time" + if current - observed_at >= timedelta(days=self.refresh_days): + return "weekly_refresh" + if search is None: + return "missing_search_baseline" + if signature(ad) != signature(search): + return "search_changed" + return "fresh_unchanged" + + +def load_policy(conn, refresh_days=7): + # A detail must have an accepted observation from a committed import. Pair it + # with the search observed before that detail, NOT the latest search overall: + # otherwise a failed enrichment could incorrectly acknowledge a change. + rows = conn.execute(""" + WITH details AS ( + SELECT DISTINCT ON (source_listing_id) + source_listing_id, observed_at, observation_id + FROM silver.observations + WHERE source = 'leboncoin' AND record_type = 'detail' + ORDER BY source_listing_id, observed_at DESC, observation_id DESC + ) + SELECT d.source_listing_id, d.observed_at, s.raw_payload + FROM details d + LEFT JOIN LATERAL ( + SELECT raw_payload FROM silver.observations + WHERE source = 'leboncoin' AND source_listing_id = d.source_listing_id + AND record_type = 'search' AND observed_at <= d.observed_at + ORDER BY observed_at DESC, observation_id DESC LIMIT 1 + ) s ON true + """).fetchall() + return DetailPolicy({identifier: (observed_at, search) for identifier, observed_at, search in rows}, refresh_days) diff --git a/src/car_tracker/pipeline.py b/src/car_tracker/pipeline.py index 06bef08..5b27758 100644 --- a/src/car_tracker/pipeline.py +++ b/src/car_tracker/pipeline.py @@ -12,6 +12,7 @@ import psycopg from .collect import collect_browser, parse_collection_args, write_json from .database import InputError, database_error_hint, import_run +from .detail_policy import load_policy def run_pipeline(args): @@ -35,6 +36,8 @@ def run_pipeline(args): try: with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn: conn.execute("SELECT 1 FROM silver.collection_runs LIMIT 0") + if getattr(args, "detail_policy", "always") == "incremental": + args.detail_decider = load_policy(conn, args.detail_refresh_days) path, status = collect_browser(args) print(f"Manifest: {path}", flush=True) sidecar = path.parent / "pipeline.json" diff --git a/tests/test_collect.py b/tests/test_collect.py index af6c2f2..2102e25 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -150,6 +150,20 @@ class CollectorTests(unittest.TestCase): self.run_collection([SourceAccessError(429, capture=capture)]) capture.assert_not_called() + def test_fresh_details_are_intentionally_skipped_without_requests(self): + path, manifest = self.run_collection([payload([123, 456], total=2)], search_only=False, + detail_policy=lambda ad: "fresh_unchanged") + self.assertEqual(manifest["details"]["attempted"], 0) + self.assertEqual(manifest["details"]["listings"]["123"]["reason"], "fresh_unchanged") + self.assertEqual(len(list(path.parent.glob("*.jsonl.gz"))), 1) + + def test_fresh_skips_do_not_consume_detail_request_limit(self): + _, manifest = self.run_collection([payload([123, 456], total=2), detail_payload(456)], + search_only=False, max_details=1, + detail_policy=lambda ad: "fresh_unchanged" if ad["list_id"] == 123 else "never_detailed") + self.assertEqual(manifest["details"]["attempted"], 1) + self.assertEqual(manifest["details"]["listings"]["456"]["status"], "complete") + def test_discovery_failure_does_not_claim_detail_completion(self): _, manifest = self.run_collection([SourceAccessError(403)], search_only=False) self.assertEqual(manifest["details"]["status"], "not_attempted") diff --git a/tests/test_database_integration.py b/tests/test_database_integration.py index 067fe15..dfa7b2f 100644 --- a/tests/test_database_integration.py +++ b/tests/test_database_integration.py @@ -21,6 +21,7 @@ from psycopg.conninfo import conninfo_to_dict from car_tracker import database from car_tracker.collect import search_url, write_json, write_records +from car_tracker.detail_policy import load_policy @unittest.skipUnless(os.environ.get("TEST_DATABASE_URL"), "TEST_DATABASE_URL is not set") @@ -128,6 +129,36 @@ class DatabaseIntegrationTests(unittest.TestCase): database.migrate(self.conn) self.assertEqual(self.state(), before) + def test_incremental_policy_pairs_search_with_last_successful_detail(self): + path, _ = self.fixture(detail={"list_id": 123, "price": ["12500"]}, day=1) + database.import_run(self.conn, path) + later, _ = self.fixture(pages=[[{"list_id": 123, "price": ["11000"]}]], day=2) + database.import_run(self.conn, later) + before = self.state() + policy = load_policy(self.conn) + policy.clock = lambda: datetime(2026, 1, 3, tzinfo=timezone.utc) + self.assertEqual(policy({"list_id": 123, "price": ["12000"]}), "fresh_unchanged") + self.assertEqual(policy({"list_id": 123, "price": ["11000"]}), "search_changed") + self.assertEqual(policy({"list_id": 456, "price": ["10000"]}), "never_detailed") + policy.clock = lambda: datetime(2026, 1, 9, tzinfo=timezone.utc) + self.assertEqual(policy({"list_id": 123, "price": ["12000"]}), "weekly_refresh") + self.assertEqual(self.state(), before) + + def test_cached_skip_import_preserves_detail_age_and_reason(self): + first, _ = self.fixture(detail={"list_id": 123, "price": ["12000"]}, day=1) + database.import_run(self.conn, first) + path, manifest = self.fixture(day=2) + manifest["status"] = "partial" + manifest["details"] = {"enabled": True, "status": "partial", "listings": { + "123": {"status": "not_attempted", "reason": "fresh_unchanged"}}} + write_json(path, manifest) + database.import_run(self.conn, path) + self.assertEqual(self.conn.execute( + "SELECT reason FROM silver.detail_coverage WHERE run_id = %s", (manifest["run_id"],) + ).fetchone()[0], "fresh_unchanged") + policy = load_policy(self.conn) + self.assertEqual(policy.history["123"][0].day, 1) + def test_applied_migration_checksum_mismatch_rolls_back(self): before = self.state() # The outer transaction also rolls back the deliberate checksum corruption. diff --git a/tests/test_detail_policy.py b/tests/test_detail_policy.py new file mode 100644 index 0000000..3a76aed --- /dev/null +++ b/tests/test_detail_policy.py @@ -0,0 +1,53 @@ +from datetime import datetime, timedelta, timezone +import unittest +from unittest.mock import Mock + +from car_tracker.detail_policy import DetailPolicy, load_policy, signature + + +class DetailPolicyTests(unittest.TestCase): + def setUp(self): + self.current = datetime(2026, 10, 12, 12, tzinfo=timezone.utc) + self.ad = {"list_id": 123, "price": [12000], "subject": "Synthetic car", + "attributes": [{"key": "mileage", "value": "100000"}]} + + def policy(self, age=1, baseline=True): + return DetailPolicy({"123": (self.current - timedelta(days=age), self.ad if baseline else None)}, + clock=lambda: self.current) + + def test_no_successful_detail_is_eligible(self): + self.assertEqual(DetailPolicy({}, clock=lambda: self.current)(self.ad), "never_detailed") + + def test_unchanged_fresh_ad_is_skipped(self): + self.assertEqual(self.policy()(self.ad), "fresh_unchanged") + + def test_exact_week_boundary_is_eligible(self): + self.assertEqual(self.policy(age=7)(self.ad), "weekly_refresh") + self.assertEqual(self.policy(age=6)(self.ad), "fresh_unchanged") + + def test_price_and_source_timestamp_changes_are_eligible(self): + self.assertEqual(self.policy()({**self.ad, "price": [11000]}), "search_changed") + self.assertEqual(self.policy()({**self.ad, "index_date": "2026-10-12"}), "search_changed") + + def test_unknown_baseline_and_future_timestamp_are_conservative(self): + self.assertEqual(self.policy(baseline=False)(self.ad), "missing_search_baseline") + self.assertEqual(self.policy(age=-1)(self.ad), "invalid_history_time") + + def test_counters_and_promotions_do_not_trigger_enrichment(self): + other = {**self.ad, "counters": {"views": 123}, "is_boosted": True, + "options": {"highlight": True}, "owner": {"name": "private"}} + self.assertEqual(self.policy()(other), "fresh_unchanged") + + def test_attribute_order_and_sensitive_fields_do_not_trigger_changes(self): + a = {**self.ad, "attributes": [*self.ad["attributes"], {"key": "fuel", "value": "petrol"}]} + b = {**a, "attributes": [*reversed(a["attributes"]), {"key": "phone", "value": "secret"}]} + self.assertEqual(signature(a), signature(b)) + + def test_loads_only_imported_history(self): + conn = Mock() + conn.execute.return_value.fetchall.return_value = [("123", self.current, self.ad)] + policy = load_policy(conn) + self.assertEqual(policy.history["123"], (self.current, self.ad)) + query = conn.execute.call_args.args[0] + self.assertIn("observed_at <= d.observed_at", query) + self.assertIn("record_type = 'detail'", query) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index dcd9606..810c6d6 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -103,6 +103,21 @@ class PipelineTests(unittest.TestCase): self.preflight.__exit__.assert_called_once() self.assertNotIn("TOP_SECRET", self.stderr.getvalue()) + def test_incremental_history_is_loaded_before_collection(self): + self.args.detail_policy = "incremental" + self.args.detail_refresh_days = 7 + with patch.object(pipeline, "load_policy", return_value="synthetic-policy") as load: + self.assertEqual(pipeline.run_pipeline(self.args), 0) + load.assert_called_once_with(self.preflight, 7) + self.assertEqual(self.args.detail_decider, "synthetic-policy") + + def test_history_failure_does_not_start_collection(self): + self.args.detail_policy = "incremental" + self.args.detail_refresh_days = 7 + with patch.object(pipeline, "load_policy", side_effect=pipeline.psycopg.OperationalError("secret")): + self.assertEqual(pipeline.run_pipeline(self.args), 3) + self.collect.assert_not_called() + def test_finalized_interruption_is_not_imported(self): self.collect.side_effect = lambda args: self.fixture(args, "partial", "KeyboardInterrupt") self.assertEqual(pipeline.run_pipeline(self.args), 130) diff --git a/tests/test_ssh_run.py b/tests/test_ssh_run.py index 2a4d875..681d5ab 100644 --- a/tests/test_ssh_run.py +++ b/tests/test_ssh_run.py @@ -97,6 +97,22 @@ class SSHRunTests(unittest.TestCase): self.assertEqual(args[args.index("--max-details") + 1], "20") self.assertEqual(args[args.index("--max-pages") + 1], "3") + def test_incremental_policy_is_host_configurable_and_validated(self): + args = host.collection_command({"COLLECT_DETAIL_POLICY": "incremental", "COLLECT_DETAIL_REFRESH_DAYS": "7"}) + self.assertEqual(args[args.index("--detail-policy") + 1], "incremental") + self.assertEqual(args[args.index("--detail-refresh-days") + 1], "7") + self.assertNotIn("--detail-policy", host.collection_command({})) + for settings in ({"COLLECT_DETAIL_POLICY": "unknown"}, {"COLLECT_DETAIL_REFRESH_DAYS": "0"}): + with self.assertRaises(ValueError): + host.collection_command(settings) + + def test_fresh_unchanged_skips_are_not_source_failures(self): + manifest, pipeline = self.bounded_outcome() + manifest["searches"][0] = {"status": "complete"} + manifest["details"]["listings"]["1"] = {"status": "not_attempted", "reason": "fresh_unchanged"} + self.assertEqual(host.limit_reasons(manifest, pipeline), ["fresh_unchanged"]) + self.assertIn("fresh unchanged details reused=1", " ".join(host.outcome_summary(manifest, pipeline))) + def test_invalid_settings_are_refused(self): for key, value in [("COLLECT_MAX_DETAILS", "0"), ("COLLECT_MAX_DETAILS", "-1"), ("COLLECT_MAX_DETAILS", "all; id"), ("COLLECT_MAX_PAGES", "all"),