/
Befun
/
FamalyControlle
Обзор
Документация
Войти
/
Befun
/
FamalyControlle
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
install.py
805 строк
32 KB
BeFun
Init project
07 май 2026, 10:43
07 май 2026, 10:43
7e96900
Код
Авторство
О чём код?
#!/usr/bin/env python3 """ Family PC Control — интерактивный установщик. Спрашивает роль устройства: 1. Сервер + Бот — устанавливает API-сервер и бота Макс на одном устройстве 2. Агент — устанавливает десктоп-агент на контролируемом ПК Для Агента спрашивает IP-адрес сервера и настраивает подключение. Для Сервера спрашивает токен бота и ID администратора. Использование: python install.py """ import os import platform import shutil import subprocess import sys import venv from pathlib import Path # ═══════════════════════════════════════════════════════════════ # Цвета и утилиты терминала # ═══════════════════════════════════════════════════════════════ class C: """ANSI-цвета.""" RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" BLUE = "\033[94m" CYAN = "\033[96m" @staticmethod def disable(): """Отключить цвета (для Windows без ANSI).""" for attr in ("RESET", "BOLD", "DIM", "RED", "GREEN", "YELLOW", "BLUE", "CYAN"): setattr(C, attr, "") # Windows 10+ поддерживает ANSI, но на всякий случай if platform.system() == "Windows": try: import ctypes kernel32 = ctypes.windll.kernel32 kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7) except Exception: C.disable() def banner(): print(f""" {C.CYAN}{C.BOLD}╔══════════════════════════════════════════════════╗ ║ Family PC Control — Установщик ║ ║ Система семейного контроля ПК ║ ╚══════════════════════════════════════════════════╝{C.RESET} """) def step(n: int, total: int, text: str): print(f"\n{C.BOLD}{C.BLUE}[{n}/{total}]{C.RESET} {text}") def info(text: str): print(f" {C.DIM}{text}{C.RESET}") def ok(text: str): print(f" {C.GREEN} {text}{C.RESET}") def warn(text: str): print(f" {C.YELLOW}! {text}{C.RESET}") def err(text: str): print(f" {C.RED} {text}{C.RESET}") def ask(prompt: str, default: str = "") -> str: suffix = f" [{C.GREEN}{default}{C.RESET}]" if default else "" line = input(f" {C.CYAN}>{C.RESET} {prompt}{suffix}: ").strip() return line or default def ask_choice(prompt: str, choices: list[str], default_idx: int = 0) -> int: """Показать нумерованный список и вернуть индекс выбора.""" print(f" {prompt}") for i, ch in enumerate(choices, 1): marker = f" {C.GREEN}(по умолчанию){C.RESET}" if i - 1 == default_idx else "" print(f" {C.BOLD}{i}{C.RESET}. {ch}{marker}") while True: try: val = input(f" {C.CYAN}>{C.RESET} Выберите [1-{len(choices)}]: ").strip() if not val: return default_idx idx = int(val) - 1 if 0 <= idx < len(choices): return idx except ValueError: pass warn(f"Введите число от 1 до {len(choices)}") def ask_yes_no(prompt: str, default: bool = True) -> bool: suffix = "[Д/y]" if default else "[д/N]" line = input(f" {C.CYAN}>{C.RESET} {prompt} {suffix}: ").strip().lower() if not line: return default return line in ("y", "yes", "д", "да", "1") def run_cmd(cmd: str, cwd: str = None, check: bool = False) -> subprocess.CompletedProcess: """Запустить shell-команду, показывая вывод.""" info(f"$ {cmd}") return subprocess.run( cmd, shell=True, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) def run_cmd_show(cmd: str, cwd: str = None) -> int: """Запустить команду с выводом на экран.""" return subprocess.call(cmd, shell=True, cwd=cwd) # ═══════════════════════════════════════════════════════════════ # Роли устройства # ═══════════════════════════════════════════════════════════════ ROLE_SERVER = "server" ROLE_AGENT = "agent" ROLE_NAMES = { ROLE_SERVER: "Сервер + Бот (управляющее устройство)", ROLE_AGENT: "Агент (контролируемый ПК)", } # ═══════════════════════════════════════════════════════════════ # Основной установщик # ═══════════════════════════════════════════════════════════════ class Installer: """Интерактивный установщик Family PC Control.""" TOTAL_STEPS = 6 def __init__(self): self.role: str = "" self.install_dir: Path = Path.cwd() / "family-pc-control" self.python: str = sys.executable self.use_venv: bool = True self.venv_dir: Path = Path() self.install_as_service: bool = False # Конфигурация self.bot_token: str = "" self.admin_id: str = "" self.server_host: str = "0.0.0.0" self.server_port: int = 8642 self.agent_server_url: str = "http://localhost:8642" # ── Запуск ──────────────────────────────────────────────── def run(self): banner() self._step1_choose_role() self._step2_install_dir() self._step3_copy_files() self._step4_python_deps() self._step5_configure() self._step6_service() self._print_summary() # ── Шаг 1: Выбор роли ───────────────────────────────────── def _step1_choose_role(self): step(1, self.TOTAL_STEPS, "Выбор типа устройства") print(f" На какое устройство вы устанавливаете Family PC Control?") print(f" Система состоит из двух частей, которые работают на разных ПК:") print() idx = ask_choice("Роль устройства:", list(ROLE_NAMES.values())) if idx == 0: self.role = ROLE_SERVER print(f"\n {C.GREEN} Сервер + Бот{C.RESET}") info("API-сервер + Бот мессенджера Макс будут установлены на этом устройстве.") info("Убедитесь, что это устройство всегда включено и доступно в сети.") else: self.role = ROLE_AGENT print(f"\n {C.GREEN} Агент{C.RESET}") info("Десктоп-агент будет установлен на этом устройстве.") info("Он будет подключаться к серверу и выполнять команды контроля.") # ── Шаг 2: Директория установки ─────────────────────────── def _step2_install_dir(self): step(2, self.TOTAL_STEPS, "Директория установки") default_dir = str(self.install_dir) user_dir = ask("Куда установить?", default_dir) self.install_dir = Path(user_dir).resolve() if self.install_dir.exists() and any(self.install_dir.iterdir()): if not ask_yes_no(f"Директория {self.install_dir} не пуста. Продолжить?"): print() err("Установка прервана.") sys.exit(0) self.install_dir.mkdir(parents=True, exist_ok=True) ok(f"Директория: {self.install_dir}") # Виртуальное окружение self.use_venv = ask_yes_no("Создать виртуальное окружение Python?", default=True) if self.use_venv: self.venv_dir = self.install_dir / ".venv" # ── Шаг 3: Копирование файлов ───────────────────────────── def _step3_copy_files(self): step(3, self.TOTAL_STEPS, "Копирование файлов проекта") # Определяем источник — там, где лежит install.py source_dir = Path(__file__).resolve().parent # Файлы, которые нужны для каждой роли common_files = [ "requirements.txt", "install_service.py", ] common_dirs = [ "core", "utils", "models", ] server_files = [ "config.yaml", "main.py", ] server_dirs = [ "server", "bot", ] agent_files = [ "config.yaml", "main.py", ] agent_dirs = [ "agent", ] files_to_copy = common_files.copy() dirs_to_copy = common_dirs.copy() if self.role == ROLE_SERVER: files_to_copy.extend(server_files) dirs_to_copy.extend(server_dirs) else: files_to_copy.extend(agent_files) dirs_to_copy.extend(agent_dirs) copied = 0 # Копируем файлы for fname in files_to_copy: src = source_dir / fname dst = self.install_dir / fname if src.exists(): dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) copied += 1 else: warn(f"Файл не найден: {fname} (пропущен)") # Копируем директории for dname in dirs_to_copy: src = source_dir / dname dst = self.install_dir / dname if src.exists() and src.is_dir(): if dst.exists(): shutil.rmtree(dst) shutil.copytree(src, dst) # Считаем файлы n = sum(1 for _ in dst.rglob("*") if _.is_file()) copied += n else: warn(f"Директория не найдена: {dname}/ (пропущена)") # Создаём data/ и logs/ (self.install_dir / "data").mkdir(parents=True, exist_ok=True) (self.install_dir / "logs").mkdir(parents=True, exist_ok=True) ok(f"Скопировано: {copied} файлов/каталогов") # ── Шаг 4: Зависимости Python ───────────────────────────── def _step4_python_deps(self): step(4, self.TOTAL_STEPS, "Установка зависимостей Python") requirements = self.install_dir / "requirements.txt" if not requirements.exists(): err("requirements.txt не найден, пропускаю.") return pip_exec = "pip" if self.use_venv: info(f"Создаю виртуальное окружение: {self.venv_dir}") venv.create(self.venv_dir, with_pip=True, clear=True) ok("Виртуальное окружение создано.") if platform.system() == "Windows": pip_exec = str(self.venv_dir / "Scripts" / "pip.exe") else: pip_exec = str(self.venv_dir / "bin" / "pip") # Фильтруем зависимости по роли self._filter_requirements(requirements) info("Устанавливаю зависимости (это может занять пару минут)...") result = subprocess.run( [pip_exec, "install", "-r", str(requirements)], cwd=str(self.install_dir), capture_output=True, text=True, ) if result.returncode == 0: ok("Все зависимости установлены.") else: err("Ошибка установки зависимостей!") # Показываем последние строки ошибки err_lines = result.stderr.strip().split("\n")[-5:] for line in err_lines: info(f" {line}") if not ask_yes_no("Продолжить несмотря на ошибку?", default=False): sys.exit(1) def _filter_requirements(self, req_file: Path): """Убрать ненужные зависимости для роли агента/сервера.""" with open(req_file, "r", encoding="utf-8") as f: lines = f.readlines() filtered = [] skip_section = None for line in lines: stripped = line.strip() # Комментарии-разделители if stripped.startswith("#"): if "Agent" in stripped and self.role == ROLE_AGENT: skip_section = "agent" elif "Server" in stripped and self.role == ROLE_SERVER: skip_section = "server" elif "Bot" in stripped and self.role == ROLE_AGENT: skip_section = "bot" elif stripped.startswith("# ==="): # Новый раздел — сбрасываем skip_section = None if skip_section is None or stripped.startswith("# =="): filtered.append(line) continue if stripped and not stripped.startswith("#"): # Если мы в секции для другой роли — пропускаем if skip_section and skip_section != self.role: continue filtered.append(line) with open(req_file, "w", encoding="utf-8") as f: f.writelines(filtered) # ── Шаг 5: Конфигурация ─────────────────────────────────── def _step5_configure(self): step(5, self.TOTAL_STEPS, "Настройка конфигурации") config_path = self.install_dir / "config.yaml" if self.role == ROLE_SERVER: self._configure_server(config_path) else: self._configure_agent(config_path) ok(f"Конфигурация записана: {config_path}") def _configure_server(self, config_path: Path): """Настроить config.yaml для сервера + бота.""" print(f"\n {C.BOLD}Настройка сервера и бота Макс{C.RESET}\n") self.bot_token = ask("Токен бота Макс") if not self.bot_token or self.bot_token == "ТОКЕН_ВАШЕГО_БОТА": warn("Токен не установлен! Бот не будет работать до его ввода в config.yaml") self.admin_id = ask("Ваш ID в мессенджере Макс (администратор)") if not self.admin_id or self.admin_id == "ВАШ_ID_В_МАКС": warn("ID администратора не установлен!") port_str = ask("Порт API-сервера", "8642") try: self.server_port = int(port_str) except ValueError: self.server_port = 8642 host_choice = ask_choice( "На каком интерфейсе слушать?", [ "Все интерфейсы (0.0.0.0) — доступен из сети LAN", "Только localhost (127.0.0.1) — только локально", ], default_idx=0, ) self.server_host = "0.0.0.0" if host_choice == 0 else "127.0.0.1" if self.server_host == "127.0.0.1": warn("Сервер будет доступен только локально!") warn("Агент на другом ПК не сможет подключиться.") if not ask_yes_no("Вы уверены?", default=False): self.server_host = "0.0.0.0" # Генерируем config.yaml config_content = f"""# === Макс бот === max_bot: token: "{self.bot_token}" poll_interval: 3 api_base: "https://api.max.ru/v1" # === Администратор === admin: max_user_id: "{self.admin_id}" # === API сервер === server: host: "{self.server_host}" port: {self.server_port} database_url: "sqlite:///./data/family_control.db" # === Десктоп-агент === # На сервере агент не запускается agent: poll_interval: 2 server_url: "http://localhost:{self.server_port}" download_dirs: - "~/Downloads" - "/home/$USER/Downloads" # === Категории блокировки === block_categories: games: - "steam.exe" - "steamwebhelper.exe" - "EpicGamesLauncher.exe" - "EpicWebHelper.exe" - "MinecraftLauncher.exe" - "javaw.exe" - "FortniteClient-Win64-Shipping.exe" - "Valorant.exe" - "LeagueClient.exe" - "cs2.exe" - "dota2.exe" browsers: - "chrome.exe" - "firefox.exe" - "msedge.exe" - "opera.exe" - "brave.exe" - "vivaldi.exe" - "YandexBrowser.exe" downloads: extensions: - ".exe" - ".msi" - ".bat" - ".cmd" - ".ps1" - ".torrent" - ".zip" - ".rar" - ".7z" - ".tar.gz" # === Роль устройства (установщик) === # НЕ МЕНЯЙТЕ — установщик определён автоматически _device_role: "server" """ with open(config_path, "w", encoding="utf-8") as f: f.write(config_content) def _configure_agent(self, config_path: Path): """Настроить config.yaml для агента.""" print(f"\n {C.BOLD}Настройка подключения к серверу{C.RESET}\n") print(f" Агенту нужен адрес сервера, на котором запущен API-сервер.") print(f" Убедитесь, что сервер уже установлен и работает.\n") self.agent_server_url = ask( "URL API-сервера", "http://192.168.1.100:8642", ) # Проверяем, что URL выглядит правильно if not self.agent_server_url.startswith("http"): warn("URL должен начинаться с http:// или https://") self.agent_server_url = "http://" + self.agent_server_url info(f"Исправлено: {self.agent_server_url}") poll_str = ask("Интервал опроса сервера (секунды)", "2") try: poll_interval = int(poll_str) except ValueError: poll_interval = 2 # Папки загрузок default_downloads = [] if platform.system() == "Windows": default_downloads = [str(Path.home() / "Downloads")] else: default_downloads = [ str(Path.home() / "Downloads"), "/home/$USER/Downloads", ] downloads_str = ask( "Папки для мониторинга загрузок (через запятую)", ", ".join(default_downloads), ) download_dirs = [d.strip() for d in downloads_str.split(",") if d.strip()] # Генерируем config.yaml downloads_yaml = "\n".join(f' - "{d}"' for d in download_dirs) config_content = f"""# === Макс бот === # На агенте бот не запускается max_bot: token: "" poll_interval: 3 api_base: "https://api.max.ru/v1" # === Администратор === admin: max_user_id: "" # === API сервер === # На агенте сервер не запускается server: host: "0.0.0.0" port: 8642 database_url: "sqlite:///./data/family_control.db" # === Десктоп-агент === agent: poll_interval: {poll_interval} server_url: "{self.agent_server_url}" download_dirs: {downloads_yaml} # === Категории блокировки === # Агент получает актуальный список ограничений с сервера, # но для локального мониторинга нужны имена процессов: block_categories: games: - "steam.exe" - "steamwebhelper.exe" - "EpicGamesLauncher.exe" - "EpicWebHelper.exe" - "MinecraftLauncher.exe" - "javaw.exe" - "FortniteClient-Win64-Shipping.exe" - "Valorant.exe" - "LeagueClient.exe" - "cs2.exe" - "dota2.exe" browsers: - "chrome.exe" - "firefox.exe" - "msedge.exe" - "opera.exe" - "brave.exe" - "vivaldi.exe" - "YandexBrowser.exe" downloads: extensions: - ".exe" - ".msi" - ".bat" - ".cmd" - ".ps1" - ".torrent" - ".zip" - ".rar" - ".7z" - ".tar.gz" # === Роль устройства (установщик) === # НЕ МЕНЯЙТЕ — установщик определён автоматически _device_role: "agent" """ with open(config_path, "w", encoding="utf-8") as f: f.write(config_content) # ── Шаг 6: Автозапуск как служба ────────────────────────── def _step6_service(self): step(6, self.TOTAL_STEPS, "Автозапуск") if self.role == ROLE_SERVER: label = "сервер и бот" else: label = "агент" self.install_as_service = ask_yes_no( f"Установить {label} как системную службу (автозапуск)?", default=True, ) if self.install_as_service: self._install_service() def _install_service(self): """Установить как системную службу.""" python = self.python if self.use_venv: if platform.system() == "Windows": python = str(self.venv_dir / "Scripts" / "python.exe") else: python = str(self.venv_dir / "bin" / "python") service_name = "fpc-server" if self.role == ROLE_SERVER else "fpc-agent" script = str(self.install_dir / "main.py") if platform.system() == "Windows": self._install_service_windows(service_name, python, script) else: self._install_service_linux(service_name, python, script) def _install_service_windows(self, service_name: str, python: str, script: str): """Установка через NSSM (Windows).""" info("Проверяю NSSM...") result = run_cmd("where nssm") if result.returncode != 0: warn("NSSM не найден. Служба не установлена.") info("Установите NSSM: choco install nssm") info("Или скачайте с https://nssm.cc/download") info(f"Затем запустите вручную:") info(f' nssm install {service_name} "{python}" "{script}"') return log_dir = self.install_dir / "logs" log_dir.mkdir(exist_ok=True) run_cmd(f'nssm install {service_name} "{python}" "{script}"') run_cmd(f'nssm set {service_name} AppDirectory "{self.install_dir}"') run_cmd(f'nssm set {service_name} DisplayName "Family PC Control ({self.role})"') run_cmd(f'nssm set {service_name} Description "Family PC Control — {ROLE_NAMES[self.role]}"') run_cmd(f'nssm set {service_name} Start SERVICE_AUTO_START') run_cmd(f'nssm set {service_name} AppStdout "{log_dir / "service.log"}"') run_cmd(f'nssm set {service_name} AppStderr "{log_dir / "service_err.log"}"') run_cmd(f'nssm set {service_name} AppEnvironmentExtra PYTHONPATH="{self.install_dir}"') ok(f'Служба "{service_name}" установлена!') info(f"Запуск: net start {service_name}") info(f"Остановка: net stop {service_name}") def _install_service_linux(self, service_name: str, python: str, script: str): """Установка через systemd (Linux).""" service_dir = Path.home() / ".config" / "systemd" / "user" service_dir.mkdir(parents=True, exist_ok=True) service_file = service_dir / f"{service_name}.service" env_file = self.install_dir / ".env" # Определяем аргументы запуска if self.role == ROLE_SERVER: start_args = "--server --bot" description = "Family PC Control — Server + Bot" else: start_args = "--agent" description = "Family PC Control — Desktop Agent" with open(env_file, "w") as f: f.write(f"PYTHONPATH={self.install_dir}\n") content = f"""[Unit] Description={description} After=network-online.target Wants=network-online.target [Service] Type=simple EnvironmentFile={env_file} ExecStart={python} {script} {start_args} WorkingDirectory={self.install_dir} Restart=on-failure RestartSec=10 StandardOutput=append:{self.install_dir / 'logs' / 'service.log'} StandardError=append:{self.install_dir / 'logs' / 'service_err.log'} [Install] WantedBy=default.target """ with open(service_file, "w") as f: f.write(content) run_cmd("systemctl --user daemon-reload") run_cmd(f"systemctl --user enable {service_name}") ok(f'Служба "{service_name}" установлена!') info(f"Запуск: systemctl --user start {service_name}") info(f"Остановка: systemctl --user stop {service_name}") info(f"Логи: journalctl --user -u {service_name} -f") # ── Итоги ───────────────────────────────────────────────── def _print_summary(self): print(f""" {C.CYAN}{C.BOLD}════════════════════════════════════════════════════{C.RESET} {C.CYAN}{C.BOLD} Установка завершена!{C.RESET} {C.CYAN}{C.BOLD}════════════════════════════════════════════════════{C.RESET} """) role_label = ROLE_NAMES[self.role] print(f" Роль: {C.BOLD}{role_label}{C.RESET}") print(f" Директория: {self.install_dir}") print(f" Конфиг: {self.install_dir / 'config.yaml'}") print(f" Логи: {self.install_dir / 'logs' / 'service.log'}") print(f" Служба: {'Да' if self.install_as_service else 'Нет'}") # Команды запуска/остановки print(f"\n{C.BOLD} Команды:{C.RESET}") if self.use_venv: if platform.system() == "Windows": activate = f"{self.venv_dir}\\Scripts\\activate" python = f"{self.venv_dir}\\Scripts\\python.exe" else: activate = f"source {self.venv_dir}/bin/activate" python = f"{self.venv_dir}/bin/python" else: activate = "" python = sys.executable if self.role == ROLE_SERVER: start_cmd = f"{python} {self.install_dir / 'main.py'} --server --bot" print(f" Ручной запуск:") print(f" {C.DIM}{activate}{C.RESET}") print(f" {C.GREEN}{start_cmd}{C.RESET}") if self.server_host == "0.0.0.0": print(f"\n API-доступ: {C.CYAN}http://<IP-этого-ПК>:{self.server_port}{C.RESET}") print(f" Swagger: {C.CYAN}http://<IP-этого-ПК>:{self.server_port}/docs{C.RESET}") else: start_cmd = f"{python} {self.install_dir / 'main.py'} --agent" print(f" Ручной запуск:") print(f" {C.DIM}{activate}{C.RESET}") print(f" {C.GREEN}{start_cmd}{C.RESET}") print(f"\n Подключение к серверу: {C.CYAN}{self.agent_server_url}{C.RESET}") if self.install_as_service: if platform.system() == "Windows": svc = "fpc-server" if self.role == ROLE_SERVER else "fpc-agent" print(f"\n Управление службой:") print(f" Запуск: {C.GREEN}net start {svc}{C.RESET}") print(f" Остановка: {C.GREEN}net stop {svc}{C.RESET}") else: svc = "fpc-server" if self.role == ROLE_SERVER else "fpc-agent" print(f"\n Управление службой:") print(f" Запуск: {C.GREEN}systemctl --user start {svc}{C.RESET}") print(f" Остановка: {C.GREEN}systemctl --user stop {svc}{C.RESET}") print(f" Логи: {C.GREEN}journalctl --user -u {svc} -f{C.RESET}") # Следующие шаги print(f"\n{C.BOLD} Следующие шаги:{C.RESET}") if self.role == ROLE_SERVER: print(f" 1. Отредактируйте {C.YELLOW}config.yaml{C.RESET} если нужно") print(f" 2. Запустите сервер: {C.YELLOW}{start_cmd}{C.RESET}") print(f" 3. Установите агент на контролируемый ПК") print(f" (запустите install.py там и выберите 'Агент')") else: print(f" 1. Убедитесь, что сервер запущен на другом ПК") print(f" 2. Проверьте подключение:") print(f" {C.YELLOW}curl {self.agent_server_url}/api/system/status{C.RESET}") print(f" 3. Запустите агент: {C.YELLOW}{start_cmd}{C.RESET}") print() # ═══════════════════════════════════════════════════════════════ # Точка входа # ═══════════════════════════════════════════════════════════════ if __name__ == "__main__": try: installer = Installer() installer.run() except KeyboardInterrupt: print(f"\n\n{C.YELLOW}Установка прервана пользователем.{C.RESET}") sys.exit(0)