fix(tradein/newbuilding): счётчики записи различают вставку и обновление (#2807) #2809
6 changed files with 322 additions and 53 deletions
|
|
@ -1961,8 +1961,9 @@ async def scrape_cian_newbuilding(
|
||||||
|
|
||||||
saved = False
|
saved = False
|
||||||
if house_id is not None:
|
if house_id is not None:
|
||||||
# save_newbuilding_enrichment — sync (def, returns None); await на sync-функции
|
# save_newbuilding_enrichment — sync (def, не корутина); await на sync-функции
|
||||||
# раньше поднимал TypeError на любом вызове с house_id.
|
# раньше поднимал TypeError на любом вызове с house_id. Возвращаемый счёт
|
||||||
|
# записанного (#2807) этой ручке не нужен — она отвечает фактом сохранения.
|
||||||
save_newbuilding_enrichment(db, house_id, enrichment)
|
save_newbuilding_enrichment(db, house_id, enrichment)
|
||||||
saved = True
|
saved = True
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -109,10 +109,17 @@ class NewbuildingEnrichBackfillResult:
|
||||||
failed_fetch: int = 0 # fetch returned None / raised
|
failed_fetch: int = 0 # fetch returned None / raised
|
||||||
failed_save: int = 0 # save raised after a good fetch
|
failed_save: int = 0 # save raised after a good fetch
|
||||||
|
|
||||||
# Row-level deltas (how much actually landed).
|
# Сколько РЕАЛЬНО записано, по словам самих писателей (#2807). Раньше здесь стоял
|
||||||
price_dynamics_rows: int = 0
|
# прирост COUNT(*) по таблице до/после сохранения — то есть «выросла ли таблица», а
|
||||||
reliability_rows: int = 0
|
# не «сколько записали»: при ON CONFLICT DO UPDATE обновление даёт ноль, а у
|
||||||
review_rows: int = 0
|
# 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)
|
duration_sec: float = field(default=0.0)
|
||||||
|
|
||||||
|
|
@ -557,15 +564,15 @@ async def backfill_newbuilding_enrichment(
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# ── Save under a SAVEPOINT so one bad house can't poison the batch ──
|
# ── Save under a SAVEPOINT so one bad house can't poison the batch ──
|
||||||
# begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally,
|
# begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally.
|
||||||
# so we snapshot the row counts BEFORE and recompute the delta AFTER its commit
|
# COUNT(*) до сохранения нужен ТОЛЬКО для had_reliability (дедуп ниже): сколько
|
||||||
# rather than relying on the nested transaction staying open.
|
# записано, теперь сообщают сами писатели, а не разница COUNT'ов (#2807).
|
||||||
pd_before, rc_before, rv_before = _house_enrichment_counts(db, house_id)
|
_, rc_before, _ = _house_enrichment_counts(db, house_id)
|
||||||
try:
|
try:
|
||||||
had_reliability = rc_before > 0
|
had_reliability = rc_before > 0
|
||||||
|
|
||||||
# 1) price_dynamics + reliability + houses UPDATE (existing, commits inside).
|
# 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).
|
# 2) reviews — added here (save_newbuilding_enrichment skips them).
|
||||||
# SAVEPOINT around the review write so a malformed review can't lose the
|
# SAVEPOINT around the review write so a malformed review can't lose the
|
||||||
|
|
@ -601,16 +608,18 @@ async def backfill_newbuilding_enrichment(
|
||||||
sp.rollback()
|
sp.rollback()
|
||||||
logger.warning("reliability dedup failed house_id=%s: %s", house_id, dexc)
|
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_inserted += saved.price_inserted
|
||||||
result.price_dynamics_rows += max(0, pd_after - pd_before)
|
result.price_dynamics_updated += saved.price_updated
|
||||||
result.reliability_rows += max(0, rc_after - rc_before)
|
result.reliability_inserted += saved.reliability_inserted
|
||||||
result.review_rows += max(0, rv_after - rv_before)
|
result.review_upserted += review_written
|
||||||
result.succeeded += 1
|
result.succeeded += 1
|
||||||
logger.info(
|
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,
|
house_id,
|
||||||
max(0, pd_after - pd_before),
|
saved.price_inserted,
|
||||||
max(0, rc_after - rc_before),
|
saved.price_updated,
|
||||||
|
saved.reliability_inserted,
|
||||||
review_written,
|
review_written,
|
||||||
len(enrichment.reviews),
|
len(enrichment.reviews),
|
||||||
)
|
)
|
||||||
|
|
@ -629,8 +638,8 @@ async def backfill_newbuilding_enrichment(
|
||||||
result.duration_sec = time.time() - t0
|
result.duration_sec = time.time() - t0
|
||||||
logger.info(
|
logger.info(
|
||||||
"newbuilding-enrich backfill done: processed=%d ok=%d skip=%d resolved=%d "
|
"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 "
|
"resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
|
||||||
"| %.1fs",
|
"%d обновлено, reliability +%d, отзывов %d | %.1fs",
|
||||||
result.processed,
|
result.processed,
|
||||||
result.succeeded,
|
result.succeeded,
|
||||||
result.skipped_already_enriched,
|
result.skipped_already_enriched,
|
||||||
|
|
@ -638,9 +647,10 @@ async def backfill_newbuilding_enrichment(
|
||||||
result.failed_resolve,
|
result.failed_resolve,
|
||||||
result.failed_fetch,
|
result.failed_fetch,
|
||||||
result.failed_save,
|
result.failed_save,
|
||||||
result.price_dynamics_rows,
|
result.price_dynamics_inserted,
|
||||||
result.reliability_rows,
|
result.price_dynamics_updated,
|
||||||
result.review_rows,
|
result.reliability_inserted,
|
||||||
|
result.review_upserted,
|
||||||
result.duration_sec,
|
result.duration_sec,
|
||||||
)
|
)
|
||||||
return result
|
return result
|
||||||
|
|
@ -787,8 +797,8 @@ async def run_newbuilding_enrich(
|
||||||
)
|
)
|
||||||
logger.info(
|
logger.info(
|
||||||
"scheduler: newbuilding_enrich run_id=%d finished — processed=%d ok=%d skip=%d "
|
"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 "
|
"resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
|
||||||
"| pending=%d %.1fs",
|
"%d обновлено, reliability +%d, отзывов %d | pending=%d %.1fs",
|
||||||
run_id,
|
run_id,
|
||||||
result.processed,
|
result.processed,
|
||||||
result.succeeded,
|
result.succeeded,
|
||||||
|
|
@ -796,9 +806,10 @@ async def run_newbuilding_enrich(
|
||||||
result.failed_resolve,
|
result.failed_resolve,
|
||||||
result.failed_fetch,
|
result.failed_fetch,
|
||||||
result.failed_save,
|
result.failed_save,
|
||||||
result.price_dynamics_rows,
|
result.price_dynamics_inserted,
|
||||||
result.reliability_rows,
|
result.price_dynamics_updated,
|
||||||
result.review_rows,
|
result.reliability_inserted,
|
||||||
|
result.review_upserted,
|
||||||
result.cian_houses_pending,
|
result.cian_houses_pending,
|
||||||
result.duration_sec,
|
result.duration_sec,
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -19,7 +19,10 @@ _wp_mock = MagicMock()
|
||||||
sys.modules.setdefault("weasyprint", _wp_mock)
|
sys.modules.setdefault("weasyprint", _wp_mock)
|
||||||
|
|
||||||
import pytest # noqa: E402
|
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
|
from app.tasks.newbuilding_enrich_backfill import ( # noqa: E402
|
||||||
NewbuildingEnrichBackfillResult,
|
NewbuildingEnrichBackfillResult,
|
||||||
|
|
@ -168,12 +171,16 @@ def _enrichment_with_everything(seed: int = 0) -> NewbuildingEnrichment:
|
||||||
|
|
||||||
|
|
||||||
def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
|
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:
|
for p in enrichment.realty_valuation_chart:
|
||||||
if p.get("price_per_sqm") is None:
|
if p.get("price_per_sqm") is None:
|
||||||
continue
|
continue
|
||||||
db.price_dynamics.add(
|
key = (
|
||||||
(
|
|
||||||
house_id,
|
house_id,
|
||||||
p["month_date"],
|
p["month_date"],
|
||||||
"cian_realty_valuation",
|
"cian_realty_valuation",
|
||||||
|
|
@ -181,11 +188,20 @@ def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
|
||||||
p.get("prices_type", "price"),
|
p.get("prices_type", "price"),
|
||||||
p.get("period", "halfYear"),
|
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:
|
for c in enrichment.reliability_checks:
|
||||||
if c.get("check_name") or c.get("check_status"):
|
if c.get("check_name") or c.get("check_status"):
|
||||||
db.reliability.append((house_id, "cian_nashdom"))
|
db.reliability.append((house_id, "cian_nashdom"))
|
||||||
|
reliability += 1
|
||||||
db.commit()
|
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.price_dynamics) == 2 # 1 chart point × 2 houses
|
||||||
assert len(db.reliability) == 2
|
assert len(db.reliability) == 2
|
||||||
assert len(db.reviews) == 4 # 2 reviews × 2 houses
|
assert len(db.reviews) == 4 # 2 reviews × 2 houses
|
||||||
assert result.price_dynamics_rows == 2
|
assert result.price_dynamics_inserted == 2
|
||||||
assert result.reliability_rows == 2
|
assert result.price_dynamics_updated == 0
|
||||||
assert result.review_rows == 4
|
assert result.reliability_inserted == 2
|
||||||
|
assert result.review_upserted == 4
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@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):
|
async def _fake_backfill(_db, *, limit, force, request_delay_sec, on_progress=None):
|
||||||
# on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
|
# on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
|
||||||
seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
|
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, "backfill_newbuilding_enrichment", _fake_backfill)
|
||||||
monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
|
monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
|
||||||
|
|
|
||||||
|
|
@ -185,7 +185,7 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
|
||||||
_stub_backfill(
|
_stub_backfill(
|
||||||
monkeypatch,
|
monkeypatch,
|
||||||
NewbuildingEnrichBackfillResult(
|
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)
|
calls = _stub_finalisers(monkeypatch)
|
||||||
|
|
@ -196,5 +196,5 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
|
||||||
assert counters["attempted"] == 10
|
assert counters["attempted"] == 10
|
||||||
assert counters["enriched"] == 3
|
assert counters["enriched"] == 3
|
||||||
assert counters["failed"] == 7
|
assert counters["failed"] == 7
|
||||||
assert counters["price_dynamics_rows"] == 7 # исходные счётчики на месте
|
assert counters["price_dynamics_inserted"] == 7 # исходные счётчики на месте
|
||||||
assert counters["succeeded"] == 3
|
assert counters["succeeded"] == 3
|
||||||
|
|
|
||||||
193
tradein-mvp/backend/tests/test_2807_write_counters_honesty.py
Normal file
193
tradein-mvp/backend/tests/test_2807_write_counters_honesty.py
Normal 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
|
||||||
|
|
@ -594,11 +594,38 @@ def _extract_nested_offers(offers_state: dict[str, Any]) -> list[dict[str, Any]]
|
||||||
# ---- save helpers ----
|
# ---- 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(
|
def save_newbuilding_enrichment(
|
||||||
db: Any,
|
db: Any,
|
||||||
house_id: int,
|
house_id: int,
|
||||||
enrichment: NewbuildingEnrichment,
|
enrichment: NewbuildingEnrichment,
|
||||||
) -> None:
|
) -> NewbuildingSaveCounts:
|
||||||
"""Persist NewbuildingEnrichment to DB.
|
"""Persist NewbuildingEnrichment to DB.
|
||||||
|
|
||||||
Steps:
|
Steps:
|
||||||
|
|
@ -606,6 +633,9 @@ def save_newbuilding_enrichment(
|
||||||
2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present)
|
2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present)
|
||||||
3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE)
|
3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE)
|
||||||
4. INSERT INTO house_reliability_checks (overall + details)
|
4. INSERT INTO house_reliability_checks (overall + details)
|
||||||
|
|
||||||
|
Returns NewbuildingSaveCounts — вставлено/обновлено раздельно (#2807). Вызывающие,
|
||||||
|
которым счёт не нужен (SERP-sweep, admin re-enrich), просто игнорируют результат.
|
||||||
"""
|
"""
|
||||||
from sqlalchemy import text
|
from sqlalchemy import text
|
||||||
|
|
||||||
|
|
@ -681,14 +711,18 @@ def save_newbuilding_enrichment(
|
||||||
# 3. INSERT houses_price_dynamics
|
# 3. INSERT houses_price_dynamics
|
||||||
# UNIQUE constraint: houses_price_dynamics_dim_key
|
# UNIQUE constraint: houses_price_dynamics_dim_key
|
||||||
# (house_id, source, room_count, prices_type, period, month_date) — per migration 029
|
# (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:
|
for point in enrichment.realty_valuation_chart:
|
||||||
if point.get("price_per_sqm") is None:
|
if point.get("price_per_sqm") is None:
|
||||||
continue
|
continue
|
||||||
room_count = point.get("room_count") or "all"
|
room_count = point.get("room_count") or "all"
|
||||||
prices_type = point.get("prices_type") or "price"
|
prices_type = point.get("prices_type") or "price"
|
||||||
period = point.get("period") or "halfYear"
|
period = point.get("period") or "halfYear"
|
||||||
db.execute(
|
# RETURNING (xmax = 0): у только что вставленной строки xmax равен нулю, у
|
||||||
|
# обновлённой конфликтом — id транзакции. Без этого «вставили» и «обновили»
|
||||||
|
# снаружи неразличимы, и обновление читается как «ничего не записали» (#2807).
|
||||||
|
written = db.execute(
|
||||||
text("""
|
text("""
|
||||||
INSERT INTO houses_price_dynamics (
|
INSERT INTO houses_price_dynamics (
|
||||||
house_id, month_date, source,
|
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
|
ON CONFLICT ON CONSTRAINT houses_price_dynamics_dim_key DO UPDATE SET
|
||||||
price_per_sqm = EXCLUDED.price_per_sqm,
|
price_per_sqm = EXCLUDED.price_per_sqm,
|
||||||
recorded_at = NOW()
|
recorded_at = NOW()
|
||||||
|
RETURNING (xmax = 0) AS is_insert
|
||||||
"""),
|
"""),
|
||||||
{
|
{
|
||||||
"hid": house_id,
|
"hid": house_id,
|
||||||
|
|
@ -714,12 +749,16 @@ def save_newbuilding_enrichment(
|
||||||
"pd": period,
|
"pd": period,
|
||||||
"pps": point["price_per_sqm"],
|
"pps": point["price_per_sqm"],
|
||||||
},
|
},
|
||||||
)
|
).fetchone()
|
||||||
chart_saved += 1
|
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)
|
# 4. INSERT house_reliability_checks (stores overall check + details array)
|
||||||
# Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at)
|
# Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at)
|
||||||
# No UNIQUE constraint — caller should manage duplicates if needed
|
# No UNIQUE constraint — caller should manage duplicates if needed
|
||||||
|
reliability_inserted = 0
|
||||||
for check in enrichment.reliability_checks:
|
for check in enrichment.reliability_checks:
|
||||||
if not check.get("check_name") and not check.get("check_status"):
|
if not check.get("check_name") and not check.get("check_status"):
|
||||||
continue
|
continue
|
||||||
|
|
@ -743,15 +782,23 @@ def save_newbuilding_enrichment(
|
||||||
"det": json.dumps(check.get("details") or [], ensure_ascii=False),
|
"det": json.dumps(check.get("details") or [], ensure_ascii=False),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
reliability_inserted += 1
|
||||||
|
|
||||||
db.commit()
|
db.commit()
|
||||||
logger.info(
|
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,
|
house_id,
|
||||||
chart_saved,
|
price_inserted,
|
||||||
len(enrichment.reliability_checks),
|
price_updated,
|
||||||
|
reliability_inserted,
|
||||||
mc_id,
|
mc_id,
|
||||||
)
|
)
|
||||||
|
return NewbuildingSaveCounts(
|
||||||
|
price_inserted=price_inserted,
|
||||||
|
price_updated=price_updated,
|
||||||
|
reliability_inserted=reliability_inserted,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
async def resolve_cian_zhk_url(
|
async def resolve_cian_zhk_url(
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue