"""Сборка и установка темы Plymouth WESP (кадры 800×600 + script theme).""" from __future__ import annotations import os import shutil import subprocess import threading import time from pathlib import Path from typing import Any, Dict, Optional from flask import Flask from app.services.plymouth_frame_renderer import ( DURATION_MS_DEFAULT, animation_end_ms, frame_count, pillow_available, render_all_frames, ) _THEME_ID = "wesp" _lock = threading.Lock() def _resolve_flask_app(app: Flask) -> Flask: """current_app (LocalProxy) в фоновом потоке не работает — нужен реальный Flask.""" get_obj = getattr(app, "_get_current_object", None) if callable(get_obj): return get_obj() # type: ignore[no-any-return] return app _job_state: Dict[str, Any] = { "running": False, "phase": "idle", "progress": 0, "total": 0, "message": "", "error": None, "result": None, "started_at": None, "finished_at": None, } def _base_dir(app: Flask) -> Path: return Path(str(app.config.get("BASE_DIR", "."))).resolve() def _logo_path(app: Flask) -> Path: return _base_dir(app) / "static" / "logo2.png" def _template_theme_dir(app: Flask) -> Path: return _base_dir(app) / "install" / "plymouth" / _THEME_ID def _build_staging_dir(app: Flask) -> Path: raw = (app.config.get("WESP_PLYMOUTH_BUILD_DIR") or "").strip() if raw: p = Path(raw).expanduser() if not p.is_absolute(): p = _base_dir(app) / p return p.resolve() data = Path(str(app.config.get("DATA_DIR") or _base_dir(app) / "data")).resolve() return data / "plymouth-build" def _install_script_path(app: Flask) -> Path: custom = (app.config.get("WESP_PLYMOUTH_INSTALL_SCRIPT") or "").strip() if custom: return Path(custom).expanduser().resolve() return _base_dir(app) / "scripts" / "plymouth" / "install_wesp_theme.sh" def _install_cmd(app: Flask) -> str: return (app.config.get("WESP_PLYMOUTH_INSTALL_CMD") or "").strip() def _system_theme_dir() -> Path: return Path("/usr/share/plymouth/themes") / _THEME_ID def _plymouth_available() -> bool: return shutil.which("plymouth") is not None def _read_default_theme() -> Optional[str]: alt = Path("/etc/alternatives/default.plymouth") if alt.is_symlink(): try: target = alt.resolve() return target.parent.name except OSError: return None return None def get_plymouth_status(app: Flask) -> Dict[str, Any]: staging = _build_staging_dir(app) anim = staging / "animation" frame_files = sorted(anim.glob("*.png")) if anim.is_dir() else [] installed_anim = _system_theme_dir() / "animation" installed_frames = ( sorted(installed_anim.glob("*.png")) if installed_anim.is_dir() else [] ) with _lock: job = dict(_job_state) return { "plymouth_installed": _plymouth_available(), "pillow_available": pillow_available(), "logo_exists": _logo_path(app).is_file(), "template_theme_exists": _template_theme_dir(app).is_dir(), "build_dir": str(staging), "built_frames": len(frame_files), "expected_frames": frame_count(DURATION_MS_DEFAULT), "installed_theme_dir": str(_system_theme_dir()), "installed_frames": len(installed_frames), "active_theme": _read_default_theme(), "resolution": "800x600", "duration_ms": DURATION_MS_DEFAULT, "animation_end_ms": animation_end_ms(DURATION_MS_DEFAULT), "install_script": str(_install_script_path(app)), "job": job, } def _copy_theme_skeleton(staging: Path, template_dir: Path) -> None: staging.mkdir(parents=True, exist_ok=True) for name in (f"{_THEME_ID}.plymouth", f"{_THEME_ID}.script"): src = template_dir / name if not src.is_file(): raise FileNotFoundError(f"Нет файла темы: {src}") shutil.copy2(src, staging / name) def _run_install_subprocess(app: Flask, staging: Path) -> Dict[str, Any]: custom_cmd = _install_cmd(app) if custom_cmd: import shlex args = shlex.split(custom_cmd) + [str(staging)] proc = subprocess.run( args, capture_output=True, text=True, timeout=int(app.config.get("WESP_PLYMOUTH_INSTALL_TIMEOUT", 600)), ) if proc.returncode != 0: err = (proc.stderr or proc.stdout or "").strip() or f"exit {proc.returncode}" return {"ok": False, "message": err, "code": "install_failed"} return {"ok": True, "message": (proc.stdout or "").strip() or "Тема установлена"} script = _install_script_path(app) if not script.is_file(): return { "ok": False, "code": "no_install_script", "message": f"Скрипт установки не найден: {script}", } proc = subprocess.run( ["sudo", "bash", str(script), str(staging)], capture_output=True, text=True, timeout=int(app.config.get("WESP_PLYMOUTH_INSTALL_TIMEOUT", 600)), ) if proc.returncode != 0: err = (proc.stderr or proc.stdout or "").strip() hint = "" if "password" in err.lower() or proc.returncode == 1: hint = ( " Настройте sudo без пароля для install_wesp_theme.sh " "или задайте WESP_PLYMOUTH_INSTALL_CMD." ) return { "ok": False, "code": "install_failed", "message": (err or f"exit {proc.returncode}") + hint, } return { "ok": True, "message": (proc.stdout or "").strip() or "Тема Plymouth WESP установлена", } def _set_job(**kwargs: Any) -> None: with _lock: _job_state.update(kwargs) def build_and_install_theme(app: Flask) -> None: """Фоновая сборка кадров + установка (вызывается из потока).""" with app.app_context(): _build_and_install_theme_impl(app) def _build_and_install_theme_impl(app: Flask) -> None: staging = _build_staging_dir(app) template = _template_theme_dir(app) logo = _logo_path(app) try: _set_job( running=True, phase="render", progress=0, total=frame_count(DURATION_MS_DEFAULT), message="Рендер кадров…", error=None, result=None, started_at=time.time(), finished_at=None, ) if not pillow_available(): raise RuntimeError( "Нет Pillow. Установите зависимости: pip install -r requirements-prod.txt" ) if not logo.is_file(): raise FileNotFoundError(f"Логотип не найден: {logo}") if not template.is_dir(): raise FileNotFoundError(f"Шаблон темы не найден: {template}") if staging.exists(): shutil.rmtree(staging) staging.mkdir(parents=True) def on_progress(done: int, total: int) -> None: _set_job(progress=done, total=total, message=f"Кадр {done}/{total}") summary = render_all_frames(logo, staging, progress_callback=on_progress) _copy_theme_skeleton(staging, template) _set_job(phase="install", message="Установка темы (sudo)…") if not _plymouth_available(): _set_job( running=False, phase="done", message="Кадры собраны; plymouth не установлен в системе", finished_at=time.time(), result={ "render": summary, "install": { "ok": False, "code": "plymouth_missing", "message": "apt install plymouth plymouth-themes", }, }, ) return install_result = _run_install_subprocess(app, staging) done_message = install_result.get("message", "") job_result: Dict[str, Any] = {"render": summary, "install": install_result} if install_result.get("ok"): try: from app.services.system_restart_service import ( maybe_auto_reboot_after_plymouth_install, ) with real_app.app_context(): reboot = maybe_auto_reboot_after_plymouth_install(real_app.config) job_result["reboot"] = reboot if reboot.get("scheduled") and reboot.get("message"): done_message = f"{done_message} {reboot['message']}".strip() except Exception as exc: job_result["reboot"] = {"scheduled": False, "message": str(exc)} _set_job( running=False, phase="done", message=done_message, error=None if install_result.get("ok") else install_result.get("message"), finished_at=time.time(), result=job_result, ) except Exception as exc: _set_job( running=False, phase="error", error=str(exc), message=str(exc), finished_at=time.time(), ) def start_build_and_install(app: Flask) -> Dict[str, Any]: real_app = _resolve_flask_app(app) with _lock: if _job_state.get("running"): return { "ok": False, "code": "busy", "message": "Установка Plymouth уже выполняется", "job": dict(_job_state), } thread = threading.Thread( target=build_and_install_theme, args=(real_app,), name="wesp-plymouth-install", daemon=True, ) thread.start() return { "ok": True, "code": "started", "message": "Сборка и установка темы Plymouth запущены", "job": get_plymouth_status(real_app)["job"], }