gendesign/tradein-mvp/backend/app/services/fns_lookup.py

"""ФНС opendata lookup по ИНН — тонкое чтение `fns_legal_entity_facts`.

CONTEXT: см. `fns_opendata_loader.py`. Этот модуль — единственная точка чтения
загруженных фактов. ПОТРЕБИТЕЛЯ У НЕГО ПОКА НЕТ: ничто в продукте (скоринг
застройщика/УК, estimator и т.п.) сюда не ходит — подключение решается отдельной
задачей вне этого PR.

psycopg v3: SQL через `text(...)` использует `CAST(:x AS type)`, НИКОГДА `:x::type`.
"""

from __future__ import annotations

from collections import defaultdict
from datetime import date
from typing import TypedDict

from sqlalchemy import text
from sqlalchemy.orm import Session

_LOOKUP_SQL = text(
    """
    SELECT dataset, series, period, value, org_name
    FROM fns_legal_entity_facts
    WHERE inn = CAST(:inn AS text)
    ORDER BY dataset, series, period
    """
)


class FnsFact(TypedDict):
    series: str
    period: date
    value: float
    org_name: str | None


def get_facts_by_inn(db: Session, inn: str) -> dict[str, list[FnsFact]]:
    """Факты по ИНН, сгруппированные по dataset (`revexp`/`sshr2019`/`debtam`/`snr`).

    Пустой/пробельный ИНН и отсутствие данных → `{}` (не исключение — вызывающий код
    не обязан оборачивать lookup в try/except ради нормального «нет данных»).
    """
    normalized = (inn or "").strip()
    if not normalized:
        return {}

    rows = db.execute(_LOOKUP_SQL, {"inn": normalized}).mappings().all()
    out: dict[str, list[FnsFact]] = defaultdict(list)
    for row in rows:
        out[row["dataset"]].append(
            {
                "series": row["series"],
                "period": row["period"],
                "value": row["value"],
                "org_name": row["org_name"],
            }
        )
    return dict(out)