отправка данных для mapping с агентом

hockey_data/agent_bridge.py

@@ -24,6 +24,11 @@ AGENT_PROTOCOL_VERSION = 1
 AGENT_ONLINE_WINDOW_SECONDS = 35
 _DEVICE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{6,128}$")
 
+# BUILD97: Mapping can easily exceed 300 links. Keep WebSocket/vMix packets small
+# and predictable instead of sending the whole profile in one giant batch.
+MAPPING_BATCH_MAX_COMMANDS = 40
+MAPPING_BATCH_MAX_BYTES = 48 * 1024
+
 
 def _utcnow() -> datetime:
     return datetime.utcnow()
@@ -241,6 +246,9 @@ class VmixAgentHub:
         # BUILD90: one FIFO lock per Agent. Shortcut batches, Mapping refreshes and
         # timer control commands can no longer interleave on the same vMix instance.
         self._vmix_send_locks: dict[str, asyncio.Lock] = {}
+        # BUILD97: serialize Mapping applications per Agent while still letting
+        # runtime/timer commands use the normal vMix FIFO between Mapping chunks.
+        self._mapping_apply_locks: dict[str, asyncio.Lock] = {}
         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] = {}
@@ -673,6 +681,173 @@ class VmixAgentHub:
             self._vmix_send_locks[device_id] = lock
         return lock
 
