305 lines
12 KiB
Python
305 lines
12 KiB
Python
"""Admin endpoints — debug + ручной запуск парсеров.
|
||
|
||
Пока без auth (Слой 7), потом закроем общим middleware.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from typing import Annotated, Literal
|
||
|
||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||
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 get_db
|
||
from app.services import cian_session as cian_session_svc
|
||
from app.services.geocoder import geocode
|
||
from app.services.scrapers.avito import AvitoScraper
|
||
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_realty import YandexRealtyScraper
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def require_admin(
|
||
x_admin_token: Annotated[str | None, Header()] = None,
|
||
) -> None:
|
||
"""Проверка admin-токена для /api/v1/admin/*.
|
||
|
||
Если settings.admin_token не задан (dev) — пропускаем (открыто).
|
||
Если задан — требуем заголовок X-Admin-Token с этим значением.
|
||
"""
|
||
if settings.admin_token is None:
|
||
return # dev-режим: токен не настроен → эндпоинты открыты
|
||
if x_admin_token != settings.admin_token:
|
||
raise HTTPException(status_code=401, detail="Invalid or missing X-Admin-Token")
|
||
|
||
|
||
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: ["avito", "cian", "yandex", "n1"])
|
||
multi_room_yandex: bool = False # Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом.
|
||
deep_yandex: bool = False # Если True — Yandex полный обход 5 rooms × 3 sorts × 2 pages = 30 запросов / ~150s.
|
||
multi_room_cian: bool = False # Если True — скрейп Cian по 4 сегментам комнат отдельно (~4× лотов).
|
||
multi_room_n1: bool = False # Если True — скрейп N1 по 4 сегментам комнат отдельно (~4× лотов).
|
||
|
||
|
||
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, dependencies=[Depends(require_admin)])
|
||
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", dependencies=[Depends(require_admin)])
|
||
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, dependencies=[Depends(require_admin)])
|
||
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, dependencies=[Depends(require_admin)])
|
||
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}
|