feat: GET /audio/devices, POST /audio/combined — PipeWire source management

api/router.py

@@ -212,6 +212,67 @@ async def open_file(body: dict, user: dict = Depends(current_user)):
     return {"ok": True}
 
 
+@router.get("/audio/devices")
+async def list_audio_devices(user: dict = Depends(current_user)):
+    import subprocess
+    if not user.get("is_admin"):
+        raise HTTPException(status_code=403, detail="Nur Administratoren")
+    try:
+        out = subprocess.check_output(
+            ["pactl", "list", "sources", "short"],
+            stderr=subprocess.DEVNULL, timeout=5,
+        ).decode()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"pactl fehlgeschlagen: {e}")
+    devices = []
+    for line in out.strip().splitlines():
+        parts = line.split("\t")
+        if len(parts) >= 2:
+            devices.append({
+                "index": parts[0],
+                "name": parts[1],
+                "state": parts[4] if len(parts) > 4 else "",
+            })
+    return devices
+
+
+@router.post("/audio/combined")
+async def create_combined_source(body: dict, user: dict = Depends(current_user)):
+    import subprocess, json, pathlib
+    if not user.get("is_admin"):
+        raise HTTPException(status_code=403, detail="Nur Administratoren")
+    mic = body.get("mic", "")
+    monitor = body.get("monitor", "")
+    if not mic or not monitor:
+        raise HTTPException(status_code=400, detail="mic und monitor erforderlich")
+    # Validate: names must come from pactl list — no shell injection via user input
+    out = subprocess.check_output(
+        ["pactl", "list", "sources", "short"], stderr=subprocess.DEVNULL, timeout=5
+    ).decode()
+    known = {line.split("\t")[1] for line in out.strip().splitlines() if "\t" in line}
+    if mic not in known or monitor not in known:
+        raise HTTPException(status_code=400, detail="Unbekanntes Audio-Device")
+    sink_id = subprocess.check_output([
+        "pactl", "load-module", "module-null-sink",
+        "sink_name=transkriptor-combined",
+        "sink_properties=device.description=Transkriptor Combined",
+    ], timeout=5).decode().strip()
+    mic_id = subprocess.check_output([
+        "pactl", "load-module", "module-loopback",
+        f"source={mic}", "sink=transkriptor-combined",
+    ], timeout=5).decode().strip()
+    mon_id = subprocess.check_output([
+        "pactl", "load-module", "module-loopback",
+        f"source={monitor}", "sink=transkriptor-combined",
+    ], timeout=5).decode().strip()
+    state_path = pathlib.Path(
+        os.path.expanduser("~/.config/tueit-transcriber/pipewire-modules.json")
+    )
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+    state_path.write_text(json.dumps({"ids": [int(sink_id), int(mic_id), int(mon_id)]}))
+    return {"device": "transkriptor-combined.monitor", "module_ids": [sink_id, mic_id, mon_id]}
+
+
 @router.websocket("/ws")
 async def websocket_endpoint(ws: WebSocket):
     from auth import get_user_for_token

tests/test_api.py

@@ -97,3 +97,36 @@ def test_login_rejects_wrong_credentials():
         with patch("auth.USERS_PATH", users_path):
             r = client.post("/login", json={"username": "nobody", "password": "wrong"})
             assert r.status_code == 401
+
+
+def test_audio_devices_returns_list(monkeypatch):
+    import subprocess
+    from main import app
+    from api.router import current_user
+    pactl_output = (
+        "1\talsa_input.pci.analog-stereo\tPipeWire\ts32le 2ch 48000Hz\tRUNNING\n"
+        "2\talsa_output.pci.analog-stereo.monitor\tPipeWire\ts32le 2ch 48000Hz\tIDLE\n"
+    )
+    monkeypatch.setattr(subprocess, "check_output", lambda *a, **kw: pactl_output.encode())
+    app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": True}
+    try:
+        client = TestClient(app)
+        r = client.get("/audio/devices", headers={"Authorization": "Bearer fake"})
+        assert r.status_code == 200
+        devices = r.json()
+        assert len(devices) == 2
+        assert devices[0]["name"] == "alsa_input.pci.analog-stereo"
+    finally:
+        app.dependency_overrides.pop(current_user, None)
+
+
+def test_audio_devices_forbidden_for_non_admin():
+    from main import app
+    from api.router import current_user
+    app.dependency_overrides[current_user] = lambda: {"username": "u", "output_dir": "/tmp", "is_admin": False}
+    try:
+        client = TestClient(app)
+        r = client.get("/audio/devices", headers={"Authorization": "Bearer fake"})
+        assert r.status_code == 403
+    finally:
+        app.dependency_overrides.pop(current_user, None)