159 lines
8.1 KiB
Python
159 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Per-user installation backend used by setup.sh, with safe path quoting."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
|
|
def xdg(variable, fallback):
|
|
path = Path(os.environ.get(variable) or Path.home() / fallback)
|
|
if not path.is_absolute() or path == Path('/') or any(ord(c) < 32 for c in str(path)):
|
|
raise ValueError('Ungültiger XDG-Pfad: ' + str(path))
|
|
return path
|
|
|
|
|
|
def systemd_arg(value):
|
|
return '"' + str(value).replace('\\', '\\\\').replace('"', '\\"').replace('%', '%%').replace('$', '$$') + '"'
|
|
|
|
|
|
def desktop_arg(value):
|
|
# Desktop Entry escaping has both a value layer and an Exec argument layer.
|
|
value = str(value).replace('%', '%%')
|
|
for char in ('\\', '"', '`', '$'):
|
|
value = value.replace(char, '\\' + char)
|
|
return '"' + value.replace('\\', '\\\\') + '"'
|
|
|
|
|
|
def atomic_write(path, content, mode=0o644):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=path.parent, mode='w', delete=False) as f:
|
|
tmp = Path(f.name)
|
|
try:
|
|
f.write(content)
|
|
f.flush()
|
|
os.fchmod(f.fileno(), mode)
|
|
tmp.replace(path)
|
|
finally:
|
|
tmp.unlink(missing_ok=True)
|
|
|
|
|
|
def install(source, autostart, start, open_browser, lan=None):
|
|
data = xdg('XDG_DATA_HOME', '.local/share')
|
|
config_root = xdg('XDG_CONFIG_HOME', '.config')
|
|
state = xdg('XDG_STATE_HOME', '.local/state')
|
|
target = data / 'voicepult' / 'app'
|
|
config = config_root / 'voicepult' / 'config.json'
|
|
log = state / 'voicepult' / 'server.log'
|
|
unit = config_root / 'systemd' / 'user' / 'voicepult.service'
|
|
startup = config_root / 'autostart' / 'voicepult.desktop'
|
|
python = Path(sys.executable).resolve()
|
|
if lan is None:
|
|
previous = target / 'install.json'
|
|
lan = json.loads(previous.read_text()).get('lan', False) if previous.exists() else False
|
|
# Check everything before writing destination files.
|
|
files = ['lan_access.py', 'audio_meter.py', 'server.py', 'start.sh', 'web/index.html', 'web/app.js', 'web/style.css', 'web/icon.svg']
|
|
for name in files + ['scripts/launcher.py']:
|
|
if not (source / name).is_file():
|
|
raise ValueError('Projektdatei fehlt: ' + name)
|
|
have_systemd = bool(shutil.which('systemctl')) and subprocess.run(
|
|
['systemctl', '--user', 'show-environment'], capture_output=True).returncode == 0
|
|
was_active = have_systemd and subprocess.run(
|
|
['systemctl', '--user', 'is-active', '--quiet', 'voicepult.service'], capture_output=True).returncode == 0
|
|
# Files are replaced atomically; local configuration is deliberately not copied
|
|
# from the download, which may have come from a different computer.
|
|
for name in files:
|
|
atomic_write(target / name, (source / name).read_text(), 0o755 if name.endswith('.sh') else 0o644)
|
|
installed_start = ('#!/usr/bin/env bash\nset -euo pipefail\nexec '
|
|
+ ' '.join(shlex.quote(str(arg)) for arg in
|
|
([python, target / 'server.py', '--config', config] + (['--lan'] if lan else [])))
|
|
+ ' "$@"\n')
|
|
atomic_write(target / 'start.sh', installed_start, 0o755)
|
|
atomic_write(target / 'launcher.py', (source / 'scripts/launcher.py').read_text(), 0o755)
|
|
config.parent.mkdir(parents=True, exist_ok=True)
|
|
log.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_write(target / 'install.json', json.dumps({'config': str(config), 'log': str(log), 'systemd': have_systemd, 'lan': lan}, indent=2))
|
|
# Icon names avoid path quoting issues in desktop entry Icon values.
|
|
atomic_write(data / 'icons/hicolor/scalable/apps/voicepult.svg', (source / 'web/icon.svg').read_text())
|
|
exec_line = f'{desktop_arg(python)} {desktop_arg(target / "launcher.py")}'
|
|
desktop = ('[Desktop Entry]\nType=Application\nVersion=1.0\nName=VoicePult\n'
|
|
'Comment=Anwendungs-Audio, Kopfhörer und virtuelles Mikrofon verbinden\n'
|
|
f'Exec={exec_line}\nIcon=voicepult\nTerminal=false\nCategories=AudioVideo;Audio;Mixer;\n')
|
|
atomic_write(data / 'applications' / 'voicepult.desktop', desktop)
|
|
service = ('[Unit]\nDescription=VoicePult Audio-Patchpanel\n'
|
|
'After=pipewire-pulse.service pulseaudio.service\n\n[Service]\nType=simple\n'
|
|
f'ExecStart={systemd_arg(python)} {systemd_arg(target / "launcher.py")} --serve\n'
|
|
'Restart=on-failure\nRestartSec=5\nTimeoutStopSec=30\n'
|
|
'Environment=PYTHONUNBUFFERED=1\n\n[Install]\nWantedBy=default.target\n')
|
|
atomic_write(unit, service)
|
|
if have_systemd:
|
|
subprocess.run(['systemctl', '--user', 'daemon-reload'], check=True)
|
|
subprocess.run(['systemctl', '--user', 'enable' if autostart else 'disable', 'voicepult.service'], check=True)
|
|
# Remove only a previous VoicePult fallback entry, never an unrelated file.
|
|
if startup.exists() and 'X-VoicePult-Managed=true' in startup.read_text():
|
|
startup.unlink()
|
|
elif autostart:
|
|
atomic_write(startup, desktop.replace(f'Exec={exec_line}', f'Exec={exec_line} --background') + 'X-VoicePult-Managed=true\n')
|
|
elif startup.exists() and 'X-VoicePult-Managed=true' in startup.read_text():
|
|
startup.unlink()
|
|
if shutil.which('update-desktop-database'):
|
|
subprocess.run(['update-desktop-database', str(data / 'applications')], check=False)
|
|
if not start:
|
|
print('Dateien installiert. Es wurden keine Dienste gestartet. Start über den Menüeintrag VoicePult.')
|
|
return
|
|
if have_systemd and was_active:
|
|
subprocess.run(['systemctl', '--user', 'restart', 'voicepult.service'], check=True)
|
|
launcher = [str(python), str(target / 'launcher.py'), '--background']
|
|
subprocess.run(launcher, check=True)
|
|
# Confirm actual audio readiness, not just the HTTP server.
|
|
sys.path.insert(0, str(target))
|
|
from launcher import status
|
|
current = None
|
|
for _ in range(30):
|
|
current = status()
|
|
if current and current.get('online') and current.get('mic_available'):
|
|
break
|
|
time.sleep(.5)
|
|
else:
|
|
raise RuntimeError('Dateien installiert, aber Audio nicht bereit: ' + str((current or {}).get('error', 'Server nicht erreichbar')))
|
|
if have_systemd and not was_active:
|
|
active_now = subprocess.run(['systemctl', '--user', 'is-active', '--quiet', 'voicepult.service'], capture_output=True).returncode == 0
|
|
if not active_now:
|
|
print('Hinweis: Eine zuvor manuell gestartete VoicePult-Instanz läuft noch. Diese beenden und VoicePult über den Menüeintrag starten, um die installierte Version zu nutzen.')
|
|
print('Audioprüfung erfolgreich: VoicePult Mikrofon ist als Eingabegerät bereit.')
|
|
if open_browser:
|
|
if not (os.environ.get('DISPLAY') or os.environ.get('WAYLAND_DISPLAY')):
|
|
print('Kein grafischer Browser erreichbar. Öffne später http://127.0.0.1:8765 im Browser.')
|
|
else:
|
|
result = subprocess.run([str(python), str(target / 'launcher.py')], check=False)
|
|
if result.returncode:
|
|
print('Browser konnte nicht automatisch geöffnet werden. URL: http://127.0.0.1:8765')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--source', type=Path, required=True)
|
|
parser.add_argument('--autostart', type=int, choices=[0, 1], required=True)
|
|
parser.add_argument('--start', type=int, choices=[0, 1], required=True)
|
|
parser.add_argument('--open', type=int, choices=[0, 1], required=True)
|
|
parser.add_argument('--lan', action='store_const', const=True, default=None)
|
|
args = parser.parse_args()
|
|
if os.geteuid() == 0:
|
|
parser.error('Bitte als normalen Desktop-Benutzer ausführen.')
|
|
try:
|
|
install(args.source.resolve(), bool(args.autostart), bool(args.start), bool(args.open), args.lan)
|
|
except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as exc:
|
|
print('VoicePult: ' + str(exc), file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|