fix(tradein): deactivate_stale для domklik/n1 + честная freshness по scraped_at (#2204) #2221

Merged
lekss361 merged 2 commits from fix/tradein-deactivate-stale-domklik-n1 into main 2026-07-02 20:10:35 +00:00
5 changed files with 326 additions and 28 deletions

View file

@ -149,6 +149,7 @@ async def _job_deactivate_stale(
listing_source: str = params.get("listing_source", "avito")
ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days)
segments: list[str] | None = params.get("segments")
staleness_column: str = params.get("staleness_column", "last_seen_at")
loop = asyncio.get_event_loop()
await loop.run_in_executor(
@ -159,6 +160,7 @@ async def _job_deactivate_stale(
listing_source=listing_source,
ttl_days=ttl_days,
segments=segments,
staleness_column=staleness_column,
),
)

View file

@ -1038,10 +1038,14 @@ async def trigger_deactivate_stale_avito_run(
async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Универсальный триггер для всех deactivate_stale_* источников.
Читает listing_source, ttl_days, segments из schedule_row["default_params"] (jsonb).
Читает listing_source, ttl_days, segments, staleness_column из
schedule_row["default_params"] (jsonb).
Для источника 'deactivate_stale_avito' с пустым default_params использует defaults:
listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None.
listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None,
staleness_column='last_seen_at'.
Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments.
Для domklik (#2204) -- staleness_column='scraped_at' (нетрекаемый bulk-touch
двигает last_seen_at, честная свежесть = scraped_at).
Запускает deactivate_stale_listings в executor (sync DB-only task).
Возвращает run_id или None (skip -- already running).
@ -1056,6 +1060,7 @@ async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]
listing_source: str = params.get("listing_source", "avito")
ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days)
segments: list[str] | None = params.get("segments") # list or None
staleness_column: str = params.get("staleness_column", "last_seen_at")
async def _run() -> None:
run_db = SessionLocal()
@ -1071,6 +1076,7 @@ async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]
listing_source=listing_source,
ttl_days=ttl_days,
segments=segments,
staleness_column=staleness_column,
),
)
except Exception:

View file

