diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py
index e70dd661..c977e568 100644
--- a/tradein-mvp/backend/app/api/v1/admin.py
+++ b/tradein-mvp/backend/app/api/v1/admin.py
@@ -1186,3 +1186,188 @@ async def scrape_cian_backfill_history(
dry_run=dry_run,
)
return CianBackfillResp(ok=True, **result.__dict__)
+
+
+# ── T8a: Cian price-history backfill ─────────────────────────────────────────
+
+
+class CianPriceHistoryRequest(BaseModel):
+ batch_size: int = Field(default=50, ge=1, le=500)
+ listing_id: int | None = Field(
+ default=None,
+ description="Если указан — обработать один конкретный листинг.",
+ )
+
+
+class BackfillCountersResp(BaseModel):
+ ok: bool
+ checked: int
+ saved: int
+ skipped: int
+ errors: int
+ duration_sec: float
+
+
+@router.post("/scrape/cian-price-history", response_model=BackfillCountersResp)
+async def scrape_cian_price_history(
+ payload: CianPriceHistoryRequest,
+ db: Session = Depends(get_db), # noqa: B008
+) -> BackfillCountersResp:
+ """T8a: Backfill offer_price_history для Cian-листингов.
+
+ Fetches Cian detail pages (curl_cffi, без Playwright) для листингов без строк
+ в offer_price_history, парсит priceChanges из _cianConfig defaultState.
+
+ batch_size: сколько листингов обработать за запуск (default 50).
+ listing_id: обработать один конкретный листинг (debug / точечная перезаливка).
+
+ Rate-limit: задержка из scraper_settings['cian'] (default 5s) между страницами.
+ Idempotent: повторный запуск безопасен (ON CONFLICT DO NOTHING).
+ """
+ from app.services.cian_price_history import (
+ CianPriceHistoryResult,
+ backfill_cian_price_history,
+ )
+
+ result: CianPriceHistoryResult = await backfill_cian_price_history(
+ db,
+ batch_size=payload.batch_size,
+ listing_id=payload.listing_id,
+ )
+ return BackfillCountersResp(
+ ok=True,
+ checked=result.checked,
+ saved=result.saved,
+ skipped=result.skipped,
+ errors=result.errors,
+ duration_sec=round(result.duration_sec, 2),
+ )
+
+
+# ── T10: Yandex address backfill ─────────────────────────────────────────────
+
+
+class YandexAddressBackfillRequest(BaseModel):
+ limit: int = Field(default=200, ge=1, le=2000)
+ request_delay_sec: float = Field(default=3.0, ge=1.0, le=15.0)
+
+
+@router.post("/scrape/yandex-address-backfill", response_model=BackfillCountersResp)
+async def scrape_yandex_address_backfill(
+ payload: YandexAddressBackfillRequest,
+ background_tasks: BackgroundTasks,
+ db: Session = Depends(get_db), # noqa: B008
+) -> BackfillCountersResp:
+ """T10: Backfill listings.address для Yandex-листингов без номера дома.
+
+ Yandex SERP-карточки содержат только улицу («улица Горького»). Этот endpoint
+ фетчит detail-страницы через curl_cffi (chrome120), извлекает полный адрес
+ из («Екатеринбург, улица Горького, 36 — id …»), обновляет
+ listings.address и сбрасывает geocode_tried_at для перегеокодирования.
+
+ limit: сколько листингов обработать за запуск (default 200).
+ request_delay_sec: пауза между запросами (default 3.0s).
+
+ Запускает задачу в foreground (не в BackgroundTasks), т.к. batch небольшой.
+ Для > 500 листингов рекомендуется вызывать несколько раз.
+ """
+ from app.services.yandex_address_backfill import (
+ YandexAddressBackfillResult,
+ backfill_yandex_addresses,
+ )
+
+ result: YandexAddressBackfillResult = await backfill_yandex_addresses(
+ db,
+ limit=payload.limit,
+ request_delay_sec=payload.request_delay_sec,
+ )
+ return BackfillCountersResp(
+ ok=True,
+ checked=result.checked,
+ saved=result.saved,
+ skipped=result.skipped,
+ errors=result.errors,
+ duration_sec=round(result.duration_sec, 2),
+ )
+
+
+# ── T7: House IMV bulk backfill ───────────────────────────────────────────────
+
+
+class HouseIMVBackfillRequest(BaseModel):
+ batch_size: int = Field(default=50, ge=1, le=500)
+ request_delay_sec: float = Field(
+ default=5.0,
+ ge=1.0,
+ le=30.0,
+ description="Пауза между Avito IMV вызовами (default 5s — anti-captcha).",
+ )
+ only_status: str = Field(
+ default="pending",
+ description="Обрабатывать дома с этим imv_status. 'transient_error' — retry.",
+ )
+ house_id: int | None = Field(
+ default=None,
+ description="Обработать один конкретный дом (debug).",
+ )
+
+
+class HouseIMVBackfillResp(BaseModel):
+ ok: bool
+ checked: int
+ saved: int
+ skipped: int
+ errors: int
+ duration_sec: float
+ status_counts: dict[str, int]
+
+
+@router.post("/scrape/house-imv-backfill", response_model=HouseIMVBackfillResp)
+async def scrape_house_imv_backfill(
+ payload: HouseIMVBackfillRequest,
+ background_tasks: BackgroundTasks,
+ db: Session = Depends(get_db), # noqa: B008
+) -> HouseIMVBackfillResp:
+ """T7: Avito IMV bulk backfill на уровне домов (house_imv_evaluations).
+
+ Итерирует дома с imv_status='pending' (или 'transient_error'), вычисляет
+ медианные параметры квартиры из связанных листингов, вызывает Avito IMV API,
+ сохраняет результаты в house_imv_evaluations + house_placement_history +
+ house_suggestions.
+
+ Resumable: дома помечаются imv_status='ok'/'not_found'/'error' после обработки.
+ Idempotent: повторный запуск не дублирует данные (UPSERT on house_id).
+
+ batch_size: сколько домов обработать за запуск (default 50).
+ request_delay_sec: пауза между IMV-вызовами (default 5s). ВАЖНО: Avito IMV
+ реагирует на частые запросы с datacenter-IP. Не снижать < 3s.
+ only_status: по умолчанию 'pending'. Для retry failed — 'transient_error'.
+ house_id: обработать один дом (debug).
+
+ Примечание по прокси: Avito IMV использует собственную curl_cffi-сессию.
+ settings.scraper_proxy_url прокидывается в неё автоматически через avito_imv.py
+ (evaluate_via_imv создаёт сессию без явного proxy-параметра — сессия использует
+ env HTTP_PROXY/HTTPS_PROXY если задан. При необходимости — добавить явный
+ proxy-параметр в evaluate_via_imv / backfill_house_imv на уровне сервиса).
+ """
+ from app.services.house_imv_backfill import (
+ HouseIMVBackfillResult,
+ backfill_house_imv,
+ )
+
+ result: HouseIMVBackfillResult = await backfill_house_imv(
+ db,
+ batch_size=payload.batch_size,
+ request_delay_sec=payload.request_delay_sec,
+ only_status=payload.only_status,
+ house_id=payload.house_id,
+ )
+ return HouseIMVBackfillResp(
+ ok=True,
+ checked=result.checked,
+ saved=result.saved,
+ skipped=result.skipped,
+ errors=result.errors,
+ duration_sec=round(result.duration_sec, 2),
+ status_counts=result.status_counts,
+ )
diff --git a/tradein-mvp/backend/app/services/cian_price_history.py b/tradein-mvp/backend/app/services/cian_price_history.py
new file mode 100644
index 00000000..5e1c2baf
--- /dev/null
+++ b/tradein-mvp/backend/app/services/cian_price_history.py
@@ -0,0 +1,153 @@
+"""T8a: Cian offer price-history backfill service.
+
+Fetches Cian detail pages (curl_cffi, без Playwright) for listings that have
+no rows in offer_price_history, extracts priceChanges from
+_cianConfig['frontend-offer-card'] defaultState, writes to offer_price_history.
+
+Deliberately NOT wired into scheduler — rate-limit risk with datacenter IPs.
+Triggered manually via POST /api/v1/admin/scrape/cian-price-history.
+"""
+
+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 CianPriceHistoryResult:
+ checked: int = 0
+ saved: int = 0
+ skipped: int = 0
+ errors: int = 0
+ duration_sec: float = field(default=0.0)
+
+
+async def backfill_cian_price_history(
+ db: Session,
+ *,
+ batch_size: int = 50,
+ listing_id: int | None = None,
+) -> CianPriceHistoryResult:
+ """Fetch Cian detail pages and write missing price-history rows.
+
+ Args:
+ db: SQLAlchemy session (caller-owned; commits internally per listing).
+ batch_size: max listings to process when listing_id is None.
+ listing_id: process a single specific listing (ignores batch_size).
+
+ Selection query picks cian listings with no existing offer_price_history rows.
+ Idempotent: re-run safe via ON CONFLICT DO NOTHING in save_detail_enrichment.
+ """
+ result = CianPriceHistoryResult()
+ t0 = time.time()
+ delay = get_scraper_delay("cian") # default 5.0s
+
+ if listing_id is not None:
+ rows = (
+ db.execute(
+ text("""
+ SELECT l.id, l.source_url
+ FROM listings l
+ WHERE l.id = CAST(:lid AS bigint)
+ AND l.source = 'cian'
+ AND l.source_url IS NOT NULL
+ """),
+ {"lid": listing_id},
+ )
+ .mappings()
+ .all()
+ )
+ else:
+ 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
+ ORDER BY l.id
+ LIMIT :lim
+ """),
+ {"lim": batch_size},
+ )
+ .mappings()
+ .all()
+ )
+
+ result.checked = len(rows)
+ logger.info(
+ "cian_price_history backfill: checked=%d delay=%.1fs",
+ result.checked,
+ delay,
+ )
+
+ for i, row in enumerate(rows):
+ lid: int = row["id"]
+ url: str = row["source_url"]
+
+ try:
+ enrichment = await fetch_detail(url)
+ except Exception as exc:
+ logger.warning(
+ "cian_price_history: fetch failed listing_id=%s url=%s: %s",
+ lid,
+ url,
+ exc,
+ )
+ result.errors += 1
+ await asyncio.sleep(delay)
+ continue
+
+ if enrichment is None:
+ logger.warning(
+ "cian_price_history: fetch returned None listing_id=%s url=%s",
+ lid,
+ url,
+ )
+ result.errors += 1
+ await asyncio.sleep(delay)
+ continue
+
+ if not enrichment.price_changes:
+ logger.debug("cian_price_history: no price_changes listing_id=%s", lid)
+ result.skipped += 1
+ else:
+ try:
+ save_detail_enrichment(db, lid, enrichment)
+ result.saved += len(enrichment.price_changes)
+ except Exception as exc:
+ logger.warning("cian_price_history: save failed listing_id=%s: %s", lid, exc)
+ result.errors += 1
+ try:
+ db.rollback()
+ except Exception:
+ pass
+ await asyncio.sleep(delay)
+ continue
+
+ if i < len(rows) - 1:
+ await asyncio.sleep(delay)
+
+ result.duration_sec = time.time() - t0
+ logger.info(
+ "cian_price_history done: checked=%d saved=%d skipped=%d errors=%d %.1fs",
+ result.checked,
+ result.saved,
+ result.skipped,
+ result.errors,
+ result.duration_sec,
+ )
+ return result
diff --git a/tradein-mvp/backend/app/services/house_imv_backfill.py b/tradein-mvp/backend/app/services/house_imv_backfill.py
new file mode 100644
index 00000000..36e27bbd
--- /dev/null
+++ b/tradein-mvp/backend/app/services/house_imv_backfill.py
@@ -0,0 +1,504 @@
+"""T7: House-level IMV backfill service.
+
+Iterates houses with imv_status='pending' (or 'transient_error') that have
+lat/lon coordinates and at least one linked listing with rooms+area data.
+
+For each house:
+ 1. Pick median lot-params from linked listings (rooms, area_m2, floor,
+ total_floors, house_type).
+ 2. Enrich address with region prefix (prevents Avito geocoder ambiguity).
+ 3. Call avito_imv.evaluate_via_imv().
+ 4. Persist house_imv_evaluations + house_placement_history + house_suggestions.
+ 5. Mark houses.imv_status accordingly (ok/not_found/no_params/error/transient_error).
+
+Resumable: re-running picks up from where the previous run stopped.
+Deliberately NOT wired into scheduler — triggered manually via admin API.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Literal
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services.scrapers.avito_imv import (
+ IMVAddressNotFoundError,
+ IMVAuthError,
+ IMVEvaluation,
+ IMVTransientError,
+ evaluate_via_imv,
+)
+
+logger = logging.getLogger(__name__)
+
+# ── house_type normalisation ─────────────────────────────────────────────────
+
+_HOUSE_TYPE_TO_IMV: dict[str, str] = {
+ "panel": "panel",
+ "brick": "brick",
+ "monolith": "monolithic",
+ "monolithic": "monolithic",
+ "monolith_brick": "monolithic", # Avito API не принимает гибриды
+ "block": "block",
+ "wood": "wood",
+}
+_HOUSE_TYPE_DEFAULT = "panel" # самый распространённый в ЕКБ
+
+
+def _map_house_type(raw: str | None) -> str:
+ if not raw:
+ return _HOUSE_TYPE_DEFAULT
+ return _HOUSE_TYPE_TO_IMV.get(raw.lower().strip(), _HOUSE_TYPE_DEFAULT)
+
+
+# ── Region bbox prefix для Avito geocoder ────────────────────────────────────
+
+_REGION_BBOX: list[tuple[str, float, float, float, float]] = [
+ # (region_name, lat_min, lat_max, lon_min, lon_max)
+ ("Свердловская область, Екатеринбург", 56.5, 57.1, 59.9, 61.0),
+ ("Свердловская область", 56.0, 61.0, 58.0, 65.0),
+ ("Кировская область, Киров", 58.4, 58.8, 49.4, 49.9),
+ ("Ставропольский край, Ставрополь", 44.9, 45.2, 41.7, 42.2),
+ ("Тюменская область, Тюмень", 57.0, 57.3, 65.3, 65.8),
+ ("Челябинская область, Челябинск", 55.0, 55.4, 61.2, 61.7),
+]
+_REGION_DEFAULT = "Свердловская область, Екатеринбург"
+
+
+def _detect_region_prefix(lat: float | None, lon: float | None) -> str:
+ if lat is None or lon is None:
+ return _REGION_DEFAULT
+ for name, lat_min, lat_max, lon_min, lon_max in _REGION_BBOX:
+ if lat_min <= lat <= lat_max and lon_min <= lon <= lon_max:
+ return name
+ return _REGION_DEFAULT
+
+
+def _enrich_address_for_imv(raw: str, lat: float | None, lon: float | None) -> str:
+ """Prepend region prefix if address lacks a region marker."""
+ addr = (raw or "").strip()
+ addr_lower = addr.lower()
+ if any(
+ m in addr_lower for m in ("область,", " обл.,", " обл,", "край,", "респ.,", "республика,")
+ ):
+ return addr
+ prefix = _detect_region_prefix(lat, lon)
+ return f"{prefix}, {addr}"
+
+
+# ── DB helpers ────────────────────────────────────────────────────────────────
+
+
+def pick_lot_params(db: Session, house_id: int) -> dict:
+ """Pick representative lot-params — median values from linked listings."""
+ row = (
+ db.execute(
+ text("""
+ SELECT
+ CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY rooms)
+ AS integer) AS rooms,
+ CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY area_m2)
+ AS numeric(8,2)) AS area_m2,
+ CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY floor)
+ AS integer) AS floor,
+ CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY total_floors)
+ AS integer) AS total_floors,
+ mode() WITHIN GROUP (ORDER BY house_type) AS house_type
+ FROM listings
+ WHERE house_id_fk = :hid
+ AND rooms IS NOT NULL
+ AND area_m2 IS NOT NULL
+ """),
+ {"hid": house_id},
+ )
+ .mappings()
+ .first()
+ )
+
+ if row is None or row["rooms"] is None:
+ return {}
+
+ house = (
+ db.execute(
+ text("SELECT house_type, total_floors FROM houses WHERE id = :hid"),
+ {"hid": house_id},
+ )
+ .mappings()
+ .first()
+ )
+
+ rooms_raw = int(row["rooms"])
+ return {
+ "rooms": max(rooms_raw, 1), # Avito IMV отклоняет rooms=0 (студия)
+ "area_m2": float(row["area_m2"]),
+ "floor": int(row["floor"] or 1),
+ "floor_at_home": int(row["total_floors"] or (house and house["total_floors"]) or 9),
+ "house_type": _map_house_type(row["house_type"] or (house and house["house_type"])),
+ "renovation_type": "cosmetic",
+ "has_balcony": True,
+ "has_loggia": False,
+ }
+
+
+def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluation) -> None:
+ """Persist evaluation to house_imv_evaluations, house_placement_history, house_suggestions."""
+ # 1. House-level evaluation (UPSERT on house_id)
+ db.execute(
+ text("""
+ INSERT INTO house_imv_evaluations (
+ house_id, cache_key, recommended_price, lower_price, higher_price,
+ market_count, raw_response, fetched_at,
+ rooms, area_m2, floor, floor_at_home, house_type,
+ renovation_type, has_balcony, has_loggia
+ ) VALUES (
+ :hid, :ck, :rec, :lo, :hi,
+ :mc, CAST(:raw AS jsonb), NOW(),
+ :rooms, CAST(:area AS numeric), :floor, :fah, :htype,
+ :rtype, :balcony, :loggia
+ )
+ ON CONFLICT (house_id) DO UPDATE SET
+ cache_key = EXCLUDED.cache_key,
+ recommended_price = EXCLUDED.recommended_price,
+ lower_price = EXCLUDED.lower_price,
+ higher_price = EXCLUDED.higher_price,
+ market_count = EXCLUDED.market_count,
+ raw_response = EXCLUDED.raw_response,
+ fetched_at = NOW(),
+ rooms = EXCLUDED.rooms,
+ area_m2 = EXCLUDED.area_m2,
+ floor = EXCLUDED.floor,
+ floor_at_home = EXCLUDED.floor_at_home,
+ house_type = EXCLUDED.house_type,
+ renovation_type = EXCLUDED.renovation_type,
+ has_balcony = EXCLUDED.has_balcony,
+ has_loggia = EXCLUDED.has_loggia
+ """),
+ {
+ "hid": house_id,
+ "ck": result.cache_key,
+ "rec": result.recommended_price,
+ "lo": result.lower_price,
+ "hi": result.higher_price,
+ "mc": result.market_count,
+ "raw": (
+ json.dumps(result.raw_response, ensure_ascii=False) if result.raw_response else None
+ ),
+ "rooms": params["rooms"],
+ "area": params["area_m2"],
+ "floor": params["floor"],
+ "fah": params["floor_at_home"],
+ "htype": params["house_type"],
+ "rtype": params["renovation_type"],
+ "balcony": params["has_balcony"],
+ "loggia": params["has_loggia"],
+ },
+ )
+
+ # 2. Placement history items
+ for item in result.placement_history:
+ db.execute(
+ text("""
+ INSERT INTO house_placement_history (
+ source, house_id, ext_item_id, title, rooms, area_m2,
+ floor, total_floors, start_price, start_price_date,
+ last_price, last_price_date, removed_date, exposure_days,
+ raw_payload, scraped_at
+ ) VALUES (
+ 'avito_imv', :hid, :ext, :title, :rooms, CAST(:area AS numeric),
+ :floor, :total_floors, :sp, :spd, :lp, :lpd, :rmd, :exp,
+ CAST(:raw AS jsonb), NOW()
+ )
+ ON CONFLICT (source, ext_item_id) DO UPDATE SET
+ house_id = EXCLUDED.house_id,
+ title = EXCLUDED.title,
+ rooms = EXCLUDED.rooms,
+ area_m2 = EXCLUDED.area_m2,
+ floor = EXCLUDED.floor,
+ total_floors = EXCLUDED.total_floors,
+ start_price = EXCLUDED.start_price,
+ start_price_date = EXCLUDED.start_price_date,
+ last_price = EXCLUDED.last_price,
+ last_price_date = EXCLUDED.last_price_date,
+ removed_date = EXCLUDED.removed_date,
+ exposure_days = EXCLUDED.exposure_days,
+ raw_payload = EXCLUDED.raw_payload,
+ scraped_at = NOW()
+ """),
+ {
+ "hid": house_id,
+ "ext": item.ext_item_id,
+ "title": item.title,
+ "rooms": item.rooms,
+ "area": item.area_m2,
+ "floor": item.floor,
+ "total_floors": item.total_floors,
+ "sp": item.start_price,
+ "spd": item.start_price_date,
+ "lp": item.last_price,
+ "lpd": item.last_price_date,
+ "rmd": item.removed_date,
+ "exp": item.exposure_days,
+ "raw": (
+ json.dumps(item.raw_payload, ensure_ascii=False) if item.raw_payload else None
+ ),
+ },
+ )
+
+ # 3. Suggestions
+ for sug in result.suggestions:
+ db.execute(
+ text("""
+ INSERT INTO house_suggestions (
+ house_id, ext_item_id, title, address, price_rub,
+ exposure_days, publish_date,
+ item_link, metro_name, metro_distance,
+ has_good_price_badge, raw_payload, fetched_at
+ ) VALUES (
+ :hid, :ext, :title, :addr, :price,
+ :exp, :pdate,
+ :link, :mname, :mdist,
+ :gpb, CAST(:raw AS jsonb), NOW()
+ )
+ ON CONFLICT (house_id, ext_item_id) DO UPDATE SET
+ title = EXCLUDED.title,
+ price_rub = EXCLUDED.price_rub,
+ exposure_days = EXCLUDED.exposure_days,
+ publish_date = EXCLUDED.publish_date,
+ item_link = EXCLUDED.item_link,
+ metro_name = EXCLUDED.metro_name,
+ metro_distance = EXCLUDED.metro_distance,
+ has_good_price_badge = EXCLUDED.has_good_price_badge,
+ raw_payload = EXCLUDED.raw_payload,
+ fetched_at = NOW()
+ """),
+ {
+ "hid": house_id,
+ "ext": sug.ext_item_id,
+ "title": sug.title,
+ "addr": sug.address,
+ "price": sug.price_rub,
+ "exp": sug.exposure_days,
+ "pdate": sug.publish_date,
+ "link": sug.item_url,
+ "mname": sug.metro_name,
+ "mdist": sug.metro_distance,
+ "gpb": sug.has_good_price_badge,
+ "raw": (
+ json.dumps(sug.raw_payload, ensure_ascii=False) if sug.raw_payload else None
+ ),
+ },
+ )
+
+ # 4. Mark house success
+ db.execute(
+ text("""
+ UPDATE houses
+ SET imv_status = 'ok',
+ last_imv_attempt_at = NOW(),
+ imv_error_reason = NULL
+ WHERE id = :hid
+ """),
+ {"hid": house_id},
+ )
+
+
+def _mark_status(
+ db: Session,
+ house_id: int,
+ status: str,
+ reason: str | None = None,
+) -> None:
+ db.execute(
+ text("""
+ UPDATE houses
+ SET imv_status = :s,
+ last_imv_attempt_at = NOW(),
+ imv_error_reason = :r
+ WHERE id = :hid
+ """),
+ {"hid": house_id, "s": status, "r": reason},
+ )
+ db.commit()
+
+
+# ── Top-level orchestrator ────────────────────────────────────────────────────
+
+_IMVStatus = Literal[
+ "ok", "no_params", "no_address", "not_found", "auth_error", "transient", "error"
+]
+
+
+@dataclass
+class HouseIMVBackfillResult:
+ checked: int = 0
+ saved: int = 0
+ skipped: int = 0
+ errors: int = 0
+ duration_sec: float = field(default=0.0)
+ status_counts: dict[str, int] = field(default_factory=dict)
+
+
+async def backfill_house_imv(
+ db: Session,
+ *,
+ batch_size: int = 50,
+ request_delay_sec: float = 5.0,
+ only_status: str = "pending",
+ house_id: int | None = None,
+) -> HouseIMVBackfillResult:
+ """Run Avito IMV evaluation for each house in scope, save results.
+
+ Args:
+ db: SQLAlchemy session.
+ batch_size: max houses to process (ignored when house_id given).
+ request_delay_sec: sleep between Avito API calls (default 5s — anti-bot).
+ only_status: process houses with this imv_status (default 'pending').
+ Use 'transient_error' to retry failures.
+ house_id: process a single specific house (debug).
+
+ Returns:
+ HouseIMVBackfillResult with aggregate counters.
+ """
+ result = HouseIMVBackfillResult()
+ t0 = time.time()
+
+ if house_id is not None:
+ rows = (
+ db.execute(
+ text("""
+ SELECT id, address, full_address, lat, lon
+ FROM houses
+ WHERE id = CAST(:hid AS bigint)
+ AND lat IS NOT NULL
+ AND lon IS NOT NULL
+ """),
+ {"hid": house_id},
+ )
+ .mappings()
+ .all()
+ )
+ else:
+ rows = (
+ db.execute(
+ text("""
+ SELECT id, address, full_address, lat, lon
+ FROM houses
+ WHERE imv_status = :status
+ AND lat IS NOT NULL
+ AND lon IS NOT NULL
+ AND address IS NOT NULL
+ ORDER BY last_imv_attempt_at NULLS FIRST, id
+ LIMIT :batch
+ """),
+ {"status": only_status, "batch": batch_size},
+ )
+ .mappings()
+ .all()
+ )
+
+ result.checked = len(rows)
+ if not rows:
+ logger.info("house_imv_backfill: nothing to process (status=%r)", only_status)
+ result.duration_sec = time.time() - t0
+ return result
+
+ logger.info(
+ "house_imv_backfill: %d houses (status=%r delay=%.1fs)",
+ result.checked,
+ only_status,
+ request_delay_sec,
+ )
+
+ for i, house in enumerate(rows):
+ hid: int = house["id"]
+ status_str = await _process_one_house(db, dict(house))
+
+ result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1
+ if status_str == "ok":
+ result.saved += 1
+ elif status_str in ("no_params", "no_address"):
+ result.skipped += 1
+ else:
+ result.errors += 1
+
+ logger.info(
+ "[%d/%d] house_id=%d status=%s",
+ i + 1,
+ result.checked,
+ hid,
+ status_str,
+ )
+
+ if i < len(rows) - 1:
+ if status_str == "auth_error":
+ logger.warning("IMV auth error — extra delay 60s before next house")
+ await asyncio.sleep(60)
+ else:
+ await asyncio.sleep(request_delay_sec)
+
+ result.duration_sec = time.time() - t0
+ logger.info(
+ "house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
+ result.checked,
+ result.saved,
+ result.skipped,
+ result.errors,
+ result.duration_sec,
+ result.status_counts,
+ )
+ return result
+
+
+async def _process_one_house(db: Session, house: dict) -> str:
+ """Process a single house. Returns final status string."""
+ hid: int = house["id"]
+
+ params = pick_lot_params(db, hid)
+ if not params:
+ _mark_status(db, hid, "no_params", "no listings with rooms+area")
+ return "no_params"
+
+ address = house.get("address") or house.get("full_address")
+ if not address:
+ _mark_status(db, hid, "no_address", "house.address is NULL")
+ return "no_address"
+
+ enriched = _enrich_address_for_imv(address, house.get("lat"), house.get("lon"))
+ if enriched != address:
+ logger.debug("house_imv: enriched address house=%d %r -> %r", hid, address, enriched)
+
+ try:
+ eval_result = await evaluate_via_imv(address=enriched, **params)
+ except IMVAddressNotFoundError as exc:
+ _mark_status(db, hid, "not_found", str(exc)[:200])
+ return "not_found"
+ except IMVAuthError as exc:
+ _mark_status(db, hid, "transient_error", f"auth: {exc!s}"[:200])
+ return "auth_error"
+ except IMVTransientError as exc:
+ _mark_status(db, hid, "transient_error", str(exc)[:200])
+ return "transient"
+ except Exception as exc:
+ _mark_status(db, hid, "error", repr(exc)[:200])
+ logger.error("house_imv: unexpected error house=%d: %r", hid, exc)
+ return "error"
+
+ try:
+ save_imv_result(db, hid, params, eval_result)
+ db.commit()
+ except Exception as exc:
+ logger.error("house_imv: save failed house=%d: %r", hid, exc)
+ try:
+ db.rollback()
+ except Exception:
+ pass
+ _mark_status(db, hid, "error", f"save_failed: {exc!s}"[:200])
+ return "error"
+
+ return "ok"
diff --git a/tradein-mvp/backend/app/services/yandex_address_backfill.py b/tradein-mvp/backend/app/services/yandex_address_backfill.py
new file mode 100644
index 00000000..ba9586aa
--- /dev/null
+++ b/tradein-mvp/backend/app/services/yandex_address_backfill.py
@@ -0,0 +1,218 @@
+"""T10: Yandex listing address backfill from detail-page <title>.
+
+Yandex SERP cards contain only street-level addresses (e.g. «улица Горького»).
+The offer detail page <title> has the full address including house number:
+«Екатеринбург, улица Горького, 36 — id 12345».
+
+This service finds active yandex listings without a house number in `address`,
+fetches each detail page via curl_cffi(chrome120), extracts the full address
+from the <title> regex, and UPDATE listings.address.
+
+Deliberately NOT wired into scheduler — manual trigger only.
+Triggered via POST /api/v1/admin/scrape/yandex-address-backfill.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import time
+from dataclasses import dataclass, field
+
+from curl_cffi.requests import AsyncSession
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Extract 'Екатеринбург, <full address>' from <title> — matches:
+# «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 на Яндекс.Недвижимости»
+# Group 1 = «Екатеринбург, улица Горького, 36» (stripped of trailing punctuation/spaces)
+_RE_TITLE_ADDRESS = re.compile(
+ r"<title>[^<]*?—\s*"
+ r"(?:[^—,]*?,\s*)?" # optional ЖК/жилой комплекс prefix
+ r"(Екатеринбург,\s*[^<—]+?)"
+ r"\s*—\s*id\s*\d+[^<]*", # allow trailing text after id NNN (e.g. 'на Яндекс...')
+ re.IGNORECASE,
+)
+
+# Listings without a house number: address contains no digit after comma-space
+# (e.g. «улица Горького» but not «улица Горького, 36»).
+# This regex matches a number right after «, » (house number pattern).
+_RE_HAS_HOUSE_NUMBER = re.compile(r",\s*\d+")
+
+
+@dataclass
+class YandexAddressBackfillResult:
+ checked: int = 0
+ saved: int = 0
+ skipped: int = 0
+ errors: int = 0
+ duration_sec: float = field(default=0.0)
+
+
+def _extract_address_from_title(html: str) -> str | None:
+ """Extract full address from Yandex offer detail page ."""
+ html_clean = html.replace("\xa0", " ")
+ m = _RE_TITLE_ADDRESS.search(html_clean)
+ if not m:
+ return None
+ addr = m.group(1).strip().rstrip(",").strip()
+ return addr if addr else None
+
+
+def _eligible_listings(db: Session, limit: int) -> list[dict]:
+ """Active EKB yandex listings without house number in address."""
+ rows = (
+ db.execute(
+ text("""
+ SELECT id, source_id, address, source_url
+ FROM listings
+ WHERE source = 'yandex'
+ AND region_code = 66
+ AND is_active = TRUE
+ AND address IS NOT NULL
+ AND NOT (address ~ ',\\s*\\d+')
+ AND source_url IS NOT NULL
+ ORDER BY scraped_at DESC
+ LIMIT :lim
+ """),
+ {"lim": limit},
+ )
+ .mappings()
+ .all()
+ )
+ return [dict(r) for r in rows]
+
+
+async def backfill_yandex_addresses(
+ db: Session,
+ *,
+ limit: int = 200,
+ request_delay_sec: float = 3.0,
+) -> YandexAddressBackfillResult:
+ """Fetch Yandex detail pages and update address with full street+house number.
+
+ Args:
+ db: SQLAlchemy session.
+ limit: max listings to process.
+ request_delay_sec: sleep between requests (anti-bot; default 3.0s).
+
+ Returns:
+ YandexAddressBackfillResult with checked/saved/skipped/errors counters.
+ """
+ from app.core.config import settings
+
+ result = YandexAddressBackfillResult()
+ t0 = time.time()
+
+ items = _eligible_listings(db, limit)
+ result.checked = len(items)
+
+ if not items:
+ logger.info("yandex_address_backfill: no eligible listings")
+ result.duration_sec = time.time() - t0
+ return result
+
+ logger.info(
+ "yandex_address_backfill: %d listings, delay=%.1fs",
+ result.checked,
+ request_delay_sec,
+ )
+
+ _proxy_url = settings.scraper_proxy_url
+ _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
+
+ async with AsyncSession(
+ impersonate="chrome120",
+ timeout=30.0,
+ proxies=_proxies,
+ headers={
+ "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
+ },
+ ) as session:
+ for i, item in enumerate(items):
+ lid: int = item["id"]
+ old_addr: str = item["address"]
+ url: str = item["source_url"]
+
+ try:
+ resp = await session.get(url, allow_redirects=True)
+ except Exception as exc:
+ logger.warning("yandex_address_backfill: fetch failed listing_id=%s: %s", lid, exc)
+ result.errors += 1
+ await asyncio.sleep(request_delay_sec)
+ continue
+
+ if resp.status_code != 200:
+ logger.warning(
+ "yandex_address_backfill: HTTP %d listing_id=%s", resp.status_code, lid
+ )
+ result.errors += 1
+ await asyncio.sleep(request_delay_sec)
+ continue
+
+ new_addr = _extract_address_from_title(resp.text)
+
+ if new_addr is None:
+ logger.debug("yandex_address_backfill: no address extracted listing_id=%s", lid)
+ result.skipped += 1
+ elif new_addr == old_addr:
+ logger.debug(
+ "yandex_address_backfill: unchanged listing_id=%s addr=%r", lid, old_addr
+ )
+ result.skipped += 1
+ elif not _RE_HAS_HOUSE_NUMBER.search(new_addr):
+ # Extracted address still has no house number — skip to avoid overwrite
+ # with another street-only value.
+ logger.debug(
+ "yandex_address_backfill: still no house number listing_id=%s addr=%r",
+ lid,
+ new_addr,
+ )
+ result.skipped += 1
+ else:
+ logger.info(
+ "yandex_address_backfill: updating listing_id=%s %r -> %r",
+ lid,
+ old_addr,
+ new_addr,
+ )
+ try:
+ db.execute(
+ text("""
+ UPDATE listings
+ SET address = :addr,
+ geocode_tried_at = NULL
+ WHERE id = CAST(:id AS bigint)
+ """),
+ {"addr": new_addr, "id": lid},
+ )
+ db.commit()
+ result.saved += 1
+ except Exception as exc:
+ logger.warning(
+ "yandex_address_backfill: DB update failed listing_id=%s: %s",
+ lid,
+ exc,
+ )
+ result.errors += 1
+ try:
+ db.rollback()
+ except Exception:
+ pass
+
+ if i < len(items) - 1:
+ await asyncio.sleep(request_delay_sec)
+
+ result.duration_sec = time.time() - t0
+ logger.info(
+ "yandex_address_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs",
+ result.checked,
+ result.saved,
+ result.skipped,
+ result.errors,
+ result.duration_sec,
+ )
+ return result
diff --git a/tradein-mvp/backend/tests/test_backfill_wave2.py b/tradein-mvp/backend/tests/test_backfill_wave2.py
new file mode 100644
index 00000000..e08ac26d
--- /dev/null
+++ b/tradein-mvp/backend/tests/test_backfill_wave2.py
@@ -0,0 +1,563 @@
+"""Unit tests for T7/T8a/T10 backfill services (wave-2).
+
+All tests are offline — no DB, no network required.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# ============================================================================
+# T10 — yandex_address_backfill: _extract_address_from_title
+# ============================================================================
+
+
+class TestYandexTitleExtract:
+ """Tests for _extract_address_from_title helper."""
+
+ def _extract(self, html: str) -> str | None:
+ from app.services.yandex_address_backfill import _extract_address_from_title
+
+ return _extract_address_from_title(html)
+
+ def test_extracts_full_address_standard(self):
+ html = (
+ "<title>Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 "
+ "на Яндекс.Недвижимости"
+ )
+ result = self._extract(html)
+ assert result == "Екатеринбург, улица Горького, 36"
+
+ def test_extracts_address_with_corpus(self):
+ html = (
+ "Продажа квартиры — Екатеринбург, улица Ленина, 52/1 — id 9999 "
+ "на Яндекс.Недвижимости"
+ )
+ result = self._extract(html)
+ assert result == "Екатеринбург, улица Ленина, 52/1"
+
+ def test_returns_none_when_no_id_marker(self):
+ html = "Продажа квартиры — Екатеринбург, улица Ленина, 52/1"
+ result = self._extract(html)
+ assert result is None
+
+ def test_returns_none_when_no_title(self):
+ html = "No title here"
+ result = self._extract(html)
+ assert result is None
+
+ def test_strips_trailing_punctuation(self):
+ html = (
+ "Купить квартиру — Екатеринбург, улица Гагарина, 11, — id 111 "
+ "на Яндекс.Недвижимости"
+ )
+ result = self._extract(html)
+ # should strip trailing comma
+ assert result is not None
+ assert not result.endswith(",")
+
+ def test_nbsp_replaced(self):
+ html = (
+ "— Екатеринбург,\xa0улица Горького, 36 — id 12 " "на Яндекс.Недвижимости"
+ )
+ result = self._extract(html)
+ assert result is not None
+ assert "\xa0" not in result
+
+ def test_case_insensitive_city(self):
+ html = (
+ "— ЕКАТЕРИНБУРГ, улица Горького, 36 — id 777 " "на Яндекс.Недвижимости"
+ )
+ # regex is case-insensitive; city name retains original case
+ result = self._extract(html)
+ assert result is not None
+ assert "Горького, 36" in result
+
+
+class TestYandexHouseNumberRegex:
+ def test_no_house_number_pattern(self):
+ from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
+
+ assert not _RE_HAS_HOUSE_NUMBER.search("улица Горького")
+ assert not _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького")
+
+ def test_has_house_number_pattern(self):
+ from app.services.yandex_address_backfill import _RE_HAS_HOUSE_NUMBER
+
+ assert _RE_HAS_HOUSE_NUMBER.search("Екатеринбург, улица Горького, 36")
+ assert _RE_HAS_HOUSE_NUMBER.search("улица Ленина, 52/1")
+
+
+# ============================================================================
+# T10 — backfill_yandex_addresses: integration (mocked)
+# ============================================================================
+
+
+@pytest.mark.asyncio
+async def test_backfill_yandex_addresses_saves_enriched():
+ """Happy path: fetch returns full address → saved=1."""
+ from app.services.yandex_address_backfill import backfill_yandex_addresses
+
+ html_with_addr = (
+ "Продажа — Екатеринбург, улица Горького, 36 — id 11 "
+ "на Яндекс.Недвижимости"
+ )
+
+ mock_resp = MagicMock()
+ mock_resp.status_code = 200
+ mock_resp.text = html_with_addr
+
+ mock_session = AsyncMock()
+ mock_session.get = AsyncMock(return_value=mock_resp)
+ mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+ mock_session.__aexit__ = AsyncMock(return_value=False)
+
+ mock_db = MagicMock()
+ mock_db.execute = MagicMock()
+ mock_db.commit = MagicMock()
+
+ items = [
+ {
+ "id": 42,
+ "source_id": "yandex-42",
+ "address": "улица Горького",
+ "source_url": "https://realty.yandex.ru/offer/11/",
+ }
+ ]
+
+ with (
+ patch(
+ "app.services.yandex_address_backfill._eligible_listings",
+ return_value=items,
+ ),
+ patch(
+ "app.services.yandex_address_backfill.AsyncSession",
+ return_value=mock_session,
+ ),
+ ):
+ result = await backfill_yandex_addresses(mock_db, limit=10)
+
+ assert result.checked == 1
+ assert result.saved == 1
+ assert result.errors == 0
+ assert result.skipped == 0
+
+
+@pytest.mark.asyncio
+async def test_backfill_yandex_addresses_skips_no_house_number():
+ """Address extracted but still has no house number → skipped."""
+ from app.services.yandex_address_backfill import backfill_yandex_addresses
+
+ # Title without house number in extracted part
+ html = "— Екатеринбург, улица Горького — id 22 на Яндекс.Недвижимости"
+
+ mock_resp = MagicMock()
+ mock_resp.status_code = 200
+ mock_resp.text = html
+
+ mock_session = AsyncMock()
+ mock_session.get = AsyncMock(return_value=mock_resp)
+ mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+ mock_session.__aexit__ = AsyncMock(return_value=False)
+
+ items = [
+ {
+ "id": 99,
+ "source_id": "y-99",
+ "address": "улица Горького",
+ "source_url": "https://realty.yandex.ru/offer/22/",
+ }
+ ]
+
+ with (
+ patch(
+ "app.services.yandex_address_backfill._eligible_listings",
+ return_value=items,
+ ),
+ patch(
+ "app.services.yandex_address_backfill.AsyncSession",
+ return_value=mock_session,
+ ),
+ ):
+ result = await backfill_yandex_addresses(MagicMock(), limit=10)
+
+ assert result.skipped == 1
+ assert result.saved == 0
+
+
+@pytest.mark.asyncio
+async def test_backfill_yandex_addresses_error_on_non_200():
+ """Non-200 response → errors counter incremented, no crash."""
+ from app.services.yandex_address_backfill import backfill_yandex_addresses
+
+ mock_resp = MagicMock()
+ mock_resp.status_code = 403
+ mock_resp.text = ""
+
+ mock_session = AsyncMock()
+ mock_session.get = AsyncMock(return_value=mock_resp)
+ mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+ mock_session.__aexit__ = AsyncMock(return_value=False)
+
+ items = [
+ {
+ "id": 55,
+ "source_id": "y-55",
+ "address": "улица Горького",
+ "source_url": "https://realty.yandex.ru/offer/55/",
+ }
+ ]
+
+ with (
+ patch(
+ "app.services.yandex_address_backfill._eligible_listings",
+ return_value=items,
+ ),
+ patch(
+ "app.services.yandex_address_backfill.AsyncSession",
+ return_value=mock_session,
+ ),
+ ):
+ result = await backfill_yandex_addresses(MagicMock(), limit=10)
+
+ assert result.errors == 1
+ assert result.saved == 0
+
+
+# ============================================================================
+# T8a — cian_price_history: backfill_cian_price_history
+# ============================================================================
+
+
+@pytest.mark.asyncio
+async def test_cian_price_history_saves_changes():
+ """Happy path: fetch returns DetailEnrichment with price_changes → saved."""
+ from app.services.cian_price_history import backfill_cian_price_history
+ from app.services.scrapers.cian_detail import DetailEnrichment
+
+ enrichment = DetailEnrichment(
+ cian_id=123456,
+ price_changes=[
+ {"change_time": "2026-04-01T00:00:00Z", "price_rub": 5000000, "diff_percent": None},
+ {"change_time": "2026-04-15T00:00:00Z", "price_rub": 4900000, "diff_percent": -2.0},
+ ],
+ )
+
+ mock_db = MagicMock()
+ # Simulate listings rows returned by SELECT
+ rows = [{"id": 10, "source_url": "https://ekb.cian.ru/sale/flat/123456/"}]
+
+ with (
+ patch(
+ "app.services.cian_price_history.fetch_detail",
+ new_callable=AsyncMock,
+ return_value=enrichment,
+ ) as mock_fetch,
+ patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
+ patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = rows
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_cian_price_history(mock_db, batch_size=10)
+
+ assert result.checked == 1
+ assert result.saved == 2 # 2 price_changes
+ assert result.errors == 0
+ mock_fetch.assert_called_once()
+ mock_save.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_cian_price_history_skips_no_changes():
+ """Fetch succeeds but no price_changes → skipped (not error)."""
+ from app.services.cian_price_history import backfill_cian_price_history
+ from app.services.scrapers.cian_detail import DetailEnrichment
+
+ enrichment = DetailEnrichment(cian_id=111, price_changes=[])
+
+ mock_db = MagicMock()
+ rows = [{"id": 20, "source_url": "https://ekb.cian.ru/sale/flat/111/"}]
+
+ with (
+ patch(
+ "app.services.cian_price_history.fetch_detail",
+ new_callable=AsyncMock,
+ return_value=enrichment,
+ ),
+ patch("app.services.cian_price_history.save_detail_enrichment") as mock_save,
+ patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = rows
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_cian_price_history(mock_db, batch_size=10)
+
+ assert result.skipped == 1
+ assert result.saved == 0
+ mock_save.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_cian_price_history_handles_fetch_none():
+ """fetch_detail returns None → errors+1, no crash."""
+ from app.services.cian_price_history import backfill_cian_price_history
+
+ mock_db = MagicMock()
+ rows = [{"id": 30, "source_url": "https://ekb.cian.ru/sale/flat/999/"}]
+
+ with (
+ patch(
+ "app.services.cian_price_history.fetch_detail",
+ new_callable=AsyncMock,
+ return_value=None,
+ ),
+ patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = rows
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_cian_price_history(mock_db, batch_size=10)
+
+ assert result.errors == 1
+ assert result.saved == 0
+
+
+# ============================================================================
+# T7 — house_imv_backfill: helpers
+# ============================================================================
+
+
+class TestHouseTypeMap:
+ def test_known_types(self):
+ from app.services.house_imv_backfill import _map_house_type
+
+ assert _map_house_type("panel") == "panel"
+ assert _map_house_type("brick") == "brick"
+ assert _map_house_type("monolith") == "monolithic"
+ assert _map_house_type("monolith_brick") == "monolithic"
+ assert _map_house_type("monolithic") == "monolithic"
+ assert _map_house_type("block") == "block"
+ assert _map_house_type("wood") == "wood"
+
+ def test_unknown_falls_back_to_panel(self):
+ from app.services.house_imv_backfill import _map_house_type
+
+ assert _map_house_type("unknown_type") == "panel"
+ assert _map_house_type(None) == "panel"
+ assert _map_house_type("") == "panel"
+
+ def test_case_insensitive(self):
+ from app.services.house_imv_backfill import _map_house_type
+
+ assert _map_house_type("PANEL") == "panel"
+ assert _map_house_type("Brick") == "brick"
+ assert _map_house_type("MONOLITH_BRICK") == "monolithic"
+
+
+class TestRegionPrefix:
+ def test_ekb_coords(self):
+ from app.services.house_imv_backfill import _detect_region_prefix
+
+ prefix = _detect_region_prefix(56.8296, 60.6001)
+ assert prefix == "Свердловская область, Екатеринбург"
+
+ def test_none_coords_default(self):
+ from app.services.house_imv_backfill import _detect_region_prefix
+
+ prefix = _detect_region_prefix(None, None)
+ assert prefix == "Свердловская область, Екатеринбург"
+
+ def test_out_of_any_region_default(self):
+ from app.services.house_imv_backfill import _detect_region_prefix
+
+ # Moscow coords — not in any bbox → default EKB
+ prefix = _detect_region_prefix(55.75, 37.6)
+ assert prefix == "Свердловская область, Екатеринбург"
+
+
+class TestEnrichAddress:
+ def test_prepends_region_when_no_marker(self):
+ from app.services.house_imv_backfill import _enrich_address_for_imv
+
+ result = _enrich_address_for_imv("улица Горького, 36", 56.83, 60.60)
+ assert result.startswith("Свердловская область, Екатеринбург,")
+ assert "улица Горького, 36" in result
+
+ def test_does_not_prepend_when_region_already_present(self):
+ from app.services.house_imv_backfill import _enrich_address_for_imv
+
+ addr = "Свердловская область, Екатеринбург, улица Горького, 36"
+ result = _enrich_address_for_imv(addr, 56.83, 60.60)
+ assert result == addr
+
+ def test_does_not_prepend_when_kraj_present(self):
+ from app.services.house_imv_backfill import _enrich_address_for_imv
+
+ addr = "Ставропольский край, Ставрополь, улица Ленина, 1"
+ result = _enrich_address_for_imv(addr, 45.05, 41.97)
+ assert result == addr
+
+
+# ============================================================================
+# T7 — backfill_house_imv: integration (mocked)
+# ============================================================================
+
+
+@pytest.mark.asyncio
+async def test_backfill_house_imv_ok_path():
+ """Happy path: picks params, calls IMV, saves → saved=1."""
+ from app.services.house_imv_backfill import backfill_house_imv
+ from app.services.scrapers.avito_imv import IMVEvaluation, IMVGeo
+
+ fake_geo = IMVGeo(geo_hash="tok", lat=56.83, lon=60.60)
+ fake_eval = IMVEvaluation(
+ cache_key="abc123",
+ address="ЕКБ ул. Горького, 36",
+ rooms=2,
+ area_m2=52.0,
+ floor=3,
+ floor_at_home=10,
+ house_type="panel",
+ renovation_type="cosmetic",
+ has_balcony=True,
+ has_loggia=False,
+ geo=fake_geo,
+ recommended_price=5_000_000,
+ lower_price=4_500_000,
+ higher_price=5_500_000,
+ market_count=24,
+ placement_history=[],
+ suggestions=[],
+ )
+
+ fake_params = {
+ "rooms": 2,
+ "area_m2": 52.0,
+ "floor": 3,
+ "floor_at_home": 10,
+ "house_type": "panel",
+ "renovation_type": "cosmetic",
+ "has_balcony": True,
+ "has_loggia": False,
+ }
+
+ mock_db = MagicMock()
+ mock_db.commit = MagicMock()
+
+ houses = [
+ {"id": 7, "address": "улица Горького, 36", "full_address": None, "lat": 56.83, "lon": 60.60}
+ ]
+
+ with (
+ patch(
+ "app.services.house_imv_backfill.pick_lot_params",
+ return_value=fake_params,
+ ),
+ patch(
+ "app.services.house_imv_backfill.evaluate_via_imv",
+ new_callable=AsyncMock,
+ return_value=fake_eval,
+ ),
+ patch("app.services.house_imv_backfill.save_imv_result") as mock_save,
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = houses
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
+
+ assert result.checked == 1
+ assert result.saved == 1
+ assert result.errors == 0
+ assert result.status_counts.get("ok") == 1
+ mock_save.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_backfill_house_imv_no_params():
+ """pick_lot_params returns {} → skipped (no_params status)."""
+ from app.services.house_imv_backfill import backfill_house_imv
+
+ mock_db = MagicMock()
+ mock_db.execute = MagicMock()
+ mock_db.commit = MagicMock()
+
+ houses = [
+ {"id": 8, "address": "улица Ленина, 1", "full_address": None, "lat": 56.83, "lon": 60.60}
+ ]
+
+ with (
+ patch(
+ "app.services.house_imv_backfill.pick_lot_params",
+ return_value={},
+ ),
+ patch("app.services.house_imv_backfill._mark_status") as mock_mark,
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = houses
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
+
+ assert result.skipped == 1
+ assert result.saved == 0
+ mock_mark.assert_called_once_with(mock_db, 8, "no_params", "no listings with rooms+area")
+
+
+@pytest.mark.asyncio
+async def test_backfill_house_imv_not_found():
+ """IMVAddressNotFoundError → status not_found, no crash."""
+ from app.services.house_imv_backfill import backfill_house_imv
+ from app.services.scrapers.avito_imv import IMVAddressNotFoundError
+
+ fake_params = {
+ "rooms": 2,
+ "area_m2": 52.0,
+ "floor": 3,
+ "floor_at_home": 10,
+ "house_type": "panel",
+ "renovation_type": "cosmetic",
+ "has_balcony": True,
+ "has_loggia": False,
+ }
+
+ mock_db = MagicMock()
+ mock_db.commit = MagicMock()
+
+ houses = [
+ {
+ "id": 9,
+ "address": "несуществующий адрес",
+ "full_address": None,
+ "lat": 56.83,
+ "lon": 60.60,
+ }
+ ]
+
+ with (
+ patch(
+ "app.services.house_imv_backfill.pick_lot_params",
+ return_value=fake_params,
+ ),
+ patch(
+ "app.services.house_imv_backfill.evaluate_via_imv",
+ new_callable=AsyncMock,
+ side_effect=IMVAddressNotFoundError("not found"),
+ ),
+ patch("app.services.house_imv_backfill._mark_status") as mock_mark,
+ ):
+ mock_mappings = MagicMock()
+ mock_mappings.all.return_value = houses
+ mock_db.execute.return_value.mappings.return_value = mock_mappings
+
+ result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
+
+ assert result.errors == 1
+ assert result.saved == 0
+ mock_mark.assert_called_once_with(mock_db, 9, "not_found", "not found")