933 lines
35 KiB
Python
933 lines
35 KiB
Python
"""Admin endpoints — debug + ручной запуск парсеров.
|
||
|
||
Auth: Caddy basic_auth gate'ит /trade-in/api/v1/admin/* — application layer открыт.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
from typing import Annotated, Any, Literal
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.core.db import SessionLocal, get_db
|
||
from app.schemas.trade_in import ScheduleConfig, ScheduleConfigUpdate
|
||
from app.services import cian_session as cian_session_svc
|
||
from app.services import scrape_runs as runs_mod
|
||
from app.services.geocoder import geocode
|
||
from app.services.scrape_pipeline import run_avito_city_sweep
|
||
from app.services.scraper_settings import invalidate_cache
|
||
from app.services.scrapers.avito import AvitoScraper
|
||
from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
|
||
from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment
|
||
from app.services.scrapers.avito_imv import (
|
||
IMVAddressNotFoundError,
|
||
evaluate_via_imv,
|
||
save_imv_evaluation,
|
||
save_imv_placement_history,
|
||
)
|
||
from app.services.scrapers.base import save_listings
|
||
from app.services.scrapers.cian import CianScraper
|
||
from app.services.scrapers.n1 import N1Scraper
|
||
from app.services.scrapers.yandex_detail import YandexDetailScraper
|
||
from app.services.scrapers.yandex_newbuilding import YandexNewbuildingScraper
|
||
from app.services.scrapers.yandex_realty import YandexRealtyScraper
|
||
from app.services.scrapers.yandex_valuation import YandexValuationScraper
|
||
from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
_ALL_SOURCES = ["avito", "cian", "yandex", "n1"]
|
||
|
||
|
||
class ScrapeRequest(BaseModel):
|
||
lat: float
|
||
lon: float
|
||
radius_m: int = Field(default=1000, ge=100, le=20000)
|
||
sources: list[Literal["avito", "cian", "yandex", "n1"]] = Field(
|
||
default_factory=lambda: list(_ALL_SOURCES)
|
||
)
|
||
# Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом.
|
||
multi_room_yandex: bool = False
|
||
# Если True — Yandex полный обход 5 rooms × 3 sorts × 2 pages = 30 запросов / ~150s.
|
||
deep_yandex: bool = False
|
||
# Если True — скрейп Cian по 4 сегментам комнат отдельно (~4× лотов).
|
||
multi_room_cian: bool = False
|
||
# Если True — скрейп N1 по 4 сегментам комнат отдельно (~4× лотов).
|
||
multi_room_n1: bool = False
|
||
|
||
|
||
class ScrapeResult(BaseModel):
|
||
source: str
|
||
fetched: int
|
||
inserted: int
|
||
updated: int
|
||
|
||
|
||
class ScrapeResponse(BaseModel):
|
||
total_fetched: int
|
||
total_inserted: int
|
||
total_updated: int
|
||
by_source: list[ScrapeResult]
|
||
|
||
|
||
@router.post("/scrape", response_model=ScrapeResponse)
|
||
async def scrape_around(
|
||
payload: ScrapeRequest,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> ScrapeResponse:
|
||
"""Запустить парсеры для точки (lat, lon) в радиусе radius_m метров.
|
||
|
||
Примеры:
|
||
curl -X POST /api/v1/admin/scrape \\
|
||
-H 'Content-Type: application/json' \\
|
||
-d '{"lat":56.8332,"lon":60.5944,"radius_m":1000,"sources":["avito"]}'
|
||
"""
|
||
results: list[ScrapeResult] = []
|
||
|
||
for source in payload.sources:
|
||
scraper_cls = {
|
||
"avito": AvitoScraper,
|
||
"cian": CianScraper,
|
||
"yandex": YandexRealtyScraper,
|
||
"n1": N1Scraper,
|
||
}.get(source)
|
||
if scraper_cls is None:
|
||
continue
|
||
async with scraper_cls() as scraper:
|
||
if source == "yandex" and payload.deep_yandex:
|
||
lots = await scraper.fetch_around_multi_room(
|
||
payload.lat, payload.lon, payload.radius_m,
|
||
sorts=("DATE_DESC", "PRICE", "AREA_DESC"),
|
||
pages=(0, 1),
|
||
)
|
||
elif source == "yandex" and payload.multi_room_yandex:
|
||
lots = await scraper.fetch_around_multi_room(
|
||
payload.lat, payload.lon, payload.radius_m
|
||
)
|
||
elif source == "cian" and payload.multi_room_cian:
|
||
lots = await scraper.fetch_around_multi_room(
|
||
payload.lat, payload.lon, payload.radius_m
|
||
)
|
||
elif source == "n1" and payload.multi_room_n1:
|
||
lots = await scraper.fetch_around_multi_room(
|
||
payload.lat, payload.lon, payload.radius_m
|
||
)
|
||
else:
|
||
lots = await scraper.fetch_around(
|
||
payload.lat, payload.lon, payload.radius_m
|
||
)
|
||
inserted, updated = save_listings(db, lots)
|
||
results.append(
|
||
ScrapeResult(source=source, fetched=len(lots), inserted=inserted, updated=updated)
|
||
)
|
||
|
||
return ScrapeResponse(
|
||
total_fetched=sum(r.fetched for r in results),
|
||
total_inserted=sum(r.inserted for r in results),
|
||
total_updated=sum(r.updated for r in results),
|
||
by_source=results,
|
||
)
|
||
|
||
|
||
def _clean_address_for_geocode(addr: str) -> str:
|
||
"""Чистим address для геокодера.
|
||
|
||
Cian отдаёт «улица Латвийская, 56/3 · р-н Чкаловский» — суффикс ' · ...'
|
||
мешает Nominatim. Берём часть до ' · '. N1 отдаёт «Репина, 75/2 стр.» — ок.
|
||
"""
|
||
main = addr.split(" · ")[0].strip()
|
||
return main or addr
|
||
|
||
|
||
@router.post("/geocode-missing")
|
||
async def geocode_missing(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = 100,
|
||
target: Literal["listings", "deals"] = "listings",
|
||
) -> dict:
|
||
"""Геокодинг listings ИЛИ deals у которых нет lat/lon (используя address).
|
||
|
||
target=listings (по умолч.) — объявления Cian/N1; target=deals — сделки Росреестра.
|
||
Чанк-обработка с бюджетом по времени (~240с, заведомо меньше cron
|
||
`curl -m 320`): за вызов геокодим сколько успеваем, остаток уходит в
|
||
`remaining`, cron вызывает в цикле пока `remaining` > 0.
|
||
|
||
geocode_tried_at: после КАЖДОЙ попытки (успех/провал) ставим NOW(). Failed-
|
||
адреса не выбираются повторно 7 дней → cron-loop завершается, не зацикливается.
|
||
geom обновляется автоматически триггером.
|
||
"""
|
||
# Доп. фильтр для listings — у Avito/N1 встречаются плейсхолдер-адреса.
|
||
extra_filter = (
|
||
"AND address NOT LIKE '%(Avito)%' AND address NOT LIKE '%(N1)%'"
|
||
if target == "listings"
|
||
else ""
|
||
)
|
||
rows = db.execute(
|
||
text(
|
||
f"""
|
||
SELECT id, address
|
||
FROM {target}
|
||
WHERE lat IS NULL
|
||
AND COALESCE(address, '') != ''
|
||
{extra_filter}
|
||
AND (geocode_tried_at IS NULL
|
||
OR geocode_tried_at < NOW() - interval '7 days')
|
||
ORDER BY geocode_tried_at NULLS FIRST
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{"limit": limit},
|
||
).mappings().all()
|
||
|
||
# Бюджет по времени: геокодинг упирается в Nominatim (1 req/sec + typo-тиры
|
||
# на провалах), 100 адресов могут не уложиться в cron-таймаут `curl -m 320`.
|
||
# Выходим из цикла заведомо раньше — remaining > 0 заставит cron продолжить.
|
||
budget_sec = 240.0
|
||
started = time.monotonic()
|
||
|
||
geocoded = 0
|
||
skipped = 0
|
||
for row in rows:
|
||
if time.monotonic() - started > budget_sec:
|
||
logger.info(
|
||
"geocode-missing: бюджет %.0fс исчерпан, обработано %d — cron продолжит",
|
||
budget_sec, geocoded + skipped,
|
||
)
|
||
break
|
||
clean = _clean_address_for_geocode(row["address"])
|
||
result = await geocode(clean, db)
|
||
if result is None:
|
||
# Помечаем что пробовали — иначе ретрай на каждом cron.
|
||
db.execute(
|
||
text(f"UPDATE {target} SET geocode_tried_at = NOW() WHERE id = :id"),
|
||
{"id": row["id"]},
|
||
)
|
||
db.commit()
|
||
skipped += 1
|
||
continue
|
||
# geom пересчитается триггером из lat/lon.
|
||
db.execute(
|
||
text(
|
||
f"UPDATE {target} SET lat = :lat, lon = :lon, "
|
||
"geocode_tried_at = NOW() WHERE id = :id"
|
||
),
|
||
{"lat": result.lat, "lon": result.lon, "id": row["id"]},
|
||
)
|
||
db.commit()
|
||
geocoded += 1
|
||
|
||
# Сколько ещё не-геокоженных адресов ждут (для cron-loop'а).
|
||
remaining = db.execute(
|
||
text(
|
||
f"""
|
||
SELECT count(*) FROM {target}
|
||
WHERE lat IS NULL
|
||
AND COALESCE(address, '') != ''
|
||
{extra_filter}
|
||
AND (geocode_tried_at IS NULL
|
||
OR geocode_tried_at < NOW() - interval '7 days')
|
||
"""
|
||
)
|
||
).scalar()
|
||
|
||
return {
|
||
"checked": len(rows),
|
||
"geocoded": geocoded,
|
||
"skipped": skipped,
|
||
"remaining": int(remaining or 0),
|
||
}
|
||
|
||
|
||
@router.post("/scrape/cian/upload-cookies", status_code=200)
|
||
async def upload_cian_cookies(
|
||
cookies: dict[str, str],
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict:
|
||
"""Upload Cian session cookies для аутентификации Valuation Calculator.
|
||
|
||
Body: JSON object вида { "_ym_uid": "...", "_cian_visitor_session_id": "...", ... }
|
||
Cookies экспортируются из Chrome DevTools → Application → Cookies → www.cian.ru
|
||
или через расширение Cookie-Editor → Export → JSON (object format).
|
||
|
||
Шаги:
|
||
1. Фильтрует payload по CIAN_REQUIRED_COOKIES.
|
||
2. Верифицирует через GET /kalkulator-nedvizhimosti/ — проверяет isAuthenticated.
|
||
3. Сохраняет зашифрованно (pgp_sym_encrypt) в cian_session_cookies.
|
||
|
||
Returns: {"ok": true, "userId": <int>, "cookieCount": <int>}
|
||
"""
|
||
if not settings.cookie_encryption_key:
|
||
raise HTTPException(status_code=503, detail="COOKIE_ENCRYPTION_KEY not configured")
|
||
|
||
cleaned = {k: v for k, v in cookies.items() if k in cian_session_svc.CIAN_REQUIRED_COOKIES}
|
||
if not cleaned:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=(
|
||
f"No recognized Cian cookies in payload. "
|
||
f"Expected any of: {sorted(cian_session_svc.CIAN_REQUIRED_COOKIES)}"
|
||
),
|
||
)
|
||
|
||
state = await cian_session_svc.verify_session(cleaned)
|
||
if state is None:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail="Cookies invalid or session not authenticated on cian.ru",
|
||
)
|
||
|
||
user_id = state.get("user", {}).get("userId")
|
||
if not user_id:
|
||
raise HTTPException(status_code=400, detail="Authenticated state missing userId")
|
||
|
||
cian_session_svc.save_session(db, account_user_id=int(user_id), cookies=cleaned)
|
||
return {"ok": True, "userId": user_id, "cookieCount": len(cleaned)}
|
||
|
||
|
||
@router.get("/scrape/cian/test-auth", status_code=200)
|
||
async def test_cian_auth(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict:
|
||
"""Проверить что текущие сохранённые Cian cookies ещё валидны.
|
||
|
||
Returns: {"authenticated": bool, "userId": <int|null>, "reason": <str|null>}
|
||
"""
|
||
if not settings.cookie_encryption_key:
|
||
return {"authenticated": False, "userId": None, "reason": "encryption_key_not_configured"}
|
||
|
||
cookies = cian_session_svc.load_session(db)
|
||
if cookies is None:
|
||
return {"authenticated": False, "userId": None, "reason": "no_session_in_db"}
|
||
|
||
state = await cian_session_svc.verify_session(cookies)
|
||
if state is None:
|
||
return {"authenticated": False, "userId": None, "reason": "session_expired_or_invalid"}
|
||
|
||
user_id = state.get("user", {}).get("userId")
|
||
return {"authenticated": True, "userId": user_id, "reason": None}
|
||
|
||
|
||
# ── Geocode backfill: batch address-dedup geocoder (all sources) ─────────────
|
||
|
||
|
||
@router.get("/scrape/geocode-missing-listings/status")
|
||
async def get_geocode_backfill_status(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Текущая статистика backfill — сколько ещё pending geocode.
|
||
|
||
Включает per-source breakdown и оценку времени при Nominatim 1 req/sec.
|
||
"""
|
||
stats = db.execute(
|
||
text(
|
||
"""
|
||
SELECT
|
||
source,
|
||
COUNT(*) AS total,
|
||
COUNT(*) FILTER (WHERE lat IS NULL AND address IS NOT NULL) AS pending_geocode,
|
||
COUNT(*) FILTER (WHERE lat IS NOT NULL) AS has_coords
|
||
FROM listings
|
||
WHERE is_active = true
|
||
GROUP BY source
|
||
ORDER BY pending_geocode DESC
|
||
"""
|
||
)
|
||
).mappings().all()
|
||
|
||
total_pending = sum(s["pending_geocode"] for s in stats)
|
||
unique_addresses_pending = db.execute(
|
||
text(
|
||
"""
|
||
SELECT COUNT(DISTINCT address) FROM listings
|
||
WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) >= 5
|
||
"""
|
||
)
|
||
).scalar() or 0
|
||
|
||
cache_size = db.execute(
|
||
text("SELECT COUNT(*) FROM geocode_cache WHERE expires_at > NOW()")
|
||
).scalar() or 0
|
||
|
||
return {
|
||
"total_pending_listings": total_pending,
|
||
"unique_addresses_pending": int(unique_addresses_pending),
|
||
"estimated_nominatim_minutes": round(int(unique_addresses_pending) / 60.0, 1),
|
||
"cache_size_active": int(cache_size),
|
||
"per_source": [dict(s) for s in stats],
|
||
}
|
||
|
||
|
||
@router.post("/scrape/geocode-missing-listings")
|
||
async def trigger_geocode_missing_listings(
|
||
background_tasks: BackgroundTasks,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
batch_size: int = Query(default=200, ge=1, le=2000),
|
||
dry_run: bool = Query(default=False),
|
||
background: bool = Query(
|
||
default=False,
|
||
description="True → запустить в BackgroundTasks (не ждать результат).",
|
||
),
|
||
) -> dict[str, Any]:
|
||
"""Manual trigger для geocode backfill всех listings с NULL coords (address-dedup).
|
||
|
||
batch_size=200 ≈ 3.5 мин при Nominatim 1 req/sec (без cache hits — быстрее с cache).
|
||
Безопасно для повторного запуска (ON CONFLICT в geocode_cache, UPDATE WHERE lat IS NULL).
|
||
|
||
dry_run=true — показывает что бы сделалось, без UPDATE listings.
|
||
background=true — запускает через BackgroundTasks, возвращает немедленно (для больших batch).
|
||
"""
|
||
from app.core.db import SessionLocal
|
||
|
||
if background:
|
||
async def _bg_task() -> None:
|
||
bg_db = SessionLocal()
|
||
try:
|
||
await geocode_missing_listings(bg_db, batch_size=batch_size, dry_run=dry_run)
|
||
except Exception:
|
||
logger.exception("geocode-missing-listings background task crashed")
|
||
finally:
|
||
bg_db.close()
|
||
|
||
background_tasks.add_task(_bg_task)
|
||
return {
|
||
"status": "queued",
|
||
"batch_size": batch_size,
|
||
"dry_run": dry_run,
|
||
"note": "Running in background. Check logs for progress.",
|
||
}
|
||
|
||
result: GeocodeBackfillResult = await geocode_missing_listings(
|
||
db, batch_size=batch_size, dry_run=dry_run
|
||
)
|
||
return {
|
||
"status": "ok",
|
||
"batch_size": batch_size,
|
||
"dry_run": dry_run,
|
||
"addresses_total": result.addresses_total,
|
||
"addresses_processed": result.addresses_processed,
|
||
"addresses_geocoded": result.addresses_geocoded,
|
||
"addresses_failed": result.addresses_failed,
|
||
"listings_updated": result.listings_updated,
|
||
"cache_hits": result.cache_hits,
|
||
"cache_misses": result.cache_misses,
|
||
"duration_sec": round(result.duration_sec, 2),
|
||
}
|
||
|
||
|
||
# ── Stage 4a: Avito enrichment endpoints (single-shot debug + manual triggers) ──
|
||
|
||
|
||
@router.post("/scrape/avito-house")
|
||
async def scrape_avito_house(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
house_url: str,
|
||
) -> dict[str, object]:
|
||
"""Enrichment ОДНОГО дома Avito Houses Catalog.
|
||
|
||
Body params (query): `house_url=/catalog/houses/<slug>/<id>` (absolute или relative path).
|
||
|
||
Flow: fetch_house_catalog → save_house_catalog_enrichment.
|
||
Returns counters {'house_id', 'reviews', 'sellers', 'listings_linked', 'placement_history'}.
|
||
"""
|
||
try:
|
||
enrichment = await fetch_house_catalog(house_url)
|
||
except Exception as e:
|
||
logger.exception("avito-house: fetch failed for %s", house_url)
|
||
raise HTTPException(status_code=502, detail=f"fetch failed: {e}") from e
|
||
|
||
try:
|
||
counters = save_house_catalog_enrichment(db, enrichment)
|
||
except Exception as e:
|
||
logger.exception("avito-house: save failed for %s", house_url)
|
||
raise HTTPException(status_code=500, detail=f"save failed: {e}") from e
|
||
|
||
logger.info("avito-house ok: %s → %s", house_url, counters)
|
||
return {"ok": True, "house_url": house_url, "counters": counters}
|
||
|
||
|
||
@router.post("/scrape/avito-detail")
|
||
async def scrape_avito_detail(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
item_url: str,
|
||
) -> dict[str, object]:
|
||
"""Detail enrichment ОДНОГО listing.
|
||
|
||
Body params (query): `item_url=/ekaterinburg/kvartiry/...` (absolute или relative).
|
||
|
||
Flow: fetch_detail → save_detail_enrichment (UPDATE listings WHERE source='avito').
|
||
Returns {'ok', 'item_id', 'updated'}.
|
||
"""
|
||
try:
|
||
enrichment = await fetch_detail(item_url)
|
||
except Exception as e:
|
||
logger.exception("avito-detail: fetch failed for %s", item_url)
|
||
raise HTTPException(status_code=502, detail=f"fetch failed: {e}") from e
|
||
|
||
try:
|
||
updated = save_detail_enrichment(db, enrichment)
|
||
except Exception as e:
|
||
logger.exception("avito-detail: save failed for %s", item_url)
|
||
raise HTTPException(status_code=500, detail=f"save failed: {e}") from e
|
||
|
||
if not updated:
|
||
# Listing с этим source_id ещё не существует в БД — был bypass через
|
||
# /admin/scrape/avito-detail для несуществующего listing.
|
||
logger.warning("avito-detail: no listing matched source_id=%s", enrichment.item_id)
|
||
logger.info("avito-detail ok: %s item_id=%s updated=%s", item_url, enrichment.item_id, updated)
|
||
return {"ok": True, "item_id": enrichment.item_id, "updated": updated}
|
||
|
||
|
||
@router.post("/scrape/avito-imv")
|
||
async def scrape_avito_imv(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
address: str,
|
||
rooms: int,
|
||
area_m2: float,
|
||
floor: int,
|
||
floor_at_home: int,
|
||
house_type: str,
|
||
renovation_type: str,
|
||
has_balcony: bool = False,
|
||
has_loggia: bool = False,
|
||
) -> dict[str, object]:
|
||
"""Debug-trigger Avito IMV evaluation. БЕЗ cache — всегда свежий fetch.
|
||
|
||
Production IMV вызывается из estimator (Stage 3, on-demand с 24h cache).
|
||
Этот endpoint — для ручной отладки / data-pipeline tools.
|
||
|
||
Returns {'ok', 'cache_key', 'recommended_price', 'range', 'market_count',
|
||
'evaluation_id', 'history_saved'}.
|
||
"""
|
||
try:
|
||
result = await evaluate_via_imv(
|
||
address=address,
|
||
rooms=rooms,
|
||
area_m2=area_m2,
|
||
floor=floor,
|
||
floor_at_home=floor_at_home,
|
||
house_type=house_type,
|
||
renovation_type=renovation_type,
|
||
has_balcony=has_balcony,
|
||
has_loggia=has_loggia,
|
||
)
|
||
except IMVAddressNotFoundError as e:
|
||
raise HTTPException(status_code=404, detail=f"IMV address not found: {e}") from e
|
||
except Exception as e:
|
||
logger.exception("avito-imv: fetch failed for %s", address)
|
||
raise HTTPException(status_code=502, detail=f"fetch failed: {e}") from e
|
||
|
||
try:
|
||
eval_id = save_imv_evaluation(db, result)
|
||
history_saved = save_imv_placement_history(db, eval_id, result.placement_history)
|
||
except Exception as e:
|
||
logger.exception("avito-imv: save failed for %s", address)
|
||
raise HTTPException(status_code=500, detail=f"save failed: {e}") from e
|
||
|
||
logger.info(
|
||
"avito-imv ok: addr=%s recommended=%d evaluation_id=%d history=%d",
|
||
address[:60],
|
||
result.recommended_price,
|
||
eval_id,
|
||
history_saved,
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"cache_key": result.cache_key,
|
||
"recommended_price": result.recommended_price,
|
||
"range": [result.lower_price, result.higher_price],
|
||
"market_count": result.market_count,
|
||
"evaluation_id": eval_id,
|
||
"history_saved": history_saved,
|
||
}
|
||
|
||
|
||
# ── City sweep: background full-ЕКБ pipeline ────────────────────────────────
|
||
|
||
|
||
class CitySweepStartRequest(BaseModel):
|
||
pages_per_anchor: int = Field(default=3, ge=1, le=10)
|
||
detail_top_n: int = Field(default=10, ge=0, le=30)
|
||
request_delay_sec: float = Field(default=7.0, ge=3.0, le=15.0)
|
||
enrich_houses: bool = True
|
||
radius_m: int = Field(default=1500, ge=500, le=5000)
|
||
|
||
|
||
class CitySweepStartResponse(BaseModel):
|
||
run_id: int
|
||
status: str
|
||
pages_per_anchor: int
|
||
detail_top_n: int
|
||
|
||
|
||
class ScrapeRunRow(BaseModel):
|
||
run_id: int
|
||
source: str
|
||
status: str
|
||
params: dict | None = None
|
||
counters: dict | None = None
|
||
error: str | None = None
|
||
started_at: str | None = None
|
||
finished_at: str | None = None
|
||
heartbeat_at: str | None = None
|
||
|
||
|
||
@router.post("/scrape/avito-city-sweep", response_model=CitySweepStartResponse)
|
||
async def start_avito_city_sweep(
|
||
payload: CitySweepStartRequest,
|
||
background_tasks: BackgroundTasks,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> CitySweepStartResponse:
|
||
"""Запустить full ЕКБ city sweep в background. Returns run_id для polling/cancel.
|
||
|
||
Sweep итерирует 5 anchor-точек ЕКБ × pages_per_anchor страниц Avito,
|
||
сохраняет listings, обогащает houses и detail. Runs in FastAPI BackgroundTasks.
|
||
|
||
Coop cancel: POST /scrape/avito-city-sweep/{run_id}/cancel помечает run,
|
||
pipeline проверяет статус каждый anchor.
|
||
"""
|
||
run_id = runs_mod.create_run(db, source="avito_city_sweep", params=payload.model_dump())
|
||
|
||
async def _sweep_task() -> None:
|
||
sweep_db = SessionLocal()
|
||
try:
|
||
await run_avito_city_sweep(
|
||
sweep_db,
|
||
run_id=run_id,
|
||
radius_m=payload.radius_m,
|
||
pages_per_anchor=payload.pages_per_anchor,
|
||
enrich_houses=payload.enrich_houses,
|
||
detail_top_n=payload.detail_top_n,
|
||
request_delay_sec=payload.request_delay_sec,
|
||
)
|
||
except Exception:
|
||
logger.exception("city-sweep background task run_id=%d crashed", run_id)
|
||
finally:
|
||
sweep_db.close()
|
||
|
||
background_tasks.add_task(_sweep_task)
|
||
logger.info("city-sweep queued run_id=%d params=%s", run_id, payload.model_dump())
|
||
return CitySweepStartResponse(
|
||
run_id=run_id,
|
||
status="running",
|
||
pages_per_anchor=payload.pages_per_anchor,
|
||
detail_top_n=payload.detail_top_n,
|
||
)
|
||
|
||
|
||
@router.get("/scrape/avito-city-sweep/runs", response_model=list[ScrapeRunRow])
|
||
def list_avito_city_sweep_runs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = 10,
|
||
) -> list[ScrapeRunRow]:
|
||
"""Список последних N runs (для UI polling). Default limit=10."""
|
||
rows = runs_mod.list_recent(db, source="avito_city_sweep", limit=limit)
|
||
return [
|
||
ScrapeRunRow(
|
||
run_id=r["run_id"],
|
||
source=r["source"],
|
||
status=r["status"],
|
||
params=r.get("params"),
|
||
counters=r.get("counters"),
|
||
error=r.get("error"),
|
||
started_at=r["started_at"].isoformat() if r.get("started_at") else None,
|
||
finished_at=r["finished_at"].isoformat() if r.get("finished_at") else None,
|
||
heartbeat_at=r["heartbeat_at"].isoformat() if r.get("heartbeat_at") else None,
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.post("/scrape/avito-city-sweep/{run_id}/cancel")
|
||
def cancel_avito_city_sweep(
|
||
run_id: int,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, object]:
|
||
"""Отменить running sweep. Cooperative: pipeline проверяет scrape_runs.status каждый anchor."""
|
||
cancelled = runs_mod.mark_cancelled(db, run_id)
|
||
return {"ok": True, "run_id": run_id, "cancelled": cancelled}
|
||
|
||
|
||
# ── Scraper settings: global + per-source delay management ───────────────────
|
||
|
||
|
||
class ScraperSettingPayload(BaseModel):
|
||
source: str = Field(..., min_length=1, max_length=64)
|
||
request_delay_sec: float = Field(..., ge=0.0, le=60.0)
|
||
description: str | None = None
|
||
|
||
|
||
@router.get("/scraper-settings", status_code=200)
|
||
def list_scraper_settings(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict:
|
||
"""Список всех настроек задержки (per-source + global)."""
|
||
rows = db.execute(
|
||
text("""
|
||
SELECT source, request_delay_sec, description, updated_at
|
||
FROM scraper_settings
|
||
ORDER BY source ASC
|
||
""")
|
||
).mappings().all()
|
||
return {"settings": [dict(r) for r in rows]}
|
||
|
||
|
||
@router.put("/scraper-settings/{source}", status_code=200)
|
||
def update_scraper_setting(
|
||
source: str,
|
||
payload: ScraperSettingPayload,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict:
|
||
"""Обновить задержку для source (или 'global'). Path source используется как ключ.
|
||
|
||
Эффект применяется немедленно — invalidate_cache() сбрасывает кеш для source,
|
||
следующий get_scraper_delay() перечитает из БД.
|
||
|
||
source='global' — нижняя планка для всех парсеров (0 = отключено).
|
||
"""
|
||
if payload.source != source:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"Path source={source!r} does not match body source={payload.source!r}",
|
||
)
|
||
|
||
row = db.execute(
|
||
text("""
|
||
INSERT INTO scraper_settings (source, request_delay_sec, description, updated_at)
|
||
VALUES (:s, CAST(:d AS numeric), :desc, NOW())
|
||
ON CONFLICT (source) DO UPDATE SET
|
||
request_delay_sec = EXCLUDED.request_delay_sec,
|
||
description = COALESCE(EXCLUDED.description, scraper_settings.description),
|
||
updated_at = NOW()
|
||
RETURNING source, request_delay_sec, description, updated_at
|
||
"""),
|
||
{
|
||
"s": source,
|
||
"d": payload.request_delay_sec,
|
||
"desc": payload.description,
|
||
},
|
||
).mappings().one()
|
||
db.commit()
|
||
|
||
invalidate_cache(source)
|
||
logger.info(
|
||
"scraper-settings: updated source=%s delay=%.1f", source, payload.request_delay_sec
|
||
)
|
||
return dict(row)
|
||
|
||
|
||
# ── In-app scheduler endpoints (Stage 4e) ────────────────────────────────────
|
||
|
||
|
||
@router.get("/scrape/schedules", response_model=list[ScheduleConfig])
|
||
def list_schedules(db: Annotated[Session, Depends(get_db)]) -> list[ScheduleConfig]:
|
||
"""Список всех schedules. Сейчас single row 'avito_city_sweep', extensible."""
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, source, enabled, window_start_hour, window_end_hour,
|
||
default_params, last_run_id, last_run_at, next_run_at,
|
||
created_at, updated_at
|
||
FROM scrape_schedules
|
||
ORDER BY source
|
||
"""
|
||
),
|
||
).mappings().all()
|
||
return [
|
||
ScheduleConfig(
|
||
id=r["id"],
|
||
source=r["source"],
|
||
enabled=r["enabled"],
|
||
window_start_hour=r["window_start_hour"],
|
||
window_end_hour=r["window_end_hour"],
|
||
default_params=r["default_params"] or {},
|
||
last_run_id=r["last_run_id"],
|
||
last_run_at=r["last_run_at"].isoformat() if r["last_run_at"] else None,
|
||
next_run_at=r["next_run_at"].isoformat() if r["next_run_at"] else None,
|
||
updated_at=r["updated_at"].isoformat() if r["updated_at"] else None,
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.put("/scrape/schedules/{source}", response_model=ScheduleConfig)
|
||
def update_schedule(
|
||
source: str,
|
||
payload: ScheduleConfigUpdate,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> ScheduleConfig:
|
||
"""UPDATE existing schedule (create если не существует, через INSERT ON CONFLICT)."""
|
||
from app.services.scheduler import compute_next_run_at
|
||
|
||
# Compute new next_run_at если window изменился — recompute, иначе keep existing
|
||
next_at = compute_next_run_at(payload.window_start_hour, payload.window_end_hour)
|
||
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO scrape_schedules
|
||
(source, enabled, window_start_hour, window_end_hour, default_params, next_run_at)
|
||
VALUES (:source, :enabled, :ws, :we, CAST(:params AS jsonb), :next_at)
|
||
ON CONFLICT (source) DO UPDATE SET
|
||
enabled = EXCLUDED.enabled,
|
||
window_start_hour = EXCLUDED.window_start_hour,
|
||
window_end_hour = EXCLUDED.window_end_hour,
|
||
default_params = EXCLUDED.default_params,
|
||
next_run_at = EXCLUDED.next_run_at,
|
||
updated_at = NOW()
|
||
RETURNING id, source, enabled, window_start_hour, window_end_hour,
|
||
default_params, last_run_id, last_run_at, next_run_at, updated_at
|
||
"""
|
||
),
|
||
{
|
||
"source": source,
|
||
"enabled": payload.enabled,
|
||
"ws": payload.window_start_hour,
|
||
"we": payload.window_end_hour,
|
||
"params": json.dumps(payload.default_params, ensure_ascii=False),
|
||
"next_at": next_at,
|
||
},
|
||
).mappings().fetchone()
|
||
db.commit()
|
||
|
||
assert row is not None, f"scrape_schedules upsert returned no row for source={source!r}"
|
||
return ScheduleConfig(
|
||
id=row["id"],
|
||
source=row["source"],
|
||
enabled=row["enabled"],
|
||
window_start_hour=row["window_start_hour"],
|
||
window_end_hour=row["window_end_hour"],
|
||
default_params=row["default_params"] or {},
|
||
last_run_id=row["last_run_id"],
|
||
last_run_at=row["last_run_at"].isoformat() if row["last_run_at"] else None,
|
||
next_run_at=row["next_run_at"].isoformat() if row["next_run_at"] else None,
|
||
updated_at=row["updated_at"].isoformat() if row["updated_at"] else None,
|
||
)
|
||
|
||
|
||
# -- Yandex ad-hoc scrape triggers --------------------------------------------
|
||
|
||
|
||
class YandexDetailTriggerResp(BaseModel):
|
||
ok: bool
|
||
offer_url: str
|
||
offer_id: str | None
|
||
price_rub: int | None
|
||
title: str | None
|
||
photo_count: int
|
||
|
||
|
||
@router.post("/scrape/yandex-detail", response_model=YandexDetailTriggerResp)
|
||
async def scrape_yandex_detail(
|
||
offer_url: str,
|
||
) -> YandexDetailTriggerResp:
|
||
"""Ad-hoc parse one Yandex offer detail page.
|
||
|
||
Returns a snapshot of the extracted DetailEnrichment (no DB write - read-only
|
||
debug; main pipeline writes via estimator on /estimate flow).
|
||
"""
|
||
async with YandexDetailScraper() as scraper:
|
||
result = await scraper.fetch_detail(offer_url)
|
||
if result is None:
|
||
raise HTTPException(404, f"Could not parse Yandex detail page: {offer_url}")
|
||
return YandexDetailTriggerResp(
|
||
ok=True,
|
||
offer_url=offer_url,
|
||
offer_id=result.offer_id,
|
||
price_rub=result.price_rub,
|
||
title=result.title,
|
||
photo_count=len(result.photo_urls),
|
||
)
|
||
|
||
|
||
class YandexNewbuildingTriggerResp(BaseModel):
|
||
ok: bool
|
||
ext_id: str
|
||
ext_slug: str
|
||
name: str | None
|
||
lat: float | None
|
||
lon: float | None
|
||
rating: float | None
|
||
ratings_count: int | None
|
||
text_reviews_count: int | None
|
||
developer_name: str | None
|
||
|
||
|
||
@router.post("/scrape/yandex-newbuilding", response_model=YandexNewbuildingTriggerResp)
|
||
async def scrape_yandex_newbuilding(
|
||
slug: str,
|
||
id: str,
|
||
city: str = "ekaterinburg",
|
||
) -> YandexNewbuildingTriggerResp:
|
||
"""Ad-hoc parse a Yandex JK landing page.
|
||
|
||
URL: /{city}/kupit/novostrojka/<slug>-<id>/
|
||
"""
|
||
async with YandexNewbuildingScraper() as scraper:
|
||
result = await scraper.fetch_jk(jk_slug=slug, jk_id=id, city=city)
|
||
if result is None:
|
||
raise HTTPException(404, f"Could not parse Yandex JK: {slug}-{id} in {city}")
|
||
return YandexNewbuildingTriggerResp(
|
||
ok=True,
|
||
ext_id=result.ext_id,
|
||
ext_slug=result.ext_slug,
|
||
name=result.name,
|
||
lat=result.lat,
|
||
lon=result.lon,
|
||
rating=result.rating,
|
||
ratings_count=result.ratings_count,
|
||
text_reviews_count=result.text_reviews_count,
|
||
developer_name=result.developer_name,
|
||
)
|
||
|
||
|
||
class YandexValuationTriggerResp(BaseModel):
|
||
ok: bool
|
||
address: str
|
||
year_built: int | None
|
||
total_floors: int | None
|
||
house_type: str | None
|
||
has_lift: bool | None
|
||
total_objects: int | None
|
||
history_items_count: int
|
||
|
||
|
||
@router.post("/scrape/yandex-valuation", response_model=YandexValuationTriggerResp)
|
||
async def scrape_yandex_valuation(
|
||
address: str,
|
||
offer_category: str = "APARTMENT",
|
||
offer_type: str = "SELL",
|
||
page: int = 1,
|
||
) -> YandexValuationTriggerResp:
|
||
"""Ad-hoc fetch Yandex Valuation house-history for an address (debug, no cache write).
|
||
|
||
Anonymous GET - works without auth.
|
||
"""
|
||
async with YandexValuationScraper() as scraper:
|
||
result = await scraper.fetch_house_history(
|
||
address=address,
|
||
offer_category=offer_category,
|
||
offer_type=offer_type,
|
||
page=page,
|
||
)
|
||
if result is None:
|
||
raise HTTPException(404, f"Could not parse Yandex Valuation: {address}")
|
||
return YandexValuationTriggerResp(
|
||
ok=True,
|
||
address=address,
|
||
year_built=result.house.year_built,
|
||
total_floors=result.house.total_floors,
|
||
house_type=result.house.house_type,
|
||
has_lift=result.house.has_lift,
|
||
total_objects=result.house.total_objects,
|
||
history_items_count=len(result.history_items),
|
||
)
|