From 464519c69bf4f341933dc5aa18d62c47b3bbacc6 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 5 Jul 2026 20:09:12 +0000 Subject: [PATCH] fix(workers): izyatie per-doc isolation + newbuilding-crossload breadcrumb (#2445-D5/D6) (#2460) --- backend/app/api/v1/admin_scrape.py | 2 +- .../tasks/etl_newbuilding_crossload.py | 101 ++++++++- .../app/workers/tasks/izyatie_ocr_ingest.py | 73 ++++--- .../services/scrapers/test_izyatie_ocr.py | 200 +++++++++++++++++- .../tasks/test_etl_newbuilding_crossload.py | 157 ++++++++++++++ 5 files changed, 498 insertions(+), 35 deletions(-) create mode 100644 backend/tests/workers/tasks/test_etl_newbuilding_crossload.py diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 5420f902..b905d216 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1152,7 +1152,7 @@ def trigger_newbuilding_crossload() -> dict[str, Any]: ) from app.workers.tasks.etl_newbuilding_crossload import etl_newbuilding_crossload - result = etl_newbuilding_crossload.apply_async() + result = etl_newbuilding_crossload.apply_async(kwargs={"triggered_by": "manual"}) return {"task_id": result.id, "queued_at": "now"} diff --git a/backend/app/workers/tasks/etl_newbuilding_crossload.py b/backend/app/workers/tasks/etl_newbuilding_crossload.py index b131f16a..a10ceadb 100644 --- a/backend/app/workers/tasks/etl_newbuilding_crossload.py +++ b/backend/app/workers/tasks/etl_newbuilding_crossload.py @@ -3,25 +3,116 @@ Расписание: 03:30 МСК (hardcoded в beat_schedule, как cbr/poi/supply-layers). Идемпотентен: UPSERT ON CONFLICT (source, ext_house_id). Если TRADEIN_DATABASE_URL не задан → warn-log, возвращает {"disabled": True}. + +Breadcrumb (issue #2445 D6): прогон логируется в objective_scrape_runs (тот же +журнал, что используют соседние nightly/weekly-таски — см. +app/workers/tasks/objective_etl.py) с group_name='newbuilding-crossload', чтобы +сбой был виден оператору в admin-таблице логов, а не только в Celery FAILURE + +GlitchTip. Таблица переиспользуется как есть — без новой миграции; поля +rows_lots/rows_history/requests_count не про Objective-домен здесь, а про +source_rows/upserted/skipped cross-load'а (см. комментарии у _finish_run). """ from __future__ import annotations import logging +from typing import Any +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal from app.services.etl.newbuilding_crossload import run_crossload from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) +_GROUP_NAME = "newbuilding-crossload" + + +def _start_run(db: Session, triggered_by: str) -> int: + """Создать breadcrumb-строку в objective_scrape_runs, статус 'running'.""" + row = db.execute( + text( + """ + INSERT INTO objective_scrape_runs (group_name, triggered_by, status) + VALUES (:gn, :tb, 'running') + RETURNING run_id + """ + ), + {"gn": _GROUP_NAME, "tb": triggered_by}, + ).scalar_one() + db.commit() + return int(row) + + +def _finish_run( + db: Session, run_id: int, *, status: str, error: str | None = None, **counts: int +) -> None: + """Обновить breadcrumb-строку финальным статусом ('done' | 'failed') + счётчиками.""" + sets = ["finished_at = NOW()", "status = :status", "error = :error"] + params: dict[str, Any] = {"rid": run_id, "status": status, "error": error} + for k, v in counts.items(): + sets.append(f"{k} = :{k}") + params[k] = v + db.execute( + text(f"UPDATE objective_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"), + params, + ) + db.commit() + @celery_app.task(name="tasks.etl_newbuilding_crossload.etl_newbuilding_crossload") -def etl_newbuilding_crossload() -> dict: +def etl_newbuilding_crossload(triggered_by: str = "beat") -> dict: """Cross-load tradein.houses → gendesign.newbuilding_listings (#976). Вызывается beat'ом в 03:30 МСК и вручную через POST /api/v1/admin/scrape/newbuilding-crossload. + + Args: + triggered_by: 'beat' | 'manual' | 'api' — для журнала objective_scrape_runs. + + Breadcrumb-строка в objective_scrape_runs логирует start/success/failure — + на исключении из run_crossload() помечает run 'failed' и ре-рейзит, чтобы + Celery ушёл в FAILURE и GlitchTip продолжил ловить сбой (issue #2445 D6). """ - logger.info("task etl_newbuilding_crossload: start") - result = run_crossload() - logger.info("task etl_newbuilding_crossload: done — %s", result) - return result + db = SessionLocal() + run_id: int | None = None + try: + run_id = _start_run(db, triggered_by) + logger.info("task etl_newbuilding_crossload: start run_id=%s", run_id) + + result = run_crossload(db=db) + + if result.get("disabled"): + _finish_run(db, run_id, status="done", error="TRADEIN_DATABASE_URL not set") + logger.info("task etl_newbuilding_crossload: done (disabled) run_id=%s", run_id) + return {"run_id": run_id, **result} + + _finish_run( + db, + run_id, + status="done", + error=None, + rows_history=result.get("source_rows", 0), + rows_lots=result.get("upserted", 0), + requests_count=result.get("skipped", 0), + ) + logger.info("task etl_newbuilding_crossload: done run_id=%s — %s", run_id, result) + return {"run_id": run_id, **result} + + except Exception as e: + logger.exception("task etl_newbuilding_crossload: failed run_id=%s: %s", run_id, e) + if run_id is not None: + try: + _finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}") + except Exception: + # Breadcrumb-flush не должен скрыть исходное исключение — Celery + # всё равно должен уйти в FAILURE + GlitchTip его поймает. + logger.exception( + "task etl_newbuilding_crossload: failed to write failure breadcrumb " + "for run_id=%s", + run_id, + ) + raise + finally: + db.close() diff --git a/backend/app/workers/tasks/izyatie_ocr_ingest.py b/backend/app/workers/tasks/izyatie_ocr_ingest.py index d3374d3f..b96952b3 100644 --- a/backend/app/workers/tasks/izyatie_ocr_ingest.py +++ b/backend/app/workers/tasks/izyatie_ocr_ingest.py @@ -151,42 +151,59 @@ def ingest_izyatie_ocr() -> dict[str, int]: logger.info("ingest_izyatie_ocr: загружаем %r → %s", doc_title[:60], doc_url) + # Весь per-document body (fetch → OCR → extract → upsert → commit) — в + # одном try/except. Сбой ЛЮБОГО шага (битый PDF на fetch, Tesseract-crash + # на OCR, баг парсера на неожиданном OCR-тексте, ошибка upsert/commit) + # логируется и переходит к следующему документу вместо падения всей + # еженедельной пачки (issue #2445 D5). db.rollback() откатывает + # незакоммиченные изменения текущей итерации, чтобы они не отравили + # commit следующего документа. try: pdf_bytes = fetch_pdf(doc_url) + + ocr_text = ocr_pdf_text(pdf_bytes) + if not ocr_text.strip(): + logger.warning( + "ingest_izyatie_ocr: OCR вернул пустой текст для %s " + "(tesseract не установлен, битый PDF или чистый лист?)", + doc_url, + ) + processed_docs += 1 + continue + + records = extract_izyatie_records(ocr_text, doc_title, doc_url) + logger.info( + "ingest_izyatie_ocr: %r → %d кад-номеров", + doc_title[:60], + len(records), + ) + + if not records: + processed_docs += 1 + continue + + count = _upsert_records(db, records) + db.commit() + total_parcels += count + processed_docs += 1 + except Exception as exc: - logger.error( - "ingest_izyatie_ocr: не удалось загрузить PDF %s: %s", + logger.warning( + "ingest_izyatie_ocr: пропуск документа %r (%s): %s", + doc_title[:60], doc_url, exc, ) + try: + db.rollback() + except Exception as rollback_exc: + logger.warning( + "ingest_izyatie_ocr: rollback после сбоя документа %s тоже упал: %s", + doc_url, + rollback_exc, + ) continue - ocr_text = ocr_pdf_text(pdf_bytes) - if not ocr_text.strip(): - logger.warning( - "ingest_izyatie_ocr: OCR вернул пустой текст для %s " - "(tesseract не установлен, битый PDF или чистый лист?)", - doc_url, - ) - processed_docs += 1 - continue - - records = extract_izyatie_records(ocr_text, doc_title, doc_url) - logger.info( - "ingest_izyatie_ocr: %r → %d кад-номеров", - doc_title[:60], - len(records), - ) - - if not records: - processed_docs += 1 - continue - - count = _upsert_records(db, records) - db.commit() - total_parcels += count - processed_docs += 1 - finally: db.close() diff --git a/backend/tests/services/scrapers/test_izyatie_ocr.py b/backend/tests/services/scrapers/test_izyatie_ocr.py index c8f68478..76ae412b 100644 --- a/backend/tests/services/scrapers/test_izyatie_ocr.py +++ b/backend/tests/services/scrapers/test_izyatie_ocr.py @@ -174,7 +174,7 @@ class TestExtractIzyatieRecords: def test_raw_excerpt_for_first_record_only(self) -> None: """raw_excerpt заполняется только для первой записи.""" - text = "Изъять: 66:41:0101001:123 участок 1;\n" "66:41:0101001:456 участок 2." + text = "Изъять: 66:41:0101001:123 участок 1;\n66:41:0101001:456 участок 2." records = extract_izyatie_records(text, self._DOC_TITLE, self._DOC_URL) assert len(records) == 2 assert records[0]["raw_excerpt"] is not None @@ -506,3 +506,201 @@ class TestIngestIzyatieOcr: # Первый документ пропущен (ошибка), второй обработан (пустой OCR). assert result["docs"] == 1 assert result["parcels"] == 0 + + def test_ocr_failure_on_one_doc_does_not_abort_batch(self) -> None: + """OCR (Tesseract) падает на документе 2 из 3 — документы 1 и 3 + всё равно обрабатываются и коммитятся (issue #2445 D5). + + Раньше только fetch_pdf был защищён try/except — падение + ocr_pdf_text пробрасывалось из цикла и валило остаток пачки. + """ + from app.workers.tasks.izyatie_ocr_ingest import ingest_izyatie_ocr + + mock_docs = [ + {"title": "Документ 1", "url": "https://example.com/file/1"}, + {"title": "Документ 2 (сломанный OCR)", "url": "https://example.com/file/2"}, + {"title": "Документ 3", "url": "https://example.com/file/3"}, + ] + + def ocr_side_effect(pdf_bytes: bytes) -> str: + if pdf_bytes == b"pdf-2": + raise Exception("Tesseract crashed on malformed PDF") + return "66:41:0101001:123" + + def fetch_side_effect(url: str) -> bytes: + return b"pdf-" + url.rsplit("/", 1)[-1].encode() + + mock_records = [ + { + "cad_num": "66:41:0101001:123", + "reservation_kind": "изъятие", + "basis_act": "test", + "act_number": None, + "act_date": None, + "purpose": None, + "doc_url": "https://example.com/file/x", + "source": "izyatie_ekb_ocr", + "raw_excerpt": None, + } + ] + + mock_db = MagicMock() + mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + + with ( + patch( + "app.services.scrapers.izyatie_client.list_izyatie_documents", + return_value=mock_docs, + ), + patch( + "app.services.scrapers.izyatie_client.fetch_pdf", + side_effect=fetch_side_effect, + ), + patch( + "app.services.scrapers.izyatie_ocr.ocr_pdf_text", + side_effect=ocr_side_effect, + ), + patch( + "app.services.scrapers.izyatie_ocr.extract_izyatie_records", + return_value=mock_records, + ), + patch("app.workers.tasks.izyatie_ocr_ingest.SessionLocal", return_value=mock_db), + ): + result = ingest_izyatie_ocr() + + # Документ 2 упал на OCR и не засчитан; 1 и 3 успешно обработаны и закоммичены. + assert result["docs"] == 2 + assert result["parcels"] == 2 + # db.commit() вызывался ровно для 1 и 3 (документ 2 не дошёл до commit). + assert mock_db.commit.call_count == 2 + # rollback() вызван один раз — после сбоя документа 2. + assert mock_db.rollback.call_count == 1 + + def test_extract_failure_on_one_doc_does_not_abort_batch(self) -> None: + """Баг парсера в extract_izyatie_records на документе 2 из 3 — документы + 1 и 3 всё равно обрабатываются (issue #2445 D5).""" + from app.workers.tasks.izyatie_ocr_ingest import ingest_izyatie_ocr + + mock_docs = [ + {"title": "Документ 1", "url": "https://example.com/file/1"}, + {"title": "Документ 2 (баг парсера)", "url": "https://example.com/file/2"}, + {"title": "Документ 3", "url": "https://example.com/file/3"}, + ] + + good_record = [ + { + "cad_num": "66:41:0101001:999", + "reservation_kind": "изъятие", + "basis_act": "test", + "act_number": None, + "act_date": None, + "purpose": None, + "doc_url": "https://example.com/file/x", + "source": "izyatie_ekb_ocr", + "raw_excerpt": None, + } + ] + + def extract_side_effect(ocr_text: str, doc_title: str, doc_url: str) -> list[dict]: + if doc_url.endswith("/2"): + raise ValueError("unexpected OCR text shape crashed the parser") + return good_record + + mock_db = MagicMock() + mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + + with ( + patch( + "app.services.scrapers.izyatie_client.list_izyatie_documents", + return_value=mock_docs, + ), + patch( + "app.services.scrapers.izyatie_client.fetch_pdf", + return_value=b"fake pdf", + ), + patch( + "app.services.scrapers.izyatie_ocr.ocr_pdf_text", + return_value="66:41:0101001:999", + ), + patch( + "app.services.scrapers.izyatie_ocr.extract_izyatie_records", + side_effect=extract_side_effect, + ), + patch("app.workers.tasks.izyatie_ocr_ingest.SessionLocal", return_value=mock_db), + ): + result = ingest_izyatie_ocr() + + assert result["docs"] == 2 + assert result["parcels"] == 2 + assert mock_db.commit.call_count == 2 + assert mock_db.rollback.call_count == 1 + + def test_upsert_failure_on_one_doc_does_not_abort_batch(self) -> None: + """db.commit()/upsert падает на документе 2 из 3 — документы 1 и 3 + всё равно обрабатываются и коммитятся, а session откатывается после + сбоя вместо отравления следующего commit (issue #2445 D5).""" + from app.workers.tasks.izyatie_ocr_ingest import ingest_izyatie_ocr + + mock_docs = [ + {"title": "Документ 1", "url": "https://example.com/file/1"}, + {"title": "Документ 2 (сбой commit)", "url": "https://example.com/file/2"}, + {"title": "Документ 3", "url": "https://example.com/file/3"}, + ] + + record = [ + { + "cad_num": "66:41:0101001:1", + "reservation_kind": "изъятие", + "basis_act": "test", + "act_number": None, + "act_date": None, + "purpose": None, + "doc_url": "https://example.com/file/x", + "source": "izyatie_ekb_ocr", + "raw_excerpt": None, + } + ] + + commit_calls = 0 + + def commit_side_effect() -> None: + nonlocal commit_calls + commit_calls += 1 + if commit_calls == 2: + raise Exception("db commit failed: connection reset") + + mock_db = MagicMock() + mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + mock_db.commit.side_effect = commit_side_effect + + with ( + patch( + "app.services.scrapers.izyatie_client.list_izyatie_documents", + return_value=mock_docs, + ), + patch( + "app.services.scrapers.izyatie_client.fetch_pdf", + return_value=b"fake pdf", + ), + patch( + "app.services.scrapers.izyatie_ocr.ocr_pdf_text", + return_value="66:41:0101001:1", + ), + patch( + "app.services.scrapers.izyatie_ocr.extract_izyatie_records", + return_value=record, + ), + patch("app.workers.tasks.izyatie_ocr_ingest.SessionLocal", return_value=mock_db), + ): + result = ingest_izyatie_ocr() + + # Документ 2 не засчитан (commit упал), 1 и 3 обработаны успешно. + assert result["docs"] == 2 + assert result["parcels"] == 2 + # commit() вызван 3 раза (для каждого документа), 2-й вызов упал. + assert mock_db.commit.call_count == 3 + # rollback() вызван ровно один раз — после сбойного commit документа 2. + assert mock_db.rollback.call_count == 1 diff --git a/backend/tests/workers/tasks/test_etl_newbuilding_crossload.py b/backend/tests/workers/tasks/test_etl_newbuilding_crossload.py new file mode 100644 index 00000000..3b609298 --- /dev/null +++ b/backend/tests/workers/tasks/test_etl_newbuilding_crossload.py @@ -0,0 +1,157 @@ +"""Тесты breadcrumb-логирования etl_newbuilding_crossload task (issue #2445 D6). + +Раньше эта таска не имела ни try/except, ни breadcrumb-записи в отличие от всех +соседних nightly/weekly-тасок в этой директории (objective_etl.py и т.д.) — +сбой run_crossload() был виден только в Celery FAILURE + GlitchTip, но не в +operational-логе (objective_scrape_runs), который проверяет оператор. + +Покрытие: +- success-path: run_id создаётся, run_crossload() вызывается с db=, breadcrumb + помечается 'done' со счётчиками source_rows/upserted/skipped. +- disabled-путь (TRADEIN_DATABASE_URL пуст): breadcrumb 'done', error поясняет причину. +- failure-path: run_crossload() бросает исключение → breadcrumb 'failed' + + error, задача ре-рейзит исключение (Celery должен уйти в FAILURE). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + + +def _make_db_mock(run_id: int = 7) -> MagicMock: + """Mock SessionLocal() — INSERT ... RETURNING run_id возвращает run_id.""" + db = MagicMock() + db.execute.return_value.scalar_one.return_value = run_id + return db + + +def _capture_finish_run(mod: Any) -> tuple[list[dict[str, Any]], Any]: + """Оборачивает mod._finish_run, чтобы записывать все вызовы (kwargs).""" + calls: list[dict[str, Any]] = [] + orig = mod._finish_run + + def capture(_db: Any, _run_id: int, **kwargs: Any) -> None: + calls.append({"run_id": _run_id, **kwargs}) + orig(_db, _run_id, **kwargs) + + return calls, capture + + +def test_success_writes_done_breadcrumb_with_counts() -> None: + """Успешный прогон: run_id создан, breadcrumb → 'done' с source_rows/upserted/skipped.""" + from app.workers.tasks import etl_newbuilding_crossload as mod + + db = _make_db_mock(run_id=7) + finish_calls, capture = _capture_finish_run(mod) + + with ( + patch.object(mod, "SessionLocal", return_value=db), + patch.object( + mod, + "run_crossload", + return_value={"source_rows": 100, "upserted": 95, "skipped": 5}, + ) as mock_run_crossload, + patch.object(mod, "_finish_run", side_effect=capture), + ): + result = mod.etl_newbuilding_crossload.run(triggered_by="manual") + + assert result["run_id"] == 7 + assert result["source_rows"] == 100 + assert result["upserted"] == 95 + assert result["skipped"] == 5 + + # run_crossload вызван с нашей db-сессией (переиспользуем транзакцию/savepoints). + mock_run_crossload.assert_called_once_with(db=db) + + assert len(finish_calls) == 1 + assert finish_calls[0]["run_id"] == 7 + assert finish_calls[0]["status"] == "done" + assert finish_calls[0]["error"] is None + assert finish_calls[0]["rows_history"] == 100 + assert finish_calls[0]["rows_lots"] == 95 + assert finish_calls[0]["requests_count"] == 5 + + db.close.assert_called_once() + + +def test_disabled_writes_done_breadcrumb_with_explanation() -> None: + """TRADEIN_DATABASE_URL не задан → run_crossload вернул {"disabled": True}; + breadcrumb всё равно 'done' (ожидаемое поведение, не ошибка), error поясняет причину.""" + from app.workers.tasks import etl_newbuilding_crossload as mod + + db = _make_db_mock(run_id=11) + finish_calls, capture = _capture_finish_run(mod) + + with ( + patch.object(mod, "SessionLocal", return_value=db), + patch.object(mod, "run_crossload", return_value={"disabled": True}), + patch.object(mod, "_finish_run", side_effect=capture), + ): + result = mod.etl_newbuilding_crossload.run(triggered_by="beat") + + assert result == {"run_id": 11, "disabled": True} + assert len(finish_calls) == 1 + assert finish_calls[0]["status"] == "done" + assert finish_calls[0]["error"] == "TRADEIN_DATABASE_URL not set" + + +def test_exception_writes_failed_breadcrumb_and_reraises() -> None: + """run_crossload() бросает исключение (postgres timeout / schema drift) → + breadcrumb помечается 'failed' с error-сообщением, задача ре-рейзит + исключение (Celery FAILURE + GlitchTip продолжает ловить сбой).""" + from app.workers.tasks import etl_newbuilding_crossload as mod + + db = _make_db_mock(run_id=13) + finish_calls, capture = _capture_finish_run(mod) + + with ( + patch.object(mod, "SessionLocal", return_value=db), + patch.object(mod, "run_crossload", side_effect=RuntimeError("tradein postgres timeout")), + patch.object(mod, "_finish_run", side_effect=capture), + ): + try: + mod.etl_newbuilding_crossload.run(triggered_by="beat") + raised = False + except RuntimeError as e: + raised = True + assert "tradein postgres timeout" in str(e) + + assert raised, "исключение должно ре-рейзиться, чтобы Celery ушёл в FAILURE" + + assert len(finish_calls) == 1 + assert finish_calls[0]["run_id"] == 13 + assert finish_calls[0]["status"] == "failed" + assert "RuntimeError" in finish_calls[0]["error"] + assert "tradein postgres timeout" in finish_calls[0]["error"] + + db.close.assert_called_once() + + +def test_breadcrumb_flush_failure_does_not_swallow_original_exception() -> None: + """Если сам breadcrumb-flush (_finish_run) на failure-пути тоже падает — + исходное исключение из run_crossload всё равно должно ре-рейзиться, не + подавляться.""" + from app.workers.tasks import etl_newbuilding_crossload as mod + + db = _make_db_mock(run_id=21) + + def _boom_finish(*_a: Any, **_kw: Any) -> None: + raise Exception("DB connection lost while writing breadcrumb") + + with ( + patch.object(mod, "SessionLocal", return_value=db), + patch.object( + mod, "run_crossload", side_effect=RuntimeError("schema drift on tradein side") + ), + patch.object(mod, "_finish_run", side_effect=_boom_finish), + ): + try: + mod.etl_newbuilding_crossload.run(triggered_by="beat") + raised_type = None + except Exception as e: # тест проверяет именно тип/сообщение исходного исключения + raised_type = type(e) + raised_msg = str(e) + + assert raised_type is RuntimeError + assert "schema drift on tradein side" in raised_msg