From 60959a58c95e09b0b0e1aa690eef1e6eab46d22b Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 2 Jul 2026 23:03:44 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20deactivate=5Fstale=20=D0=B4?= =?UTF-8?q?=D0=BB=D1=8F=20domklik/n1=20+=20=D1=87=D0=B5=D1=81=D1=82=D0=BD?= =?UTF-8?q?=D0=B0=D1=8F=20freshness=20=D0=BF=D0=BE=20scraped=5Fat=20(#2204?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - deactivate_stale_listings: kwarg staleness_column (whitelist last_seen_at/scraped_at, ValueError до SQL; имя колонки в f-string только после whitelist, значения — binding) - параметр прокинут в оба dispatch-пути (product_handlers kit + legacy scheduler) - seed 160: deactivate_stale_domklik (scraped_at, ttl=14 — last_seen_at двигается нетрекаемым bulk-touch и для TTL бесполезен), deactivate_stale_n1 (ttl=14, dead source) - domklik blanket по сегментам осознанно: 100% строк vtorichka (prod 2026-07-02) - +13 тестов; полный сьют 2959 passed Refs #2204 --- .../backend/app/services/product_handlers.py | 2 + tradein-mvp/backend/app/services/scheduler.py | 10 +- .../app/tasks/deactivate_stale_avito.py | 101 +++++++++--- .../160_seed_deactivate_stale_domklik_n1.sql | 93 +++++++++++ .../tests/test_deactivate_stale_listings.py | 148 ++++++++++++++++++ 5 files changed, 326 insertions(+), 28 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/160_seed_deactivate_stale_domklik_n1.sql diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py index fb38506d..2810bc2e 100644 --- a/tradein-mvp/backend/app/services/product_handlers.py +++ b/tradein-mvp/backend/app/services/product_handlers.py @@ -149,6 +149,7 @@ async def _job_deactivate_stale( listing_source: str = params.get("listing_source", "avito") ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days) segments: list[str] | None = params.get("segments") + staleness_column: str = params.get("staleness_column", "last_seen_at") loop = asyncio.get_event_loop() await loop.run_in_executor( @@ -159,6 +160,7 @@ async def _job_deactivate_stale( listing_source=listing_source, ttl_days=ttl_days, segments=segments, + staleness_column=staleness_column, ), ) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 2c2f7277..c233e498 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -1038,10 +1038,14 @@ async def trigger_deactivate_stale_avito_run( async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Универсальный триггер для всех deactivate_stale_* источников. - Читает listing_source, ttl_days, segments из schedule_row["default_params"] (jsonb). + Читает listing_source, ttl_days, segments, staleness_column из + schedule_row["default_params"] (jsonb). Для источника 'deactivate_stale_avito' с пустым default_params использует defaults: - listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None. + listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None, + staleness_column='last_seen_at'. Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments. + Для domklik (#2204) -- staleness_column='scraped_at' (нетрекаемый bulk-touch + двигает last_seen_at, честная свежесть = scraped_at). Запускает deactivate_stale_listings в executor (sync DB-only task). Возвращает run_id или None (skip -- already running). @@ -1056,6 +1060,7 @@ async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any] listing_source: str = params.get("listing_source", "avito") ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days) segments: list[str] | None = params.get("segments") # list or None + staleness_column: str = params.get("staleness_column", "last_seen_at") async def _run() -> None: run_db = SessionLocal() @@ -1071,6 +1076,7 @@ async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any] listing_source=listing_source, ttl_days=ttl_days, segments=segments, + staleness_column=staleness_column, ), ) except Exception: diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py index 216f5b78..3fdc40af 100644 --- a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -31,30 +31,57 @@ from app.services import scrape_runs as runs_mod logger = logging.getLogger(__name__) -# Базовый UPDATE без фильтра по сегменту: все сегменты для данного source. -# CAST(:ttl_days || ' days' AS interval) -- psycopg v3 safe (никаких :param::type). -_DEACTIVATE_SQL_ALL_SEGMENTS = text( - """ - UPDATE listings - SET is_active = false - WHERE source = :listing_source - AND is_active = true - AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval) - """ -) +# Разрешённые колонки-таймстемпы для проверки свежести (staleness). +# Whitelist ЖЁСТКО ограничивает f-string-интерполяцию имени колонки в SQL: +# значение колонки НИКОГДА не берётся из пользовательского ввода без этой проверки. +# - last_seen_at: дефолт; двигается при каждом «увиденном» объявлении в прогоне. +# - scraped_at: честная свежесть для источников с нетрекаемым bulk-touch +# last_seen_at (domklik: #2204 — bulk UPDATE двигает last_seen_at всем строкам +# одним timestamp, поэтому TTL по last_seen_at бесполезен; scraped_at двигает +# только реальный скрейп). +_ALLOWED_STALENESS_COLUMNS = frozenset({"last_seen_at", "scraped_at"}) -# UPDATE с фильтром по сегменту (segments задан): затрагивает только указанные сегменты. -# = ANY(CAST(:segments AS text[])) -- psycopg v3 адаптирует Python list -> text[]. -_DEACTIVATE_SQL_SEGMENTS = text( + +def _build_all_segments_sql(staleness_column: str) -> Any: + """UPDATE без фильтра по сегменту: все сегменты для данного source. + + staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings, + поэтому f-string-подстановка имени колонки безопасна. Значения (:listing_source, + :ttl_days) остаются param-binding — psycopg v3 safe (никаких :param::type). """ - UPDATE listings - SET is_active = false - WHERE source = :listing_source - AND is_active = true - AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval) - AND listing_segment = ANY(CAST(:segments AS text[])) + return text( + f""" + UPDATE listings + SET is_active = false + WHERE source = :listing_source + AND is_active = true + AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval) + """ + ) + + +def _build_segments_sql(staleness_column: str) -> Any: + """UPDATE с фильтром по сегменту (segments задан): только указанные сегменты. + + staleness_column уже прошёл whitelist-проверку. = ANY(CAST(:segments AS text[])) + -- psycopg v3 адаптирует Python list -> text[]. """ -) + return text( + f""" + UPDATE listings + SET is_active = false + WHERE source = :listing_source + AND is_active = true + AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval) + AND listing_segment = ANY(CAST(:segments AS text[])) + """ + ) + + +# Дефолтные (last_seen_at) варианты SQL -- сохранены как модульные константы для +# обратной совместимости (тесты читают .text, product_handlers/scheduler не менялись). +_DEACTIVATE_SQL_ALL_SEGMENTS = _build_all_segments_sql("last_seen_at") +_DEACTIVATE_SQL_SEGMENTS = _build_segments_sql("last_seen_at") # Алиас для обратной совместимости -- использовался в тестах через task_mod._DEACTIVATE_SQL _DEACTIVATE_SQL = _DEACTIVATE_SQL_ALL_SEGMENTS @@ -67,22 +94,43 @@ def deactivate_stale_listings( listing_source: str, ttl_days: int, segments: list[str] | None = None, + staleness_column: str = "last_seen_at", ) -> dict[str, int]: - """Пометить is_active=false объявления с last_seen_at > ttl_days дней назад. + """Пометить is_active=false объявления, чья свежесть старше ttl_days дней. Параметры: - listing_source: значение поля source в таблице listings ('avito', 'yandex', 'cian'). + listing_source: значение поля source в таблице listings ('avito', 'yandex', + 'cian', 'domklik', 'n1'). ttl_days: количество дней TTL; объявления старше этого порога деактивируются. segments: если задан -- деактивировать только объявления с указанными listing_segment значениями. None -> все сегменты (поведение avito по умолчанию). + staleness_column: колонка-таймстемп, по которой считается свежесть. Whitelist + {"last_seen_at", "scraped_at"} — иначе ValueError ДО любого SQL. Дефолт + last_seen_at. Для domklik (#2204) — scraped_at: нетрекаемый bulk-touch + двигает last_seen_at всем строкам одним timestamp, поэтому честная + свежесть = scraped_at (двигается только реальным скрейпом). Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed). Returns {"deactivated": N} -- количество обновлённых строк. + + Raises: + ValueError: если staleness_column не входит в whitelist (проверка ДО SQL, + никакой интерполяции пользовательского ввода в запрос). """ counters: dict[str, int] = {"deactivated": 0} try: + # Whitelist-проверка ДО построения/выполнения SQL: только после неё имя колонки + # интерполируется f-string'ом. Значения по-прежнему идут через param-binding. + # Внутри try -> невалидная колонка финализирует run как failed (mark_failed), + # а не оставляет его «running» (единый контракт с прочими сбоями). + if staleness_column not in _ALLOWED_STALENESS_COLUMNS: + raise ValueError( + f"invalid staleness_column={staleness_column!r}; " + f"allowed: {sorted(_ALLOWED_STALENESS_COLUMNS)}" + ) + # segments is None -> все сегменты (поведение avito). segments=[...] -> только # перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список [] # означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0), @@ -93,10 +141,10 @@ def deactivate_stale_listings( "ttl_days": ttl_days, "segments": segments, } - result = db.execute(_DEACTIVATE_SQL_SEGMENTS, params) + result = db.execute(_build_segments_sql(staleness_column), params) else: params = {"listing_source": listing_source, "ttl_days": ttl_days} - result = db.execute(_DEACTIVATE_SQL_ALL_SEGMENTS, params) + result = db.execute(_build_all_segments_sql(staleness_column), params) counters["deactivated"] = result.rowcount or 0 @@ -104,12 +152,13 @@ def deactivate_stale_listings( runs_mod.mark_done(db, run_id, counters) logger.info( "deactivate_stale source=%s run_id=%d done: deactivated=%d " - "(ttl_days=%d, segments=%r)", + "(ttl_days=%d, segments=%r, staleness_column=%s)", listing_source, run_id, counters["deactivated"], ttl_days, segments, + staleness_column, ) return counters except Exception as exc: diff --git a/tradein-mvp/backend/data/sql/160_seed_deactivate_stale_domklik_n1.sql b/tradein-mvp/backend/data/sql/160_seed_deactivate_stale_domklik_n1.sql new file mode 100644 index 00000000..cefc5ba2 --- /dev/null +++ b/tradein-mvp/backend/data/sql/160_seed_deactivate_stale_domklik_n1.sql @@ -0,0 +1,93 @@ +-- 160_seed_deactivate_stale_domklik_n1.sql +-- Issue #2204 — seed scrape_schedules rows for nightly stale-listing deactivation +-- of domklik and n1 (wildcard deactivate_stale_* family, #759 generalisation). +-- +-- Context (verified on prod 2026-07-02): +-- n1: 381 is_active rows; last_seen_at frozen at 2026-06-16; no scraper runs +-- anymore -> source is DEAD. Nothing keeps these rows fresh, yet they +-- inflate the active-listing denominator used by coverage/estimator. +-- domklik: 6296 is_active rows; scraped_at frozen at 2026-06-28, BUT last_seen_at +-- is advanced every day by an UNtracked bulk UPDATE (all rows share one +-- timestamp) -> a TTL over last_seen_at can NEVER fire for domklik. +-- Honest freshness for domklik = scraped_at, which only moves when a real +-- scrape actually re-sees the listing. +-- +-- Fix: +-- 1. app/tasks/deactivate_stale_avito.py — deactivate_stale_listings() gains a +-- staleness_column kwarg (whitelist {last_seen_at, scraped_at}); the WHERE +-- clause compares the chosen column against NOW() - ttl_days. +-- 2. app/services/product_handlers.py + app/services/scheduler.py — both +-- deactivate_stale_* dispatch paths read staleness_column from default_params +-- (default 'last_seen_at'). +-- 3. This migration — seeds the two scrape_schedules rows below. +-- +-- Why staleness_column='scraped_at' for domklik (NOT last_seen_at): +-- The daily bulk-touch advances last_seen_at for every domklik row to the same +-- instant, so `last_seen_at < NOW() - INTERVAL '14 days'` matches 0 rows forever — +-- the TTL is useless. scraped_at moves ONLY on a genuine scrape, so it reflects +-- true visibility. TTL=14 mirrors the estimator's LISTINGS_FRESH_DAYS: a listing +-- invisible to valuations for >14 days should also be deactivated, keeping the +-- active set synchronous with what the estimator actually consumes. +-- +-- Why domklik is blanket (no "segments" key), unlike yandex/cian (#115 vtorichka-only): +-- verified on prod 2026-07-02 — ALL 6296 domklik rows are listing_segment='vtorichka'; +-- the BFF sweep ingests secondary only, so there are no novostroyki to protect. +-- +-- Why n1 uses the default last_seen_at (staleness_column omitted): +-- n1 is dead — last_seen_at is frozen at 2026-06-16, so the default column already +-- crosses the TTL threshold. The very first run will deactivate the stale inventory. +-- TTL=14 days keeps it conservative and consistent with domklik. +-- +-- Schedule window 07:00-08:00 UTC: +-- After the avito deactivate_stale window (06:00-07:00 UTC) and after the main city +-- sweeps (02:00-05:00 UTC), same slot as the yandex/cian deactivate rows (#115). +-- Running after the sweeps ensures listings genuinely re-seen are NOT deactivated. +-- +-- next_run_at bootstrapped to tomorrow 07:00 UTC so the scheduler does not fire +-- immediately on deploy (same pattern as 090/115). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- listings table with last_seen_at + scraped_at columns. +-- app/tasks/deactivate_stale_avito.py — deactivate_stale_listings() (staleness_column). +-- app/services/scheduler.py + product_handlers.py — deactivate_stale_* dispatch that +-- forwards staleness_column from default_params. +-- +-- Deploy order: +-- Apply after deploying the scheduler.py + product_handlers.py + task changes so the +-- dispatch can forward staleness_column correctly on the first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'deactivate_stale_domklik', + true, -- SAFE: pure internal DB UPDATE, no ext calls + 7, + 8, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC', + -- staleness_column=scraped_at: last_seen_at bulk-touched daily -> useless TTL. + '{"listing_source":"domklik","ttl_days":14,"staleness_column":"scraped_at"}'::jsonb +), +( + 'deactivate_stale_n1', + true, -- SAFE: pure internal DB UPDATE, no ext calls + 7, + 8, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC', + -- n1 dead (last_seen_at frozen 2026-06-16); default last_seen_at column fires first run. + '{"listing_source":"n1","ttl_days":14}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_listings.py b/tradein-mvp/backend/tests/test_deactivate_stale_listings.py index cae3aeb9..1018896a 100644 --- a/tradein-mvp/backend/tests/test_deactivate_stale_listings.py +++ b/tradein-mvp/backend/tests/test_deactivate_stale_listings.py @@ -303,3 +303,151 @@ def test_migration_115_ttl_30_days() -> None: def test_migration_115_no_psycopg_trap() -> None: sql = _MIGRATION_115.read_text("utf-8") assert not re.search(r":\w+::", sql) + + +# ── staleness_column (#2204: domklik scraped_at vs last_seen_at) ─────────────── + + +def _sql_text(stmt: Any) -> str: + """Достать SQL-текст из TextClause, захваченного _FakeDB.execute.""" + return str(getattr(stmt, "text", stmt)) + + +def test_default_staleness_column_is_last_seen_at(monkeypatch: pytest.MonkeyPatch) -> None: + """Без явного staleness_column WHERE фильтрует по last_seen_at (обратная совместимость).""" + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + db = _FakeDB(rowcount=1) + task_mod.deactivate_stale_listings( + db, run_id=10, listing_source="avito", ttl_days=10, segments=None + ) # type: ignore[arg-type] + stmt, _params = db.executed[0] + sql = _sql_text(stmt) + assert "last_seen_at <" in sql + assert "scraped_at" not in sql + + +def test_scraped_at_column_used_in_all_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None: + """domklik bulk-touch: staleness_column='scraped_at' -> WHERE по scraped_at, не last_seen_at. + + Кейс: last_seen_at свежий (ежедневный bulk UPDATE), но scraped_at заморожен -> + деактивация должна сработать по scraped_at. Проверяем, что именно эта колонка в SQL. + """ + marked: dict[str, Any] = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + db = _FakeDB(rowcount=6296) + out = task_mod.deactivate_stale_listings( + db, + run_id=11, + listing_source="domklik", + ttl_days=14, + segments=None, + staleness_column="scraped_at", + ) # type: ignore[arg-type] + stmt, params = db.executed[0] + sql = _sql_text(stmt) + assert "scraped_at <" in sql + assert "last_seen_at" not in sql + assert params is not None + assert params["listing_source"] == "domklik" + assert params["ttl_days"] == 14 + assert out == {"deactivated": 6296} + assert marked["counters"] == {"deactivated": 6296} + + +def test_scraped_at_column_used_in_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None: + """staleness_column='scraped_at' также уважается в segment-варианте SQL.""" + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + db = _FakeDB(rowcount=0) + task_mod.deactivate_stale_listings( + db, + run_id=12, + listing_source="domklik", + ttl_days=14, + segments=["vtorichka"], + staleness_column="scraped_at", + ) # type: ignore[arg-type] + stmt, _params = db.executed[0] + sql = _sql_text(stmt) + assert "scraped_at <" in sql + assert "listing_segment = ANY(CAST(:segments AS text[]))" in sql + + +def test_invalid_staleness_column_raises_and_marks_failed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Невалидная колонка -> ValueError ДО SQL, run помечен failed, никакого execute.""" + failed: dict[str, Any] = {} + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + task_mod.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + db = _FakeDB(rowcount=0) + with pytest.raises(ValueError, match="invalid staleness_column"): + task_mod.deactivate_stale_listings( + db, + run_id=13, + listing_source="domklik", + ttl_days=14, + segments=None, + staleness_column="created_at; DROP TABLE listings", + ) # type: ignore[arg-type] + # Никакого SQL не выполнено (проверка ДО execute). + assert db.executed == [] + assert db.rolled_back is True + assert failed["run_id"] == 13 + + +def test_allowed_staleness_columns_whitelist() -> None: + assert task_mod._ALLOWED_STALENESS_COLUMNS == frozenset({"last_seen_at", "scraped_at"}) + + +# ── Migration 160 (deactivate_stale_domklik / _n1) ──────────────────────────── + +_MIGRATION_160 = _SQL_DIR / "160_seed_deactivate_stale_domklik_n1.sql" + + +def test_migration_160_exists() -> None: + assert _MIGRATION_160.is_file(), f"missing migration: {_MIGRATION_160}" + + +def test_migration_160_seeds_domklik_and_n1() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert "'deactivate_stale_domklik'" in sql + assert "'deactivate_stale_n1'" in sql + + +def test_migration_160_domklik_uses_scraped_at() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert '"listing_source":"domklik"' in sql + assert '"staleness_column":"scraped_at"' in sql + + +def test_migration_160_n1_default_column_ttl14() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert '"listing_source":"n1"' in sql + assert '"ttl_days":14' in sql + + +def test_migration_160_is_idempotent() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert "ON CONFLICT (source) DO NOTHING" in sql + + +def test_migration_160_is_transactional() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert "BEGIN;" in sql + assert "COMMIT;" in sql + + +def test_migration_160_no_psycopg_trap() -> None: + sql = _MIGRATION_160.read_text("utf-8") + assert not re.search(r":\w+::", sql) -- 2.45.3