fix(fam-edge): schema 对齐新架构 - parse_vlm_json 必填改 global_summary/events（兼容旧 frame_details/entities_json 转换）；NVIDIA PUT content-type 值改为 video/mp4 与 POST 一致（预签名校验）

fam-edge/src/fam_edge/ai_orchestrator/json_parser.py

@@ -48,55 +48,60 @@ def parse_vlm_json(raw: str) -> dict:
 def validate_schema(data: dict) -> dict:
     """Schema 校验 + 脏数据清洗
 
+    新架构（整视频分析）schema: global_summary / events / people_mentioned
+    兼容旧结构: 模型若输出 frame_details / entities_json 也接受并转换。
     注意: compute_provider 是内部记账字段（orchestrator 在解析成功后填充），
     提示词从不要求模型输出它，故不做必填校验。
     """
-    required = ["global_summary", "entities_json", "frame_details"]
+    required = ["global_summary", "events"]
     for k in required:
         if k not in data:
+            if k == "events" and "frame_details" in data:
+                continue
             raise VLMOutputInvalidError(f"缺失字段: {k}")
 
-    # entities_json 结构校验
-    if not isinstance(data["entities_json"], list):
-        raise VLMOutputInvalidError("entities_json 必须为数组")
-
-    cleaned_entities = []
-    for ent in data["entities_json"]:
-        if not isinstance(ent, dict):
-            continue
-        if "person" not in ent or "action" not in ent:
-            raise VLMOutputInvalidError("entity 缺少 person 或 action 字段")
-        cleaned_entities.append({
-            "person": str(ent["person"]),
-            "action": str(ent["action"]),
-            "clothing": str(ent.get("clothing", ""))
+    # events 结构校验（兼容 frame_details 旧结构转换）
+    events = data.get("events")
+    if events is None and "frame_details" in data:
+        events = []
+        for f in data["frame_details"]:
+            events.append({
+                "timestamp": f.get("frame_timestamp", ""),
+                "description": f.get("action", ""),
+                "people": [f.get("person", "")] if f.get("person") else [],
+                "is_attention_event": bool(f.get("is_attention_event", False)),
             })
-    data["entities_json"] = cleaned_entities
+    if not isinstance(events, list):
+        raise VLMOutputInvalidError("events 必须为数组")
 
-    # frame_details 结构校验
-    if not isinstance(data["frame_details"], list):
-        raise VLMOutputInvalidError("frame_details 必须为数组")
-
-    cleaned_frames = []
-    for frame in data["frame_details"]:
-        if not isinstance(frame, dict):
+    cleaned = []
+    for ev in events:
+        if not isinstance(ev, dict):
             continue
-        for k in ["frame_index", "frame_timestamp", "person", "action", "source_providers"]:
-            if k not in frame:
-                raise VLMOutputInvalidError(f"frame_details 缺少字段: {k}")
-        sp = frame["source_providers"]
-        if not isinstance(sp, list) or len(sp) == 0:
-            raise VLMOutputInvalidError("frame_details.source_providers 必须为非空数组")
-        cleaned_frames.append({
-            "frame_index": int(frame["frame_index"]),
-            "frame_timestamp": str(frame["frame_timestamp"]),
-            "person": str(frame["person"]),
-            "action": str(frame["action"]),
-            "clothing": str(frame.get("clothing", "")),
-            "is_attention_event": bool(frame.get("is_attention_event", False)),
-            "source_providers": [str(p) for p in sp]
+        if "timestamp" not in ev or "description" not in ev:
+            raise VLMOutputInvalidError("event 缺少 timestamp 或 description 字段")
+        people = ev.get("people", [])
+        if isinstance(people, str):
+            people = [people]
+        if not isinstance(people, list):
+            people = []
+        cleaned.append({
+            "timestamp": str(ev["timestamp"]),
+            "description": str(ev["description"]),
+            "people": [str(p) for p in people if p],
+            "is_attention_event": bool(ev.get("is_attention_event", False)),
         })
-    data["frame_details"] = cleaned_frames
+    data["events"] = cleaned
+
+    # people_mentioned（兼容 entities_json 旧结构）
+    people = data.get("people_mentioned")
+    if people is None and "entities_json" in data:
+        people = data["entities_json"]
+    if isinstance(people, list) and people and isinstance(people[0], dict):
+        people = [p.get("person", "") for p in people]
+    if not isinstance(people, list):
+        people = []
+    data["people_mentioned"] = [str(p) for p in people if p]
 
     # compute_provider: 模型通常不输出（提示词不要求），缺失/非法时归一为空数组，
     # 由 orchestrator.format_cloud_result 用实际成功的 provider 覆盖填充

fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py

@@ -74,6 +74,7 @@ class NvidiaVisionAdapter(BaseModelAdapter):
 
     def _upload_asset(self, video_path: str) -> Optional[str]:
         """用 NVIDIA Assets API 上传大视频文件，返回 asset_id 供 video_url 引用。"""
+        content_type = "video/mp4"
         upload_timeout = max(self.timeout, 900)
         try:
             r = requests.post(
@@ -82,7 +83,7 @@ class NvidiaVisionAdapter(BaseModelAdapter):
                     "Authorization": f"Bearer {self.api_key}",
                     "Content-Type": "application/json",
                 },
-                json={"contentType": "video/mp4", "description": "fam-edge video asset"},
+                json={"contentType": content_type, "description": "fam-edge video asset"},
                 timeout=60,
             )
         except Exception as e:
@@ -106,8 +107,9 @@ class NvidiaVisionAdapter(BaseModelAdapter):
                 up = requests.put(
                     upload_url,
                     data=f,
-                    # 注意：必须用小写 header 名，预签名 S3 URL 的签名覆盖小写 content-type
-                    headers={"content-type": "application/octet-stream",
+                    # 必须全小写 header 名且 content-type 值与 POST 的 contentType 一致：
+                    # 预签名 S3 URL 签名覆盖这两个值，不一致会 SignatureDoesNotMatch
+                    headers={"content-type": content_type,
                              "x-amz-meta-nvcf-asset-description": "fam-edge video asset"},
                     timeout=upload_timeout,
                 )