From 3e1c568d04a8634ca106c67c04a8eca4c1ce9c75 Mon Sep 17 00:00:00 2001
From: chakcy <947105045@qq.com>
Date: Tue, 11 Aug 2026 14:47:19 +0800
Subject: [PATCH] =?UTF-8?q?modify:=20=E6=8E=A5=E5=8F=A3=E8=BD=AC=E5=8F=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 aibox_sophon_adapter/edge_app_controller.py | 24 +++++++-------
 aibox_sophon_adapter/routes.py              | 36 +++++++++++++--------
 2 files changed, 34 insertions(+), 26 deletions(-)

diff --git a/aibox_sophon_adapter/edge_app_controller.py b/aibox_sophon_adapter/edge_app_controller.py
index 976fbba..74adcc7 100644
--- a/aibox_sophon_adapter/edge_app_controller.py
+++ b/aibox_sophon_adapter/edge_app_controller.py
@@ -56,12 +56,12 @@ async def edge_app_controller(payload: dict, topic: str) -> dict | None:
     )
 
     if command is None:
-        return _reply(-1, "缺少 command 字段")
+        return None
 
     handler = _COMMANDS.get(str(command))
     if handler is None:
         logger.warning(f"[EdgeAppController] 未知命令: {command}")
-        return _reply(-2, f"未知命令: {command}")
+        return None
 
     try:
         data = await handler(payload)
@@ -73,7 +73,7 @@ async def edge_app_controller(payload: dict, topic: str) -> dict | None:
         return data
     except Exception as e:
         logger.error(f"[EdgeAppController] 命令 {command} 执行失败: {e}")
-        return _reply(-3, f"命令执行异常: {e}")
+        return None
 
 
 # ============================================================
@@ -124,15 +124,15 @@ async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
 
     board_id = payload.get("BoardId")
     if not board_id:
-        return _reply(-4, "缺少 BoardId 参数")
+        return None
 
     print(f"[alg_media_fetch]Processing for board: {board_id}")
     print(f"[alg_media_fetch]Expected payload: {payload}")
 
     if db is None:
-        return _reply(-5, "数据库未初始化")
+        return None
     if internal_mqtt is None:
-        return _reply(-5, "internal_mqtt 未初始化")
+        return None
 
     async with db.get_session() as session:
         from .models.orm import BoardHeartbeat
@@ -144,7 +144,7 @@ async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
 
         if record is None:
             print(f"[alg_media_fetch]Board {board_id} not found in database.")
-            return _reply(-4, f"设备 {board_id} 不在数据库中")
+            return None
 
         # 内部请求-响应主题由 [workers.config.internal_mqtt] 配置，与盒子侧约定一致
         # （盒子只设根主题 box，子路径固定 aibox_sophon/...，回复主题固定为 ..._reply）。
@@ -161,7 +161,7 @@ async def _cmd_alg_media_fetch(payload: dict) -> dict | None:
         print(f"[alg_media_fetch]Response from internal_mqtt: {resp}")
 
         if resp is None:
-            return _reply(-5, "获取算法媒体信息失败（内部请求无响应）")
+            return None
         return resp
 
 async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
@@ -180,15 +180,15 @@ async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
 
     board_id = payload.get("BoardId")
     if not board_id:
-        return _reply(-4, "缺少 BoardId 参数")
+        return None
 
     print(f"[alg_task_fetch]Processing for board: {board_id}")
     print(f"[alg_task_fetch]Expected payload: {payload}")
 
     if db is None:
-        return _reply(-5, "数据库未初始化")
+        return None
     if internal_mqtt is None:
-        return _reply(-5, "internal_mqtt 未初始化")
+        return None
 
     async with db.get_session() as session:
         from .models.orm import BoardHeartbeat
@@ -200,7 +200,7 @@ async def _cmd_alg_task_fetch(payload: dict) -> dict | None:
 
         if record is None:
             print(f"[alg_task_fetch]Board {board_id} not found in database.")
-            return _reply(-4, f"设备 {board_id} 不在数据库中")
+            return None
 
         
         resp = await internal_mqtt.request(
diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py
index 662f081..936330b 100644
--- a/aibox_sophon_adapter/routes.py
+++ b/aibox_sophon_adapter/routes.py
@@ -38,31 +38,37 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/aibox_sophon_adapter", tags=["aibox_sophon_adapter"])
 
 
-class _FileStream(httpx.AsyncByteStream):
+class _FileStream:
     """
-    异步文件流 — 用于流式上传视频，避免一次性将整个文件读入内存。
+    文件流 — 用于流式上传视频，避免一次性将整个文件读入内存。
 
-    以 1MB 分块异步读取文件，供 httpx multipart 上传使用。
+    注意: httpx 0.28 的 multipart 上传（FileField）要求文件对象提供同步的
+    read()/seek() 方法，httpx 内部以 64KB 分块同步读取；AsyncByteStream 只能
+    作为整个请求体 content= 使用，无法与 multipart 字段共存。因此这里提供
+    同步文件接口，配合同步上下文管理器 with 使用（本地磁盘小分块读取，
+    内存占用依然有界）。
     """
 
-    _CHUNK_SIZE = 1024 * 1024
-
     def __init__(self, file_path: str):
         self._file_path = file_path
         self._file = None
 
-    async def __aenter__(self) -> "_FileStream":
-        self._file = await aiofiles.open(self._file_path, "rb")
+    def __enter__(self) -> "_FileStream":
+        self._file = open(self._file_path, "rb")
         return self
 
-    async def __aexit__(self, exc_type, exc, tb) -> None:
+    def __exit__(self, exc_type, exc, tb) -> None:
         if self._file is not None:
-            await self._file.close()
+            self._file.close()
+            self._file = None
 
-    async def __aiter__(self):
-        assert self._file is not None, "必须在 async with 上下文内使用"
-        while chunk := await self._file.read(self._CHUNK_SIZE):
-            yield chunk
+    def read(self, size: int = -1) -> bytes:
+        assert self._file is not None, "必须在 with 上下文内使用"
+        return self._file.read(size)
+
+    def seek(self, offset: int, whence: int = 0) -> int:
+        assert self._file is not None, "必须在 with 上下文内使用"
+        return self._file.seek(offset, whence)
 
 
 # ============================================================
@@ -206,7 +212,9 @@ async def _upload_video_to_external(
 
     video_path = os.path.join(video_dir_path, video_files[0])
     try:
-        async with _FileStream(video_path) as video_stream:
+        # httpx 0.28 的 multipart 以同步 read() 流式读取文件（64KB 分块），
+        # 因此这里用同步 with 打开文件流
+        with _FileStream(video_path) as video_stream:
             resp = await http_client.post(
                 video_url,
                 data={