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
2 changed files with 219 additions and 0 deletions
Showing only changes of commit 6851c55090 - Show all commits

View file

@ -1020,3 +1020,51 @@ async def scrape_cian_newbuilding(
saved=saved,
house_id=house_id,
)
class CianBackfillResp(BaseModel):
ok: bool
listings_total: int
listings_processed: int
listings_succeeded: int
listings_failed_fetch: int
listings_failed_save: int
price_changes_inserted: int
houses_total: int
houses_processed: int
houses_succeeded: int
houses_failed_fetch: int
houses_failed_save: int
duration_sec: float
@router.post("/scrape/cian-backfill-history", response_model=CianBackfillResp)
async def scrape_cian_backfill_history(
batch_size: int = 50,
do_listings: bool = True,
do_houses: bool = True,
dry_run: bool = False,
db: Session = Depends(get_db), # noqa: B008
) -> CianBackfillResp:
"""Batch backfill для Cian historical data.
Iterates Cian listings with missing offer_price_history, fetches Cian detail
pages, persists enrichment (price changes, views, agent, ceiling_height, etc.).
Houses block (houses_price_dynamics) is currently a no-op requires
a cian_zhk_url column on houses (migration pending, see TODO in
app/tasks/cian_history_backfill.py).
Rate-limit: scraper_settings 'cian' delay (default 5s) between requests.
Idempotent: skips listings that already have offer_price_history rows.
"""
from app.tasks.cian_history_backfill import CianBackfillResult, backfill_cian_history
result: CianBackfillResult = await backfill_cian_history(
db,
batch_size=batch_size,
do_listings=do_listings,
do_houses=do_houses,
dry_run=dry_run,
)
return CianBackfillResp(ok=True, **result.__dict__)

View file

@ -0,0 +1,171 @@
"""Batch backfill для Cian historical data.
Запускается:
- Manual через POST /admin/scrape/cian-backfill-history
- Future: in-app scheduler (после bootstrap'а Celery beat)
Что делает:
1. SELECT listings WHERE source='cian' AND source_url IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM offer_price_history WHERE listing_id=...)
LIMIT batch_size
fetch_detail + save_detail_enrichment per listing
2. Houses block TODO (PR #525): houses не имеют cian_url / cian_zhk_url колонки.
Колонка cian_internal_house_id (020_houses_alter_cian.sql) хранит numeric Cian ID,
но не ZHK URL, необходимый для fetch_newbuilding().
Backfill ЖК-динамики требует отдельной миграции, добавляющей cian_zhk_url к houses.
До тех пор do_houses=True no-op (houses_total=0).
Rate limit: scraper_settings.get_scraper_delay('cian') between requests.
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass, field
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment
logger = logging.getLogger(__name__)
@dataclass
class CianBackfillResult:
listings_total: int = 0
listings_processed: int = 0
listings_succeeded: int = 0
listings_failed_fetch: int = 0
listings_failed_save: int = 0
price_changes_inserted: int = 0
houses_total: int = 0
houses_processed: int = 0
houses_succeeded: int = 0
houses_failed_fetch: int = 0
houses_failed_save: int = 0
duration_sec: float = field(default=0.0)
async def backfill_cian_history(
db: Session,
*,
batch_size: int = 50,
do_listings: bool = True,
do_houses: bool = True,
dry_run: bool = False,
) -> CianBackfillResult:
"""Iterate Cian listings + houses with missing history, fetch+save.
Args:
db: SQLAlchemy session (caller-owned; each entity commits internally via
save_detail_enrichment / save_newbuilding_enrichment).
batch_size: max entities to process per call (listings + houses counted separately).
do_listings: process listings batch (offer_price_history backfill).
do_houses: reserved for future houses_price_dynamics backfill (currently no-op
see module docstring TODO).
dry_run: skip all fetch+save; only count and log pending rows.
Returns:
CianBackfillResult with per-domain counters + total wall-clock duration.
"""
result = CianBackfillResult()
t0 = time.time()
delay = get_scraper_delay("cian") # seconds; default 5.0
# ── 1. Listings: missing offer_price_history ──────────────────────────────
if do_listings:
rows = (
db.execute(
text("""
SELECT l.id, l.source_url
FROM listings l
LEFT JOIN offer_price_history oph ON oph.listing_id = l.id
WHERE l.source = 'cian'
AND l.source_url IS NOT NULL
AND oph.listing_id IS NULL
LIMIT :lim
"""),
{"lim": batch_size},
)
.mappings()
.all()
)
result.listings_total = len(rows)
if dry_run:
logger.info("dry_run: would process %d cian listings", result.listings_total)
else:
for row in rows:
listing_id: int = row["id"]
source_url: str = row["source_url"]
result.listings_processed += 1
enrichment = None
try:
enrichment = await fetch_detail(source_url)
except Exception as exc:
logger.warning(
"cian_detail fetch failed for listing_id=%s url=%s: %s",
listing_id,
source_url,
exc,
)
result.listings_failed_fetch += 1
await asyncio.sleep(delay)
continue
if enrichment is None:
logger.warning(
"cian_detail fetch returned None for listing_id=%s url=%s",
listing_id,
source_url,
)
result.listings_failed_fetch += 1
await asyncio.sleep(delay)
continue
try:
save_detail_enrichment(db, listing_id, enrichment)
result.listings_succeeded += 1
result.price_changes_inserted += len(enrichment.price_changes or [])
except Exception as exc:
logger.warning(
"cian_detail save failed for listing_id=%s: %s", listing_id, exc
)
result.listings_failed_save += 1
await asyncio.sleep(delay)
# ── 2. Houses: missing houses_price_dynamics ──────────────────────────────
# TODO (PR #525): houses table has cian_internal_house_id (bigint) but no
# cian_zhk_url column. fetch_newbuilding() requires a ZHK URL
# (e.g. https://zhk-...-i.cian.ru/). Add migration `063_houses_add_cian_url.sql`
# with `ALTER TABLE houses ADD COLUMN IF NOT EXISTS cian_zhk_url text` and populate
# it during the existing SERP scrape (CianScraper → save_listings → house upsert).
# Until that column exists, do_houses is accepted but silently no-ops.
if do_houses:
logger.info(
"cian_backfill: houses block skipped — cian_zhk_url column not yet added "
"(see TODO in cian_history_backfill.py)"
)
result.duration_sec = time.time() - t0
logger.info(
"cian_backfill done: listings=%d/%d (ok=%d fetch_fail=%d save_fail=%d "
"price_rows=%d) houses=%d/%d (ok=%d fail=%d) %.1fs",
result.listings_processed,
result.listings_total,
result.listings_succeeded,
result.listings_failed_fetch,
result.listings_failed_save,
result.price_changes_inserted,
result.houses_processed,
result.houses_total,
result.houses_succeeded,
result.houses_failed_fetch + result.houses_failed_save,
result.duration_sec,
)
return result