@ -31,30 +31,57 @@ from app.services import scrape_runs as runs_mod
logger = logging.getLogger(__name__)
# Базовый UPDATE без фильтра по сегменту: все сегменты для данного source.
# CAST(:ttl_days || ' days' AS interval) -- psycopg v3 safe (никаких :param::type).
_DEACTIVATE_SQL_ALL_SEGMENTS = text(
"""
UPDATE listings
SET is_active = false
WHERE source = :listing_source
AND is_active = true
AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
"""
)
# Разрешённые колонки-таймстемпы для проверки свежести (staleness).
# Whitelist ЖЁСТКО ограничивает f-string-интерполяцию имени колонки в SQL:
# значение колонки НИКОГДА не берётся из пользовательского ввода без этой проверки.
# - last_seen_at: дефолт; двигается при каждом «увиденном» объявлении в прогоне.
# - scraped_at: честная свежесть для источников с нетрекаемым bulk-touch
# last_seen_at (domklik: #2204 — bulk UPDATE двигает last_seen_at всем строкам
# одним timestamp, поэтому TTL по last_seen_at бесполезен; scraped_at двигает
# только реальный скрейп).
_ALLOWED_STALENESS_COLUMNS = frozenset({"last_seen_at", "scraped_at"})
# UPDATE с фильтром по сегменту (segments задан): затрагивает только указанные сегменты.
# = ANY(CAST(:segments AS text[])) -- psycopg v3 адаптирует Python list -> text[].
_DEACTIVATE_SQL_SEGMENTS = text(
def _build_all_segments_sql(staleness_column: str) -> Any:
"""UPDATE без фильтра по сегменту: все сегменты для данного source.
staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings,
поэтому f-string-подстановка имени колонки безопасна. Значения (:listing_source,
:ttl_days) остаются param-binding psycopg v3 safe (никаких :param::type).
"""
UPDATE listings
SET is_active = false
WHERE source = :listing_source
AND is_active = true
AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
AND listing_segment = ANY(CAST(:segments AS text[]))
return text(
f"""
UPDATE listings
SET is_active = false
WHERE source = :listing_source
AND is_active = true
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
"""
)
def _build_segments_sql(staleness_column: str) -> Any:
"""UPDATE с фильтром по сегменту (segments задан): только указанные сегменты.
staleness_column уже прошёл whitelist-проверку. = ANY(CAST(:segments AS text[]))
-- psycopg v3 адаптирует Python list -> text[].
"""
)
return text(
f"""
UPDATE listings
SET is_active = false
WHERE source = :listing_source
AND is_active = true
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
AND listing_segment = ANY(CAST(:segments AS text[]))
"""
)
# Дефолтные (last_seen_at) варианты SQL -- сохранены как модульные константы для
# обратной совместимости (тесты читают .text, product_handlers/scheduler не менялись).
_DEACTIVATE_SQL_ALL_SEGMENTS = _build_all_segments_sql("last_seen_at")
_DEACTIVATE_SQL_SEGMENTS = _build_segments_sql("last_seen_at")
# Алиас для обратной совместимости -- использовался в тестах через task_mod._DEACTIVATE_SQL
_DEACTIVATE_SQL = _DEACTIVATE_SQL_ALL_SEGMENTS
@ -67,22 +94,43 @@ def deactivate_stale_listings(
listing_source: str,
ttl_days: int,
segments: list[str] | None = None,
staleness_column: str = "last_seen_at",
) -> dict[str, int]:
"""Пометить is_active=false объявления с last_seen_at > ttl_days дней назад.
"""Пометить is_active=false объявления, чья свежесть старше ttl_days дней.
Параметры:
listing_source: значение поля source в таблице listings ('avito', 'yandex', 'cian').
listing_source: значение поля source в таблице listings ('avito', 'yandex',
'cian', 'domklik', 'n1').
ttl_days: количество дней TTL; объявления старше этого порога деактивируются.
segments: если задан -- деактивировать только объявления с указанными
listing_segment значениями. None -> все сегменты (поведение avito по умолчанию).
staleness_column: колонка-таймстемп, по которой считается свежесть. Whitelist
{"last_seen_at", "scraped_at"} иначе ValueError ДО любого SQL. Дефолт
last_seen_at. Для domklik (#2204) — scraped_at: нетрекаемый bulk-touch
двигает last_seen_at всем строкам одним timestamp, поэтому честная
свежесть = scraped_at (двигается только реальным скрейпом).
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
Returns {"deactivated": N} -- количество обновлённых строк.
Raises:
ValueError: если staleness_column не входит в whitelist (проверка ДО SQL,
никакой интерполяции пользовательского ввода в запрос).
"""
counters: dict[str, int] = {"deactivated": 0}
try:
# Whitelist-проверка ДО построения/выполнения SQL: только после неё имя колонки
# интерполируется f-string'ом. Значения по-прежнему идут через param-binding.
# Внутри try -> невалидная колонка финализирует run как failed (mark_failed),
# а не оставляет его «running» (единый контракт с прочими сбоями).
if staleness_column not in _ALLOWED_STALENESS_COLUMNS:
raise ValueError(
f"invalid staleness_column={staleness_column!r}; "
f"allowed: {sorted(_ALLOWED_STALENESS_COLUMNS)}"
)
# segments is None -> все сегменты (поведение avito). segments=[...] -> только
# перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список []
# означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0),
@ -93,10 +141,10 @@ def deactivate_stale_listings(
"ttl_days": ttl_days,
"segments": segments,
}
result = db.execute(_DEACTIVATE_SQL_SEGMENTS, params)
result = db.execute(_build_segments_sql(staleness_column), params)
else:
params = {"listing_source": listing_source, "ttl_days": ttl_days}
result = db.execute(_DEACTIVATE_SQL_ALL_SEGMENTS, params)
result = db.execute(_build_all_segments_sql(staleness_column), params)
counters["deactivated"] = result.rowcount or 0
@ -104,12 +152,13 @@ def deactivate_stale_listings(
runs_mod.mark_done(db, run_id, counters)
logger.info(
"deactivate_stale source=%s run_id=%d done: deactivated=%d "
"(ttl_days=%d, segments=%r)",
"(ttl_days=%d, segments=%r, staleness_column=%s)",
listing_source,
run_id,
counters["deactivated"],
ttl_days,
segments,
staleness_column,
)
return counters
except Exception as exc:

View file

@ -0,0 +1,93 @@
-- 160_seed_deactivate_stale_domklik_n1.sql
-- Issue #2204 — seed scrape_schedules rows for nightly stale-listing deactivation
-- of domklik and n1 (wildcard deactivate_stale_* family, #759 generalisation).
--
-- Context (verified on prod 2026-07-02):
-- n1: 381 is_active rows; last_seen_at frozen at 2026-06-16; no scraper runs
-- anymore -> source is DEAD. Nothing keeps these rows fresh, yet they
-- inflate the active-listing denominator used by coverage/estimator.
-- domklik: 6296 is_active rows; scraped_at frozen at 2026-06-28, BUT last_seen_at
-- is advanced every day by an UNtracked bulk UPDATE (all rows share one
-- timestamp) -> a TTL over last_seen_at can NEVER fire for domklik.
-- Honest freshness for domklik = scraped_at, which only moves when a real
-- scrape actually re-sees the listing.
--
-- Fix:
-- 1. app/tasks/deactivate_stale_avito.py — deactivate_stale_listings() gains a
-- staleness_column kwarg (whitelist {last_seen_at, scraped_at}); the WHERE
-- clause compares the chosen column against NOW() - ttl_days.
-- 2. app/services/product_handlers.py + app/services/scheduler.py — both
-- deactivate_stale_* dispatch paths read staleness_column from default_params
-- (default 'last_seen_at').
-- 3. This migration — seeds the two scrape_schedules rows below.
--
-- Why staleness_column='scraped_at' for domklik (NOT last_seen_at):
-- The daily bulk-touch advances last_seen_at for every domklik row to the same
-- instant, so `last_seen_at < NOW() - INTERVAL '14 days'` matches 0 rows forever —
-- the TTL is useless. scraped_at moves ONLY on a genuine scrape, so it reflects
-- true visibility. TTL=14 mirrors the estimator's LISTINGS_FRESH_DAYS: a listing
-- invisible to valuations for >14 days should also be deactivated, keeping the
-- active set synchronous with what the estimator actually consumes.
--
-- Why domklik is blanket (no "segments" key), unlike yandex/cian (#115 vtorichka-only):
-- verified on prod 2026-07-02 — ALL 6296 domklik rows are listing_segment='vtorichka';
-- the BFF sweep ingests secondary only, so there are no novostroyki to protect.
--
-- Why n1 uses the default last_seen_at (staleness_column omitted):
-- n1 is dead — last_seen_at is frozen at 2026-06-16, so the default column already
-- crosses the TTL threshold. The very first run will deactivate the stale inventory.
-- TTL=14 days keeps it conservative and consistent with domklik.
--
-- Schedule window 07:00-08:00 UTC:
-- After the avito deactivate_stale window (06:00-07:00 UTC) and after the main city
-- sweeps (02:00-05:00 UTC), same slot as the yandex/cian deactivate rows (#115).
-- Running after the sweeps ensures listings genuinely re-seen are NOT deactivated.
--
-- next_run_at bootstrapped to tomorrow 07:00 UTC so the scheduler does not fire
-- immediately on deploy (same pattern as 090/115).
--
-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
--
-- Dependencies:
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
-- listings table with last_seen_at + scraped_at columns.
-- app/tasks/deactivate_stale_avito.py — deactivate_stale_listings() (staleness_column).
-- app/services/scheduler.py + product_handlers.py — deactivate_stale_* dispatch that
-- forwards staleness_column from default_params.
--
-- Deploy order:
-- Apply after deploying the scheduler.py + product_handlers.py + task changes so the
-- dispatch can forward staleness_column correctly on the first fire.
BEGIN;
INSERT INTO scrape_schedules (
source,
enabled,
window_start_hour,
window_end_hour,
next_run_at,
default_params
)
VALUES
(
'deactivate_stale_domklik',
true, -- SAFE: pure internal DB UPDATE, no ext calls
7,
8,
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
-- staleness_column=scraped_at: last_seen_at bulk-touched daily -> useless TTL.
'{"listing_source":"domklik","ttl_days":14,"staleness_column":"scraped_at"}'::jsonb
),
(
'deactivate_stale_n1',
true, -- SAFE: pure internal DB UPDATE, no ext calls
7,
8,
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
-- n1 dead (last_seen_at frozen 2026-06-16); default last_seen_at column fires first run.
'{"listing_source":"n1","ttl_days":14}'::jsonb
)
ON CONFLICT (source) DO NOTHING;
COMMIT;

View file

@ -303,3 +303,151 @@ def test_migration_115_ttl_30_days() -> None:
def test_migration_115_no_psycopg_trap() -> None:
sql = _MIGRATION_115.read_text("utf-8")
assert not re.search(r":\w+::", sql)
# ── staleness_column (#2204: domklik scraped_at vs last_seen_at) ───────────────
def _sql_text(stmt: Any) -> str:
"""Достать SQL-текст из TextClause, захваченного _FakeDB.execute."""
return str(getattr(stmt, "text", stmt))
def test_default_staleness_column_is_last_seen_at(monkeypatch: pytest.MonkeyPatch) -> None:
"""Без явного staleness_column WHERE фильтрует по last_seen_at (обратная совместимость)."""
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
db = _FakeDB(rowcount=1)
task_mod.deactivate_stale_listings(
db, run_id=10, listing_source="avito", ttl_days=10, segments=None
) # type: ignore[arg-type]
stmt, _params = db.executed[0]
sql = _sql_text(stmt)
assert "last_seen_at <" in sql
assert "scraped_at" not in sql
def test_scraped_at_column_used_in_all_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None:
"""domklik bulk-touch: staleness_column='scraped_at' -> WHERE по scraped_at, не last_seen_at.
Кейс: last_seen_at свежий (ежедневный bulk UPDATE), но scraped_at заморожен ->
деактивация должна сработать по scraped_at. Проверяем, что именно эта колонка в SQL.
"""
marked: dict[str, Any] = {}
monkeypatch.setattr(
task_mod.runs_mod,
"mark_done",
lambda _db, run_id, counters: marked.update(counters=dict(counters)),
)
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
db = _FakeDB(rowcount=6296)
out = task_mod.deactivate_stale_listings(
db,
run_id=11,
listing_source="domklik",
ttl_days=14,
segments=None,
staleness_column="scraped_at",
) # type: ignore[arg-type]
stmt, params = db.executed[0]
sql = _sql_text(stmt)
assert "scraped_at <" in sql
assert "last_seen_at" not in sql
assert params is not None
assert params["listing_source"] == "domklik"
assert params["ttl_days"] == 14
assert out == {"deactivated": 6296}
assert marked["counters"] == {"deactivated": 6296}
def test_scraped_at_column_used_in_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None:
"""staleness_column='scraped_at' также уважается в segment-варианте SQL."""
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
db = _FakeDB(rowcount=0)
task_mod.deactivate_stale_listings(
db,
run_id=12,
listing_source="domklik",
ttl_days=14,
segments=["vtorichka"],
staleness_column="scraped_at",
) # type: ignore[arg-type]
stmt, _params = db.executed[0]
sql = _sql_text(stmt)
assert "scraped_at <" in sql
assert "listing_segment = ANY(CAST(:segments AS text[]))" in sql
def test_invalid_staleness_column_raises_and_marks_failed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Невалидная колонка -> ValueError ДО SQL, run помечен failed, никакого execute."""
failed: dict[str, Any] = {}
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
monkeypatch.setattr(
task_mod.runs_mod,
"mark_failed",
lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
)
db = _FakeDB(rowcount=0)
with pytest.raises(ValueError, match="invalid staleness_column"):
task_mod.deactivate_stale_listings(
db,
run_id=13,
listing_source="domklik",
ttl_days=14,
segments=None,
staleness_column="created_at; DROP TABLE listings",
) # type: ignore[arg-type]
# Никакого SQL не выполнено (проверка ДО execute).
assert db.executed == []
assert db.rolled_back is True
assert failed["run_id"] == 13
def test_allowed_staleness_columns_whitelist() -> None:
assert task_mod._ALLOWED_STALENESS_COLUMNS == frozenset({"last_seen_at", "scraped_at"})
# ── Migration 160 (deactivate_stale_domklik / _n1) ────────────────────────────
_MIGRATION_160 = _SQL_DIR / "160_seed_deactivate_stale_domklik_n1.sql"
def test_migration_160_exists() -> None:
assert _MIGRATION_160.is_file(), f"missing migration: {_MIGRATION_160}"
def test_migration_160_seeds_domklik_and_n1() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert "'deactivate_stale_domklik'" in sql
assert "'deactivate_stale_n1'" in sql
def test_migration_160_domklik_uses_scraped_at() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert '"listing_source":"domklik"' in sql
assert '"staleness_column":"scraped_at"' in sql
def test_migration_160_n1_default_column_ttl14() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert '"listing_source":"n1"' in sql
assert '"ttl_days":14' in sql
def test_migration_160_is_idempotent() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert "ON CONFLICT (source) DO NOTHING" in sql
def test_migration_160_is_transactional() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert "BEGIN;" in sql
assert "COMMIT;" in sql
def test_migration_160_no_psycopg_trap() -> None:
sql = _MIGRATION_160.read_text("utf-8")
assert not re.search(r":\w+::", sql)