fix(tradein/deactivate): TTL не снимает объявления по порогу ниже собственного цикла обхода #2797
3 changed files with 494 additions and 4 deletions
|
|
@ -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,
|
||||
),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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.<col> и
|
||||
# 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,
|
||||
|
|
|
|||
334
tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py
Normal file
334
tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue