402 lines
13 KiB
Python
402 lines
13 KiB
Python
"""Настройка Pi: /etc/wesp/wesp.env + systemd unit (как scripts/install_fresh.sh)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
import secrets
|
||
import subprocess
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from app.services.system_restart_service import schedule_service_restart
|
||
|
||
_DEFAULT_ENV_PATH = Path("/etc/wesp/wesp.env")
|
||
_ENV_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=")
|
||
|
||
|
||
def _is_root() -> bool:
|
||
geteuid = getattr(os, "geteuid", None)
|
||
if geteuid is None:
|
||
return False
|
||
try:
|
||
return geteuid() == 0
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _wesp_home(app_config: Dict[str, Any]) -> Path:
|
||
return Path(app_config.get("BASE_DIR") or ".").resolve()
|
||
|
||
|
||
def _venv_dir(app_config: Dict[str, Any]) -> Path:
|
||
raw = (
|
||
(app_config.get("WESP_VENV") or os.environ.get("WESP_VENV") or "").strip()
|
||
)
|
||
if raw:
|
||
return Path(raw)
|
||
base = _wesp_home(app_config)
|
||
for candidate in (base.parent / "wes", Path("/opt/wes")):
|
||
if (candidate / "bin" / "python3").is_file():
|
||
return candidate
|
||
return Path("/opt/wes")
|
||
|
||
|
||
def _python_bin(app_config: Dict[str, Any], venv: Path) -> Path:
|
||
raw = (app_config.get("WESP_PYTHON") or os.environ.get("WESP_PYTHON") or "").strip()
|
||
if raw:
|
||
return Path(raw)
|
||
py3 = venv / "bin" / "python3"
|
||
return py3 if py3.is_file() else venv / "bin" / "python"
|
||
|
||
|
||
def _env_file_path(app_config: Dict[str, Any]) -> Path:
|
||
raw = (
|
||
app_config.get("WESP_ENV_FILE")
|
||
or os.environ.get("WESP_ENV_FILE")
|
||
or ""
|
||
).strip()
|
||
return Path(raw) if raw else _DEFAULT_ENV_PATH
|
||
|
||
|
||
def _systemd_unit_name(app_config: Dict[str, Any]) -> str:
|
||
return (
|
||
app_config.get("WESP_SYSTEMD_UNIT")
|
||
or os.environ.get("WESP_SYSTEMD_UNIT")
|
||
or "myscript"
|
||
).strip() or "myscript"
|
||
|
||
|
||
def _patch_paths(text: str, home: Path, venv: Path) -> str:
|
||
return (
|
||
text.replace("/opt/wesp", str(home))
|
||
.replace("/opt/wes", str(venv))
|
||
)
|
||
|
||
|
||
def _env_defaults(app_config: Dict[str, Any], home: Path, venv: Path, py: Path) -> Dict[str, str]:
|
||
unit = _systemd_unit_name(app_config)
|
||
return {
|
||
"WESP_SYSTEMD_UNIT": unit,
|
||
"WESP_ADMIN_RESTART_CMD": f"systemctl restart {unit}",
|
||
"WESP_ADMIN_STOP_CMD": f"systemctl stop {unit}",
|
||
"WESP_UPDATE_RESTART_CMD": f"bash {home}/scripts/post_update.sh",
|
||
"WESP_LISTEN": "0.0.0.0:80",
|
||
"WESP_HEALTH_URL": "http://127.0.0.1/api/health",
|
||
"WESP_ADMIN_REBOOT_CMD": (
|
||
app_config.get("WESP_ADMIN_REBOOT_CMD") or "sudo /sbin/reboot"
|
||
).strip(),
|
||
"WESP_AUTO_REBOOT_ENABLED": "1",
|
||
"WESP_AUTO_REBOOT_DELAY_SEC": str(
|
||
app_config.get("WESP_AUTO_REBOOT_DELAY_SEC", 3)
|
||
),
|
||
"WESP_FULL_SETUP_REBOOT_DELAY_SEC": str(
|
||
app_config.get("WESP_FULL_SETUP_REBOOT_DELAY_SEC", 10)
|
||
),
|
||
"WESP_KIOSK_AUTO_REBOOT": "1",
|
||
"WESP_PI_BOOT_AUTO_REBOOT": "1",
|
||
"WESP_PLYMOUTH_AUTO_REBOOT": "1",
|
||
"WESP_KIOSK_GUI_USER": (
|
||
app_config.get("WESP_KIOSK_GUI_USER") or "komton"
|
||
).strip(),
|
||
"WESP_KIOSK_TARGET_URL": (
|
||
app_config.get("WESP_KIOSK_TARGET_URL") or "http://localhost/scales"
|
||
).strip(),
|
||
"WESP_KIOSK_STARTUP_SCREEN": "1",
|
||
"WESP_KIOSK_STARTUP_THEME": (
|
||
app_config.get("WESP_KIOSK_STARTUP_THEME") or "dark"
|
||
).strip(),
|
||
"WESP_VENV": str(venv),
|
||
"WESP_PYTHON": str(py),
|
||
}
|
||
|
||
|
||
def _ensure_secret_key_in_env(env_path: Path) -> bool:
|
||
"""
|
||
WESP_CONFIG=production требует WESP_SECRET_KEY.
|
||
Если ключ отсутствует или равен дефолту — генерируем и дописываем в конец файла.
|
||
"""
|
||
try:
|
||
text = env_path.read_text(encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
# если не смогли прочитать — пусть упадёт дальше в validate_production_config
|
||
return False
|
||
|
||
# Если уже задан НЕ дефолтный ключ — ничего не делаем.
|
||
# Берём последнее вхождение ключа (как сделает shell при source).
|
||
last_key_line = ""
|
||
for line in text.splitlines():
|
||
if line.strip().startswith("WESP_SECRET_KEY="):
|
||
last_key_line = line.strip()
|
||
if last_key_line and "change-me-in-production" not in last_key_line:
|
||
return False
|
||
|
||
secret = secrets.token_urlsafe(64)
|
||
new_line = f"WESP_SECRET_KEY={secret}"
|
||
|
||
# 1) Если есть дефолтная строка — заменяем её (чтобы не копить дубли).
|
||
replaced = False
|
||
def_pat = re.compile(r"^WESP_SECRET_KEY=change-me-in-production\s*$", re.MULTILINE)
|
||
if def_pat.search(text):
|
||
text = def_pat.sub(new_line, text, count=1)
|
||
replaced = True
|
||
|
||
# 2) Иначе, если ключа вообще нет — дописываем в конец.
|
||
if not replaced and "WESP_SECRET_KEY=" not in text:
|
||
if text and not text.endswith("\n"):
|
||
text += "\n"
|
||
text += new_line + "\n"
|
||
replaced = True
|
||
|
||
if not replaced:
|
||
# Ключ есть, но дефолтный и не в точной форме — добавим в конец.
|
||
if text and not text.endswith("\n"):
|
||
text += "\n"
|
||
text += new_line + "\n"
|
||
|
||
try:
|
||
env_path.write_text(text, encoding="utf-8")
|
||
try:
|
||
env_path.chmod(0o600)
|
||
except OSError:
|
||
pass
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _read_env_keys(path: Path) -> Dict[str, str]:
|
||
if not path.is_file():
|
||
return {}
|
||
keys: Dict[str, str] = {}
|
||
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
|
||
m = _ENV_KEY_RE.match(line.strip())
|
||
if m:
|
||
keys[m.group(1)] = line
|
||
return keys
|
||
|
||
|
||
def _merge_env_file(
|
||
path: Path,
|
||
defaults: Dict[str, str],
|
||
*,
|
||
home: Path,
|
||
venv: Path,
|
||
) -> Tuple[List[str], List[str]]:
|
||
"""Вернуть (добавленные ключи, обновлённые при apply_force)."""
|
||
added: List[str] = []
|
||
if not path.is_file():
|
||
example = home / "install" / "wesp.env.example"
|
||
if example.is_file():
|
||
text = _patch_paths(example.read_text(encoding="utf-8"), home, venv)
|
||
else:
|
||
text = "WESP_CONFIG=production\n"
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
try:
|
||
path.chmod(0o600)
|
||
except OSError:
|
||
pass
|
||
added.append("(создан из wesp.env.example)")
|
||
|
||
existing = _read_env_keys(path)
|
||
lines_to_append: List[str] = []
|
||
for key, value in defaults.items():
|
||
if key not in existing:
|
||
lines_to_append.append(f"{key}={value}")
|
||
added.append(key)
|
||
|
||
if lines_to_append:
|
||
with path.open("a", encoding="utf-8") as fh:
|
||
if path.stat().st_size and not existing:
|
||
pass
|
||
fh.write("\n".join(lines_to_append) + "\n")
|
||
|
||
# Подправить устаревшие пути /opt/wesp в существующем файле
|
||
try:
|
||
original = path.read_text(encoding="utf-8")
|
||
patched = _patch_paths(original, home, venv)
|
||
if patched != original:
|
||
path.write_text(patched, encoding="utf-8")
|
||
added.append("(пути /opt/wesp → актуальные)")
|
||
except OSError:
|
||
pass
|
||
|
||
return added, []
|
||
|
||
|
||
def _unit_template_path(home: Path) -> Optional[Path]:
|
||
for name in ("myscript.service.example", "wesp.service.example"):
|
||
p = home / "install" / name
|
||
if p.is_file():
|
||
return p
|
||
return None
|
||
|
||
|
||
def _install_systemd_unit(
|
||
app_config: Dict[str, Any],
|
||
home: Path,
|
||
venv: Path,
|
||
py: Path,
|
||
) -> Tuple[bool, str, Path]:
|
||
unit = _systemd_unit_name(app_config)
|
||
dest = Path(f"/etc/systemd/system/{unit}.service")
|
||
template = _unit_template_path(home)
|
||
if not template:
|
||
return False, "Шаблон install/*.service.example не найден", dest
|
||
|
||
# Шаблон unit должен задавать реальный WSGI-сервер (waitress/gunicorn),
|
||
# а не Flask dev-server (run.py debug=True). Поэтому не переписываем ExecStart автоматически.
|
||
text = _patch_paths(template.read_text(encoding="utf-8"), home, venv)
|
||
|
||
try:
|
||
dest.write_text(text, encoding="utf-8")
|
||
dest.chmod(0o644)
|
||
except OSError as exc:
|
||
return False, str(exc), dest
|
||
|
||
for cmd in (
|
||
["systemctl", "daemon-reload"],
|
||
["systemctl", "enable", f"{unit}.service"],
|
||
):
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "").strip()
|
||
return False, f"{' '.join(cmd)}: {err}", dest
|
||
|
||
return True, f"Установлен {dest}", dest
|
||
|
||
|
||
def get_pi_platform_setup_status(app_config: Dict[str, Any]) -> Dict[str, Any]:
|
||
home = _wesp_home(app_config)
|
||
venv = _venv_dir(app_config)
|
||
env_path = _env_file_path(app_config)
|
||
unit = _systemd_unit_name(app_config)
|
||
unit_path = Path(f"/etc/systemd/system/{unit}.service")
|
||
defaults = _env_defaults(app_config, home, venv, _python_bin(app_config, venv))
|
||
present = _read_env_keys(env_path)
|
||
missing = [k for k in defaults if k not in present]
|
||
|
||
enabled = False
|
||
if unit_path.is_file():
|
||
proc = subprocess.run(
|
||
["systemctl", "is-enabled", f"{unit}.service"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=15,
|
||
)
|
||
enabled = proc.returncode == 0 and "enabled" in (proc.stdout or "")
|
||
|
||
active: Optional[bool] = None
|
||
active_state: Optional[str] = None
|
||
status_tail: Optional[str] = None
|
||
if unit_path.is_file():
|
||
try:
|
||
proc = subprocess.run(
|
||
["systemctl", "is-active", f"{unit}.service"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=15,
|
||
)
|
||
active_state = (proc.stdout or proc.stderr or "").strip() or None
|
||
if active_state is not None:
|
||
active = active_state == "active"
|
||
except Exception:
|
||
pass
|
||
|
||
# Короткий хвост status — чтобы понять причину "не стартует" прямо из админки.
|
||
try:
|
||
proc = subprocess.run(
|
||
["systemctl", "status", "--no-pager", "-n", "20", f"{unit}.service"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=20,
|
||
)
|
||
if proc.stdout:
|
||
status_tail = proc.stdout.strip()[-4000:]
|
||
except Exception:
|
||
pass
|
||
|
||
ready = (
|
||
env_path.is_file()
|
||
and not missing
|
||
and unit_path.is_file()
|
||
and enabled
|
||
)
|
||
return {
|
||
"wesp_home": str(home),
|
||
"venv_dir": str(venv),
|
||
"env_path": str(env_path),
|
||
"env_exists": env_path.is_file(),
|
||
"env_keys_missing": missing,
|
||
"systemd_unit": unit,
|
||
"systemd_unit_path": str(unit_path),
|
||
"systemd_unit_exists": unit_path.is_file(),
|
||
"systemd_unit_enabled": enabled,
|
||
"systemd_unit_active": active,
|
||
"systemd_unit_active_state": active_state,
|
||
"systemd_unit_status_tail": status_tail,
|
||
"running_as_root": _is_root(),
|
||
"platform_ready": ready,
|
||
"message": (
|
||
"Настройка Pi применена"
|
||
if ready
|
||
else "Нажмите «Применить» для wesp.env и systemd"
|
||
),
|
||
}
|
||
|
||
|
||
def apply_pi_platform_setup(
|
||
app_config: Dict[str, Any],
|
||
*,
|
||
restart_service: bool = True,
|
||
) -> Dict[str, Any]:
|
||
if not _is_root():
|
||
return {
|
||
"ok": False,
|
||
"code": "not_root",
|
||
"message": (
|
||
"Нужен запуск WESP от root (или systemd User=root), "
|
||
"иначе нельзя записать /etc/wesp и systemd."
|
||
),
|
||
}
|
||
|
||
home = _wesp_home(app_config)
|
||
venv = _venv_dir(app_config)
|
||
py = _python_bin(app_config, venv)
|
||
env_path = _env_file_path(app_config)
|
||
defaults = _env_defaults(app_config, home, venv, py)
|
||
|
||
added, _ = _merge_env_file(env_path, defaults, home=home, venv=venv)
|
||
if _ensure_secret_key_in_env(env_path):
|
||
added.append("WESP_SECRET_KEY")
|
||
unit_ok, unit_msg, unit_path = _install_systemd_unit(app_config, home, venv, py)
|
||
if not unit_ok:
|
||
return {
|
||
"ok": False,
|
||
"code": "systemd_error",
|
||
"message": unit_msg,
|
||
"env_path": str(env_path),
|
||
"env_keys_added": added,
|
||
}
|
||
|
||
restart: Dict[str, Any] = {"scheduled": False}
|
||
if restart_service:
|
||
restart = schedule_service_restart(app_config, delay_sec=2.0)
|
||
st = get_pi_platform_setup_status(app_config)
|
||
|
||
msg = f"Готово: {env_path}, {unit_path.name}."
|
||
if restart.get("scheduled"):
|
||
msg = f"{msg} {restart.get('message', 'Перезапуск службы…')}"
|
||
|
||
return {
|
||
"ok": True,
|
||
"code": "applied",
|
||
"message": msg,
|
||
"env_path": str(env_path),
|
||
"env_keys_added": added,
|
||
"systemd_unit_path": str(unit_path),
|
||
"service_restart_scheduled": bool(restart.get("scheduled")),
|
||
"platform_ready": st.get("platform_ready"),
|
||
}
|