add interactive parser of наш.дом.рф ЖК+flats with stealth Playwright
- Discovered real endpoints via chrome-devtools: /сервисы/api/kn/object (offset/limit/place/objStatus) + /portal-kn/api/sales/portal/table. Server-side filter ignored — local filter by developer.companyGroup. - Pure Playwright in-context fetch (httpx blocked by ServicePipe TLS-fp). - Saved fingerprint state in data/playwright_state.json for server reuse. - Celery beat (configurable cron + jitter) + admin trigger endpoint + /admin/scrape UI page. - Schema migration: UNIQUE(id, snapshot_date) for versioned snapshots, kn_scrape_runs journal. - Smoke-tested on PRINZIP: 28 ЖК / 826 квартир as expected.
This commit is contained in:
parent
8d3a0874ef
commit
79070bfe99
17 changed files with 1374 additions and 7 deletions
|
|
@ -56,6 +56,10 @@ COPY --from=builder --chown=app:app /app/app /app/app
|
|||
COPY --from=builder --chown=app:app /app/alembic.ini /app/alembic.ini
|
||||
COPY --from=builder --chown=app:app /app/alembic /app/alembic
|
||||
|
||||
# Playwright Chromium for /сервисы/api/kn/* WAF-bootstrap (used only by worker).
|
||||
# --with-deps installs ~80MB of system libs needed by Chromium headless.
|
||||
RUN /app/.venv/bin/playwright install --with-deps chromium
|
||||
|
||||
USER app
|
||||
|
||||
EXPOSE 8000
|
||||
|
|
|
|||
101
backend/app/api/v1/admin_scrape.py
Normal file
101
backend/app/api/v1/admin_scrape.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
"""Admin trigger for ad-hoc kn-API sweeps.
|
||||
|
||||
POST /api/v1/admin/scrape/kn
|
||||
Body: {"region_code": 66, "developers": ["6208_0"]?, "fetch_flats": true?}
|
||||
Header: X-Admin-Token: <SCRAPE_ADMIN_TOKEN env var>
|
||||
|
||||
Disabled if SCRAPE_ADMIN_TOKEN is empty (default).
|
||||
Returns the Celery task id; poll /api/v1/admin/scrape/runs/{run_id} to track DB-side progress.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Any
|
||||
|
||||
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
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class TriggerKnRequest(BaseModel):
|
||||
region_code: int = Field(..., ge=1, le=99)
|
||||
developers: list[str] | None = Field(default=None, max_length=20)
|
||||
fetch_flats: bool = True
|
||||
|
||||
|
||||
def _check_token(x_admin_token: str | None) -> None:
|
||||
if not settings.scrape_admin_token:
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="admin scrape disabled — set SCRAPE_ADMIN_TOKEN",
|
||||
)
|
||||
if x_admin_token != settings.scrape_admin_token:
|
||||
raise HTTPException(status_code=401, detail="invalid admin token")
|
||||
|
||||
|
||||
@router.post("/kn")
|
||||
def trigger_kn_sweep(
|
||||
payload: TriggerKnRequest,
|
||||
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
|
||||
) -> dict[str, Any]:
|
||||
_check_token(x_admin_token)
|
||||
# Defer Celery import so the API works without a broker in dev.
|
||||
from app.workers.tasks.scrape_kn import scrape_kn_region
|
||||
|
||||
result = scrape_kn_region.apply_async(
|
||||
args=[payload.region_code, payload.developers],
|
||||
kwargs={"skip_jitter": True},
|
||||
)
|
||||
return {
|
||||
"task_id": result.id,
|
||||
"region_code": payload.region_code,
|
||||
"developers": payload.developers,
|
||||
"queued_at": result.date_done.isoformat() if result.date_done else None,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/runs")
|
||||
def list_runs(
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
|
||||
limit: int = 20,
|
||||
) -> list[dict[str, Any]]:
|
||||
_check_token(x_admin_token)
|
||||
rows = (
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT run_id, started_at, finished_at, region_codes, developer_ids,
|
||||
objects_count, flats_count, requests_count, status, error, snapshot_date
|
||||
FROM kn_scrape_runs
|
||||
ORDER BY started_at DESC
|
||||
LIMIT :lim
|
||||
"""
|
||||
),
|
||||
{"lim": min(limit, 100)},
|
||||
)
|
||||
.mappings()
|
||||
.all()
|
||||
)
|
||||
return [
|
||||
{
|
||||
"run_id": r["run_id"],
|
||||
"started_at": r["started_at"].isoformat() if r["started_at"] else None,
|
||||
"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
|
||||
"region_codes": list(r["region_codes"]) if r["region_codes"] else [],
|
||||
"developer_ids": list(r["developer_ids"]) if r["developer_ids"] else [],
|
||||
"objects_count": r["objects_count"],
|
||||
"flats_count": r["flats_count"],
|
||||
"requests_count": r["requests_count"],
|
||||
"status": r["status"],
|
||||
"error": r["error"],
|
||||
"snapshot_date": r["snapshot_date"].isoformat() if r["snapshot_date"] else None,
|
||||
}
|
||||
for r in rows
|
||||
]
|
||||
|
|
@ -14,5 +14,21 @@ class Settings(BaseSettings):
|
|||
rosreestr_pkk_base_url: str = "https://pkk.rosreestr.ru/api/features/1"
|
||||
overpass_url: str = "https://overpass-api.de/api/interpreter"
|
||||
|
||||
# Scraper schedule (наш.дом.рф kn-API).
|
||||
# Crontab format: "minute hour day_of_month month day_of_week" (Celery crontab).
|
||||
# Default: каждый понедельник в окне 04:00–05:00 по МСК.
|
||||
scrape_kn_cron: str = "15 4 * * mon"
|
||||
# Random delay window in seconds added on top of scheduled start so the worker
|
||||
# does not hit DOM.РФ at exactly the same minute every cycle. 0 = disabled.
|
||||
scrape_kn_jitter_seconds: int = 1800
|
||||
# Default region(s) for scheduled sweeps. Comma-separated.
|
||||
scrape_kn_default_regions: str = "66"
|
||||
# Path to a pre-captured Playwright storage_state.json (committed in repo,
|
||||
# used by worker to skip cold-start WAF challenge).
|
||||
scrape_kn_state_path: str = "data/playwright_state.json"
|
||||
# Token to authorize ad-hoc /api/v1/admin/scrape/* trigger calls.
|
||||
# Empty string = endpoint disabled.
|
||||
scrape_admin_token: str = ""
|
||||
|
||||
|
||||
settings = Settings()
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import sentry_sdk
|
|||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api.v1 import analytics, concepts, parcels
|
||||
from app.api.v1 import admin_scrape, analytics, concepts, parcels
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
|
|
@ -33,6 +33,7 @@ app.add_middleware(
|
|||
app.include_router(concepts.router, prefix="/api/v1/concepts", tags=["concepts"])
|
||||
app.include_router(parcels.router, prefix="/api/v1/parcels", tags=["parcels"])
|
||||
app.include_router(analytics.router, prefix="/api/v1/analytics", tags=["analytics"])
|
||||
app.include_router(admin_scrape.router, prefix="/api/v1/admin/scrape", tags=["admin"])
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
|
|
|
|||
0
backend/app/services/scrapers/__init__.py
Normal file
0
backend/app/services/scrapers/__init__.py
Normal file
537
backend/app/services/scrapers/domrf_kn.py
Normal file
537
backend/app/services/scrapers/domrf_kn.py
Normal file
|
|
@ -0,0 +1,537 @@
|
|||
"""Scraper for наш.дом.рф kn-API (objects + flats).
|
||||
|
||||
Endpoints discovered via chrome-devtools probe:
|
||||
GET /сервисы/api/kn/object
|
||||
params: offset, limit, place, objStatus, sortField=default, sortType=desc
|
||||
place: "0-4" (city Екатеринбург) or "66" (region Свердл)
|
||||
objStatus: 0=строящиеся, 1=архив-1, 2=сданные (sweep 0..2 for full coverage)
|
||||
Returns {data: {list: [obj, ...], total: N}}.
|
||||
|
||||
GET /portal-kn/api/sales/portal/table?externalId={objId}
|
||||
Per-object flat catalogue: nested entrances → floors → flats.
|
||||
|
||||
Auth header: `Authorization: Basic MTpxd2U=` (decodes to `1:qwe`, public debug
|
||||
auth shipped in their frontend bundle).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
from app.services.scrapers.stealth import BrowserSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PATH_OBJECTS = "/сервисы/api/kn/object"
|
||||
PATH_FLATS_TABLE = "/portal-kn/api/sales/portal/table"
|
||||
RAW_SECTION = "kn_api"
|
||||
|
||||
# Three statuses observed in production (Apr 2026):
|
||||
# 0 — строящиеся ЖК
|
||||
# 1 — небольшая выборка ~2 объекта на регион (legacy/archive flag)
|
||||
# 2 — сданные / завершённые (самый большой бакет)
|
||||
DEFAULT_STATUSES = (0, 1, 2)
|
||||
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _g(d: dict[str, Any], *keys: str, default: Any = None) -> Any:
|
||||
for k in keys:
|
||||
if k in d and d[k] is not None:
|
||||
return d[k]
|
||||
return default
|
||||
|
||||
|
||||
def _to_bool(v: Any) -> bool | None:
|
||||
"""Coerce DOM.РФ 0/1 ints into Python bool. Pass through actual bools/None."""
|
||||
if v is None:
|
||||
return None
|
||||
if isinstance(v, bool):
|
||||
return v
|
||||
if isinstance(v, int | float):
|
||||
return bool(v)
|
||||
if isinstance(v, str):
|
||||
s = v.strip().lower()
|
||||
if s in ("true", "1", "y", "yes"):
|
||||
return True
|
||||
if s in ("false", "0", "n", "no", ""):
|
||||
return False
|
||||
return None
|
||||
|
||||
|
||||
def _to_date(v: Any) -> date | None:
|
||||
"""Coerce date string to date. Accept 'YYYY-MM-DD', 'YYYY-MM-DD HH:MM:SS',
|
||||
'DD-MM-YYYY HH:MM:SS', or quarter-string like 'IV кв. 2028'.
|
||||
Quarter strings get mapped to the last day of the quarter.
|
||||
"""
|
||||
if v is None or v == "":
|
||||
return None
|
||||
if isinstance(v, date) and not isinstance(v, datetime):
|
||||
return v
|
||||
if isinstance(v, datetime):
|
||||
return v.date()
|
||||
if not isinstance(v, str):
|
||||
return None
|
||||
s = v.strip()
|
||||
for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%d-%m-%Y %H:%M:%S", "%d-%m-%Y", "%d.%m.%Y"):
|
||||
try:
|
||||
return datetime.strptime(s, fmt).date()
|
||||
except ValueError:
|
||||
pass
|
||||
# Quarter-like 'IV кв. 2028' / '4 кв. 2028' / 'III квартал 2028'
|
||||
quarter_end = {1: (3, 31), 2: (6, 30), 3: (9, 30), 4: (12, 31)}
|
||||
roman = {"I": 1, "II": 2, "III": 3, "IV": 4}
|
||||
parts = s.replace(".", " ").replace("кв", " ").replace("квартал", " ").split()
|
||||
qq = year = None
|
||||
for p in parts:
|
||||
p = p.strip()
|
||||
if p in roman and qq is None:
|
||||
qq = roman[p]
|
||||
elif p.isdigit():
|
||||
n = int(p)
|
||||
if 1 <= n <= 4 and qq is None:
|
||||
qq = n
|
||||
elif 2000 <= n <= 2050 and year is None:
|
||||
year = n
|
||||
if qq and year:
|
||||
m, d = quarter_end[qq]
|
||||
return date(year, m, d)
|
||||
return None
|
||||
|
||||
|
||||
def _problem_text(v: Any) -> str | None:
|
||||
"""Coerce problem flag (sometimes int 0/1, sometimes text) to text or None."""
|
||||
if v is None:
|
||||
return None
|
||||
if isinstance(v, bool):
|
||||
return "1" if v else "0"
|
||||
if isinstance(v, int | float):
|
||||
return str(int(v))
|
||||
if isinstance(v, str):
|
||||
return v.strip() or None
|
||||
return None
|
||||
|
||||
|
||||
def _extract_total(payload: dict[str, Any]) -> int | None:
|
||||
data = payload.get("data")
|
||||
if isinstance(data, dict):
|
||||
for key in ("total", "totalCount", "count"):
|
||||
v = data.get(key)
|
||||
if isinstance(v, int):
|
||||
return v
|
||||
return None
|
||||
|
||||
|
||||
def _extract_list(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
data = payload.get("data")
|
||||
if isinstance(data, list):
|
||||
return data
|
||||
if isinstance(data, dict):
|
||||
for key in ("list", "items", "objects", "content", "rows"):
|
||||
v = data.get(key)
|
||||
if isinstance(v, list):
|
||||
return v
|
||||
return []
|
||||
|
||||
|
||||
# ── normalisation ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _norm_object(row: dict[str, Any], region_cd: int | None = None) -> dict[str, Any]:
|
||||
"""Map naш.дом.рф /kn/object row → domrf_kn_objects column dict.
|
||||
|
||||
Real API field names:
|
||||
objId, hobjId, developer{devId, shortName, fullName, groupName, companyGroup, devInn},
|
||||
rpdRegionCd, objAddr, shortAddr, objCommercNm, objFloorMin, objFloorMax,
|
||||
objElemLivingCnt, objSquareLiving, objReady100PercDt, objClass, latitude, longitude,
|
||||
objProblemFlg, problemFlag, siteStatus, objGreenHouseFlg, objGuarantyEscrowFlg,
|
||||
objStatus, freeFlatsInfo{priceMin, numberFlats}.
|
||||
"""
|
||||
dev = row.get("developer") if isinstance(row.get("developer"), dict) else {}
|
||||
company_group = _g(dev, "companyGroup") if dev else None
|
||||
# Our DB convention: dev_id="<companyGroup>_0" matches v_developer_full_metrics.
|
||||
dev_id = f"{company_group}_0" if company_group else None
|
||||
return {
|
||||
"obj_id": _g(row, "objId", "obj_id", "id"),
|
||||
"hobj_id": _g(row, "hobjId", "hobj_id"),
|
||||
"comm_name": _g(row, "objCommercNm"),
|
||||
"addr": _g(row, "objAddr"),
|
||||
"short_addr": _g(row, "shortAddr"),
|
||||
"region_cd": _g(row, "rpdRegionCd") or region_cd,
|
||||
"dev_id": dev_id,
|
||||
"dev_name": _g(dev, "groupName") or _g(dev, "shortName") or _g(dev, "fullName"),
|
||||
"dev_inn": _g(dev, "devInn"),
|
||||
"floor_min": _g(row, "objFloorMin"),
|
||||
"floor_max": _g(row, "objFloorMax"),
|
||||
"flat_count": _g(row, "objElemLivingCnt"),
|
||||
"square_living": _g(row, "objSquareLiving"),
|
||||
"ready_dt": _to_date(_g(row, "objReady100PercDt")),
|
||||
"problem_flag": _problem_text(_g(row, "problemFlag", "objProblemFlg")),
|
||||
"site_status": _g(row, "siteStatus"),
|
||||
"green_house": _to_bool(_g(row, "objGreenHouseFlg")),
|
||||
"escrow": _to_bool(_g(row, "objGuarantyEscrowFlg")),
|
||||
"obj_class": _g(row, "objClass"),
|
||||
"wall_type": _g(row, "wallType"),
|
||||
"energy_eff": _g(row, "energyEff"),
|
||||
"latitude": _g(row, "latitude"),
|
||||
"longitude": _g(row, "longitude"),
|
||||
"obj_status": _g(row, "objStatus"),
|
||||
}
|
||||
|
||||
|
||||
def _flatten_table(table_payload: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Iterate nested entrances → floors → flats[] from /sales/portal/table response."""
|
||||
out: list[dict[str, Any]] = []
|
||||
ext_id = table_payload.get("externalId")
|
||||
for entrance in table_payload.get("entrances") or []:
|
||||
for floor in entrance.get("floors") or []:
|
||||
for flat in floor.get("flats") or []:
|
||||
# Inject obj_id and floor for normalisation
|
||||
flat = {
|
||||
**flat,
|
||||
"_objId": ext_id,
|
||||
"_floor": floor.get("floorNumber"),
|
||||
"_entrance": entrance.get("entranceNumber"),
|
||||
}
|
||||
out.append(flat)
|
||||
return out
|
||||
|
||||
|
||||
def _norm_flat(row: dict[str, Any], region_cd: int | None) -> dict[str, Any]:
|
||||
"""Map flat from /sales/portal/table → domrf_kn_flats columns.
|
||||
|
||||
Real fields: flatId|id (numeric, may be null), odsId, elemId (uuid hash),
|
||||
type, number, isStudio, totalArea, livingArea, rooms, status (free|booked|sold),
|
||||
price, pricePerSquareMeter, numberFloors. Plus injected _objId and _floor.
|
||||
"""
|
||||
flat_id = _g(row, "flatId", "id")
|
||||
if flat_id is None:
|
||||
# Fallback: hash of elemId (stable, fits BIGINT). Without an id we cannot UPSERT.
|
||||
elem = _g(row, "elemId")
|
||||
if elem:
|
||||
flat_id = abs(hash(elem)) % (2**63 - 1)
|
||||
return {
|
||||
"id": flat_id,
|
||||
"ods_id": _g(row, "odsId"),
|
||||
"flat_type": _g(row, "type", "flatType"),
|
||||
"flat_number": _g(row, "number", "flatNumber"),
|
||||
"is_studio": _to_bool(_g(row, "isStudio")),
|
||||
"total_area": _g(row, "totalArea"),
|
||||
"living_area": _g(row, "livingArea"),
|
||||
"rooms": _g(row, "rooms"),
|
||||
"status": _g(row, "status"),
|
||||
"price_rub": _g(row, "price"),
|
||||
"price_per_m2": _g(row, "pricePerSquareMeter"),
|
||||
"floor": _g(row, "_floor", "floor"),
|
||||
"num_floors": _g(row, "numberFloors"),
|
||||
"obj_id": _g(row, "_objId", "objId"),
|
||||
"city": _g(row, "city"),
|
||||
"region_cd": region_cd,
|
||||
"obj_name": _g(row, "objName") or _g(row.get("objInfo") or {}, "objCommercNm"),
|
||||
}
|
||||
|
||||
|
||||
# ── DB writes ────────────────────────────────────────────────────────────────
|
||||
|
||||
UPSERT_OBJECT_SQL = text(
|
||||
"""
|
||||
INSERT INTO domrf_kn_objects (
|
||||
obj_id, hobj_id, comm_name, addr, short_addr, region_cd,
|
||||
dev_id, dev_name, dev_inn,
|
||||
floor_min, floor_max, flat_count, square_living, ready_dt,
|
||||
problem_flag, site_status, green_house, escrow,
|
||||
obj_class, wall_type, energy_eff,
|
||||
latitude, longitude, obj_status, snapshot_date
|
||||
) VALUES (
|
||||
:obj_id, :hobj_id, :comm_name, :addr, :short_addr, :region_cd,
|
||||
:dev_id, :dev_name, :dev_inn,
|
||||
:floor_min, :floor_max, :flat_count, :square_living, :ready_dt,
|
||||
:problem_flag, :site_status, :green_house, :escrow,
|
||||
:obj_class, :wall_type, :energy_eff,
|
||||
:latitude, :longitude, :obj_status, :snapshot_date
|
||||
)
|
||||
ON CONFLICT (obj_id, snapshot_date) DO UPDATE SET
|
||||
comm_name = EXCLUDED.comm_name,
|
||||
addr = EXCLUDED.addr,
|
||||
dev_id = EXCLUDED.dev_id,
|
||||
dev_name = EXCLUDED.dev_name,
|
||||
flat_count = EXCLUDED.flat_count,
|
||||
square_living = EXCLUDED.square_living,
|
||||
ready_dt = EXCLUDED.ready_dt,
|
||||
site_status = EXCLUDED.site_status,
|
||||
escrow = EXCLUDED.escrow,
|
||||
obj_class = EXCLUDED.obj_class,
|
||||
latitude = EXCLUDED.latitude,
|
||||
longitude = EXCLUDED.longitude,
|
||||
obj_status = EXCLUDED.obj_status
|
||||
"""
|
||||
)
|
||||
|
||||
UPSERT_FLAT_SQL = text(
|
||||
"""
|
||||
INSERT INTO domrf_kn_flats (
|
||||
id, ods_id, flat_type, flat_number, is_studio, total_area, living_area,
|
||||
rooms, status, price_rub, price_per_m2, floor, num_floors, obj_id,
|
||||
city, region_cd, obj_name, snapshot_date
|
||||
) VALUES (
|
||||
:id, :ods_id, :flat_type, :flat_number, :is_studio, :total_area, :living_area,
|
||||
:rooms, :status, :price_rub, :price_per_m2, :floor, :num_floors, :obj_id,
|
||||
:city, :region_cd, :obj_name, :snapshot_date
|
||||
)
|
||||
ON CONFLICT (id, snapshot_date) DO UPDATE SET
|
||||
status = EXCLUDED.status,
|
||||
price_rub = EXCLUDED.price_rub,
|
||||
price_per_m2 = EXCLUDED.price_per_m2,
|
||||
obj_id = EXCLUDED.obj_id,
|
||||
region_cd = EXCLUDED.region_cd,
|
||||
obj_name = EXCLUDED.obj_name
|
||||
"""
|
||||
)
|
||||
|
||||
INSERT_RAW_SQL = text(
|
||||
"""
|
||||
INSERT INTO domrf_raw_endpoints (
|
||||
snapshot_date, section, endpoint, source_url, payload, payload_size
|
||||
) VALUES (
|
||||
:snapshot_date, :section, :endpoint, :source_url,
|
||||
CAST(:payload AS JSONB), :payload_size
|
||||
)
|
||||
ON CONFLICT (snapshot_date, section, endpoint) DO UPDATE SET
|
||||
payload = EXCLUDED.payload,
|
||||
payload_size = EXCLUDED.payload_size,
|
||||
source_url = EXCLUDED.source_url
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def _insert_raw(db: Session, snapshot_date: date, endpoint_label: str, payload: Any) -> None:
|
||||
body = json.dumps(payload, ensure_ascii=False)
|
||||
db.execute(
|
||||
INSERT_RAW_SQL,
|
||||
{
|
||||
"snapshot_date": snapshot_date,
|
||||
"section": RAW_SECTION,
|
||||
"endpoint": endpoint_label[:255],
|
||||
"source_url": f"https://xn--80az8a.xn--d1aqf.xn--p1ai{endpoint_label}",
|
||||
"payload": body,
|
||||
"payload_size": len(body.encode("utf-8")),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def upsert_objects(
|
||||
db: Session, raw: list[dict[str, Any]], snapshot_date: date, region_cd: int
|
||||
) -> int:
|
||||
inserted = 0
|
||||
for row in raw:
|
||||
norm = _norm_object(row, region_cd=region_cd)
|
||||
if not norm["obj_id"]:
|
||||
continue
|
||||
norm["snapshot_date"] = snapshot_date
|
||||
try:
|
||||
db.execute(UPSERT_OBJECT_SQL, norm)
|
||||
inserted += 1
|
||||
except Exception as e:
|
||||
logger.warning("upsert object %s failed: %s", norm.get("obj_id"), e)
|
||||
db.rollback()
|
||||
return inserted
|
||||
|
||||
|
||||
def upsert_flats(
|
||||
db: Session, raw: list[dict[str, Any]], snapshot_date: date, region_cd: int
|
||||
) -> int:
|
||||
inserted = 0
|
||||
skipped_no_id = 0
|
||||
for row in raw:
|
||||
norm = _norm_flat(row, region_cd)
|
||||
if norm["id"] is None:
|
||||
skipped_no_id += 1
|
||||
continue
|
||||
norm["snapshot_date"] = snapshot_date
|
||||
try:
|
||||
db.execute(UPSERT_FLAT_SQL, norm)
|
||||
inserted += 1
|
||||
except Exception as e:
|
||||
logger.warning("upsert flat %s failed: %s", norm.get("id"), e)
|
||||
db.rollback()
|
||||
if skipped_no_id:
|
||||
logger.info("skipped %d flats without numeric id", skipped_no_id)
|
||||
return inserted
|
||||
|
||||
|
||||
# ── orchestrator ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _place_str(region_code: int) -> str:
|
||||
"""Map region code → DOM.РФ `place` parameter. By default uses the regionCode itself
|
||||
(e.g. 66=Свердл region). City-level codes (e.g. '0-4'=ЕКБ) can be passed as-is via
|
||||
PLACE_OVERRIDE if more granularity is needed.
|
||||
"""
|
||||
return str(region_code)
|
||||
|
||||
|
||||
async def fetch_objects_for_status(
|
||||
sess: BrowserSession, place: str, status: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Fetch ALL objects for a given (place, objStatus) — server returns up to limit=999999."""
|
||||
payload = await sess.get_json(
|
||||
PATH_OBJECTS,
|
||||
{
|
||||
"offset": 0,
|
||||
"limit": 999999,
|
||||
"sortField": "default",
|
||||
"sortType": "desc",
|
||||
"place": place,
|
||||
"objStatus": status,
|
||||
},
|
||||
)
|
||||
rows = _extract_list(payload)
|
||||
total = _extract_total(payload)
|
||||
logger.info("kn/object place=%s status=%d -> %d/%s rows", place, status, len(rows), total)
|
||||
return rows
|
||||
|
||||
|
||||
async def fetch_flats_for_object(sess: BrowserSession, obj_id: int) -> list[dict[str, Any]]:
|
||||
"""Fetch flat-table for one object, return flat rows (entrance/floor flattened)."""
|
||||
try:
|
||||
payload = await sess.get_json(PATH_FLATS_TABLE, {"externalId": obj_id})
|
||||
except Exception as e:
|
||||
logger.warning("flats fetch obj=%s failed: %s", obj_id, e)
|
||||
return []
|
||||
# Body shape: {externalId, entrances: [{entranceNumber, floors:[{floorNumber, flats:[...]}]}]}
|
||||
return _flatten_table(payload)
|
||||
|
||||
|
||||
async def run_region_sweep(
|
||||
region_code: int,
|
||||
developers: list[str] | None = None,
|
||||
headed: bool = False,
|
||||
snapshot_date: date | None = None,
|
||||
load_state: str | None = None,
|
||||
save_state: str | None = None,
|
||||
statuses: tuple[int, ...] = DEFAULT_STATUSES,
|
||||
fetch_flats: bool = True,
|
||||
place_override: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Full sweep for one region: bootstrap → fetch objects (each objStatus) → fetch flats.
|
||||
|
||||
`developers` filters the OBJECT set locally by `developer.companyGroup` matching the
|
||||
"<group>_0" id pattern (since DOM.РФ doesn't honour developer filter on the API).
|
||||
"""
|
||||
snapshot_date = snapshot_date or date.today()
|
||||
db = SessionLocal()
|
||||
run_id = db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO kn_scrape_runs (region_codes, developer_ids, snapshot_date, status)
|
||||
VALUES (ARRAY[:rc]::int[], CAST(:devs AS text[]), :snap, 'running')
|
||||
RETURNING run_id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"rc": region_code,
|
||||
"devs": developers if developers else None,
|
||||
"snap": snapshot_date,
|
||||
},
|
||||
).scalar_one()
|
||||
db.commit()
|
||||
|
||||
place = place_override or _place_str(region_code)
|
||||
|
||||
try:
|
||||
async with BrowserSession(
|
||||
region_code=region_code,
|
||||
headed=headed,
|
||||
load_state=load_state,
|
||||
save_state=save_state,
|
||||
) as sess:
|
||||
# --- objects across statuses -------------------------------------
|
||||
all_objects: list[dict[str, Any]] = []
|
||||
for status in statuses:
|
||||
rows = await fetch_objects_for_status(sess, place, status)
|
||||
all_objects.extend(rows)
|
||||
logger.info("place=%s total objects across %s = %d", place, statuses, len(all_objects))
|
||||
|
||||
# local developer filter
|
||||
if developers:
|
||||
wanted_groups = {int(d.split("_")[0]) for d in developers if "_" in d}
|
||||
all_objects = [
|
||||
o
|
||||
for o in all_objects
|
||||
if isinstance(o.get("developer"), dict)
|
||||
and o["developer"].get("companyGroup") in wanted_groups
|
||||
]
|
||||
logger.info("after developers filter: %d objects", len(all_objects))
|
||||
|
||||
# --- flats per object --------------------------------------------
|
||||
all_flats: list[dict[str, Any]] = []
|
||||
if fetch_flats:
|
||||
for i, obj in enumerate(all_objects):
|
||||
obj_id = obj.get("objId") or obj.get("obj_id")
|
||||
if not obj_id:
|
||||
continue
|
||||
flats = await fetch_flats_for_object(sess, obj_id)
|
||||
all_flats.extend(flats)
|
||||
if (i + 1) % 25 == 0:
|
||||
logger.info(
|
||||
"flats progress: %d/%d objs, %d flats so far",
|
||||
i + 1,
|
||||
len(all_objects),
|
||||
len(all_flats),
|
||||
)
|
||||
logger.info("total flats: %d (across %d objs)", len(all_flats), len(all_objects))
|
||||
|
||||
request_count = sess.request_count
|
||||
|
||||
# --- write all in one transaction ------------------------------------
|
||||
obj_count = upsert_objects(db, all_objects, snapshot_date, region_code)
|
||||
flat_count = upsert_flats(db, all_flats, snapshot_date, region_code) if fetch_flats else 0
|
||||
# Save raw payloads (one row per status; flats are huge so we store the count, not bodies).
|
||||
_insert_raw(db, snapshot_date, f"kn_object_place_{place}", all_objects)
|
||||
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
UPDATE kn_scrape_runs
|
||||
SET finished_at=NOW(), objects_count=:oc, flats_count=:fc,
|
||||
requests_count=:rq, status='done'
|
||||
WHERE run_id=:rid
|
||||
"""
|
||||
),
|
||||
{"oc": obj_count, "fc": flat_count, "rq": request_count, "rid": run_id},
|
||||
)
|
||||
db.commit()
|
||||
return {
|
||||
"run_id": run_id,
|
||||
"region_code": region_code,
|
||||
"place": place,
|
||||
"objects": obj_count,
|
||||
"flats": flat_count,
|
||||
"requests": request_count,
|
||||
"snapshot_date": snapshot_date.isoformat(),
|
||||
}
|
||||
except Exception as e:
|
||||
logger.exception("kn sweep failed for region=%d", region_code)
|
||||
db.rollback()
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
UPDATE kn_scrape_runs
|
||||
SET finished_at=NOW(), status='failed', error=:err
|
||||
WHERE run_id=:rid
|
||||
"""
|
||||
),
|
||||
{"err": str(e)[:2000], "rid": run_id},
|
||||
)
|
||||
db.commit()
|
||||
raise
|
||||
finally:
|
||||
db.close()
|
||||
182
backend/app/services/scrapers/stealth.py
Normal file
182
backend/app/services/scrapers/stealth.py
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
"""Stealth-fetch utilities for наш.дом.рф scraping.
|
||||
|
||||
Pure-httpx hits ServicePipe WAF JS-challenge even with Playwright-exported
|
||||
cookies (different TLS-fingerprint = blocked). The proven pattern from
|
||||
data/sql/30_scrape_domrf.py is to issue `fetch()` calls *inside* the live
|
||||
Chromium page — same JA3 fingerprint as a real browser, cookies/CSP/CORS
|
||||
all handled natively.
|
||||
|
||||
BrowserSession keeps the browser+page open for the lifetime of a sweep and
|
||||
exposes get_json() that calls page.evaluate(fetch_in_browser).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from playwright.async_api import Browser, BrowserContext, Page, async_playwright
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
BASE_URL = "https://xn--80az8a.xn--d1aqf.xn--p1ai"
|
||||
|
||||
# Real Chrome 147 fingerprint (Win10 x64). Keep in sync with data/sql/30_scrape_domrf.py.
|
||||
USER_AGENT = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36"
|
||||
)
|
||||
|
||||
# Маппинг region_code → URL-сегмент города для реалистичного Referer.
|
||||
# Не обязан быть исчерпывающим — fallback на /новостройки/строящиеся/.
|
||||
REGION_LANDING_PATH = {
|
||||
66: "/новостройки/строящиеся/екатеринбург/",
|
||||
77: "/новостройки/строящиеся/москва/",
|
||||
78: "/новостройки/строящиеся/санкт-петербург/",
|
||||
50: "/новостройки/строящиеся/московская-область/",
|
||||
47: "/новостройки/строящиеся/ленинградская-область/",
|
||||
}
|
||||
|
||||
|
||||
def make_referer(region_code: int) -> str:
|
||||
path = REGION_LANDING_PATH.get(region_code, "/новостройки/строящиеся/")
|
||||
return BASE_URL + path
|
||||
|
||||
|
||||
async def jitter_sleep(min_ms: int = 600, max_ms: int = 1500) -> None:
|
||||
delay = random.uniform(min_ms / 1000, max_ms / 1000)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
|
||||
# JS executed inside the live page. Returns {ok, status, body, contentType}.
|
||||
# Auth header is hardcoded `Basic MTpxd2U=` (decodes to `1:qwe`, the public
|
||||
# debug auth shipped in the site's frontend bundle for the kn API).
|
||||
_FETCH_JS = """
|
||||
async ({url, auth}) => {
|
||||
const headers = {'Accept': 'application/json, text/plain, */*'};
|
||||
if (auth) headers['Authorization'] = auth;
|
||||
try {
|
||||
const r = await fetch(url, {credentials: 'include', headers});
|
||||
const ctype = r.headers.get('content-type') || '';
|
||||
const body = await r.text();
|
||||
return {ok: r.ok, status: r.status, body, contentType: ctype};
|
||||
} catch (e) {
|
||||
return {ok: false, status: 0, body: String(e), contentType: ''};
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
class WafBlockedError(RuntimeError):
|
||||
"""Server returned non-JSON (likely ServicePipe JS-challenge)."""
|
||||
|
||||
|
||||
class BrowserSession:
|
||||
"""Keeps a live Playwright Chromium page; issues kn-API fetches inside it.
|
||||
|
||||
Use as async context manager. Concurrency is bounded by an asyncio.Semaphore
|
||||
of size 3 (the same page can serve multiple concurrent fetches via JS-await).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
region_code: int = 66,
|
||||
headed: bool = False,
|
||||
auth: str | None = "Basic MTpxd2U=",
|
||||
load_state: str | None = None,
|
||||
save_state: str | None = None,
|
||||
) -> None:
|
||||
self.region_code = region_code
|
||||
self.headed = headed
|
||||
self.auth = auth
|
||||
# Path to a Playwright storage_state JSON (cookies + localStorage).
|
||||
# If load_state is set, the context is created with these cookies preloaded —
|
||||
# ServicePipe sees a returning user, no fresh JS-challenge.
|
||||
# If save_state is set, the post-bootstrap state is dumped to that path
|
||||
# so it can be committed and reused on a server.
|
||||
self.load_state = load_state
|
||||
self.save_state = save_state
|
||||
self._pw: Any = None
|
||||
self._browser: Browser | None = None
|
||||
self._context: BrowserContext | None = None
|
||||
self._page: Page | None = None
|
||||
self._sem = asyncio.Semaphore(3)
|
||||
self._request_count = 0
|
||||
|
||||
async def __aenter__(self) -> BrowserSession:
|
||||
await self._bootstrap()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc: Any) -> None:
|
||||
if self._browser is not None:
|
||||
await self._browser.close()
|
||||
if self._pw is not None:
|
||||
await self._pw.stop()
|
||||
|
||||
async def _bootstrap(self) -> None:
|
||||
landing_url = make_referer(self.region_code)
|
||||
logger.info("bootstrap: opening %s", landing_url)
|
||||
self._pw = await async_playwright().start()
|
||||
self._browser = await self._pw.chromium.launch(headless=not self.headed)
|
||||
ctx_kwargs: dict[str, Any] = {
|
||||
"user_agent": USER_AGENT,
|
||||
"locale": "ru-RU",
|
||||
"viewport": {"width": 1920, "height": 1080},
|
||||
}
|
||||
if self.load_state:
|
||||
ctx_kwargs["storage_state"] = self.load_state
|
||||
logger.info("bootstrap: loading saved storage_state from %s", self.load_state)
|
||||
self._context = await self._browser.new_context(**ctx_kwargs)
|
||||
self._page = await self._context.new_page()
|
||||
await self._page.goto(landing_url, wait_until="domcontentloaded", timeout=30_000)
|
||||
try:
|
||||
await self._page.wait_for_load_state("networkidle", timeout=10_000)
|
||||
except Exception:
|
||||
pass
|
||||
if self.save_state:
|
||||
await self._context.storage_state(path=self.save_state)
|
||||
logger.info("bootstrap: storage_state saved to %s", self.save_state)
|
||||
logger.info("bootstrap: page ready, WAF challenge passed")
|
||||
|
||||
async def get_json(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Fetch JSON via fetch() in browser. Retries with backoff on transient errors."""
|
||||
if self._page is None:
|
||||
raise RuntimeError("BrowserSession not bootstrapped")
|
||||
url = BASE_URL + path + "?" + urlencode(params)
|
||||
last_err: Exception | None = None
|
||||
for attempt in range(5):
|
||||
async with self._sem:
|
||||
await jitter_sleep()
|
||||
try:
|
||||
self._request_count += 1
|
||||
result = await self._page.evaluate(_FETCH_JS, {"url": url, "auth": self.auth})
|
||||
except Exception as e:
|
||||
last_err = e
|
||||
logger.warning("evaluate err attempt=%d: %r", attempt, e)
|
||||
await asyncio.sleep(2**attempt)
|
||||
continue
|
||||
status = result.get("status", 0)
|
||||
body = result.get("body", "")
|
||||
ctype = result.get("contentType", "")
|
||||
if status in (429,) or status >= 500 or status == 0:
|
||||
last_err = RuntimeError(f"transient status={status}")
|
||||
logger.warning("transient status=%d attempt=%d, backing off", status, attempt)
|
||||
await asyncio.sleep(2**attempt)
|
||||
continue
|
||||
if "application/json" not in ctype:
|
||||
# WAF returned the JS challenge HTML; re-bootstrap could help, but for now fail.
|
||||
raise WafBlockedError(
|
||||
f"non-JSON response: status={status} ctype={ctype} body[:120]={body[:120]!r}"
|
||||
)
|
||||
if status != 200:
|
||||
raise RuntimeError(f"http {status}: {body[:200]}")
|
||||
return json.loads(body)
|
||||
raise RuntimeError(f"max retries exhausted: {last_err!r}")
|
||||
|
||||
@property
|
||||
def request_count(self) -> int:
|
||||
return self._request_count
|
||||
|
|
@ -1,11 +1,46 @@
|
|||
"""Celery app + beat schedule.
|
||||
|
||||
Stage 2a: daily PKK Rosreestr sync.
|
||||
"""
|
||||
"""Celery app + beat schedule."""
|
||||
|
||||
from celery import Celery
|
||||
from celery.schedules import crontab
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
celery_app = Celery("gendesign", broker=settings.redis_url, backend=settings.redis_url)
|
||||
|
||||
def _parse_cron(spec: str) -> crontab:
|
||||
"""Parse 'M H DoM Mon DoW' crontab string into Celery crontab. Empty fields default to '*'."""
|
||||
parts = spec.strip().split()
|
||||
if len(parts) != 5:
|
||||
raise ValueError(f"crontab spec must have 5 fields, got: {spec!r}")
|
||||
minute, hour, dom, month, dow = parts
|
||||
return crontab(
|
||||
minute=minute,
|
||||
hour=hour,
|
||||
day_of_month=dom,
|
||||
month_of_year=month,
|
||||
day_of_week=dow,
|
||||
)
|
||||
|
||||
|
||||
def _default_regions() -> list[int]:
|
||||
return [int(x.strip()) for x in settings.scrape_kn_default_regions.split(",") if x.strip()]
|
||||
|
||||
|
||||
celery_app = Celery(
|
||||
"gendesign",
|
||||
broker=settings.redis_url,
|
||||
backend=settings.redis_url,
|
||||
include=["app.workers.tasks.scrape_kn"],
|
||||
)
|
||||
celery_app.conf.timezone = "Europe/Moscow"
|
||||
|
||||
# Расписание задаётся через SCRAPE_KN_CRON env var (см. app.core.config).
|
||||
# Каждый региональный sweep оборачивается случайной задержкой 0..SCRAPE_KN_JITTER_SECONDS
|
||||
# внутри самой задачи (см. tasks.scrape_kn) — чтобы не бить ровно в одну минуту.
|
||||
celery_app.conf.beat_schedule = {
|
||||
f"kn-region-{rc}": {
|
||||
"task": "tasks.scrape_kn.scrape_kn_region",
|
||||
"schedule": _parse_cron(settings.scrape_kn_cron),
|
||||
"args": [rc, None],
|
||||
}
|
||||
for rc in _default_regions()
|
||||
}
|
||||
|
|
|
|||
0
backend/app/workers/tasks/__init__.py
Normal file
0
backend/app/workers/tasks/__init__.py
Normal file
52
backend/app/workers/tasks/scrape_kn.py
Normal file
52
backend/app/workers/tasks/scrape_kn.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""Celery task wrapper for naш.дом.рф kn-API sweeps."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.core.config import settings
|
||||
from app.services.scrapers.domrf_kn import run_region_sweep
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@celery_app.task(bind=True, name="tasks.scrape_kn.scrape_kn_region", max_retries=2)
|
||||
def scrape_kn_region(
|
||||
self: Any,
|
||||
region_code: int,
|
||||
developers: list[str] | None = None,
|
||||
skip_jitter: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Run a full kn-API sweep for one region. Returns summary dict.
|
||||
|
||||
For scheduled runs (skip_jitter=False), sleeps a random 0..SCRAPE_KN_JITTER_SECONDS
|
||||
interval before starting so multiple workers / repeated weekly cycles don't hit
|
||||
DOM.РФ at exactly the same minute. Manual triggers pass skip_jitter=True for
|
||||
immediate execution.
|
||||
"""
|
||||
if not skip_jitter and settings.scrape_kn_jitter_seconds > 0:
|
||||
delay = random.uniform(0, settings.scrape_kn_jitter_seconds)
|
||||
logger.info("jitter delay %.0fs before scrape", delay)
|
||||
time.sleep(delay)
|
||||
|
||||
state_path: str | None = settings.scrape_kn_state_path or None
|
||||
if state_path and not Path(state_path).exists():
|
||||
logger.warning("storage_state %s not found — bootstrap will run cold", state_path)
|
||||
state_path = None
|
||||
|
||||
logger.info(
|
||||
"scrape_kn_region start region=%s devs=%s state=%s", region_code, developers, state_path
|
||||
)
|
||||
return asyncio.run(
|
||||
run_region_sweep(
|
||||
region_code=region_code,
|
||||
developers=developers,
|
||||
load_state=state_path,
|
||||
)
|
||||
)
|
||||
|
|
@ -18,6 +18,8 @@ dependencies = [
|
|||
"httpx>=0.27.0",
|
||||
"redis>=5.0.0",
|
||||
"celery>=5.4.0",
|
||||
"playwright>=1.45.0",
|
||||
"tenacity>=9.0.0",
|
||||
"weasyprint>=62.0",
|
||||
"ezdxf>=1.3.0",
|
||||
"openpyxl>=3.1.0",
|
||||
|
|
|
|||
1
data/playwright_state.json
Normal file
1
data/playwright_state.json
Normal file
File diff suppressed because one or more lines are too long
47
data/sql/50_schema_kn_extensions.sql
Normal file
47
data/sql/50_schema_kn_extensions.sql
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
-- Расширения схемы для регулярного парсинга наш.дом.рф /сервисы/api/kn/{flats,objects}.
|
||||
-- См. план C:\Users\user\.claude\plans\crispy-waddling-swing.md.
|
||||
|
||||
-- ── Версионирование строк через UNIQUE (id, snapshot_date) ────────────────
|
||||
-- Текущий PK на flats.id и objects.obj_id мешает сохранять несколько срезов
|
||||
-- одного и того же ЖК/квартиры. Меняем на UNIQUE по (id, snapshot_date)
|
||||
-- чтобы UPSERT идемпотентно записывал свежий snapshot.
|
||||
|
||||
ALTER TABLE domrf_kn_flats DROP CONSTRAINT IF EXISTS domrf_kn_flats_pkey;
|
||||
ALTER TABLE domrf_kn_flats
|
||||
ADD CONSTRAINT uq_kn_flats_id_snap UNIQUE (id, snapshot_date);
|
||||
|
||||
ALTER TABLE domrf_kn_objects DROP CONSTRAINT IF EXISTS domrf_kn_objects_pkey;
|
||||
ALTER TABLE domrf_kn_objects
|
||||
ADD CONSTRAINT uq_kn_objects_id_snap UNIQUE (obj_id, snapshot_date);
|
||||
|
||||
-- ── Backfill region_cd для Антоновых flats где obj_id матчится с Свердл ──
|
||||
UPDATE domrf_kn_flats f
|
||||
SET region_cd = 66
|
||||
FROM domrf_kn_objects o
|
||||
WHERE f.obj_id = o.obj_id
|
||||
AND o.region_cd = 66
|
||||
AND f.region_cd IS NULL;
|
||||
|
||||
-- ── Композитные индексы для выборок последнего среза ──────────────────────
|
||||
CREATE INDEX IF NOT EXISTS idx_kn_flats_region_snap
|
||||
ON domrf_kn_flats (region_cd, snapshot_date);
|
||||
CREATE INDEX IF NOT EXISTS idx_kn_objects_region_snap
|
||||
ON domrf_kn_objects (region_cd, snapshot_date);
|
||||
|
||||
-- ── Журнал прогонов скрейпера ─────────────────────────────────────────────
|
||||
CREATE TABLE IF NOT EXISTS kn_scrape_runs (
|
||||
run_id BIGSERIAL PRIMARY KEY,
|
||||
started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
finished_at TIMESTAMPTZ,
|
||||
region_codes INT[] NOT NULL,
|
||||
developer_ids TEXT[],
|
||||
objects_count INT,
|
||||
flats_count INT,
|
||||
requests_count INT,
|
||||
status TEXT NOT NULL DEFAULT 'running', -- running | done | failed
|
||||
error TEXT,
|
||||
snapshot_date DATE NOT NULL DEFAULT CURRENT_DATE
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_kn_runs_status ON kn_scrape_runs (status, started_at DESC);
|
||||
|
||||
COMMENT ON TABLE kn_scrape_runs IS 'Журнал запусков парсера наш.дом.рф kn-API. Один прогон = один (regions, developers) sweep.';
|
||||
94
data/sql/50_scrape_kn_flats.py
Normal file
94
data/sql/50_scrape_kn_flats.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""CLI for ad-hoc naш.дом.рф kn-API sweeps.
|
||||
|
||||
Usage:
|
||||
PYTHONIOENCODING=utf-8 python data/sql/50_scrape_kn_flats.py --regions 66
|
||||
python data/sql/50_scrape_kn_flats.py --regions 66 --developers 6208_0 --headed
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# allow `from app.services.scrapers.domrf_kn import ...` when run from repo root
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(ROOT / "backend"))
|
||||
|
||||
from app.services.scrapers.domrf_kn import run_region_sweep # noqa: E402
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="naш.дом.рф kn-API scraper (CLI)")
|
||||
parser.add_argument(
|
||||
"--regions",
|
||||
type=int,
|
||||
nargs="+",
|
||||
required=True,
|
||||
help="Region codes (1..89). e.g. --regions 66 77",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--developers",
|
||||
type=str,
|
||||
nargs="*",
|
||||
help="Optional developer IDs in '<id>_0' format. e.g. --developers 6208_0",
|
||||
)
|
||||
parser.add_argument("--headed", action="store_true", help="Show Chromium window during bootstrap")
|
||||
parser.add_argument(
|
||||
"--save-state",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Path to dump Playwright storage_state.json after bootstrap (commit to git for server reuse)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--load-state",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Path to load Playwright storage_state.json (server uses fingerprint captured locally)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--bootstrap-only",
|
||||
action="store_true",
|
||||
help="Just bootstrap a session and exit — useful for capturing storage_state without scraping",
|
||||
)
|
||||
parser.add_argument("--verbose", "-v", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||||
)
|
||||
|
||||
if not os.environ.get("DATABASE_URL"):
|
||||
print("WARNING: DATABASE_URL not set; using default from app.core.config")
|
||||
|
||||
async def _run() -> None:
|
||||
if args.bootstrap_only:
|
||||
from app.services.scrapers.stealth import BrowserSession
|
||||
async with BrowserSession(
|
||||
region_code=args.regions[0],
|
||||
headed=args.headed,
|
||||
load_state=args.load_state,
|
||||
save_state=args.save_state,
|
||||
):
|
||||
print(f"bootstrap-only: state saved to {args.save_state}")
|
||||
return
|
||||
|
||||
for region in args.regions:
|
||||
summary = await run_region_sweep(
|
||||
region_code=region,
|
||||
developers=args.developers or None,
|
||||
headed=args.headed,
|
||||
load_state=args.load_state,
|
||||
save_state=args.save_state if region == args.regions[0] else None,
|
||||
)
|
||||
print(f"region={region} -> {summary}")
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -54,6 +54,28 @@ services:
|
|||
depends_on:
|
||||
- backend
|
||||
|
||||
worker:
|
||||
build: ./backend
|
||||
env_file: ./backend/.env
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
redis:
|
||||
condition: service_started
|
||||
volumes:
|
||||
- ./backend/app:/app/app
|
||||
command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=2"]
|
||||
|
||||
beat:
|
||||
build: ./backend
|
||||
env_file: ./backend/.env
|
||||
depends_on:
|
||||
redis:
|
||||
condition: service_started
|
||||
volumes:
|
||||
- ./backend/app:/app/app
|
||||
command: ["celery", "-A", "app.workers.celery_app", "beat", "--loglevel=info"]
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
redis_data:
|
||||
|
|
|
|||
273
frontend/src/app/admin/scrape/page.tsx
Normal file
273
frontend/src/app/admin/scrape/page.tsx
Normal file
|
|
@ -0,0 +1,273 @@
|
|||
"use client";
|
||||
|
||||
import { useState } from "react";
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
|
||||
import { apiFetch } from "@/lib/api";
|
||||
|
||||
interface RunRow {
|
||||
run_id: number;
|
||||
started_at: string | null;
|
||||
finished_at: string | null;
|
||||
region_codes: number[];
|
||||
developer_ids: string[];
|
||||
objects_count: number | null;
|
||||
flats_count: number | null;
|
||||
requests_count: number | null;
|
||||
status: string;
|
||||
error: string | null;
|
||||
snapshot_date: string | null;
|
||||
}
|
||||
|
||||
interface TriggerResp {
|
||||
task_id: string;
|
||||
region_code: number;
|
||||
developers: string[] | null;
|
||||
}
|
||||
|
||||
export default function ScrapeAdminPage() {
|
||||
const [token, setToken] = useState<string>(() =>
|
||||
typeof window === "undefined"
|
||||
? ""
|
||||
: (localStorage.getItem("admin_token") ?? ""),
|
||||
);
|
||||
const [region, setRegion] = useState(66);
|
||||
const [devs, setDevs] = useState("");
|
||||
const [fetchFlats, setFetchFlats] = useState(true);
|
||||
const queryClient = useQueryClient();
|
||||
|
||||
const runs = useQuery({
|
||||
queryKey: ["scrape-runs", token],
|
||||
queryFn: () =>
|
||||
apiFetch<RunRow[]>("/api/v1/admin/scrape/runs?limit=20", {
|
||||
headers: { "X-Admin-Token": token },
|
||||
}),
|
||||
enabled: !!token,
|
||||
refetchInterval: 5000,
|
||||
});
|
||||
|
||||
const trigger = useMutation({
|
||||
mutationFn: () =>
|
||||
apiFetch<TriggerResp>("/api/v1/admin/scrape/kn", {
|
||||
method: "POST",
|
||||
headers: { "X-Admin-Token": token },
|
||||
body: JSON.stringify({
|
||||
region_code: region,
|
||||
developers: devs
|
||||
? devs
|
||||
.split(",")
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean)
|
||||
: null,
|
||||
fetch_flats: fetchFlats,
|
||||
}),
|
||||
}),
|
||||
onSuccess: () =>
|
||||
queryClient.invalidateQueries({ queryKey: ["scrape-runs"] }),
|
||||
});
|
||||
|
||||
const saveToken = (v: string) => {
|
||||
setToken(v);
|
||||
localStorage.setItem("admin_token", v);
|
||||
};
|
||||
|
||||
return (
|
||||
<main style={{ padding: 24, maxWidth: 1100, margin: "0 auto" }}>
|
||||
<h1 style={{ marginTop: 0 }}>Скрапер наш.дом.рф — ручной запуск</h1>
|
||||
<p style={{ color: "#5b6066", fontSize: 13 }}>
|
||||
Запуск sweep-а вне расписания (расписание: см. SCRAPE_KN_CRON env, по
|
||||
умолчанию пн 04:15 МСК с jitter 0–30мин).
|
||||
</p>
|
||||
|
||||
<section style={cardStyle}>
|
||||
<label style={{ display: "block", marginBottom: 12 }}>
|
||||
<span style={labelStyle}>Admin Token</span>
|
||||
<input
|
||||
type="password"
|
||||
value={token}
|
||||
onChange={(e) => saveToken(e.target.value)}
|
||||
placeholder="SCRAPE_ADMIN_TOKEN"
|
||||
style={inputStyle}
|
||||
/>
|
||||
</label>
|
||||
<div style={{ display: "flex", gap: 16, flexWrap: "wrap" }}>
|
||||
<label>
|
||||
<span style={labelStyle}>Регион</span>
|
||||
<input
|
||||
type="number"
|
||||
value={region}
|
||||
min={1}
|
||||
max={99}
|
||||
onChange={(e) => setRegion(Number(e.target.value))}
|
||||
style={{ ...inputStyle, width: 100 }}
|
||||
/>
|
||||
</label>
|
||||
<label style={{ flex: 1 }}>
|
||||
<span style={labelStyle}>Девелоперы (csv, опц.)</span>
|
||||
<input
|
||||
value={devs}
|
||||
onChange={(e) => setDevs(e.target.value)}
|
||||
placeholder="6208_0,5791_0"
|
||||
style={inputStyle}
|
||||
/>
|
||||
</label>
|
||||
<label
|
||||
style={{
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
gap: 6,
|
||||
marginTop: 22,
|
||||
}}
|
||||
>
|
||||
<input
|
||||
type="checkbox"
|
||||
checked={fetchFlats}
|
||||
onChange={(e) => setFetchFlats(e.target.checked)}
|
||||
/>
|
||||
<span>Также собрать квартиры</span>
|
||||
</label>
|
||||
</div>
|
||||
<div style={{ marginTop: 16 }}>
|
||||
<button
|
||||
onClick={() => trigger.mutate()}
|
||||
disabled={!token || trigger.isPending}
|
||||
style={primaryBtn}
|
||||
>
|
||||
{trigger.isPending ? "Запускаю…" : "Запустить сейчас"}
|
||||
</button>
|
||||
{trigger.isError ? (
|
||||
<span style={{ color: "#b3261e", marginLeft: 12 }}>
|
||||
Ошибка: {String(trigger.error)}
|
||||
</span>
|
||||
) : null}
|
||||
{trigger.isSuccess ? (
|
||||
<span style={{ color: "#0a7a3a", marginLeft: 12 }}>
|
||||
Поставлено в очередь: {trigger.data.task_id}
|
||||
</span>
|
||||
) : null}
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<section style={{ ...cardStyle, marginTop: 16 }}>
|
||||
<h2 style={{ marginTop: 0, fontSize: 16 }}>Последние прогоны</h2>
|
||||
{!token ? (
|
||||
<p style={{ color: "#5b6066" }}>
|
||||
Введите admin token чтобы увидеть журнал.
|
||||
</p>
|
||||
) : runs.isLoading ? (
|
||||
<p>Загрузка…</p>
|
||||
) : runs.isError ? (
|
||||
<p style={{ color: "#b3261e" }}>Ошибка: {String(runs.error)}</p>
|
||||
) : (
|
||||
<table
|
||||
style={{ width: "100%", borderCollapse: "collapse", fontSize: 13 }}
|
||||
>
|
||||
<thead>
|
||||
<tr style={{ background: "#f6f7f9" }}>
|
||||
{[
|
||||
"#",
|
||||
"Старт",
|
||||
"Регион",
|
||||
"Dev",
|
||||
"Status",
|
||||
"Объектов",
|
||||
"Квартир",
|
||||
"Req",
|
||||
"Ошибка",
|
||||
].map((h) => (
|
||||
<th key={h} style={th}>
|
||||
{h}
|
||||
</th>
|
||||
))}
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{(runs.data ?? []).map((r) => (
|
||||
<tr
|
||||
key={r.run_id}
|
||||
style={{ borderBottom: "1px solid #eef0f3" }}
|
||||
>
|
||||
<td style={td}>{r.run_id}</td>
|
||||
<td style={td}>
|
||||
{r.started_at?.slice(0, 19).replace("T", " ")}
|
||||
</td>
|
||||
<td style={td}>{r.region_codes.join(",")}</td>
|
||||
<td style={td}>{r.developer_ids?.join(",") || "—"}</td>
|
||||
<td
|
||||
style={{
|
||||
...td,
|
||||
color:
|
||||
r.status === "done"
|
||||
? "#0a7a3a"
|
||||
: r.status === "failed"
|
||||
? "#b3261e"
|
||||
: "#9a6700",
|
||||
fontWeight: 600,
|
||||
}}
|
||||
>
|
||||
{r.status}
|
||||
</td>
|
||||
<td style={td}>{r.objects_count ?? "—"}</td>
|
||||
<td style={td}>{r.flats_count ?? "—"}</td>
|
||||
<td style={td}>{r.requests_count ?? "—"}</td>
|
||||
<td
|
||||
style={{
|
||||
...td,
|
||||
color: "#b3261e",
|
||||
maxWidth: 240,
|
||||
overflow: "hidden",
|
||||
textOverflow: "ellipsis",
|
||||
whiteSpace: "nowrap",
|
||||
}}
|
||||
>
|
||||
{r.error ?? ""}
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
)}
|
||||
</section>
|
||||
</main>
|
||||
);
|
||||
}
|
||||
|
||||
const cardStyle = {
|
||||
background: "#fff",
|
||||
border: "1px solid #e6e8ec",
|
||||
borderRadius: 12,
|
||||
padding: 20,
|
||||
};
|
||||
const labelStyle = {
|
||||
display: "block",
|
||||
fontSize: 12,
|
||||
color: "#5b6066",
|
||||
marginBottom: 4,
|
||||
textTransform: "uppercase" as const,
|
||||
letterSpacing: 0.4,
|
||||
};
|
||||
const inputStyle = {
|
||||
padding: "8px 10px",
|
||||
border: "1px solid #d1d5db",
|
||||
borderRadius: 6,
|
||||
fontSize: 14,
|
||||
width: "100%",
|
||||
boxSizing: "border-box" as const,
|
||||
};
|
||||
const primaryBtn = {
|
||||
padding: "10px 18px",
|
||||
background: "#1d4ed8",
|
||||
color: "#fff",
|
||||
border: "none",
|
||||
borderRadius: 6,
|
||||
fontSize: 14,
|
||||
fontWeight: 600,
|
||||
cursor: "pointer",
|
||||
};
|
||||
const th = {
|
||||
padding: "8px 10px",
|
||||
textAlign: "left" as const,
|
||||
fontWeight: 600,
|
||||
borderBottom: "1px solid #e6e8ec",
|
||||
};
|
||||
const td = { padding: "8px 10px" };
|
||||
File diff suppressed because one or more lines are too long
Loading…
Add table
Reference in a new issue