805 lines
26 KiB
Python
805 lines
26 KiB
Python
"""Автозапуск Chromium в kiosk (labwc-pi / XDG autostart)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import shlex
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
from urllib.parse import quote, urlparse
|
||
|
||
from app.services.hardware_settings_service import default_app_landing_path
|
||
|
||
try:
|
||
import pwd
|
||
except ImportError: # Windows / non-Unix
|
||
pwd = None # type: ignore[assignment,misc]
|
||
|
||
_DESKTOP_ENTRY_NAME = "chromium-kiosk.desktop"
|
||
_STATE_DIR_NAME = "wesp-kiosk"
|
||
_STATE_FILE = "state.json"
|
||
_LABWC_MARKER = "Managed by WESP kiosk boot"
|
||
_DEFAULT_SYSTEM_LABWC = "/etc/xdg/labwc/autostart"
|
||
_DEFAULT_CHROMIUM_POLICY = Path("/etc/chromium/policies/managed/wesp-kiosk.json")
|
||
_CHROMIUM_PROFILE_DIRNAME = "wesp-kiosk-chromium"
|
||
# MAP → 127.0.0.1: сеть перевода не достигается (дополнение к policy и meta notranslate).
|
||
_DEFAULT_CHROMIUM_HOST_RULES = (
|
||
"MAP translate.googleapis.com 127.0.0.1,"
|
||
" MAP translate-pa.googleapis.com 127.0.0.1,"
|
||
" MAP translation.googleapis.com 127.0.0.1,"
|
||
" MAP translate.google.com 127.0.0.1"
|
||
)
|
||
|
||
def _is_root() -> bool:
|
||
geteuid = getattr(os, "geteuid", None)
|
||
if geteuid is None:
|
||
return False
|
||
try:
|
||
return geteuid() == 0
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
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 _chromium_user_data_dir(app_config: Dict[str, Any], home: Path) -> Path:
|
||
custom = (app_config.get("WESP_KIOSK_CHROMIUM_USER_DATA_DIR") or "").strip()
|
||
if custom:
|
||
return Path(custom)
|
||
return home / ".config" / _CHROMIUM_PROFILE_DIRNAME
|
||
|
||
|
||
def _chromium_policy_path(app_config: Dict[str, Any]) -> Path:
|
||
custom = (app_config.get("WESP_KIOSK_CHROMIUM_POLICY_PATH") or "").strip()
|
||
if custom:
|
||
return Path(custom)
|
||
return _DEFAULT_CHROMIUM_POLICY
|
||
|
||
|
||
def _managed_policy_content() -> str:
|
||
template = Path(__file__).resolve().parents[2] / "install" / "kiosk" / "chromium-managed-policy.json"
|
||
if template.is_file():
|
||
return template.read_text(encoding="utf-8")
|
||
return (
|
||
'{"TranslateEnabled":false,"OfferTranslateEnabled":false,'
|
||
'"DefaultNotificationsSetting":2}\n'
|
||
)
|
||
|
||
|
||
def _chromium_preferences_content() -> str:
|
||
return json.dumps(
|
||
{
|
||
"translate": {"enabled": False},
|
||
"translate_site_blocklist": ["localhost", "127.0.0.1"],
|
||
"intl": {"accept_languages": "ru-RU,ru"},
|
||
"profile": {
|
||
"default_content_setting_values": {"notifications": 2},
|
||
},
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
|
||
def _ensure_chromium_translate_block(
|
||
home: Path,
|
||
user: str,
|
||
app_config: Dict[str, Any],
|
||
) -> Tuple[bool, str]:
|
||
"""Профиль Chromium + enterprise policy (перевод и уведомления)."""
|
||
profile_root = _chromium_user_data_dir(app_config, home)
|
||
prefs_path = profile_root / "Default" / "Preferences"
|
||
ok, msg = _write_file_as_user(
|
||
prefs_path,
|
||
_chromium_preferences_content(),
|
||
user,
|
||
)
|
||
if not ok:
|
||
return False, f"Preferences: {msg}"
|
||
# Каталоги могли создаться под root — зафиксируем владельца и доступы.
|
||
if _is_root():
|
||
_chown_tree(profile_root, user)
|
||
try:
|
||
profile_root.chmod(0o700)
|
||
except OSError:
|
||
pass
|
||
|
||
policy_path = _chromium_policy_path(app_config)
|
||
policy_path.parent.mkdir(parents=True, exist_ok=True)
|
||
ok, msg = _write_system_file(policy_path, _managed_policy_content(), executable=False)
|
||
if not ok:
|
||
return False, f"policy: {msg}"
|
||
return True, "Chromium profile + policy"
|
||
|
||
|
||
def _chromium_host_rules(app_config: Dict[str, Any]) -> str:
|
||
custom = (app_config.get("WESP_KIOSK_CHROMIUM_HOST_RULES") or "").strip()
|
||
return custom or _DEFAULT_CHROMIUM_HOST_RULES
|
||
|
||
|
||
def _chromium_flag_list(
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
home: Path,
|
||
) -> Tuple[str, ...]:
|
||
"""Флаги kiosk-Chromium (Debian 13 / Chromium 147+)."""
|
||
extra = (app_config.get("WESP_KIOSK_CHROMIUM_EXTRA_FLAGS") or "").strip()
|
||
gl = (app_config.get("WESP_KIOSK_CHROMIUM_GL") or "").strip().lower()
|
||
user_data = _chromium_user_data_dir(app_config, home)
|
||
host_rules = _chromium_host_rules(app_config)
|
||
flags: List[str] = [
|
||
# Wayland (labwc): без Ozone часть сборок Chromium рисует чёрный экран.
|
||
"--enable-features=UseOzonePlatform",
|
||
"--ozone-platform=wayland",
|
||
"--kiosk",
|
||
"--no-first-run",
|
||
"--no-default-browser-check",
|
||
"--disable-session-crashed-bubble",
|
||
"--disable-restore-session-state",
|
||
f"--user-data-dir={user_data}",
|
||
"--password-store=basic",
|
||
"--disable-infobars",
|
||
"--disable-notifications",
|
||
"--disable-translate",
|
||
"--disable-translate-new-ux",
|
||
"--lang=ru",
|
||
"--accept-lang=ru-RU,ru",
|
||
(
|
||
"--disable-features="
|
||
"Translate,TranslateUI,TranslateOffer,TranslateBubble,"
|
||
"TranslateNewUX,OptimizationHints"
|
||
),
|
||
f"--host-rules={host_rules}",
|
||
]
|
||
if gl in ("swiftshader", "software"):
|
||
flags.insert(0, "--use-gl=swiftshader")
|
||
elif gl in ("egl",):
|
||
flags.insert(0, "--use-gl=egl")
|
||
if extra:
|
||
flags.extend(shlex.split(extra))
|
||
return tuple(flags)
|
||
|
||
|
||
def _gui_user(app_config: Dict[str, Any]) -> str:
|
||
return (app_config.get("WESP_KIOSK_GUI_USER") or "komton").strip() or "komton"
|
||
|
||
|
||
def _kiosk_target_url(app_config: Dict[str, Any]) -> str:
|
||
"""Конечная страница после /starting (весы и т.д.)."""
|
||
explicit = (app_config.get("WESP_KIOSK_TARGET_URL") or "").strip()
|
||
if explicit:
|
||
return explicit
|
||
legacy = (app_config.get("WESP_KIOSK_URL") or "").strip()
|
||
if legacy and "/starting" not in legacy:
|
||
return legacy
|
||
landing = default_app_landing_path()
|
||
return f"http://127.0.0.1{landing}"
|
||
|
||
|
||
def _kiosk_launch_url(app_config: Dict[str, Any]) -> str:
|
||
"""URL в Chromium: /starting с тёмной темой или сразу target."""
|
||
forced = (app_config.get("WESP_KIOSK_URL") or "").strip()
|
||
if forced:
|
||
return forced
|
||
|
||
target = _kiosk_target_url(app_config)
|
||
if not _env_bool(app_config.get("WESP_KIOSK_STARTUP_SCREEN"), True):
|
||
return target
|
||
|
||
parsed = urlparse(target)
|
||
if parsed.scheme and parsed.netloc:
|
||
base = f"{parsed.scheme}://{parsed.netloc}"
|
||
else:
|
||
base = "http://127.0.0.1"
|
||
next_path = parsed.path or default_app_landing_path()
|
||
if parsed.query:
|
||
next_path = f"{next_path}?{parsed.query}"
|
||
theme = (app_config.get("WESP_KIOSK_STARTUP_THEME") or "dark").strip() or "dark"
|
||
q = f"next={quote(next_path, safe='/?:=&')}&theme={quote(theme, safe='')}"
|
||
return f"{base}/starting?{q}"
|
||
|
||
|
||
def _chromium_bin(app_config: Dict[str, Any]) -> str:
|
||
return (app_config.get("WESP_KIOSK_CHROMIUM") or "/usr/bin/chromium").strip()
|
||
|
||
|
||
def _system_labwc_paths(app_config: Dict[str, Any]) -> Tuple[Path, Path]:
|
||
base = (
|
||
app_config.get("WESP_KIOSK_SYSTEM_LABWC_AUTOSTART") or _DEFAULT_SYSTEM_LABWC
|
||
).strip()
|
||
path = Path(base)
|
||
return path, Path(f"{path}.wesp-backup")
|
||
|
||
|
||
def _home_dir(user: str) -> Path:
|
||
if pwd is not None:
|
||
try:
|
||
return Path(pwd.getpwnam(user).pw_dir)
|
||
except KeyError:
|
||
pass
|
||
return Path(f"/home/{user}")
|
||
|
||
|
||
def _state_path(home: Path) -> Path:
|
||
return home / ".config" / _STATE_DIR_NAME / _STATE_FILE
|
||
|
||
|
||
def _autostart_dir(home: Path) -> Path:
|
||
return home / ".config" / "autostart"
|
||
|
||
|
||
def _user_labwc_autostart_path(home: Path) -> Path:
|
||
return home / ".config" / "labwc" / "autostart"
|
||
|
||
|
||
def _chromium_exec_line(
|
||
chromium: str,
|
||
url: str,
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
home: Path,
|
||
) -> str:
|
||
# --app=URL: один документ, без «залипания» на / или NTP; & в query не ломает shell.
|
||
return " ".join(
|
||
[
|
||
shlex.quote(chromium),
|
||
*_chromium_flag_list(app_config, home=home),
|
||
f"--app={shlex.quote(url)}",
|
||
]
|
||
)
|
||
|
||
|
||
def _chown_path(path: Path, user: str) -> None:
|
||
if pwd is None:
|
||
return
|
||
try:
|
||
pw = pwd.getpwnam(user)
|
||
os.chown(path, pw.pw_uid, pw.pw_gid)
|
||
except (KeyError, OSError, AttributeError):
|
||
pass
|
||
|
||
|
||
def _chown_tree(path: Path, user: str) -> None:
|
||
if pwd is None:
|
||
return
|
||
try:
|
||
pw = pwd.getpwnam(user)
|
||
uid, gid = pw.pw_uid, pw.pw_gid
|
||
except KeyError:
|
||
return
|
||
if path.is_dir():
|
||
for root, _dirs, files in os.walk(path):
|
||
for name in files:
|
||
try:
|
||
os.chown(os.path.join(root, name), uid, gid)
|
||
except OSError:
|
||
pass
|
||
try:
|
||
os.chown(path, uid, gid)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def build_chromium_desktop_content(
|
||
*, chromium: str, url: str, app_config: Dict[str, Any], home: Path
|
||
) -> str:
|
||
return (
|
||
"[Desktop Entry]\n"
|
||
"Type=Application\n"
|
||
"Name=Chromium Kiosk\n"
|
||
f"Exec={_chromium_exec_line(chromium, url, app_config, home=home)}\n"
|
||
"Hidden=false\n"
|
||
"X-GNOME-Autostart-enabled=true\n"
|
||
)
|
||
|
||
|
||
def _kiosk_chromium_boot_delay_sec(app_config: Dict[str, Any]) -> int:
|
||
"""Доп. пауза после появления HTTP (миграции / SD). 0 = только wait-for-http."""
|
||
raw = app_config.get("WESP_KIOSK_CHROMIUM_BOOT_DELAY_SEC", 2)
|
||
try:
|
||
return max(0, min(30, int(raw)))
|
||
except (TypeError, ValueError):
|
||
return 2
|
||
|
||
|
||
def _kiosk_wait_http_max_sec(app_config: Dict[str, Any]) -> int:
|
||
raw = app_config.get("WESP_KIOSK_WAIT_HTTP_MAX_SEC", 90)
|
||
try:
|
||
return max(5, min(180, int(raw)))
|
||
except (TypeError, ValueError):
|
||
return 90
|
||
|
||
|
||
def build_labwc_autostart_content(
|
||
*, chromium: str, url: str, app_config: Dict[str, Any], home: Path
|
||
) -> str:
|
||
"""Минимальный autostart для labwc-pi: kanshi + Chromium, без pcmanfm/панели."""
|
||
exec_line = _chromium_exec_line(chromium, url, app_config, home=home)
|
||
log_path = home / ".cache" / "wesp-kiosk-chromium.log"
|
||
wait_max = _kiosk_wait_http_max_sec(app_config)
|
||
boot_delay = _kiosk_chromium_boot_delay_sec(app_config)
|
||
inner = (
|
||
f"for i in $(seq 1 {wait_max}); do "
|
||
'curl -sS -m 2 -o /dev/null "http://127.0.0.1/starting" 2>/dev/null && break; '
|
||
"sleep 1; done; "
|
||
f"[ {boot_delay} -gt 0 ] && sleep {boot_delay}; "
|
||
f"{exec_line} --enable-logging=stderr --v=1 >>\"{log_path}\" 2>&1"
|
||
)
|
||
wrapped = f"bash -lc {shlex.quote(inner)}"
|
||
return (
|
||
"#!/bin/sh\n"
|
||
f"# {_LABWC_MARKER}\n"
|
||
"/usr/bin/kanshi &\n"
|
||
f"{wrapped} &\n"
|
||
)
|
||
|
||
|
||
def _is_wesp_labwc_autostart(path: Path) -> bool:
|
||
if not path.is_file():
|
||
return False
|
||
try:
|
||
return _LABWC_MARKER in path.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _read_url_from_text(text: str) -> Optional[str]:
|
||
m = re.search(r"https?://\S+", text)
|
||
return m.group(0).rstrip("'\"") if m else None
|
||
|
||
|
||
def _read_desktop_url(path: Path) -> Optional[str]:
|
||
if not path.is_file():
|
||
return None
|
||
try:
|
||
text = path.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
return None
|
||
for line in text.splitlines():
|
||
if line.strip().lower().startswith("exec="):
|
||
return _read_url_from_text(line)
|
||
return None
|
||
|
||
|
||
def _load_state(state_path: Path) -> Dict[str, Any]:
|
||
if not state_path.is_file():
|
||
return {}
|
||
try:
|
||
return json.loads(state_path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return {}
|
||
|
||
|
||
def _save_state(state_path: Path, data: Dict[str, Any], user: str) -> None:
|
||
state_path.parent.mkdir(parents=True, exist_ok=True)
|
||
state_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||
_chown_tree(state_path.parent, user)
|
||
|
||
|
||
def _write_system_file(path: Path, content: str, *, executable: bool = True) -> Tuple[bool, str]:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
path.write_text(content, encoding="utf-8")
|
||
if executable:
|
||
path.chmod(0o755)
|
||
return True, "Записано"
|
||
except OSError as exc:
|
||
with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False, suffix=".sh") 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()
|
||
return False, f"{exc}; sudo: {err}"
|
||
if executable:
|
||
subprocess.run(["sudo", "chmod", "755", str(path)], check=False, timeout=30)
|
||
return True, "Записано (sudo)"
|
||
finally:
|
||
try:
|
||
Path(tmp_path).unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _ensure_system_backup(system: Path, backup: Path) -> Tuple[bool, str]:
|
||
if backup.is_file():
|
||
return True, "Резервная копия уже есть"
|
||
if not system.is_file():
|
||
return False, f"Не найден {system}"
|
||
try:
|
||
shutil.copy2(system, backup)
|
||
return True, "Создана резервная копия autostart"
|
||
except OSError as exc:
|
||
return False, str(exc)
|
||
|
||
|
||
def _restore_system_autostart(system: Path, backup: Path) -> Tuple[bool, str]:
|
||
if not backup.is_file():
|
||
return True, "Восстановление не требуется"
|
||
try:
|
||
shutil.copy2(backup, system)
|
||
system.chmod(0o755)
|
||
return True, "Восстановлен исходный /etc/xdg/labwc/autostart"
|
||
except OSError as exc:
|
||
proc = subprocess.run(
|
||
["sudo", "cp", str(backup), str(system)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "").strip()
|
||
return False, f"{exc}; sudo: {err}"
|
||
subprocess.run(["sudo", "chmod", "755", str(system)], check=False, timeout=30)
|
||
return True, "Восстановлено (sudo)"
|
||
|
||
|
||
def _patch_system_labwc(
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
chromium: str,
|
||
url: str,
|
||
) -> Tuple[bool, str]:
|
||
system, backup = _system_labwc_paths(app_config)
|
||
ok, msg = _ensure_system_backup(system, backup)
|
||
if not ok:
|
||
return False, msg
|
||
|
||
home = _home_dir(_gui_user(app_config))
|
||
content = build_labwc_autostart_content(
|
||
chromium=chromium, url=url, app_config=app_config, home=home
|
||
)
|
||
ok, msg = _write_system_file(system, content, executable=True)
|
||
if not ok:
|
||
return False, msg
|
||
return True, f"Патч {system}"
|
||
|
||
|
||
def _unpatch_system_labwc(app_config: Dict[str, Any]) -> Tuple[bool, str]:
|
||
system, backup = _system_labwc_paths(app_config)
|
||
if _is_wesp_labwc_autostart(system):
|
||
return _restore_system_autostart(system, backup)
|
||
return True, "Системный autostart не изменён WESP"
|
||
|
||
|
||
def _write_file_as_user(
|
||
path: Path,
|
||
content: str,
|
||
user: str,
|
||
*,
|
||
write_cmd: str = "",
|
||
executable: bool = False,
|
||
) -> Tuple[bool, str]:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
if _is_root():
|
||
path.write_text(content, encoding="utf-8")
|
||
if executable:
|
||
path.chmod(0o755)
|
||
_chown_path(path, user)
|
||
_chown_path(path.parent, user)
|
||
# Профиль Chromium часто в ~/.config/... (user-data-dir).
|
||
# Если каталоги созданы от root, Chromium не сможет создать SingletonLock.
|
||
try:
|
||
_chown_path(path.parent.parent, user)
|
||
except Exception:
|
||
pass
|
||
return True, "Записано"
|
||
path.write_text(content, encoding="utf-8")
|
||
if executable:
|
||
path.chmod(0o755)
|
||
return True, "Записано"
|
||
except OSError:
|
||
pass
|
||
|
||
with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False, suffix=".tmp") as tmp:
|
||
tmp.write(content)
|
||
tmp_path = tmp.name
|
||
|
||
try:
|
||
if write_cmd:
|
||
args = shlex.split(write_cmd) + [str(path)]
|
||
proc = subprocess.run(
|
||
args,
|
||
input=content,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
)
|
||
else:
|
||
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()
|
||
return False, (err or "write failed") + " Нужен root или WESP_KIOSK_WRITE_CMD."
|
||
if executable:
|
||
subprocess.run(["sudo", "chmod", "755", str(path)], check=False, timeout=30)
|
||
if _is_root():
|
||
_chown_path(path, user)
|
||
return True, "Записано (sudo)"
|
||
finally:
|
||
try:
|
||
Path(tmp_path).unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _remove_path(path: Path, *, write_cmd: str = "") -> Tuple[bool, str]:
|
||
if not path.exists():
|
||
return True, "Уже отсутствует"
|
||
try:
|
||
if _is_root() or (path.is_file() and os.access(path, os.W_OK)):
|
||
path.unlink(missing_ok=True)
|
||
return True, "Удалено"
|
||
except OSError:
|
||
pass
|
||
if write_cmd:
|
||
args = shlex.split(write_cmd) + ["rm", "-f", str(path)]
|
||
proc = subprocess.run(args, capture_output=True, text=True, timeout=30)
|
||
else:
|
||
proc = subprocess.run(
|
||
["sudo", "rm", "-f", str(path)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30,
|
||
)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "").strip()
|
||
return False, err or "rm failed"
|
||
return True, "Удалено (sudo)"
|
||
|
||
|
||
def get_kiosk_boot_status(app_config: Dict[str, Any]) -> Dict[str, Any]:
|
||
user = _gui_user(app_config)
|
||
home = _home_dir(user)
|
||
autostart = _autostart_dir(home)
|
||
desktop_path = autostart / _DESKTOP_ENTRY_NAME
|
||
user_labwc = _user_labwc_autostart_path(home)
|
||
system_labwc, system_backup = _system_labwc_paths(app_config)
|
||
chromium = _chromium_bin(app_config)
|
||
launch_url = _kiosk_launch_url(app_config)
|
||
target_url = _kiosk_target_url(app_config)
|
||
|
||
state = _load_state(_state_path(home))
|
||
hide_desktop = bool(state.get("hide_desktop"))
|
||
system_active = _is_wesp_labwc_autostart(system_labwc)
|
||
user_labwc_active = _is_wesp_labwc_autostart(user_labwc)
|
||
xdg_enabled = desktop_path.is_file()
|
||
|
||
enabled = system_active or user_labwc_active or xdg_enabled
|
||
detected_url = None
|
||
if system_active:
|
||
try:
|
||
detected_url = _read_url_from_text(
|
||
system_labwc.read_text(encoding="utf-8", errors="replace"),
|
||
)
|
||
except OSError:
|
||
pass
|
||
elif user_labwc_active:
|
||
try:
|
||
detected_url = _read_url_from_text(
|
||
user_labwc.read_text(encoding="utf-8", errors="replace"),
|
||
)
|
||
except OSError:
|
||
pass
|
||
elif xdg_enabled:
|
||
detected_url = _read_desktop_url(desktop_path)
|
||
|
||
if system_active:
|
||
mode = "system_labwc"
|
||
msg = f"Киоск: патч {system_labwc} (без рабочего стола)"
|
||
elif user_labwc_active:
|
||
mode = "user_labwc_legacy"
|
||
msg = (
|
||
f"Устаревший {user_labwc} — включите киоск снова "
|
||
f"(нужен патч {system_labwc})"
|
||
)
|
||
elif xdg_enabled:
|
||
mode = "xdg"
|
||
msg = "Киоск XDG (рабочий стол возможен до браузера)"
|
||
else:
|
||
mode = "off"
|
||
msg = "Киоск выключен"
|
||
|
||
return {
|
||
"gui_user": user,
|
||
"home_dir": str(home),
|
||
"kiosk_enabled": enabled,
|
||
"kiosk_mode": mode,
|
||
"hide_desktop": hide_desktop or system_active,
|
||
"system_labwc_autostart_path": str(system_labwc),
|
||
"system_labwc_patched": system_active,
|
||
"system_labwc_backup_path": str(system_backup),
|
||
"system_labwc_backup_exists": system_backup.is_file(),
|
||
"user_labwc_autostart_path": str(user_labwc),
|
||
"user_labwc_autostart_active": user_labwc_active,
|
||
"autostart_path": str(desktop_path),
|
||
"configured_url": launch_url,
|
||
"target_url": target_url,
|
||
"active_url": detected_url or (launch_url if enabled else None),
|
||
"startup_screen": _env_bool(app_config.get("WESP_KIOSK_STARTUP_SCREEN"), True),
|
||
"chromium_path": chromium,
|
||
"chromium_found": Path(chromium).is_file(),
|
||
"home_exists": home.is_dir(),
|
||
"running_as_root": _is_root(),
|
||
"message": msg,
|
||
}
|
||
|
||
|
||
def set_kiosk_boot(
|
||
enabled: bool,
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
hide_desktop: Optional[bool] = None,
|
||
schedule_reboot: bool = True,
|
||
) -> Dict[str, Any]:
|
||
user = _gui_user(app_config)
|
||
home = _home_dir(user)
|
||
if not home.is_dir():
|
||
return {
|
||
"ok": False,
|
||
"code": "no_home",
|
||
"message": f"Домашний каталог не найден: {home}",
|
||
}
|
||
|
||
autostart = _autostart_dir(home)
|
||
desktop_path = autostart / _DESKTOP_ENTRY_NAME
|
||
user_labwc = _user_labwc_autostart_path(home)
|
||
state_path = _state_path(home)
|
||
write_cmd = (app_config.get("WESP_KIOSK_WRITE_CMD") or "").strip()
|
||
chromium = _chromium_bin(app_config)
|
||
launch_url = _kiosk_launch_url(app_config)
|
||
target_url = _kiosk_target_url(app_config)
|
||
system_labwc, _backup = _system_labwc_paths(app_config)
|
||
|
||
if not Path(chromium).is_file() and enabled:
|
||
return {
|
||
"ok": False,
|
||
"code": "chromium_missing",
|
||
"message": f"Chromium не найден: {chromium}",
|
||
}
|
||
|
||
state = _load_state(state_path)
|
||
if hide_desktop is not None:
|
||
state["hide_desktop"] = bool(hide_desktop)
|
||
apply_hide = bool(state.get("hide_desktop"))
|
||
|
||
if not enabled:
|
||
ok, msg = _unpatch_system_labwc(app_config)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": msg}
|
||
|
||
for path in (desktop_path, user_labwc):
|
||
ok, rm_msg = _remove_path(path, write_cmd=write_cmd)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": rm_msg}
|
||
|
||
state = {"hide_desktop": False, "system_patched": False}
|
||
_save_state(state_path, state, user)
|
||
|
||
return _maybe_attach_auto_reboot(
|
||
app_config,
|
||
{
|
||
"ok": True,
|
||
"code": "disabled",
|
||
"message": f"{msg} Киоск отключён.",
|
||
"kiosk_enabled": False,
|
||
"hide_desktop": False,
|
||
},
|
||
schedule_reboot=schedule_reboot,
|
||
)
|
||
|
||
# Всегда убираем устаревший ~/.config/labwc/autostart (labwc-pi его не использует).
|
||
_remove_path(user_labwc, write_cmd=write_cmd)
|
||
|
||
ok, chrome_msg = _ensure_chromium_translate_block(home, user, app_config)
|
||
if not ok:
|
||
return {"ok": False, "code": "chromium_profile_error", "message": chrome_msg}
|
||
|
||
if apply_hide:
|
||
ok, patch_msg = _patch_system_labwc(
|
||
app_config, chromium=chromium, url=launch_url
|
||
)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": patch_msg}
|
||
|
||
ok, msg = _remove_path(desktop_path, write_cmd=write_cmd)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": msg}
|
||
|
||
hide_note = (
|
||
f" Патч {system_labwc}: без pcmanfm/панели; Chromium сразу."
|
||
)
|
||
kiosk_mode = "system_labwc"
|
||
state = {"hide_desktop": True, "system_patched": True}
|
||
else:
|
||
ok, unpatch_msg = _unpatch_system_labwc(app_config)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": unpatch_msg}
|
||
|
||
content = build_chromium_desktop_content(
|
||
chromium=chromium,
|
||
url=launch_url,
|
||
app_config=app_config,
|
||
home=home,
|
||
)
|
||
ok, msg = _write_file_as_user(
|
||
desktop_path,
|
||
content,
|
||
user,
|
||
write_cmd=write_cmd,
|
||
)
|
||
if not ok:
|
||
return {"ok": False, "code": "write_error", "message": msg}
|
||
|
||
hide_note = (
|
||
" XDG autostart; рабочий стол может мелькать — включите «Без рабочего стола»."
|
||
)
|
||
kiosk_mode = "xdg"
|
||
state = {"hide_desktop": False, "system_patched": False}
|
||
|
||
_save_state(state_path, state, user)
|
||
|
||
return _maybe_attach_auto_reboot(
|
||
app_config,
|
||
{
|
||
"ok": True,
|
||
"code": "enabled",
|
||
"message": (
|
||
f"Киоск включён → {launch_url} (далее {target_url}).{hide_note}"
|
||
),
|
||
"kiosk_enabled": True,
|
||
"hide_desktop": apply_hide,
|
||
"kiosk_mode": kiosk_mode,
|
||
"system_labwc_autostart_path": str(system_labwc),
|
||
"user_labwc_autostart_path": str(user_labwc),
|
||
"autostart_path": str(desktop_path),
|
||
"home_dir": str(home),
|
||
"target_url": target_url,
|
||
},
|
||
schedule_reboot=schedule_reboot,
|
||
)
|
||
|
||
|
||
def _maybe_attach_auto_reboot(
|
||
app_config: Dict[str, Any],
|
||
result: Dict[str, Any],
|
||
*,
|
||
schedule_reboot: bool,
|
||
) -> Dict[str, Any]:
|
||
if not schedule_reboot:
|
||
result["reboot_scheduled"] = False
|
||
return result
|
||
return _attach_auto_reboot(app_config, result)
|
||
|
||
|
||
def _attach_auto_reboot(app_config: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]:
|
||
from app.services.system_restart_service import maybe_auto_reboot_after_kiosk_change
|
||
|
||
reboot = maybe_auto_reboot_after_kiosk_change(app_config)
|
||
result["reboot_scheduled"] = bool(reboot.get("scheduled"))
|
||
if reboot.get("scheduled") and reboot.get("message"):
|
||
base = str(result.get("message") or "").rstrip(".")
|
||
result["message"] = f"{base}. {reboot['message']}"
|
||
return result
|