from __future__ import annotations import logging from collections.abc import Coroutine from dataclasses import asdict, is_dataclass from datetime import date from decimal import Decimal from typing import Any, TypeVar import streamlit as st from app.db.base import run_db from app.exports.xlsx import MIME, table_xlsx from app.services import business_calendar from app.services.dates import previous_business_day logger = logging.getLogger(__name__) T = TypeVar("T") def current_user() -> str: """email из профиля Keycloak (AuthService гарантирует, что он есть).""" profile = st.session_state.get("user_profile", {}) if isinstance(profile, dict) and profile.get("email"): return str(profile["email"]) raise RuntimeError("Пользователь не авторизован.") def default_report_date() -> date: """Предыдущий рабочий день по производственному календарю (D30); если БД недоступна — без праздников, чтобы экран всё равно открылся.""" try: return run_db(business_calendar.report_date()) except Exception: # noqa: BLE001 logger.exception("Не удалось загрузить производственный календарь") return previous_business_day() def call_db(coro: Coroutine[Any, Any, T], error_message: str) -> T | None: """run_db() для экранов: непредвиденная ошибка -> st.error, а не трейсбек. Ожидаемые ошибки сервисов (с понятным текстом) ловятся на месте вызова. """ try: return run_db(coro) except Exception as exc: # noqa: BLE001 logger.exception(error_message) st.error(f"{error_message}: {exc}") return None def as_table(rows: list[Any], columns: dict[str, str]) -> list[dict[str, Any]]: """Список dataclass -> строки для st.dataframe с русскими заголовками. Decimal выводим строкой: так видно ровно то, что лежит в БД, без float. """ table = [] for row in rows: data = asdict(row) if is_dataclass(row) else dict(row) table.append( {title: _display(data.get(name)) for name, title in columns.items()} ) return table def download_xlsx( rows: list[Any], columns: dict[str, str], sheet: str, file_name: str ) -> None: """Кнопка «Скачать .xlsx» с той же таблицей, что на экране (D13).""" st.download_button( "Скачать .xlsx", data=table_xlsx(rows, columns, sheet), file_name=file_name, mime=MIME, key=f"xlsx_{file_name}", ) def decimal_text(value: Decimal | None) -> str: """Decimal -> текст для поля ввода ('' для None), без экспоненты и хвостовых нулей.""" return "" if value is None else format(value.normalize(), "f") def _display(value: Any) -> Any: if isinstance(value, Decimal): return format(value.normalize(), "f") return value