313 lines
10 KiB
Python
313 lines
10 KiB
Python
"""Сборка и установка темы 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"],
|
||
}
|