fix(scrape-kn): real documents endpoints (5x) + skip obj_checks + sem 3→8 #325
2 changed files with 129 additions and 25 deletions
|
|
@ -28,7 +28,10 @@ from sqlalchemy.orm import Session
|
|||
|
||||
from app.core.db import SessionLocal
|
||||
from app.services.scrapers.documents import extract_documents, upsert_documents
|
||||
from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks
|
||||
|
||||
# obj_checks import temporarily disabled — endpoint /checks returns 404 (run #19).
|
||||
# Re-enable with _fetch_obj_checks_safe when endpoint is found (see TODO in Phase B/C).
|
||||
# from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks
|
||||
from app.services.scrapers.stealth import BASE_URL, BrowserSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -39,7 +42,13 @@ PATH_SALE_GRAPH = "/сервисы/api/object/{obj_id}/sale_graph" # ?type=apar
|
|||
PATH_SALES_AGG = "/сервисы/api/object/{obj_id}/sales_agg"
|
||||
PATH_INFRA = "/сервисы/api/object/{obj_id}/infrastructure"
|
||||
PATH_PHOTOS = "/сервисы/api/object/construction/progress/photo/{obj_id}"
|
||||
PATH_DOCUMENTS = "/сервисы/api/object/{obj_id}/documents"
|
||||
# Five separate document endpoints discovered via Playwright on obj=65136 (2026-05-17).
|
||||
# Former single PATH_DOCUMENTS (/documents) returned 404 — replaced by these 5 real paths.
|
||||
PATH_DOC_RPD = "/сервисы/api/object/{obj_id}/document/rpd"
|
||||
PATH_DOC_DEVELOPER_REPORT = "/сервисы/api/object/{obj_id}/developer/report"
|
||||
PATH_DOC_PROJECT_DOCUMENTATION = "/сервисы/api/object/{obj_id}/project/documentation"
|
||||
PATH_DOC_DOCUMENTATION_OTHER = "/сервисы/api/object/{obj_id}/documentation/other"
|
||||
PATH_DOC_PERMITS = "/сервисы/api/object/{obj_id}/document/permits"
|
||||
# NOTE: PATH_CHECKS URL not verified via devtools — likely pattern analogous to /infrastructure.
|
||||
# If endpoint returns 404/error the failure is logged to kn_scrape_failures and scrape continues.
|
||||
PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks"
|
||||
|
|
@ -997,14 +1006,50 @@ async def fetch_photos(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str
|
|||
return (payload.get("data") or []), full_url
|
||||
|
||||
|
||||
async def fetch_documents(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Fetch PDF-document list for one object. Returns (items, full_url)."""
|
||||
path = PATH_DOCUMENTS.format(obj_id=obj_id)
|
||||
async def _fetch_doc_endpoint(
|
||||
sess: BrowserSession, path_template: str, obj_id: int
|
||||
) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Generic helper: fetch one document endpoint, return (items, full_url)."""
|
||||
path = path_template.format(obj_id=obj_id)
|
||||
payload = await sess.get_json(path, {})
|
||||
full_url = f"{BASE_URL}{path}"
|
||||
if isinstance(payload, list):
|
||||
return payload, full_url
|
||||
return (payload.get("data") or []), full_url
|
||||
data = payload.get("data")
|
||||
if isinstance(data, list):
|
||||
return data, full_url
|
||||
return [], full_url
|
||||
|
||||
|
||||
async def fetch_doc_rpd(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||||
"""214-ФЗ отчёт (РПД)."""
|
||||
return await _fetch_doc_endpoint(sess, PATH_DOC_RPD, obj_id)
|
||||
|
||||
|
||||
async def fetch_doc_developer_report(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Отчёт застройщика."""
|
||||
return await _fetch_doc_endpoint(sess, PATH_DOC_DEVELOPER_REPORT, obj_id)
|
||||
|
||||
|
||||
async def fetch_doc_project_documentation(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Проектная декларация."""
|
||||
return await _fetch_doc_endpoint(sess, PATH_DOC_PROJECT_DOCUMENTATION, obj_id)
|
||||
|
||||
|
||||
async def fetch_doc_documentation_other(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Прочие документы."""
|
||||
return await _fetch_doc_endpoint(sess, PATH_DOC_DOCUMENTATION_OTHER, obj_id)
|
||||
|
||||
|
||||
async def fetch_doc_permits(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]:
|
||||
"""Разрешения на строительство."""
|
||||
return await _fetch_doc_endpoint(sess, PATH_DOC_PERMITS, obj_id)
|
||||
|
||||
|
||||
async def fetch_obj_checks(sess: BrowserSession, obj_id: int) -> tuple[Any, str]:
|
||||
|
|
@ -1080,15 +1125,59 @@ async def _fetch_photos_safe(
|
|||
return ("photos", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_documents_safe(
|
||||
async def _fetch_doc_rpd_safe(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||||
full_url = f"{BASE_URL}{PATH_DOCUMENTS.format(obj_id=obj_id)}"
|
||||
full_url = f"{BASE_URL}{PATH_DOC_RPD.format(obj_id=obj_id)}"
|
||||
try:
|
||||
items, url = await fetch_documents(sess, obj_id)
|
||||
return ("documents", url, (items, url))
|
||||
items, url = await fetch_doc_rpd(sess, obj_id)
|
||||
return ("doc_rpd", url, (items, url))
|
||||
except Exception as e:
|
||||
return ("documents", full_url, e)
|
||||
return ("doc_rpd", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_doc_developer_report_safe(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||||
full_url = f"{BASE_URL}{PATH_DOC_DEVELOPER_REPORT.format(obj_id=obj_id)}"
|
||||
try:
|
||||
items, url = await fetch_doc_developer_report(sess, obj_id)
|
||||
return ("doc_developer_report", url, (items, url))
|
||||
except Exception as e:
|
||||
return ("doc_developer_report", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_doc_project_documentation_safe(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||||
full_url = f"{BASE_URL}{PATH_DOC_PROJECT_DOCUMENTATION.format(obj_id=obj_id)}"
|
||||
try:
|
||||
items, url = await fetch_doc_project_documentation(sess, obj_id)
|
||||
return ("doc_project_documentation", url, (items, url))
|
||||
except Exception as e:
|
||||
return ("doc_project_documentation", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_doc_documentation_other_safe(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||||
full_url = f"{BASE_URL}{PATH_DOC_DOCUMENTATION_OTHER.format(obj_id=obj_id)}"
|
||||
try:
|
||||
items, url = await fetch_doc_documentation_other(sess, obj_id)
|
||||
return ("doc_documentation_other", url, (items, url))
|
||||
except Exception as e:
|
||||
return ("doc_documentation_other", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_doc_permits_safe(
|
||||
sess: BrowserSession, obj_id: int
|
||||
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]:
|
||||
full_url = f"{BASE_URL}{PATH_DOC_PERMITS.format(obj_id=obj_id)}"
|
||||
try:
|
||||
items, url = await fetch_doc_permits(sess, obj_id)
|
||||
return ("doc_permits", url, (items, url))
|
||||
except Exception as e:
|
||||
return ("doc_permits", full_url, e)
|
||||
|
||||
|
||||
async def _fetch_obj_checks_safe(
|
||||
|
|
@ -1691,8 +1780,15 @@ async def run_region_sweep(
|
|||
coros.append(_fetch_sales_agg_safe(sess, obj_id))
|
||||
coros.append(_fetch_infrastructure_safe(sess, obj_id))
|
||||
coros.append(_fetch_photos_safe(sess, obj_id))
|
||||
coros.append(_fetch_documents_safe(sess, obj_id))
|
||||
coros.append(_fetch_obj_checks_safe(sess, obj_id))
|
||||
coros.append(_fetch_doc_rpd_safe(sess, obj_id))
|
||||
coros.append(_fetch_doc_developer_report_safe(sess, obj_id))
|
||||
coros.append(_fetch_doc_project_documentation_safe(sess, obj_id))
|
||||
coros.append(_fetch_doc_documentation_other_safe(sess, obj_id))
|
||||
coros.append(_fetch_doc_permits_safe(sess, obj_id))
|
||||
# TODO: obj_checks endpoint not found at /api/object/{id}/checks (404).
|
||||
# 6 чек-боксов "Проверено на наш.дом.рф" вероятно inline в kn/object payload.
|
||||
# Re-enable после investigation структуры объекта (separate PR).
|
||||
# coros.append(_fetch_obj_checks_safe(sess, obj_id))
|
||||
|
||||
if not coros:
|
||||
continue
|
||||
|
|
@ -1701,6 +1797,16 @@ async def run_region_sweep(
|
|||
results = await asyncio.gather(*coros, return_exceptions=False)
|
||||
|
||||
# Sequential upsert — DB session не thread-safe
|
||||
all_docs: list[dict[str, Any]] = []
|
||||
_doc_kinds = frozenset(
|
||||
(
|
||||
"doc_rpd",
|
||||
"doc_developer_report",
|
||||
"doc_project_documentation",
|
||||
"doc_documentation_other",
|
||||
"doc_permits",
|
||||
)
|
||||
)
|
||||
for kind_tag, full_url, result in results:
|
||||
if isinstance(result, Exception):
|
||||
_classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
|
||||
|
|
@ -1743,18 +1849,16 @@ async def run_region_sweep(
|
|||
db, obj_id, photos_data, local_paths, thumb_paths
|
||||
)
|
||||
|
||||
elif kind_tag == "documents":
|
||||
docs_payload, _ = result # type: ignore[misc]
|
||||
docs = extract_documents(docs_payload or [])
|
||||
if docs:
|
||||
ins, _skip = upsert_documents(db, obj_id, docs)
|
||||
extras_counts["documents_rows"] += ins
|
||||
elif kind_tag in _doc_kinds:
|
||||
# Каждый из 5 doc-endpoint'ов отдаёт свой список документов.
|
||||
# Накапливаем в all_docs — единый upsert после цикла.
|
||||
doc_items, _ = result # type: ignore[misc]
|
||||
all_docs.extend(extract_documents(doc_items or []))
|
||||
|
||||
elif kind_tag == "obj_checks":
|
||||
checks_payload, _ = result # type: ignore[misc]
|
||||
checks = extract_obj_checks(checks_payload)
|
||||
if checks:
|
||||
extras_counts["checks_rows"] += upsert_obj_checks(db, obj_id, checks)
|
||||
# Единый upsert всех документов объекта после обработки 5 endpoint'ов.
|
||||
if all_docs:
|
||||
ins, _skip = upsert_documents(db, obj_id, all_docs)
|
||||
extras_counts["documents_rows"] += ins
|
||||
|
||||
# Checkpoint раз в 10 объектов: запись прогресса + heartbeat.
|
||||
if (i + 1) % 10 == 0:
|
||||
|
|
|
|||
|
|
@ -104,7 +104,7 @@ class BrowserSession:
|
|||
self._browser: Browser | None = None
|
||||
self._context: BrowserContext | None = None
|
||||
self._page: Page | None = None
|
||||
self._sem = asyncio.Semaphore(3)
|
||||
self._sem = asyncio.Semaphore(8)
|
||||
self._request_count = 0
|
||||
|
||||
async def __aenter__(self) -> BrowserSession:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue