Добавлен 4-й источник N1.ru (ekaterinburg.n1.ru) — региональный портал, хорошо представлен на Урале. DOM-парсинг карточек data-test=offers-list-item через curl_cffi. Multi-room сегментация 1/2/3/4к. cron-scrape.sh переписан под прод: запрос к backend через docker exec (без публичного admin/scrape), все 4 источника (avito/cian/yandex/n1).
158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
"""Admin endpoints — debug + ручной запуск парсеров.
|
||
|
||
Пока без auth (Слой 7), потом закроем общим middleware.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import Annotated, Literal
|
||
|
||
from fastapi import APIRouter, Depends
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import get_db
|
||
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()
|
||
|
||
|
||
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)
|
||
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,
|
||
)
|
||
|
||
|
||
@router.post("/geocode-missing")
|
||
async def geocode_missing(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = 50,
|
||
) -> dict:
|
||
"""Геокодинг listings у которых нет lat/lon (используя address).
|
||
|
||
Запускать после scrape если парсер не достал координаты.
|
||
Nominatim rate-limited 1 req/sec — поэтому небольшой batch.
|
||
"""
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, address, raw_payload
|
||
FROM listings
|
||
WHERE (lat IS NULL OR lon IS NULL)
|
||
AND COALESCE(address, '') != ''
|
||
ORDER BY scraped_at DESC
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{"limit": limit},
|
||
).mappings().all()
|
||
|
||
geocoded = 0
|
||
skipped = 0
|
||
for row in rows:
|
||
result = await geocode(row["address"], db)
|
||
if result is None:
|
||
skipped += 1
|
||
continue
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE listings
|
||
SET lat = :lat, lon = :lon
|
||
WHERE id = :id
|
||
"""
|
||
),
|
||
{"lat": result.lat, "lon": result.lon, "id": row["id"]},
|
||
)
|
||
db.commit()
|
||
geocoded += 1
|
||
|
||
return {"checked": len(rows), "geocoded": geocoded, "skipped": skipped}
|