+    def _device_mapping_apply_lock(self, device_id: str) -> asyncio.Lock:
+        lock = self._mapping_apply_locks.get(device_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._mapping_apply_locks[device_id] = lock
+        return lock
+
+    @staticmethod
+    def _mapping_batch_chunks(entries: list[dict[str, Any]]) -> list[list[tuple[int, dict[str, Any]]]]:
+        """Group Mapping commands by vMix Input, then cap packet count and bytes.
+
+        The returned tuples keep the original index so ACKs can be mapped back to
+        the original Mapping links even though commands are grouped by Input.
+        """
+        groups: dict[str, list[tuple[int, dict[str, Any]]]] = {}
+        for index, entry in enumerate(entries):
+            command = entry.get("command") if isinstance(entry, dict) else None
+            input_ref = str((command or {}).get("Input") or "").strip()
+            groups.setdefault(input_ref, []).append((index, entry))
+
+        chunks: list[list[tuple[int, dict[str, Any]]]] = []
+        for group in groups.values():
+            current: list[tuple[int, dict[str, Any]]] = []
+            current_bytes = 0
+            for indexed_entry in group:
+                command = indexed_entry[1].get("command") or {}
+                try:
+                    command_bytes = len(json.dumps(command, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) + 96
+                except Exception:
+                    command_bytes = 512
+                if current and (
+                    len(current) >= MAPPING_BATCH_MAX_COMMANDS
+                    or current_bytes + command_bytes > MAPPING_BATCH_MAX_BYTES
+                ):
+                    chunks.append(current)
+                    current = []
+                    current_bytes = 0
+                current.append(indexed_entry)
+                current_bytes += command_bytes
+            if current:
+                chunks.append(current)
+        return chunks
+
+    async def _send_mapping_entries_resilient(
+        self,
+        device_id: str,
+        *,
+        assignment_id: str,
+        match_id: str,
+        entries: list[dict[str, Any]],
+        use_batch: bool,
+    ) -> dict[str, Any]:
+        """Deliver Mapping commands reliably, with chunking and adaptive fallback.
+
+        Large mappings are grouped by Input and sent in bounded batches. A timed
+        out/failed batch is recursively split; a single failed command falls back
+        to the legacy one-command transport. Mapping commands are idempotent
+        (SetText/SetImage/SetColor/visibility), so this retry strategy is safe.
+        """
+        outcomes: list[dict[str, Any] | None] = [None] * len(entries)
+        stats = {
+            "chunks_total": 0,
+            "chunks_ok": 0,
+            "chunks_failed": 0,
+            "packets_sent": 0,
+            "retries": 0,
+            "fallback_commands": 0,
+            "input_groups": len({
+                str(((entry.get("command") or {}).get("Input") or "")).strip()
+                for entry in entries
+            }),
+        }
+
+        async def send_single(
+            index: int, entry: dict[str, Any], previous_reason: str = "", *, count_fallback: bool = True
+        ) -> None:
+            if count_fallback:
+                stats["fallback_commands"] += 1
+            try:
+                ack = await self.send_vmix_command(
+                    device_id, assignment_id=assignment_id, match_id=match_id,
+                    command=entry.get("command") or {}, timeout=5.0,
+                )
+                if bool(ack.get("ok")):
+                    outcomes[index] = dict(ack)
+                else:
+                    outcomes[index] = {
+                        "ok": False,
+                        "reason": str(ack.get("reason") or ack.get("error") or previous_reason or "vmix_error"),
+                    }
+            except Exception as error:
+                outcomes[index] = {
+                    "ok": False,
+                    "reason": str(getattr(error, "detail", error) or previous_reason or "vmix_error")[:300],
+                }
+
+        async def send_chunk(chunk: list[tuple[int, dict[str, Any]]]) -> None:
+            if not chunk:
+                return
+            commands = [entry.get("command") or {} for _, entry in chunk]
+            stats["packets_sent"] += 1
+            try:
+                ack = await self.send_vmix_batch(
+                    device_id, assignment_id=assignment_id, match_id=match_id, commands=commands,
+                    timeout=max(6.0, min(10.0, 4.0 + len(commands) * 0.06)),
+                )
+                ack_results = ack.get("results") if isinstance(ack.get("results"), list) else None
+                if ack_results is None:
+                    if bool(ack.get("ok")):
+                        for index, _entry in chunk:
+                            outcomes[index] = {"ok": True}
+                        return
+                    raise RuntimeError(str(ack.get("reason") or ack.get("error") or "vmix_batch_error"))
+
+                failed: list[tuple[int, dict[str, Any], str]] = []
+                for position, (index, entry) in enumerate(chunk):
+                    item_ack = ack_results[position] if position < len(ack_results) and isinstance(ack_results[position], dict) else {}
+                    if bool(item_ack.get("ok")):
+                        outcomes[index] = dict(item_ack)
+                    else:
+                        failed.append((
+                            index, entry,
+                            str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"),
+                        ))
+                # A batch ACK can contain isolated command failures. Retry only
+                # those commands once through the single-command transport.
+                for index, entry, reason in failed:
+                    stats["retries"] += 1
+                    await send_single(index, entry, reason)
+                return
+            except Exception as error:
+                reason = str(getattr(error, "detail", error))[:300]
+                if len(chunk) > 1:
+                    stats["retries"] += 1
+                    middle = max(1, len(chunk) // 2)
+                    await send_chunk(chunk[:middle])
+                    await send_chunk(chunk[middle:])
+                    return
+                index, entry = chunk[0]
+                stats["retries"] += 1
+                await send_single(index, entry, reason)
+
+        async with self._device_mapping_apply_lock(device_id):
+            if use_batch:
+                chunks = self._mapping_batch_chunks(entries)
+                stats["chunks_total"] = len(chunks)
+                for chunk in chunks:
+                    await send_chunk(chunk)
+                    failed_now = sum(1 for index, _entry in chunk if outcomes[index] is None or not bool(outcomes[index].get("ok")))
+                    if failed_now == 0:
+                        stats["chunks_ok"] += 1
+                    else:
+                        stats["chunks_failed"] += 1
+            else:
+                stats["chunks_total"] = len(entries)
+                for index, entry in enumerate(entries):
+                    await send_single(index, entry, count_fallback=False)
+                    if outcomes[index] is not None and bool(outcomes[index].get("ok")):
+                        stats["chunks_ok"] += 1
+                    else:
+                        stats["chunks_failed"] += 1
+
+        return {
+            "results": [item if isinstance(item, dict) else {"ok": False, "reason": "mapping_transport_no_ack"} for item in outcomes],
+            **stats,
+        }
+
     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:
@@ -1037,55 +1212,40 @@ class VmixAgentHub:
             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]})
