site/WESP_REL/app/services/pi_boot_config_service.py

"""Настройки загрузки Raspberry Pi в config.txt (disable_splash и т.д.)."""

from __future__ import annotations

import re
import shlex
import subprocess
import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

_DISABLE_SPLASH_RE = re.compile(r"^\s*#?\s*disable_splash\s*=", re.IGNORECASE)
_SECTION_RE = re.compile(r"^\s*\[(.+)]\s*$")


def _config_candidates() -> List[Path]:
    return [
        Path("/boot/firmware/config.txt"),
        Path("/boot/config.txt"),
    ]


def find_config_txt(custom: str = "") -> Optional[Path]:
    if custom:
        p = Path(custom).expanduser()
        return p if p.is_file() else None
    for p in _config_candidates():
        if p.is_file():
            return p
    return None


def _read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="replace")


def _parse_disable_splash_enabled(text: str) -> bool:
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            continue
        if stripped == "disable_splash=1":
            return True
    return False


def _patch_disable_splash(text: str, enabled: bool) -> str:
    lines = text.splitlines()
    kept: List[str] = []
    for line in lines:
        if _DISABLE_SPLASH_RE.match(line):
            continue
        kept.append(line)

    if not enabled:
        return "\n".join(kept).rstrip() + "\n"

    insert_at = len(kept)
    for i, line in enumerate(kept):
        if _SECTION_RE.match(line) and _SECTION_RE.match(line).group(1).strip().lower() == "all":
            insert_at = i + 1
            break

    kept.insert(insert_at, "disable_splash=1")
    return "\n".join(kept).rstrip() + "\n"


def _write_via_sudo(path: Path, content: str, custom_cmd: str = "") -> Tuple[bool, str]:
    if custom_cmd:
        args = shlex.split(custom_cmd) + [str(path)]
        proc = subprocess.run(
            args,
            input=content,
            capture_output=True,
            text=True,
            timeout=60,
        )
        if proc.returncode != 0:
            err = (proc.stderr or proc.stdout or "").strip() or f"exit {proc.returncode}"
            return False, err
        return True, "Конфиг обновлён"

    try:
        if path.is_file() and path.stat().st_mode & 0o200:
            path.write_text(content, encoding="utf-8")
            return True, "Конфиг обновлён"
    except OSError:
        pass

    with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False, suffix=".txt") as tmp:
        tmp.write(content)
        tmp_path = tmp.name

    try:
        proc = subprocess.run(
            ["sudo", "cp", tmp_path, str(path)],
            capture_output=True,
            text=True,
            timeout=60,
        )
        if proc.returncode != 0:
            err = (proc.stderr or proc.stdout or "").strip()
            hint = " Нужен sudo без пароля или WESP_PI_BOOT_CONFIG_WRITE_CMD."
            return False, (err or f"exit {proc.returncode}") + hint
        subprocess.run(["sudo", "sync"], capture_output=True, timeout=30)
        return True, "Конфиг обновлён (sudo)"
    finally:
        try:
            Path(tmp_path).unlink(missing_ok=True)
        except OSError:
            pass


def get_rainbow_splash_status(config_path: str = "") -> Dict[str, Any]:
    path = find_config_txt(config_path)
    if not path:
        return {
            "config_found": False,
            "config_path": config_path or str(_config_candidates()[0]),
            "disable_splash": False,
            "message": "config.txt не найден (не Raspberry Pi или другой путь)",
        }

    try:
        text = _read_text(path)
    except OSError as exc:
        return {
            "config_found": True,
            "config_path": str(path),
            "disable_splash": False,
            "readable": False,
            "message": str(exc),
        }

    enabled = _parse_disable_splash_enabled(text)
    return {
        "config_found": True,
        "config_path": str(path),
        "readable": True,
        "disable_splash": enabled,
        "message": (
            "Радужный splash прошивки отключён (disable_splash=1)"
            if enabled
            else "Радужный splash прошивки включён (стандарт Pi)"
        ),
    }


def set_rainbow_splash_disabled(
    enabled: bool,
    *,
    config_path: str = "",
    write_cmd: str = "",
) -> Dict[str, Any]:
    """enabled=True → записать disable_splash=1; False → убрать строку."""
    path = find_config_txt(config_path)
    if not path:
        return {
            "ok": False,
            "code": "config_missing",
            "message": "config.txt не найден",
        }

    try:
        original = _read_text(path)
    except OSError as exc:
        return {"ok": False, "code": "read_error", "message": str(exc)}

    patched = _patch_disable_splash(original, enabled)
    if patched == original and enabled == _parse_disable_splash_enabled(original):
        return {
            "ok": True,
            "code": "unchanged",
            "message": "Уже в нужном состоянии",
            "disable_splash": enabled,
            "config_path": str(path),
        }

    ok, msg = _write_via_sudo(path, patched, write_cmd)
    if not ok:
        return {"ok": False, "code": "write_error", "message": msg, "config_path": str(path)}

    action = "отключён" if enabled else "включён (строка убрана)"
    return {
        "ok": True,
        "code": "updated",
        "message": f"Радужный splash прошивки {action}.",
        "disable_splash": enabled,
        "config_path": str(path),
    }