From 54714f0aa2575f3cd769b26bfce41de852dddc6f Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 6 Aug 2026 21:12:17 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein/scrapers):=20=D0=BC=D0=B5=D1=82?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BD=D0=B0=D0=B1=D0=BB=D1=8E=D0=B4=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=E2=80=94=20=D0=B2=D1=80=D0=B5=D0=BC=D1=8F=20?= =?UTF-8?q?=D1=81=D1=82=D1=80=D0=BE=D0=BA=D0=B8,=20=D0=B0=20=D0=BD=D0=B5?= =?UTF-8?q?=20=D1=81=D1=82=D0=B0=D1=80=D1=82=D0=B0=20=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=BD=D0=B7=D0=B0=D0=BA=D1=86=D0=B8=D0=B8=20(#2731)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit save_listings и upsert_listing_snapshot писали observed_at/scraped_at/last_seen_at через NOW() == transaction_timestamp(). Коммит у писателя один, в конце всего batch'а, поэтому все строки вызова несли одну метку: прод 2026-08-06 — run 3303 219 строк на 1 метку при семи минутах работы, run 3229 2643 строки на 55 меток (метка на вызов save_listings, не на строку). Цена — не фильтры свежести: смещение равно длительности прогона (минуты; худший случай 17.3 ч у cian_full_load) против окна в 14 суток, то есть 0.03-5%. Теряется разрешение во времени — темп сбора и порядок строк внутри прогона по данным не восстановить, отсюда же следовала невозможность бэкфилла run_id (#2701). statement_timestamp(), а НЕ clock_timestamp() как в #2702/#2718: здесь одним statement'ом пишутся ДВЕ колонки, обязанные совпадать (после #2206 scraped_at = last_seen_at у 100% строк, на этом равенстве стоит предикат миграции 161). Прод-проверка: clock_timestamp() = clock_timestamp() → false, а statement_timestamp() = statement_timestamp() → true; при этом statement_timestamp() двигается от запроса к запросу внутри одной транзакции, а каждый upsert — свой запрос. Set-based писатели (listing_source_snapshots 89827 строк одним statement'ом, deactivate_stale_avito) намеренно НЕ тронуты: там одна метка — правда о statement'е, а clock_timestamp() дал бы ложное разрешение (прод: 27783 разных значения на 200000 строк одного statement'а, кодирующих порядок обработки, а не наблюдения). Миграция 232 — только COMMENT ON COLUMN: фиксирует дату перехода и то, что исторические строки невосстановимы (внутрипрогонное время нигде не сохранялось). Refs #2731 --- .../232_listings_observation_time_meaning.sql | 84 +++++ .../tests/test_2731_observation_timestamps.py | 292 ++++++++++++++++++ .../tests/test_scraped_at_bump_2206.py | 20 +- .../scraper-kit/src/scraper_kit/base.py | 23 +- .../src/scraper_kit/snapshot_writer.py | 17 +- 5 files changed, 423 insertions(+), 13 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/232_listings_observation_time_meaning.sql create mode 100644 tradein-mvp/backend/tests/test_2731_observation_timestamps.py diff --git a/tradein-mvp/backend/data/sql/232_listings_observation_time_meaning.sql b/tradein-mvp/backend/data/sql/232_listings_observation_time_meaning.sql new file mode 100644 index 00000000..adf9e522 --- /dev/null +++ b/tradein-mvp/backend/data/sql/232_listings_observation_time_meaning.sql @@ -0,0 +1,84 @@ +-- 232_listings_observation_time_meaning.sql +-- Purpose (#2731): зафиксировать в схеме, что содержательные метки наблюдения до +-- этой правки несли время НАЧАЛА транзакции сбора, а не момент наблюдения строки, +-- и назвать дату, с которой их смысл изменился. +-- +-- Dependencies: 002_core_tables.sql (listings.scraped_at/last_seen_at), +-- 016_listings_snapshots.sql (listings_snapshots.observed_at), +-- 161_backfill_scraped_at_active_recent.sql (ретро-выравнивание scraped_at по last_seen_at). +-- Apply after: 230_house_merge_log.sql +-- Идемпотентно: только COMMENT ON COLUMN (перезаписывает сам себя), данных не трогает. +-- +-- ── ЧТО БЫЛО ───────────────────────────────────────────────────────────────── +-- Писатель объявлений (packages/scraper-kit/.../base.py::save_listings + snapshot_writer) +-- ставил NOW(). В PostgreSQL NOW() == transaction_timestamp() — время старта транзакции. +-- save_listings коммитит ОДИН раз в конце всего batch'а, поэтому одну метку получали все +-- строки одного вызова, сколько бы он ни работал. +-- +-- Прод-замер 2026-08-06 (listings_snapshots, строк / различных меток на прогон): +-- run 3303 — 219 / 1 (прогон шёл 7 минут, все метки на нулевой секунде); +-- run 3299 — 235 / 1; run 3293 — 297 / 1; +-- run 3229 — 2643 / 55 (метка на вызов save_listings, а не на строку). +-- listings за те же сутки: 219/1, 235/1, 297/1 … — и scraped_at, и last_seen_at, причём +-- у 100% строк они РАВНЫ между собой (замер по часам: eq == rows во всех корзинах). +-- +-- ── ЧЕГО ЭТО НЕ ЛОМАЛО ─────────────────────────────────────────────────────── +-- Фильтры свежести и TTL — НЕ искажены. Смещение равно длительности прогона: типично +-- минуты, худший случай на проде 17.3 ч (cian_full_load). Против окна свежести в 14 суток +-- это 0.03-5%. Утверждение «долгие прогоны ломают фильтр свежести» проверено и снято. +-- +-- ── ЧТО ЭТО ЛОМАЛО ─────────────────────────────────────────────────────────── +-- Разрешение во времени. По данным нельзя восстановить ни темп сбора, ни порядок строк +-- внутри прогона: все они выглядят одномоментными. Отсюда же следовала невозможность +-- бэкфилла run_id (#2701). Ошибка тихая — значения правдоподобны. +-- +-- ── ПОЧЕМУ ИСТОРИЮ НЕ ЧИНИМ ────────────────────────────────────────────────── +-- Внутрипрогонное время НИГДЕ БОЛЬШЕ НЕ СОХРАНЯЛОСЬ: у прогона есть только started_at и +-- finished_at, а распределение строк между ними неизвестно. Строки ДО перехода +-- невосстановимы — их метки помечаются, а не переписываются. +-- +-- ── ПОЧЕМУ statement_timestamp(), А НЕ clock_timestamp() ───────────────────── +-- В #2702/#2718 (служебные колонки прогона) взяли clock_timestamp() — там колонка одна. +-- Здесь в одном statement'е пишутся ДВЕ колонки, и они обязаны совпадать: после #2206 +-- scraped_at и last_seen_at равны, и на этом равенстве стоит предикат миграции 161 +-- (`WHERE last_seen_at > scraped_at` как признак «видели живым, но не пере-скрейпили»). +-- Проверка на проде: `clock_timestamp() = clock_timestamp()` → false, +-- `statement_timestamp() = statement_timestamp()` → true. При этом statement_timestamp() +-- двигается ОТ STATEMENT'А К STATEMENT'У внутри одной транзакции (проверено: 1.2 с между +-- соседними запросами при неподвижном now()), а каждый upsert объявления — свой statement. +-- Итог: построчная метка без ложного расхождения колонок. +-- +-- Set-based писатели (listing_source_snapshots, deactivate_stale_avito) НАМЕРЕННО оставлены +-- на now(): там один statement пишет десятки тысяч строк, и одна метка — это правда о нём. +-- clock_timestamp() выдал бы там ложное разрешение: на проде +-- `count(DISTINCT clock_timestamp())` по 200 000 строк одного statement'а = 27 783 разных +-- значения, кодирующих порядок обработки строк планировщиком, а не порядок наблюдения. + +BEGIN; + +COMMENT ON COLUMN listings_snapshots.observed_at IS + 'Момент наблюдения снимка. С #2731 (2026-08-06) — statement_timestamp(), то есть ' + 'время записи КОНКРЕТНОЙ строки. У строк ДО этой даты значение общее на весь вызов ' + 'save_listings (писалось now() = старт транзакции batch''а): на проде 219 снимков ' + 'семиминутного прогона несли одну метку. Для строк до перехода observed_at читать как ' + '«прогон, в котором строку увидели», а НЕ как момент; темп сбора и порядок строк ' + 'внутри прогона по историческим данным невосстановимы — внутрипрогонное время нигде ' + 'не сохранялось.'; + +COMMENT ON COLUMN listings.scraped_at IS + 'Момент последнего скрейпа объявления. С #2206 двигается и при ре-подтверждении живым ' + '(не только при вставке), с #2731 (2026-08-06) пишется statement_timestamp() — ' + 'построчно. У строк ДО этой даты — время старта транзакции batch''а, общее на все ' + 'строки вызова save_listings. На фильтры свежести это не влияло: смещение равно ' + 'длительности прогона (минуты; худший случай 17.3 ч у cian_full_load) против окна в ' + '14 суток. Невосстановимо потеряно другое — разрешение внутри прогона.'; + +COMMENT ON COLUMN listings.last_seen_at IS + 'Момент последнего подтверждения, что объявление живо. Пишется тем же statement''ом, ' + 'что и scraped_at, и потому РАВЕН ему — это свойство сохранено намеренно: ' + 'с #2731 (2026-08-06) обе колонки берут statement_timestamp(), стабильный в statement''е ' + '(clock_timestamp() развёл бы их на микросекунды и сделал бы предикат ' + '`last_seen_at > scraped_at` из миграции 161 истинным почти для всех строк). ' + 'У строк ДО перехода — время старта транзакции batch''а, см. комментарий к scraped_at.'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_2731_observation_timestamps.py b/tradein-mvp/backend/tests/test_2731_observation_timestamps.py new file mode 100644 index 00000000..d78fad6d --- /dev/null +++ b/tradein-mvp/backend/tests/test_2731_observation_timestamps.py @@ -0,0 +1,292 @@ +"""#2731: метка наблюдения — время строки, а не старта транзакции сбора. + +Что было. `save_listings` и `upsert_listing_snapshot` писали содержательные колонки +времени через `NOW()`, а `NOW()` в PostgreSQL — синоним `transaction_timestamp()`: +он замерзает на СТАРТЕ транзакции. Коммит у писателя объявлений ОДИН, в конце всего +batch'а, поэтому все строки вызова получали одну и ту же метку — сколько бы он ни работал. + +Прод-замер 2026-08-06 (listings_snapshots, строк / различных меток на прогон): + run 3303 — 219 / 1, при семи минутах работы (11:59:59 → 12:07:19); + run 3299 — 235 / 1; run 3293 — 297 / 1; + run 3229 — 2643 / 55 — метка на ВЫЗОВ save_listings, а не на строку. +`listings.scraped_at` и `last_seen_at` за те же сутки схлопнуты так же и при этом равны +друг другу у 100% строк (по часам: eq == rows во всех корзинах). + +Цена дефекта — НЕ фильтры свежести (смещение равно длительности прогона: минуты, худший +случай 17.3 ч, против окна 14 суток), а разрешение во времени: темп сбора и порядок строк +внутри прогона по данным не восстановить (отсюда же невозможность бэкфилла run_id, #2701). + +Почему `statement_timestamp()`, а не `clock_timestamp()` как в #2702/#2718: здесь в одном +statement'е пишутся ДВЕ колонки, обязанные совпадать (после #2206 scraped_at = last_seen_at, +на этом равенстве стоит предикат миграции 161). На проде проверено: +`clock_timestamp() = clock_timestamp()` → false, `statement_timestamp() = ...` → true. +Модель БД ниже воспроизводит ровно это различие, поэтому «починка» через clock_timestamp() +провалит тест на согласованность. + +Фальсификация: на старом коде (`NOW()`) тесты 1 и 2 дают одну метку на все строки. +""" + +from __future__ import annotations + +import inspect +import os +import re +from contextlib import contextmanager +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit import base as kit_base +from scraper_kit import snapshot_writer as kit_snapshot +from scraper_kit.base import ScrapedLot, save_listings + +# Сколько «работает» писатель на одну строку. Пять лотов → 2 с работы одной транзакции: +# требование задачи — писатель, отработавший ДОЛЬШЕ СЕКУНДЫ, оставляет различные метки. +_STEP_SEC = 0.4 + +# Содержательные колонки времени, за которыми следим, по таблицам-писателям. +_WATCHED: dict[str, tuple[str, ...]] = { + "listings": ("scraped_at", "last_seen_at"), + "listings_snapshots": ("observed_at",), +} + + +def _strip_sql_comments(sql: str) -> str: + """Убрать `--`-комментарии: в них есть и скобки, и слово NOW() в объяснительной прозе.""" + return re.sub(r"--[^\n]*", "", sql) + + +def _split_top(items: str) -> list[str]: + """Разбить список SQL-элементов по запятым ВЕРХНЕГО уровня (CAST(:x AS t) — один).""" + out: list[str] = [] + depth = 0 + cur = "" + for ch in items: + if ch == "," and depth == 0: + out.append(cur.strip()) + cur = "" + continue + depth += (ch == "(") - (ch == ")") + cur += ch + out.append(cur.strip()) + return out + + +def _balanced(sql: str, open_idx: int) -> tuple[str, int]: + """Содержимое скобки, открытой на open_idx, и индекс её закрывающей пары.""" + depth = 0 + for i in range(open_idx, len(sql)): + depth += (sql[i] == "(") - (sql[i] == ")") + if depth == 0: + return sql[open_idx + 1 : i], i + raise AssertionError("несбалансированные скобки в SQL") + + +class _Row: + """Строка ответа: id/inserted для INSERT INTO listings ... RETURNING.""" + + id = 42 + inserted = False + card_hash = None + + +class _FakeResult: + def __init__(self, row: _Row | None) -> None: + self._row = row + + def fetchone(self) -> _Row | None: + return self._row + + def first(self) -> _Row | None: + return self._row + + def scalar_one_or_none(self) -> None: + return None + + +class _FakePg: + """Мини-модель PostgreSQL на три функции времени и ленивую транзакцию. + + `now()` == `transaction_timestamp()` — замерзает на старте транзакции (autobegin на + первом execute, сброс на commit/rollback); + `statement_timestamp()` — момент начала ТЕКУЩЕГО statement'а: двигается от запроса к + запросу и СТАБИЛЕН внутри запроса; + `clock_timestamp()` — настоящие часы, вычисляются на КАЖДЫЙ вызов отдельно, поэтому + два вызова в одном statement'е дают разные значения (проверено на проде). + + Каждый execute «работает» _STEP_SEC — так модель отличает построчную метку от общей. + """ + + def __init__(self) -> None: + self.wall: float = 0.0 + self.tx_start: float | None = None + self.stmt_start: float = 0.0 + self._clock_calls: int = 0 + # Что осело в таблицах: список строк {колонка: значение}. + self.rows: dict[str, list[dict[str, float]]] = {t: [] for t in _WATCHED} + + # ── модель времени ──────────────────────────────────────────────────────── + def _value(self, func: str) -> float: + if func == "now": + assert self.tx_start is not None + return self.tx_start + if func == "statement_timestamp": + return self.stmt_start + if func == "clock_timestamp": + self._clock_calls += 1 # каждый вызов — своё значение + return self.stmt_start + self._clock_calls * 1e-6 + raise AssertionError(f"неизвестная функция времени: {func}") + + # ── разбор писателя ─────────────────────────────────────────────────────── + def _record(self, sql: str) -> None: + ins = re.search(r"INSERT INTO (listings|listings_snapshots)\s*\(", sql) + if ins is not None: + table = ins.group(1) + cols, end = _balanced(sql, ins.end() - 1) + vals_kw = re.compile(r"\bVALUES\s*\(").search(sql, end) + assert vals_kw is not None, "INSERT без VALUES" + vals, _ = _balanced(sql, vals_kw.end() - 1) + row: dict[str, float] = {} + for col, val in zip(_split_top(cols), _split_top(vals), strict=False): + f = re.fullmatch(r"(\w+)\s*\(\s*\)", val) + if col in _WATCHED[table] and f is not None: + row[col] = self._value(f.group(1).lower()) + if row: + self.rows[table].append(row) + return + upd = re.search(r"UPDATE (listings)\b", sql) + if upd is not None: # reconcile-путь при дрейфе dedup_hash + table = upd.group(1) + row = {} + for col in _WATCHED[table]: + m = re.search(rf"\b{col}\s*=\s*(\w+)\s*\(\s*\)", sql) + if m is not None: + row[col] = self._value(m.group(1).lower()) + if row: + self.rows[table].append(row) + + # ── интерфейс сессии ────────────────────────────────────────────────────── + def execute(self, stmt: Any, params: Any = None) -> _FakeResult: + if self.tx_start is None: + self.tx_start = self.wall + self.stmt_start = self.wall + self.wall += _STEP_SEC # statement отработал + sql = _strip_sql_comments(str(stmt)) + self._record(sql) + if "INSERT INTO listings (" in sql: + return _FakeResult(_Row()) + return _FakeResult(None) + + def commit(self) -> None: + self.tx_start = None + + def rollback(self) -> None: + self.tx_start = None + + def begin_nested(self) -> Any: + @contextmanager + def _ctx() -> Any: + yield MagicMock() + + return _ctx() + + +def _matcher() -> MagicMock: + matcher = MagicMock() + matcher.match_or_create_house.return_value = (101, 1.0, "new") + matcher.upsert_listing_source.return_value = None + return matcher + + +def _lots(n: int = 5) -> list[ScrapedLot]: + return [ + ScrapedLot( + source="cian", + source_url=f"https://ekb.cian.ru/sale/flat/{i}/", + source_id=str(i), + price_rub=5_000_000 + i, + ) + for i in range(n) + ] + + +def _run_writer() -> _FakePg: + db = _FakePg() + save_listings(db, _lots(), matcher=_matcher(), region_code=66) + return db + + +# ── 1. Снимки одного batch'а датируются построчно ──────────────────────────── + + +def test_snapshot_marks_differ_per_row() -> None: + """Прогон 3303 дословно: строки одного вызова — разные моменты, а не один. + + На старом коде (`NOW()`) все снимки batch'а несут метку старта транзакции — ровно + 219 строк на одну метку, ради чего заведена задача. + """ + db = _run_writer() + marks = [row["observed_at"] for row in db.rows["listings_snapshots"]] + + assert len(marks) == 5, "снимок пишется на каждый сохранённый лот" + assert max(marks) - min(marks) > 1.0, "писатель отработал дольше секунды" + assert len(set(marks)) == len(marks), "у строк одного batch'а обязаны быть разные метки" + + +# ── 2. scraped_at/last_seen_at: построчно, но по-прежнему равны между собой ── + + +def test_listing_marks_differ_per_row_and_stay_equal_to_each_other() -> None: + """Две колонки одного statement'а: построчная метка без нового расхождения. + + Равенство scraped_at = last_seen_at — не косметика: на нём стоит предикат миграции + 161 (`last_seen_at > scraped_at` как признак «видели живым, но не пере-скрейпили»). + `clock_timestamp()` вычисляется на каждый вызов отдельно и развёл бы колонки на + микросекунды — модель БД это воспроизводит, поэтому такая «починка» провалит тест. + """ + db = _run_writer() + rows = db.rows["listings"] + + assert len(rows) == 5 + for row in rows: + assert row["scraped_at"] == row["last_seen_at"], "колонки statement'а обязаны совпадать" + scraped = [row["scraped_at"] for row in rows] + assert max(scraped) - min(scraped) > 1.0 + assert len(set(scraped)) == len(scraped), "метки строк одного batch'а обязаны различаться" + + +# ── 3. Инвариант источника: содержательная колонка не пишется now() ───────── + + +@pytest.mark.parametrize("mod", [kit_base, kit_snapshot], ids=["base", "snapshot_writer"]) +def test_no_content_timestamp_is_written_with_now(mod: Any) -> None: + """`<колонка> = NOW()` в этих писателях не имеет корректного применения. + + Комментарии вырезаются: `NOW()` там остаётся в объяснительной прозе (и должен — + без неё следующий читатель повторит дефект). + """ + src = _strip_sql_comments(inspect.getsource(mod)) + cols = "|".join(sorted({c for t in _WATCHED.values() for c in t})) + assert re.search(rf"\b({cols})\s*=\s*NOW\s*\(\s*\)", src, re.I) is None + assert "statement_timestamp()" in src + + +# ── 4. Историческая граница зафиксирована в схеме ──────────────────────────── + +_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" +_MIGRATION = _SQL_DIR / "232_listings_observation_time_meaning.sql" + + +def test_migration_232_marks_the_transition_date_and_unrecoverable_history() -> None: + """Аналитике нужно знать, где сменился смысл колонки и что до него не чинится.""" + sql = _MIGRATION.read_text("utf-8") + assert "BEGIN;" in sql and "COMMIT;" in sql + for col in ("listings_snapshots.observed_at", "listings.scraped_at", "listings.last_seen_at"): + assert f"COMMENT ON COLUMN {col} IS" in sql + assert sql.count("#2731 (2026-08-06)") >= 3, "дата перехода — в каждом комментарии" + assert "невосстановим" in sql + assert not re.search(r":\w+::", sql), "psycopg v3: никаких :param::type" diff --git a/tradein-mvp/backend/tests/test_scraped_at_bump_2206.py b/tradein-mvp/backend/tests/test_scraped_at_bump_2206.py index f5e3b247..784df63d 100644 --- a/tradein-mvp/backend/tests/test_scraped_at_bump_2206.py +++ b/tradein-mvp/backend/tests/test_scraped_at_bump_2206.py @@ -7,7 +7,8 @@ эстиматору было видно лишь ~45.6% активного инвентаря (CIAN — 5.9%). Фикс: и ON CONFLICT DO UPDATE, и dedup-drift reconcile UPDATE теперь выставляют -scraped_at = NOW() рядом с last_seen_at = NOW(). Проверяем оба пути в +scraped_at рядом с last_seen_at (с #2731 — statement_timestamp(), см. +test_2731_observation_timestamps.py; до него NOW()). Проверяем оба пути в `scraper_kit.base` (единственный живой модуль — `app.services.scrapers.base` удалён #2397 финальный шаг E, mirror-тесты через legacy убраны) и свойства ретро-бэкфилл-миграции 161. @@ -120,7 +121,12 @@ def _kit_matcher() -> MagicMock: def test_kit_on_conflict_bumps_scraped_at() -> None: - """kit ON CONFLICT DO UPDATE двигает scraped_at = NOW().""" + """kit ON CONFLICT DO UPDATE двигает scraped_at рядом с last_seen_at. + + #2731: обе метки пишутся statement_timestamp() (построчно, но одинаково внутри + statement'а) вместо NOW() — инвариант #2206 «scraped_at двигается вместе с + last_seen_at» от этого не меняется, меняется только источник времени. + """ db = _mock_db_update_path(inserted=False) lot = KitLot( source="cian", @@ -134,12 +140,12 @@ def test_kit_on_conflict_bumps_scraped_at() -> None: assert (inserted, updated) == (0, 1) sql = _find_sql(db, "INSERT INTO listings (") - assert "scraped_at = NOW()" in sql - assert "last_seen_at = NOW()" in sql + assert "scraped_at = statement_timestamp()" in sql + assert "last_seen_at = statement_timestamp()" in sql def test_kit_reconcile_bumps_scraped_at() -> None: - """kit dedup-drift reconcile UPDATE двигает scraped_at = NOW().""" + """kit dedup-drift reconcile UPDATE двигает scraped_at рядом с last_seen_at (#2731).""" db = _mock_db_reconcile(reconcile_id=88) lot = KitLot( source="avito", @@ -152,8 +158,8 @@ def test_kit_reconcile_bumps_scraped_at() -> None: kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66) sql = _find_sql(db, "SET dedup_hash") - assert "scraped_at = NOW()" in sql - assert "last_seen_at = NOW()" in sql + assert "scraped_at = statement_timestamp()" in sql + assert "last_seen_at = statement_timestamp()" in sql # ── Migration 161: retro-backfill scraped_at ────────────────────────────────── diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py index 4c4c4d36..44ddd374 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py @@ -553,7 +553,16 @@ def save_listings( :predicted_price_min, :predicted_price_max, :price_trend, :price_previous_rub, :geo_precision, :card_hash, - NOW(), NOW() + -- scraped_at / last_seen_at — statement_timestamp(), НЕ NOW() (#2731). + -- NOW() == transaction_timestamp() замерзает на старте транзакции, а + -- save_listings коммитит один раз в конце всего batch'а → все строки + -- прогона несли ОДНУ метку (прод 2026-08-06: 219 строк — 1 метка). + -- statement_timestamp() двигается построчно (каждый upsert — свой + -- statement) и СТАБИЛЕН внутри statement'а, поэтому обе колонки + -- получают одно и то же значение. clock_timestamp() здесь нельзя: + -- он вычисляется на каждый вызов отдельно и развёл бы scraped_at и + -- last_seen_at на микросекунды — расхождение там, где его нет. + statement_timestamp(), statement_timestamp() ) -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)). -- Для одного (source, source_id) формула даёт ОДИН dedup_hash, @@ -563,10 +572,12 @@ def save_listings( -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL, -- а NULL не годится как conflict-target → дубли вернулись бы (#1773). ON CONFLICT (dedup_hash) DO UPDATE - SET last_seen_at = NOW(), + SET last_seen_at = statement_timestamp(), -- #2206: ре-подтверждение живым = свежий скрейп; без bump'а -- эстиматор (scraped_at > NOW()-14d) терял ~55% живого инвентаря. - scraped_at = NOW(), + -- #2731: обе метки — statement_timestamp() (см. VALUES выше); + -- их равенство сохраняется, потому что оно стабильно в statement'е. + scraped_at = statement_timestamp(), is_active = true, -- если цена изменилась — обновляем price_rub = EXCLUDED.price_rub, @@ -679,10 +690,12 @@ def save_listings( """ UPDATE listings SET dedup_hash = :dedup, - last_seen_at = NOW(), + last_seen_at = statement_timestamp(), -- #2206: ре-подтверждение живым = свежий скрейп; -- без bump'а эстиматор терял ~55% живого инвентаря. - scraped_at = NOW(), + -- #2731: statement_timestamp() — построчная метка, + -- одинаковая для обеих колонок (см. INSERT выше). + scraped_at = statement_timestamp(), is_active = true, price_rub = :price_rub, price_per_m2 = :ppm2, diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py index 3847d323..641caf61 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py @@ -15,6 +15,21 @@ (data-modifying CTE в app/tasks/deactivate_stale_avito.py), не через этот per-row хелпер. +observed_at пишется `statement_timestamp()`, а НЕ `NOW()` (#2731). `NOW()` в PostgreSQL — +синоним `transaction_timestamp()`: он замерзает на СТАРТЕ транзакции, а save_listings +коммитит ОДИН раз в конце всего batch'а — то есть все снимки прогона получали ОДНУ метку. +Прод-замер 2026-08-06: run 3303 — 219 строк, 1 различная метка, при семи минутах работы; +run 3229 — 2643 строки на 55 меток (метка на save_listings-вызов, не на строку). +Цена — не фильтры свежести (14 суток против минут прогона), а разрешение во времени: +темп сбора и порядок внутри прогона по данным не восстановить (отсюда же #2701). + +Почему statement_timestamp(), а не clock_timestamp() как в #2702/#2718: этот INSERT — один +statement на строку (вызов в цикле save_listings), поэтому statement_timestamp() двигается +ПОСТРОЧНО и при этом СТАБИЛЕН внутри строки. clock_timestamp() дал бы разные значения даже +двум колонкам одного statement'а (прод-проверка: `clock_timestamp() = clock_timestamp()` → f, +`statement_timestamp() = statement_timestamp()` → t) — это сломало бы равенство +listings.scraped_at = last_seen_at, на которое опирается предикат миграции 161. + position_in_serp (#2674): параметр УДАЛЁН, колонка осталась и остаётся NULL. Это не оборванная проводка — задуманное отношение НЕВЫРАЗИМО в этой таблице. Позиция — свойство пары (объявление, конкретный прогон выдачи с конкретными @@ -87,7 +102,7 @@ def upsert_listing_snapshot( CAST(:price AS bigint), CAST(:ppm2 AS int), :status, - NOW() + statement_timestamp() ) ON CONFLICT (listing_id, snapshot_date) DO UPDATE SET run_id = COALESCE(EXCLUDED.run_id, listings_snapshots.run_id), -- 2.45.3