diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py index 29b6aaab..64ba11e2 100644 --- a/tradein-mvp/backend/app/services/product_handlers.py +++ b/tradein-mvp/backend/app/services/product_handlers.py @@ -218,6 +218,7 @@ async def _job_deactivate_stale( from app.core.config import settings as _settings from app.tasks.deactivate_stale_avito import ( DEFAULT_MIN_CONFIRMATIONS, + DEFAULT_REVISIT_FLOOR_QUANTILE, deactivate_stale_listings, ) @@ -229,6 +230,12 @@ async def _job_deactivate_stale( # получает страховочный порог, а не «деактивируй вслепую». Посчитанные по # источнику пороги приходят из default_params (миграция 219). min_confirmations: int = params.get("min_confirmations", DEFAULT_MIN_CONFIRMATIONS) + # Пол TTL по измеренному циклу переобхода (#2659) — тоже включён по умолчанию: + # незасеянное расписание не должно снимать объявления по порогу ниже собственного + # хвоста обхода. Снять ручку вручную: revisit_floor_quantile = 0. + revisit_floor_quantile: float = params.get( + "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE + ) loop = asyncio.get_event_loop() await loop.run_in_executor( @@ -241,6 +248,7 @@ async def _job_deactivate_stale( segments=segments, staleness_column=staleness_column, min_confirmations=min_confirmations, + revisit_floor_quantile=revisit_floor_quantile, ), ) diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py index 1f800160..cdab053b 100644 --- a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -24,6 +24,7 @@ TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_ST from __future__ import annotations import logging +from math import ceil from typing import Any from sqlalchemy import text @@ -127,6 +128,107 @@ DEFAULT_MIN_CONFIRMATIONS = 500 _CONFIRMATIONS_SEGMENT_FILTER = "\n AND listing_segment = ANY(CAST(:segments AS text[]))" +# ── Пол TTL по измеренному циклу переобхода (#2659) ─────────────────────────── +# Гейт выше отвечает на вопрос «источник вообще собирается?». Он НЕ отвечает на +# вопрос, из-за которого заведён #2659: «а достаточно ли ttl_days, чтобы молчание +# означало снятие?». Пока свип возвращается к строке реже, чем раз в ttl_days, +# TTL меряет НАШУ выборку, а не жизнь объявления, — и источник при этом полностью +# здоров, так что гейт молчит. +# +# ЗАМЕР НА ПРОДЕ 2026-08-09, из-за которого этот пол существует. +# С момента деплоя гейта (06.08) TTL снял 1 028 строк; 127 из них (12.4%) УЖЕ снова +# активны — свип нашёл их живыми через 1-3 суток и вернул сам (upsert в +# scraper_kit/base.py ставит is_active = true). В единственном городе с настоящим +# покрытием доля ложных снятий 100%: +# cian Екатеринбург 103 снято → 103 снова активны +# yandex Екатеринбург 24 снято → 24 снова активны +# cian/yandex без города 901 снято → 0 вернулись (их свип не обходит вовсе) +# Возраст на момент снятия у всех 127: 29.9..30.3 суток при TTL=30 — то есть TTL +# срабатывал ровно на границе, а свип возвращался к строке на 31-34-е сутки. +# +# ПОЧЕМУ ЭТО НЕ ЛЕЧИТСЯ НОВОЙ КОНСТАНТОЙ. Разрывы переобхода, суток +# (listing_source_snapshots, 40 суток, посчитано по срезу TTL-джобы): +# источник/сегмент p90 p99 TTL сейчас TTL/p99 +# domklik vtorichka 1.9 3.1 14 4.5 ← сплошное суточное покрытие +# cian vtorichka 10.9 26.6 30 1.1 +# yandex vtorichka 5.7 43.0 30 0.7 +# avito vtorichka 29.1 42.1 10 0.24 ← отсюда 9 033 строки +# Домклик — контрольная группа: при почти полном суточном обходе TTL=14 лежит в +# 4.5 раза выше хвоста, и снятие у него действительно означает снятие. У остальных +# трёх порог ниже собственного хвоста обхода — руками подобранное число и есть +# корень #2659, поэтому чинить его вторым руками подобранным числом бессмысленно. +# +# ЧТО МЕРЯЕМ ВМЕСТО КОНСТАНТЫ: факт, а не оценку. «Какой самый большой возраст, при +# котором свип за последнее окно ДОКАЗАЛ, что объявление живо» — то есть насколько +# старую строку он только что нашёл на площадке. Если свип буквально вчера вернул к +# жизни строку, молчавшую 40 суток, то 30 суток молчания не доказывают ничего. +# Пол = квантиль этого распределения, эффективный TTL = max(ttl_days, пол). +# +# Считается по ТОМУ ЖЕ срезу (source + segments) и по ТОЙ ЖЕ колонке свежести, что +# и UPDATE. Предыдущее наблюдение берётся из listing_source_snapshots — единственной +# истории свежести, что у нас есть; расхождение listings. и +# listing_sources.last_seen_at замерено на проде и не превышает 0.5 суток в среднем +# (максимум 0), что на шкале 30-70 суток шум. +# +# КВАНТИЛЬ — калибровочная ручка, не догма. 0.99 подобран по требованию «пол обязан +# накрыть 127 доказанных ложных снятий», у которых возраст был 29.9..30.3: замер +# того же запроса на проде даёт 34.0 для cian/vtorichka и 74.3 для yandex/vtorichka. +# Ниже 0.99 опускать нельзя без нового замера. Ручка живёт в default_params +# расписания (revisit_floor_quantile), 0 -> пол выключен. +# +# ПОБОЧНЫЙ ЭФФЕКТ, КОТОРЫЙ ЗДЕСЬ НАМЕРЕННЫЙ: после провала сбора хвост разрывов +# распухает (свип разгребает завал и находит очень старые строки), пол поднимается, +# и деактивация замирает сама — без отдельного детектора банов. Когда завал разобран, +# хвост схлопывается и пол опускается обратно. Это ровно то поведение, которого +# issue просил от «гейта по банам», но выраженное через результат, а не через причину. +# +# ПОТОЛОК: пол не может превысить глубину истории снимков. Если снимок за нужную +# дату не писался (дыры на проде есть — 30.07, 01.08), берётся ближайший более +# ранний; при полном отсутствии снимков пол не считается и TTL остаётся как задан. +DEFAULT_REVISIT_FLOOR_QUANTILE = 0.99 + +_REVISIT_FLOOR_SEGMENT_FILTER = "\n AND l.listing_segment = ANY(CAST(:segments AS text[]))" + + +def _build_revisit_floor_sql(staleness_column: str, *, with_segments: bool) -> Any: + """Квантиль возраста, при котором свип за окно ДОКАЗАЛ, что строка жива. + + Пара «предыдущее наблюдение (снимок) → текущее наблюдение (listings)» даёт + разрыв переобхода в сутках; берём его квантиль по срезу source+segments. + Только строки, у которых свежесть реально сдвинулась, — то есть выжившие, + а не «мы к ним не приходили». + + staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings. + Значения — param-binding, psycopg v3 safe (CAST(... AS ...), никаких :param::type). + """ + segment_filter = _REVISIT_FLOOR_SEGMENT_FILTER if with_segments else "" + return text( + f""" + SELECT percentile_disc(CAST(:revisit_quantile AS double precision)) + WITHIN GROUP ( + ORDER BY EXTRACT(epoch FROM (l.{staleness_column} - prev.last_seen_at)) + / 86400.0 + ) + FROM listings l + JOIN listing_sources ls + ON ls.listing_id = l.id + AND ls.ext_source = l.source + JOIN listing_source_snapshots prev + ON prev.listing_source_id = ls.id + AND prev.snapshot_date = ( + SELECT max(snapshot_date) + FROM listing_source_snapshots + WHERE snapshot_date + <= CURRENT_DATE - CAST(:health_window_days AS integer) + ) + WHERE l.source = :listing_source + AND l.{staleness_column} + > NOW() - CAST(:health_window_days || ' days' AS interval) + AND l.{staleness_column} > prev.last_seen_at{segment_filter} + """ + ) + + def _build_confirmations_sql(staleness_column: str, *, with_segments: bool) -> Any: """SELECT count(*) подтверждённых за окно строк — тот же срез, что и у UPDATE. @@ -209,6 +311,7 @@ def deactivate_stale_listings( staleness_column: str = "last_seen_at", min_confirmations: int = 0, health_window_days: int = _HEALTH_WINDOW_DAYS, + revisit_floor_quantile: float = 0.0, ) -> dict[str, int]: """Пометить is_active=false объявления, чья свежесть старше ttl_days дней. @@ -229,6 +332,11 @@ def deactivate_stale_listings( вызывают старые тесты и совместимая обёртка); реальные значения приходят из default_params расписания, см. миграцию 219 и комментарий выше. health_window_days: окно подтверждений для гейта, суток. Дефолт 3. + revisit_floor_quantile: пол TTL по измеренному циклу переобхода (#2659). + Квантиль возраста, при котором свип за окно ДОКАЗАЛ строку живой; + эффективный TTL = max(ttl_days, этот пол). 0 -> пол выключен (так + вызывают старые тесты и совместимая обёртка), рабочее значение — + DEFAULT_REVISIT_FLOOR_QUANTILE, см. комментарий выше. Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). Один statement в транзакции: UPDATE флага + снимок 'stale' в listings_snapshots @@ -236,7 +344,8 @@ def deactivate_stale_listings( Returns {"deactivated": N} -- количество обновлённых строк (1:1 со снимками). Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N, - "skipped_unhealthy": 1} и НИ ОДНА строка не тронута. + "skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял + TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}. Raises: ValueError: если staleness_column не входит в whitelist (проверка ДО SQL, @@ -292,6 +401,44 @@ def deactivate_stale_listings( ) return counters + # Пол TTL по измеренному циклу переобхода (#2659) — тоже ДО UPDATE и по тому же + # срезу. Поднимает порог, никогда не опускает: max(), а не замена. + effective_ttl_days = ttl_days + if revisit_floor_quantile > 0: + floor_params: dict[str, Any] = { + "listing_source": listing_source, + "health_window_days": health_window_days, + "revisit_quantile": revisit_floor_quantile, + } + if segments is not None: + floor_params["segments"] = segments + floor_days = db.execute( + _build_revisit_floor_sql(staleness_column, with_segments=segments is not None), + floor_params, + ).scalar() + # NULL = истории снимков за окно нет вовсе (свежая БД, дыра в снимках). + # Тогда пола нет и TTL остаётся как задан: выдумывать пол не из чего. + if floor_days is not None: + counters["revisit_floor_days"] = ceil(float(floor_days)) + effective_ttl_days = max(ttl_days, counters["revisit_floor_days"]) + counters["ttl_days_effective"] = effective_ttl_days + if effective_ttl_days > ttl_days: + logger.warning( + "deactivate_stale source=%s run_id=%d TTL поднят с %d до %d сут: " + "свип за %d сут доказал живой строку, молчавшую %d сут " + "(квантиль %.3f, segments=%r) — при ttl_days=%d снятие означало бы " + "«мы не дошли», а не «объявление снято»", + listing_source, + run_id, + ttl_days, + effective_ttl_days, + health_window_days, + counters["revisit_floor_days"], + revisit_floor_quantile, + segments, + ttl_days, + ) + # segments is None -> все сегменты (поведение avito). segments=[...] -> только # перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список [] # означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0), @@ -299,7 +446,7 @@ def deactivate_stale_listings( if segments is not None: params: dict[str, Any] = { "listing_source": listing_source, - "ttl_days": ttl_days, + "ttl_days": effective_ttl_days, "segments": segments, "run_id": run_id, } @@ -307,7 +454,7 @@ def deactivate_stale_listings( else: params = { "listing_source": listing_source, - "ttl_days": ttl_days, + "ttl_days": effective_ttl_days, "run_id": run_id, } result = db.execute(_build_all_segments_sql(staleness_column), params) @@ -318,10 +465,11 @@ 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, staleness_column=%s)", + "(ttl_days=%d эффективный, задан %d, segments=%r, staleness_column=%s)", listing_source, run_id, counters["deactivated"], + effective_ttl_days, ttl_days, segments, staleness_column, diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py b/tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py new file mode 100644 index 00000000..76f34665 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py @@ -0,0 +1,334 @@ +"""Пол TTL по измеренному циклу переобхода (#2659). + +Гейт здоровья (#2710) отвечает «источник собирается?». Этот пол отвечает на второй +вопрос issue — «а достаточно ли ttl_days, чтобы молчание означало снятие?». Пока свип +возвращается к строке реже, чем раз в ttl_days, TTL меряет нашу выборку, а не жизнь +объявления, и источник при этом ЗДОРОВ — гейт молчит. + +Ключевой тест — test_effective_ttl_covers_every_proven_false_kill: он проигрывает +РЕАЛЬНЫЙ прод-случай (127 строк, снятых на возрасте 29.9..30.3 суток при TTL=30 и +доказанно вернувшихся живыми) и требует, чтобы эффективный TTL накрыл каждую. На +старом коде — без пола — эффективный TTL остаётся 30, и тест падает на всех срезах. +""" + +from __future__ import annotations + +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.tasks import deactivate_stale_avito as task_mod + +# ── Прод-замер 2026-08-09 (read-only), из-за которого пол существует ─────────── +# С деплоя гейта 06.08 TTL снял 1 028 строк, 127 уже снова активны (12.4%). +# В Екатеринбурге — единственном городе с настоящим покрытием — доля ложных 100%. +_FALSE_KILLS_BY_CITY: dict[str, tuple[int, int]] = { # срез -> (снято, снова активны) + "cian/Екатеринбург": (103, 103), + "yandex/Екатеринбург": (24, 24), + "cian/без города": (560, 0), + "yandex/без города": (341, 0), +} +# Возраст строки на момент снятия у всех 127 доказанно ложных снятий, суток. +_FALSE_KILL_AGE_MIN = 29.9 +_FALSE_KILL_AGE_MAX = 30.3 + +# Пол, который отдаёт ТОТ ЖЕ запрос на проде (percentile_disc 0.99, окно 3 суток, +# срез = срез TTL-джобы). Ключ -> (listing_source, segments, ttl_days сейчас, пол). +_PROD_FLOORS: dict[str, tuple[str, list[str] | None, int, float]] = { + "cian/vtorichka": ("cian", ["vtorichka"], 30, 34.0), + "yandex/vtorichka": ("yandex", ["vtorichka"], 30, 74.3), + "avito/все сегменты": ("avito", None, 10, 69.7), +} + +# Разрывы переобхода по срезу TTL-джобы (listing_source_snapshots, 40 суток): +# источник -> (p90, p99, ttl_days сейчас). Домклик — контрольная группа: почти +# полное суточное покрытие, TTL лежит в 4.5 раза выше хвоста, и снятие у него +# действительно означает снятие. +_REVISIT_TAIL: dict[str, tuple[float, float, int]] = { + "domklik/vtorichka": (1.9, 3.1, 14), + "cian/vtorichka": (10.9, 26.6, 30), + "yandex/vtorichka": (5.7, 43.0, 30), + "avito/vtorichka": (29.1, 42.1, 10), +} + + +# ── Фейковая сессия ─────────────────────────────────────────────────────────── + + +class _FakeResult: + def __init__(self, rowcount: int = 0, scalar_value: Any = None) -> None: + self.rowcount = rowcount + self._scalar = scalar_value + + def scalar(self) -> Any: + return self._scalar + + +class _FakeDB: + """Session-заглушка: percentile_disc -> пол, count(*) -> подтверждения, UPDATE -> rowcount.""" + + def __init__( + self, + *, + floor_days: float | None, + confirmations: int = 10_000, + rowcount: int = 137, + ) -> None: + self._floor = floor_days + self._confirmations = confirmations + self._rowcount = rowcount + self.executed: list[tuple[str, dict[str, Any] | None]] = [] + self.committed = False + self.rolled_back = False + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + sql = str(stmt.text) + self.executed.append((sql, params)) + if "percentile_disc" in sql: + return _FakeResult(scalar_value=self._floor) + if "SELECT count(*)" in sql: + return _FakeResult(scalar_value=self._confirmations) + return _FakeResult(rowcount=self._rowcount) + + def commit(self) -> None: + self.committed = True + + def rollback(self) -> None: + self.rolled_back = True + + @property + def floor_query(self) -> tuple[str, dict[str, Any] | None]: + return next((e for e in self.executed if "percentile_disc" in e[0]), ("", None)) + + @property + def update_query(self) -> tuple[str, dict[str, Any] | None]: + return next((e for e in self.executed if "UPDATE listings" in e[0]), ("", None)) + + +def _run(db: _FakeDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[str, int]: + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + return task_mod.deactivate_stale_listings( + db, # type: ignore[arg-type] + 1, + listing_source=kwargs.pop("listing_source", "cian"), + ttl_days=kwargs.pop("ttl_days", 30), + **kwargs, + ) + + +# ── Исторический случай: 127 доказанных ложных снятий ───────────────────────── + + +def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.MonkeyPatch) -> None: + """Ни одно из 127 доказанно ложных снятий не должно повториться. + + Все они произошли на возрасте 29.9..30.3 суток. Эффективный TTL обязан быть + строго выше этого возраста на КАЖДОМ прод-срезе — иначе следующий прогон + снимет ту же строку снова. + """ + for slice_name, (source, segments, ttl_days, floor) in _PROD_FLOORS.items(): + db = _FakeDB(floor_days=floor) + out = _run( + db, + monkeypatch, + listing_source=source, + ttl_days=ttl_days, + segments=segments, + revisit_floor_quantile=task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE, + ) + effective = out["ttl_days_effective"] + assert effective > _FALSE_KILL_AGE_MAX, ( + f"{slice_name}: эффективный TTL {effective} не накрывает возраст ложного " + f"снятия {_FALSE_KILL_AGE_MAX} — 127 строк снимутся снова" + ) + _, update_params = db.update_query + assert update_params is not None + assert ( + update_params["ttl_days"] == effective + ), f"{slice_name}: UPDATE получил не поднятый TTL — пол посчитан и выброшен" + + +def test_false_kill_ages_sit_inside_the_old_ttl(monkeypatch: pytest.MonkeyPatch) -> None: + """Замер согласован сам с собой: снимали ровно на границе TTL=30, не раньше.""" + assert _FALSE_KILL_AGE_MIN < 30.0 <= _FALSE_KILL_AGE_MAX + for source, _segments, ttl_days, _floor in _PROD_FLOORS.values(): + if source in ("cian", "yandex"): + assert ttl_days == 30, f"{source}: прод-TTL разошёлся с замером" + + +def test_false_kill_rate_is_total_where_coverage_is_real() -> None: + """В городе с настоящим покрытием ложны ВСЕ снятия — это и есть корень #2659.""" + for slice_name in ("cian/Екатеринбург", "yandex/Екатеринбург"): + killed, returned = _FALSE_KILLS_BY_CITY[slice_name] + assert returned == killed, f"{slice_name}: замер разошёлся" + total_killed = sum(k for k, _ in _FALSE_KILLS_BY_CITY.values()) + total_returned = sum(r for _, r in _FALSE_KILLS_BY_CITY.values()) + assert total_killed == 1028 + assert total_returned == 127 + + +def test_control_group_domklik_needs_no_floor() -> None: + """Домклик — контроль: при почти полном суточном обходе TTL с запасом выше хвоста. + + Если бы пол был нужен всем подряд, он был бы нужен и источнику со сплошным + покрытием. Ему не нужен — значит меряем именно покрытие, а не «TTL маловат». + + Разделяет источники ЗАПАС над хвостом, а не сам факт превышения: у домклика + TTL/p99 = 4.5, у остальных трёх 0.24..1.13, то есть порог сидит вплотную к + хвосту или внутри него. Именно у самого «благополучного» из трёх (cian, 1.13) + и случились 103 доказанно ложных снятия: p99 запаса не даёт, снимает 1% живых. + """ + _p90, p99, ttl = _REVISIT_TAIL["domklik/vtorichka"] + assert ttl / p99 > 4, "домклик перестал быть контрольной группой — перемерить" + for name, (_p90, p99, ttl) in _REVISIT_TAIL.items(): + if name.startswith("domklik"): + continue + assert ttl / p99 < 1.5, f"{name}: TTL отошёл от хвоста обхода, замер устарел" + + +# ── Контракт пола ───────────────────────────────────────────────────────────── + + +def test_floor_never_lowers_configured_ttl(monkeypatch: pytest.MonkeyPatch) -> None: + """Пол поднимает порог и только. Здоровый источник с коротким циклом не даёт + поводу снять больше, чем задано оператором.""" + db = _FakeDB(floor_days=3.1) + out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) + assert out["ttl_days_effective"] == 30 + _, update_params = db.update_query + assert update_params is not None + assert update_params["ttl_days"] == 30 + + +def test_floor_is_rounded_up_not_down(monkeypatch: pytest.MonkeyPatch) -> None: + """34.0 суток разрыва -> TTL 34, а 34.2 -> 35: округление в сторону осторожности.""" + db = _FakeDB(floor_days=34.2) + out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) + assert out["revisit_floor_days"] == 35 + assert out["ttl_days_effective"] == 35 + + +def test_floor_disabled_keeps_old_behaviour(monkeypatch: pytest.MonkeyPatch) -> None: + """quantile=0 -> ни одного лишнего запроса, поведение как до правки.""" + db = _FakeDB(floor_days=74.3) + out = _run(db, monkeypatch, ttl_days=30) + assert out == {"deactivated": 137} + assert len(db.executed) == 1 + assert "percentile_disc" not in db.executed[0][0] + + +def test_missing_snapshot_history_leaves_ttl_as_configured( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """NULL (истории снимков за окно нет) -> пол не выдумывается, TTL как задан.""" + db = _FakeDB(floor_days=None) + out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) + assert "revisit_floor_days" not in out + _, update_params = db.update_query + assert update_params is not None + assert update_params["ttl_days"] == 30 + + +def test_floor_runs_before_any_write(monkeypatch: pytest.MonkeyPatch) -> None: + """Пол считается ДО UPDATE: снятое объявление возвращает только повторный сбор.""" + db = _FakeDB(floor_days=74.3) + _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) + kinds = ["floor" if "percentile_disc" in sql else "update" for sql, _ in db.executed] + assert kinds.index("floor") < kinds.index("update") + + +def test_floor_measures_same_slice_as_update(monkeypatch: pytest.MonkeyPatch) -> None: + """Срез пола совпадает со срезом UPDATE: тот же source и те же сегменты.""" + db = _FakeDB(floor_days=74.3) + _run( + db, + monkeypatch, + listing_source="yandex", + segments=["vtorichka"], + revisit_floor_quantile=0.99, + ) + floor_sql, floor_params = db.floor_query + assert "ANY(CAST(:segments AS text[]))" in floor_sql + assert floor_params is not None + assert floor_params["segments"] == ["vtorichka"] + assert floor_params["listing_source"] == "yandex" + + +def test_floor_uses_same_staleness_column_as_ttl(monkeypatch: pytest.MonkeyPatch) -> None: + """domklik считает свежесть по scraped_at (#2204) — пол обязан мерить ту же колонку, + иначе bulk-touch по last_seen_at показал бы цикл обхода там, где сбора нет.""" + db = _FakeDB(floor_days=20.0) + _run( + db, + monkeypatch, + listing_source="domklik", + ttl_days=14, + staleness_column="scraped_at", + revisit_floor_quantile=0.99, + ) + floor_sql, _ = db.floor_query + assert "l.scraped_at" in floor_sql + assert "l.last_seen_at" not in floor_sql + + +def test_floor_rejects_invalid_staleness_column(monkeypatch: pytest.MonkeyPatch) -> None: + """Whitelist колонки работает и на пути пола — интерполяции чужого имени нет.""" + db = _FakeDB(floor_days=20.0) + with pytest.raises(ValueError): + _run(db, monkeypatch, staleness_column="is_active", revisit_floor_quantile=0.99) + assert db.executed == [] + + +def test_health_gate_still_wins_over_floor(monkeypatch: pytest.MonkeyPatch) -> None: + """Нездоровый источник блокируется гейтом ДО того, как считается пол: + лишний тяжёлый запрос по мёртвому источнику не нужен.""" + db = _FakeDB(floor_days=74.3, confirmations=10) + out = _run( + db, + monkeypatch, + min_confirmations=500, + revisit_floor_quantile=0.99, + ) + assert out["skipped_unhealthy"] == 1 + assert db.floor_query[0] == "" + assert db.update_query[0] == "" + + +def test_floor_sql_is_psycopg_v3_safe() -> None: + sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=True).text) + assert "CAST(:revisit_quantile AS double precision)" in sql + assert "CAST(:health_window_days AS integer)" in sql + assert not re.search(r":\w+::", sql) + assert "UPDATE" not in sql.upper() + assert "DELETE" not in sql.upper() + + +def test_floor_only_counts_rows_the_sweep_actually_refound() -> None: + """Пол меряет выживших, а не «мы к ним не приходили»: свежесть обязана СДВИНУТЬСЯ + относительно предыдущего снимка, иначе разрыв нулевой и хвост схлопнется в ноль.""" + sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=False).text) + assert "l.last_seen_at > prev.last_seen_at" in sql + + +def test_default_quantile_is_high_enough_for_the_prod_case() -> None: + """Ниже 0.99 опускать нельзя без нового замера: именно на 0.99 прод-запрос даёт + 34.0 для cian/vtorichka, что накрывает возраст ложных снятий 30.3.""" + assert task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE >= 0.99 + assert _PROD_FLOORS["cian/vtorichka"][3] > _FALSE_KILL_AGE_MAX + + +def test_handler_wires_revisit_floor_from_schedule_params() -> None: + """Читаем исходник файлом: product_handlers тянет scraper_kit, которого в + юнит-окружении может не быть, а проверяем мы проводку, а не импорт.""" + handlers = Path(__file__).resolve().parents[1] / "app" / "services" / "product_handlers.py" + src = handlers.read_text("utf-8") + job = src.split("async def _job_deactivate_stale")[1].split("\nasync def ")[0] + flat = " ".join(job.split()) + assert 'params.get( "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE )' in flat + assert "revisit_floor_quantile=revisit_floor_quantile" in job