217 lines
7.4 KiB
Python
217 lines
7.4 KiB
Python
#!/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}")
|