fix(workers): count lots_pf failure correctly, no silent 'done' (#1220)

scrape_objective.sync_objective_group инкрементировал n_requests и reports_ok
ДО HTTP-стрима client.stream_report(...). Любая ошибка стрима (ObjectiveAuthError,
ObjectiveAPIError — оба ⊂ RuntimeError ⊂ Exception, ijson-сбой, обрыв сети)
проглатывалась generic except Exception в lots_pf-ветке без инкремента
reports_failed и без re-raise. Внешний except (ObjectiveAuthError|APIError)
был НЕДОСТИЖИМ для lots_pf. _finish_run(status='done' if reports_failed==0)
помечал run УСПЕШНЫМ при полностью провалившемся главном 600МБ отчёте «Лоты».
objective_lots/history оставались stale молча, админка показывала last_run_at,
downstream supply-layers L2 видел 'done', rows_lots=0.

Patch (зеркалит corp_sum — там паттерн уже правильный):
- n_requests/reports_ok перенесены ИЗ pre-HTTP в after-success (после
  успешного commit'а внутри `with stream_report(...)`).
- В except Exception lots_pf-ветке: n_requests += 1 + reports_failed += 1
  + logger.exception (с traceback). Graceful (no re-raise) сохранён —
  sibling jobs продолжают, но статус run теперь честный.
- corp_sum ветка не тронута. _finish_run сигнатура не меняется.

5 новых регрессионных тестов: success keeps reports_ok=1, auth-error /
api-error / generic Exception в lots_pf инкрементируют reports_failed (не
ok), _finish_run status='failed' при reports_failed>0. ruff clean.

Closes #1220

backend/app/workers/tasks/scrape_objective.py

@@ -263,8 +263,14 @@ def sync_objective_group(
                             use_dkp=params.get("use_dkp", False),
                             payload=None,
                         )
-                        n_requests += 1
-                        reports_ok += 1
+                        # NB: n_requests / reports_ok инкрементируем ТОЛЬКО ПОСЛЕ
+                        # успешного завершения stream+parse+commit. Иначе при провале
+                        # внутреннего generic-except (issue #1220) счётчики бы росли
+                        # даже на упавшем главном отчёте, и _finish_run помечал бы
+                        # run status='done' → silent failure для downstream.
+                        # ObjectiveAuthError/ObjectiveAPIError ⊂ RuntimeError ⊂ Exception
+                        # → ловятся именно этим inner-except (внешний (Auth|API)Error-блок
+                        # на строке ~355 для lots_pf недостижим).
                         try:
                             with client.stream_report(
                                 report_type="Поквартирные",
@@ -290,8 +296,12 @@ def sync_objective_group(
                                 {"n": n_lots, "rid": raw_id},
                             )
                             db.commit()
+                            n_requests += 1
+                            reports_ok += 1
                         except Exception as parse_err:
                             db.rollback()
+                            n_requests += 1
+                            reports_failed += 1
                             logger.exception(
                                 "sync_objective_group: lots_pf stream failed raw_id=%s: %s",
                                 raw_id,

backend/tests/workers/tasks/test_scrape_objective.py

@@ -0,0 +1,202 @@
+"""Тесты для backend/app/workers/tasks/scrape_objective.py — fix P2 silent-failure
+из issue #1220.
+
+Покрытие:
+- lots_pf success-path: `reports_ok=1, reports_failed=0` (старое поведение сохранено);
+- lots_pf бросает `ObjectiveAuthError` (⊂ RuntimeError ⊂ Exception): теперь
+  `reports_failed=1, reports_ok=0` — БЫЛО `reports_ok=1, reports_failed=0` (silent
+  done на провале главного отчёта);
+- lots_pf бросает `ObjectiveAPIError`: симметрично — `reports_failed=1, reports_ok=0`;
+- lots_pf бросает generic `RuntimeError`: тот же graceful-fail + counter;
+- `_finish_run` при `reports_failed>0` помечает run `status='failed'`, не `'done'`.
+
+Перед патчем счётчик `reports_ok += 1` стоял ДО `with client.stream_report(...)` блока
+и не откатывался во внутреннем `except Exception` — отчёт «Лоты» (~600МБ) падал,
+а run помечался `status='done'`. Сосед `corp_sum` уже считал корректно (после save_raw).
+"""
+
+from __future__ import annotations
+
+from types import ModuleType
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _make_fake_parser(parse_lots_pf_stream: Any, parse_corp_sum: Any) -> Any:
+    """Возвращает callable spec_from_file_location-подобный объект, который
+    подменяет загрузку 70_parse_objective_raw.py и впрыскивает fake-parser.
+    """
+
+    def fake_spec_from_file_location(_name: str, _path: Any) -> Any:
+        spec = MagicMock()
+
+        class _Loader:
+            def exec_module(self, module: ModuleType) -> None:
+                module.parse_lots_pf_stream = parse_lots_pf_stream
+                module.parse_corp_sum = parse_corp_sum
+
+        spec.loader = _Loader()
+        return spec
+
+    def fake_module_from_spec(_spec: Any) -> ModuleType:
+        return ModuleType("objective_parser_70_fake")
+
+    return fake_spec_from_file_location, fake_module_from_spec
+
+
+def _make_db_mock() -> MagicMock:
+    """Mock SessionLocal() — все execute().scalar_one() возвращают целые ID."""
+    db = MagicMock()
+    # _start_run → INSERT ... RETURNING run_id
+    # _save_raw → INSERT ... RETURNING raw_id
+    db.execute.return_value.scalar_one.return_value = 42
+    return db
+
+
+class _FakeStreamCtx:
+    """Контекст-менеджер, имитирующий client.stream_report(...) → resp.iter_bytes."""
+
+    def __init__(self, exc: Exception | None = None) -> None:
+        self._exc = exc
+        self.iter_bytes = MagicMock(return_value=iter([b"{}"]))
+
+    def __enter__(self) -> Any:
+        if self._exc is not None:
+            raise self._exc
+        return self
+
+    def __exit__(self, *a: Any) -> None:
+        return None
+
+
+def _patched_run(
+    *,
+    stream_exc: Exception | None,
+    parse_lots_fn: Any = None,
+) -> tuple[dict[str, Any], list[Any]]:
+    """Запускает sync_objective_group.run() с замоканными:
+    SessionLocal, ObjectiveClient, parser_mod, settings.
+
+    Возвращает (return-value task, [список _finish_run kwargs]) для assert'ов.
+    """
+    from app.workers.tasks import scrape_objective as mod
+
+    db = _make_db_mock()
+    client = MagicMock()
+    client.stream_report = MagicMock(return_value=_FakeStreamCtx(exc=stream_exc))
+    # corp_sum ветка — успешно
+    client.report_corpuses_summary = MagicMock(return_value={"some": "payload"})
+
+    parse_lots = parse_lots_fn or (lambda *a, **kw: (10, 20))  # n_lots, n_hist
+    parse_corp = MagicMock(return_value=5)
+
+    fake_spec, fake_mod_from_spec = _make_fake_parser(parse_lots, parse_corp)
+
+    finish_calls: list[dict[str, Any]] = []
+    orig_finish = mod._finish_run
+
+    def capture_finish(_db: Any, _run_id: int, **kwargs: Any) -> None:
+        finish_calls.append(kwargs)
+        orig_finish(_db, _run_id, **kwargs)
+
+    with (
+        patch.object(mod, "SessionLocal", return_value=db),
+        patch.object(mod, "ObjectiveClient", return_value=client),
+        patch("importlib.util.spec_from_file_location", side_effect=fake_spec),
+        patch("importlib.util.module_from_spec", side_effect=fake_mod_from_spec),
+        patch.object(mod.settings, "objective_api_key", "test-key"),
+        patch.object(mod.settings, "objective_default_group", "Свердловская область"),
+        patch.object(mod, "_finish_run", side_effect=capture_finish),
+    ):
+        result = mod.sync_objective_group.run(
+            group_name="Свердловская область",
+            triggered_by="manual",
+        )
+    return result, finish_calls
+
+
+# ── tests ────────────────────────────────────────────────────────────────────
+
+
+def test_lots_pf_success_keeps_reports_ok_one_failed_zero() -> None:
+    """Success-path: stream+parse прошли → reports_ok=1, reports_failed=0
+    (старое поведение, регресс-страховка).
+    """
+    result, finish_calls = _patched_run(stream_exc=None)
+
+    # corp_sum + lots_pf оба успешны → reports_ok=2, reports_failed=0
+    assert result["reports_ok"] == 2
+    assert result["reports_failed"] == 0
+    assert result["rows_lots"] == 10
+    assert result["rows_history"] == 20
+    assert result["rows_corpus_room"] == 5
+
+    assert len(finish_calls) == 1
+    assert finish_calls[0]["status"] == "done"
+    assert finish_calls[0]["error"] is None
+    assert finish_calls[0]["reports_ok"] == 2
+    assert finish_calls[0]["reports_failed"] == 0
+
+
+def test_lots_pf_auth_error_increments_failed_not_ok() -> None:
+    """ObjectiveAuthError ⊂ RuntimeError ⊂ Exception — ловится во внутреннем
+    generic-except (внешний (Auth|API)Error-блок недостижим). Должен НЕ
+    инкрементить reports_ok и инкрементить reports_failed.
+
+    Регресс fix #1220: до патча reports_ok=1, reports_failed=0 → status='done'.
+    """
+    from app.services.scrapers.objective import ObjectiveAuthError
+
+    result, finish_calls = _patched_run(stream_exc=ObjectiveAuthError("HTTP 401"))
+
+    # corp_sum успешен (reports_ok=1), lots_pf упал (reports_failed=1)
+    assert result["reports_ok"] == 1
+    assert result["reports_failed"] == 1
+    assert result["rows_lots"] == 0
+    assert result["rows_history"] == 0
+
+    assert finish_calls[0]["status"] == "failed"
+    assert finish_calls[0]["error"] is not None
+    assert finish_calls[0]["reports_ok"] == 1
+    assert finish_calls[0]["reports_failed"] == 1
+
+
+def test_lots_pf_api_error_increments_failed_not_ok() -> None:
+    """ObjectiveAPIError (HTTP 500/429/etc) — то же поведение, что Auth."""
+    from app.services.scrapers.objective import ObjectiveAPIError
+
+    result, finish_calls = _patched_run(stream_exc=ObjectiveAPIError("HTTP 500"))
+
+    assert result["reports_ok"] == 1  # только corp_sum
+    assert result["reports_failed"] == 1
+    assert finish_calls[0]["status"] == "failed"
+
+
+def test_lots_pf_generic_exception_increments_failed_not_ok() -> None:
+    """Generic RuntimeError из parser_mod.parse_lots_pf_stream — также
+    graceful-fail, считается как reports_failed."""
+
+    def _boom(*_a: Any, **_kw: Any) -> tuple[int, int]:
+        raise RuntimeError("ijson parse exploded mid-stream")
+
+    result, finish_calls = _patched_run(stream_exc=None, parse_lots_fn=_boom)
+
+    assert result["reports_ok"] == 1  # corp_sum ОК
+    assert result["reports_failed"] == 1
+    assert result["rows_lots"] == 0
+    assert finish_calls[0]["status"] == "failed"
+    assert "1 reports failed" in (finish_calls[0]["error"] or "")
+
+
+def test_finish_run_status_failed_when_any_report_failed() -> None:
+    """Контракт _finish_run: при reports_failed>0 status='failed', не 'done'.
+    error содержит '<n> reports failed'.
+    """
+    from app.services.scrapers.objective import ObjectiveAuthError
+
+    _, finish_calls = _patched_run(stream_exc=ObjectiveAuthError("nope"))
+
+    assert finish_calls[0]["status"] == "failed"
+    assert finish_calls[0]["error"] == "1 reports failed"