bot-backend
f1f2bca2e9
All checks were successful
Deploy Trade-In / changes (push) Successful in 11s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 3m5s
Deploy Trade-In / build-backend (push) Successful in 56s
Deploy Trade-In / deploy (push) Successful in 1m45s
500 lines
33 KiB
Python
500 lines
33 KiB
Python
"""Deactivate stale listings that have not been seen in >TTL days.
|
||
|
||
Первоначально создано для avito (#759). Теперь поддерживает любой источник
|
||
(listing_source) с опциональной фильтрацией по listing_segment.
|
||
|
||
Ключевые решения:
|
||
- Cian/Yandex не поддерживают full-coverage sweep -> паушальный TTL сломает живой
|
||
инвентарь. DECISION: для yandex/cian деактивировать ТОЛЬКО listing_segment='vtorichka',
|
||
TTL=30. novostroyki (9659 активных первичных строк) и NULL-сегмент не трогаем.
|
||
- avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений.
|
||
- Строки НЕ удаляются -- история нужна для бэктеста (#667).
|
||
- #2674: деактивация в той же транзакции пишет снимок listings_snapshots со статусом
|
||
'stale' за текущую дату -- «мы N суток не видели». Жёсткое 'closed' (площадка
|
||
ответила 404) пишет только avito_detail_backfill: смешивать факт с догадкой дорого.
|
||
|
||
Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается kit-scheduler'ом
|
||
через product_handlers._job_deactivate_stale (wildcard-handler deactivate_stale_*),
|
||
по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios
|
||
(sync task в run_in_executor).
|
||
|
||
TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from math import ceil
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Разрешённые колонки-таймстемпы для проверки свежести (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"})
|
||
|
||
# ── Снимок «протухло» в дневной истории (#2674) ───────────────────────────────
|
||
# listings_snapshots.status до этого фикса был константой 'active' у всех строк
|
||
# (394 299 на момент находки) — оба места вызова upsert_listing_snapshot передавали
|
||
# литерал 'active', и это честно: там объявление ДЕЙСТВИТЕЛЬНО видели. А деактивация
|
||
# TTL-задачей не оставляла в истории вообще никакого следа. Из-за этого дата снятия
|
||
# объявления (лучший доступный сигнал «скорее всего продано») не запрашивалась из
|
||
# истории, а восстанавливалась на глаз: последний показ + предполагаемый срок жизни.
|
||
#
|
||
# ПОЧЕМУ 'stale', А НЕ 'closed'. Эта задача НЕ знает, что объявление снято, — она
|
||
# знает только, что МЫ его N суток не видели, а это разные факты, когда TTL короче
|
||
# простоя обхода. Замер: прогон по домклику 02.08 снял 6131 объявление за раз (TTL
|
||
# 14 суток против 12 суток простоя обхода) — с общим статусом это были бы 6131
|
||
# фальшивая «дата продажи» одной датой. Продукт про цены, смешивать факт с догадкой
|
||
# дорого. Поэтому:
|
||
# 'closed' — только путь 404: площадка ответила «нет» (avito_detail_backfill);
|
||
# 'stale' — этот путь: «мы N суток не смотрели».
|
||
# Дата всё равно фиксируется, но читатель отличает одно от другого. Ограничения
|
||
# CHECK на колонке нет (проверено на проде), миграция не нужна — только COMMENT.
|
||
#
|
||
# Пишем снимок в ТОЙ ЖЕ транзакции, что и UPDATE флага: деактивация без снимка (или
|
||
# наоборот) невозможна по построению — один statement, data-modifying CTE.
|
||
# 1:1 по строкам: `stale` возвращает уникальные listings.id (PK), каждая даёт ровно
|
||
# одну затронутую строку listings_snapshots (INSERT либо DO UPDATE — оба считаются
|
||
# в rowcount), поэтому rowcount statement'а по-прежнему равен числу деактивированных.
|
||
# price_rub берём из listings (NOT NULL в схеме) — это последняя известная цена.
|
||
# ON CONFLICT: если снимок за сегодня уже есть (объявление видели активным утром,
|
||
# а вечером сработал TTL) — только переводим статус в 'stale', цену не переписываем.
|
||
_STALE_SNAPSHOT_TAIL = """
|
||
INSERT INTO listings_snapshots
|
||
(listing_id, snapshot_date, run_id, price_rub, status, observed_at)
|
||
SELECT id, CURRENT_DATE, CAST(:run_id AS bigint), price_rub, 'stale', NOW()
|
||
FROM stale
|
||
ON CONFLICT (listing_id, snapshot_date) DO UPDATE SET
|
||
status = 'stale',
|
||
observed_at = EXCLUDED.observed_at
|
||
"""
|
||
|
||
|
||
# ── Гейт по здоровью сбора (#2659) ────────────────────────────────────────────
|
||
# TTL отвечает на вопрос «объявление сняли?», а меряет «мы его давно не видели».
|
||
# Пока обход здоров, разница мала. Когда обход лёг — разница равна всему инвентарю.
|
||
#
|
||
# Замер на проде, из-за которого этот гейт существует. Авито 10.07-26.07.2026:
|
||
# 17 суток подряд без единой успешно собранной страницы, TTL=10 снял за этот отрезок
|
||
# 9 033 строки; 1 270 из них потом доказанно вернулись живыми (снимки
|
||
# listing_source_snapshots + текущий last_seen_at) — сбор восстановился, и объявления
|
||
# оказались на месте. То есть «мы не смогли зайти» было прочитано как «объявление снято».
|
||
#
|
||
# ПОЧЕМУ НЕ ПО СТАТУСУ ban. Соблазн взять scrape_runs.status='banned' — ловушка:
|
||
# Яндекс 18.07-30.07 — 5 прогонов в сутки, ВСЕ 'done', НОЛЬ 'banned', total_seen=0
|
||
# 13 суток подряд (снято ~839 строк vtorichka);
|
||
# Домклик 20.07-30.07 — то же самое, 11 суток 'done' с total_seen=0, а 02.08 TTL
|
||
# снял 6 131 строку разом (см. комментарий про 'stale' выше).
|
||
# Оба провала для ban-детектора невидимы. Поэтому здоровье меряем НЕ статусом прогона,
|
||
# а результатом: сколько строк источник реально подтвердил свежими за последние сутки.
|
||
#
|
||
# МЕТРИКА: count(*) по той же колонке свежести, что и сам TTL (last_seen_at или
|
||
# scraped_at) и по тому же срезу source+segment, что и UPDATE. Одна колонка на обе
|
||
# стороны — гейт нельзя обмануть bulk-touch'ем, который не двигает scraped_at (#2204).
|
||
#
|
||
# ПОРОГ. Ряд «подтверждений за 3 суток» по дням (восстановлен из listing_source_snapshots):
|
||
# avito здоровые сутки 3542..6079, провал 10.07-26.07 — 0..970 → порог 1500;
|
||
# yandex vtorichka здоровые 897..2206, провал — 0 → порог 500;
|
||
# cian vtorichka 748..4329, провала не было → порог 500;
|
||
# domklik по scraped_at сейчас 62/3 суток (сбор фактически стоит) → порог 200.
|
||
# Пороги живут в default_params расписания (миграция 219), здесь только страховка
|
||
# на случай незасеянного расписания. Асимметрия цены ошибки намеренная: пропущенная
|
||
# деактивация чинится следующим прогоном, ложная — только повторным сбором, которого
|
||
# может не быть. Поэтому при сомнении — пропускаем прогон.
|
||
#
|
||
# ПОТОЛОК: окно 3 суток годится, пока свипы источника ходят не реже чем раз в 3 дня.
|
||
# Источник с более редкой каденцией будет блокироваться всегда — тогда окно нужно
|
||
# растить до каденции, а не понижать порог.
|
||
_HEALTH_WINDOW_DAYS = 3
|
||
|
||
# Страховка для расписаний без явного min_confirmations в default_params: ловит
|
||
# полный ноль и близкое к нулю, но НЕ ловит частичный провал вроде avito 936-970 —
|
||
# для этого нужен посчитанный по источнику порог из миграции 219.
|
||
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.
|
||
|
||
staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings.
|
||
Значения (:listing_source, :health_window_days, :segments) — param-binding,
|
||
psycopg v3 safe (CAST(... AS ...), никаких :param::type).
|
||
"""
|
||
segment_filter = _CONFIRMATIONS_SEGMENT_FILTER if with_segments else ""
|
||
return text(
|
||
f"""
|
||
SELECT count(*)
|
||
FROM listings
|
||
WHERE source = :listing_source
|
||
AND {staleness_column}
|
||
> NOW() - CAST(:health_window_days || ' days' AS interval){segment_filter}
|
||
"""
|
||
)
|
||
|
||
|
||
def _build_all_segments_sql(staleness_column: str) -> Any:
|
||
"""UPDATE без фильтра по сегменту: все сегменты для данного source.
|
||
|
||
staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings,
|
||
поэтому f-string-подстановка имени колонки безопасна. Значения (:listing_source,
|
||
:ttl_days, :run_id) остаются param-binding — psycopg v3 safe (никаких :param::type).
|
||
"""
|
||
return text(
|
||
f"""
|
||
WITH stale AS (
|
||
UPDATE listings
|
||
SET is_active = false
|
||
WHERE source = :listing_source
|
||
AND is_active = true
|
||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||
RETURNING id, price_rub
|
||
)
|
||
{_STALE_SNAPSHOT_TAIL}
|
||
"""
|
||
)
|
||
|
||
|
||
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"""
|
||
WITH stale AS (
|
||
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[]))
|
||
RETURNING id, price_rub
|
||
)
|
||
{_STALE_SNAPSHOT_TAIL}
|
||
"""
|
||
)
|
||
|
||
|
||
# Дефолтные (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
|
||
|
||
|
||
def deactivate_stale_listings(
|
||
db: Session,
|
||
run_id: int,
|
||
*,
|
||
listing_source: str,
|
||
ttl_days: int,
|
||
segments: list[str] | None = None,
|
||
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 дней.
|
||
|
||
Параметры:
|
||
listing_source: значение поля source в таблице listings ('avito', 'yandex',
|
||
'cian', 'domklik').
|
||
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 (двигается только реальным скрейпом).
|
||
min_confirmations: гейт по здоровью сбора (#2659). Сколько строк источник
|
||
должен был подтвердить свежими за health_window_days суток, чтобы
|
||
деактивации вообще разрешалось исполниться. 0 -> гейт выключен (так
|
||
вызывают старые тесты и совместимая обёртка); реальные значения приходят
|
||
из 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
|
||
(data-modifying CTE, #2674). Финализирует scrape_runs (mark_done / mark_failed).
|
||
|
||
Returns {"deactivated": N} -- количество обновлённых строк (1:1 со снимками).
|
||
Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N,
|
||
"skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял
|
||
TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": 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)}"
|
||
)
|
||
|
||
# Гейт по здоровью сбора (#2659) — ДО любого UPDATE. Деактивация необратима
|
||
# на практике (вернуть «живость» может только повторный сбор), поэтому
|
||
# проверяем ПЕРЕД записью, а не откатываем после.
|
||
if min_confirmations > 0:
|
||
health_params: dict[str, Any] = {
|
||
"listing_source": listing_source,
|
||
"health_window_days": health_window_days,
|
||
}
|
||
if segments is not None:
|
||
health_params["segments"] = segments
|
||
confirmations = (
|
||
db.execute(
|
||
_build_confirmations_sql(staleness_column, with_segments=segments is not None),
|
||
health_params,
|
||
).scalar()
|
||
or 0
|
||
)
|
||
counters["confirmations"] = int(confirmations)
|
||
if confirmations < min_confirmations:
|
||
counters["skipped_unhealthy"] = 1
|
||
# Ничего не писали (был только SELECT) — rollback закрывает транзакцию
|
||
# чисто, чтобы mark_done стартовал со своей.
|
||
db.rollback()
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.warning(
|
||
"deactivate_stale source=%s run_id=%d SKIPPED: сбор нездоров — "
|
||
"подтверждений за %d сут %d < порога %d "
|
||
"(segments=%r, staleness_column=%s); ни одна строка не тронута",
|
||
listing_source,
|
||
run_id,
|
||
health_window_days,
|
||
confirmations,
|
||
min_confirmations,
|
||
segments,
|
||
staleness_column,
|
||
)
|
||
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),
|
||
# а НЕ "все сегменты" — иначе случайный [] стёр бы весь источник.
|
||
if segments is not None:
|
||
params: dict[str, Any] = {
|
||
"listing_source": listing_source,
|
||
"ttl_days": effective_ttl_days,
|
||
"segments": segments,
|
||
"run_id": run_id,
|
||
}
|
||
result = db.execute(_build_segments_sql(staleness_column), params)
|
||
else:
|
||
params = {
|
||
"listing_source": listing_source,
|
||
"ttl_days": effective_ttl_days,
|
||
"run_id": run_id,
|
||
}
|
||
result = db.execute(_build_all_segments_sql(staleness_column), params)
|
||
|
||
counters["deactivated"] = result.rowcount or 0
|
||
|
||
db.commit()
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.info(
|
||
"deactivate_stale source=%s run_id=%d done: deactivated=%d "
|
||
"(ttl_days=%d эффективный, задан %d, segments=%r, staleness_column=%s)",
|
||
listing_source,
|
||
run_id,
|
||
counters["deactivated"],
|
||
effective_ttl_days,
|
||
ttl_days,
|
||
segments,
|
||
staleness_column,
|
||
)
|
||
return counters
|
||
except Exception as exc:
|
||
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,
|
||
)
|