import asyncio import json import time import socket import uuid from urllib.parse import urlencode from xml.etree import ElementTree as ET import requests import websockets VMIX_API = "http://127.0.0.1:8088/api" WS_BASE = "wss://wfl.tvstart.ru/ws/vmix-client" # WS_BASE = "ws://127.0.0.1:8000/ws/vmix-client" CLIENT_ID = f"vmix-{socket.gethostname().lower()}-{uuid.uuid4().hex[:6]}" POLL_INTERVAL = 3 REQUEST_TIMEOUT = 3 RECONNECT_DELAY = 5 def read_vmix_dynamic_values(): try: resp = requests.get(VMIX_API, timeout=REQUEST_TIMEOUT) resp.raise_for_status() xml_text = resp.text root = ET.fromstring(xml_text) def get_text(path: str) -> str | None: node = root.find(path) if node is None or node.text is None: return None value = node.text.strip() return value or None session_token = get_text("./dynamic/value1") match_id = get_text("./dynamic/value2") group_name = get_text("./dynamic/value3") operator_login = get_text("./dynamic/value4") return { "ok": True, "session_token": session_token, "match_id": int(match_id) if match_id and match_id.isdigit() else None, "group_name": group_name, "operator_login": operator_login, } except Exception as e: return { "ok": False, "error": str(e), "session_token": None, "match_id": None, "group_name": None, "operator_login": None, } def execute_vmix_command(path: str): path = str(path or "").strip() if not path.startswith("/api/"): return {"ok": False, "error": "invalid command", "path": path} url = f"http://127.0.0.1:8088{path}" try: resp = requests.get(url, timeout=REQUEST_TIMEOUT) return { "ok": resp.ok, "status_code": resp.status_code, "path": path, "response": resp.text[:300], } except Exception as e: return { "ok": False, "path": path, "error": str(e), } async def wait_for_routing(): while True: data = read_vmix_dynamic_values() if data["ok"] and data["match_id"] is not None and data["operator_login"]: return data print("[agent] waiting for match_id + operator_login...", data) await asyncio.sleep(POLL_INTERVAL) def build_ws_url(match_id: int, group_name: str | None, operator_login: str): params = { "client_id": CLIENT_ID, "match_id": match_id, "operator_name": operator_login, } if group_name: params["group_name"] = group_name return f"{WS_BASE}?{urlencode(params)}" async def ping_loop(ws): while True: await asyncio.sleep(15) await ws.send(json.dumps({"type": "ping"})) async def run_agent(): current_match_id = None current_group_name = None current_operator_login = None while True: vmix_data = await wait_for_routing() match_id = vmix_data["match_id"] group_name = vmix_data["group_name"] operator_login = vmix_data["operator_login"] if current_match_id != match_id or current_operator_login != operator_login or current_group_name != group_name: print( f"[agent] routing: match_id={match_id}, operator_login={operator_login}, group_name={group_name}" ) current_match_id = match_id current_group_name = group_name current_operator_login = operator_login ws_url = build_ws_url(match_id, group_name, operator_login) try: async with websockets.connect(ws_url, ping_interval=None, max_size=2**20) as ws: print("[agent] connected to server") await ws.send(json.dumps({ "type": "register", "client_id": CLIENT_ID, "match_id": match_id, "group_name": group_name, "operator_name": operator_login, })) pinger = asyncio.create_task(ping_loop(ws)) try: while True: latest = read_vmix_dynamic_values() if latest["ok"]: latest_match_id = latest.get("match_id") latest_group_name = latest.get("group_name") latest_operator_login = latest.get("operator_login") if ( latest_match_id != match_id or latest_group_name != group_name or latest_operator_login != operator_login ): print("[agent] routing meta changed in vMix, reconnecting") break try: raw = await asyncio.wait_for(ws.recv(), timeout=3) except asyncio.TimeoutError: continue msg = json.loads(raw) if msg.get("type") == "vmix_command": print("[agent] got commands:", msg.get("commands", [])) results = [execute_vmix_command(cmd) for cmd in msg.get("commands", [])] print("[agent] results:", results) await ws.send(json.dumps({ "type": "vmix_result", "client_id": CLIENT_ID, "match_id": match_id, "operator_name": operator_login, "results": results, "sent_at": time.time(), })) finally: pinger.cancel() except Exception as e: print("[agent] websocket error:", e) await asyncio.sleep(RECONNECT_DELAY) if __name__ == "__main__": asyncio.run(run_agent())