From 652ea75f546502d0392f9bba2190151bd4442d6e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 31 May 2026 10:10:59 +0300 Subject: [PATCH] feat(tradein): nightly task to deactivate stale avito listings (#759) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Avito listings re-seen by the scraper get is_active=true but were never deactivated when they stop appearing (sold/removed) — unlike cian's snapshot scope. Result: all avito rows stay is_active=true, inflating the coverage denominator and polluting estimator analog medians with dead lots. Add scheduled task deactivate_stale_avito_listings: marks source='avito' is_active=false when last_seen_at older than AVITO_STALE_TTL_DAYS (10d), window 06:00-07:00 UTC (after avito_city_sweep). No DELETE — history kept for backtest #667. cian/yandex untouched. Wired into scheduler_loop dispatch + seed migration 090 (ON CONFLICT(source) idempotent). Refs #759 --- tradein-mvp/backend/app/core/config.py | 7 + tradein-mvp/backend/app/services/scheduler.py | 41 +++ .../app/tasks/deactivate_stale_avito.py | 70 +++++ ..._schedules_seed_deactivate_stale_avito.sql | 65 +++++ .../tests/test_deactivate_stale_avito.py | 254 ++++++++++++++++++ 5 files changed, 437 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/deactivate_stale_avito.py create mode 100644 tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql create mode 100644 tradein-mvp/backend/tests/test_deactivate_stale_avito.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 39d61f86..d47bf69f 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -182,6 +182,13 @@ class Settings(BaseSettings): # Сколько раз сменить IP при блоке прежде чем сдаться (на одну страницу). avito_proxy_max_rotations: int = 2 + # ── #759: deactivate stale avito listings ─────────────────────────────── + # Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней, + # помечаются is_active=false ночной задачей deactivate_stale_avito_listings. + # TTL=10 — баланс между "снятое объявление" (обычно 3-7 дней молчания) и + # допуском на перерыв в работе scraper'а. ENV: AVITO_STALE_TTL_DAYS. + avito_stale_ttl_days: int = 10 + # ── Yandex SERP cookies (#801/T4) ─────────────────────────────────────── # Путь к JSON-файлу с cookies браузера (формат: [{name, value, ...}, ...]). # Если задан и файл существует — cookies передаются в curl_cffi-сессию при diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index fbff867c..8433ed74 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -19,6 +19,10 @@ Sources: (tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB re-derivation of the asking→sold ratios — no external calls, enabled by default; window after rosreestr_dkp_import) + - deactivate_stale_avito → deactivate_stale_avito_listings + (tasks/deactivate_stale_avito.py, #759; marks avito listings + with last_seen_at > TTL days as is_active=false — no DELETE, + no external calls, enabled by default; window after avito sweep) """ from __future__ import annotations @@ -513,6 +517,41 @@ async def trigger_refresh_search_matview_run( return run_id +async def trigger_deactivate_stale_avito_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch deactivate_stale_avito_listings в executor (sync DB-only task). + + Помечает avito-объявления, не виденные >TTL дней, как is_active=false (#759). + Задача синхронная (чистый internal UPDATE, БЕЗ внешних HTTP-вызовов) — гоняем в + run_in_executor по образцу trigger_listing_source_snapshot_run. SAFE to enable — + schedule seed enabled=true (090), pure internal DB op. + Окно 06:00-07:00 UTC — после avito_city_sweep (02:00-05:00 UTC). + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.deactivate_stale_avito import deactivate_stale_avito_listings + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, deactivate_stale_avito_listings, run_db, run_id) + except Exception: + logger.exception("scheduler: deactivate_stale_avito crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered deactivate_stale_avito run_id=%d", run_id) + return run_id + + async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task). @@ -829,6 +868,8 @@ async def scheduler_loop() -> None: await trigger_asking_to_sold_ratio_run(db, sch) elif source == "refresh_search_matview": await trigger_refresh_search_matview_run(db, sch) + elif source == "deactivate_stale_avito": + await trigger_deactivate_stale_avito_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py new file mode 100644 index 00000000..88386463 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -0,0 +1,70 @@ +"""Deactivate stale avito listings that have not been seen in >TTL days (#759). + +Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются +is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми +объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667). + +Только source='avito'. Cian/Yandex не затрагиваются. + +Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app +scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py), +по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios +(sync task в run_in_executor). + +TTL берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10). +""" + +from __future__ import annotations + +import logging + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.config import settings +from app.services import scrape_runs as runs_mod + +logger = logging.getLogger(__name__) + +# UPDATE: помечаем is_active=false для avito-объявлений, не виденных >TTL дней. +# CAST(:ttl_days || ' days' AS interval) — psycopg v3 safe (никаких :param::type). +# Только source='avito' и is_active=true — чтобы не трогать уже неактивные строки +# и не затрагивать cian/yandex. +_DEACTIVATE_SQL = text( + """ + UPDATE listings + SET is_active = false + WHERE source = 'avito' + AND is_active = true + AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval) + """ +) + + +def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]: + """Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад. + + Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). + Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed). + + Returns {"deactivated": N} — количество обновлённых строк. + """ + counters: dict[str, int] = {"deactivated": 0} + try: + result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days}) + counters["deactivated"] = result.rowcount or 0 + + db.commit() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "deactivate_stale_avito run_id=%d done: deactivated=%d (ttl_days=%d)", + run_id, + counters["deactivated"], + settings.avito_stale_ttl_days, + ) + return counters + except Exception as exc: + logger.exception("deactivate_stale_avito run_id=%d failed", run_id) + db.rollback() + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql b/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql new file mode 100644 index 00000000..c741e58a --- /dev/null +++ b/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql @@ -0,0 +1,65 @@ +-- 090_scrape_schedules_seed_deactivate_stale_avito.sql +-- Issue #759 — seed scrape_schedules row for nightly deactivation of stale avito listings. +-- +-- Context: +-- Avito listings that haven't been seen by the scraper for >AVITO_STALE_TTL_DAYS days +-- (default 10) remain is_active=true indefinitely. This inflates the active-listing +-- denominator used by the coverage and estimator services, degrading accuracy. +-- +-- Fix: +-- 1. app/tasks/deactivate_stale_avito.py — deactivate_stale_avito_listings() +-- issues a single UPDATE listings SET is_active=false WHERE source='avito' +-- AND is_active=true AND last_seen_at < NOW()-INTERVAL(:ttl_days days). +-- Rows are NOT deleted — history is required for backtest (#667). +-- 2. scheduler.py — added trigger_deactivate_stale_avito_run() + dispatch branch +-- `elif source == "deactivate_stale_avito"` in scheduler_loop(). +-- 3. app/core/config.py — AVITO_STALE_TTL_DAYS setting (default 10). +-- 4. This migration — seeds the scrape_schedules row. +-- +-- Schedule window 06:00-07:00 UTC: +-- After the avito_city_sweep window (02:00-05:00 UTC per #814). Running after the +-- sweep ensures listings genuinely seen in the nightly pass are NOT deactivated. +-- +-- next_run_at bootstrapped to tomorrow 06:00 UTC so the scheduler does not fire +-- immediately on deploy (same pattern as 079/082/088). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- listings table with last_seen_at column (002_core_tables.sql). +-- scheduler.py change (trigger_deactivate_stale_avito_run must be deployed). +-- +-- Deploy order: +-- Apply this migration after deploying the scheduler.py + task changes so the +-- scheduler can dispatch the new source name correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'deactivate_stale_avito', + true, -- SAFE: pure internal DB UPDATE, no ext calls + 6, + 7, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC', + '{}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), ' + 'deactivate_stale_avito (#759).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_avito.py b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py new file mode 100644 index 00000000..aed99581 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py @@ -0,0 +1,254 @@ +"""Tests for the stale avito listing deactivation task (#759). + +Static assertions (SQL shape, psycopg-v3 cast discipline, scheduler wiring) mirror +the pattern established in test_listing_source_snapshot.py. One behavioural test drives +the counter logic via a fake db without a live database. +""" + +import inspect +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +# Stub DATABASE_URL before any app import (same guard as test_scheduler.py / +# test_listing_source_snapshot.py — these tests are static/fake-db, no live DB). +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scheduler +from app.tasks import deactivate_stale_avito as task_mod + +_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" +_MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sql" + +_DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text) +_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings) + + +# ── SQL shape ───────────────────────────────────────────────────────────────── + + +def test_sql_targets_only_avito_source() -> None: + """UPDATE must filter source='avito' — cian/yandex rows must not be touched.""" + assert "source = 'avito'" in _DEACTIVATE_SQL + + +def test_sql_only_deactivates_not_deletes() -> None: + """Rows must be marked is_active=false, never DELETEd (history needed for #667).""" + assert "SET is_active = false" in _DEACTIVATE_SQL + assert "DELETE" not in _DEACTIVATE_SQL.upper() + + +def test_sql_guards_already_inactive_rows() -> None: + """Filter AND is_active=true avoids redundant updates on rows already deactivated.""" + assert "is_active = true" in _DEACTIVATE_SQL + + +def test_sql_uses_last_seen_at_threshold() -> None: + """Threshold is last_seen_at < NOW() - interval, not any other column.""" + assert "last_seen_at" in _DEACTIVATE_SQL + assert "NOW()" in _DEACTIVATE_SQL + + +def test_sql_uses_cast_interval_not_double_colon() -> None: + """psycopg v3: bind param via CAST(:ttl_days || ' days' AS interval), never ::interval.""" + assert "CAST(:ttl_days || ' days' AS interval)" in _DEACTIVATE_SQL + # No bind-param (:name) followed by ::type anywhere. + assert not re.search(r":\w+::", _DEACTIVATE_SQL) + + +def test_sql_ttl_param_named_ttl_days() -> None: + """Bind param must be :ttl_days (wired from settings.avito_stale_ttl_days).""" + assert ":ttl_days" in _DEACTIVATE_SQL + + +# ── Settings wiring ─────────────────────────────────────────────────────────── + + +def test_settings_has_avito_stale_ttl_days() -> None: + from app.core.config import settings + + assert hasattr(settings, "avito_stale_ttl_days") + assert settings.avito_stale_ttl_days == 10 # default + + +def test_task_reads_ttl_from_settings() -> None: + """Task must use settings.avito_stale_ttl_days, not a hardcoded literal.""" + assert "settings.avito_stale_ttl_days" in _TASK_SRC + + +# ── Task function contract ──────────────────────────────────────────────────── + + +def test_task_returns_deactivated_counter() -> None: + assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC + + +def test_task_finalises_run() -> None: + assert "mark_done" in _TASK_SRC + assert "mark_failed" in _TASK_SRC + + +def test_task_commits_and_rollbacks_on_error() -> None: + assert "db.commit()" in _TASK_SRC + assert "db.rollback()" in _TASK_SRC + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_trigger_fn_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_deactivate_stale_avito_run", None) + assert fn is not None, "trigger_deactivate_stale_avito_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_wired() -> None: + """scheduler_loop dispatches source='deactivate_stale_avito' to the trigger.""" + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "deactivate_stale_avito"' in loop_src + assert "trigger_deactivate_stale_avito_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + """Sync DB-only task → run_in_executor, not a bare await.""" + trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_avito_run) + # run_in_executor call may span two lines; check both parts separately. + assert "run_in_executor(" in trig_src + assert "deactivate_stale_avito_listings" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +# ── Migration 090 ───────────────────────────────────────────────────────────── + + +def test_migration_090_exists() -> None: + assert _MIGRATION_090.is_file(), f"missing migration: {_MIGRATION_090}" + + +def test_migration_090_seeds_correct_source_and_window() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "'deactivate_stale_avito'" in sql + # Window 06:00-07:00 UTC — after avito sweep (02:00-05:00). + assert re.search(r"\b6\b", sql), "window_start_hour 6 missing" + assert re.search(r"\b7\b", sql), "window_end_hour 7 missing" + + +def test_migration_090_is_idempotent() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "ON CONFLICT (source) DO NOTHING" in sql + + +def test_migration_090_enabled_true() -> None: + sql = _MIGRATION_090.read_text("utf-8") + insert_block = sql.split("INSERT INTO scrape_schedules")[1] + # enabled=true (SAFE — pure internal DB UPDATE). + assert "true," in insert_block + + +def test_migration_090_is_transactional() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "BEGIN;" in sql + assert "COMMIT;" in sql + + +def test_migration_090_no_psycopg_trap() -> None: + """Migration is plain DDL/DML seed (no bind params), guard :x::type trap anyway.""" + sql = _MIGRATION_090.read_text("utf-8") + assert not re.search(r":\w+::", sql) + + +# ── Behavioural test: counter logic via fake db ─────────────────────────────── + + +class _FakeResult: + def __init__(self, rowcount: int) -> None: + self.rowcount = rowcount + + +class _FakeDB: + """Minimal stand-in for SQLAlchemy Session.""" + + def __init__(self, rowcount: int) -> None: + self._rowcount = rowcount + self.executed: list[Any] = [] + self.committed = False + self.rolled_back = False + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + self.executed.append((stmt, params)) + return _FakeResult(self._rowcount) + + def commit(self) -> None: + self.committed = True + + def rollback(self) -> None: + self.rolled_back = True + + +def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None: + """deactivate_stale_avito_listings maps execute() rowcount → counters + marks done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + db = _FakeDB(rowcount=137) + out = task_mod.deactivate_stale_avito_listings(db, run_id=42) # type: ignore[arg-type] + + assert out == {"deactivated": 137} + assert db.committed is True + assert len(db.executed) == 1 + # Bind params must contain ttl_days wired from settings. + _stmt, params = db.executed[0] + assert params is not None + assert "ttl_days" in params + assert params["ttl_days"] == 10 # default from settings + # run_id finalised via mark_done. + assert marked["run_id"] == 42 + assert marked["counters"] == {"deactivated": 137} + + +def test_counter_logic_zero_rows(monkeypatch: pytest.MonkeyPatch) -> None: + """rowcount=0 → deactivated=0, still commits and marks done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + db = _FakeDB(rowcount=0) + out = task_mod.deactivate_stale_avito_listings(db, run_id=1) # type: ignore[arg-type] + + assert out == {"deactivated": 0} + assert db.committed is True + assert marked["counters"] == {"deactivated": 0} + + +def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None: + """On execute error: rollback + mark_failed + re-raise (no silent swallow).""" + 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), + ) + + class _BoomDB(_FakeDB): + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + raise RuntimeError("db exploded") + + db = _BoomDB(rowcount=0) + with pytest.raises(RuntimeError, match="db exploded"): + task_mod.deactivate_stale_avito_listings(db, run_id=7) # type: ignore[arg-type] + + assert db.rolled_back is True + assert failed["run_id"] == 7