+        use_batch = self._agent_supports_batch(agent_version)
+        result["transport"] = "chunked_batch" if use_batch else "legacy"
+        transport = await self._send_mapping_entries_resilient(
+            device_id,
+            assignment_id=assignment_id,
+            match_id=match_id,
+            entries=pending_entries,
+            use_batch=use_batch,
+        )
+        result["batch_chunks_total"] = int(transport.get("chunks_total") or 0)
+        result["batch_chunks_applied"] = int(transport.get("chunks_ok") or 0)
+        result["batch_chunks_failed"] = int(transport.get("chunks_failed") or 0)
+        result["batch_packets_sent"] = int(transport.get("packets_sent") or 0)
+        result["batch_retries"] = int(transport.get("retries") or 0)
+        result["fallback_commands"] = int(transport.get("fallback_commands") or 0)
+        result["input_groups"] = int(transport.get("input_groups") or 0)
+        result["batch_max_commands"] = MAPPING_BATCH_MAX_COMMANDS
+
+        ack_results = transport.get("results") if isinstance(transport.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"],
+                    "input": str((entry.get("command") or {}).get("Input") or ""),
+                    "field": str((entry.get("command") or {}).get("SelectedName") or ""),
+                    "reason": str(item_ack.get("reason") or item_ack.get("error") or "mapping_transport_error"),
+                })
         result["ok"] = not result["errors"]
         return result
 
@@ -1519,41 +1679,30 @@ class VmixAgentHub:
                 "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]})
+        if prepared:
+            use_batch = self._agent_supports_batch(agent_version)
+            result["transport"] = "chunked_batch" if use_batch else "legacy"
+            transport = await self._send_mapping_entries_resilient(
+                device_id, assignment_id=assignment_id, match_id=match_id,
+                entries=prepared, use_batch=use_batch,
+            )
+            result["batch_chunks_total"] = int(transport.get("chunks_total") or 0)
+            result["batch_chunks_applied"] = int(transport.get("chunks_ok") or 0)
+            result["batch_chunks_failed"] = int(transport.get("chunks_failed") or 0)
+            result["batch_packets_sent"] = int(transport.get("packets_sent") or 0)
+            result["batch_retries"] = int(transport.get("retries") or 0)
+            result["fallback_commands"] = int(transport.get("fallback_commands") or 0)
+            result["input_groups"] = int(transport.get("input_groups") or 0)
+            ack_results = transport.get("results") if isinstance(transport.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 "mapping_transport_error"),
+                    })
         result["ok"] = not result["errors"]
         return result
 
@@ -3815,7 +3964,10 @@ def create_hockey_agent_router(
             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"):
+        # BUILD97: field-level vMix errors are returned as diagnostics instead of
+        # being collapsed into a misleading HTTP 409 (e.g. "manual_apply").
+        fatal_reasons = {"device_not_found", "device_not_paired", "no_match_assignment", "no_project_inventory", "vmix_not_connected", "mapping_missing"}
+        if not result.get("ok") and str(result.get("reason") or "") in fatal_reasons:
             raise HTTPException(status_code=409, detail=str(result.get("reason")))
         return result
 
@@ -3854,7 +4006,8 @@ def create_hockey_agent_router(
     @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"):
+        fatal_reasons = {"device_not_found", "device_not_paired", "no_match_assignment", "no_project_inventory", "vmix_not_connected", "mapping_missing"}
+        if not result.get("ok") and str(result.get("reason") or "") in fatal_reasons:
             raise HTTPException(status_code=409, detail=str(result.get("reason")))
         return result