From 3aa0db75749982e3f28ebef1f1db0fb13b7fd40f Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 4 Jul 2026 09:17:53 +0500 Subject: [PATCH] =?UTF-8?q?fix(gisogd66):=20=D0=BA=D0=B0=D1=80=D1=82=D0=BE?= =?UTF-8?q?=D1=87=D0=BA=D0=B8=20=D0=B1=D0=B5=D0=B7=205xx-=D1=80=D0=B5?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D0=B5=D0=B2=20+=20circuit=20breaker=20+=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B8=D0=BE=D0=B4=D0=B8=D1=87=D0=B5=D1=81?= =?UTF-8?q?=D0=BA=D0=B8=D0=B9=20commit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Холодный прод-прогон вскрыл: сегмент старых ЕКБ-ключей (~100001007xxx) детерминированно отдаёт 500 на карточках (портал-сайд, с любого IP), каждая жгла ~12с на 3 ретраях (181 фейл / 40 мин), а единственный commit в конце терял всё при soft_time_limit-kill. - fetch_doc_card: retries_5xx=0 (5xx карточки — битые записи, не транзиент); список — с прежними ретраями; сетевые ретраи не тронуты. - _sync_group: commit каждые 200 строк; load_gisogd_permits — после каждой (schema, group): партиал переживает time-limit kill. - SCHEMAS: регион (agate_sverdregion) первым — карточки стабильны, покрывает ЕКБ-доки через агрегат; ekbgo вторым (freshness). - circuit breaker: 50 подряд card-фейлов прерывают группу (битый сегмент), успех/skip сбрасывает. Новые счётчики: card_failed, group_aborted. Refs #2367 --- backend/app/services/scrapers/gisogd66.py | 89 +++++++++--- backend/tests/test_gisogd66_loader.py | 166 ++++++++++++++++++++-- 2 files changed, 227 insertions(+), 28 deletions(-) diff --git a/backend/app/services/scrapers/gisogd66.py b/backend/app/services/scrapers/gisogd66.py index a62691d8..fa2f86b1 100644 --- a/backend/app/services/scrapers/gisogd66.py +++ b/backend/app/services/scrapers/gisogd66.py @@ -13,15 +13,21 @@ WGS84 lon/lat, может отсутствовать — hasGeometry=false), approvedOrganization, docNum, dateDoc, dateReg. -Две схемы (тянем обе): ``agate_ekbgo`` (ЕКБ, свежая) и ``agate_sverdregion`` (регион- -агрегат, отстаёт ~неделю). Один документ может встретиться в обеих → бизнес-ключ -``(doc_group, doc_num)``; при конфликте предпочитаем запись с более поздним ``date_reg``. +Две схемы (тянем обе, sverdregion первым): ``agate_sverdregion`` (регион-агрегат, +стабильные карточки, покрывает и ЕКБ) и ``agate_ekbgo`` (ЕКБ, свежее на ~неделю, но +часть старых ключей битая на портале). Один документ может встретиться в обеих → +бизнес-ключ ``(doc_group, doc_num)``; при конфликте предпочитаем запись с более +поздним ``date_reg``. Инкрементальность: список группы сверяется с БД по (source_schema, source_key, date_reg); карточки (второй запрос — дорогой) тянутся ТОЛЬКО для новых / изменившихся документов. -Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на 5xx, -малый sleep между карточками. UPSERT — per-row SAVEPOINT (битая строка не валит батч). +Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на +сетевых ошибках. Карточки (5xx-битые записи портала) НЕ ретраятся — deterministically +broken, не транзиент. Circuit breaker: 50 карточек подряд с ошибкой → прерываем группу +(битый сегмент). UPSERT — per-row SAVEPOINT (битая строка не валит батч). Периодический +db.commit() (каждые 200 записей + после каждой schema/group), чтобы партиал пережил +soft_time_limit-kill. """ from __future__ import annotations @@ -46,17 +52,26 @@ logger = logging.getLogger(__name__) GISOGD_BASE = "https://gisogd66.midural.ru/api/v1" _SECTION = "razdel13" -# Схемы источника — обе тянем (ekbgo свежее, sverdregion — регион-агрегат). -SCHEMAS: tuple[str, ...] = ("agate_ekbgo", "agate_sverdregion") +# Схемы источника — обе тянем. Порядок ВАЖЕН: sverdregion первым. +# sverdregion (регион-агрегат) отдаёт карточки стабильно и покрывает в т.ч. ЕКБ-доки +# через агрегацию, поэтому он — базовый источник. ekbgo вторым как freshness-энхансер +# (свежее на ~неделю), НО у него большой сегмент старых ключей (префикс ~100001007xxx) +# детерминированно отдаёт 500 на карточках (сломано на стороне портала, 2026-07-04) — +# гоняем его после того, как регион уже покрыл основную массу (см. circuit breaker ниже). +SCHEMAS: tuple[str, ...] = ("agate_sverdregion", "agate_ekbgo") # Группа источника (doc group key) → наш doc_group-код в БД (CHECK IN ('RS','RV')). GROUP_CODE: dict[str, str] = {"DocRS": "RS", "DocRV": "RV"} _HTTP_TIMEOUT = 20.0 _MAX_CONNECTIONS = 5 -_MAX_RETRIES = 2 # ретраи на 5xx (helper делает суммарно до 1 + _MAX_RETRIES попыток) +_MAX_RETRIES = 2 # ретраи (helper делает суммарно до 1 + _MAX_RETRIES попыток) _RETRY_BACKOFF_S = 1.5 _CARD_SLEEP_S = 0.05 # вежливая пауза между карточками (второй, «дорогой» запрос) +_COMMIT_EVERY = 200 # периодический commit внутри группы (партиал переживает time-limit) +# Circuit breaker: столько карточек ПОДРЯД с ошибкой fetch → прерываем группу (портал +# отдаёт битый сегмент старых ключей детерминированным 5xx; остаток догоним будущими ранами). +_CARD_FAIL_BREAKER = 50 USER_AGENT = "GenDesign/1.0 (+https://gendsgn.ru) Site Finder gisogd permits loader" @@ -78,20 +93,25 @@ def _make_client() -> httpx.Client: ) -def _get_json(client: httpx.Client, path: str) -> Any: +def _get_json(client: httpx.Client, path: str, *, retries_5xx: int = _MAX_RETRIES) -> Any: """GET path → JSON с ретраями на 5xx / сетевых ошибках. - До (1 + _MAX_RETRIES) попыток с линейным backoff. 4xx → сразу raise (не ретраим). + Сетевые Timeout/TransportError ретраятся всегда (до 1 + _MAX_RETRIES попыток) — + это транзиент. 5xx ретраятся ``retries_5xx`` раз: списки группы — с дефолтом + (транзиентная нагрузка портала), карточки — с ``retries_5xx=0`` (5xx карточки + детерминированно битые записи сегмента портала, не транзиент). 4xx → сразу raise. """ last_exc: Exception | None = None for attempt in range(_MAX_RETRIES + 1): try: resp = client.get(path) if resp.status_code >= 500: - # 5xx — транзиентная ошибка сервера, ретраим. last_exc = httpx.HTTPStatusError( f"{resp.status_code} on {path}", request=resp.request, response=resp ) + # 5xx: ретраим только пока attempt в пределах retries_5xx (0 → сразу raise). + if attempt >= retries_5xx: + raise last_exc logger.warning( "gisogd66: %s (попытка %d) на %s", resp.status_code, attempt + 1, path ) @@ -128,8 +148,11 @@ def fetch_doc_card(client: httpx.Client, schema: str, key: str | int) -> dict[st Поля: spatialUnit, infosetGeoJson, hasGeometry, approvedOrganization, docNum, dateDoc, dateReg, name. + + 5xx карточки НЕ ретраятся (retries_5xx=0): битый сегмент старых ключей портала + отдаёт детерминированный 500 — 3 попытки с backoff = ~12с потерянного времени/карточку. """ - data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}") + data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}", retries_5xx=0) if not isinstance(data, dict): logger.warning("gisogd66: карточка %s/%s не-dict (%s)", schema, key, type(data).__name__) return None @@ -304,7 +327,7 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]: SessionLocal. Returns: счётчики {listed, fetched_details, inserted, updated, skipped_unchanged, - no_cad, no_geom}. + no_cad, no_geom, card_failed, group_aborted}. """ owns_session = db is None if db is None: @@ -318,6 +341,8 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]: "skipped_unchanged": 0, "no_cad": 0, "no_geom": 0, + "card_failed": 0, # каждая упавшая карточка (5xx / сеть / не-dict) + "group_aborted": 0, # групп прервано circuit breaker'ом (битый сегмент портала) } client = _make_client() try: @@ -331,7 +356,9 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]: continue counts["listed"] += len(items) _sync_group(db, client, schema, doc_group, items, existing, counts) - db.commit() + # Commit после КАЖДОЙ (schema, group), а не один раз в конце: партиал + # должен пережить soft_time_limit-kill (иначе теряем весь прогон). + db.commit() except Exception as exc: db.rollback() logger.exception("gisogd66: outer tx rolled back: %s", exc) @@ -354,7 +381,15 @@ def _sync_group( existing: dict[str, date | None], counts: dict[str, int], ) -> None: - """Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT.""" + """Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT. + + Периодический commit каждые ``_COMMIT_EVERY`` обработанных записей (партиал переживёт + time-limit kill). Circuit breaker: ``_CARD_FAIL_BREAKER`` карточек ПОДРЯД с ошибкой + fetch → прерываем группу (битый сегмент портала — остаток догоним будущими ранами). + Любая успешная карточка сбрасывает счётчик подряд-фейлов. + """ + processed = 0 # успешно обработанных (upsert/skip) с последнего commit + consecutive_fail = 0 # карточек подряд с ошибкой fetch (для circuit breaker) for item in items: key = item.get("key") if key is None: @@ -365,17 +400,36 @@ def _sync_group( # Инкрементальность: пропускаем карточку, если key уже есть И date_reg совпал. if source_key in existing and existing[source_key] == list_reg: counts["skipped_unchanged"] += 1 + consecutive_fail = 0 # skip — не фейл карточки, сбрасываем breaker + processed += 1 + if processed % _COMMIT_EVERY == 0: + db.commit() continue + card: dict[str, Any] | None try: card = fetch_doc_card(client, schema, source_key) except Exception as exc: logger.warning("gisogd66: карточка %s/%s упала: %s", schema, source_key, exc) - continue + card = None finally: time.sleep(_CARD_SLEEP_S) if card is None: + # Неуспех fetch (5xx / сеть / не-dict): считаем и двигаем circuit breaker. + counts["card_failed"] += 1 + consecutive_fail += 1 + if consecutive_fail >= _CARD_FAIL_BREAKER: + logger.error( + "gisogd66: %s/%s: %d карточек подряд 5xx — прерываю группу " + "(битый сегмент портала)", + schema, + doc_group, + _CARD_FAIL_BREAKER, + ) + counts["group_aborted"] += 1 + return continue + consecutive_fail = 0 # успешная карточка сбрасывает breaker counts["fetched_details"] += 1 rec = _build_record(item, card, schema, doc_group, source_key, list_reg) @@ -393,6 +447,9 @@ def _sync_group( logger.warning("gisogd66: upsert %s %r упал: %s", doc_group, rec["doc_num"], exc) continue counts[outcome] += 1 + processed += 1 + if processed % _COMMIT_EVERY == 0: + db.commit() def _build_record( diff --git a/backend/tests/test_gisogd66_loader.py b/backend/tests/test_gisogd66_loader.py index 058eec20..7c99aa86 100644 --- a/backend/tests/test_gisogd66_loader.py +++ b/backend/tests/test_gisogd66_loader.py @@ -18,6 +18,8 @@ from contextlib import contextmanager from datetime import date from typing import Any +import httpx + from app.services.scrapers import gisogd66 as g # ── Реальные фрагменты API (разведка 2026-07-04) ────────────────────────────── @@ -221,6 +223,7 @@ class _FakeSession: self._scalar = scalar_value self._select_rows = select_rows or [] self.committed = False + self.commit_count = 0 @contextmanager def begin_nested(self): # type: ignore[no-untyped-def] @@ -235,6 +238,7 @@ class _FakeSession: def commit(self) -> None: self.committed = True + self.commit_count += 1 def rollback(self) -> None: pass @@ -284,6 +288,23 @@ def test_upsert_permit_no_geom_passes_null() -> None: # ── Инкрементальность через _sync_group ─────────────────────────────────────── +_COUNT_KEYS = ( + "inserted", + "updated", + "skipped_unchanged", + "fetched_details", + "no_cad", + "no_geom", + "card_failed", + "group_aborted", +) + + +def _fresh_counts() -> dict[str, int]: + """Свежий словарь счётчиков _sync_group (все ключи, что лоадер инкрементит).""" + return dict.fromkeys(_COUNT_KEYS, 0) + + class _RecordingClient: """Заглушка httpx-клиента: считает, сколько карточек запрошено (fetch_doc_card).""" @@ -302,9 +323,7 @@ def test_sync_group_incremental_skip(monkeypatch: Any) -> None: _patch_card_fetch(monkeypatch, fetched) db = _FakeSession(scalar_value=True) existing = {"1000210001947595": date(2026, 6, 19)} # тот же dateReg, что в списке - counts = dict.fromkeys( - ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0 - ) + counts = _fresh_counts() g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts) # type: ignore[arg-type] assert fetched == [] # карточку не тянули assert counts["skipped_unchanged"] == 1 @@ -317,9 +336,7 @@ def test_sync_group_changed_datereg_fetches(monkeypatch: Any) -> None: _patch_card_fetch(monkeypatch, fetched) db = _FakeSession(scalar_value=False) # update existing = {"1000210001947595": date(2026, 6, 10)} # старее списочного 06-19 - counts = dict.fromkeys( - ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0 - ) + counts = _fresh_counts() g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts) # type: ignore[arg-type] assert fetched == ["1000210001947595"] # карточку тянули assert counts["fetched_details"] == 1 @@ -331,9 +348,7 @@ def test_sync_group_new_key_inserts_and_counts_geom(monkeypatch: Any) -> None: fetched: list[Any] = [] _patch_card_fetch(monkeypatch, fetched) db = _FakeSession(scalar_value=True) # insert - counts = dict.fromkeys( - ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0 - ) + counts = _fresh_counts() g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], {}, counts) # type: ignore[arg-type] assert counts["inserted"] == 1 assert counts["fetched_details"] == 1 @@ -350,15 +365,142 @@ def test_sync_group_no_cad_no_geom_counts(monkeypatch: Any) -> None: monkeypatch.setattr(g, "fetch_doc_card", _fake) db = _FakeSession(scalar_value=True) item = {**_LIST_ITEM, "key": 42, "dateReg": "2026-06-01T10:00:00"} - counts = dict.fromkeys( - ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0 - ) + counts = _fresh_counts() g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RV", [item], {}, counts) # type: ignore[arg-type] assert counts["no_cad"] == 1 assert counts["no_geom"] == 1 assert counts["inserted"] == 1 +# ── Хотфикс #2370: 5xx-карточки без ретраев, circuit breaker, периодический commit ── + + +class _FakeResponse: + """Мини-заглушка httpx.Response: только status_code + json/raise_for_status/request.""" + + def __init__(self, status_code: int, payload: Any = None) -> None: + self.status_code = status_code + self._payload = payload + self.request = httpx.Request("GET", "https://example/x") + + def json(self) -> Any: + return self._payload + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise httpx.HTTPStatusError( + f"{self.status_code}", + request=self.request, + response=None, # type: ignore[arg-type] + ) + + +class _CountingClient: + """Заглушка httpx.Client: считает GET-и, всегда отдаёт заданный status_code.""" + + def __init__(self, status_code: int, payload: Any = None) -> None: + self.status_code = status_code + self.payload = payload + self.get_calls: list[str] = [] + + def get(self, path: str) -> _FakeResponse: + self.get_calls.append(path) + return _FakeResponse(self.status_code, self.payload) + + +def test_fetch_doc_card_500_single_attempt(monkeypatch: Any) -> None: + """Хотфикс: карточка отдаёт 500 → ровно 1 HTTP-попытка (retries_5xx=0), затем raise. + + Битый сегмент старых ключей портала детерминированно 5xx — ретраи = ~12с/карточку + впустую. Убираем sleep, чтоб тест был мгновенным. + """ + monkeypatch.setattr(g.time, "sleep", lambda _s: None) + client = _CountingClient(status_code=500) + raised = False + try: + g.fetch_doc_card(client, "agate_ekbgo", "100001007000123") # type: ignore[arg-type] + except httpx.HTTPStatusError: + raised = True + assert raised is True + assert len(client.get_calls) == 1 # НИ одного ретрая на 5xx карточки + + +def test_fetch_group_list_500_retries(monkeypatch: Any) -> None: + """Список группы 5xx ретраится с дефолтом (1 + _MAX_RETRIES попыток) — контраст с карточкой.""" + monkeypatch.setattr(g.time, "sleep", lambda _s: None) + client = _CountingClient(status_code=500) + raised = False + try: + g.fetch_group_list(client, "agate_ekbgo", "DocRS") # type: ignore[arg-type] + except httpx.HTTPStatusError: + raised = True + assert raised is True + assert len(client.get_calls) == 1 + g._MAX_RETRIES # 3 попытки + + +def test_sync_group_circuit_breaker_aborts(monkeypatch: Any) -> None: + """50 карточек ПОДРЯД с 5xx → группа прервана: group_aborted=1, остаток НЕ запрошен.""" + monkeypatch.setattr(g.time, "sleep", lambda _s: None) + requested: list[Any] = [] + + def _always_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None: + requested.append(key) + raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None) # type: ignore[arg-type] + + monkeypatch.setattr(g, "fetch_doc_card", _always_fail) + db = _FakeSession(scalar_value=True) + # 60 новых ключей — breaker (50) должен сработать раньше, чем дойдём до 60-го. + items = [{**_LIST_ITEM, "key": 100001007000000 + i} for i in range(60)] + counts = _fresh_counts() + g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type] + assert counts["group_aborted"] == 1 + assert counts["card_failed"] == g._CARD_FAIL_BREAKER # ровно 50 фейлов до аборта + assert len(requested) == g._CARD_FAIL_BREAKER # 51-й ключ НЕ запрашивался + assert counts["inserted"] == 0 + + +def test_sync_group_success_resets_breaker(monkeypatch: Any) -> None: + """Успешная карточка между фейлами сбрасывает breaker → группа НЕ прерывается. + + 49 фейлов, потом успех (reset), потом ещё 49 фейлов — суммарно 98 фейлов, но НИ разу + 50 подряд → group_aborted=0, вся группа обработана. + """ + monkeypatch.setattr(g.time, "sleep", lambda _s: None) + # Индексы 49 и 99 (0-based) — успех; остальные — фейл. Между ними ≤49 подряд. + good_idx = {49, 99} + seq: list[int] = [] + + def _mostly_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None: + idx = len(seq) + seq.append(idx) + if idx in good_idx: + return dict(_CARD) + raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None) # type: ignore[arg-type] + + monkeypatch.setattr(g, "fetch_doc_card", _mostly_fail) + db = _FakeSession(scalar_value=True) + items = [{**_LIST_ITEM, "key": 200000000 + i} for i in range(100)] + counts = _fresh_counts() + g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type] + assert counts["group_aborted"] == 0 # breaker сбрасывался — не прервались + assert counts["card_failed"] == 98 # 100 − 2 успешных + assert counts["fetched_details"] == 2 # 2 успешные карточки + assert len(seq) == 100 # прошли ВСЮ группу + + +def test_sync_group_periodic_commit(monkeypatch: Any) -> None: + """>200 обработанных записей → db.commit() вызывается периодически (каждые 200).""" + monkeypatch.setattr(g.time, "sleep", lambda _s: None) + _patch_card_fetch(monkeypatch, []) + db = _FakeSession(scalar_value=True) # каждый upsert → inserted + # 450 новых ключей → 2 периодических commit (на 200-й и 400-й записи). + items = [{**_LIST_ITEM, "key": 300000000 + i} for i in range(450)] + counts = _fresh_counts() + g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts) # type: ignore[arg-type] + assert counts["inserted"] == 450 + assert db.commit_count == 2 # 450 // 200 = 2 периодических commit + + # ── Конфликт двух схем: поздний date_reg параметризуется в UPSERT ────────────── -- 2.45.3