"""Data-access для analysis_runs (#994, 961-C3, ТЗ §22, EPIC 17 «GG-форсайт»). Тонкий repository поверх таблицы `analysis_runs` + view `v_analysis_runs_latest` (миграция 127 / #993). Держит endpoint `analyze_parcel` тонким: persist одного завершённого рана + чтение «даты последнего анализа» (powers by-bbox карту). Контракт миграции 127 (см. data/sql/127_analysis_runs.sql): - result/params/segment — JSONB; пишем через CAST(:x AS jsonb) + json.dumps (psycopg v3: НИКОГДА :x::type — backend.md). result может содержать Pydantic- модели (RiskZone/OpportunityParcel/RedLine) → прогоняем через jsonable_encoder перед json.dumps, иначе TypeError на сериализации. - CHECK'и: status ∈ {complete,partial,error}; confidence ∈ {high,medium,low} ИЛИ NULL. Чтобы НИКОГДА не словить IntegrityError на write — валидируем обе колонки локально (`_normalize_confidence` / `_normalize_status`): неизвестное → NULL/дефолт, не сырое значение. Плохой ран не должен превращать успешный analyze в 500. """ from __future__ import annotations import json import logging from datetime import datetime from typing import Any from fastapi.encoders import jsonable_encoder from sqlalchemy import Row, text from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # Версия схемы рана, когда result — это dict эндпоинта analyze (НЕ SiteFinderReport). # SiteFinderReport.as_dict() несёт собственный _SCHEMA_VERSION ("1.0"); для inline- # результата analyze, у которого своей версии формы нет, фиксируем ОТДЕЛЬНУЮ строку # с префиксом, чтобы re-open мог различить два источника (#992 формализует контракт). ANALYZE_SCHEMA_VERSION = "analyze-1.0" # CHECK-зеркала миграции 127 — единственный источник допустимых значений на write. _ALLOWED_CONFIDENCE: frozenset[str] = frozenset({"high", "medium", "low"}) _ALLOWED_STATUS: frozenset[str] = frozenset({"complete", "partial", "error"}) _DEFAULT_STATUS = "complete" def _normalize_confidence(confidence: str | None) -> str | None: """Привести confidence к CHECK-допустимому high|medium|low или None. Пустая строка / неизвестное значение → None (а НЕ '' — это нарушило бы CHECK). """ if confidence is None: return None value = confidence.strip().lower() return value if value in _ALLOWED_CONFIDENCE else None def _normalize_status(status: str | None) -> str: """Привести status к CHECK-допустимому complete|partial|error (дефолт complete).""" if status is None: return _DEFAULT_STATUS value = status.strip().lower() return value if value in _ALLOWED_STATUS else _DEFAULT_STATUS def _jsonb_param(value: Any) -> str: """Сериализовать произвольный (возможно Pydantic-содержащий) объект в JSON-строку. jsonable_encoder разворачивает Pydantic-модели / даты / Enum в JSON-native типы; json.dumps(..., ensure_ascii=False) — кириллица как есть (зеркало pzz_loader). """ return json.dumps(jsonable_encoder(value), ensure_ascii=False) def persist_analysis_run( db: Session, *, cad_num: str, result: dict[str, Any], params: dict[str, Any], district: str | None, confidence: str | None, status: str, schema_version: str, created_by: str | None, segment: dict[str, Any] | None = None, ) -> int | None: """Сохранить ОДИН завершённый ран анализа в analysis_runs (#994). Best-effort: обёрнут в SAVEPOINT (db.begin_nested, backend.md) + try/except — провал write НЕ ломает ответ analyze и НЕ отравляет outer-транзакцию (откатывает только свой savepoint, прошлые reads сессии остаются валидны). На успехе — возвращает id новой строки, на провале — None (+ logger.warning/exception). confidence/status нормализуются под CHECK'и миграции 127 ДО write — IntegrityError на enum'ах структурно невозможен. advisory всегда TRUE (форсайт-стек советующий). """ norm_confidence = _normalize_confidence(confidence) norm_status = _normalize_status(status) try: with db.begin_nested(): row = ( db.execute( text(""" INSERT INTO analysis_runs ( cad_num, district, segment, params, result, schema_version, advisory, confidence, status, created_by ) VALUES ( CAST(:cad_num AS text), CAST(:district AS text), CAST(:segment AS jsonb), CAST(:params AS jsonb), CAST(:result AS jsonb), CAST(:schema_version AS text), TRUE, CAST(:confidence AS text), CAST(:status AS text), CAST(:created_by AS text) ) RETURNING id """), { "cad_num": cad_num, "district": district, "segment": _jsonb_param(segment) if segment is not None else None, "params": _jsonb_param(params), "result": _jsonb_param(result), "schema_version": schema_version, "confidence": norm_confidence, "status": norm_status, "created_by": created_by, }, ) .mappings() .first() ) new_id = int(row["id"]) if row else None # SAVEPOINT RELEASE (выход из begin_nested) НЕ персистит — лишь сливает вложенную # tx в outer. get_db() (core/db.py) коммита-на-успехе НЕ делает (yield→finally:close # = ROLLBACK), а analyze — единственный write на success-path. Без явного commit # строка откатится на teardown (silent no-op). Коммитим ВНУТРИ try → провал commit # тоже best-effort. Паттерн как в trade_in.py / photos.py (explicit commit на write). db.commit() logger.info( "analysis_run persisted: cad=%s id=%s status=%s confidence=%s", cad_num, new_id, norm_status, norm_confidence, ) return new_id except Exception: # Persist — best-effort: ран уже посчитан, ответ отдаём в любом случае. # exception (не warning) — нужен traceback, write-провалы должны быть видны. logger.exception("analysis_run persist failed for cad=%s (response unaffected)", cad_num) return None def latest_run_dates(db: Session, cad_nums: list[str]) -> dict[str, datetime]: """Дата последнего анализа на участок для набора cad_num (batch, НЕ N+1). Один запрос к v_analysis_runs_latest (canonical latest-per-parcel из 127): `WHERE cad_num = ANY(CAST(:cads AS text[]))`. Powers last_analysis_date в by-bbox. Участки без ранов в результат не попадают (caller трактует отсутствие как None). Пустой вход → пустой dict (без обращения к БД). """ if not cad_nums: return {} rows = ( db.execute( text(""" SELECT cad_num, created_at FROM v_analysis_runs_latest WHERE cad_num = ANY(CAST(:cads AS text[])) """), {"cads": cad_nums}, ) .mappings() .all() ) return {row["cad_num"]: row["created_at"] for row in rows} def latest_run_for(db: Session, cad_num: str) -> Row[Any] | None: """Последний ран на участок целиком (для re-open / «текущий анализ участка»). Читает из v_analysis_runs_latest (DISTINCT ON cad_num, max created_at). None если участок ещё не анализировался. Возвращает Row (id/result/params/.../created_at). """ return db.execute( text(""" SELECT id, cad_num, district, segment, params, result, schema_version, advisory, confidence, status, created_by, created_at FROM v_analysis_runs_latest WHERE cad_num = CAST(:cad_num AS text) """), {"cad_num": cad_num}, ).first()