From e4d44025e5d7d50fd5dcb967484b2a1fed1918b3 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 17 Sep 2026 14:46:36 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein/snapshot):=20=D1=81=D0=BD=D0=B8?= =?UTF-8?q?=D0=BC=D0=BE=D0=BA=20=D0=B8=D1=81=D1=82=D0=BE=D1=87=D0=BD=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BF=D0=B8=D1=88=D0=B5=D1=82=D1=81=D1=8F=20?= =?UTF-8?q?=D1=82=D0=BE=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BF=D1=80=D0=B8=20?= =?UTF-8?q?=D0=B8=D0=B7=D0=BC=D0=B5=D0=BD=D0=B5=D0=BD=D0=B8=D0=B8=20(#2993?= =?UTF-8?q?,=20PR-C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit listing_source_snapshot каждую ночь копировал listing_sources целиком: прод 14-17.09 — 290-293 тыс. строк в сутки, из них отличались от предыдущего снимка 7.7-11.4 тыс. (3-4 %). Решение владельца 2026-08-23 — строка на изменение; читатель (пол переобхода, PR-A #3056) и рельсы объёма (PR-B #3066) уже смержены. Строка пишется, если снимка ещё нет или (price_rub, is_active, last_seen_at, payload_hash) отличается от последнего снимка источника, включая сегодняшний (повторный прогон в те же сутки перезаписывает строку, только если значение снова сдвинулось). Состояние на дату D = последняя строка с snapshot_date <= D — для всех четырёх колонок то же, что дала бы суточная копия, поэтому пол переобхода и события не меняются. Старые суточные снимки не трогаются. Замер чтением на проде 17.09: отбор с условием — 1.0 с на 293 602 источника (бюджет 900 с). Миграция 325 — только COMMENT на таблицу и view: прежние «ежедневный снимок»/«на дату снимка» стали бы неправдой. Co-Authored-By: Claude Opus 5 --- .../app/tasks/listing_source_snapshot.py | 64 ++++- ...g_source_snapshots_change_only_comment.sql | 29 +++ .../tests/test_2993_snapshot_change_only.py | 246 ++++++++++++++++++ 3 files changed, 328 insertions(+), 11 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/325_listing_source_snapshots_change_only_comment.sql create mode 100644 tradein-mvp/backend/tests/test_2993_snapshot_change_only.py diff --git a/tradein-mvp/backend/app/tasks/listing_source_snapshot.py b/tradein-mvp/backend/app/tasks/listing_source_snapshot.py index 81dcbf16..b745a5c2 100644 --- a/tradein-mvp/backend/app/tasks/listing_source_snapshot.py +++ b/tradein-mvp/backend/app/tasks/listing_source_snapshot.py @@ -1,7 +1,8 @@ """Daily per-source snapshot writer (#570). Берёт текущее состояние listing_sources (последний снимок на canonical listing × source) -и пишет ежедневный ряд в listing_source_snapshots + change-log price_change в +и раз в сутки пишет в listing_source_snapshots строку на КАЖДОЕ ИЗМЕНЕНИЕ (#2993, см. +_SNAPSHOT_SQL; раньше — полная копия каждые сутки) + change-log в listing_source_events. Так история per-source цены копится СРАЗУ, независимо от (сейчас DORMANT) скраперов — см. шапку data/sql/079_listing_source_history.sql. @@ -72,9 +73,28 @@ def _clamp_budget_sec(raw: Any) -> float: return max(_MIN_BUDGET_SEC, min(val, _MAX_BUDGET_SEC)) -# ── Daily snapshot upsert ───────────────────────────────────────────────────── -# Снимок на (listing_source_id, CURRENT_DATE). ON CONFLICT → last-write-wins за день -# (повторный прогон в те же сутки перезаписывает снимок свежими значениями). +# ── Snapshot upsert: строка только на изменение (#2993) ─────────────────────── +# Снимок на (listing_source_id, CURRENT_DATE) пишется, только если у источника ещё нет +# ни одного снимка ИЛИ хоть одно из четырёх значений (price_rub, is_active, last_seen_at, +# payload_hash) отличается от его ПОСЛЕДНЕГО снимка. Было: полная копия listing_sources +# каждые сутки, прод 14-17.09 — 290-293 тыс. строк в сутки, из них отличались от +# предыдущего снимка 7.7-11.4 тыс. (3-4 %). Решение владельца 2026-08-23. +# +# Состояние источника на дату D = его последний снимок с snapshot_date <= D. Для всех +# четырёх колонок это то же значение, что записала бы суточная модель: сутки, в которые +# ни одна из них не менялась, и есть пропущенные строки. Поэтому last_seen_at обязан +# быть в сравнении: пол переобхода (deactivate_stale_avito._revisit_floor_from_where_sql, +# PR-A #3056) берёт last_seen_at последнего снимка не позже якоря — без него пары +# «прошлое наблюдение → текущее» считались бы от устаревшей свежести. is_active там же: +# он выводится из now(), и переход «свежий → протух» при неизменном last_seen_at — тоже +# изменение состояния на дату. +# +# p — последний снимок ВКЛЮЧАЯ сегодняшний: повторный прогон в те же сутки сравнивает +# с тем, что уже записано сегодня, и перезаписывает строку только если значение снова +# сдвинулось (ON CONFLICT → last-write-wins, как раньше). Пер-строчный LATERAL +# point-lookup по PK, как в event-diff (#2607); прод 17.09: 293 602 lookup'а — 1.0 с. +# Старые суточные снимки не трогаются — история до перехода остаётся как есть. +# # is_active derived: last_seen_at в пределах окна свежести на момент снимка. # payload_hash = md5(raw_payload::text) — ::text на колонке допустим (это не bind-param). # run_id через CAST(:run_id AS bigint) — psycopg v3 (никогда :run_id::bigint). @@ -85,15 +105,33 @@ _SNAPSHOT_SQL = text( last_seen_at, payload_hash, observed_at, run_id ) SELECT - id, + cur.id, CURRENT_DATE, - price_rub, - (last_seen_at > now() - make_interval(days => :freshness_days)) AS is_active, - last_seen_at, - md5(raw_payload::text), + cur.price_rub, + cur.is_active, + cur.last_seen_at, + cur.payload_hash, now(), CAST(:run_id AS bigint) - FROM listing_sources + FROM ( + SELECT + id, + price_rub, + (last_seen_at > now() - make_interval(days => :freshness_days)) AS is_active, + last_seen_at, + md5(raw_payload::text) AS payload_hash + FROM listing_sources + ) cur + LEFT JOIN LATERAL ( + SELECT s.snapshot_date, s.price_rub, s.is_active, s.last_seen_at, s.payload_hash + FROM listing_source_snapshots s + WHERE s.listing_source_id = cur.id + ORDER BY s.snapshot_date DESC + LIMIT 1 + ) p ON true + WHERE p.snapshot_date IS NULL + OR (cur.price_rub, cur.is_active, cur.last_seen_at, cur.payload_hash) + IS DISTINCT FROM (p.price_rub, p.is_active, p.last_seen_at, p.payload_hash) ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE SET price_rub = EXCLUDED.price_rub, is_active = EXCLUDED.is_active, @@ -108,6 +146,9 @@ _SNAPSHOT_SQL = text( # Для каждого источника сравниваем сегодняшний снимок (snapshot_date = CURRENT_DATE) с # самым свежим ПРЕДЫДУЩИМ (snapshot_date < CURRENT_DATE). # today — снимок за сегодня (только что записан _SNAPSHOT_SQL, в той же транзакции). +# С #2993 здесь только изменившиеся и новые источники: у остальных все четыре +# значения равны последнему снимку, а значит ни одно событие ниже сработать +# не могло бы — набор событий тот же, что при суточной копии. # p — последний снимок строго ДО сегодня, per-row LATERAL point-lookup (#2607). # # #2674: схема (079) знает пять типов событий, писатель умел один — price_change, @@ -249,7 +290,8 @@ def snapshot_listing_sources( Sync (вызывается scheduler-триггером в executor, как import_rosreestr_dkp). Два set-based statement'а в одной транзакции: - 1. upsert снимка на (listing_source_id, CURRENT_DATE) — last-write-wins. + 1. upsert снимка на (listing_source_id, CURRENT_DATE) только для источников, + изменившихся с последнего снимка (#2993) — last-write-wins. 2. diff сегодняшнего снимка против последнего предыдущего → три события, выводимые из наших данных (#2674). delisted/relisted схема разрешает, но они НЕ выводимы при покрытии обхода 10-35% — см. _EVENT_DIFF_SQL. diff --git a/tradein-mvp/backend/data/sql/325_listing_source_snapshots_change_only_comment.sql b/tradein-mvp/backend/data/sql/325_listing_source_snapshots_change_only_comment.sql new file mode 100644 index 00000000..ef1eac6b --- /dev/null +++ b/tradein-mvp/backend/data/sql/325_listing_source_snapshots_change_only_comment.sql @@ -0,0 +1,29 @@ +-- 325_listing_source_snapshots_change_only_comment.sql +-- #2993 — listing_source_snapshots переходит на «строку на изменение» +-- (app/tasks/listing_source_snapshot.py, _SNAPSHOT_SQL; решение владельца 2026-08-23). +-- +-- Схема не меняется, меняется смысл строк — поэтому только комментарии. Прежние +-- («Ежедневный снимок…», «на дату снимка») после перехода стали бы неправдой: +-- `WHERE snapshot_date = D` отдаёт теперь не состояние всех источников на D, а только +-- изменившиеся в D. Состояние на дату — последний снимок с snapshot_date <= D. +-- Суточные строки до перехода остаются в таблице как есть. +-- +-- COMMENT берёт ShareUpdateExclusiveLock (конфликтует с autovacuum/ANALYZE, не с +-- чтением и записью) — lock_timeout, чтобы деплой не повис за VACUUM 8-млн таблицы. + +BEGIN; + +SET LOCAL lock_timeout = '5s'; + +COMMENT ON TABLE listing_source_snapshots IS + 'Снимок listing_sources на источник (#570), строка только при изменении (#2993): ' + 'пишется, если price_rub/is_active/last_seen_at/payload_hash отличаются от последнего ' + 'снимка источника или снимка ещё нет. Состояние на дату D — последняя строка с ' + 'snapshot_date <= D. До перехода (#2993) — полная суточная копия. ' + 'PK (listing_source_id, snapshot_date), last-write-wins за день.'; + +COMMENT ON VIEW v_listing_source_price_on_date IS + 'Цена/активность источника с даты снимка (#570): listing_source_snapshots ⋈ listing_sources. ' + 'С #2993 строки только на даты изменений — цена на дату D = последняя строка с snapshot_date <= D.'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_2993_snapshot_change_only.py b/tradein-mvp/backend/tests/test_2993_snapshot_change_only.py new file mode 100644 index 00000000..947e505a --- /dev/null +++ b/tradein-mvp/backend/tests/test_2993_snapshot_change_only.py @@ -0,0 +1,246 @@ +"""#2993: listing_source_snapshots получает строку только при изменении источника. + +Проверки идут по ЗНАЧЕНИЯМ в живом Postgres: какие источники получили строку за +сегодня, с какими значениями, и какие события записал event-diff поверх них. +Без реальной БД файл self-skip'ается, как соседние live-тесты (в CI есть +postgres, см. ci-tradein.yml). Вся синтетика живёт в одной транзакции, очистка — +rollback. +""" + +from __future__ import annotations + +import os +import uuid +from typing import Any + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import pytest +from sqlalchemy import text + +from app.tasks import listing_source_snapshot as snap_mod + + +def _live_session() -> Any | None: + """Тот же контракт, что у test_revisit_floor_lateral_lookup.py.""" + try: + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "") + if not dsn or "localhost:5432/test" in dsn: + return None + engine = create_engine(dsn, future=True) + conn = engine.connect() + conn.execute(text("SELECT 1")) + conn.close() + return sessionmaker(bind=engine, future=True)() + except Exception: + return None + + +pytestmark = pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") + +_PRICE = 5_000_000 + + +def _source(db: Any, tag: str, *, price: int, payload: str, seen_days_ago: int) -> int: + ext_id = f"zzz2993-{tag}-{uuid.uuid4().hex[:8]}" + listing_id = db.execute( + text( + """ + INSERT INTO listings + (source, source_url, source_id, dedup_hash, price_rub, + is_active, scraped_at, last_seen_at) + VALUES ('cian', :url, :ext_id, :ext_id, :price, true, now(), now()) + RETURNING id + """ + ), + {"url": f"https://example.test/2993/{ext_id}", "ext_id": ext_id, "price": price}, + ).scalar_one() + return int( + db.execute( + text( + """ + INSERT INTO listing_sources + (listing_id, ext_source, ext_id, confidence, matched_method, + price_rub, raw_payload, last_seen_at) + VALUES (:listing_id, 'cian', :ext_id, 1.0, 'test', + :price, CAST(:payload AS jsonb), + now() - make_interval(days => :seen)) + RETURNING id + """ + ), + { + "listing_id": listing_id, + "ext_id": ext_id, + "price": price, + "payload": payload, + "seen": seen_days_ago, + }, + ).scalar_one() + ) + + +def _snapshot( + db: Any, + lsid: int, + *, + days_ago: int, + price: int, + payload: str, + seen_days_ago: int, + is_active: bool = True, +) -> None: + # now() постоянен внутри транзакции — last_seen_at снимка совпадает с + # listing_sources.last_seen_at до микросекунды, если seen_days_ago тот же. + db.execute( + text( + """ + INSERT INTO listing_source_snapshots + (listing_source_id, snapshot_date, price_rub, is_active, + last_seen_at, payload_hash) + VALUES (:lsid, CURRENT_DATE - :days_ago, :price, :is_active, + now() - make_interval(days => :seen), + md5(CAST(:payload AS jsonb)::text)) + """ + ), + { + "lsid": lsid, + "days_ago": days_ago, + "price": price, + "is_active": is_active, + "seen": seen_days_ago, + "payload": payload, + }, + ) + + +def _run_writer(db: Any) -> None: + db.execute( + snap_mod._SNAPSHOT_SQL, + {"freshness_days": snap_mod.FRESHNESS_WINDOW_DAYS, "run_id": None}, + ) + + +def _today(db: Any, ids: list[int]) -> dict[int, tuple[int, bool]]: + rows = db.execute( + text( + """ + SELECT listing_source_id, price_rub, is_active + FROM listing_source_snapshots + WHERE snapshot_date = CURRENT_DATE AND listing_source_id = ANY(:ids) + """ + ), + {"ids": ids}, + ).fetchall() + return {int(r[0]): (int(r[1]), bool(r[2])) for r in rows} + + +def _seed(db: Any) -> dict[str, int]: + """Семь источников: по одному на каждую ветку условия записи.""" + p = '{"v": 1}' + ids = { + "same": _source(db, "same", price=_PRICE, payload=p, seen_days_ago=1), + "same_gap": _source(db, "same_gap", price=_PRICE, payload=p, seen_days_ago=1), + "price": _source(db, "price", price=_PRICE, payload=p, seen_days_ago=1), + "payload": _source(db, "payload", price=_PRICE, payload=p, seen_days_ago=1), + "seen": _source(db, "seen", price=_PRICE, payload=p, seen_days_ago=1), + "active": _source(db, "active", price=_PRICE, payload=p, seen_days_ago=10), + "new": _source(db, "new", price=_PRICE, payload=p, seen_days_ago=1), + } + _snapshot(db, ids["same"], days_ago=1, price=_PRICE, payload=p, seen_days_ago=1) + # Последний снимок пятидневной давности совпадает, а более старый — нет: + # сравнивать обязаны с ПОСЛЕДНИМ, а не с любым из прошлых. + _snapshot(db, ids["same_gap"], days_ago=9, price=_PRICE - 1, payload=p, seen_days_ago=9) + _snapshot(db, ids["same_gap"], days_ago=5, price=_PRICE, payload=p, seen_days_ago=1) + _snapshot(db, ids["price"], days_ago=1, price=4_900_000, payload=p, seen_days_ago=1) + _snapshot(db, ids["payload"], days_ago=1, price=_PRICE, payload='{"v": 0}', seen_days_ago=1) + _snapshot(db, ids["seen"], days_ago=1, price=_PRICE, payload=p, seen_days_ago=3) + # last_seen_at тот же, но 10 суток > окна свежести: вчера снимок был активным. + _snapshot(db, ids["active"], days_ago=1, price=_PRICE, payload=p, seen_days_ago=10) + return ids + + +def test_only_changed_or_new_sources_get_a_row_today() -> None: + db = _live_session() + try: + ids = _seed(db) + _run_writer(db) + today = _today(db, list(ids.values())) + by_tag = {tag: today.get(lsid) for tag, lsid in ids.items()} + + assert by_tag == { + "same": None, + "same_gap": None, + "price": (_PRICE, True), + "payload": (_PRICE, True), + "seen": (_PRICE, True), + "active": (_PRICE, False), + "new": (_PRICE, True), + } + finally: + db.rollback() + db.close() + + +def test_events_survive_change_only_writes() -> None: + """event-diff видит только записанные сегодня строки — и не теряет ни одного события.""" + db = _live_session() + try: + ids = _seed(db) + _run_writer(db) + db.execute(snap_mod._EVENT_DIFF_SQL).fetchall() + rows = db.execute( + text( + """ + SELECT listing_source_id, event_type, diff_percent + FROM listing_source_events + WHERE listing_source_id = ANY(:ids) + """ + ), + {"ids": list(ids.values())}, + ).fetchall() + tag_of = {lsid: tag for tag, lsid in ids.items()} + events = {(tag_of[int(r[0])], r[1], None if r[2] is None else float(r[2])) for r in rows} + + assert events == { + ("price", "price_change", 2.0408), # round((5.0-4.9)/4.9*100, 4) + ("payload", "edited", None), + ("new", "first_seen", None), + } + finally: + db.rollback() + db.close() + + +def test_rerun_same_day_compares_with_todays_row() -> None: + """Второй прогон в те же сутки сравнивает с уже записанной сегодня строкой. + + price вернулся к вчерашней цене — сегодняшняя строка обязана перезаписаться + (иначе на сегодня осталась бы цена, которой у источника больше нет); same + изменился после первого прогона — строка появляется; seen не менялся после + первого прогона — строка остаётся как была. + """ + db = _live_session() + try: + ids = _seed(db) + _run_writer(db) + db.execute( + text("UPDATE listing_sources SET price_rub = 4900000 WHERE id = :id"), + {"id": ids["price"]}, + ) + db.execute( + text("UPDATE listing_sources SET price_rub = 5100000 WHERE id = :id"), + {"id": ids["same"]}, + ) + _run_writer(db) + today = _today(db, list(ids.values())) + + assert today.get(ids["price"]) == (4_900_000, True) + assert today.get(ids["same"]) == (5_100_000, True) + assert today.get(ids["seen"]) == (_PRICE, True) + assert ids["same_gap"] not in today + finally: + db.rollback() + db.close() -- 2.45.3