gendesign/backend/app/workers/tasks/forecast.py

"""Celery task: посчитать §22 SiteFinderReport в фоне и сохранить в analysis_runs.

Step 3b-i проводки форсайта. §22-оркестратор ``build_site_finder_report`` (#1056,
forecasting/orchestrator.py) считает форсайт из §9.x-слоёв — это занимает ~30-180с
(ЕКБ-wide ~184с, scoped-сегмент меньше), поэтому НЕ может крутиться inline на
синхронном /analyze-эндпоинте. Здесь — фоновая Celery-таска, которая его запускает
и персистит результат рядом с site-analysis раном в той же таблице ``analysis_runs``.

ПОТОК (read → build → persist):
  1. Читаем ПОСЛЕДНИЙ site-analysis ран ('analyze-1.0') для cad_num через
     schema-фильтрованный ``latest_run_for(..., schema_version=ANALYZE_SCHEMA_VERSION)``
     (#3b-i, repository.py) — НЕ через view: поверх analyze-рана уже могут лежать
     §22-раны ('1.0'), view вернул бы их, а нам нужна именно site-analysis база.
     Нет такого рана → logger.warning + return None (нечего форсайтить — graceful,
     без падения воркера).
  2. ``analyze = row.result`` (полный персиснутый dict analyze — несёт district +
     competitors, которые нужны оркестратору). district — из денорм-колонки рана,
     fallback в analyze["district"]["district_name"].
  3. ``build_site_finder_report(db, analyze=…, cad_num=…, district=…, horizons=…)``
     — горизонты = §9.x default grid, с гарантией что запрошенный ``horizon`` в сетке.
  4. ``report.as_dict()`` → ``persist_analysis_run(... schema_version="1.0",
     status="done" ...)``. persist сам best-effort (SAVEPOINT + commit), нормализует
     status/confidence под CHECK'и миграции 127 (status='done' → нормализуется в
     'complete'; форма CHECK не знает 'done'). Возвращаем id новой строки (или None).

NB схема: site-analysis раны несут schema_version='analyze-1.0' (ANALYZE_SCHEMA_VERSION),
§22-форсайт-раны — '1.0' (собственный SiteFinderReport._SCHEMA_VERSION, прокидывается
из ``d["schema_version"]``). Оба сосуществуют в analysis_runs, различаются по
schema_version — поэтому read берёт именно 'analyze-1.0', а write кладёт '1.0'.

Mirror conventions (supply_layers_refresh.py / cbr_macro_sync.py): ``@celery_app.task``,
``SessionLocal()`` + try/finally close, ``logger`` (не print), синхронный ``def`` (НЕ
async внутри таски), ``CAST(:x AS type)`` живёт в repository (тут SQL не пишем).

TIME LIMIT: build_site_finder_report может занять минуты — ставим щедрый per-task лимит
(time_limit=900с / soft_time_limit=840с), чтобы таску не убили mid-compute. Глобального
task_time_limit в celery_app.py нет (только timezone + beat_schedule), так что override
тут — единственный лимит для этой таски.

REGISTRATION: добавлено в ``include=[...]`` celery_app.py (зеркало того, как
зарегистрирован supply_layers_refresh) — beat-расписания НЕТ намеренно (таска
enqueue-on-demand; триггер из /analyze — отдельный шаг 3b-ii, вне scope).
"""

from __future__ import annotations

import logging
from typing import Any

from app.core.db import SessionLocal
from app.services.analysis_runs.repository import (
    ANALYZE_SCHEMA_VERSION,
    latest_run_for,
    persist_analysis_run,
)
from app.services.forecasting.orchestrator import build_site_finder_report
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# Горизонты прогноза по умолчанию (мес) — зеркало §9.x default grid (orchestrator.py
# _DEFAULT_HORIZONS / #952/#984/#986). Запрошенный horizon домешивается, если его нет.
_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)


