Files
hockey_new/hockey_data/agent_bridge.py

3220 lines
156 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
import asyncio
import hashlib
import hmac
import json
import re
import secrets
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Callable
from fastapi import APIRouter, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import and_, delete, desc, select
from .auth_bridge import HockeyUser
from .database import HockeyDatabase
from .mapping_context import MappingDataService
from .models import (MappingSqlDataSource, OperatorSession, UserPreference, VmixAssignment, VmixDevice, VmixMappingProfile, VmixMappingField)
AGENT_PROTOCOL_VERSION = 1
AGENT_ONLINE_WINDOW_SECONDS = 35
_DEVICE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{6,128}$")
def _utcnow() -> datetime:
return datetime.utcnow()
def _secret_hash(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
_MAPPING_TEST_FUNCTIONS = {
"SetText", "SetImage", "SetColor",
"SetTextColour",
"SetTextVisibleOn", "SetTextVisibleOff",
"SetImageVisibleOn", "SetImageVisibleOff",
}
def _mapping_value_function(field_type: Any) -> str:
kind = str(field_type or "text").strip().lower()
if kind in {"image", "source"}:
return "SetImage"
if kind in {"color", "colour"}:
return "SetColor"
return "SetText"
def _mapping_rule_payload(value: Any) -> dict[str, Any]:
if isinstance(value, dict):
return dict(value)
try:
parsed = json.loads(str(value or "{}"))
except Exception:
return {}
return parsed if isinstance(parsed, dict) else {}
def _stable_vmix_inventory_fingerprint(inputs: list[dict[str, Any]]) -> str:
"""Fingerprint a vMix project without positional Input numbers/order.
vMix ``number`` is a slot position and changes when an operator moves an
Input. ``key`` is the durable identity, with title as fallback for old
inventories. Field indices are positional too, so only names/types are part
of the identity.
"""
stable_inputs: list[dict[str, Any]] = []
for item in inputs:
if not isinstance(item, dict):
continue
key = str(item.get("key") or "").strip()
title = str(item.get("title") or "").strip()
identity = key or title
if not identity:
# Last-resort legacy inventory. This is intentionally weaker, but
# avoids dropping projects produced by very old Agents.
identity = f"legacy:{str(item.get('number') or '').strip()}"
fields = item.get("fields") if isinstance(item.get("fields"), list) else []
stable_fields = sorted(
(
{
"name": str(field.get("name") or "").strip(),
"type": str(field.get("type") or "text").strip().lower(),
}
for field in fields
if isinstance(field, dict) and str(field.get("name") or "").strip()
),
key=lambda value: (value["name"].casefold(), value["type"]),
)
stable_inputs.append({
"identity": identity,
"key": key,
# A title rename must not create a new project when vMix supplied a
# durable key. For legacy inputs without key the title is identity.
"title": "" if key else title,
"type": str(item.get("type") or "").strip().lower(),
"fields": stable_fields,
})
stable_inputs.sort(key=lambda value: (value["identity"].casefold(), value["title"].casefold()))
payload = json.dumps(stable_inputs, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _stored_inventory_stable_fingerprint(value: Any) -> str:
try:
payload = json.loads(str(value or "{}")) if not isinstance(value, dict) else value
except Exception:
return ""
inputs = payload.get("inputs") if isinstance(payload, dict) and isinstance(payload.get("inputs"), list) else []
return _stable_vmix_inventory_fingerprint(inputs) if inputs else ""
def _mapping_numeric_value(value: Any) -> float | None:
if value is None or isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value)
text_value = str(value).strip()
if not text_value:
return None
compact = text_value.replace(" ", "").replace("%", "")
if re.fullmatch(r"[-+]?\d{1,4}:\d{1,2}(?::\d{1,2}(?:[.,]\d+)?)?", compact):
parts = compact.replace(",", ".").split(":")
try:
numbers = [float(part) for part in parts]
except ValueError:
return None
if len(numbers) == 2:
return numbers[0] * 60 + numbers[1]
if len(numbers) == 3:
return numbers[0] * 3600 + numbers[1] * 60 + numbers[2]
try:
return float(compact.replace(",", "."))
except ValueError:
return None
def _mapping_rule_matches(left: Any, operator: Any, right: Any = None) -> bool:
op = str(operator or "eq").strip().lower()
left_text = "" if left is None else str(left).strip()
right_text = "" if right is None else str(right).strip()
if op in {"empty", "is_empty"}:
return left_text == ""
if op in {"not_empty", "is_not_empty"}:
return left_text != ""
if op in {"contains", "not_contains"}:
matched = right_text.casefold() in left_text.casefold()
return (not matched) if op == "not_contains" else matched
left_number = _mapping_numeric_value(left)
right_number = _mapping_numeric_value(right)
if left_number is not None and right_number is not None:
a: Any = left_number
b: Any = right_number
else:
a = left_text.casefold()
b = right_text.casefold()
if op in {"gt", ">"}: return a > b
if op in {"lt", "<"}: return a < b
if op in {"gte", ">="}: return a >= b
if op in {"lte", "<="}: return a <= b
if op in {"neq", "!=", "<>"}: return a != b
return a == b
def _mapping_rule_command(
rule: dict[str, Any],
*,
field_type: Any,
input_ref: str,
selected_name: str,
default_data_key: str,
by_key: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any] | None, str]:
if not rule or not bool(rule.get("enabled")):
return None, ""
kind = str(field_type or "text").strip().lower()
action = str(rule.get("action") or ("visibility" if kind in {"image", "source"} else "text_colour")).strip().lower()
if action == "text_colour" and kind != "text":
return None, "unsupported_action"
if action == "visibility" and kind not in {"text", "image", "source"}:
return None, "unsupported_action"
left_key = str(rule.get("left_key") or default_data_key or "").strip()
left_item = by_key.get(left_key)
if left_item is None:
return None, f"missing_left:{left_key}"
left_value = left_item.get("value")
operator = str(rule.get("operator") or "gt").strip().lower()
right_mode = str(rule.get("right_mode") or "field").strip().lower()
right_value: Any = None
if operator not in {"empty", "is_empty", "not_empty", "is_not_empty"}:
if right_mode == "value":
right_value = rule.get("right_value", "")
else:
right_key = str(rule.get("right_key") or "").strip()
right_item = by_key.get(right_key)
if right_item is None:
return None, f"missing_right:{right_key}"
right_value = right_item.get("value")
matched = _mapping_rule_matches(left_value, operator, right_value)
if action == "text_colour":
chosen = rule.get("true_value", "#E5CEA8") if matched else rule.get("false_value", "#FFFFFF")
if chosen is None or str(chosen).strip() == "":
return None, ""
return {
"Function": "SetTextColour",
"Input": input_ref,
"SelectedName": selected_name,
"Value": str(chosen).strip(),
}, ""
raw_state = rule.get("true_value", "on") if matched else rule.get("false_value", "off")
visible = str(raw_state).strip().lower() not in {"0", "false", "off", "hide", "hidden", "no"}
if kind in {"image", "source"}:
function = "SetImageVisibleOn" if visible else "SetImageVisibleOff"
else:
function = "SetTextVisibleOn" if visible else "SetTextVisibleOff"
return {"Function": function, "Input": input_ref, "SelectedName": selected_name}, ""
@dataclass(slots=True)
class LiveAgent:
websocket: WebSocket
connected_at: datetime
class VmixAgentHub:
"""Persistent device registry + live WebSocket routing for vMix agents."""
def __init__(self, database: HockeyDatabase, settings: Any | None = None) -> None:
self.database = database
self._live: dict[str, LiveAgent] = {}
self._lock = asyncio.Lock()
self._pending_commands: dict[str, tuple[str, asyncio.Future[dict[str, Any]]]] = {}
self.mapping_data = MappingDataService(database, settings=settings)
self._auto_refresh_task: asyncio.Task[None] | None = None
self._auto_refresh_next: dict[tuple[str, str], float] = {}
self._mapping_value_cache: dict[tuple[str, str, int, str, str, str, str], str] = {}
self._auto_refresh_last_error = ""
@staticmethod
def normalise_device_id(value: Any) -> str:
device_id = str(value or "").strip()
if not _DEVICE_ID_RE.fullmatch(device_id):
raise ValueError("Некорректный device_id")
return device_id
@staticmethod
def _mapping_source_code(data_key: Any) -> str:
key = str(data_key or "").strip()
return key.split(".", 1)[0] if key else ""
@staticmethod
def _agent_supports_batch(version: Any) -> bool:
match = re.search(r"(\d+)\.(\d+)\.(\d+)", str(version or ""))
if not match:
return False
return tuple(int(part) for part in match.groups()) >= (1, 4, 0)
@staticmethod
def _inventory_target_pairs(value: Any) -> set[tuple[str, str]]:
"""Return stable ``(Input key, field name)`` pairs from an inventory."""
try:
payload = json.loads(str(value or "{}")) if not isinstance(value, dict) else value
except Exception:
return set()
inputs = payload.get("inputs") if isinstance(payload, dict) and isinstance(payload.get("inputs"), list) else []
pairs: set[tuple[str, str]] = set()
for item in inputs:
if not isinstance(item, dict):
continue
input_key = str(item.get("key") or "").strip()
if not input_key:
continue
fields = item.get("fields") if isinstance(item.get("fields"), list) else []
for field in fields:
if not isinstance(field, dict):
continue
field_name = str(field.get("name") or "").strip()
if field_name:
pairs.add((input_key, field_name))
return pairs
def _compatible_mapping_profile_for_device(self, session: Any, device: VmixDevice) -> VmixMappingProfile | None:
"""Recover a profile when the project changed only in unrelated Inputs.
A profile is considered compatible only when *all* enabled mapped targets
identified by stable vMix Input keys still exist in the current project. If
several profiles tie for the best coverage we deliberately do not guess.
"""
current_pairs = self._inventory_target_pairs(device.project_inventory_json)
if not current_pairs:
return None
candidates: list[tuple[int, VmixMappingProfile]] = []
for profile in session.scalars(
select(VmixMappingProfile)
.where(VmixMappingProfile.active.is_(True))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
):
rows = list(session.scalars(
select(VmixMappingField).where(and_(
VmixMappingField.profile_id == profile.id,
VmixMappingField.enabled.is_(True),
))
))
targets = {
(str(row.vmix_input_key or "").strip(), str(row.vmix_field or "").strip())
for row in rows
if str(row.vmix_input_key or "").strip() and str(row.vmix_field or "").strip()
}
if not targets or not targets.issubset(current_pairs):
continue
candidates.append((len(targets), profile))
if not candidates:
return None
best_score = max(score for score, _ in candidates)
best = [profile for score, profile in candidates if score == best_score]
return best[0] if len(best) == 1 else None
def _active_mapping_profile_for_device(self, session: Any, device: VmixDevice) -> VmixMappingProfile | None:
"""Find/migrate the mapping profile for a device using stable inventory identity."""
fingerprint = str(device.project_fingerprint or "").strip()
if fingerprint:
profile = session.scalar(
select(VmixMappingProfile)
.where(and_(VmixMappingProfile.project_fingerprint == fingerprint, VmixMappingProfile.active.is_(True)))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
)
if profile is not None:
return profile
try:
inventory = json.loads(device.project_inventory_json or "{}")
except Exception:
inventory = {}
stable_fingerprint = _stored_inventory_stable_fingerprint(inventory)
agent_fingerprint = str(inventory.get("agent_fingerprint") or "").strip().lower() if isinstance(inventory, dict) else ""
# Legacy build <=39 profile: raw Agent fingerprint was stored directly.
if agent_fingerprint:
profile = session.scalar(
select(VmixMappingProfile)
.where(and_(VmixMappingProfile.project_fingerprint == agent_fingerprint, VmixMappingProfile.active.is_(True)))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
)
if profile is not None:
if stable_fingerprint:
profile.project_fingerprint = stable_fingerprint
device.project_fingerprint = stable_fingerprint
return profile
if stable_fingerprint:
candidates = list(session.scalars(
select(VmixMappingProfile)
.where(VmixMappingProfile.active.is_(True))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
))
profile = next((candidate for candidate in candidates
if _stored_inventory_stable_fingerprint(candidate.inventory_json) == stable_fingerprint), None)
if profile is not None:
profile.project_fingerprint = stable_fingerprint
device.project_fingerprint = stable_fingerprint
return profile
# Final safe recovery: if the operator added/removed unrelated Inputs, the
# whole-project fingerprint changes even though every linked target is still
# present. Reuse a profile only when the target set identifies one profile
# unambiguously. Do not rewrite its primary fingerprint here, because another
# active Agent may legitimately use the original project variant.
return self._compatible_mapping_profile_for_device(session, device)
def start_auto_refresh(self) -> None:
"""Start the server-side SQL → Mapping → vMix refresh scheduler."""
if self._auto_refresh_task is None or self._auto_refresh_task.done():
self._auto_refresh_task = asyncio.create_task(self._auto_refresh_loop(), name="hockey-mapping-auto-refresh")
async def stop_auto_refresh(self) -> None:
task = self._auto_refresh_task
self._auto_refresh_task = None
if task is None:
return
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
async def _auto_refresh_loop(self) -> None:
while True:
try:
await self.auto_refresh_once()
self._auto_refresh_last_error = ""
except asyncio.CancelledError:
raise
except Exception as error:
# The scheduler must never take the HTTP/WebSocket server down. A broken
# source is isolated to its next tick and can be fixed from the admin UI.
self._auto_refresh_last_error = str(error)[:500]
await asyncio.sleep(0.25)
async def auto_refresh_once(self) -> dict[str, Any]:
"""Apply due auto-refresh SQL sources to live devices, sending only changed values."""
async with self._lock:
live_ids = set(self._live.keys())
if not live_ids:
return {"ok": True, "devices": 0, "sources": 0, "applied": 0}
with self.database.session() as session:
source_rows = list(session.scalars(
select(MappingSqlDataSource).where(and_(
MappingSqlDataSource.enabled.is_(True),
MappingSqlDataSource.auto_refresh_enabled.is_(True),
))
))
source_intervals = {
str(row.code): max(1.0, min(3600.0, float(int(row.refresh_interval_ms or 1000)) / 1000.0))
for row in source_rows
}
if not source_intervals:
return {"ok": True, "devices": 0, "sources": 0, "applied": 0}
devices = list(session.scalars(
select(VmixDevice).where(and_(
VmixDevice.device_uuid.in_(live_ids),
VmixDevice.is_active_for_account.is_(True),
VmixDevice.vmix_connected.is_(True),
))
))
jobs: list[tuple[str, set[str]]] = []
loop_now = asyncio.get_running_loop().time()
for device in devices:
if not device.current_assignment_key or not device.current_match_external_id or not device.project_fingerprint:
continue
profile = self._active_mapping_profile_for_device(session, device)
if profile is None:
continue
data_keys = list(session.scalars(
select(VmixMappingField.data_key).where(and_(
VmixMappingField.profile_id == profile.id,
VmixMappingField.enabled.is_(True),
))
))
used_codes = {self._mapping_source_code(key) for key in data_keys}
due: set[str] = set()
for code in used_codes.intersection(source_intervals):
schedule_key = (str(device.device_uuid), code)
if loop_now + 1e-9 < self._auto_refresh_next.get(schedule_key, 0.0):
continue
due.add(code)
self._auto_refresh_next[schedule_key] = loop_now + source_intervals[code]
if due:
jobs.append((str(device.device_uuid), due))
if not jobs:
return {"ok": True, "devices": 0, "sources": 0, "applied": 0}
raw_results = await asyncio.gather(*(
self.apply_mapping_to_device(
device_id, reason="sql_auto_refresh", source_codes=source_codes, only_changed=True
)
for device_id, source_codes in jobs
), return_exceptions=True)
applied = 0
errors: list[str] = []
for value in raw_results:
if isinstance(value, Exception):
errors.append(str(value)[:300])
elif isinstance(value, dict):
applied += int(value.get("applied") or 0)
errors.extend(str(item)[:300] for item in (value.get("errors") or []))
return {
"ok": not errors,
"devices": len(jobs),
"sources": sum(len(codes) for _, codes in jobs),
"applied": applied,
"errors": errors,
}
async def register(self, websocket: WebSocket, hello: dict[str, Any]) -> dict[str, Any]:
device_id = self.normalise_device_id(hello.get("device_id"))
device_secret = str(hello.get("device_secret") or "").strip()
if len(device_secret) < 24:
raise PermissionError("device_secret слишком короткий")
now = _utcnow()
secret_hash = _secret_hash(device_secret)
vmix = hello.get("vmix") if isinstance(hello.get("vmix"), dict) else {}
client = websocket.client
client_ip = client.host if client else ""
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None:
row = VmixDevice(
device_uuid=device_id,
device_secret_hash=secret_hash,
name=str(hello.get("device_name") or hello.get("hostname") or device_id)[:200],
hostname=str(hello.get("hostname") or "")[:200],
agent_version=str(hello.get("agent_version") or "")[:64],
first_seen_at=now,
last_seen_at=now,
last_ip=client_ip[:128],
vmix_connected=bool(vmix.get("connected")),
vmix_version=str(vmix.get("version") or "")[:64],
vmix_url=str(vmix.get("url") or "")[:500],
)
session.add(row)
session.flush()
elif not hmac.compare_digest(row.device_secret_hash, secret_hash):
raise PermissionError("Этот device_id уже зарегистрирован с другим ключом")
else:
row.name = str(hello.get("device_name") or row.name or row.hostname or device_id)[:200]
row.hostname = str(hello.get("hostname") or row.hostname or "")[:200]
row.agent_version = str(hello.get("agent_version") or row.agent_version or "")[:64]
row.last_seen_at = now
row.last_ip = client_ip[:128]
row.vmix_connected = bool(vmix.get("connected"))
row.vmix_version = str(vmix.get("version") or row.vmix_version or "")[:64]
row.vmix_url = str(vmix.get("url") or row.vmix_url or "")[:500]
row.last_error = ""
session.flush()
payload = self._device_payload(row, viewer_user_id=row.wfl_user_id or "")
async with self._lock:
previous = self._live.get(device_id)
self._live[device_id] = LiveAgent(websocket=websocket, connected_at=now)
if previous is not None and previous.websocket is not websocket:
try:
await previous.websocket.close(code=4001, reason="Device reconnected")
except Exception:
pass
return payload
async def unregister(self, device_id: str, websocket: WebSocket) -> None:
async with self._lock:
current = self._live.get(device_id)
if current is not None and current.websocket is websocket:
self._live.pop(device_id, None)
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is not None:
row.last_seen_at = _utcnow()
async def receive_status(self, device_id: str, message: dict[str, Any]) -> None:
vmix = message.get("vmix") if isinstance(message.get("vmix"), dict) else {}
now = _utcnow()
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None:
return
row.last_seen_at = now
if "connected" in vmix:
row.vmix_connected = bool(vmix.get("connected"))
if vmix.get("version") is not None:
row.vmix_version = str(vmix.get("version") or "")[:64]
if vmix.get("url") is not None:
row.vmix_url = str(vmix.get("url") or "")[:500]
if message.get("error") is not None:
row.last_error = str(message.get("error") or "")[:1000]
async def send(self, device_id: str, payload: dict[str, Any]) -> bool:
async with self._lock:
live = self._live.get(device_id)
if live is None:
return False
try:
await live.websocket.send_json(payload)
return True
except Exception:
async with self._lock:
if self._live.get(device_id) is live:
self._live.pop(device_id, None)
return False
async def receive_command_ack(self, device_id: str, message: dict[str, Any]) -> None:
request_id = str(message.get("request_id") or "").strip()
if not request_id:
return
pending = self._pending_commands.get(request_id)
if pending is None:
return
expected_device_id, future = pending
if expected_device_id != device_id or future.done():
return
payload = dict(message)
payload["device_id"] = device_id
future.set_result(payload)
async def send_vmix_command(
self,
device_id: str,
*,
assignment_id: str,
match_id: str,
command: dict[str, Any],
timeout: float = 5.0,
) -> dict[str, Any]:
request_id = secrets.token_urlsafe(12)
loop = asyncio.get_running_loop()
future: asyncio.Future[dict[str, Any]] = loop.create_future()
self._pending_commands[request_id] = (device_id, future)
try:
delivered = await self.send(
device_id,
{
"type": "vmix.command",
"protocol": AGENT_PROTOCOL_VERSION,
"request_id": request_id,
"device_id": device_id,
"assignment_id": assignment_id,
"match_id": match_id,
"command": command,
},
)
if not delivered:
raise HTTPException(status_code=409, detail="Agent сейчас offline")
try:
ack = await asyncio.wait_for(future, timeout=timeout)
except asyncio.TimeoutError as error:
raise HTTPException(status_code=504, detail="Agent не подтвердил команду vMix") from error
return ack
finally:
pending = self._pending_commands.pop(request_id, None)
if pending is not None:
_, pending_future = pending
if not pending_future.done():
pending_future.cancel()
async def send_vmix_batch(
self,
device_id: str,
*,
assignment_id: str,
match_id: str,
commands: list[dict[str, Any]],
timeout: float = 8.0,
) -> dict[str, Any]:
"""Send many concrete vMix commands in one WebSocket frame and wait for one ACK."""
request_id = secrets.token_urlsafe(12)
loop = asyncio.get_running_loop()
future: asyncio.Future[dict[str, Any]] = loop.create_future()
self._pending_commands[request_id] = (device_id, future)
try:
delivered = await self.send(
device_id,
{
"type": "vmix.batch",
"protocol": AGENT_PROTOCOL_VERSION,
"request_id": request_id,
"device_id": device_id,
"assignment_id": assignment_id,
"match_id": match_id,
"commands": commands,
},
)
if not delivered:
raise HTTPException(status_code=409, detail="Agent сейчас offline")
try:
return await asyncio.wait_for(future, timeout=timeout)
except asyncio.TimeoutError as error:
raise HTTPException(status_code=504, detail="Agent не подтвердил пакет команд vMix") from error
finally:
pending = self._pending_commands.pop(request_id, None)
if pending is not None:
_, pending_future = pending
if not pending_future.done():
pending_future.cancel()
async def test_set_text(
self,
device_id: str,
user: HockeyUser,
*,
input_ref: str,
selected_name: str,
value: str,
) -> dict[str, Any]:
device_id = self.normalise_device_id(device_id)
input_ref = str(input_ref or "").strip()
selected_name = str(selected_name or "").strip()
if not input_ref:
raise HTTPException(status_code=422, detail="Укажите Input (имя, номер или key)")
if not selected_name:
raise HTTPException(status_code=422, detail="Укажите имя текстового поля vMix")
if len(input_ref) > 300 or len(selected_name) > 300 or len(value) > 4000:
raise HTTPException(status_code=422, detail="Слишком длинные параметры тестовой команды")
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or device.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
if not device.is_active_for_account:
raise HTTPException(status_code=409, detail="Включите для этого Agent режим «Получать данные»")
assignment_id = str(device.current_assignment_key or "")
match_id = str(device.current_match_external_id or "")
if not assignment_id or not match_id:
raise HTTPException(status_code=409, detail="Сначала выберите матч в web-интерфейсе")
ack = await self.send_vmix_command(
device_id,
assignment_id=assignment_id,
match_id=match_id,
command={
"Function": "SetText",
"Input": input_ref,
"SelectedName": selected_name,
"Value": value,
},
)
if not bool(ack.get("ok")):
reason = str(ack.get("reason") or ack.get("error") or "vMix отклонил команду")
raise HTTPException(status_code=502, detail=f"Команда дошла до agent, но не выполнена: {reason}")
return {
"ok": True,
"device_id": device_id,
"assignment_id": assignment_id,
"match_id": match_id,
"function": "SetText",
"input": input_ref,
"selected_name": selected_name,
"value": value,
"agent_ack": ack,
}
async def apply_mapping_to_device(
self,
device_id: str,
*,
reason: str = "manual",
source_codes: set[str] | None = None,
only_changed: bool = False,
extra_context: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Resolve the active mapping against the device's current match and push values to vMix.
Mapping configuration stays server-side. The Agent receives only concrete vMix commands,
so it never needs database access or knowledge of SQL/context keys.
"""
device_id = self.normalise_device_id(device_id)
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None:
return {"ok": False, "reason": "device_not_found", "device_id": device_id}
if not device.wfl_user_id:
return {"ok": False, "reason": "device_not_paired", "device_id": device_id}
assignment_id = str(device.current_assignment_key or "")
match_id = str(device.current_match_external_id or "")
fingerprint = str(device.project_fingerprint or "")
vmix_connected = bool(device.vmix_connected)
login = str(device.login_snapshot or "")
user_id = str(device.wfl_user_id or "")
agent_version = str(device.agent_version or "")
if not assignment_id or not match_id:
return {"ok": False, "reason": "no_match_assignment", "device_id": device_id}
if not fingerprint:
return {"ok": False, "reason": "no_project_inventory", "device_id": device_id}
if not vmix_connected:
return {"ok": False, "reason": "vmix_not_connected", "device_id": device_id}
assignment = session.scalar(
select(VmixAssignment).where(
and_(
VmixAssignment.device_id == device.id,
VmixAssignment.assignment_key == assignment_id,
VmixAssignment.active.is_(True),
)
)
)
tournament_id = str(assignment.tournament_external_id or "") if assignment is not None else ""
preference = session.get(UserPreference, user_id)
mapping_language = ""
if preference is not None:
preferred_vmix = str(preference.vmix_language or "").strip().lower()
preferred_display = str(preference.display_language or "").strip().lower()
if preferred_vmix in {"ru", "en"}:
mapping_language = preferred_vmix
elif preferred_display in {"ru", "en"}:
mapping_language = preferred_display
profile = self._active_mapping_profile_for_device(session, device)
if profile is None:
return {"ok": False, "reason": "mapping_missing", "device_id": device_id}
fields = list(
session.scalars(
select(VmixMappingField)
.where(and_(VmixMappingField.profile_id == profile.id, VmixMappingField.enabled.is_(True)))
.order_by(VmixMappingField.sort_order, VmixMappingField.id)
)
)
if source_codes is not None:
wanted_sources = {str(code) for code in source_codes}
fields = [field for field in fields if self._mapping_source_code(field.data_key) in wanted_sources]
profile_id = profile.id
profile_name = profile.name
profile_version = profile.version
user = HockeyUser(id=user_id, login=login or user_id, display_name=login or user_id)
supplied_context = {
"game_id": match_id,
"tournament_id": tournament_id,
"device_id": device_id,
"assignment_id": assignment_id,
}
if isinstance(extra_context, dict):
for raw_key, raw_value in list(extra_context.items())[:32]:
key = re.sub(r"[^A-Za-z0-9_.:-]+", "_", str(raw_key or "").strip())[:96]
if not key or raw_value is None:
continue
supplied_context[key] = str(raw_value)[:8000]
if mapping_language:
supplied_context["ui_language"] = mapping_language
catalog_source_codes = None if source_codes is None else {str(code) for code in source_codes}
if catalog_source_codes is not None:
for field in fields:
rule = _mapping_rule_payload(getattr(field, "rule_json", "{}"))
if not bool(rule.get("enabled")):
continue
for dependency in (rule.get("left_key"), rule.get("right_key") if str(rule.get("right_mode") or "field").lower() != "value" else ""):
code = self._mapping_source_code(dependency)
if code:
catalog_source_codes.add(code)
catalog = (
self.mapping_data.data_catalog(user, supplied_context, source_codes=catalog_source_codes)
if catalog_source_codes is not None
else self.mapping_data.data_catalog(user, supplied_context)
)
by_key = {str(item.get("key") or ""): item for item in (catalog.get("items") or []) if item.get("key")}
result: dict[str, Any] = {
"ok": True,
"reason": reason,
"device_id": device_id,
"assignment_id": assignment_id,
"match_id": match_id,
"profile_id": profile_id,
"profile_name": profile_name,
"profile_version": profile_version,
"total": len(fields),
"applied": 0,
"rules_total": 0,
"rules_applied": 0,
"unchanged": 0,
"rules_unchanged": 0,
"rule_missing": [],
"source_codes": sorted(source_codes) if source_codes is not None else [],
"missing": [],
"errors": [],
"transport": "batch" if self._agent_supports_batch(agent_version) else "legacy",
}
pending_entries: list[dict[str, Any]] = []
for field in fields:
data_key = str(field.data_key or "")
item = by_key.get(data_key)
if item is None:
result["missing"].append(data_key)
continue
raw_value = item.get("value", "")
value = "" if raw_value is None else str(raw_value)
input_ref = str(field.vmix_input_key or field.vmix_input_title or field.vmix_input_number or "").strip()
selected_name = str(field.vmix_field or "").strip()
if not input_ref or not selected_name:
result["errors"].append({"key": data_key, "reason": "missing_vmix_target"})
continue
field_type = str(field.field_type or "text").lower()
function = _mapping_value_function(field_type)
cache_key = (device_id, assignment_id, int(profile_id), input_ref, selected_name, function, data_key)
if only_changed and self._mapping_value_cache.get(cache_key) == value:
result["unchanged"] += 1
else:
pending_entries.append({
"entry_kind": "value",
"data_key": data_key,
"value": value,
"cache_key": cache_key,
"command": {
"Function": function,
"Input": input_ref,
"SelectedName": selected_name,
"Value": value,
},
})
rule = _mapping_rule_payload(getattr(field, "rule_json", "{}"))
rule_command, rule_reason = _mapping_rule_command(
rule, field_type=field_type, input_ref=input_ref, selected_name=selected_name,
default_data_key=data_key, by_key=by_key,
)
if rule_reason.startswith("missing_"):
result["rule_missing"].append({"key": data_key, "reason": rule_reason})
elif rule_reason:
result["errors"].append({"key": data_key, "reason": f"rule:{rule_reason}"})
if rule_command is not None:
result["rules_total"] += 1
rule_function = str(rule_command.get("Function") or "")
rule_value = str(rule_command.get("Value") or rule_function)
rule_cache_key = (device_id, assignment_id, int(profile_id), input_ref, selected_name, rule_function, f"rule:{data_key}")
if only_changed and self._mapping_value_cache.get(rule_cache_key) == rule_value:
result["rules_unchanged"] += 1
else:
pending_entries.append({
"entry_kind": "rule",
"data_key": data_key,
"value": rule_value,
"cache_key": rule_cache_key,
"command": rule_command,
})
result["total_commands"] = len(pending_entries)
if not pending_entries:
result["ok"] = not result["errors"]
return result
if self._agent_supports_batch(agent_version):
try:
ack = await self.send_vmix_batch(
device_id,
assignment_id=assignment_id,
match_id=match_id,
commands=[entry["command"] for entry in pending_entries],
timeout=max(6.0, min(15.0, 4.0 + len(pending_entries) * 0.03)),
)
ack_results = ack.get("results") if isinstance(ack.get("results"), list) else []
for index, entry in enumerate(pending_entries):
item_ack = ack_results[index] if index < len(ack_results) and isinstance(ack_results[index], dict) else {}
if bool(item_ack.get("ok")):
if entry.get("entry_kind") == "rule":
result["rules_applied"] += 1
else:
result["applied"] += 1
self._mapping_value_cache[entry["cache_key"]] = entry["value"]
else:
result["errors"].append({
"key": entry["data_key"],
"reason": str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"),
})
except HTTPException as error:
result["errors"].append({"key": "*batch*", "reason": str(error.detail)})
except Exception as error:
result["errors"].append({"key": "*batch*", "reason": str(error)[:300]})
else:
for entry in pending_entries:
try:
ack = await self.send_vmix_command(
device_id,
assignment_id=assignment_id,
match_id=match_id,
command=entry["command"],
timeout=3.0,
)
if bool(ack.get("ok")):
if entry.get("entry_kind") == "rule":
result["rules_applied"] += 1
else:
result["applied"] += 1
self._mapping_value_cache[entry["cache_key"]] = entry["value"]
else:
result["errors"].append({"key": entry["data_key"], "reason": str(ack.get("reason") or ack.get("error") or "vmix_error")})
except HTTPException as error:
result["errors"].append({"key": entry["data_key"], "reason": str(error.detail)})
except Exception as error:
result["errors"].append({"key": entry["data_key"], "reason": str(error)[:300]})
result["ok"] = not result["errors"]
return result
async def apply_mapping_for_user(
self,
user: HockeyUser,
*,
reason: str = "context_changed",
device_id: str = "",
session_token: str = "",
only_changed: bool = False,
) -> dict[str, Any] | None:
"""Apply Mapping to exactly one browser/session Agent, never fan out by account."""
requested = str(device_id or "").strip()
token = str(session_token or "").strip()
with self.database.session() as session:
if not requested and token:
operator = session.scalar(select(OperatorSession).where(and_(
OperatorSession.session_token == token,
OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active",
)))
if operator is not None:
requested = str(operator.vmix_device_uuid or "").strip()
if requested:
requested = self.normalise_device_id(requested)
device = session.scalar(select(VmixDevice).where(and_(
VmixDevice.device_uuid == requested,
VmixDevice.wfl_user_id == user.id,
VmixDevice.is_active_for_account.is_(True),
)))
if device is None:
return {"ok": False, "reason": "device_not_available", "device_id": requested}
target = device.device_uuid
else:
devices = list(session.scalars(select(VmixDevice).where(and_(
VmixDevice.wfl_user_id == user.id,
VmixDevice.is_active_for_account.is_(True),
)).order_by(desc(VmixDevice.last_seen_at))))
if not devices:
return None
if len(devices) != 1:
return {"ok": False, "reason": "device_selection_required", "total_devices": len(devices)}
target = devices[0].device_uuid
return await self.apply_mapping_to_device(target, reason=reason, only_changed=only_changed)
async def apply_mapping_to_all_active_devices(self, *, reason: str = "global_runtime_changed") -> dict[str, Any]:
"""Admin-only explicit fan-out used for truly global maintenance actions."""
with self.database.session() as session:
device_ids = [
str(value)
for value in session.scalars(
select(VmixDevice.device_uuid)
.where(and_(
VmixDevice.wfl_user_id.is_not(None),
VmixDevice.is_active_for_account.is_(True),
VmixDevice.vmix_connected.is_(True),
))
.order_by(VmixDevice.id)
)
if str(value or "").strip()
]
if not device_ids:
return {"ok": True, "reason": reason, "total_devices": 0, "applied_devices": 0, "devices": []}
async def apply_one(target_device_id: str) -> dict[str, Any]:
try:
result = await self.apply_mapping_to_device(target_device_id, reason=reason)
return {"device_id": target_device_id, **result}
except Exception as error:
return {"ok": False, "device_id": target_device_id, "reason": str(error)[:500]}
results = list(await asyncio.gather(*(apply_one(target_device_id) for target_device_id in device_ids)))
applied_devices = sum(1 for item in results if bool(item.get("ok")))
return {
"ok": applied_devices == len(device_ids),
"reason": reason,
"total_devices": len(device_ids),
"applied_devices": applied_devices,
"devices": results,
}
async def _run_vmix_sequence_on_device(
self,
device_id: str,
assignment_id: str,
match_id: str,
commands: list[dict[str, Any]],
*,
timeout: float,
) -> dict[str, Any]:
"""Execute one sequence serially on one Agent."""
results: list[dict[str, Any]] = []
for index, raw in enumerate(commands):
function = str(raw.get("Function") or raw.get("function") or "").strip()
if not function:
raise HTTPException(status_code=422, detail=f"Команда {index + 1}: не указана Function")
command: dict[str, Any] = {"Function": function}
for key, value in raw.items():
if str(key).lower() == "function" or value is None:
continue
if str(key).lower() != "value" and str(value) == "":
continue
command[str(key)] = value
ack = await self.send_vmix_command(
device_id,
assignment_id=assignment_id,
match_id=match_id,
command=command,
timeout=timeout,
)
item = {
"index": index,
"function": function,
"ok": bool(ack.get("ok")),
"reason": str(ack.get("reason") or ack.get("error") or ""),
}
results.append(item)
if not item["ok"]:
raise HTTPException(
status_code=502,
detail={
"message": f"vMix command failed: {function}",
"index": index,
"results": results,
},
)
return {
"ok": True,
"device_id": device_id,
"match_id": match_id,
"assignment_id": assignment_id,
"applied": len(results),
"results": results,
}
async def run_vmix_sequence_for_user(
self,
user: HockeyUser,
commands: list[dict[str, Any]],
*,
device_id: str = "",
session_token: str = "",
timeout: float = 4.0,
) -> dict[str, Any]:
"""Execute commands on exactly one Agent bound to this browser/match session."""
requested_device_id = str(device_id or "").strip()
session_token = str(session_token or "").strip()
operator_game_id = ""
operator_tournament_id = ""
bound_device_id = ""
with self.database.session() as session:
if session_token:
operator = session.scalar(
select(OperatorSession).where(
and_(
OperatorSession.session_token == session_token,
OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active",
)
)
)
if operator is None:
raise HTTPException(status_code=404, detail="Матчевая сессия не найдена")
operator_game_id = str(operator.game_external_id or "")
operator_tournament_id = str(operator.tournament_external_id or "")
bound_device_id = str(operator.vmix_device_uuid or "").strip()
target_device_id = requested_device_id or bound_device_id
if target_device_id:
target_device_id = self.normalise_device_id(target_device_id)
device = session.scalar(
select(VmixDevice).where(
and_(
VmixDevice.device_uuid == target_device_id,
VmixDevice.wfl_user_id == user.id,
VmixDevice.is_active_for_account.is_(True),
)
)
)
if device is None:
raise HTTPException(status_code=409, detail="Выбранный Hockey Agent недоступен или отключён")
else:
candidates = list(session.scalars(
select(VmixDevice)
.where(and_(VmixDevice.wfl_user_id == user.id, VmixDevice.is_active_for_account.is_(True)))
.order_by(desc(VmixDevice.last_seen_at))
))
if not candidates:
raise HTTPException(status_code=409, detail="Hockey Agent не выбран")
if len(candidates) > 1:
raise HTTPException(status_code=409, detail="Для этой панели выберите конкретный Hockey Agent")
device = candidates[0]
target_device_id = device.device_uuid
# An explicit browser device_id always wins. OperatorSession keeps a
# fallback device only for legacy/single-panel clients; several browser
# panels of the same account may legitimately target different Agents.
assignment = session.scalar(
select(VmixAssignment)
.where(
and_(
VmixAssignment.device_id == device.id,
VmixAssignment.active.is_(True),
)
)
.order_by(desc(VmixAssignment.id))
)
assignment_matches = bool(
assignment is not None
and (not operator_game_id or assignment.game_external_id == operator_game_id)
and (not session_token or not assignment.operator_session_token or assignment.operator_session_token == session_token)
)
need_assignment = not assignment_matches
if need_assignment:
if not operator_game_id:
raise HTTPException(status_code=409, detail="Agent не привязан к текущему матчу")
assigned = await self.assign_match(
wfl_user_id=user.id,
game_external_id=operator_game_id,
tournament_external_id=operator_tournament_id,
device_id=target_device_id,
operator_session_token=session_token,
)
if not assigned:
raise HTTPException(status_code=409, detail="Не удалось назначить матч выбранному Agent")
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == target_device_id))
if device is None:
raise HTTPException(status_code=404, detail="Agent не найден")
assignment_id = str(device.current_assignment_key or "")
match_id = str(device.current_match_external_id or "")
if not assignment_id or not match_id:
raise HTTPException(status_code=409, detail="Agent не привязан к текущему матчу")
if operator_game_id and match_id != operator_game_id:
raise HTTPException(status_code=409, detail="Agent назначен на другой матч")
results: list[dict[str, Any]] = []
for index, raw in enumerate(commands):
function = str(raw.get("Function") or raw.get("function") or "").strip()
if not function:
raise HTTPException(status_code=422, detail=f"Команда {index + 1}: не указана Function")
command: dict[str, Any] = {"Function": function}
for key, value in raw.items():
if str(key).lower() == "function" or value is None:
continue
if str(key).lower() != "value" and str(value) == "":
continue
command[str(key)] = value
ack = await self.send_vmix_command(
target_device_id,
assignment_id=assignment_id,
match_id=match_id,
command=command,
timeout=timeout,
)
item = {
"index": index,
"function": function,
"ok": bool(ack.get("ok")),
"reason": str(ack.get("reason") or ack.get("error") or ""),
}
results.append(item)
if not item["ok"]:
raise HTTPException(
status_code=502,
detail={
"message": f"vMix command failed: {function}",
"index": index,
"results": results,
},
)
return {
"ok": True,
"device_id": target_device_id,
"match_id": match_id,
"assignment_id": assignment_id,
"session_token": session_token,
"applied": len(results),
"results": results,
}
async def admin_test_mapping_value(
self,
device_id: str,
*,
input_ref: str,
selected_name: str,
value: str,
field_type: str = "text",
) -> dict[str, Any]:
"""Send one explicit Mapping Editor preview value to a live Agent.
This method is exposed only through the admin-protected mapping route.
It deliberately uses the device's *current* assignment so a stale
Mapping Editor cannot address a different match after the operator has
switched games.
"""
device_id = self.normalise_device_id(device_id)
input_ref = str(input_ref or "").strip()
selected_name = str(selected_name or "").strip()
value = str(value or "")
field_type = str(field_type or "text").strip().lower()
if not input_ref:
raise HTTPException(status_code=422, detail="Не определён vMix Input")
if not selected_name:
raise HTTPException(status_code=422, detail="Не определено поле vMix")
if len(input_ref) > 300 or len(selected_name) > 300 or len(value) > 4000:
raise HTTPException(status_code=422, detail="Слишком длинные параметры тестовой команды")
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None:
raise HTTPException(status_code=404, detail="Agent не найден")
assignment_id = str(device.current_assignment_key or "")
match_id = str(device.current_match_external_id or "")
if not assignment_id or not match_id:
raise HTTPException(status_code=409, detail="У этого Agent сейчас не назначен матч")
if not device.vmix_connected:
raise HTTPException(status_code=409, detail="vMix этого Agent сейчас не подключён")
function = _mapping_value_function(field_type)
ack = await self.send_vmix_command(
device_id,
assignment_id=assignment_id,
match_id=match_id,
command={
"Function": function,
"Input": input_ref,
"SelectedName": selected_name,
"Value": value,
},
)
if not bool(ack.get("ok")):
reason = str(ack.get("reason") or ack.get("error") or "vMix отклонил команду")
raise HTTPException(status_code=502, detail=f"Команда дошла до agent, но не выполнена: {reason}")
return {
"ok": True,
"device_id": device_id,
"assignment_id": assignment_id,
"match_id": match_id,
"function": function,
"input": input_ref,
"selected_name": selected_name,
"value": value,
"agent_ack": ack,
}
async def admin_test_mapping_batch(self, device_id: str, commands: list[Any]) -> dict[str, Any]:
"""Apply many unsaved Mapping Editor values using batch transport when supported."""
device_id = self.normalise_device_id(device_id)
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None:
raise HTTPException(status_code=404, detail="Agent не найден")
assignment_id = str(device.current_assignment_key or "")
match_id = str(device.current_match_external_id or "")
agent_version = str(device.agent_version or "")
if not assignment_id or not match_id:
raise HTTPException(status_code=409, detail="У этого Agent сейчас не назначен матч")
if not device.vmix_connected:
raise HTTPException(status_code=409, detail="vMix этого Agent сейчас не подключён")
result: dict[str, Any] = {
"ok": True, "device_id": device_id, "assignment_id": assignment_id,
"match_id": match_id, "total": len(commands), "applied": 0, "errors": [],
"transport": "batch" if self._agent_supports_batch(agent_version) else "legacy",
}
prepared: list[dict[str, Any]] = []
for index, item in enumerate(commands):
input_ref = str(getattr(item, "input", "") or "").strip()
selected_name = str(getattr(item, "selected_name", "") or "").strip()
value = str(getattr(item, "value", "") or "")
field_type = str(getattr(item, "field_type", "text") or "text").strip().lower()
requested_function = str(getattr(item, "function", "") or "").strip()
data_key = str(getattr(item, "data_key", "") or "")
if not input_ref or not selected_name:
result["errors"].append({"index": index, "key": data_key, "reason": "missing_vmix_target"})
continue
if len(input_ref) > 300 or len(selected_name) > 300 or len(value) > 4000:
result["errors"].append({"index": index, "key": data_key, "reason": "parameters_too_long"})
continue
function = requested_function or _mapping_value_function(field_type)
if function not in _MAPPING_TEST_FUNCTIONS:
result["errors"].append({"index": index, "key": data_key, "reason": "unsupported_function"})
continue
prepared.append({
"index": index, "key": data_key, "field": selected_name,
"command": {"Function": function, "Input": input_ref, "SelectedName": selected_name, "Value": value},
})
if self._agent_supports_batch(agent_version) and prepared:
try:
ack = await self.send_vmix_batch(
device_id, assignment_id=assignment_id, match_id=match_id,
commands=[item["command"] for item in prepared],
timeout=max(6.0, min(15.0, 4.0 + len(prepared) * 0.03)),
)
ack_results = ack.get("results") if isinstance(ack.get("results"), list) else []
for pos, item in enumerate(prepared):
item_ack = ack_results[pos] if pos < len(ack_results) and isinstance(ack_results[pos], dict) else {}
if bool(item_ack.get("ok")):
result["applied"] += 1
else:
result["errors"].append({
"index": item["index"], "key": item["key"], "field": item["field"],
"reason": str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"),
})
except Exception as error:
result["errors"].append({"index": -1, "key": "*batch*", "reason": str(getattr(error, "detail", error))[:300]})
else:
for item in prepared:
try:
ack = await self.send_vmix_command(
device_id, assignment_id=assignment_id, match_id=match_id,
command=item["command"],
)
if bool(ack.get("ok")):
result["applied"] += 1
else:
result["errors"].append({
"index": item["index"], "key": item["key"], "field": item["field"],
"reason": str(ack.get("reason") or ack.get("error") or "vmix_error"),
})
except Exception as error:
result["errors"].append({"index": item["index"], "key": item["key"], "field": item["field"], "reason": str(error)[:300]})
result["ok"] = not result["errors"]
return result
async def list_for_user(self, user: HockeyUser) -> dict[str, Any]:
now = _utcnow()
visible_since = now - timedelta(minutes=5)
with self.database.session() as session:
rows = list(
session.scalars(
select(VmixDevice)
.where(
(VmixDevice.wfl_user_id == user.id)
| (VmixDevice.last_seen_at >= visible_since)
)
.order_by(desc(VmixDevice.is_active_for_account), desc(VmixDevice.last_seen_at))
)
)
devices = [self._device_payload(row, viewer_user_id=user.id) for row in rows]
live_ids = set(self._live)
for device in devices:
device["online"] = device["device_id"] in live_ids
if device["paired_to_me"]:
device["pair_state"] = "mine"
elif device["paired"]:
device["pair_state"] = "busy"
else:
device["pair_state"] = "free"
active_device_ids = [
item["device_id"] for item in devices
if item["paired_to_me"] and item["active_for_account"]
]
return {
"protocol": AGENT_PROTOCOL_VERSION,
"devices": devices,
# Kept for backward compatibility with older browser bundles.
"active_device_id": active_device_ids[0] if active_device_ids else "",
"active_device_ids": active_device_ids,
}
async def pair_device(self, device_id: str, user: HockeyUser, *, make_active: bool = True) -> dict[str, Any]:
"""Pair one Agent without disabling other Agents owned by the account."""
device_id = self.normalise_device_id(device_id)
if device_id not in self._live:
raise HTTPException(status_code=409, detail="Agent сейчас не подключён к серверу")
now = _utcnow()
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None:
raise HTTPException(status_code=404, detail="Device ID не найден")
if row.wfl_user_id and row.wfl_user_id != user.id:
raise HTTPException(status_code=409, detail="Это устройство уже прикреплено к другому аккаунту")
row.wfl_user_id = user.id
row.login_snapshot = user.login
row.paired_at = row.paired_at or now
if make_active:
row.is_active_for_account = True
session.flush()
payload = self._device_payload(row, viewer_user_id=user.id)
await self.send(
device_id,
{
"type": "pairing.confirmed",
"protocol": AGENT_PROTOCOL_VERSION,
"device_id": device_id,
"account": {"id": user.id, "login": user.login},
"active_for_account": payload["active_for_account"],
},
)
# Do not steal the browser's current match automatically. A match is
# assigned only when this concrete device is selected for that panel.
payload["assignment"] = None
return payload
async def activate_device(self, device_id: str, user: HockeyUser) -> dict[str, Any]:
"""Enable one Agent. Other enabled Agents remain enabled."""
device_id = self.normalise_device_id(device_id)
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None or row.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
row.is_active_for_account = True
session.flush()
payload = self._device_payload(row, viewer_user_id=user.id)
await self.send(device_id, {"type": "device.activated", "device_id": device_id})
return payload
async def deactivate_device(self, device_id: str, user: HockeyUser) -> dict[str, Any]:
"""Stop routing this account's data/commands to one Agent only."""
device_id = self.normalise_device_id(device_id)
now = _utcnow()
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None or row.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
for assignment in session.scalars(
select(VmixAssignment).where(
and_(VmixAssignment.device_id == row.id, VmixAssignment.active.is_(True))
)
):
assignment.active = False
assignment.ended_at = now
assignment.last_activity_at = now
row.is_active_for_account = False
row.current_match_external_id = ""
row.current_assignment_key = ""
session.flush()
payload = self._device_payload(row, viewer_user_id=user.id)
await self.send(device_id, {"type": "device.deactivated", "device_id": device_id, "reason": "disabled_by_operator"})
return payload
async def unpair_device(self, device_id: str, user: HockeyUser) -> dict[str, Any]:
device_id = self.normalise_device_id(device_id)
now = _utcnow()
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None or row.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
for assignment in session.scalars(
select(VmixAssignment).where(
and_(VmixAssignment.device_id == row.id, VmixAssignment.active.is_(True))
)
):
assignment.active = False
assignment.ended_at = now
row.wfl_user_id = None
row.login_snapshot = ""
row.paired_at = None
row.is_active_for_account = False
row.current_match_external_id = ""
row.current_assignment_key = ""
await self.send(device_id, {"type": "pairing.revoked", "device_id": device_id})
return {"ok": True, "device_id": device_id}
async def assign_current_match(self, user: HockeyUser, *, device_id: str = "") -> dict[str, Any] | None:
"""Assign the user's current operator-session match to one concrete Agent."""
with self.database.session() as session:
operator = session.scalar(
select(OperatorSession)
.where(
and_(
OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active",
)
)
.order_by(desc(OperatorSession.id))
)
if operator is None or not operator.game_external_id:
return None
game_id = str(operator.game_external_id)
tournament_id = str(operator.tournament_external_id)
session_token = str(operator.session_token or "")
selected_device_id = str(device_id or operator.vmix_device_uuid or "").strip()
return await self.assign_match(
wfl_user_id=user.id,
game_external_id=game_id,
tournament_external_id=tournament_id,
device_id=selected_device_id,
operator_session_token=session_token,
)
async def assign_match(
self,
*,
wfl_user_id: str,
game_external_id: str,
tournament_external_id: str = "",
device_id: str = "",
operator_session_token: str = "",
) -> dict[str, Any] | None:
"""Assign a match to exactly one Agent. Never broadcast to all active Agents."""
game_external_id = str(game_external_id or "").strip()
requested_device_id = str(device_id or "").strip()
operator_session_token = str(operator_session_token or "").strip()
if not game_external_id:
return None
now = _utcnow()
with self.database.session() as session:
if requested_device_id:
requested_device_id = self.normalise_device_id(requested_device_id)
device = session.scalar(
select(VmixDevice).where(
and_(
VmixDevice.device_uuid == requested_device_id,
VmixDevice.wfl_user_id == str(wfl_user_id),
VmixDevice.is_active_for_account.is_(True),
)
)
)
else:
# Legacy/single-Agent fallback only. If several devices are enabled,
# routing must be explicit so a browser cannot control the wrong vMix.
candidates = list(session.scalars(
select(VmixDevice)
.where(
and_(
VmixDevice.wfl_user_id == str(wfl_user_id),
VmixDevice.is_active_for_account.is_(True),
)
)
.order_by(desc(VmixDevice.last_seen_at))
))
device = candidates[0] if len(candidates) == 1 else None
if device is None:
return None
if operator_session_token:
operator = session.scalar(
select(OperatorSession).where(
and_(
OperatorSession.session_token == operator_session_token,
OperatorSession.wfl_user_id == str(wfl_user_id),
OperatorSession.status == "active",
)
)
)
if operator is not None:
operator.vmix_device_uuid = device.device_uuid
operator.last_activity_at = now
current = session.scalar(
select(VmixAssignment)
.where(
and_(
VmixAssignment.device_id == device.id,
VmixAssignment.active.is_(True),
)
)
.order_by(desc(VmixAssignment.id))
)
same_binding = bool(
current is not None
and current.game_external_id == game_external_id
and (not operator_session_token or current.operator_session_token == operator_session_token)
)
if same_binding:
current.last_activity_at = now
assignment = current
else:
if current is not None:
current.active = False
current.ended_at = now
current.last_activity_at = now
assignment = VmixAssignment(
assignment_key=secrets.token_urlsafe(18),
device_id=device.id,
wfl_user_id=str(wfl_user_id),
operator_session_token=operator_session_token,
tournament_external_id=str(tournament_external_id or ""),
game_external_id=game_external_id,
active=True,
created_at=now,
last_activity_at=now,
)
session.add(assignment)
session.flush()
device.current_match_external_id = game_external_id
device.current_assignment_key = assignment.assignment_key
device.last_seen_at = device.last_seen_at or now
resolved_device_id = device.device_uuid
payload = self._assignment_payload(assignment, device_id=resolved_device_id)
delivered = await self.send(
resolved_device_id,
{
"type": "match.assign",
"protocol": AGENT_PROTOCOL_VERSION,
**payload,
},
)
payload["delivered"] = delivered
if delivered:
payload["mapping_apply"] = await self.apply_mapping_to_device(resolved_device_id, reason="match_assigned")
return payload
async def select_device_for_session(
self,
device_id: str,
user: HockeyUser,
*,
session_token: str = "",
) -> dict[str, Any]:
"""Bind this browser/operator session to one enabled Agent."""
device_id = self.normalise_device_id(device_id)
session_token = str(session_token or "").strip()
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or device.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
if not device.is_active_for_account:
raise HTTPException(status_code=409, detail="Сначала включите получение данных для этого Agent")
operator = None
if session_token:
operator = session.scalar(
select(OperatorSession).where(
and_(
OperatorSession.session_token == session_token,
OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active",
)
)
)
if operator is None:
raise HTTPException(status_code=404, detail="Матчевая сессия не найдена")
operator.vmix_device_uuid = device_id
game_id = str(operator.game_external_id or "")
tournament_id = str(operator.tournament_external_id or "")
else:
game_id = ""
tournament_id = ""
assignment = None
if game_id:
assignment = await self.assign_match(
wfl_user_id=user.id,
game_external_id=game_id,
tournament_external_id=tournament_id,
device_id=device_id,
operator_session_token=session_token,
)
return {
"ok": True,
"device_id": device_id,
"session_token": session_token,
"assignment": assignment,
}
async def reconcile_device(self, device_id: str) -> None:
"""Restore only this Agent's own assignment after reconnect."""
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or not device.wfl_user_id:
return
user_id = device.wfl_user_id
login = device.login_snapshot
active = device.is_active_for_account
assignment = None
if active and device.current_assignment_key:
assignment = session.scalar(
select(VmixAssignment).where(
and_(
VmixAssignment.device_id == device.id,
VmixAssignment.assignment_key == device.current_assignment_key,
VmixAssignment.active.is_(True),
)
)
)
assignment_payload = self._assignment_payload(assignment, device_id=device_id) if assignment is not None else None
await self.send(
device_id,
{
"type": "pairing.confirmed",
"protocol": AGENT_PROTOCOL_VERSION,
"device_id": device_id,
"account": {"id": user_id, "login": login},
"active_for_account": active,
},
)
if active and assignment_payload is not None:
delivered = await self.send(
device_id,
{"type": "match.assign", "protocol": AGENT_PROTOCOL_VERSION, **assignment_payload},
)
if delivered:
await self.apply_mapping_to_device(device_id, reason="agent_reconnected")
async def receive_inventory(self, device_id: str, message: dict[str, Any]) -> None:
inventory = message.get("inventory") if isinstance(message.get("inventory"), dict) else {}
fingerprint = str(inventory.get("fingerprint") or "").strip().lower()
inputs = inventory.get("inputs") if isinstance(inventory.get("inputs"), list) else []
if not re.fullmatch(r"[0-9a-f]{64}", fingerprint) or len(inputs) > 1000:
return
clean_inputs: list[dict[str, Any]] = []
field_count = 0
for raw_input in inputs:
if not isinstance(raw_input, dict):
continue
raw_fields = raw_input.get("fields") if isinstance(raw_input.get("fields"), list) else []
fields: list[dict[str, str]] = []
for raw_field in raw_fields[:500]:
if not isinstance(raw_field, dict):
continue
name = str(raw_field.get("name") or "")[:300]
if not name:
continue
fields.append({
"name": name,
"type": str(raw_field.get("type") or "text")[:32],
"index": str(raw_field.get("index") or "")[:32],
})
field_count += len(fields)
clean_inputs.append({
"key": str(raw_input.get("key") or "")[:128],
"number": str(raw_input.get("number") or "")[:32],
"title": str(raw_input.get("title") or "")[:300],
"type": str(raw_input.get("type") or "")[:64],
"fields": fields,
})
agent_fingerprint = fingerprint
stable_fingerprint = _stable_vmix_inventory_fingerprint(clean_inputs)
clean_inventory = {
"fingerprint": stable_fingerprint,
"agent_fingerprint": agent_fingerprint,
"inputs": clean_inputs,
"input_count": len(clean_inputs),
"field_count": field_count,
"vmix_version": str(inventory.get("vmix_version") or "")[:64],
}
now = _utcnow()
profile_payload: dict[str, Any] | None = None
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None:
return
# Server-side stable fingerprint deliberately ignores Input numbers
# and ordering. Moving an Input in vMix must not detach its Mapping.
device.project_fingerprint = stable_fingerprint
device.project_inventory_json = json.dumps(clean_inventory, ensure_ascii=False, separators=(",", ":"))
device.project_input_count = len(clean_inputs)
device.project_field_count = field_count
device.project_scanned_at = now
profile = self._active_mapping_profile_for_device(session, device)
if profile is not None:
# Refresh the profile inventory only when this project is its exact
# stable identity. A compatibility-recovered profile may be shared
# by another active Agent whose project contains a harmless extra
# Input, so do not make the primary identity bounce between devices.
if str(profile.project_fingerprint or "") == stable_fingerprint:
profile.inventory_json = device.project_inventory_json
profile.updated_at = now
profile_payload = self._mapping_profile_payload(session, profile, include_inventory=False, include_fields=True)
if profile_payload is not None:
await self.send(device_id, {"type": "mapping.assigned", **profile_payload})
else:
await self.send(device_id, {"type": "mapping.missing", "project_fingerprint": stable_fingerprint})
def _mapping_profile_payload(
self,
session: Any,
profile: VmixMappingProfile,
*,
include_inventory: bool = True,
include_fields: bool = True,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"id": profile.id,
"name": profile.name,
"description": profile.description,
"project_fingerprint": profile.project_fingerprint,
"version": profile.version,
"active": profile.active,
"created_by": profile.created_by,
"updated_by": profile.updated_by,
"created_at": profile.created_at.isoformat() if profile.created_at else "",
"updated_at": profile.updated_at.isoformat() if profile.updated_at else "",
}
if include_inventory:
try:
payload["inventory"] = json.loads(profile.inventory_json or "{}")
except Exception:
payload["inventory"] = {}
if include_fields:
rows = list(session.scalars(
select(VmixMappingField)
.where(VmixMappingField.profile_id == profile.id)
.order_by(VmixMappingField.sort_order, VmixMappingField.id)
))
payload["fields"] = [
{
"id": row.id,
"graphic": row.graphic,
"data_key": row.data_key,
"vmix_input_key": row.vmix_input_key,
"vmix_input_number": row.vmix_input_number,
"vmix_input_title": row.vmix_input_title,
"vmix_field": row.vmix_field,
"field_type": row.field_type,
"rule": _mapping_rule_payload(row.rule_json),
"enabled": row.enabled,
"sort_order": row.sort_order,
}
for row in rows
]
return payload
def _mapping_device_display_payload(self, session: Any, profile: VmixMappingProfile) -> dict[str, Any]:
source_id = self._runtime_mapping_source_id(profile)
if source_id:
source = session.get(VmixMappingProfile, source_id)
if source is not None:
return {
"id": source.id,
"name": source.name,
"version": source.version,
"runtime_profile_id": profile.id,
"source_profile_id": source.id,
}
return {"id": profile.id, "name": profile.name, "version": profile.version, "source_profile_id": profile.id}
async def list_mapping_devices(self) -> dict[str, Any]:
with self.database.session() as session:
rows = list(session.scalars(select(VmixDevice).order_by(desc(VmixDevice.last_seen_at))))
items = []
for row in rows:
profile = self._active_mapping_profile_for_device(session, row)
items.append({
"device_id": row.device_uuid,
"name": row.name or row.hostname or row.device_uuid,
"hostname": row.hostname,
"online": row.device_uuid in self._live,
"vmix_connected": bool(row.vmix_connected),
"vmix_version": row.vmix_version,
"owner": row.login_snapshot,
"project_fingerprint": row.project_fingerprint,
"input_count": row.project_input_count,
"field_count": row.project_field_count,
"scanned_at": row.project_scanned_at.isoformat() if row.project_scanned_at else "",
"mapping": (self._mapping_device_display_payload(session, profile) if profile else None),
})
return {"devices": items}
async def latest_vmix_inventory(self) -> dict[str, Any]:
"""Return the most recent connected vMix project inventory for editor helpers.
Inventory is reported by Hockey Agent and stored on the device row, so the
UI Builder can populate Input/SelectedName controls without talking to the
operator's local vMix HTTP API directly.
"""
with self.database.session() as session:
rows = list(session.scalars(select(VmixDevice).order_by(desc(VmixDevice.last_seen_at))))
devices: list[dict[str, Any]] = []
selected: VmixDevice | None = None
for row in rows:
has_inventory = bool(str(row.project_inventory_json or "").strip())
online = row.device_uuid in self._live
devices.append({
"device_id": row.device_uuid,
"name": row.name or row.hostname or row.device_uuid,
"hostname": row.hostname or "",
"online": online,
"vmix_connected": bool(row.vmix_connected),
"vmix_version": row.vmix_version or "",
"input_count": int(row.project_input_count or 0),
"field_count": int(row.project_field_count or 0),
"scanned_at": row.project_scanned_at.isoformat() if row.project_scanned_at else "",
"has_inventory": has_inventory,
})
if selected is None and has_inventory and online and bool(row.vmix_connected):
selected = row
if selected is None:
selected = next((row for row in rows if str(row.project_inventory_json or "").strip() and bool(row.vmix_connected)), None)
if selected is None:
selected = next((row for row in rows if str(row.project_inventory_json or "").strip()), None)
inventory: dict[str, Any] = {}
if selected is not None:
try:
parsed = json.loads(selected.project_inventory_json or "{}")
inventory = parsed if isinstance(parsed, dict) else {}
except Exception:
inventory = {}
return {
"device_id": selected.device_uuid if selected is not None else "",
"device_name": (selected.name or selected.hostname or selected.device_uuid) if selected is not None else "",
"online": bool(selected is not None and selected.device_uuid in self._live),
"vmix_connected": bool(selected.vmix_connected) if selected is not None else False,
"inventory": inventory,
"devices": devices,
}
async def vmix_inventory_for_user(self, user: HockeyUser, *, device_id: str = "") -> dict[str, Any]:
"""Return vMix inventory for one device owned by the current operator.
This endpoint is intentionally read-only and available to regular hockey
operators so settings pages can store stable Input keys without requiring
the UI Builder editor PIN.
"""
requested = self.normalise_device_id(device_id) if str(device_id or "").strip() else ""
with self.database.session() as session:
query = select(VmixDevice).where(VmixDevice.wfl_user_id == user.id)
rows = list(session.scalars(query.order_by(desc(VmixDevice.last_seen_at))))
selected: VmixDevice | None = None
if requested:
selected = next((row for row in rows if row.device_uuid == requested), None)
if selected is None:
raise HTTPException(status_code=404, detail="Agent не принадлежит текущему аккаунту")
else:
selected = next((row for row in rows if row.device_uuid in self._live and bool(row.vmix_connected) and str(row.project_inventory_json or "").strip()), None)
if selected is None:
selected = next((row for row in rows if str(row.project_inventory_json or "").strip()), None)
inventory: dict[str, Any] = {}
if selected is not None:
try:
parsed = json.loads(selected.project_inventory_json or "{}")
inventory = parsed if isinstance(parsed, dict) else {}
except Exception:
inventory = {}
return {
"device_id": selected.device_uuid if selected is not None else "",
"device_name": (selected.name or selected.hostname or selected.device_uuid) if selected is not None else "",
"online": bool(selected is not None and selected.device_uuid in self._live),
"vmix_connected": bool(selected.vmix_connected) if selected is not None else False,
"inventory": inventory,
}
@staticmethod
def _portable_text(value: Any) -> str:
return " ".join(str(value or "").strip().casefold().split())
@classmethod
def _rebind_portable_mapping_fields(
cls,
fields: list[dict[str, Any]],
inventory: dict[str, Any],
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
"""Rebind saved Mapping targets to another vMix project inventory.
Resolution order is deliberately conservative: stable Input key first,
then a unique title, then a unique positional number as a legacy fallback.
The selected vMix field itself must also exist on the resolved Input.
"""
inputs = inventory.get("inputs") if isinstance(inventory, dict) and isinstance(inventory.get("inputs"), list) else []
by_key: dict[str, dict[str, Any]] = {}
by_title: dict[str, list[dict[str, Any]]] = {}
by_number: dict[str, list[dict[str, Any]]] = {}
for item in inputs:
if not isinstance(item, dict):
continue
key = str(item.get("key") or "").strip()
title = cls._portable_text(item.get("title"))
number = str(item.get("number") or "").strip()
if key:
by_key[key] = item
if title:
by_title.setdefault(title, []).append(item)
if number:
by_number.setdefault(number, []).append(item)
rebound: list[dict[str, Any]] = []
skipped: list[dict[str, Any]] = []
matched_by = {"key": 0, "title": 0, "number": 0}
seen: set[tuple[str, str, str]] = set()
for index, raw in enumerate((fields or [])[:2000]):
if not isinstance(raw, dict):
skipped.append({"index": index, "reason": "invalid_field"})
continue
data_key = str(raw.get("data_key") or "").strip()
vmix_field = str(raw.get("vmix_field") or "").strip()
if not data_key or not vmix_field:
skipped.append({"index": index, "data_key": data_key, "vmix_field": vmix_field, "reason": "missing_mapping_key"})
continue
source_key = str(raw.get("vmix_input_key") or "").strip()
source_title = str(raw.get("vmix_input_title") or "").strip()
source_number = str(raw.get("vmix_input_number") or "").strip()
target = by_key.get(source_key) if source_key else None
resolution = "key" if target is not None else ""
if target is None and source_title:
title_matches = by_title.get(cls._portable_text(source_title), [])
if len(title_matches) == 1:
target = title_matches[0]
resolution = "title"
elif len(title_matches) > 1:
skipped.append({"index": index, "data_key": data_key, "input": source_title, "vmix_field": vmix_field, "reason": "ambiguous_input_title"})
continue
if target is None and source_number:
number_matches = by_number.get(source_number, [])
if len(number_matches) == 1:
target = number_matches[0]
resolution = "number"
elif len(number_matches) > 1:
skipped.append({"index": index, "data_key": data_key, "input": source_number, "vmix_field": vmix_field, "reason": "ambiguous_input_number"})
continue
if target is None:
skipped.append({"index": index, "data_key": data_key, "input": source_title or source_key or source_number, "vmix_field": vmix_field, "reason": "input_not_found"})
continue
target_fields = target.get("fields") if isinstance(target.get("fields"), list) else []
exact = next((f for f in target_fields if isinstance(f, dict) and str(f.get("name") or "").strip() == vmix_field), None)
selected_name = vmix_field
if exact is None:
folded = cls._portable_text(vmix_field)
folded_matches = [f for f in target_fields if isinstance(f, dict) and cls._portable_text(f.get("name")) == folded]
if len(folded_matches) == 1:
selected_name = str(folded_matches[0].get("name") or "").strip()
else:
skipped.append({
"index": index, "data_key": data_key,
"input": str(target.get("title") or target.get("key") or target.get("number") or ""),
"vmix_field": vmix_field,
"reason": "field_not_found" if not folded_matches else "ambiguous_field",
})
continue
target_key = str(target.get("key") or "").strip()
target_title = str(target.get("title") or "").strip()
target_number = str(target.get("number") or "").strip()
dedupe = (data_key, target_key or target_title or target_number, selected_name)
if dedupe in seen:
skipped.append({"index": index, "data_key": data_key, "input": target_title, "vmix_field": selected_name, "reason": "duplicate_target"})
continue
seen.add(dedupe)
matched_by[resolution or "title"] = matched_by.get(resolution or "title", 0) + 1
rebound.append({
"graphic": str(raw.get("graphic") or "")[:100],
"data_key": data_key[:200],
"vmix_input_key": target_key[:128],
"vmix_input_number": target_number[:32],
"vmix_input_title": target_title[:300],
"vmix_field": selected_name[:300],
"field_type": str(raw.get("field_type") or "text")[:32],
"rule": dict(raw.get("rule") or {}) if isinstance(raw.get("rule"), dict) else {},
"enabled": bool(raw.get("enabled", True)),
"sort_order": len(rebound),
})
return rebound, {
"total": min(len(fields or []), 2000),
"mapped": len(rebound),
"skipped": len(skipped),
"matched_by": matched_by,
"skipped_items": skipped[:200],
}
@staticmethod
def _portable_profile_name(session: Any, desired: str, fallback: str) -> str:
base = str(desired or fallback or "Imported Mapping").strip()[:200] or "Imported Mapping"
if session.scalar(select(VmixMappingProfile).where(VmixMappingProfile.name == base)) is None:
return base
stem = base[:185]
for number in range(2, 1000):
candidate = f"{stem} ({number})"[:200]
if session.scalar(select(VmixMappingProfile).where(VmixMappingProfile.name == candidate)) is None:
return candidate
return f"{stem} {secrets.token_hex(3)}"[:200]
def _create_portable_profile_for_device(
self,
session: Any,
*,
device: VmixDevice,
fields: list[dict[str, Any]],
name: str,
description: str,
user: HockeyUser,
replace_existing: bool,
) -> tuple[VmixMappingProfile, dict[str, Any]]:
try:
inventory = json.loads(device.project_inventory_json or "{}")
except Exception:
inventory = {}
rebound, report = self._rebind_portable_mapping_fields(fields, inventory)
if not rebound:
raise HTTPException(status_code=409, detail="Не удалось сопоставить ни одной связи Mapping с выбранным vMix")
conflict = session.scalar(
select(VmixMappingProfile)
.where(and_(VmixMappingProfile.project_fingerprint == device.project_fingerprint, VmixMappingProfile.active.is_(True)))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
)
if conflict is not None:
if not replace_existing:
raise HTTPException(status_code=409, detail=f"Для этого vMix уже активен Mapping «{conflict.name}». Разрешите замену при переносе.")
conflict.active = False
conflict.updated_by = user.login
conflict.updated_at = _utcnow()
now = _utcnow()
profile = VmixMappingProfile(
name=self._portable_profile_name(session, name, "Imported Mapping"),
description=str(description or "")[:4000],
project_fingerprint=str(device.project_fingerprint or "")[:64],
inventory_json=device.project_inventory_json or "{}",
version=1,
active=True,
created_by=user.login,
updated_by=user.login,
created_at=now,
updated_at=now,
)
session.add(profile)
session.flush()
for index, item in enumerate(rebound):
session.add(VmixMappingField(
profile_id=profile.id,
graphic=str(item.get("graphic") or "")[:100],
data_key=str(item.get("data_key") or "")[:200],
vmix_input_key=str(item.get("vmix_input_key") or "")[:128],
vmix_input_number=str(item.get("vmix_input_number") or "")[:32],
vmix_input_title=str(item.get("vmix_input_title") or "")[:300],
vmix_field=str(item.get("vmix_field") or "")[:300],
field_type=str(item.get("field_type") or "text")[:32],
rule_json=json.dumps(dict(item.get("rule") or {}), ensure_ascii=False, separators=(",", ":"))[:12000],
enabled=bool(item.get("enabled", True)),
sort_order=index,
))
session.flush()
return profile, report
@staticmethod
def _runtime_mapping_profile_name(source_profile_id: int, target_fingerprint: str) -> str:
"""Deterministic hidden profile used only to adapt one saved config to another vMix."""
safe_fp = re.sub(r"[^a-zA-Z0-9]+", "", str(target_fingerprint or ""))[:32] or "project"
return f"__AUTO_MAPPING__{int(source_profile_id)}__{safe_fp}"[:200]
@staticmethod
def _runtime_mapping_source_id(profile: VmixMappingProfile | None) -> int:
if profile is None:
return 0
match = re.match(r"^__AUTO_MAPPING__(\d+)__", str(profile.name or ""))
return int(match.group(1)) if match else 0
def _upsert_runtime_mapping_for_device(
self,
session: Any,
*,
source: VmixMappingProfile,
source_fields: list[dict[str, Any]],
device: VmixDevice,
user: HockeyUser,
) -> tuple[VmixMappingProfile, dict[str, Any]]:
"""Create/update a hidden rebinding of a saved config for one concrete vMix project."""
try:
inventory = json.loads(device.project_inventory_json or "{}")
except Exception:
inventory = {}
rebound, report = self._rebind_portable_mapping_fields(source_fields, inventory)
if not rebound:
raise HTTPException(status_code=409, detail="Не удалось сопоставить ни одной связи выбранного Mapping с вашим vMix")
target_fingerprint = str(device.project_fingerprint or "")[:64]
runtime_name = self._runtime_mapping_profile_name(source.id, target_fingerprint)
runtime = session.scalar(select(VmixMappingProfile).where(VmixMappingProfile.name == runtime_name))
now = _utcnow()
# Only one concrete mapping can drive a vMix project at a time.
for other in session.scalars(
select(VmixMappingProfile).where(and_(
VmixMappingProfile.project_fingerprint == target_fingerprint,
VmixMappingProfile.active.is_(True),
))
):
if runtime is not None and other.id == runtime.id:
continue
other.active = False
other.updated_by = user.login
other.updated_at = now
if runtime is None:
runtime = VmixMappingProfile(
name=runtime_name,
description=f"Техническая адаптация Mapping #{source.id}: {source.name}"[:4000],
project_fingerprint=target_fingerprint,
inventory_json=device.project_inventory_json or "{}",
version=1,
active=True,
created_by=user.login,
updated_by=user.login,
created_at=now,
updated_at=now,
)
session.add(runtime)
session.flush()
else:
runtime.description = f"Техническая адаптация Mapping #{source.id}: {source.name}"[:4000]
runtime.project_fingerprint = target_fingerprint
runtime.inventory_json = device.project_inventory_json or "{}"
runtime.version = int(runtime.version or 0) + 1
runtime.active = True
runtime.updated_by = user.login
runtime.updated_at = now
session.execute(delete(VmixMappingField).where(VmixMappingField.profile_id == runtime.id))
session.flush()
for index, item in enumerate(rebound):
session.add(VmixMappingField(
profile_id=runtime.id,
graphic=str(item.get("graphic") or "")[:100],
data_key=str(item.get("data_key") or "")[:200],
vmix_input_key=str(item.get("vmix_input_key") or "")[:128],
vmix_input_number=str(item.get("vmix_input_number") or "")[:32],
vmix_input_title=str(item.get("vmix_input_title") or "")[:300],
vmix_field=str(item.get("vmix_field") or "")[:300],
field_type=str(item.get("field_type") or "text")[:32],
rule_json=json.dumps(dict(item.get("rule") or {}), ensure_ascii=False, separators=(",", ":"))[:12000],
enabled=bool(item.get("enabled", True)),
sort_order=index,
))
session.flush()
return runtime, report
async def use_mapping_profile_for_user(self, profile_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
"""Use any saved Mapping config on the Agent selected for the user's current match."""
requested = str(getattr(payload, "device_id", "") or "").strip()
session_token = str(getattr(payload, "session_token", "") or "").strip()
with self.database.session() as session:
source = session.get(VmixMappingProfile, profile_id)
if source is None or self._runtime_mapping_source_id(source):
raise HTTPException(status_code=404, detail="Mapping-конфиг не найден")
operator = None
if session_token:
operator = session.scalar(select(OperatorSession).where(and_(
OperatorSession.session_token == session_token,
OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active",
)))
if operator is None:
operator = session.scalar(
select(OperatorSession)
.where(and_(OperatorSession.wfl_user_id == user.id, OperatorSession.status == "active"))
.order_by(desc(OperatorSession.id))
)
if not requested and operator is not None:
requested = str(operator.vmix_device_uuid or "").strip()
if requested:
requested = self.normalise_device_id(requested)
device = session.scalar(select(VmixDevice).where(and_(
VmixDevice.device_uuid == requested,
VmixDevice.wfl_user_id == user.id,
VmixDevice.is_active_for_account.is_(True),
)))
else:
candidates = list(session.scalars(
select(VmixDevice)
.where(and_(VmixDevice.wfl_user_id == user.id, VmixDevice.is_active_for_account.is_(True)))
.order_by(desc(VmixDevice.last_seen_at))
))
device = candidates[0] if len(candidates) == 1 else None
if device is not None:
requested = str(device.device_uuid)
if device is None:
raise HTTPException(status_code=409, detail="Не удалось определить ваш Agent. Выберите Agent для текущей сессии матча.")
if not device.project_fingerprint or not str(device.project_inventory_json or "").strip():
raise HTTPException(status_code=409, detail="Ваш Agent ещё не передал структуру vMix")
if not bool(device.vmix_connected):
raise HTTPException(status_code=409, detail="На вашем Agent сейчас не подключён vMix")
# Make sure this concrete Agent is attached to the current match before sending values.
assignment = await self.assign_current_match(user, device_id=requested)
if assignment is None:
raise HTTPException(status_code=409, detail="Сначала откройте матч в основном интерфейсе")
with self.database.session() as session:
source = session.get(VmixMappingProfile, profile_id)
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == requested))
if source is None or device is None:
raise HTTPException(status_code=404, detail="Mapping или Agent больше недоступен")
source_payload = self._mapping_profile_payload(session, source, include_inventory=False, include_fields=True)
source_fields = list(source_payload.get("fields") or [])
target_fingerprint = str(device.project_fingerprint or "")
if str(source.project_fingerprint or "") == target_fingerprint:
now = _utcnow()
for other in session.scalars(select(VmixMappingProfile).where(and_(
VmixMappingProfile.project_fingerprint == target_fingerprint,
VmixMappingProfile.active.is_(True),
VmixMappingProfile.id != source.id,
))):
other.active = False
other.updated_by = user.login
other.updated_at = now
source.active = True
source.updated_by = user.login
source.updated_at = now
runtime = source
report = {
"total": len(source_fields), "mapped": len(source_fields), "skipped": 0,
"matched_by": {"key": len(source_fields), "title": 0, "number": 0},
"skipped_items": [], "reused": True,
}
else:
runtime, report = self._upsert_runtime_mapping_for_device(
session, source=source, source_fields=source_fields, device=device, user=user,
)
runtime_id = runtime.id
await self._broadcast_mapping_for_fingerprint(target_fingerprint)
applied = await self.apply_mapping_to_device(requested, reason="mapping_config_selected")
return {
"ok": True,
"device_id": requested,
"profile": {"id": source_payload.get("id"), "name": source_payload.get("name"), "version": source_payload.get("version")},
"runtime_profile_id": runtime_id,
"report": report,
"applied": applied,
}
async def export_mapping_profile(self, profile_id: int) -> dict[str, Any]:
with self.database.session() as session:
profile = session.get(VmixMappingProfile, profile_id)
if profile is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
payload = self._mapping_profile_payload(session, profile, include_inventory=True, include_fields=True)
fields = []
for row in payload.get("fields") or []:
fields.append({key: value for key, value in row.items() if key != "id"})
inventory = payload.get("inventory") if isinstance(payload.get("inventory"), dict) else {}
return {
"format": "hockey-vmix-mapping",
"schema_version": 1,
"exported_at": _utcnow().isoformat(),
"profile": {
"name": payload.get("name") or "",
"description": payload.get("description") or "",
"source_project_fingerprint": payload.get("project_fingerprint") or "",
"source_inventory": {
"input_count": inventory.get("input_count", 0),
"field_count": inventory.get("field_count", 0),
"vmix_version": inventory.get("vmix_version", ""),
},
"fields": fields,
},
}
async def copy_mapping_profile_to_device(self, profile_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
device_id = self.normalise_device_id(payload.device_id)
with self.database.session() as session:
source = session.get(VmixMappingProfile, profile_id)
if source is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or not device.project_fingerprint or not str(device.project_inventory_json or "").strip():
raise HTTPException(status_code=409, detail="Выбранный Agent ещё не передал структуру vMix")
source_payload = self._mapping_profile_payload(session, source, include_inventory=False, include_fields=True)
target_fingerprint = str(device.project_fingerprint or "")
if str(source.project_fingerprint or "") == target_fingerprint:
other = session.scalar(
select(VmixMappingProfile)
.where(and_(
VmixMappingProfile.project_fingerprint == target_fingerprint,
VmixMappingProfile.active.is_(True),
VmixMappingProfile.id != source.id,
))
.order_by(desc(VmixMappingProfile.updated_at), desc(VmixMappingProfile.id))
)
if other is not None and not bool(payload.replace_existing):
raise HTTPException(status_code=409, detail=f"Для этого vMix уже активен Mapping «{other.name}». Разрешите замену при переносе.")
if other is not None:
other.active = False
other.updated_by = user.login
other.updated_at = _utcnow()
source.active = True
source.updated_by = user.login
source.updated_at = _utcnow()
result_profile = self._mapping_profile_payload(session, source, include_inventory=True, include_fields=True)
report = {"total": len(source_payload.get("fields") or []), "mapped": len(source_payload.get("fields") or []), "skipped": 0, "matched_by": {"key": len(source_payload.get("fields") or []), "title": 0, "number": 0}, "skipped_items": [], "reused": True}
else:
fallback_name = f"{source.name} · {device.name or device.hostname or device.device_uuid}"
created, report = self._create_portable_profile_for_device(
session, device=device, fields=list(source_payload.get("fields") or []),
name=str(payload.name or "").strip() or fallback_name, description=source.description, user=user,
replace_existing=bool(payload.replace_existing),
)
result_profile = self._mapping_profile_payload(session, created, include_inventory=True, include_fields=True)
await self._broadcast_mapping_for_fingerprint(target_fingerprint)
applied = None
if bool(payload.apply_now):
applied = await self.apply_mapping_to_device(device_id, reason="portable_mapping_copy")
return {"ok": True, "profile": result_profile, "report": report, "applied": applied}
async def import_mapping_profile_to_device(self, payload: Any, user: HockeyUser) -> dict[str, Any]:
device_id = self.normalise_device_id(payload.device_id)
document = dict(payload.document or {})
if document.get("format") not in {None, "", "hockey-vmix-mapping"}:
raise HTTPException(status_code=422, detail="Это не файл Hockey vMix Mapping")
try:
schema_version = int(document.get("schema_version") or 1)
except (TypeError, ValueError):
raise HTTPException(status_code=422, detail="Некорректная версия файла Mapping")
if schema_version != 1:
raise HTTPException(status_code=422, detail="Версия файла Mapping пока не поддерживается")
profile_doc = document.get("profile") if isinstance(document.get("profile"), dict) else document
fields = profile_doc.get("fields") if isinstance(profile_doc, dict) and isinstance(profile_doc.get("fields"), list) else []
if not fields:
raise HTTPException(status_code=422, detail="В файле Mapping нет связей")
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or not device.project_fingerprint or not str(device.project_inventory_json or "").strip():
raise HTTPException(status_code=409, detail="Выбранный Agent ещё не передал структуру vMix")
target_fingerprint = str(device.project_fingerprint or "")
imported_name = str(payload.name or "").strip() or str(profile_doc.get("name") or "Imported Mapping").strip()
created, report = self._create_portable_profile_for_device(
session, device=device, fields=fields, name=imported_name,
description=str(profile_doc.get("description") or ""), user=user,
replace_existing=bool(payload.replace_existing),
)
result_profile = self._mapping_profile_payload(session, created, include_inventory=True, include_fields=True)
await self._broadcast_mapping_for_fingerprint(target_fingerprint)
applied = None
if bool(payload.apply_now):
applied = await self.apply_mapping_to_device(device_id, reason="portable_mapping_import")
return {"ok": True, "profile": result_profile, "report": report, "applied": applied}
async def list_mapping_profiles(self) -> dict[str, Any]:
with self.database.session() as session:
rows = list(session.scalars(
select(VmixMappingProfile)
.where(~VmixMappingProfile.name.like("__AUTO_MAPPING__%"))
.order_by(desc(VmixMappingProfile.updated_at), VmixMappingProfile.name)
))
items = [self._mapping_profile_payload(session, row, include_inventory=False, include_fields=False) for row in rows]
for item, row in zip(items, rows):
item["field_count"] = len(list(session.scalars(select(VmixMappingField.id).where(VmixMappingField.profile_id == row.id))))
return {"profiles": items}
async def get_mapping_profile(self, profile_id: int) -> dict[str, Any]:
with self.database.session() as session:
row = session.get(VmixMappingProfile, profile_id)
if row is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
return self._mapping_profile_payload(session, row, include_inventory=True, include_fields=True)
async def create_mapping_profile(self, payload: Any, user: HockeyUser) -> dict[str, Any]:
name = str(payload.name or "").strip()
device_id = self.normalise_device_id(payload.device_id)
if not name:
raise HTTPException(status_code=422, detail="Укажите название mapping-профиля")
now = _utcnow()
with self.database.session() as session:
if session.scalar(select(VmixMappingProfile).where(VmixMappingProfile.name == name)) is not None:
raise HTTPException(status_code=409, detail="Mapping с таким названием уже существует")
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or not device.project_fingerprint or not device.project_inventory_json:
raise HTTPException(status_code=409, detail="Agent ещё не передал структуру текущего vMix")
existing = session.scalar(select(VmixMappingProfile).where(and_(VmixMappingProfile.project_fingerprint == device.project_fingerprint, VmixMappingProfile.active.is_(True))))
if existing is not None:
raise HTTPException(status_code=409, detail=f"Для этого проекта уже назначен mapping «{existing.name}»")
row = VmixMappingProfile(
name=name[:200],
description=str(payload.description or "")[:4000],
project_fingerprint=device.project_fingerprint,
inventory_json=device.project_inventory_json,
version=1,
active=True,
created_by=user.login,
updated_by=user.login,
created_at=now,
updated_at=now,
)
session.add(row)
session.flush()
result = self._mapping_profile_payload(session, row, include_inventory=True, include_fields=True)
await self._broadcast_mapping_for_fingerprint(result["project_fingerprint"])
return result
async def update_mapping_profile(self, profile_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
now = _utcnow()
with self.database.session() as session:
row = session.get(VmixMappingProfile, profile_id)
if row is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
name = str(payload.name or "").strip()
if not name:
raise HTTPException(status_code=422, detail="Укажите название mapping-профиля")
duplicate = session.scalar(select(VmixMappingProfile).where(and_(VmixMappingProfile.name == name, VmixMappingProfile.id != profile_id)))
if duplicate is not None:
raise HTTPException(status_code=409, detail="Mapping с таким названием уже существует")
row.name = name[:200]
row.description = str(payload.description or "")[:4000]
row.active = bool(payload.active)
# The web editor saves metadata and field links as one operation.
# Version is incremented by replace_mapping_fields(), so one click
# on Save produces exactly one new mapping version.
row.updated_by = user.login
row.updated_at = now
result = self._mapping_profile_payload(session, row, include_inventory=True, include_fields=True)
await self._broadcast_mapping_for_fingerprint(result["project_fingerprint"])
return result
async def refresh_mapping_inventory(self, profile_id: int, device_id: str, user: HockeyUser) -> dict[str, Any]:
device_id = self.normalise_device_id(device_id)
now = _utcnow()
with self.database.session() as session:
row = session.get(VmixMappingProfile, profile_id)
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
if device is None or not device.project_fingerprint or not device.project_inventory_json:
raise HTTPException(status_code=409, detail="Устройство не передало структуру vMix")
conflicting = session.scalar(select(VmixMappingProfile).where(and_(VmixMappingProfile.project_fingerprint == device.project_fingerprint, VmixMappingProfile.id != row.id, VmixMappingProfile.active.is_(True))))
if conflicting is not None:
raise HTTPException(status_code=409, detail=f"Эта структура уже используется mapping «{conflicting.name}»")
old_fingerprint = row.project_fingerprint
row.project_fingerprint = device.project_fingerprint
row.inventory_json = device.project_inventory_json
row.version = int(row.version or 0) + 1
row.updated_by = user.login
row.updated_at = now
result = self._mapping_profile_payload(session, row, include_inventory=True, include_fields=True)
if old_fingerprint and old_fingerprint != result["project_fingerprint"]:
await self._broadcast_mapping_for_fingerprint(old_fingerprint)
await self._broadcast_mapping_for_fingerprint(result["project_fingerprint"])
return result
async def replace_mapping_fields(self, profile_id: int, payload: Any, user: HockeyUser) -> dict[str, Any]:
now = _utcnow()
with self.database.session() as session:
profile = session.get(VmixMappingProfile, profile_id)
if profile is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
for row in list(session.scalars(select(VmixMappingField).where(VmixMappingField.profile_id == profile_id))):
session.delete(row)
session.flush()
for index, item in enumerate(payload.fields[:2000]):
data_key = str(item.data_key or "").strip()
vmix_field = str(item.vmix_field or "").strip()
if not data_key or not vmix_field:
continue
session.add(VmixMappingField(
profile_id=profile_id,
graphic=str(item.graphic or "")[:100],
data_key=data_key[:200],
vmix_input_key=str(item.vmix_input_key or "")[:128],
vmix_input_number=("" if (str(item.vmix_input_key or "").strip() or str(item.vmix_input_title or "").strip()) else str(item.vmix_input_number or "")[:32]),
vmix_input_title=str(item.vmix_input_title or "")[:300],
vmix_field=vmix_field[:300],
field_type=str(item.field_type or "text")[:32],
rule_json=json.dumps(dict(getattr(item, "rule", {}) or {}), ensure_ascii=False, separators=(",", ":"))[:12000],
enabled=bool(item.enabled),
sort_order=index,
))
profile.version = int(profile.version or 0) + 1
profile.updated_by = user.login
profile.updated_at = now
session.flush()
result = self._mapping_profile_payload(session, profile, include_inventory=True, include_fields=True)
await self._broadcast_mapping_for_fingerprint(result["project_fingerprint"])
return result
async def delete_mapping_profile(self, profile_id: int) -> dict[str, Any]:
with self.database.session() as session:
profile = session.get(VmixMappingProfile, profile_id)
if profile is None:
raise HTTPException(status_code=404, detail="Mapping-профиль не найден")
fingerprint = profile.project_fingerprint
runtime_prefix = f"__AUTO_MAPPING__{profile_id}__%"
runtime_profiles = list(session.scalars(select(VmixMappingProfile).where(VmixMappingProfile.name.like(runtime_prefix))))
for runtime in runtime_profiles:
session.execute(delete(VmixMappingField).where(VmixMappingField.profile_id == runtime.id))
session.delete(runtime)
for field in list(session.scalars(select(VmixMappingField).where(VmixMappingField.profile_id == profile_id))):
session.delete(field)
session.delete(profile)
if fingerprint:
await self._broadcast_mapping_for_fingerprint(fingerprint)
return {"ok": True, "id": profile_id}
async def _broadcast_mapping_for_fingerprint(self, fingerprint: str) -> None:
if not fingerprint:
return
profile_payload: dict[str, Any] | None = None
device_ids: list[str] = []
with self.database.session() as session:
profile = session.scalar(
select(VmixMappingProfile)
.where(and_(VmixMappingProfile.project_fingerprint == fingerprint, VmixMappingProfile.active.is_(True)))
.order_by(desc(VmixMappingProfile.updated_at))
)
if profile is not None:
profile_payload = self._mapping_profile_payload(session, profile, include_inventory=False, include_fields=True)
# Include devices recovered by stable-key compatibility, not only
# devices whose whole-project fingerprint is byte-for-byte equal.
for row in session.scalars(select(VmixDevice).where(VmixDevice.project_fingerprint != "")):
resolved = self._active_mapping_profile_for_device(session, row)
if resolved is not None and resolved.id == profile.id:
device_ids.append(str(row.device_uuid))
else:
device_ids = [
str(row.device_uuid)
for row in session.scalars(select(VmixDevice).where(VmixDevice.project_fingerprint == fingerprint))
]
for device_id in dict.fromkeys(device_ids):
if profile_payload is None:
await self.send(device_id, {"type": "mapping.missing", "project_fingerprint": fingerprint})
else:
delivered = await self.send(device_id, {"type": "mapping.assigned", **profile_payload})
if delivered:
await self.apply_mapping_to_device(device_id, reason="mapping_loaded")
async def probe_device(self, device_id: str, user: HockeyUser) -> dict[str, Any]:
device_id = self.normalise_device_id(device_id)
with self.database.session() as session:
row = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if row is None or row.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
delivered = await self.send(device_id, {"type": "vmix.probe", "request_id": secrets.token_urlsafe(8)})
if not delivered:
raise HTTPException(status_code=409, detail="Agent сейчас offline")
return {"ok": True, "device_id": device_id}
@staticmethod
def _assignment_payload(row: VmixAssignment, *, device_id: str) -> dict[str, Any]:
return {
"assignment_id": row.assignment_key,
"device_id": device_id,
"game_id": row.game_external_id,
"tournament_id": row.tournament_external_id,
"operator_session_token": row.operator_session_token,
"created_at": row.created_at.isoformat(),
}
def _device_payload(self, row: VmixDevice, *, viewer_user_id: str) -> dict[str, Any]:
paired = bool(row.wfl_user_id)
paired_to_me = bool(viewer_user_id and row.wfl_user_id == viewer_user_id)
mapping = None
if row.project_fingerprint or str(row.project_inventory_json or "").strip():
with self.database.session() as mapping_session:
managed = mapping_session.get(VmixDevice, row.id)
profile = self._active_mapping_profile_for_device(mapping_session, managed) if managed is not None else None
if profile is not None:
mapping = {"id": profile.id, "name": profile.name, "version": profile.version}
return {
"device_id": row.device_uuid,
"name": row.name or row.hostname or row.device_uuid,
"hostname": row.hostname,
"agent_version": row.agent_version,
"paired": paired,
"paired_to_me": paired_to_me,
"active_for_account": bool(row.is_active_for_account and paired_to_me),
"owner": row.login_snapshot if paired_to_me else ("Другой оператор" if paired else ""),
"vmix_connected": bool(row.vmix_connected),
"vmix_version": row.vmix_version,
"vmix_url": row.vmix_url,
"current_match_id": row.current_match_external_id if paired_to_me else "",
"assignment_id": row.current_assignment_key if paired_to_me else "",
"last_seen": row.last_seen_at.isoformat() if row.last_seen_at else "",
"last_error": row.last_error if paired_to_me else "",
"project_fingerprint": row.project_fingerprint,
"project_input_count": row.project_input_count,
"project_field_count": row.project_field_count,
"mapping": mapping,
"online": row.device_uuid in self._live,
}
class PairDevicePayload(BaseModel):
make_active: bool = True
class TestSetTextPayload(BaseModel):
input: str = Field(min_length=1, max_length=300)
selected_name: str = Field(min_length=1, max_length=300)
value: str = Field(default="", max_length=4000)
class RuntimeVmixCommandPayload(BaseModel):
model_config = ConfigDict(extra="allow")
Function: str = Field(min_length=1, max_length=128)
Input: str = Field(default="", max_length=300)
Value: Any = ""
SelectedName: str = Field(default="", max_length=300)
Duration: str = Field(default="", max_length=64)
Mix: str = Field(default="", max_length=32)
class RuntimeVmixSequencePayload(BaseModel):
commands: list[RuntimeVmixCommandPayload] = Field(default_factory=list, min_length=1, max_length=80)
device_id: str = Field(default="", max_length=128)
session_token: str = Field(default="", max_length=128)
class SelectSessionDevicePayload(BaseModel):
session_token: str = Field(default="", max_length=128)
class MappingTestValuePayload(BaseModel):
device_id: str = Field(min_length=6, max_length=128)
input: str = Field(min_length=1, max_length=300)
selected_name: str = Field(min_length=1, max_length=300)
value: str = Field(default="", max_length=4000)
field_type: str = Field(default="text", max_length=32)
class MappingTestBatchCommandPayload(BaseModel):
input: str = Field(min_length=1, max_length=300)
selected_name: str = Field(min_length=1, max_length=300)
value: str = Field(default="", max_length=4000)
field_type: str = Field(default="text", max_length=32)
function: str = Field(default="", max_length=64)
data_key: str = Field(default="", max_length=200)
class MappingTestBatchPayload(BaseModel):
device_id: str = Field(min_length=6, max_length=128)
commands: list[MappingTestBatchCommandPayload] = Field(default_factory=list, min_length=1, max_length=500)
class MappingProfileCreatePayload(BaseModel):
name: str = Field(min_length=1, max_length=200)
device_id: str = Field(min_length=6, max_length=128)
description: str = Field(default="", max_length=4000)
class MappingProfileUpdatePayload(BaseModel):
name: str = Field(min_length=1, max_length=200)
description: str = Field(default="", max_length=4000)
active: bool = True
class MappingInventoryRefreshPayload(BaseModel):
device_id: str = Field(min_length=6, max_length=128)
class MappingFieldPayload(BaseModel):
graphic: str = Field(default="", max_length=100)
data_key: str = Field(min_length=1, max_length=200)
vmix_input_key: str = Field(default="", max_length=128)
vmix_input_number: str = Field(default="", max_length=32)
vmix_input_title: str = Field(default="", max_length=300)
vmix_field: str = Field(min_length=1, max_length=300)
field_type: str = Field(default="text", max_length=32)
rule: dict[str, Any] = Field(default_factory=dict)
enabled: bool = True
class MappingFieldsReplacePayload(BaseModel):
fields: list[MappingFieldPayload] = Field(default_factory=list, max_length=2000)
class MappingProfileUsePayload(BaseModel):
device_id: str = Field(default="", max_length=128)
session_token: str = Field(default="", max_length=128)
class MappingProfileCopyPayload(BaseModel):
device_id: str = Field(min_length=6, max_length=128)
name: str = Field(default="", max_length=200)
replace_existing: bool = False
apply_now: bool = True
class MappingPortableImportPayload(BaseModel):
device_id: str = Field(min_length=6, max_length=128)
document: dict[str, Any] = Field(default_factory=dict)
name: str = Field(default="", max_length=200)
replace_existing: bool = False
apply_now: bool = True
class ContextVariablePayload(BaseModel):
key: str = Field(min_length=1, max_length=128)
label: str = Field(min_length=1, max_length=200)
category: str = Field(default="Пользовательские", max_length=100)
description: str = Field(default="", max_length=4000)
value_type: str = Field(default="id", max_length=32)
entity_type: str = Field(default="", max_length=32)
scope: str = Field(default="match", max_length=32)
source_type: str = Field(default="manual", max_length=32)
default_value: str = Field(default="", max_length=4000)
enabled: bool = True
sort_order: int = 1000
class ContextValuePayload(BaseModel):
value: str = Field(default="", max_length=8000)
context: dict[str, Any] = Field(default_factory=dict)
class ContextValuesPayload(BaseModel):
values: dict[str, Any] = Field(default_factory=dict)
context: dict[str, Any] = Field(default_factory=dict)
class SqlDataSourcePayload(BaseModel):
code: str = Field(min_length=1, max_length=100)
name: str = Field(min_length=1, max_length=200)
category: str = Field(default="Данные", max_length=100)
description: str = Field(default="", max_length=4000)
sql_text: str = Field(min_length=1, max_length=50000)
field_metadata: dict[str, Any] = Field(default_factory=dict)
enabled: bool = True
auto_refresh_enabled: bool = False
refresh_interval_ms: int = Field(default=1000, ge=1000, le=3600000)
sort_order: int = 1000
class MappingDataContextPayload(BaseModel):
context: dict[str, Any] = Field(default_factory=dict)
class MappingSqlPreviewPayload(BaseModel):
sql_text: str = Field(min_length=1, max_length=50000)
context: dict[str, Any] = Field(default_factory=dict)
class AgentHelloPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
type: str = "hello"
protocol: int = Field(default=AGENT_PROTOCOL_VERSION)
device_id: str
device_secret: str
device_name: str = ""
hostname: str = ""
agent_version: str = ""
vmix: dict[str, Any] = Field(default_factory=dict)
def create_hockey_agent_router(
hub: VmixAgentHub,
*,
auth_dependency: Callable[..., Any],
admin_dependency: Callable[..., Any] | None = None,
) -> APIRouter:
router = APIRouter(tags=["Hockey vMix Agent"])
admin = [Depends(admin_dependency)] if admin_dependency else []
@router.get("/api/hockey/agents/devices")
async def list_devices(user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.list_for_user(user)
@router.get("/api/hockey/agents/vmix-inventory")
async def operator_vmix_inventory(
device_id: str = Query("", max_length=128),
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.vmix_inventory_for_user(user, device_id=device_id)
@router.post("/api/hockey/agents/devices/{device_id}/pair")
async def pair_device(
device_id: str,
payload: PairDevicePayload,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.pair_device(device_id, user, make_active=payload.make_active)
@router.post("/api/hockey/agents/devices/{device_id}/activate")
async def activate_device(
device_id: str,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.activate_device(device_id, user)
@router.post("/api/hockey/agents/devices/{device_id}/deactivate")
async def deactivate_device(
device_id: str,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.deactivate_device(device_id, user)
@router.post("/api/hockey/agents/devices/{device_id}/select-session")
async def select_session_device(
device_id: str,
payload: SelectSessionDevicePayload,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.select_device_for_session(device_id, user, session_token=payload.session_token)
@router.delete("/api/hockey/agents/devices/{device_id}/pair")
async def unpair_device(
device_id: str,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.unpair_device(device_id, user)
@router.post("/api/hockey/agents/devices/{device_id}/probe")
async def probe_device(
device_id: str,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.probe_device(device_id, user)
@router.post("/api/hockey/agents/devices/{device_id}/test-set-text")
async def test_set_text(
device_id: str,
payload: TestSetTextPayload,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
return await hub.test_set_text(
device_id,
user,
input_ref=payload.input,
selected_name=payload.selected_name,
value=payload.value,
)
@router.post("/api/hockey/agents/devices/{device_id}/apply-mapping")
async def apply_device_mapping(
device_id: str,
only_changed: bool = False,
active_tab: str = "",
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
device_id = hub.normalise_device_id(device_id)
with hub.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or device.wfl_user_id != user.id:
raise HTTPException(status_code=404, detail="Устройство не прикреплено к вашему аккаунту")
if not device.is_active_for_account:
raise HTTPException(status_code=409, detail="Для этого Agent выключено получение данных")
result = await hub.apply_mapping_to_device(
device_id,
reason="runtime_strength_changed" if only_changed else "manual_apply",
only_changed=only_changed,
extra_context={"active_tab": str(active_tab or "").strip()} if str(active_tab or "").strip() else None,
)
if not result.get("ok") and result.get("reason"):
raise HTTPException(status_code=409, detail=str(result.get("reason")))
return result
@router.post("/api/hockey/agents/apply-active-mapping")
async def apply_active_mapping(user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
result = await hub.apply_mapping_for_user(user, reason="language_or_runtime_changed")
return result or {"ok": False, "reason": "no_active_device"}
@router.post("/api/hockey/vmix/sequence")
async def run_runtime_vmix_sequence(
payload: RuntimeVmixSequencePayload,
user: HockeyUser = Depends(auth_dependency),
) -> dict[str, Any]:
commands = [item.model_dump(exclude_none=True) for item in payload.commands]
return await hub.run_vmix_sequence_for_user(
user,
commands,
device_id=payload.device_id,
session_token=payload.session_token,
)
@router.post("/api/hockey/admin/vmix-mapping/apply-all-active", dependencies=admin)
async def admin_apply_all_active_mappings() -> dict[str, Any]:
return await hub.apply_mapping_to_all_active_devices(reason="ui_language_changed")
@router.post("/api/hockey/admin/vmix-mapping/apply/{device_id}", dependencies=admin)
async def admin_apply_device_mapping(device_id: str) -> dict[str, Any]:
result = await hub.apply_mapping_to_device(device_id, reason="admin_manual_apply")
if not result.get("ok") and result.get("reason"):
raise HTTPException(status_code=409, detail=str(result.get("reason")))
return result
@router.post("/api/hockey/admin/vmix-mapping/test-value", dependencies=admin)
async def mapping_test_value(payload: MappingTestValuePayload) -> dict[str, Any]:
return await hub.admin_test_mapping_value(
payload.device_id,
input_ref=payload.input,
selected_name=payload.selected_name,
value=payload.value,
field_type=payload.field_type,
)
@router.post("/api/hockey/admin/vmix-mapping/test-batch", dependencies=admin)
async def mapping_test_batch(payload: MappingTestBatchPayload) -> dict[str, Any]:
return await hub.admin_test_mapping_batch(payload.device_id, payload.commands)
@router.get("/api/hockey/admin/vmix-mapping/devices", dependencies=admin)
async def mapping_devices() -> dict[str, Any]:
return await hub.list_mapping_devices()
@router.get("/api/hockey/admin/vmix-mapping/profiles", dependencies=admin)
async def mapping_profiles() -> dict[str, Any]:
return await hub.list_mapping_profiles()
@router.get("/api/hockey/admin/vmix-mapping/profiles/{profile_id}", dependencies=admin)
async def mapping_profile(profile_id: int) -> dict[str, Any]:
return await hub.get_mapping_profile(profile_id)
@router.post("/api/hockey/admin/vmix-mapping/profiles", dependencies=admin)
async def mapping_profile_create(payload: MappingProfileCreatePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.create_mapping_profile(payload, user)
@router.post("/api/hockey/admin/vmix-mapping/profiles/{profile_id}/use", dependencies=admin)
async def mapping_profile_use(profile_id: int, payload: MappingProfileUsePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.use_mapping_profile_for_user(profile_id, payload, user)
@router.get("/api/hockey/admin/vmix-mapping/profiles/{profile_id}/export", dependencies=admin)
async def mapping_profile_export(profile_id: int) -> dict[str, Any]:
return await hub.export_mapping_profile(profile_id)
@router.post("/api/hockey/admin/vmix-mapping/profiles/{profile_id}/copy-to-device", dependencies=admin)
async def mapping_profile_copy_to_device(profile_id: int, payload: MappingProfileCopyPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.copy_mapping_profile_to_device(profile_id, payload, user)
@router.post("/api/hockey/admin/vmix-mapping/import", dependencies=admin)
async def mapping_profile_import(payload: MappingPortableImportPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.import_mapping_profile_to_device(payload, user)
@router.put("/api/hockey/admin/vmix-mapping/profiles/{profile_id}", dependencies=admin)
async def mapping_profile_update(profile_id: int, payload: MappingProfileUpdatePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.update_mapping_profile(profile_id, payload, user)
@router.post("/api/hockey/admin/vmix-mapping/profiles/{profile_id}/inventory", dependencies=admin)
async def mapping_profile_inventory(profile_id: int, payload: MappingInventoryRefreshPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.refresh_mapping_inventory(profile_id, payload.device_id, user)
@router.put("/api/hockey/admin/vmix-mapping/profiles/{profile_id}/fields", dependencies=admin)
async def mapping_profile_fields(profile_id: int, payload: MappingFieldsReplacePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return await hub.replace_mapping_fields(profile_id, payload, user)
@router.delete("/api/hockey/admin/vmix-mapping/profiles/{profile_id}", dependencies=admin)
async def mapping_profile_delete(profile_id: int) -> dict[str, Any]:
return await hub.delete_mapping_profile(profile_id)
@router.get("/api/hockey/admin/mapping-context/variables", dependencies=admin)
async def mapping_context_variables() -> dict[str, Any]:
return hub.mapping_data.list_variables()
@router.post("/api/hockey/admin/mapping-context/variables", dependencies=admin)
async def mapping_context_variable_create(payload: ContextVariablePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.create_variable(payload, user)
@router.put("/api/hockey/admin/mapping-context/variables/{variable_id}", dependencies=admin)
async def mapping_context_variable_update(variable_id: int, payload: ContextVariablePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.update_variable(variable_id, payload, user)
@router.delete("/api/hockey/admin/mapping-context/variables/{variable_id}", dependencies=admin)
async def mapping_context_variable_delete(variable_id: int) -> dict[str, Any]:
return hub.mapping_data.delete_variable(variable_id)
@router.get("/api/hockey/admin/mapping-data/sources", dependencies=admin)
async def mapping_data_sources() -> dict[str, Any]:
return hub.mapping_data.list_sources()
@router.post("/api/hockey/admin/mapping-data/sources", dependencies=admin)
async def mapping_data_source_create(payload: SqlDataSourcePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.create_source(payload, user)
@router.put("/api/hockey/admin/mapping-data/sources/{source_id}", dependencies=admin)
async def mapping_data_source_update(source_id: int, payload: SqlDataSourcePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.update_source(source_id, payload, user)
@router.delete("/api/hockey/admin/mapping-data/sources/{source_id}", dependencies=admin)
async def mapping_data_source_delete(source_id: int) -> dict[str, Any]:
return hub.mapping_data.delete_source(source_id)
@router.post("/api/hockey/admin/mapping-data/preview-sql", dependencies=admin)
async def mapping_data_preview_sql(payload: MappingSqlPreviewPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.preview_sql(payload.sql_text, user, payload.context)
@router.post("/api/hockey/admin/mapping-data/sources/{source_id}/preview", dependencies=admin)
async def mapping_data_source_preview(source_id: int, payload: MappingDataContextPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.preview_source(source_id, user, payload.context)
@router.post("/api/hockey/admin/mapping-data/catalog", dependencies=admin)
async def mapping_data_catalog(payload: MappingDataContextPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
return hub.mapping_data.data_catalog(user, payload.context)
@router.post("/api/hockey/context/{key}")
async def hockey_context_set(key: str, payload: ContextValuePayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
result = hub.mapping_data.set_context_value(key, payload.value, user, payload.context)
result["mapping_apply"] = await hub.apply_mapping_for_user(user, reason=f"context_changed:{key}")
return result
@router.post("/api/hockey/context/batch")
async def hockey_context_batch(payload: ContextValuesPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
if len(payload.values) > 32:
raise HTTPException(status_code=422, detail="Можно изменить не более 32 переменных за один запрос")
results = []
for key, value in payload.values.items():
results.append(hub.mapping_data.set_context_value(str(key), value, user, payload.context))
context = payload.context if isinstance(payload.context, dict) else {}
mapping_apply = await hub.apply_mapping_for_user(
user,
reason="context_batch_changed",
device_id=str(context.get("device_id") or ""),
session_token=str(context.get("session_token") or ""),
only_changed=True,
)
return {"ok": True, "items": results, "mapping_apply": mapping_apply}
@router.post("/api/hockey/context/resolve")
async def hockey_context_resolve(payload: MappingDataContextPayload, user: HockeyUser = Depends(auth_dependency)) -> dict[str, Any]:
context, variables = hub.mapping_data.resolve_context(user, payload.context)
return {"context": context, "variables": variables}
@router.websocket("/ws/hockey-agent")
async def agent_socket(websocket: WebSocket) -> None:
await websocket.accept()
device_id = ""
try:
raw = await asyncio.wait_for(websocket.receive_json(), timeout=12.0)
hello = AgentHelloPayload.model_validate(raw)
if hello.type != "hello" or hello.protocol != AGENT_PROTOCOL_VERSION:
await websocket.close(code=4400, reason="Unsupported agent protocol")
return
hello_data = hello.model_dump()
registered = await hub.register(websocket, hello_data)
device_id = registered["device_id"]
await websocket.send_json(
{
"type": "hello.ok",
"protocol": AGENT_PROTOCOL_VERSION,
"device": registered,
"server_time": _utcnow().isoformat(),
}
)
await hub.reconcile_device(device_id)
while True:
message = await websocket.receive_json()
if not isinstance(message, dict):
continue
message_type = str(message.get("type") or "")
if message_type in {"heartbeat", "vmix.status", "command.ack", "command.batch.ack", "match.accepted", "vmix.inventory"}:
await hub.receive_status(device_id, message)
if message_type == "vmix.inventory":
await hub.receive_inventory(device_id, message)
if message_type in {"command.ack", "command.batch.ack"}:
await hub.receive_command_ack(device_id, message)
if message_type == "heartbeat":
await websocket.send_json({"type": "heartbeat.ack", "server_time": _utcnow().isoformat()})
except asyncio.TimeoutError:
await websocket.close(code=4408, reason="Agent hello timeout")
except (ValueError, PermissionError) as error:
await websocket.send_json({"type": "hello.error", "detail": str(error)})
await websocket.close(code=4403, reason="Agent registration rejected")
except WebSocketDisconnect:
pass
except Exception as error:
try:
await websocket.send_json({"type": "server.error", "detail": str(error)[:500]})
except Exception:
pass
finally:
if device_id:
await hub.unregister(device_id, websocket)
return router