fix(backend): reservation_ingest пейсинг + короткоживущая DB-сессия (#2445-D4) (#2458)
Some checks failed
Deploy / build-backend (push) Blocked by required conditions
Deploy / build-worker (push) Blocked by required conditions
Deploy / build-frontend (push) Blocked by required conditions
Deploy / deploy (push) Blocked by required conditions
Deploy / changes (push) Has been cancelled
Some checks failed
Deploy / build-backend (push) Blocked by required conditions
Deploy / build-worker (push) Blocked by required conditions
Deploy / build-frontend (push) Blocked by required conditions
Deploy / deploy (push) Blocked by required conditions
Deploy / changes (push) Has been cancelled
This commit is contained in:
parent
be4b2ee1fa
commit
d1fa450f48
4 changed files with 332 additions and 108 deletions
|
|
@ -5,6 +5,12 @@ HTML server-rendered (нет JS-рендеринга), httpx достаточн
|
||||||
|
|
||||||
TLS: сайт использует российский УЦ → verify=False (аналогично ekburg_permits #242).
|
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:
|
Публичный API:
|
||||||
search_documents(query, law_type, since) → list[dict]
|
search_documents(query, law_type, since) → list[dict]
|
||||||
get_document_pdf_url(doc_id) → str | None
|
get_document_pdf_url(doc_id) → str | None
|
||||||
|
|
@ -15,6 +21,7 @@ from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
|
import time
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
|
@ -28,6 +35,14 @@ _SEARCH_URL = f"{_BASE_URL}/search/"
|
||||||
# Тайм-аут щедрый: сервер иногда медленный.
|
# Тайм-аут щедрый: сервер иногда медленный.
|
||||||
_TIMEOUT = httpx.Timeout(60.0, connect=15.0)
|
_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 на сайте).
|
# Числовые коды типов документов (select#id_law_type на сайте).
|
||||||
# Текстовое значение «Постановление» — неверный параметр, сервер не матчит → 0 хитов.
|
# Текстовое значение «Постановление» — неверный параметр, сервер не матчит → 0 хитов.
|
||||||
_LAW_TYPE_POSTANOVLENIE = "1" # Постановление Правительства Свердловской области
|
_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)
|
_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(
|
def search_documents(
|
||||||
query: str,
|
query: str,
|
||||||
law_type: str = _LAW_TYPE_POSTANOVLENIE,
|
law_type: str = _LAW_TYPE_POSTANOVLENIE,
|
||||||
|
|
@ -65,8 +117,7 @@ def search_documents(
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
||||||
resp = client.get(_SEARCH_URL, params=params)
|
resp = _get_with_backoff(client, _SEARCH_URL, params=params)
|
||||||
resp.raise_for_status()
|
|
||||||
html = resp.text
|
html = resp.text
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error("pravo_gov66_client.search_documents: ошибка сети q=%r: %s", query, 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}/"
|
doc_url = f"{_BASE_URL}/{doc_id}/"
|
||||||
try:
|
try:
|
||||||
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
||||||
resp = client.get(doc_url)
|
resp = _get_with_backoff(client, doc_url)
|
||||||
resp.raise_for_status()
|
|
||||||
html = resp.text
|
html = resp.text
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
|
|
@ -151,6 +201,10 @@ def _extract_pdf_url(html: str) -> str | None:
|
||||||
def fetch_pdf(url: str) -> bytes:
|
def fetch_pdf(url: str) -> bytes:
|
||||||
"""Загружает PDF по URL.
|
"""Загружает PDF по URL.
|
||||||
|
|
||||||
|
Пейсинг + backoff на 429/5xx через ``_get_with_backoff`` (#2445 D4) — caller
|
||||||
|
(reservation_ingest) НЕ должен добавлять свой sleep перед этим вызовом, пауза
|
||||||
|
уже внутри клиента.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
url: абсолютный URL PDF-файла.
|
url: абсолютный URL PDF-файла.
|
||||||
|
|
||||||
|
|
@ -161,8 +215,7 @@ def fetch_pdf(url: str) -> bytes:
|
||||||
httpx.HTTPError: при сетевой ошибке (caller должен обработать gracefully).
|
httpx.HTTPError: при сетевой ошибке (caller должен обработать gracefully).
|
||||||
"""
|
"""
|
||||||
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
|
||||||
resp = client.get(url)
|
resp = _get_with_backoff(client, url)
|
||||||
resp.raise_for_status()
|
|
||||||
return resp.content
|
return resp.content
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -10,12 +10,17 @@ Beat НЕ настроен: постановления выходят редко
|
||||||
celery call tasks.reservation_ingest.ingest_reservations
|
celery call tasks.reservation_ingest.ingest_reservations
|
||||||
|
|
||||||
Conventions (зеркало ird_harvest.py / backend.md):
|
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():``) — битая строка не валит батч.
|
• SAVEPOINT per-row (``with db.begin_nested():``) — битая строка не валит батч.
|
||||||
• CAST(:x AS type) — НИКОГДА :x::text (psycopg v3).
|
• CAST(:x AS type) — НИКОГДА :x::text (psycopg v3).
|
||||||
• ON CONFLICT (cad_num, act_number) → обновляет метаданные + fetched_at.
|
• ON CONFLICT (cad_num, act_number) → обновляет метаданные + fetched_at.
|
||||||
• httpx.Client (не requests). Сбой одного дока не валит прогон.
|
• httpx.Client (не requests). Сбой одного дока не валит прогон.
|
||||||
• q-запросы — 2-3 слова (движок pravo.gov66 строгий AND/фраза, 1 или 4+ → 0 хитов).
|
• 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
|
from __future__ import annotations
|
||||||
|
|
@ -97,8 +102,11 @@ def _fetch_pdf_safe(pdf_url: str) -> bytes | None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _upsert_records(db: Session, records_meta: list[dict[str, Any]]) -> int:
|
def _upsert_records_in_session(db: Session, records_meta: list[dict[str, Any]]) -> int:
|
||||||
"""UPSERT записей в land_reservation. Возвращает число обработанных строк."""
|
"""UPSERT записей в land_reservation на уже открытой сессии.
|
||||||
|
|
||||||
|
Возвращает число upsert'нутых строк.
|
||||||
|
"""
|
||||||
count = 0
|
count = 0
|
||||||
for row in records_meta:
|
for row in records_meta:
|
||||||
try:
|
try:
|
||||||
|
|
@ -114,6 +122,27 @@ def _upsert_records(db: Session, records_meta: list[dict[str, Any]]) -> int:
|
||||||
return count
|
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]]:
|
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]:
|
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
|
total_parcels = 0
|
||||||
db: Session = SessionLocal()
|
for doc in docs:
|
||||||
try:
|
url = doc.get("url", "")
|
||||||
for doc in docs:
|
act_number = doc.get("act_number")
|
||||||
url = doc.get("url", "")
|
basis_act = doc.get("basis_act", "")
|
||||||
act_number = doc.get("act_number")
|
doc_purpose = doc.get("purpose")
|
||||||
basis_act = doc.get("basis_act", "")
|
doc_kind = doc.get("kind", "изъятие")
|
||||||
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)
|
pdf_bytes = _fetch_pdf_safe(url) # без открытой сессии — только сетевой запрос
|
||||||
if pdf_bytes is None:
|
if pdf_bytes is None:
|
||||||
logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url)
|
logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
records = extract_from_pdf(pdf_bytes)
|
records = extract_from_pdf(pdf_bytes)
|
||||||
logger.info("ingest_reservations(explicit): url=%s → %d КН", url, len(records))
|
logger.info("ingest_reservations(explicit): url=%s → %d КН", url, len(records))
|
||||||
if not records:
|
if not records:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
rows = [
|
rows = [
|
||||||
{
|
{
|
||||||
"cad_num": rec.cad_num,
|
"cad_num": rec.cad_num,
|
||||||
"reservation_kind": rec.reservation_kind or doc_kind,
|
"reservation_kind": rec.reservation_kind or doc_kind,
|
||||||
"basis_act": basis_act,
|
"basis_act": basis_act,
|
||||||
"act_number": rec.act_number or act_number,
|
"act_number": rec.act_number or act_number,
|
||||||
"act_date": rec.act_date,
|
"act_date": rec.act_date,
|
||||||
"purpose": rec.purpose or doc_purpose,
|
"purpose": rec.purpose or doc_purpose,
|
||||||
"doc_url": url,
|
"doc_url": url,
|
||||||
"source": "page_pdf",
|
"source": "page_pdf",
|
||||||
"raw_excerpt": rec.raw_excerpt,
|
"raw_excerpt": rec.raw_excerpt,
|
||||||
}
|
}
|
||||||
for rec in records
|
for rec in records
|
||||||
]
|
]
|
||||||
count = _upsert_records(db, rows)
|
total_parcels += _upsert_records(rows) # short-lived session: open→upsert→commit→close
|
||||||
db.commit()
|
|
||||||
total_parcels += count
|
|
||||||
|
|
||||||
logger.info(
|
|
||||||
"ingest_reservations(explicit): готово docs=%d parcels=%d",
|
|
||||||
len(docs),
|
|
||||||
total_parcels,
|
|
||||||
)
|
|
||||||
finally:
|
|
||||||
db.close()
|
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
"ingest_reservations(explicit): готово docs=%d parcels=%d",
|
||||||
|
len(docs),
|
||||||
|
total_parcels,
|
||||||
|
)
|
||||||
return {"docs": len(docs), "parcels": total_parcels}
|
return {"docs": len(docs), "parcels": total_parcels}
|
||||||
|
|
||||||
|
|
||||||
def _ingest_enumerated_docs(enumerated: list[dict[str, Any]]) -> dict[str, int]:
|
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
|
total_parcels = 0
|
||||||
fetched_docs = 0
|
fetched_docs = 0
|
||||||
db: Session = SessionLocal()
|
for hit in enumerated:
|
||||||
try:
|
doc_id: str = hit["doc_id"]
|
||||||
for hit in enumerated:
|
title: str = hit.get("title", "")
|
||||||
doc_id: str = hit["doc_id"]
|
doc_page_url: str = hit["url"]
|
||||||
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)
|
pdf_url = get_document_pdf_url(doc_id) # сеть, без открытой сессии
|
||||||
if pdf_url is None:
|
if pdf_url is None:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"ingest_reservations(enum): doc_id=%s — PDF не найден на странице", doc_id
|
"ingest_reservations(enum): doc_id=%s — PDF не найден на странице", doc_id
|
||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
pdf_bytes = _fetch_pdf_safe(pdf_url)
|
pdf_bytes = _fetch_pdf_safe(pdf_url) # сеть, без открытой сессии
|
||||||
if pdf_bytes is None:
|
if pdf_bytes is None:
|
||||||
logger.warning(
|
logger.warning("ingest_reservations(enum): doc_id=%s — не удалось скачать PDF", doc_id)
|
||||||
"ingest_reservations(enum): doc_id=%s — не удалось скачать PDF", doc_id
|
continue
|
||||||
)
|
|
||||||
continue
|
|
||||||
|
|
||||||
fetched_docs += 1
|
fetched_docs += 1
|
||||||
records = extract_from_pdf(pdf_bytes)
|
records = extract_from_pdf(pdf_bytes)
|
||||||
logger.info("ingest_reservations(enum): doc_id=%s → %d КН", doc_id, len(records))
|
logger.info("ingest_reservations(enum): doc_id=%s → %d КН", doc_id, len(records))
|
||||||
if not records:
|
if not records:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
rows = [
|
rows = [
|
||||||
{
|
{
|
||||||
"cad_num": rec.cad_num,
|
"cad_num": rec.cad_num,
|
||||||
"reservation_kind": rec.reservation_kind,
|
"reservation_kind": rec.reservation_kind,
|
||||||
"basis_act": title,
|
"basis_act": title,
|
||||||
"act_number": rec.act_number,
|
"act_number": rec.act_number,
|
||||||
"act_date": rec.act_date,
|
"act_date": rec.act_date,
|
||||||
"purpose": rec.purpose,
|
"purpose": rec.purpose,
|
||||||
"doc_url": doc_page_url,
|
"doc_url": doc_page_url,
|
||||||
"source": "pravo_gov66",
|
"source": "pravo_gov66",
|
||||||
"raw_excerpt": rec.raw_excerpt,
|
"raw_excerpt": rec.raw_excerpt,
|
||||||
}
|
}
|
||||||
for rec in records
|
for rec in records
|
||||||
]
|
]
|
||||||
count = _upsert_records(db, rows)
|
total_parcels += _upsert_records(rows) # short-lived session: open→upsert→commit→close
|
||||||
db.commit()
|
|
||||||
total_parcels += count
|
|
||||||
|
|
||||||
logger.info(
|
|
||||||
"ingest_reservations(enum): готово fetched=%d parcels=%d",
|
|
||||||
fetched_docs,
|
|
||||||
total_parcels,
|
|
||||||
)
|
|
||||||
finally:
|
|
||||||
db.close()
|
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
"ingest_reservations(enum): готово fetched=%d parcels=%d",
|
||||||
|
fetched_docs,
|
||||||
|
total_parcels,
|
||||||
|
)
|
||||||
return {"docs": fetched_docs, "parcels": total_parcels}
|
return {"docs": fetched_docs, "parcels": total_parcels}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,14 +1,19 @@
|
||||||
"""Тесты pravo_gov66_client — offline, без живой сети.
|
"""Тесты pravo_gov66_client — offline, без живой сети.
|
||||||
|
|
||||||
Покрывают: парсинг search-HTML (doc_id/title/url), парсинг страницы документа
|
Покрывают: парсинг 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
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
import pytest
|
||||||
|
|
||||||
from app.services.scrapers.pravo_gov66_client import (
|
from app.services.scrapers.pravo_gov66_client import (
|
||||||
_LAW_TYPE_POSTANOVLENIE,
|
_LAW_TYPE_POSTANOVLENIE,
|
||||||
_extract_pdf_url,
|
_extract_pdf_url,
|
||||||
|
_get_with_backoff,
|
||||||
_parse_search_html,
|
_parse_search_html,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -169,3 +174,96 @@ class TestExtractPdfUrl:
|
||||||
url = _extract_pdf_url(_DOC_HTML_ECP_ONLY)
|
url = _extract_pdf_url(_DOC_HTML_ECP_ONLY)
|
||||||
assert url is not None
|
assert url is not None
|
||||||
assert url.startswith("https://www.pravo.gov66.ru/")
|
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
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,9 @@
|
||||||
"""Тесты ingest_reservations (#1062) — offline, без живой сети и реальной БД.
|
"""Тесты ingest_reservations (#1062) — offline, без живой сети и реальной БД.
|
||||||
|
|
||||||
Мокаем: pravo_gov66_client (search/get_pdf_url/fetch_pdf) + extract_from_pdf + SessionLocal.
|
Мокаем: 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
|
from __future__ import annotations
|
||||||
|
|
@ -42,6 +44,25 @@ class _FakeDB:
|
||||||
self.closed = True
|
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 = [
|
_SAMPLE_RECORDS = [
|
||||||
|
|
@ -213,11 +234,40 @@ class TestIngestEnumerate:
|
||||||
assert params["cad_num"] == "66:41:0101001:1"
|
assert params["cad_num"] == "66:41:0101001:1"
|
||||||
assert params["source"] == "pravo_gov66"
|
assert params["source"] == "pravo_gov66"
|
||||||
|
|
||||||
def test_db_closed_after_run(self, monkeypatch: Any) -> None:
|
def test_db_closed_after_each_doc(self, monkeypatch: Any) -> None:
|
||||||
db = _FakeDB()
|
"""Session lifecycle (#2445 D4): сессия открывается/закрывается НА КАЖДЫЙ документ,
|
||||||
_patch_enumerate(monkeypatch, db, hits=[])
|
не одна на весь прогон — доказывает, что она не держится через 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()
|
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:
|
def test_skips_empty_records(self, monkeypatch: Any) -> None:
|
||||||
"""PDF без кад-номеров → ни одного execute."""
|
"""PDF без кад-номеров → ни одного execute."""
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue