import asyncio import json import time import socket import uuid from urllib.parse import urlencode from xml.etree import ElementTree as ET import os import sys import ctypes 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 resource_path(relative_path: str) -> str: try: base_path = sys._MEIPASS except Exception: base_path = os.path.abspath(".") return os.path.join(base_path, relative_path) def setup_green_console(): if os.name != "nt": return kernel32 = ctypes.windll.kernel32 h_out = kernel32.GetStdHandle(-11) # STD_OUTPUT_HANDLE mode = ctypes.c_uint() if kernel32.GetConsoleMode(h_out, ctypes.byref(mode)): kernel32.SetConsoleMode(h_out, mode.value | 0x0004) # ENABLE_VIRTUAL_TERMINAL_PROCESSING # зелёный текст по умолчанию print("\033[92m", end="") def set_console_icon(): if os.name != "nt": return ico_path = resource_path("app.ico") if not os.path.exists(ico_path): return user32 = ctypes.windll.user32 # AppUserModelID помогает Windows корректнее показывать иконку приложения ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("my.agent.console") hwnd = kernel32_get_console_window() if not hwnd: return IMAGE_ICON = 1 LR_LOADFROMFILE = 0x00000010 WM_SETICON = 0x0080 ICON_SMALL = 0 ICON_BIG = 1 hicon = user32.LoadImageW( None, ico_path, IMAGE_ICON, 0, 0, LR_LOADFROMFILE ) if hicon: user32.SendMessageW(hwnd, WM_SETICON, ICON_SMALL, hicon) user32.SendMessageW(hwnd, WM_SETICON, ICON_BIG, hicon) def kernel32_get_console_window(): try: return ctypes.windll.kernel32.GetConsoleWindow() except Exception: return 0 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__": setup_green_console() set_console_icon() asyncio.run(run_agent())