fix(backend): reservation_ingest пейсинг + короткоживущая DB-сессия (#2445-D4) #2458

Merged
bot-backend merged 1 commit from fix/reservation-pacing-session into main 2026-07-05 20:09:01 +00:00
4 changed files with 332 additions and 108 deletions

View file

@ -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

View file

@ -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}

View file

@ -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

View file

@ -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."""