#!/usr/bin/env python3 """Модели серверов и управление ими.""" import os import logging import getpass from pathlib import Path from typing import Dict, List, Optional from dataclasses import dataclass, field from dotenv import load_dotenv from telegram import InlineKeyboardButton, InlineKeyboardMarkup logger = logging.getLogger(__name__) @dataclass class Server: """Конфигурация сервера.""" name: str host: str port: int user: str tags: List[str] = field(default_factory=list) password: str = "" @property def display_name(self) -> str: """Отображаемое имя с иконкой.""" icon = "🖥️" if "local" in self.tags: icon = "💻" elif "db" in self.tags: icon = "🗄️" elif "web" in self.tags: icon = "🌐" return f"{icon} {self.name}" @property def description(self) -> str: """Краткое описание сервера.""" return f"{self.user}@{self.host}:{self.port}" class ServerManager: """Управление серверами.""" def __init__(self): self._servers: Dict[str, Server] = {} self._default_server: str = "local" self._ssh_key_path: Optional[str] = None # Локальный сервер всегда доступен try: local_user = getpass.getuser() except Exception: local_user = "user" self._servers["local"] = Server( name="local", host="localhost", port=22, user=local_user, tags=["local", "dev"] ) def load_from_env(self): """Загрузка серверов из переменных окружения.""" self._ssh_key_path = os.getenv("SSH_KEY_PATH") self._default_server = os.getenv("DEFAULT_SERVER", "local") servers_str = os.getenv("SERVERS", "") if not servers_str.strip(): return # Парсинг формата: name|host|port|user|tags|password,name|host|port|user|tags|password for server_line in servers_str.split(","): if not server_line.strip(): continue parts = server_line.strip().split("|") if len(parts) < 4: continue try: name = parts[0].strip() host = parts[1].strip() port = int(parts[2].strip()) user = parts[3].strip() # Теги (часть 4) и пароль (часть 5) могут отсутствовать tags = [] password = "" if len(parts) >= 5 and parts[4].strip(): tags = [t.strip() for t in parts[4].split(",") if t.strip()] if len(parts) >= 6: password = parts[5].strip() server = Server(name=name, host=host, port=port, user=user, tags=tags, password=password) self._servers[name] = server logger.info(f"Загружен сервер: {server.display_name} ({server.description})") except ValueError as e: logger.warning(f"Ошибка парсинга сервера: {parts} - {e}") def get(self, name: str) -> Optional[Server]: """Получить сервер по имени.""" return self._servers.get(name) def list_servers(self) -> List[Server]: """Список всех серверов.""" return list(self._servers.values()) def get_by_tags(self, tags: List[str]) -> List[Server]: """Получить серверы по тегам.""" result = [] for server in self._servers.values(): if any(tag in server.tags for tag in tags): result.append(server) return result @property def default_server(self) -> str: """Имя сервера по умолчанию.""" return self._default_server @property def ssh_key_path(self) -> Optional[str]: """Путь к SSH ключу.""" return self._ssh_key_path def get_keyboard(self, exclude_local: bool = False) -> InlineKeyboardMarkup: """Создать клавиатуру с выбором сервера.""" keyboard = [] for server in self._servers.values(): if exclude_local and server.name == "local": continue button = InlineKeyboardButton( server.display_name, callback_data=f"server_select_{server.name}" ) keyboard.append([button]) return InlineKeyboardMarkup(keyboard) def add_server(self, name: str, host: str, port: int, user: str, tags: List[str] = None, password: str = "") -> bool: """Добавить сервер.""" if name in self._servers: return False self._servers[name] = Server(name=name, host=host, port=port, user=user, tags=tags or [], password=password) self.save_to_env() return True def update_server(self, name: str, host: str = None, port: int = None, user: str = None, tags: List[str] = None, password: str = None) -> bool: """Обновить сервер.""" if name not in self._servers or name == "local": return False server = self._servers[name] if host: server.host = host if port: server.port = port if user: server.user = user if tags is not None: server.tags = tags if password is not None: server.password = password self.save_to_env() return True def delete_server(self, name: str) -> bool: """Удалить сервер.""" if name not in self._servers or name == "local": return False del self._servers[name] self.save_to_env() return True def save_to_env(self): """Сохранить серверы в .env файл.""" env_file = Path(__file__).parent.parent.parent / ".env" # Читаем существующий файл lines = [] if env_file.exists(): with open(env_file, "r", encoding="utf-8") as f: lines = f.readlines() # Формируем строку серверов server_parts = [] for server in self._servers.values(): if server.name == "local": continue tags_str = ",".join(server.tags) if server.tags else "" # Формат: name|host|port|user|tags|password server_parts.append(f"{server.name}|{server.host}|{server.port}|{server.user}|{tags_str}|{server.password}") servers_line = f"SERVERS={','.join(server_parts)}\n" # Ищем и обновляем или добавляем строку SERVERS found = False for i, line in enumerate(lines): if line.startswith("SERVERS="): lines[i] = servers_line found = True break if not found: lines.append("\n" + servers_line) # Записываем обратно with open(env_file, "w", encoding="utf-8") as f: f.writelines(lines) logger.debug(f"Серверы сохранены в {env_file}")