def _horizons_for(horizon: int) -> list[int]:
    """Default-сетка горизонтов с ГАРАНТИЕЙ что запрошенный ``horizon`` в ней. PURE.

    §9.x single-горизонтные сервисы (score_card/overlay/future_supply) берут primary
    из сетки (orchestrator._primary_horizon → 12 если есть, иначе первый). Чтобы
    запрошенный horizon реально влиял на прогноз, добавляем его в сетку (отсортировано,
    без дублей). horizon ≤ 0 (бессмысленный) → возвращаем чистую default-сетку.
    """
    if horizon <= 0:
        return list(_DEFAULT_HORIZONS)
    return sorted(set(_DEFAULT_HORIZONS) | {horizon})


@celery_app.task(
    bind=True,
    name="tasks.forecast.forecast_site_finder_report",
    time_limit=900,
    soft_time_limit=840,
)
def forecast_site_finder_report(
    self: Any,
    cad_num: str,
    horizon: int = 12,
    created_by: str | None = None,
) -> int | None:
    """Посчитать §22 SiteFinderReport для участка в фоне и сохранить в analysis_runs.

    Берёт последний site-analysis ран ('analyze-1.0') как базу, считает §22-форсайт
    через оркестратор (#1056) и персистит ``SiteFinderReport.as_dict()`` отдельной
    строкой ('1.0') в той же таблице. Best-effort на каждом шаге: нет базы → None
    (warning, не crash); неожиданная ошибка compute → logger.exception + None (таска
    не валит воркер — persist сам best-effort, и форсайт может честно не собраться).

    Args:
        cad_num: кадастровый номер участка.
        horizon: основной горизонт прогноза (мес); домешивается в default-сетку.
        created_by: автор рана (прокидывается в persist; None допустимо).

    Returns:
        id новой строки analysis_runs (§22-форсайт, schema_version='1.0') или None,
        если базы для прогноза нет / персист не удался / compute упал.
    """
    db = SessionLocal()
    try:
        # 1. База прогноза — последний site-analysis ран ('analyze-1.0'), НЕ через view.
        row = latest_run_for(db, cad_num, schema_version=ANALYZE_SCHEMA_VERSION)
        if row is None:
            logger.warning(
                "forecast_site_finder_report: no %s run for cad=%s — nothing to forecast "
                "from (run /analyze first); returning None",
                ANALYZE_SCHEMA_VERSION,
                cad_num,
            )
            return None

        # 2. analyze-dict (несёт district + competitors) + district (денорм → fallback).
        analyze: dict[str, Any] = row.result or {}
        district = row.district or (analyze.get("district") or {}).get("district_name")

        horizons = _horizons_for(horizon)
        logger.info(
            "forecast_site_finder_report: building §22 report cad=%s district=%s "
            "horizon=%s horizons=%s (base run id=%s)",
            cad_num,
            district,
            horizon,
            horizons,
            row.id,
        )

        # 3. §22-форсайт через оркестратор (#1056) — может занять минуты (time_limit=900).
        report = build_site_finder_report(
            db,
            analyze=analyze,
            cad_num=cad_num,
            district=district,
            horizons=horizons,
        )
        d = report.as_dict()

        # 4. Персист отдельной строкой '1.0'. status='done' нормализуется repository'ем
        # под CHECK миграции 127 (→ 'complete'); confidence/segment — из as_dict().
        new_id = persist_analysis_run(
            db,
            cad_num=cad_num,
            result=d,
            params={"horizon": horizon, "source": "forecast-task"},
            district=district,
            confidence=(d.get("confidence") or {}).get("level"),
            status="done",
            schema_version=d["schema_version"],  # "1.0" (SiteFinderReport._SCHEMA_VERSION)
            created_by=created_by,
            segment=(d.get("meta") or {}).get("segment"),
        )
        logger.info(
            "forecast_site_finder_report: persisted §22 report cad=%s run_id=%s schema_version=%s",
            cad_num,
            new_id,
            d.get("schema_version"),
        )
        return new_id
    except Exception:
        # Неожиданный сбой compute/чтения: логируем traceback и НЕ роняем воркер —
        # форсайт может честно не собраться (тонкие данные / сбой §9.x-шва), а persist
        # сам best-effort. Возвращаем None — таска failed gracefully.
        logger.exception("forecast_site_finder_report failed for cad=%s (returning None)", cad_num)
        return None
    finally:
        db.close()