diff --git a/backend/app/services/scrapers/pravo_gov66_client.py b/backend/app/services/scrapers/pravo_gov66_client.py index a4a28da7..428a3c3e 100644 --- a/backend/app/services/scrapers/pravo_gov66_client.py +++ b/backend/app/services/scrapers/pravo_gov66_client.py @@ -5,6 +5,12 @@ HTML server-rendered (нет JS-рендеринга), httpx достаточн TLS: сайт использует российский УЦ → verify=False (аналогично ekburg_permits #242). +Politeness (#2445 D4): пауза ``_RATE_S`` перед каждым запросом (search / карточка +документа / PDF) — тот же идиом, что nspd_client/objective/gisogd66 (300-500мс). Этот +VPS уже словил WAF-бан от DOM.РФ 24.05.2026 (#2443) за непейсованные последовательные +запросы; pravo.gov66.ru — сайт той же формы (гос-портал region, строгий WAF). Ретрай с +backoff на 429/5xx — идиом gisogd66._get_json. + Публичный API: search_documents(query, law_type, since) → list[dict] get_document_pdf_url(doc_id) → str | None @@ -15,6 +21,7 @@ from __future__ import annotations import logging import re +import time from typing import Any import httpx @@ -28,6 +35,14 @@ _SEARCH_URL = f"{_BASE_URL}/search/" # Тайм-аут щедрый: сервер иногда медленный. _TIMEOUT = httpx.Timeout(60.0, connect=15.0) +# Politeness pacing (#2445 D4): пауза ПЕРЕД каждым запросом к pravo.gov66.ru — тот же +# идиом (300-500мс), что nspd_client/objective/gisogd66. Не непейсованный WAF-риск, +# аналогичный DOM.РФ-бану #2443. +_RATE_S = 0.4 +# Ретраи на 429/5xx: экспоненциальный backoff, суммарно до 1 + _MAX_RETRIES попыток. +_MAX_RETRIES = 2 +_RETRY_BACKOFF_S = 2.0 + # Числовые коды типов документов (select#id_law_type на сайте). # Текстовое значение «Постановление» — неверный параметр, сервер не матчит → 0 хитов. _LAW_TYPE_POSTANOVLENIE = "1" # Постановление Правительства Свердловской области @@ -41,6 +56,43 @@ _RE_PDF_PATH = re.compile(r"/media/pravo/(?!ecp/)[^\s\"']+\.pdf", re.IGNORECASE) _RE_PDF_ECP_PATH = re.compile(r"/media/pravo/ecp/[^\s\"']+\.pdf", re.IGNORECASE) +def _get_with_backoff( + client: httpx.Client, + url: str, + *, + params: dict[str, str] | None = None, +) -> httpx.Response: + """GET с politeness-паузой перед запросом + backoff-ретраем на 429/5xx. + + Пауза ``_RATE_S`` — ПЕРЕД каждой попыткой (в т.ч. первой): пейсинг + последовательных запросов к pravo.gov66.ru, не только ретраев (#2445 D4). + 429/5xx ретраятся до ``_MAX_RETRIES`` раз с экспоненциальным backoff; + прочие статусы (включая 4xx кроме 429) — сразу ``raise_for_status()``. + """ + last_exc: Exception | None = None + for attempt in range(_MAX_RETRIES + 1): + time.sleep(_RATE_S) + resp = client.get(url, params=params) + if resp.status_code == 429 or resp.status_code >= 500: + last_exc = httpx.HTTPStatusError( + f"{resp.status_code} on {url}", request=resp.request, response=resp + ) + if attempt >= _MAX_RETRIES: + raise last_exc + logger.warning( + "pravo_gov66_client: %s (попытка %d) на %s — backoff", + resp.status_code, + attempt + 1, + url, + ) + time.sleep(_RETRY_BACKOFF_S * (attempt + 1)) + continue + resp.raise_for_status() + return resp + assert last_exc is not None + raise last_exc + + def search_documents( query: str, law_type: str = _LAW_TYPE_POSTANOVLENIE, @@ -65,8 +117,7 @@ def search_documents( try: with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client: - resp = client.get(_SEARCH_URL, params=params) - resp.raise_for_status() + resp = _get_with_backoff(client, _SEARCH_URL, params=params) html = resp.text except Exception as exc: logger.error("pravo_gov66_client.search_documents: ошибка сети q=%r: %s", query, exc) @@ -117,8 +168,7 @@ def get_document_pdf_url(doc_id: str) -> str | None: doc_url = f"{_BASE_URL}/{doc_id}/" try: with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client: - resp = client.get(doc_url) - resp.raise_for_status() + resp = _get_with_backoff(client, doc_url) html = resp.text except Exception as exc: logger.error( @@ -151,6 +201,10 @@ def _extract_pdf_url(html: str) -> str | None: def fetch_pdf(url: str) -> bytes: """Загружает PDF по URL. + Пейсинг + backoff на 429/5xx через ``_get_with_backoff`` (#2445 D4) — caller + (reservation_ingest) НЕ должен добавлять свой sleep перед этим вызовом, пауза + уже внутри клиента. + Args: url: абсолютный URL PDF-файла. @@ -161,8 +215,7 @@ def fetch_pdf(url: str) -> bytes: httpx.HTTPError: при сетевой ошибке (caller должен обработать gracefully). """ with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client: - resp = client.get(url) - resp.raise_for_status() + resp = _get_with_backoff(client, url) return resp.content diff --git a/backend/app/workers/tasks/reservation_ingest.py b/backend/app/workers/tasks/reservation_ingest.py index 56ee8006..bac86f72 100644 --- a/backend/app/workers/tasks/reservation_ingest.py +++ b/backend/app/workers/tasks/reservation_ingest.py @@ -10,12 +10,17 @@ Beat НЕ настроен: постановления выходят редко celery call tasks.reservation_ingest.ingest_reservations Conventions (зеркало ird_harvest.py / backend.md): - • SessionLocal() + try/finally close. + • Session lifecycle (#2445 D4): PDF fetch происходит БЕЗ открытой сессии — pravo.gov66 + иногда медленный (таймаут до 60с), держать connection-pool slot на всё время внешнего + запроса нельзя. Короткоживущий ``SessionLocal()`` открывается ТОЛЬКО на запись строк + одного документа (open → upsert → commit → close per item), не на весь цикл. • SAVEPOINT per-row (``with db.begin_nested():``) — битая строка не валит батч. • CAST(:x AS type) — НИКОГДА :x::text (psycopg v3). • ON CONFLICT (cad_num, act_number) → обновляет метаданные + fetched_at. • httpx.Client (не requests). Сбой одного дока не валит прогон. • q-запросы — 2-3 слова (движок pravo.gov66 строгий AND/фраза, 1 или 4+ → 0 хитов). + • Pacing/backoff к pravo.gov66.ru — внутри pravo_gov66_client (_get_with_backoff), + НЕ здесь: каждый search/page/pdf запрос уже пейсится на уровне клиента. """ from __future__ import annotations @@ -97,8 +102,11 @@ def _fetch_pdf_safe(pdf_url: str) -> bytes | None: return None -def _upsert_records(db: Session, records_meta: list[dict[str, Any]]) -> int: - """UPSERT записей в land_reservation. Возвращает число обработанных строк.""" +def _upsert_records_in_session(db: Session, records_meta: list[dict[str, Any]]) -> int: + """UPSERT записей в land_reservation на уже открытой сессии. + + Возвращает число upsert'нутых строк. + """ count = 0 for row in records_meta: try: @@ -114,6 +122,27 @@ def _upsert_records(db: Session, records_meta: list[dict[str, Any]]) -> int: return count +def _upsert_records(records_meta: list[dict[str, Any]]) -> int: + """Открывает короткоживущую сессию ТОЛЬКО на запись одного документа → commit → close. + + Session lifecycle (#2445 D4): вызывается ПОСЛЕ внешнего PDF-fetch, никогда во время + него — держать открытую транзакцию/connection-pool slot на время сетевого запроса + к pravo.gov66.ru (таймаут до 60с) недопустимо (см. WAF-инцидент #2443). + """ + if not records_meta: + return 0 + db: Session = SessionLocal() + try: + count = _upsert_records_in_session(db, records_meta) + db.commit() + return count + except Exception: + db.rollback() + raise + finally: + db.close() + + def _enumerate_documents(since: str | None = None) -> list[dict[str, Any]]: """Обходит все поисковые запросы и возвращает дедуплицированный список документов. @@ -182,117 +211,111 @@ def ingest_reservations( def _ingest_explicit_docs(docs: list[dict[str, Any]]) -> dict[str, int]: - """Ингестирует явно переданный список документов (back-compat с ручным вызовом).""" + """Ингестирует явно переданный список документов (back-compat с ручным вызовом). + + Session lifecycle (#2445 D4): PDF fetch (сетевой, может занять до 60с) выполняется + БЕЗ открытой DB-сессии; ``_upsert_records`` открывает свою короткоживущую сессию + только на запись строк одного документа (commit+close сразу после). + """ total_parcels = 0 - db: Session = SessionLocal() - try: - for doc in docs: - url = doc.get("url", "") - act_number = doc.get("act_number") - basis_act = doc.get("basis_act", "") - doc_purpose = doc.get("purpose") - doc_kind = doc.get("kind", "изъятие") + for doc in docs: + url = doc.get("url", "") + act_number = doc.get("act_number") + basis_act = doc.get("basis_act", "") + doc_purpose = doc.get("purpose") + doc_kind = doc.get("kind", "изъятие") - logger.info("ingest_reservations(explicit): загрузка url=%s act=%s", url, act_number) + logger.info("ingest_reservations(explicit): загрузка url=%s act=%s", url, act_number) - pdf_bytes = _fetch_pdf_safe(url) - if pdf_bytes is None: - logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url) - continue + pdf_bytes = _fetch_pdf_safe(url) # без открытой сессии — только сетевой запрос + if pdf_bytes is None: + logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url) + continue - records = extract_from_pdf(pdf_bytes) - logger.info("ingest_reservations(explicit): url=%s → %d КН", url, len(records)) - if not records: - continue + records = extract_from_pdf(pdf_bytes) + logger.info("ingest_reservations(explicit): url=%s → %d КН", url, len(records)) + if not records: + continue - rows = [ - { - "cad_num": rec.cad_num, - "reservation_kind": rec.reservation_kind or doc_kind, - "basis_act": basis_act, - "act_number": rec.act_number or act_number, - "act_date": rec.act_date, - "purpose": rec.purpose or doc_purpose, - "doc_url": url, - "source": "page_pdf", - "raw_excerpt": rec.raw_excerpt, - } - for rec in records - ] - count = _upsert_records(db, rows) - db.commit() - total_parcels += count - - logger.info( - "ingest_reservations(explicit): готово docs=%d parcels=%d", - len(docs), - total_parcels, - ) - finally: - db.close() + rows = [ + { + "cad_num": rec.cad_num, + "reservation_kind": rec.reservation_kind or doc_kind, + "basis_act": basis_act, + "act_number": rec.act_number or act_number, + "act_date": rec.act_date, + "purpose": rec.purpose or doc_purpose, + "doc_url": url, + "source": "page_pdf", + "raw_excerpt": rec.raw_excerpt, + } + for rec in records + ] + total_parcels += _upsert_records(rows) # short-lived session: open→upsert→commit→close + logger.info( + "ingest_reservations(explicit): готово docs=%d parcels=%d", + len(docs), + total_parcels, + ) return {"docs": len(docs), "parcels": total_parcels} def _ingest_enumerated_docs(enumerated: list[dict[str, Any]]) -> dict[str, int]: - """Ингестирует документы, найденные через enumerate с pravo.gov66.""" + """Ингестирует документы, найденные через enumerate с pravo.gov66. + + Session lifecycle (#2445 D4): оба сетевых вызова на документ (get_document_pdf_url, + fetch_pdf) выполняются БЕЗ открытой DB-сессии; ``_upsert_records`` открывает свою + короткоживущую сессию только на запись строк одного документа. + """ total_parcels = 0 fetched_docs = 0 - db: Session = SessionLocal() - try: - for hit in enumerated: - doc_id: str = hit["doc_id"] - title: str = hit.get("title", "") - doc_page_url: str = hit["url"] + for hit in enumerated: + doc_id: str = hit["doc_id"] + title: str = hit.get("title", "") + doc_page_url: str = hit["url"] - logger.info("ingest_reservations(enum): doc_id=%s title=%r", doc_id, title) + logger.info("ingest_reservations(enum): doc_id=%s title=%r", doc_id, title) - pdf_url = get_document_pdf_url(doc_id) - if pdf_url is None: - logger.warning( - "ingest_reservations(enum): doc_id=%s — PDF не найден на странице", doc_id - ) - continue + pdf_url = get_document_pdf_url(doc_id) # сеть, без открытой сессии + if pdf_url is None: + logger.warning( + "ingest_reservations(enum): doc_id=%s — PDF не найден на странице", doc_id + ) + continue - pdf_bytes = _fetch_pdf_safe(pdf_url) - if pdf_bytes is None: - logger.warning( - "ingest_reservations(enum): doc_id=%s — не удалось скачать PDF", doc_id - ) - continue + pdf_bytes = _fetch_pdf_safe(pdf_url) # сеть, без открытой сессии + if pdf_bytes is None: + logger.warning("ingest_reservations(enum): doc_id=%s — не удалось скачать PDF", doc_id) + continue - fetched_docs += 1 - records = extract_from_pdf(pdf_bytes) - logger.info("ingest_reservations(enum): doc_id=%s → %d КН", doc_id, len(records)) - if not records: - continue + fetched_docs += 1 + records = extract_from_pdf(pdf_bytes) + logger.info("ingest_reservations(enum): doc_id=%s → %d КН", doc_id, len(records)) + if not records: + continue - rows = [ - { - "cad_num": rec.cad_num, - "reservation_kind": rec.reservation_kind, - "basis_act": title, - "act_number": rec.act_number, - "act_date": rec.act_date, - "purpose": rec.purpose, - "doc_url": doc_page_url, - "source": "pravo_gov66", - "raw_excerpt": rec.raw_excerpt, - } - for rec in records - ] - count = _upsert_records(db, rows) - db.commit() - total_parcels += count - - logger.info( - "ingest_reservations(enum): готово fetched=%d parcels=%d", - fetched_docs, - total_parcels, - ) - finally: - db.close() + rows = [ + { + "cad_num": rec.cad_num, + "reservation_kind": rec.reservation_kind, + "basis_act": title, + "act_number": rec.act_number, + "act_date": rec.act_date, + "purpose": rec.purpose, + "doc_url": doc_page_url, + "source": "pravo_gov66", + "raw_excerpt": rec.raw_excerpt, + } + for rec in records + ] + total_parcels += _upsert_records(rows) # short-lived session: open→upsert→commit→close + logger.info( + "ingest_reservations(enum): готово fetched=%d parcels=%d", + fetched_docs, + total_parcels, + ) return {"docs": fetched_docs, "parcels": total_parcels} diff --git a/backend/tests/scrapers/test_pravo_gov66_client.py b/backend/tests/scrapers/test_pravo_gov66_client.py index cdb06a0a..d3c9dd47 100644 --- a/backend/tests/scrapers/test_pravo_gov66_client.py +++ b/backend/tests/scrapers/test_pravo_gov66_client.py @@ -1,14 +1,19 @@ """Тесты pravo_gov66_client — offline, без живой сети. Покрывают: парсинг search-HTML (doc_id/title/url), парсинг страницы документа -(PDF-ссылка), fallback на ecp-вариант, отсутствие PDF. +(PDF-ссылка), fallback на ecp-вариант, отсутствие PDF, pacing/backoff (#2445 D4) +между последовательными запросами к pravo.gov66.ru. """ from __future__ import annotations +import httpx +import pytest + from app.services.scrapers.pravo_gov66_client import ( _LAW_TYPE_POSTANOVLENIE, _extract_pdf_url, + _get_with_backoff, _parse_search_html, ) @@ -169,3 +174,96 @@ class TestExtractPdfUrl: url = _extract_pdf_url(_DOC_HTML_ECP_ONLY) assert url is not None assert url.startswith("https://www.pravo.gov66.ru/") + + +# ── Тесты pacing/backoff (#2445 D4) — offline через httpx.MockTransport ─────── + + +def _client_with_responses(responses: list[httpx.Response]) -> httpx.Client: + """httpx.Client на MockTransport, отдающий responses по очереди (без реальной сети).""" + calls = iter(responses) + + def _handler(request: httpx.Request) -> httpx.Response: + return next(calls) + + return httpx.Client(transport=httpx.MockTransport(_handler)) + + +class TestGetWithBackoff: + def test_sleeps_before_request(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Пейсинг: time.sleep вызывается ПЕРЕД запросом, даже на первой/единственной попытке.""" + sleeps: list[float] = [] + monkeypatch.setattr( + "app.services.scrapers.pravo_gov66_client.time.sleep", + lambda s: sleeps.append(s), + ) + client = _client_with_responses([httpx.Response(200, text="ok")]) + _get_with_backoff(client, "https://www.pravo.gov66.ru/1/") + assert len(sleeps) == 1 + + def test_paces_between_successive_fetches(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Несколько последовательных fetch → time.sleep между КАЖДЫМ (не только ретраи).""" + sleeps: list[float] = [] + monkeypatch.setattr( + "app.services.scrapers.pravo_gov66_client.time.sleep", + lambda s: sleeps.append(s), + ) + client = _client_with_responses( + [ + httpx.Response(200, text="a"), + httpx.Response(200, text="b"), + httpx.Response(200, text="c"), + ] + ) + for i in range(3): + _get_with_backoff(client, f"https://www.pravo.gov66.ru/{i}/") + assert len(sleeps) == 3 + + def test_retries_on_429_then_succeeds(self, monkeypatch: pytest.MonkeyPatch) -> None: + """429 → backoff sleep + ретрай → 200 успешно, без исключения.""" + sleeps: list[float] = [] + monkeypatch.setattr( + "app.services.scrapers.pravo_gov66_client.time.sleep", + lambda s: sleeps.append(s), + ) + client = _client_with_responses( + [httpx.Response(429, text="rate limited"), httpx.Response(200, text="ok")] + ) + resp = _get_with_backoff(client, "https://www.pravo.gov66.ru/1/") + assert resp.status_code == 200 + # 1 pacing-sleep перед 1й попыткой + 1 backoff-sleep перед ретраем + 1 pacing перед 2й. + assert len(sleeps) >= 2 + + def test_retries_on_5xx_then_succeeds(self, monkeypatch: pytest.MonkeyPatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr( + "app.services.scrapers.pravo_gov66_client.time.sleep", + lambda s: sleeps.append(s), + ) + client = _client_with_responses( + [httpx.Response(503, text="unavailable"), httpx.Response(200, text="ok")] + ) + resp = _get_with_backoff(client, "https://www.pravo.gov66.ru/1/") + assert resp.status_code == 200 + + def test_raises_after_exhausting_retries_on_429(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("app.services.scrapers.pravo_gov66_client.time.sleep", lambda s: None) + # _MAX_RETRIES=2 → до 3 попыток, все 429. + client = _client_with_responses([httpx.Response(429, text="x") for _ in range(3)]) + with pytest.raises(httpx.HTTPStatusError): + _get_with_backoff(client, "https://www.pravo.gov66.ru/1/") + + def test_4xx_other_than_429_raises_immediately_without_retry( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """404 не ретраится — сразу raise_for_status(), задача не тратит время на 3 попытки.""" + sleeps: list[float] = [] + monkeypatch.setattr( + "app.services.scrapers.pravo_gov66_client.time.sleep", + lambda s: sleeps.append(s), + ) + client = _client_with_responses([httpx.Response(404, text="not found")]) + with pytest.raises(httpx.HTTPStatusError): + _get_with_backoff(client, "https://www.pravo.gov66.ru/1/") + # Только 1 pacing-sleep перед единственной попыткой, никаких backoff-ретраев. + assert len(sleeps) == 1 diff --git a/backend/tests/workers/test_reservation_ingest.py b/backend/tests/workers/test_reservation_ingest.py index a6807dde..34e6784c 100644 --- a/backend/tests/workers/test_reservation_ingest.py +++ b/backend/tests/workers/test_reservation_ingest.py @@ -1,7 +1,9 @@ """Тесты ingest_reservations (#1062) — offline, без живой сети и реальной БД. Мокаем: pravo_gov66_client (search/get_pdf_url/fetch_pdf) + extract_from_pdf + SessionLocal. -Проверяем: фильтрацию заголовков, дедупликацию doc_id, upsert-параметры, graceful на ошибках. +Проверяем: фильтрацию заголовков, дедупликацию doc_id, upsert-параметры, graceful на ошибках, +session lifecycle (#2445 D4: сессия НЕ держится через весь цикл — открывается заново на +каждый документ) и pacing к pravo.gov66.ru (time.sleep между запросами клиента). """ from __future__ import annotations @@ -42,6 +44,25 @@ class _FakeDB: self.closed = True +class _FakeSessionFactory: + """SessionLocal-двойник: считает, сколько раз вызывалась фабрика (open) и сколько + раз каждая выданная сессия была закрыта СРАЗУ (до следующего open) — доказательство, + что сессия не держится открытой через весь цикл, а открывается/закрывается per-item. + """ + + def __init__(self) -> None: + self.sessions: list[_FakeDB] = [] + # True если на момент следующего SessionLocal() предыдущая сессия уже была closed. + self.closed_before_next_open: list[bool] = [] + + def __call__(self) -> _FakeDB: + if self.sessions: + self.closed_before_next_open.append(self.sessions[-1].closed) + db = _FakeDB() + self.sessions.append(db) + return db + + # ── Фикстуры ────────────────────────────────────────────────────────────────── _SAMPLE_RECORDS = [ @@ -213,11 +234,40 @@ class TestIngestEnumerate: assert params["cad_num"] == "66:41:0101001:1" assert params["source"] == "pravo_gov66" - def test_db_closed_after_run(self, monkeypatch: Any) -> None: - db = _FakeDB() - _patch_enumerate(monkeypatch, db, hits=[]) + def test_db_closed_after_each_doc(self, monkeypatch: Any) -> None: + """Session lifecycle (#2445 D4): сессия открывается/закрывается НА КАЖДЫЙ документ, + не одна на весь прогон — доказывает, что она не держится через PDF-fetch цикла. + """ + factory = _FakeSessionFactory() + monkeypatch.setattr(reservation_ingest, "SessionLocal", factory) + monkeypatch.setattr(reservation_ingest, "search_documents", lambda q, **kw: _RELEVANT_HITS) + monkeypatch.setattr( + reservation_ingest, + "get_document_pdf_url", + lambda doc_id: "https://www.pravo.gov66.ru/media/pravo/test.pdf", + ) + monkeypatch.setattr(reservation_ingest, "fetch_pdf", lambda url: _PDF_BYTES) + monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: _SAMPLE_RECORDS[:1]) + reservation_ingest.ingest_reservations() - assert db.closed + + # 2 релевантных хита → 2 документа с записями → SessionLocal() вызван 2 раза. + assert len(factory.sessions) == 2 + # Каждая сессия должна быть closed к моменту следующего open (per-item lifecycle, + # не одна сессия держится открытой на весь цикл enumerate). + assert factory.closed_before_next_open == [True] + # Последняя сессия тоже закрыта после прогона. + assert factory.sessions[-1].closed + # И каждая закоммичена (per-item commit, идемпотентность сохранена). + assert all(s.commit_count == 1 for s in factory.sessions) + + def test_no_session_opened_when_no_hits(self, monkeypatch: Any) -> None: + """Нет хитов → нет документов → SessionLocal вообще не вызывается (нечего писать).""" + factory = _FakeSessionFactory() + monkeypatch.setattr(reservation_ingest, "SessionLocal", factory) + monkeypatch.setattr(reservation_ingest, "search_documents", lambda q, **kw: []) + reservation_ingest.ingest_reservations() + assert factory.sessions == [] def test_skips_empty_records(self, monkeypatch: Any) -> None: """PDF без кад-номеров → ни одного execute."""