feat(tradein/cian): batch backfill task для offer_price_history + houses_price_dynamics #535

Merged

lekss361 merged 2 commits from feat/cian-history-backfill into main

2026-05-24 14:55:48 +00:00

lekss361 commented

2026-05-24 14:44:55 +00:00

Owner

What

Batch backfill для уже собранных Cian listings — iterates DB, fetches Cian detail pages for listings WITHOUT offer_price_history rows, persists historical price journey via existing cian_detail.save_detail_enrichment. Idempotent (skips entities already with history).

Project has NO Celery infra (verified: only app/tasks/refresh_search_matview.py + geocode_missing.py are sync funcs called from admin endpoints). This PR mirrors that pattern — function + admin trigger, no broker bootstrap.

Backend

tradein-mvp/backend/app/tasks/cian_history_backfill.py — NEW. backfill_cian_history(db, batch_size, do_listings, do_houses, dry_run) -> CianBackfillResult. Listings query: LEFT JOIN offer_price_history anti-join. Per-iteration: fetch_detail → save_detail_enrichment → asyncio.sleep(scraper_settings.cian_delay). Counters: total, processed, succeeded, failed_fetch, failed_save, price_changes_attempted.
tradein-mvp/backend/app/api/v1/admin.py — NEW endpoint POST /scrape/cian-backfill-history?batch_size=...&do_listings=...&do_houses=...&dry_run=.... batch_size capped 1≤N≤200 via Query(ge=1, le=200) (prevents proxy timeouts on >200×5s runs).

Houses block

Stubbed as TODO. Houses lack a cian_zhk_url column (verified: 020_houses_alter_cian.sql only adds cian_internal_house_id bigint, no URL). To activate houses backfill: future migration 065_houses_add_cian_url.sql to add column, then populate during SERP scrape. Comment in code points to this.

Why now

Predecessors:

PR #523 — UNIQUE(listing_id, change_time) so ON CONFLICT DO NOTHING dedups correctly during backfill.
PR #525 — endpoints + UI for one-shot detail fills (this PR is the batch version).
PR #530 — UI surface for Cian chart (this PR feeds the underlying data store offer_price_history).

Test plan

ruff clean (verified)
Pre-push review ✅ APPROVE with minor cosmetic notes (applied: rename counter + batch_size cap)
Manual smoke: curl -X POST .../admin/scrape/cian-backfill-history?batch_size=10&dry_run=true → expect listings_total > 0, listings_processed = 0 (dry-run).
Real run: ?batch_size=10&dry_run=false → expect price_changes_attempted > 0 + rows in offer_price_history for those listing_ids.
Verify ON CONFLICT dedup: re-run same batch → succeeded count high, price_changes_attempted high, but offer_price_history row count unchanged (UNIQUE from PR #523 047).

## What Batch backfill для уже собранных Cian listings — iterates DB, fetches Cian detail pages for listings WITHOUT `offer_price_history` rows, persists historical price journey via existing `cian_detail.save_detail_enrichment`. Idempotent (skips entities already with history). Project has NO Celery infra (verified: only `app/tasks/refresh_search_matview.py` + `geocode_missing.py` are sync funcs called from admin endpoints). This PR mirrors that pattern — function + admin trigger, no broker bootstrap. ## Backend - `tradein-mvp/backend/app/tasks/cian_history_backfill.py` — NEW. `backfill_cian_history(db, batch_size, do_listings, do_houses, dry_run) -> CianBackfillResult`. Listings query: `LEFT JOIN offer_price_history` anti-join. Per-iteration: fetch_detail → save_detail_enrichment → asyncio.sleep(scraper_settings.cian_delay). Counters: total, processed, succeeded, failed_fetch, failed_save, price_changes_attempted. - `tradein-mvp/backend/app/api/v1/admin.py` — NEW endpoint `POST /scrape/cian-backfill-history?batch_size=...&do_listings=...&do_houses=...&dry_run=...`. `batch_size` capped 1≤N≤200 via `Query(ge=1, le=200)` (prevents proxy timeouts on >200×5s runs). ## Houses block **Stubbed as TODO.** Houses lack a `cian_zhk_url` column (verified: `020_houses_alter_cian.sql` only adds `cian_internal_house_id bigint`, no URL). To activate houses backfill: future migration `065_houses_add_cian_url.sql` to add column, then populate during SERP scrape. Comment in code points to this. ## Why now Predecessors: - PR #523 — UNIQUE(listing_id, change_time) so ON CONFLICT DO NOTHING dedups correctly during backfill. - PR #525 — endpoints + UI for one-shot detail fills (this PR is the batch version). - PR #530 — UI surface for Cian chart (this PR feeds the underlying data store offer_price_history). ## Test plan - [x] ruff clean (verified) - [x] Pre-push review ✅ APPROVE with minor cosmetic notes (applied: rename counter + batch_size cap) - [ ] Manual smoke: `curl -X POST .../admin/scrape/cian-backfill-history?batch_size=10&dry_run=true` → expect listings_total > 0, listings_processed = 0 (dry-run). - [ ] Real run: `?batch_size=10&dry_run=false` → expect price_changes_attempted > 0 + rows in offer_price_history for those listing_ids. - [ ] Verify ON CONFLICT dedup: re-run same batch → succeeded count high, price_changes_attempted high, but offer_price_history row count unchanged (UNIQUE from PR #523 047).

lekss361 added 2 commits 2026-05-24 14:44:56 +00:00

feat(tradein/cian): batch backfill task для offer_price_history + houses_price_dynamics 6851c55090

fix(tradein/cian): rename counter to price_changes_attempted + cap batch_size <=200 e77bc9fe06

lekss361 commented

2026-05-24 14:55:40 +00:00

Author

Owner

Deep Code Review — PR #535

Verdict: APPROVE
SHA verified: e77bc9f (head matches prompt)
Files: 2 (admin.py +48, cian_history_backfill.py +171 new) · 219 LoC

Summary

Batch backfill для offer_price_history: iterates listings WHERE source='cian' anti-joined against offer_price_history, per-listing fetch→save, idempotent through ON CONFLICT DO NOTHING (relies on offer_price_history_listing_change_uq from migration 047 — verified present in main). Houses block explicitly stubbed (no cian_zhk_url column yet). No Celery — admin-triggered sync, matches existing geocode_missing.py / refresh_search_matview.py pattern.

Critical / High

None.

Medium

tradein-mvp/backend/app/api/v1/admin.py:1037-1067 — no background=true option. At batch_size=200 × cian_delay≈5s = ~17min, single HTTP request will exceed typical Caddy/gunicorn timeouts. Peer POST /scrape/geocode-missing-listings (admin.py:370) accepts background: bool + uses BackgroundTasks + own SessionLocal() for long batches. Recommend adopting same shape — non-blocking since idempotent re-runs are safe.

Low

tradein-mvp/backend/app/tasks/cian_history_backfill.py:131 — enrichment.price_changes or [] is defensive but redundant (field(default_factory=list) guarantees list). Keep or remove, no functional impact.
price_changes_attempted counter (line 131) counts everything fetch_detail returned, not what save_detail_enrichment actually INSERTed past the if not change.get('change_time') or not change.get('price_rub'): continue filter and ON CONFLICT. Name says "attempted" so OK, but logged metric ≠ row-count delta.
Concurrent invocations: two simultaneous ?batch_size=N calls could pick overlapping rows (anti-join not under lock) → wasted HTTP fetches. ON CONFLICT saves DB integrity. Not in scope to fix, but consider an advisory_lock guard in future iteration.

Positives

Idempotent: anti-join LEFT JOIN offer_price_history oph ON oph.listing_id = l.id WHERE oph.listing_id IS NULL + ON CONFLICT against verified (listing_id, change_time) UNIQUE (migration 047). Re-runs only process new listings.
Resumable: save_detail_enrichment does its own db.commit() per listing (verified at cian_detail.py:save_detail_enrichment final commit). Interrupt loses at most one in-flight row.
Error isolation: separate try/except for fetch and save, sleep honored on failure paths — rate limit к Cian не нарушен при ошибках.
Rate limit: pulls from runtime-tunable scraper_settings.get_scraper_delay('cian') (default 5.0s) — admin can throttle without redeploy.
Houses TODO honest: doesn't pretend to work; clear path documented (need migration 063_houses_add_cian_url.sql first). Endpoint accepts do_houses=True to log skip — not a silent no-op trap.
psycopg v3 conventions OK: no :param IS NOT NULL without CAST (this PR uses oph.listing_id IS NULL against a column, not a parameter, so no AmbiguousParameter risk — per memory rule about CAST in IS NOT NULL predicates).
Dependency contract verified: fetch_detail source confirms it returns None on HTTP≠200 or parse failure, and the loop explicitly handles None (lines 121-128) — not relying on raised exceptions like some service-layer functions in this repo silently defaulting.
Pattern parity: shape matches geocode_missing.py (dataclass result, structured logger.info summary, per-domain counters).

Cross-PR context

PR #523 (047 UNIQUE on offer_price_history) — merged, verified — sentence of idempotency: ✅
PR #525 (one-shot detail admin endpoints) — merged — batch version is logical follow-up: ✅
PR #530 (chart/rent surface) — merged — this PR feeds the data store the chart reads: ✅
PR #533 (Yandex sweep) / #534 (Avito Phase C) — not adjacent (different sources, different schemas); no shared files, no overlap.

Test plan

Manual smoke flow in PR body is reasonable. Recommend post-merge:

POST .../admin/scrape/cian-backfill-history?batch_size=5&dry_run=true → expect listings_total reflects pending count, processed=0.
POST .../admin/scrape/cian-backfill-history?batch_size=5&dry_run=false → verify rows appear in offer_price_history WHERE listing_id IN (...).
Re-run #2 → succeeded stays high, price_changes_attempted high, but SELECT COUNT(*) FROM offer_price_history unchanged (UNIQUE dedup).
Watch for any cian_detail save failed warnings in logs (would indicate save-path edge cases).

Approving for merge (any-scope policy).

## Deep Code Review — PR #535  **Verdict:** APPROVE **SHA verified:** e77bc9f (head matches prompt) **Files:** 2 (admin.py +48, cian_history_backfill.py +171 new) · 219 LoC ### Summary Batch backfill для `offer_price_history`: iterates `listings WHERE source='cian'` anti-joined against `offer_price_history`, per-listing fetch→save, idempotent through `ON CONFLICT DO NOTHING` (relies on `offer_price_history_listing_change_uq` from migration 047 — verified present in main). Houses block explicitly stubbed (no `cian_zhk_url` column yet). No Celery — admin-triggered sync, matches existing `geocode_missing.py` / `refresh_search_matview.py` pattern. ### Critical / High None. ### Medium - **`tradein-mvp/backend/app/api/v1/admin.py:1037-1067`** — no `background=true` option. At `batch_size=200 × cian_delay≈5s = ~17min`, single HTTP request will exceed typical Caddy/gunicorn timeouts. Peer `POST /scrape/geocode-missing-listings` (admin.py:370) accepts `background: bool` + uses `BackgroundTasks` + own `SessionLocal()` for long batches. Recommend adopting same shape — non-blocking since idempotent re-runs are safe. ### Low - `tradein-mvp/backend/app/tasks/cian_history_backfill.py:131` — `enrichment.price_changes or []` is defensive but redundant (`field(default_factory=list)` guarantees list). Keep or remove, no functional impact. - `price_changes_attempted` counter (line 131) counts everything `fetch_detail` returned, not what `save_detail_enrichment` actually INSERTed past the `if not change.get('change_time') or not change.get('price_rub'): continue` filter and ON CONFLICT. Name says "attempted" so OK, but logged metric ≠ row-count delta. - Concurrent invocations: two simultaneous `?batch_size=N` calls could pick overlapping rows (anti-join not under lock) → wasted HTTP fetches. ON CONFLICT saves DB integrity. Not in scope to fix, but consider an advisory_lock guard in future iteration. ### Positives - **Idempotent**: anti-join `LEFT JOIN offer_price_history oph ON oph.listing_id = l.id WHERE oph.listing_id IS NULL` + ON CONFLICT against verified `(listing_id, change_time)` UNIQUE (migration 047). Re-runs only process new listings. - **Resumable**: `save_detail_enrichment` does its own `db.commit()` per listing (verified at `cian_detail.py:save_detail_enrichment` final commit). Interrupt loses at most one in-flight row. - **Error isolation**: separate try/except for fetch and save, sleep honored on failure paths — rate limit к Cian не нарушен при ошибках. - **Rate limit**: pulls from runtime-tunable `scraper_settings.get_scraper_delay('cian')` (default 5.0s) — admin can throttle without redeploy. - **Houses TODO honest**: doesn't pretend to work; clear path documented (need migration 063_houses_add_cian_url.sql first). Endpoint accepts `do_houses=True` to log skip — not a silent no-op trap. - **psycopg v3 conventions OK**: no `:param IS NOT NULL` without CAST (this PR uses `oph.listing_id IS NULL` against a column, not a parameter, so no AmbiguousParameter risk — per memory rule about CAST in IS NOT NULL predicates). - **Dependency contract verified**: `fetch_detail` source confirms it returns `None` on HTTP≠200 or parse failure, and the loop explicitly handles `None` (lines 121-128) — not relying on raised exceptions like some service-layer functions in this repo silently defaulting. - **Pattern parity**: shape matches `geocode_missing.py` (dataclass result, structured logger.info summary, per-domain counters). ### Cross-PR context - PR #523 (047 UNIQUE on offer_price_history) — merged, verified — sentence of idempotency: ✅ - PR #525 (one-shot detail admin endpoints) — merged — batch version is logical follow-up: ✅ - PR #530 (chart/rent surface) — merged — this PR feeds the data store the chart reads: ✅ - PR #533 (Yandex sweep) / #534 (Avito Phase C) — not adjacent (different sources, different schemas); no shared files, no overlap. ### Test plan Manual smoke flow in PR body is reasonable. Recommend post-merge: 1. `POST .../admin/scrape/cian-backfill-history?batch_size=5&dry_run=true` → expect listings_total reflects pending count, processed=0. 2. `POST .../admin/scrape/cian-backfill-history?batch_size=5&dry_run=false` → verify rows appear in `offer_price_history WHERE listing_id IN (...)`. 3. Re-run #2 → succeeded stays high, price_changes_attempted high, but `SELECT COUNT(*) FROM offer_price_history` unchanged (UNIQUE dedup). 4. Watch for any `cian_detail save failed` warnings in logs (would indicate save-path edge cases). Approving for merge (any-scope policy).