fix(scrape-kn): real documents endpoints (5x) + skip obj_checks 404 + sem 3->6

PR #322 used placeholder URLs:
- /api/object/{id}/documents -> 404 (Playwright showed: 5 real endpoints:
  document/rpd / developer/report / project/documentation /
  documentation/other / document/permits). Replaced with 5 parallel
  fetches + merge into single upsert_documents call.
- /api/object/{id}/checks -> 404. 6 checkboxes likely inline in bulk
  kn/object payload; endpoint /checks does not exist. Commented out
  _fetch_obj_checks_safe with TODO for separate investigation PR.
- BrowserSession Semaphore(3) -> Semaphore(6) — double concurrency
  for intra-object gather.

GlitchTip: kn_scrape_failures run #19 showed 40 failures each for
both placeholder endpoints across first 30 objects of 1532.
This commit is contained in:
lekss361 2026-05-17 22:59:44 +03:00
parent 79c9a8a034
commit 0f289c05b1
2 changed files with 129 additions and 25 deletions

View file

@ -28,7 +28,10 @@ from sqlalchemy.orm import Session
from app.core.db import SessionLocal from app.core.db import SessionLocal
from app.services.scrapers.documents import extract_documents, upsert_documents 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 from app.services.scrapers.stealth import BASE_URL, BrowserSession
logger = logging.getLogger(__name__) 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_SALES_AGG = "/сервисы/api/object/{obj_id}/sales_agg"
PATH_INFRA = "/сервисы/api/object/{obj_id}/infrastructure" PATH_INFRA = "/сервисы/api/object/{obj_id}/infrastructure"
PATH_PHOTOS = "/сервисы/api/object/construction/progress/photo/{obj_id}" 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. # 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. # If endpoint returns 404/error the failure is logged to kn_scrape_failures and scrape continues.
PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks" 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 return (payload.get("data") or []), full_url
async def fetch_documents(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]: async def _fetch_doc_endpoint(
"""Fetch PDF-document list for one object. Returns (items, full_url).""" sess: BrowserSession, path_template: str, obj_id: int
path = PATH_DOCUMENTS.format(obj_id=obj_id) ) -> 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, {}) payload = await sess.get_json(path, {})
full_url = f"{BASE_URL}{path}" full_url = f"{BASE_URL}{path}"
if isinstance(payload, list): if isinstance(payload, list):
return payload, full_url 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]: 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) return ("photos", full_url, e)
async def _fetch_documents_safe( async def _fetch_doc_rpd_safe(
sess: BrowserSession, obj_id: int sess: BrowserSession, obj_id: int
) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]: ) -> 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: try:
items, url = await fetch_documents(sess, obj_id) items, url = await fetch_doc_rpd(sess, obj_id)
return ("documents", url, (items, url)) return ("doc_rpd", url, (items, url))
except Exception as e: 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( 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_sales_agg_safe(sess, obj_id))
coros.append(_fetch_infrastructure_safe(sess, obj_id)) coros.append(_fetch_infrastructure_safe(sess, obj_id))
coros.append(_fetch_photos_safe(sess, obj_id)) coros.append(_fetch_photos_safe(sess, obj_id))
coros.append(_fetch_documents_safe(sess, obj_id)) coros.append(_fetch_doc_rpd_safe(sess, obj_id))
coros.append(_fetch_obj_checks_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: if not coros:
continue continue
@ -1701,6 +1797,16 @@ async def run_region_sweep(
results = await asyncio.gather(*coros, return_exceptions=False) results = await asyncio.gather(*coros, return_exceptions=False)
# Sequential upsert — DB session не thread-safe # 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: for kind_tag, full_url, result in results:
if isinstance(result, Exception): if isinstance(result, Exception):
_classify_and_log(db, run_id, obj_id, kind_tag, full_url, result) _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 db, obj_id, photos_data, local_paths, thumb_paths
) )
elif kind_tag == "documents": elif kind_tag in _doc_kinds:
docs_payload, _ = result # type: ignore[misc] # Каждый из 5 doc-endpoint'ов отдаёт свой список документов.
docs = extract_documents(docs_payload or []) # Накапливаем в all_docs — единый upsert после цикла.
if docs: doc_items, _ = result # type: ignore[misc]
ins, _skip = upsert_documents(db, obj_id, docs) all_docs.extend(extract_documents(doc_items or []))
extras_counts["documents_rows"] += ins
elif kind_tag == "obj_checks": # Единый upsert всех документов объекта после обработки 5 endpoint'ов.
checks_payload, _ = result # type: ignore[misc] if all_docs:
checks = extract_obj_checks(checks_payload) ins, _skip = upsert_documents(db, obj_id, all_docs)
if checks: extras_counts["documents_rows"] += ins
extras_counts["checks_rows"] += upsert_obj_checks(db, obj_id, checks)
# Checkpoint раз в 10 объектов: запись прогресса + heartbeat. # Checkpoint раз в 10 объектов: запись прогресса + heartbeat.
if (i + 1) % 10 == 0: if (i + 1) % 10 == 0:

View file

@ -104,7 +104,7 @@ class BrowserSession:
self._browser: Browser | None = None self._browser: Browser | None = None
self._context: BrowserContext | None = None self._context: BrowserContext | None = None
self._page: Page | None = None self._page: Page | None = None
self._sem = asyncio.Semaphore(3) self._sem = asyncio.Semaphore(6)
self._request_count = 0 self._request_count = 0
async def __aenter__(self) -> BrowserSession: async def __aenter__(self) -> BrowserSession: