from __future__ import annotations
import platform
import re
from dataclasses import dataclass
from src.core.config import load_settings
from src.core.constants import APP_NAME, APP_VERSION
from src.integrations.exchange.service import ExchangeService
from src.storage.session import check_database_health
@dataclass(slots=True)
class ComponentStatus:
name: str
state: str
details: str = ""
@dataclass(slots=True)
class SystemSnapshot:
app_name: str
app_version: str
db_label: str
timezone_name: str
mode_label: str
default_symbol: str
components: list[ComponentStatus]
def _extract_postgres_version(raw: str) -> str:
if not raw:
return "PostgreSQL"
match = re.search(r"PostgreSQL\s+(\d+(?:\.\d+)?)", raw)
if match:
return f"PostgreSQL {match.group(1)}"
return "PostgreSQL"
def _build_exchange_status(exchange_service: ExchangeService, default_symbol: str) -> ComponentStatus:
try:
symbol_validation = exchange_service.validate_symbol(default_symbol)
except Exception as exc:
return ComponentStatus(
name="Биржа",
state="🔴",
details=f"Не удалось проверить инструмент: {exc}",
)
exchange_health = exchange_service.get_health()
if exchange_health.ok and symbol_validation.is_valid:
return ComponentStatus(name="Биржа", state="🟢")
if not exchange_health.ok:
return ComponentStatus(
name="Биржа",
state="🔴",
details=exchange_health.message or "Ошибка подключения к API биржи.",
)
return ComponentStatus(
name="Биржа",
state="🔴",
details=symbol_validation.message or "Инструмент не прошел проверку.",
)
def _build_account_status(exchange_service: ExchangeService) -> ComponentStatus:
private_auth_health = exchange_service.get_private_auth_health()
if private_auth_health.ok:
return ComponentStatus(name="Аккаунт", state="🟢")
return ComponentStatus(
name="Аккаунт",
state="🔴",
details=private_auth_health.message or "Ошибка private API.",
)
def _build_database_status() -> tuple[ComponentStatus, str]:
db_ok, db_message = check_database_health()
db_label = _extract_postgres_version(db_message)
if db_ok:
return ComponentStatus(name="База данных", state="🟢"), db_label
return (
ComponentStatus(
name="База данных",
state="🔴",
details=db_message or "Ошибка подключения к БД.",
),
db_label,
)
def _resolve_mode_label(exchange_testnet: bool) -> str:
return "ДЕМО аккаунт" if exchange_testnet else "РЕАЛЬНЫЙ аккаунт"
def get_system_snapshot() -> SystemSnapshot:
settings = load_settings()
exchange_service = ExchangeService()
database_status, db_label = _build_database_status()
exchange_status = _build_exchange_status(exchange_service, settings.default_symbol)
account_status = _build_account_status(exchange_service)
components = [
ComponentStatus(name="Приложение", state="🟢"),
database_status,
ComponentStatus(name="Telegram", state="🟢"),
exchange_status,
account_status,
]
return SystemSnapshot(
app_name=APP_NAME,
app_version=APP_VERSION,
db_label=db_label,
timezone_name=settings.tz,
mode_label=_resolve_mode_label(settings.exchange_testnet),
default_symbol=settings.default_symbol,
components=components,
)
def _render_component(component: ComponentStatus) -> str:
if component.state == "🟢":
return f"{component.state} {component.name}"
return f"{component.state} {component.name}\n— {component.details}"
def build_system_text() -> str:
snapshot = get_system_snapshot()
components_block = "\n".join(_render_component(component) for component in snapshot.components)
return (
"⚙️ Система\n\n"
f"{components_block}\n\n"
"🌐 Окружение\n"
f"• приложение: {snapshot.app_name} {snapshot.app_version}\n"
f"• база данных: {snapshot.db_label}\n"
f"• часовой пояс: {snapshot.timezone_name}\n"
f"• режим: {snapshot.mode_label}\n"
f"• инструмент: {snapshot.default_symbol}"
)