149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
"""Отложенный перезапуск службы WESP и перезагрузка Pi (админка / OTA)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import shlex
|
||
import subprocess
|
||
import time
|
||
from typing import Any, Dict, Optional, Tuple
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _env_bool(value: Any, default: bool = True) -> bool:
|
||
if value is None:
|
||
return default
|
||
if isinstance(value, bool):
|
||
return value
|
||
return str(value).strip().lower() not in ("0", "false", "no", "off", "")
|
||
|
||
|
||
def _delay_sec(app_config: Dict[str, Any]) -> float:
|
||
try:
|
||
return max(1.0, float(app_config.get("WESP_AUTO_REBOOT_DELAY_SEC", 3)))
|
||
except (TypeError, ValueError):
|
||
return 3.0
|
||
|
||
|
||
def default_systemd_restart_cmd(app_config: Dict[str, Any]) -> str:
|
||
raw = app_config.get("WESP_SYSTEMD_UNIT")
|
||
if raw is None:
|
||
unit = "myscript"
|
||
else:
|
||
unit = str(raw).strip()
|
||
if not unit:
|
||
return ""
|
||
return f"systemctl restart {unit}"
|
||
|
||
|
||
def resolve_admin_restart_cmd(app_config: Dict[str, Any]) -> str:
|
||
explicit = (app_config.get("WESP_ADMIN_RESTART_CMD") or "").strip()
|
||
if explicit:
|
||
return explicit
|
||
return default_systemd_restart_cmd(app_config)
|
||
|
||
|
||
def resolve_update_restart_cmd(app_config: Dict[str, Any]) -> str:
|
||
explicit = (app_config.get("UPDATE_RESTART_CMD") or "").strip()
|
||
if explicit:
|
||
return explicit
|
||
base = (app_config.get("BASE_DIR") or "").strip()
|
||
if base:
|
||
script = f"{base}/scripts/post_update.sh"
|
||
return f'bash "{script}"'
|
||
return ""
|
||
|
||
|
||
def _spawn_shell(cmd: str, *, label: str) -> Tuple[bool, str]:
|
||
cmd = (cmd or "").strip()
|
||
if not cmd:
|
||
return False, f"Команда не задана ({label})"
|
||
try:
|
||
wrapped = f"nohup bash -c {shlex.quote(cmd)} >/dev/null 2>&1 &"
|
||
subprocess.Popen(
|
||
["bash", "-c", wrapped],
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
start_new_session=True,
|
||
)
|
||
logger.info("Запланировано в фоне (%s): %s", label, cmd)
|
||
return True, cmd
|
||
except OSError as exc:
|
||
logger.warning("Не удалось запустить %s: %s", label, exc)
|
||
return False, str(exc)
|
||
|
||
|
||
def schedule_delayed_command(cmd: str, *, delay_sec: float, label: str) -> Dict[str, Any]:
|
||
delay = max(1.0, float(delay_sec))
|
||
inner = f"sleep {int(delay)}; {cmd}"
|
||
ok, detail = _spawn_shell(inner, label=label)
|
||
return {
|
||
"scheduled": ok,
|
||
"delay_sec": delay,
|
||
"command": cmd,
|
||
"detail": detail,
|
||
}
|
||
|
||
|
||
def schedule_service_restart(
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
delay_sec: Optional[float] = None,
|
||
) -> Dict[str, Any]:
|
||
cmd = resolve_admin_restart_cmd(app_config)
|
||
delay = delay_sec if delay_sec is not None else _delay_sec(app_config)
|
||
result = schedule_delayed_command(cmd, delay_sec=delay, label="service_restart")
|
||
if result.get("scheduled"):
|
||
result["message"] = (
|
||
f"Перезапуск WESP через {int(delay)} с ({cmd})."
|
||
)
|
||
else:
|
||
result["message"] = (
|
||
"Задайте WESP_ADMIN_RESTART_CMD или WESP_SYSTEMD_UNIT (например myscript)."
|
||
)
|
||
return result
|
||
|
||
|
||
def schedule_system_reboot(
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
delay_sec: Optional[float] = None,
|
||
reason: str = "",
|
||
) -> Dict[str, Any]:
|
||
if not _env_bool(app_config.get("WESP_AUTO_REBOOT_ENABLED"), True):
|
||
return {
|
||
"scheduled": False,
|
||
"message": "Автоперезагрузка отключена (WESP_AUTO_REBOOT_ENABLED=0).",
|
||
}
|
||
cmd = (app_config.get("WESP_ADMIN_REBOOT_CMD") or "sudo /sbin/reboot").strip()
|
||
delay = delay_sec if delay_sec is not None else _delay_sec(app_config)
|
||
result = schedule_delayed_command(cmd, delay_sec=delay, label="reboot")
|
||
if result.get("scheduled"):
|
||
suffix = f" ({reason})" if reason else ""
|
||
result["message"] = (
|
||
f"Перезагрузка устройства через {int(delay)} с{suffix}."
|
||
)
|
||
else:
|
||
result["message"] = "Не удалось запланировать reboot (WESP_ADMIN_REBOOT_CMD)."
|
||
return result
|
||
|
||
|
||
def maybe_auto_reboot_after_kiosk_change(app_config: Dict[str, Any]) -> Dict[str, Any]:
|
||
if not _env_bool(app_config.get("WESP_KIOSK_AUTO_REBOOT"), True):
|
||
return {"scheduled": False}
|
||
return schedule_system_reboot(app_config, reason="киоск")
|
||
|
||
|
||
def maybe_auto_reboot_after_pi_boot_change(app_config: Dict[str, Any]) -> Dict[str, Any]:
|
||
if not _env_bool(app_config.get("WESP_PI_BOOT_AUTO_REBOOT"), True):
|
||
return {"scheduled": False}
|
||
return schedule_system_reboot(app_config, reason="config.txt")
|
||
|
||
|
||
def maybe_auto_reboot_after_plymouth_install(app_config: Dict[str, Any]) -> Dict[str, Any]:
|
||
if not _env_bool(app_config.get("WESP_PLYMOUTH_AUTO_REBOOT"), True):
|
||
return {"scheduled": False}
|
||
return schedule_system_reboot(app_config, reason="Plymouth")
|