diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 2017fff4..696cf7a7 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -23,10 +23,13 @@ Sources: yandex detail pages, extracts full address from , updates listings.address for street-only addresses; nightly 02:00-03:00 UTC, before refresh_search_matview) - - 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) + - deactivate_stale_* → trigger_deactivate_stale_run → deactivate_stale_listings + (tasks/deactivate_stale_avito.py, #759 + segment-aware + generalisation; marks listings with last_seen_at > TTL days as + is_active=false — no DELETE, no external calls. Source prefix + dispatch covers deactivate_stale_avito (all segments, TTL=10) + + deactivate_stale_yandex/cian (vtorichka only, TTL=30, segments + from default_params). Window after the source sweep.) - sber_index_pull → run_sber_index_pull (tasks/sber_index_pull.py, #887; monthly pull of СберИндекс city-level secondary-housing price index — pure HTTP+DB, no @@ -708,6 +711,59 @@ async def trigger_deactivate_stale_avito_run( return run_id +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). + Для источника 'deactivate_stale_avito' с пустым default_params использует defaults: + listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None. + Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments. + + Запускает deactivate_stale_listings в executor (sync DB-only task). + Возвращает run_id или None (skip -- already running). + """ + from app.core.config import settings as _settings + + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + 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 + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.deactivate_stale_avito import deactivate_stale_listings + + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + lambda: deactivate_stale_listings( + run_db, + run_id, + listing_source=listing_source, + ttl_days=ttl_days, + segments=segments, + ), + ) + except Exception: + logger.exception( + "scheduler: deactivate_stale source=%s crashed run_id=%d", + listing_source, + 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 source=%s run_id=%d", listing_source, run_id) + return run_id + + async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch run_sber_index_pull в asyncio.create_task. @@ -1268,8 +1324,8 @@ async def scheduler_loop() -> None: await trigger_refresh_search_matview_run(db, sch) elif source == "yandex_address_backfill": await trigger_yandex_address_backfill_run(db, sch) - elif source == "deactivate_stale_avito": - await trigger_deactivate_stale_avito_run(db, sch) + elif source.startswith("deactivate_stale_"): + await trigger_deactivate_stale_run(db, sch) elif source == "sber_index_pull": await trigger_sber_index_pull_run(db, sch) elif source == "rosreestr_quarter_poll": diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py index 88386463..216f5b78 100644 --- a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -1,22 +1,27 @@ -"""Deactivate stale avito listings that have not been seen in >TTL days (#759). +"""Deactivate stale listings that have not been seen in >TTL days. -Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются -is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми -объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667). +Первоначально создано для avito (#759). Теперь поддерживает любой источник +(listing_source) с опциональной фильтрацией по listing_segment. -Только source='avito'. Cian/Yandex не затрагиваются. +Ключевые решения: +- Cian/Yandex не поддерживают full-coverage sweep -> паушальный TTL сломает живой + инвентарь. DECISION: для yandex/cian деактивировать ТОЛЬКО listing_segment='vtorichka', + TTL=30. novostroyki (9659 активных первичных строк) и NULL-сегмент не трогаем. +- avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений. +- Строки НЕ удаляются -- история нужна для бэктеста (#667). -Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app -scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py), +Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается in-app +scheduler'ом через trigger_deactivate_stale_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). +TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10). """ from __future__ import annotations import logging +from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session @@ -26,45 +31,107 @@ 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 без фильтра по сегменту: все сегменты для данного 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 = 'avito' + WHERE source = :listing_source AND is_active = true AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval) """ ) +# UPDATE с фильтром по сегменту (segments задан): затрагивает только указанные сегменты. +# = ANY(CAST(:segments AS text[])) -- psycopg v3 адаптирует Python list -> text[]. +_DEACTIVATE_SQL_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) + AND listing_segment = ANY(CAST(:segments AS text[])) + """ +) -def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]: - """Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад. +# Алиас для обратной совместимости -- использовался в тестах через task_mod._DEACTIVATE_SQL +_DEACTIVATE_SQL = _DEACTIVATE_SQL_ALL_SEGMENTS + + +def deactivate_stale_listings( + db: Session, + run_id: int, + *, + listing_source: str, + ttl_days: int, + segments: list[str] | None = None, +) -> dict[str, int]: + """Пометить is_active=false объявления с last_seen_at > ttl_days дней назад. + + Параметры: + listing_source: значение поля source в таблице listings ('avito', 'yandex', 'cian'). + ttl_days: количество дней TTL; объявления старше этого порога деактивируются. + segments: если задан -- деактивировать только объявления с указанными + listing_segment значениями. None -> все сегменты (поведение avito по умолчанию). Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed). - Returns {"deactivated": N} — количество обновлённых строк. + Returns {"deactivated": N} -- количество обновлённых строк. """ counters: dict[str, int] = {"deactivated": 0} try: - result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days}) + # segments is None -> все сегменты (поведение avito). segments=[...] -> только + # перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список [] + # означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0), + # а НЕ "все сегменты" — иначе случайный [] стёр бы весь источник. + if segments is not None: + params: dict[str, Any] = { + "listing_source": listing_source, + "ttl_days": ttl_days, + "segments": segments, + } + result = db.execute(_DEACTIVATE_SQL_SEGMENTS, params) + else: + params = {"listing_source": listing_source, "ttl_days": ttl_days} + result = db.execute(_DEACTIVATE_SQL_ALL_SEGMENTS, params) + 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)", + "deactivate_stale source=%s run_id=%d done: deactivated=%d " + "(ttl_days=%d, segments=%r)", + listing_source, run_id, counters["deactivated"], - settings.avito_stale_ttl_days, + ttl_days, + segments, ) return counters except Exception as exc: - logger.exception("deactivate_stale_avito run_id=%d failed", run_id) + logger.exception("deactivate_stale source=%s run_id=%d failed", listing_source, run_id) db.rollback() runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) raise + + +def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]: + """Обёртка для обратной совместимости -- deactivate_stale_avito (все сегменты, TTL из settings). + + Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад. + Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). + Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed). + + Returns {"deactivated": N} -- количество обновлённых строк. + """ + return deactivate_stale_listings( + db, + run_id, + listing_source="avito", + ttl_days=settings.avito_stale_ttl_days, + segments=None, + ) diff --git a/tradein-mvp/backend/data/sql/115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql b/tradein-mvp/backend/data/sql/115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql new file mode 100644 index 00000000..78820916 --- /dev/null +++ b/tradein-mvp/backend/data/sql/115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql @@ -0,0 +1,68 @@ +-- 115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql +-- Seed scrape_schedules rows for nightly segment-aware stale-listing deactivation +-- of yandex and cian vtorichka (#759 generalisation). +-- +-- Rationale (segment-aware vs blanket TTL): +-- Yandex and Cian city sweeps do NOT maintain full inventory coverage -- they +-- sample a rolling window rather than crawling every live listing on every run. +-- A blanket TTL (as used for Avito) would therefore wrongly deactivate live +-- novostroyki and unknown-segment (NULL) rows that simply were not re-visited +-- in the sample window. +-- +-- DECISION (user-approved): deactivate ONLY listing_segment='vtorichka' rows, +-- with TTL=30 days. This is conservative enough that at first deploy it will +-- touch 0 rows (no vtorichka listing has been absent for 30 days yet), but will +-- catch genuinely-stale secondhand inventory going forward. +-- +-- novostroyki rows (9659 active first-party rows as of 2026-06-16) are protected: +-- they are managed by dedicated newbuilding pipelines and must not be deactivated +-- by this task. NULL-segment rows are also excluded as their segment is unknown. +-- +-- Schedule window 07:00-08:00 UTC: +-- After avito deactivate_stale (06:00-07:00 UTC) and after the main city sweeps +-- (02:00-05:00 UTC) and Rosreestr OS-cron (05:30/05:45 UTC). +-- Pattern: next_run_at bootstrapped to tomorrow 07:00 UTC so the scheduler does +-- not fire immediately on deploy. +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING -- safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- listings table with listing_segment column. +-- app/tasks/deactivate_stale_avito.py -- deactivate_stale_listings() (generic fn). +-- app/services/scheduler.py -- trigger_deactivate_stale_run() + dispatch branch +-- source.startswith("deactivate_stale_"). +-- +-- Deploy order: +-- Apply after deploying the scheduler.py + task changes. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'deactivate_stale_yandex', + true, + 7, + 8, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC', + '{"listing_source":"yandex","ttl_days":30,"segments":["vtorichka"]}'::jsonb +), +( + 'deactivate_stale_cian', + true, + 7, + 8, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC', + '{"listing_source":"cian","ttl_days":30,"segments":["vtorichka"]}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_avito.py b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py index ee1da4ab..b9ae6189 100644 --- a/tradein-mvp/backend/tests/test_deactivate_stale_avito.py +++ b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py @@ -25,14 +25,18 @@ _MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sq _DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text) _TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings) +_GENERIC_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_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 + """Compat wrapper enforces source='avito' via listing_source param — cian/yandex not touched.""" + # _DEACTIVATE_SQL is now the generic all-segments SQL using :listing_source bind param. + assert ":listing_source" in _DEACTIVATE_SQL + # The avito constraint is enforced by the compat wrapper passing listing_source="avito". + assert 'listing_source="avito"' in _TASK_SRC def test_sql_only_deactivates_not_deletes() -> None: @@ -83,17 +87,20 @@ def test_task_reads_ttl_from_settings() -> None: def test_task_returns_deactivated_counter() -> None: - assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC + # counter logic lives in the generic deactivate_stale_listings function + assert '"deactivated"' in _GENERIC_TASK_SRC or "'deactivated'" in _GENERIC_TASK_SRC def test_task_finalises_run() -> None: - assert "mark_done" in _TASK_SRC - assert "mark_failed" in _TASK_SRC + # mark_done/mark_failed live in the generic deactivate_stale_listings function + assert "mark_done" in _GENERIC_TASK_SRC + assert "mark_failed" in _GENERIC_TASK_SRC def test_task_commits_and_rollbacks_on_error() -> None: - assert "db.commit()" in _TASK_SRC - assert "db.rollback()" in _TASK_SRC + # commit/rollback live in the generic deactivate_stale_listings function + assert "db.commit()" in _GENERIC_TASK_SRC + assert "db.rollback()" in _GENERIC_TASK_SRC # ── Scheduler wiring ────────────────────────────────────────────────────────── @@ -106,10 +113,10 @@ def test_trigger_fn_exists_and_is_async() -> None: def test_dispatch_branch_wired() -> None: - """scheduler_loop dispatches source='deactivate_stale_avito' to the trigger.""" + """scheduler_loop dispatches deactivate_stale_* sources to the generic 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 + assert 'source.startswith("deactivate_stale_")' in loop_src + assert "trigger_deactivate_stale_run(db, sch)" in loop_src def test_trigger_runs_sync_task_in_executor() -> None: diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_listings.py b/tradein-mvp/backend/tests/test_deactivate_stale_listings.py new file mode 100644 index 00000000..cae3aeb9 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deactivate_stale_listings.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +import inspect +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +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_115 = _SQL_DIR / "115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql" + +_GENERIC_FN_SRC = inspect.getsource(task_mod.deactivate_stale_listings) +_COMPAT_FN_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings) + + +def test_segment_sql_uses_cast_interval_not_double_colon() -> None: + sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text + assert "CAST(:ttl_days || ' days' AS interval)" in sql + assert not re.search(r"\w+::", sql) + + +def test_segment_sql_uses_any_cast_text_array() -> None: + sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text + assert "ANY(CAST(:segments AS text[]))" in sql + + +def test_segment_sql_filters_is_active_true() -> None: + sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text + assert "is_active = true" in sql + + +def test_segment_sql_only_deactivates_not_deletes() -> None: + sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text + assert "SET is_active = false" in sql + assert "DELETE" not in sql.upper() + + +def test_all_segments_sql_uses_listing_source_param() -> None: + sql = task_mod._DEACTIVATE_SQL_ALL_SEGMENTS.text + assert ":listing_source" in sql + assert "source = 'avito'" not in sql + + +def test_deactivate_sql_alias_is_all_segments() -> None: + assert task_mod._DEACTIVATE_SQL is task_mod._DEACTIVATE_SQL_ALL_SEGMENTS + + +def test_generic_fn_exists() -> None: + assert callable(task_mod.deactivate_stale_listings) + + +def test_generic_fn_returns_deactivated_counter() -> None: + assert "deactivated" in _GENERIC_FN_SRC + + +def test_generic_fn_finalises_run() -> None: + assert "mark_done" in _GENERIC_FN_SRC + assert "mark_failed" in _GENERIC_FN_SRC + + +def test_generic_fn_commits_and_rollbacks() -> None: + assert "db.commit()" in _GENERIC_FN_SRC + assert "db.rollback()" in _GENERIC_FN_SRC + + +def test_compat_wrapper_calls_generic_with_avito_defaults() -> None: + assert "deactivate_stale_listings(" in _COMPAT_FN_SRC + assert 'listing_source="avito"' in _COMPAT_FN_SRC + assert "settings.avito_stale_ttl_days" in _COMPAT_FN_SRC + assert "segments=None" in _COMPAT_FN_SRC + + +def test_trigger_deactivate_stale_run_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_deactivate_stale_run", None) + assert fn is not None, "trigger_deactivate_stale_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_uses_startswith() -> None: + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source.startswith("deactivate_stale_")' in loop_src + assert "trigger_deactivate_stale_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run) + assert "run_in_executor(" in trig_src + assert "deactivate_stale_listings" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +def test_trigger_reads_params_from_default_params() -> None: + trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run) + assert "listing_source" in trig_src + assert "ttl_days" in trig_src + assert "segments" in trig_src + + +class _FakeResult: + def __init__(self, rowcount: int) -> None: + self.rowcount = rowcount + + +class _FakeDB: + def __init__(self, rowcount: int = 0) -> None: + self._rowcount = rowcount + self.executed: list[tuple[Any, 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_vtorichka_stale_beyond_ttl_deactivated(monkeypatch: pytest.MonkeyPatch) -> None: + 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=5) + out = task_mod.deactivate_stale_listings( + db, run_id=1, listing_source="yandex", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + assert out == {"deactivated": 5} + assert db.committed is True + _stmt, params = db.executed[0] + assert params is not None + assert params["segments"] == ["vtorichka"] + assert params["listing_source"] == "yandex" + assert params["ttl_days"] == 30 + assert marked["counters"] == {"deactivated": 5} + + +def test_novostroyki_not_in_segment_filter(monkeypatch: pytest.MonkeyPatch) -> None: + 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=2, listing_source="yandex", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + _stmt, params = db.executed[0] + assert params is not None + assert "segments" in params + assert "novostroyki" not in params["segments"] + + +def test_null_segment_not_in_filter(monkeypatch: pytest.MonkeyPatch) -> None: + 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=3, listing_source="cian", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + _stmt, params = db.executed[0] + assert params is not None + assert None not in params["segments"] + + +def test_other_source_listing_source_param_matches(monkeypatch: pytest.MonkeyPatch) -> None: + 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=4, listing_source="yandex", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + _stmt, params = db.executed[0] + assert params is not None + assert params["listing_source"] == "yandex" + + +def test_fresh_listing_zero_rowcount(monkeypatch: pytest.MonkeyPatch) -> None: + 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_listings( + db, run_id=5, listing_source="cian", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + assert out == {"deactivated": 0} + assert db.committed is True + assert marked["counters"] == {"deactivated": 0} + + +def test_segments_none_uses_all_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None: + 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=3) + out = task_mod.deactivate_stale_listings( + db, run_id=6, listing_source="avito", ttl_days=10, segments=None + ) # type: ignore[arg-type] + assert out == {"deactivated": 3} + _stmt, params = db.executed[0] + assert params is not None + assert "segments" not in params + assert params["listing_source"] == "avito" + assert params["ttl_days"] == 10 + + +def test_avito_compat_wrapper_counter(monkeypatch: pytest.MonkeyPatch) -> None: + 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=42) + out = task_mod.deactivate_stale_avito_listings(db, run_id=7) # type: ignore[arg-type] + assert out == {"deactivated": 42} + assert db.committed is True + _stmt, params = db.executed[0] + assert params is not None + assert params["listing_source"] == "avito" + assert "segments" not in params + assert marked["run_id"] == 7 + + +def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None: + 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() + with pytest.raises(RuntimeError, match="db exploded"): + task_mod.deactivate_stale_listings( + db, run_id=8, listing_source="yandex", ttl_days=30, segments=["vtorichka"] + ) # type: ignore[arg-type] + assert db.rolled_back is True + assert failed["run_id"] == 8 + + +def test_migration_115_exists() -> None: + assert _MIGRATION_115.is_file(), f"missing migration: {_MIGRATION_115}" + + +def test_migration_115_seeds_yandex_and_cian() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "'deactivate_stale_yandex'" in sql + assert "'deactivate_stale_cian'" in sql + + +def test_migration_115_is_idempotent() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "ON CONFLICT (source) DO NOTHING" in sql + + +def test_migration_115_enabled_true() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "true" in sql + + +def test_migration_115_is_transactional() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "BEGIN;" in sql + assert "COMMIT;" in sql + + +def test_migration_115_window_7_to_8_utc() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert re.search(r"\b7\b", sql), "window_start_hour 7 missing" + assert re.search(r"\b8\b", sql), "window_end_hour 8 missing" + + +def test_migration_115_segments_vtorichka_in_default_params() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "vtorichka" in sql + + +def test_migration_115_ttl_30_days() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert "30" in sql + + +def test_migration_115_no_psycopg_trap() -> None: + sql = _MIGRATION_115.read_text("utf-8") + assert not re.search(r":\w+::", sql)