fix(tradein/newbuilding): счётчики записи различают вставку и обновление (#2807) #2809

Merged
bot-backend merged 1 commit from fix/2807-price-dynamics-counter into main 2026-08-10 08:50:57 +00:00
6 changed files with 322 additions and 53 deletions

View file

@ -1961,8 +1961,9 @@ async def scrape_cian_newbuilding(
saved = False
if house_id is not None:
# save_newbuilding_enrichment — sync (def, returns None); await на sync-функции
# раньше поднимал TypeError на любом вызове с house_id.
# save_newbuilding_enrichment — sync (def, не корутина); await на sync-функции
# раньше поднимал TypeError на любом вызове с house_id. Возвращаемый счёт
# записанного (#2807) этой ручке не нужен — она отвечает фактом сохранения.
save_newbuilding_enrichment(db, house_id, enrichment)
saved = True

View file

@ -109,10 +109,17 @@ class NewbuildingEnrichBackfillResult:
failed_fetch: int = 0 # fetch returned None / raised
failed_save: int = 0 # save raised after a good fetch
# Row-level deltas (how much actually landed).
price_dynamics_rows: int = 0
reliability_rows: int = 0
review_rows: int = 0
# Сколько РЕАЛЬНО записано, по словам самих писателей (#2807). Раньше здесь стоял
# прирост COUNT(*) по таблице до/после сохранения — то есть «выросла ли таблица», а
# не «сколько записали»: при ON CONFLICT DO UPDATE обновление даёт ноль, а у
# reliability ноль давал ещё и _dedup_reliability, схлопывающий дубль сразу после
# вставки. Ключи переименованы намеренно: у price_dynamics_rows/reliability_rows/
# review_rows в истории прогонов старый смысл, и молча поменять его под тем же
# именем — ровно тот дефект, ради которого правка и делается.
price_dynamics_inserted: int = 0 # новых точек динамики цен
price_dynamics_updated: int = 0 # существующих точек переписано свежей ценой
reliability_inserted: int = 0 # строк house_reliability_checks вставлено
review_upserted: int = 0 # отзывов записано (вставка+обновление, ключ ext_review_id)
duration_sec: float = field(default=0.0)
@ -557,15 +564,15 @@ async def backfill_newbuilding_enrichment(
continue
# ── Save under a SAVEPOINT so one bad house can't poison the batch ──
# begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally,
# so we snapshot the row counts BEFORE and recompute the delta AFTER its commit
# rather than relying on the nested transaction staying open.
pd_before, rc_before, rv_before = _house_enrichment_counts(db, house_id)
# begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally.
# COUNT(*) до сохранения нужен ТОЛЬКО для had_reliability (дедуп ниже): сколько
# записано, теперь сообщают сами писатели, а не разница COUNT'ов (#2807).
_, rc_before, _ = _house_enrichment_counts(db, house_id)
try:
had_reliability = rc_before > 0
# 1) price_dynamics + reliability + houses UPDATE (existing, commits inside).
save_newbuilding_enrichment(db, house_id, enrichment)
saved = save_newbuilding_enrichment(db, house_id, enrichment)
# 2) reviews — added here (save_newbuilding_enrichment skips them).
# SAVEPOINT around the review write so a malformed review can't lose the
@ -601,16 +608,18 @@ async def backfill_newbuilding_enrichment(
sp.rollback()
logger.warning("reliability dedup failed house_id=%s: %s", house_id, dexc)
pd_after, rc_after, rv_after = _house_enrichment_counts(db, house_id)
result.price_dynamics_rows += max(0, pd_after - pd_before)
result.reliability_rows += max(0, rc_after - rc_before)
result.review_rows += max(0, rv_after - rv_before)
result.price_dynamics_inserted += saved.price_inserted
result.price_dynamics_updated += saved.price_updated
result.reliability_inserted += saved.reliability_inserted
result.review_upserted += review_written
result.succeeded += 1
logger.info(
"enriched house_id=%s: +pd=%d +reliability=%d +reviews=%d (parsed reviews=%d)",
"enriched house_id=%s: динамика цен +%d новых / %d обновлено, "
"reliability +%d, отзывов записано %d (распознано %d)",
house_id,
max(0, pd_after - pd_before),
max(0, rc_after - rc_before),
saved.price_inserted,
saved.price_updated,
saved.reliability_inserted,
review_written,
len(enrichment.reviews),
)
@ -629,8 +638,8 @@ async def backfill_newbuilding_enrichment(
result.duration_sec = time.time() - t0
logger.info(
"newbuilding-enrich backfill done: processed=%d ok=%d skip=%d resolved=%d "
"resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
"| %.1fs",
"resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
"%d обновлено, reliability +%d, отзывов %d | %.1fs",
result.processed,
result.succeeded,
result.skipped_already_enriched,
@ -638,9 +647,10 @@ async def backfill_newbuilding_enrichment(
result.failed_resolve,
result.failed_fetch,
result.failed_save,
result.price_dynamics_rows,
result.reliability_rows,
result.review_rows,
result.price_dynamics_inserted,
result.price_dynamics_updated,
result.reliability_inserted,
result.review_upserted,
result.duration_sec,
)
return result
@ -787,8 +797,8 @@ async def run_newbuilding_enrich(
)
logger.info(
"scheduler: newbuilding_enrich run_id=%d finished — processed=%d ok=%d skip=%d "
"resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
"| pending=%d %.1fs",
"resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
"%d обновлено, reliability +%d, отзывов %d | pending=%d %.1fs",
run_id,
result.processed,
result.succeeded,
@ -796,9 +806,10 @@ async def run_newbuilding_enrich(
result.failed_resolve,
result.failed_fetch,
result.failed_save,
result.price_dynamics_rows,
result.reliability_rows,
result.review_rows,
result.price_dynamics_inserted,
result.price_dynamics_updated,
result.reliability_inserted,
result.review_upserted,
result.cian_houses_pending,
result.duration_sec,
)

View file

@ -19,7 +19,10 @@ _wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
import pytest # noqa: E402
from scraper_kit.providers.cian.newbuilding import NewbuildingEnrichment # noqa: E402
from scraper_kit.providers.cian.newbuilding import ( # noqa: E402
NewbuildingEnrichment,
NewbuildingSaveCounts,
)
from app.tasks.newbuilding_enrich_backfill import ( # noqa: E402
NewbuildingEnrichBackfillResult,
@ -168,24 +171,37 @@ def _enrichment_with_everything(seed: int = 0) -> NewbuildingEnrichment:
def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
"""Stand-in for the real saver: lands price_dynamics + reliability into FakeDB."""
"""Stand-in for the real saver: lands price_dynamics + reliability into FakeDB.
Возвращает NewbuildingSaveCounts, как настоящий (#2807): вставку от обновления
различает сам писатель снаружи по таблице их не отличить (UPSERT по dim_key).
"""
inserted = updated = 0
for p in enrichment.realty_valuation_chart:
if p.get("price_per_sqm") is None:
continue
db.price_dynamics.add(
(
house_id,
p["month_date"],
"cian_realty_valuation",
p.get("room_count", "all"),
p.get("prices_type", "price"),
p.get("period", "halfYear"),
)
key = (
house_id,
p["month_date"],
"cian_realty_valuation",
p.get("room_count", "all"),
p.get("prices_type", "price"),
p.get("period", "halfYear"),
)
if key in db.price_dynamics:
updated += 1
else:
inserted += 1
db.price_dynamics.add(key)
reliability = 0
for c in enrichment.reliability_checks:
if c.get("check_name") or c.get("check_status"):
db.reliability.append((house_id, "cian_nashdom"))
reliability += 1
db.commit()
return NewbuildingSaveCounts(
price_inserted=inserted, price_updated=updated, reliability_inserted=reliability
)
# ---------------------------------------------------------------------------
@ -279,9 +295,10 @@ async def test_backfill_populates_all_three_tables() -> None:
assert len(db.price_dynamics) == 2 # 1 chart point × 2 houses
assert len(db.reliability) == 2
assert len(db.reviews) == 4 # 2 reviews × 2 houses
assert result.price_dynamics_rows == 2
assert result.reliability_rows == 2
assert result.review_rows == 4
assert result.price_dynamics_inserted == 2
assert result.price_dynamics_updated == 0
assert result.reliability_inserted == 2
assert result.review_upserted == 4
@pytest.mark.asyncio
@ -544,7 +561,7 @@ async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.Monk
async def _fake_backfill(_db, *, limit, force, request_delay_sec, on_progress=None):
# on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2)
return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_inserted=2)
monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)

View file

@ -185,7 +185,7 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
_stub_backfill(
monkeypatch,
NewbuildingEnrichBackfillResult(
processed=10, succeeded=3, failed_fetch=6, failed_resolve=1, price_dynamics_rows=7
processed=10, succeeded=3, failed_fetch=6, failed_resolve=1, price_dynamics_inserted=7
),
)
calls = _stub_finalisers(monkeypatch)
@ -196,5 +196,5 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
assert counters["attempted"] == 10
assert counters["enriched"] == 3
assert counters["failed"] == 7
assert counters["price_dynamics_rows"] == 7 # исходные счётчики на месте
assert counters["price_dynamics_inserted"] == 7 # исходные счётчики на месте
assert counters["succeeded"] == 3

View file

@ -0,0 +1,193 @@
"""#2807: счётчик мерил прирост таблицы, а читался как «сколько записали».
`newbuilding_enrich_backfill` считал свою работу разницей `COUNT(*)` до и после
сохранения. Вставка в houses_price_dynamics идёт `ON CONFLICT DO UPDATE`, поэтому
обновление существующей точки давало ноль. Прод 10.08: прогон 3578 отчитался
`price_dynamics_rows: 0`, обновив за своё окно **64 строки по 10 домам** те самые,
что вставил прогон 3563 накануне (у него в тех же counters стояло 64). Ноль читался
как «динамика цен снова не пишется».
Соседние счётчики врали в том же месте по своим причинам: `reliability_rows` обнулял
`_dedup_reliability`, схлопывающий строку сразу после вставки, а `review_rows`
игнорировал число, которое `_save_cian_reviews` УЖЕ возвращал, в пользу разницы COUNT'ов.
Фальсификация (см. прогон в PR): на коде до правки `test_second_pass_reports_updates`
даёт `price_dynamics_rows == 0` при 64 переписанных точках ровно прод-симптом.
Отдельно проверяется, что правка НЕ ослабила сторожа нулевого результата: он смотрит на
`attempted`/`enriched`/`gone`/`blocked` (#2695), а не на счётчики записи, и прогон,
который ничего не обогатил, обязан остаться 'failed' при любых числах в `*_written`.
"""
from __future__ import annotations
import os
import sys
from unittest.mock import MagicMock, patch
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
sys.modules.setdefault("weasyprint", MagicMock())
from scraper_kit.providers.cian.newbuilding import ( # noqa: E402
NewbuildingSaveCounts,
save_newbuilding_enrichment,
)
from app.services import scrape_runs as runs_mod # noqa: E402
from app.tasks.newbuilding_enrich_backfill import ( # noqa: E402
NewbuildingEnrichBackfillResult,
)
# Прод-масштаб прогона 3578: 10 домов × 64/10 точек. Держим ровно 64, чтобы число в
# тесте совпадало с числом в задаче.
PROD_POINTS = 64
class _UpsertDB:
"""Сессия, у которой houses_price_dynamics уже населена (второй проход).
`RETURNING (xmax = 0)` возвращает False на конфликте это и есть «обновили».
"""
def __init__(self, *, already_present: bool) -> None:
self.already_present = already_present
self.price_writes = 0
self.reliability_writes = 0
self.committed = False
def execute(self, statement, params=None):
sql = str(statement)
res = MagicMock()
if "INSERT INTO houses_price_dynamics" in sql:
self.price_writes += 1
assert "RETURNING (xmax = 0)" in sql, "писатель обязан различать вставку и update"
res.fetchone.return_value = (not self.already_present,)
return res
if "INSERT INTO house_reliability_checks" in sql:
self.reliability_writes += 1
res.fetchone.return_value = None
return res
def commit(self) -> None:
self.committed = True
def _enrichment(points: int):
from scraper_kit.providers.cian.newbuilding import NewbuildingEnrichment
return NewbuildingEnrichment(
cian_internal_house_id=1,
cian_zhk_url="https://zhk-x.cian.ru/",
name="ЖК Тест",
realty_valuation_chart=[
{
"month_date": f"2026-{(i % 12) + 1:02d}-01",
"room_count": "all",
"prices_type": "price",
"period": "halfYear",
"price_per_sqm": 150000.0 + i,
}
for i in range(points)
],
reliability_checks=[{"check_name": "Надёжный застройщик", "check_status": "reliable"}],
reviews=[],
)
# ── 1. Писатель различает вставку и обновление ───────────────────────────────
def test_first_pass_reports_inserts() -> None:
db = _UpsertDB(already_present=False)
counts = save_newbuilding_enrichment(db, 42, _enrichment(PROD_POINTS))
assert counts.price_inserted == PROD_POINTS
assert counts.price_updated == 0
assert counts.reliability_inserted == 1
assert db.price_writes == PROD_POINTS
def test_second_pass_reports_updates() -> None:
"""Прод-симптом: те же 64 точки, ничего нового — но записаны все 64.
До правки этот прогон отчитывался нулём по всем трём счётчикам.
"""
db = _UpsertDB(already_present=True)
counts = save_newbuilding_enrichment(db, 42, _enrichment(PROD_POINTS))
assert counts.price_inserted == 0
assert counts.price_updated == PROD_POINTS
assert counts.price_written == PROD_POINTS
assert db.price_writes == PROD_POINTS
def test_nothing_to_write_stays_zero() -> None:
"""Встречная проверка: пустой график — ноль и во «вставлено», и в «обновлено»."""
db = _UpsertDB(already_present=True)
counts = save_newbuilding_enrichment(db, 42, _enrichment(0))
assert (counts.price_inserted, counts.price_updated, counts.price_written) == (0, 0, 0)
def test_points_without_price_are_not_counted_as_written() -> None:
"""Точка без price_per_sqm пропускается писателем — и не попадает в счёт."""
enrichment = _enrichment(2)
enrichment.realty_valuation_chart[0]["price_per_sqm"] = None
db = _UpsertDB(already_present=False)
counts = save_newbuilding_enrichment(db, 42, enrichment)
assert counts.price_written == 1
assert db.price_writes == 1
# ── 2. Сторож нулевого результата не ослаблен ────────────────────────────────
def _finalize(counters: dict[str, int]) -> str:
"""Прогнать counters через боевой финализатор и вернуть выбранный статус."""
chosen: dict[str, str] = {}
with (
patch.object(runs_mod, "mark_done", lambda *a, **k: chosen.setdefault("s", "done")),
patch.object(runs_mod, "mark_failed", lambda *a, **k: chosen.setdefault("s", "failed")),
patch.object(runs_mod, "mark_banned", lambda *a, **k: chosen.setdefault("s", "banned")),
):
runs_mod.mark_backfill_finished(MagicMock(), 1, counters, source="newbuilding_enrich")
return chosen["s"]
def test_watchdog_still_fails_a_run_that_enriched_nothing() -> None:
"""Прогон без обогащений остаётся 'failed', сколько бы записей ни насчитали.
Числа записи в решение сторожа не входят вовсе он судит по attempted/enriched.
Если бы входили, честный счётчик «обновлено» превратил бы холостой прогон в успех.
"""
result = NewbuildingEnrichBackfillResult(
processed=25,
succeeded=0,
failed_fetch=25,
price_dynamics_updated=PROD_POINTS, # «что-то писали» — но никого не обогатили
)
assert _finalize(result.to_backfill_counters()) == "failed"
def test_watchdog_verdict_ignores_the_new_keys() -> None:
"""Явно: добавление/убирание новых ключей не двигает вердикт ни в одну сторону."""
base = {"attempted": 25, "enriched": 3, "failed": 22}
assert _finalize(dict(base)) == "done"
assert _finalize({**base, "price_dynamics_inserted": 0, "price_dynamics_updated": 0}) == "done"
zero = {"attempted": 25, "enriched": 0, "failed": 25}
assert _finalize(dict(zero)) == "failed"
assert _finalize({**zero, "price_dynamics_updated": 999}) == "failed"
def test_counters_carry_both_numbers_into_the_run() -> None:
"""В scrape_runs.counters уезжают ОБА числа — ноль одного больше не читается как ноль."""
counters = NewbuildingEnrichBackfillResult(
processed=25, succeeded=25, price_dynamics_updated=PROD_POINTS
).to_backfill_counters()
assert counters["price_dynamics_inserted"] == 0
assert counters["price_dynamics_updated"] == PROD_POINTS
# Старые имена не должны остаться: у них в истории прогонов другой смысл.
assert "price_dynamics_rows" not in counters
assert "reliability_rows" not in counters
assert "review_rows" not in counters
def test_save_counts_written_is_the_sum() -> None:
assert NewbuildingSaveCounts(price_inserted=3, price_updated=4).price_written == 7

View file

@ -594,11 +594,38 @@ def _extract_nested_offers(offers_state: dict[str, Any]) -> list[dict[str, Any]]
# ---- save helpers ----
@dataclass(frozen=True)
class NewbuildingSaveCounts:
"""Что прогон РЕАЛЬНО записал — вставил и обновил отдельно (#2807).
Заводится потому, что вызывающий мерил свою работу разницей ``COUNT(*)`` по таблице
до и после сохранения. Это прирост ЧИСЛА СТРОК, а не число записанных точек: у
houses_price_dynamics вставка идёт ``ON CONFLICT DO UPDATE``, поэтому обновление
уже существующей точки даёт ноль. Прод 10.08: прогон 3578 отчитался
``price_dynamics_rows: 0``, обновив за своё окно 64 строки по 10 домам (их вставил
прогон 3563 накануне) ноль читался как «динамика цен не пишется».
Единственный, кто знает разницу, сам писатель: ``RETURNING (xmax = 0)`` отличает
вставку от обновления (та же идиома, что в
``backend/app/services/scrapers/gisogd66.py``). Поэтому число возвращается отсюда, а
не восстанавливается снаружи по таблице.
"""
price_inserted: int = 0
price_updated: int = 0
reliability_inserted: int = 0
@property
def price_written(self) -> int:
"""Сколько точек динамики прошло через запись (вставка + обновление)."""
return self.price_inserted + self.price_updated
def save_newbuilding_enrichment(
db: Any,
house_id: int,
enrichment: NewbuildingEnrichment,
) -> None:
) -> NewbuildingSaveCounts:
"""Persist NewbuildingEnrichment to DB.
Steps:
@ -606,6 +633,9 @@ def save_newbuilding_enrichment(
2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present)
3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE)
4. INSERT INTO house_reliability_checks (overall + details)
Returns NewbuildingSaveCounts вставлено/обновлено раздельно (#2807). Вызывающие,
которым счёт не нужен (SERP-sweep, admin re-enrich), просто игнорируют результат.
"""
from sqlalchemy import text
@ -681,14 +711,18 @@ def save_newbuilding_enrichment(
# 3. INSERT houses_price_dynamics
# UNIQUE constraint: houses_price_dynamics_dim_key
# (house_id, source, room_count, prices_type, period, month_date) — per migration 029
chart_saved = 0
price_inserted = 0
price_updated = 0
for point in enrichment.realty_valuation_chart:
if point.get("price_per_sqm") is None:
continue
room_count = point.get("room_count") or "all"
prices_type = point.get("prices_type") or "price"
period = point.get("period") or "halfYear"
db.execute(
# RETURNING (xmax = 0): у только что вставленной строки xmax равен нулю, у
# обновлённой конфликтом — id транзакции. Без этого «вставили» и «обновили»
# снаружи неразличимы, и обновление читается как «ничего не записали» (#2807).
written = db.execute(
text("""
INSERT INTO houses_price_dynamics (
house_id, month_date, source,
@ -705,6 +739,7 @@ def save_newbuilding_enrichment(
ON CONFLICT ON CONSTRAINT houses_price_dynamics_dim_key DO UPDATE SET
price_per_sqm = EXCLUDED.price_per_sqm,
recorded_at = NOW()
RETURNING (xmax = 0) AS is_insert
"""),
{
"hid": house_id,
@ -714,12 +749,16 @@ def save_newbuilding_enrichment(
"pd": period,
"pps": point["price_per_sqm"],
},
)
chart_saved += 1
).fetchone()
if written is not None and written[0]:
price_inserted += 1
else:
price_updated += 1
# 4. INSERT house_reliability_checks (stores overall check + details array)
# Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at)
# No UNIQUE constraint — caller should manage duplicates if needed
reliability_inserted = 0
for check in enrichment.reliability_checks:
if not check.get("check_name") and not check.get("check_status"):
continue
@ -743,15 +782,23 @@ def save_newbuilding_enrichment(
"det": json.dumps(check.get("details") or [], ensure_ascii=False),
},
)
reliability_inserted += 1
db.commit()
logger.info(
"Cian newbuilding saved house_id=%s (chart=%d points, reliability=%d checks, mc_id=%s)",
"Cian newbuilding saved house_id=%s (chart: +%d new / %d updated, "
"reliability=%d checks, mc_id=%s)",
house_id,
chart_saved,
len(enrichment.reliability_checks),
price_inserted,
price_updated,
reliability_inserted,
mc_id,
)
return NewbuildingSaveCounts(
price_inserted=price_inserted,
price_updated=price_updated,
reliability_inserted=reliability_inserted,
)
async def resolve_cian_zhk_url(