fix(workers): izyatie per-doc isolation + newbuilding-crossload breadcrumb (#2445-D5/D6) (#2460)

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

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()

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()
 

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

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