gendesign/backend/app/services/analysis_runs/repository.py
Light1YT c94810f3a7
All checks were successful
Deploy / changes (push) Successful in 6s
Deploy / build-frontend (push) Has been skipped
Deploy / build-backend (push) Successful in 1m21s
Deploy / build-worker (push) Successful in 2m39s
Deploy / deploy (push) Successful in 1m11s
feat(forecast): forecast_site_finder_report Celery task + schema-filtered run read (3b-i)
build_site_finder_report (§22) takes ~30-180s → runs in a background Celery task,
not inline on the sync /analyze endpoint.

- repository: latest_run_for gains keyword-only schema_version (default None keeps
  v_analysis_runs_latest behavior, backward-compat); when given, reads base
  analysis_runs filtered by schema_version ORDER BY created_at DESC LIMIT 1 — fetches
  the latest analyze-1.0 site-analysis run even when newer 1.0 (§22) rows exist on top
  (index-served via 127's (cad_num, created_at DESC)).
- new workers/tasks/forecast.py::forecast_site_finder_report: reads latest analyze-1.0,
  runs the §22 orchestrator, persists SiteFinderReport.as_dict() as a 1.0 run via
  persist_analysis_run. Graceful: no base run / compute error → logger + return None
  (worker not crashed). time_limit=900/soft=840 (no global limit). Registered in include.

Prod-confirmed: analyze-1.0 result carries the full analyze dict (competitors+district)
→ orchestrator input valid. Endpoint trigger (3b-ii) + §9.x untouched. 943 tests pass;
code-review APPROVE (contracts verified vs real as_dict(); status done→complete normalized,
no IntegrityError). Refs #994 #961.
2026-06-05 09:02:15 +05:00

223 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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,
*,
schema_version: str | None = None,
) -> Row[Any] | None:
"""Последний ран на участок (для re-open / база форсайта по конкретной схеме).
Два режима:
• schema_version=None (дефолт, backward-compatible): читает из
v_analysis_runs_latest (DISTINCT ON cad_num, max created_at) — последний ран
ЛЮБОЙ схемы. Это исходное поведение, существующие вызовы не меняются.
• schema_version задан: читает из БАЗОВОЙ таблицы analysis_runs с фильтром по
schema_version — последний ран ИМЕННО этой схемы. Нужно форсайт-таске (3b-i),
чтобы достать последний site-analysis ран ('analyze-1.0') как базу прогноза,
даже когда поверх него уже легли §22-раны ('1.0') — view бы вернул §22-ран.
None если подходящего рана нет. Возвращает Row (id/result/params/.../created_at).
psycopg v3: CAST(:x AS text) — никогда :x::type (backend.md).
"""
if schema_version is None:
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()
return db.execute(
text("""
SELECT id, cad_num, district, segment, params, result,
schema_version, advisory, confidence, status, created_by, created_at
FROM analysis_runs
WHERE cad_num = CAST(:cad_num AS text)
AND schema_version = CAST(:schema_version AS text)
ORDER BY created_at DESC
LIMIT 1
"""),
{"cad_num": cad_num, "schema_version": schema_version},
).first()