feat(v3): 人物模块重构 - 大模型结构化特征值替代 OpenCV 帧定位。prompt 增加 person_appearances(uid+7特征+action)并重写合并 prompt(稳定特征优先比对)；gemini 透传特征字段；oracle_db events/people 加 person_appearances_json/features_json/display_uid + _merge_features；video_processor 去 cv2 改 ffprobe 校验、_store_result 聚合 uid 特征落 people、删缩略图/事件截图；person_service 聚合特征后基于特征文本 LLM 合并；api_gateway 删 3 个图接口；NAS 镜像+DDL 加新字段；fam-ui 特征卡替代头像、事件时间线人物特征块

fam-edge/config/config.yaml

@@ -43,7 +43,7 @@ video_processing:
   timeout_multiplier: 2        # 模型消费超时倍数：在 models[i].timeout 原值上 ×2（大视频上传+分析耗时）
   max_retries: 10              # 单视频失败最大重试次数（配额/过载等瞬时故障给足重试机会）
   retry_interval_sec: 3600     # 失败重试最小间隔：距上次失败 ≥1h 才重新入队，等配额恢复
-  file_validate: true          # 登记入队前用 OpenCV 校验文件可解码；失败标记 invalid 不入队
+  file_validate: true          # 登记入队前用 ffprobe 校验文件可解码；失败标记 invalid 不入队
   stable_window_sec: 60        # 文件 mtime 稳定窗口：写入中（rclone 同步未完成）的文件跳过本轮
   # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed
   vision_order: ["gemini", "nvidia"]

fam-edge/requirements.txt

@@ -2,9 +2,8 @@ flask>=2.0.0
 gunicorn>=20.0.0
 requests>=2.28.0
 PyYAML>=6.0
-opencv-python-headless>=4.5.0
-numpy>=1.21.0
 # google-generativeai 和 openai 为可选依赖（代码用 requests 直接调 REST API）
 # 如需 SDK 方式调用，取消注释并在 Python 3.9+ 环境安装:
 # google-generativeai>=0.5.0
 # openai>=1.10.0
+# 视频文件校验依赖系统 ffprobe（ffmpeg 套件自带，无需 Python 包）

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

@@ -6,8 +6,10 @@ Prompt 模板 - 集中管理，避免 gemini/nvidia 适配器各维护一份导
   2. timestamp 统一"视频内相对时间 HH:MM:SS"，消除绝对/相对歧义
   3. 人物命名: 已知成员用真名，未知用"人物A/B/C"本视频内临时编号，
      并强制 people_mentioned = events 内出现人物去重后的集合（下游合并依赖）
-  4. 输出硬约束: 首字符必须是 {，禁止思考过程/markdown/解释
-  5. 边界情况: 无人/空视频/看不清 -> 空 events + summary 说明，不凑数
+  4. 人物特征: 每个人物在每个 event 里输出 person_appearances，含 uid + 结构化特征
+     文本（性别/年龄段/身形/发型/衣着/面部/辨识点）；下游靠特征值跨视频绑定同一身份
+  5. 输出硬约束: 首字符必须是 {，禁止思考过程/markdown/解释
+  6. 边界情况: 无人/空视频/看不清 -> 空 events + summary 说明，不凑数
 """
 from typing import Optional
 
@@ -44,6 +46,21 @@ def build_video_prompt(known_members: str, event_start_time: str,
       "timestamp": "HH:MM:SS",
       "description": "该时刻画面的详细描述：人物身份、动作细节、位置移动、交互对象、姿态/手势/朝向、手中物品、周围环境",
       "people": ["人物标识"],
+      "person_appearances": [
+        {{
+          "uid": "人物A",
+          "features": {{
+            "gender": "男",
+            "age_band": "中年",
+            "build": "瘦高",
+            "hair": "短发黑色",
+            "clothing": "红色卫衣+深色长裤",
+            "face": "蓄须",
+            "distinguishing": "左手戴手表"
+          }},
+          "action": "走向沙发坐下"
+        }}
+      ],
       "is_attention_event": false
     }}
   ],
@@ -74,12 +91,27 @@ def build_video_prompt(known_members: str, event_start_time: str,
 4. people: 该时刻出现的人物标识。已知成员用真名，未知人物用「人物A」「人物B」
    本视频内连续编号（同一人保持同一编号）。只填标识本身，不要带括号注释
    （如只写"人物A"，不要写"人物A（别名/标识：人物B）"）。
-5. people_mentioned: 必须等于 events 中所有 people 字段出现过的标识去重后的集合。
+5. person_appearances: 该时刻出现的每个人物的结构化特征 + 动作。必填字段说明:
+   - uid: 与 people 数组里的标识完全一致（同一人同一 uid）。
+   - features: 客观可见特征，必须包含以下 7 个子字段，看不清的写 "unknown"，绝不留空:
+     * gender: 性别（男/女/unknown）
+     * age_band: 年龄段（幼儿/儿童/少年/青年/中年/老年/unknown）
+     * build: 身材（如 瘦高/中等/偏胖/壮实/矮小/unknown）
+     * hair: 发型与颜色（如 短发黑色/长发棕色/秃顶/unknown）
+     * clothing: 当下衣着（如 红色卫衣+深色长裤/白色T恤+牛仔裤/unknown）
+     * face: 面部特征（如 蓄须/戴眼镜/圆脸/unknown）
+     * distinguishing: 辨识点（如 左手戴手表/右脸有痣/跛行/无）
+   - action: 该人物在本时刻的动作（与 description 里该人物动作一致，单独抽出便于检索）。
+   特征硬约束:
+     * 客观描述可见特征，不猜测、不推断、不编造（看不清的字段写 unknown，不要靠常识猜性别/年龄）。
+     * 同一 uid 在视频多个 event 出现时，features 字段保持一致（衣着变了再如实更新 clothing，
+       但 gender/age_band/build/face 必须稳定）。
+6. people_mentioned: 必须等于 events 中所有 people 字段出现过的标识去重后的集合。
    一致性强制：events 里出现的标识必须都在 people_mentioned 里，反之亦然。
-6. is_attention_event: 跌倒、危险动作、异常哭闹、陌生人闯入、身体不适等需关注事件
+7. is_attention_event: 跌倒、危险动作、异常哭闹、陌生人闯入、身体不适等需关注事件
    填 true，否则 false。关注事件的 event 仍按上述密度规则抽取，但 description 须明确
    说明"异常"点（如"张三在 00:01:15 跌坐在地，身体向右侧倾，双手撑地"）。
-7. global_summary: 客观描述，不猜测、不想象、不编造。须包含：谁在画面中、主要活动、
+8. global_summary: 客观描述，不猜测、不想象、不编造。须包含：谁在画面中、主要活动、
    是否有关注事件、时段大致结构。
 
 【已知家庭成员】
@@ -124,9 +156,10 @@ def build_person_merge_prompt(unnamed_lines: str) -> str:
     """构建人物合并 prompt（person_service._llm_merge 用）。
 
     Args:
-      unnamed_lines: 待合并人物的场景描述，每行 "- label：出现场景 ..."
+      unnamed_lines: 待合并人物的特征文本，每行 "- uid：特征=...; 特征=..."
+        （features_json 渲染而来的文本，供 LLM 判断是否同一人）
     """
-    return f"""你是家庭监控人物汇总助手。下面是若干人物标识及其出现场景描述。
+    return f"""你是家庭监控人物汇总助手。下面是若干人物标识及其结构化特征描述。
 请判断哪些标识指向同一个人，并为每个人输出一个稳定的规范名。
 
 【输出格式】
@@ -140,5 +173,12 @@ def build_person_merge_prompt(unnamed_lines: str) -> str:
 3. 无法判断是否同一人的，保守不合并（各保留独立规范名）。
 4. 规范名必须在输出中唯一：多个原标识可映射到同一规范名，但同一规范名只指向一个人。
 
+【判断依据】
+- 优先比对 gender / age_band / build / face 这四个稳定特征（不会在同一天内变化）。
+- hair / clothing 会变化，仅作辅助；仅靠 clothing 相同不足以合并，仅靠 hair 不同不足以拆分。
+- distinguishing 辨识点（如手表/痣/跛行）是强证据，一致时倾向合并。
+- 任一稳定特征明确冲突（如一个写"男"一个写"女"），绝不合并。
+- 任一关键特征缺失（"unknown"）时，其他特征一致性需更强才合并；拿不准保守不合并。
+
 【待处理人物】
 {unnamed_lines}"""

fam-edge/src/fam_edge/api_gateway/api_gateway.py

@@ -1,18 +1,23 @@
 """
-API-Gateway - Flask 蓝图（新架构 v2）
+API-Gateway - Flask 蓝图（新架构 v3）
 
 端点:
   GET  /api/oracle/sync            NAS 每 30 分钟拉取增量（since + token 校验）
   POST /api/oracle/people/correct  NAS 推送手动命名校正（label -> canonical_name）
   POST /api/edge/chat/ask         智能问答编排（Gemini -> NVIDIA -> Ollama）
+  GET  /api/oracle/activity        实时服务状态 + 最近活动流
   GET  /health                    健康检查
 
+已移除（v3 去除帧图/avatar 依赖，改用大模型特征值）:
+  /api/oracle/video/<id>/thumb, /api/oracle/event/<id>/thumb,
+  /api/oracle/person/avatar —— 不再生成 jpg，UI 读 sync_people.features_json
+
 已移除（旧推送/分块/队列模式）: /video/push, /enqueue, /chunk, /assemble,
 /results, /queue/stats, /mark_frames
 """
 import os
 
-from flask import Blueprint, request, jsonify, send_file
+from flask import Blueprint, request, jsonify
 
 from ..logger import setup_logger
 from .. import state
@@ -121,61 +126,6 @@ def chat_ask():
     return jsonify({"answer": answer, "provider": provider}), 200
 
 
-@api_bp.route('/api/oracle/video/<int:video_id>/thumb', methods=['GET'])
-def video_thumb(video_id):
-    """返回视频首帧缩略图（token 校验，不公网裸奔）。
-
-    缩略图由 video_processor 处理成功后生成于 /opt/fam-edge/thumbs/{video_id}.jpg。
-    """
-    if not _check_token():
-        return jsonify({"error": "unauthorized"}), 401
-    path = f"/opt/fam-edge/thumbs/{video_id}.jpg"
-    if not os.path.isfile(path):
-        return jsonify({"error": "thumb_not_found"}), 404
-    return send_file(path, mimetype='image/jpeg')
-
-
-@api_bp.route('/api/oracle/event/<int:event_id>/thumb', methods=['GET'])
-def event_thumb(event_id):
-    """返回事件对应时间点的画面截图（token 校验）。
-
-    由 video_processor 处理成功后按事件时间戳定位视频帧生成：
-    /opt/fam-edge/thumbs/ev_{event_id}.jpg
-    截图缺失返回 404（前端隐藏，不拿视频首帧冒充该事件画面）。
-    """
-    if not _check_token():
-        return jsonify({"error": "unauthorized"}), 401
-    path = f"/opt/fam-edge/thumbs/ev_{event_id}.jpg"
-    if not os.path.isfile(path):
-        return jsonify({"error": "thumb_not_found"}), 404
-    return send_file(path, mimetype='image/jpeg')
-
-
-@api_bp.route('/api/oracle/person/avatar', methods=['GET'])
-def person_avatar():
-    """人物代表画面（头像）：在该人物出现的所有事件中，返回第一个有截图的事件画面。
-
-    人物出现在多个视频/事件中，任选一个截图存在的即可（比视频首帧准确——
-    首帧可能根本没有该人物）。
-    参数: label=人物标识（如 人物A）；匹配 person_list_json 数组中的元素。
-    """
-    if not _check_token():
-        return jsonify({"error": "unauthorized"}), 401
-    label = (request.args.get('label') or '').strip()
-    if not label:
-        return jsonify({"error": "缺少 label"}), 400
-    # JSON 数组元素精确匹配（person_list_json 形如 ["人物A","人物C"]）
-    pat = f'%"{label}"%'
-    rows = state.get_db()._conn.execute(
-        "SELECT id FROM events WHERE person_list_json LIKE ? "
-        "ORDER BY id DESC", (pat,)).fetchall()
-    for r in rows:
-        p = f"/opt/fam-edge/thumbs/ev_{r['id']}.jpg"
-        if os.path.isfile(p):
-            return send_file(p, mimetype='image/jpeg')
-    return jsonify({"error": "no_avatar_found"}), 404
-
-
 @api_bp.route('/api/oracle/activity', methods=['GET'])
 def activity():
     """实时服务状态 + 最近活动流（token 校验）。

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

@@ -12,7 +12,13 @@
        "events": [                                  # 有用时间点 + 画面信息
          {"timestamp": "2026-08-21 08:15:30",       # 绝对北京时间（event_start_time 推算）
           "description": str,
-          "people": [str],
+          "people": [str],                          # 该时刻出现的人物标识
+          "person_appearances": [                   # 该时刻每个人物的结构化特征
+             {"uid": str,                            # 与 people 数组里的标识一致
+              "features": {                          # 客观可见特征，看不清写 "unknown"
+                "gender, age_band, build, hair, clothing, face, distinguishing"
+              },
+              "action": str}],
           "is_attention_event": bool}, ...],
        "people_mentioned": [str],                   # 本视频出现的人物标识/真名
      }

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

@@ -295,7 +295,7 @@ class GeminiAdapter(BaseModelAdapter):
 
     @staticmethod
     def _normalize(result: dict) -> dict:
-        """统一字段名：frame_details -> events（兼容旧结构）。"""
+        """统一字段名：frame_details -> events（兼容旧结构），保留 person_appearances 特征。"""
         events = result.get('events')
         if events is None and 'frame_details' in result:
             events = []
@@ -308,13 +308,25 @@ class GeminiAdapter(BaseModelAdapter):
                 })
         if events is None:
             events = []
+        # 透传 events 内全部字段（含 person_appearances + features），不丢特征
+        norm_events = []
+        for ev in events:
+            if not isinstance(ev, dict):
+                continue
+            item = dict(ev)  # 保留原模型输出的所有字段（含 person_appearances）
+            # 保证 people 字段为字符串数组
+            people = ev.get('people') or []
+            if isinstance(people, str):
+                people = [people]
+            item['people'] = [str(p) for p in people if p]
+            norm_events.append(item)
         people = result.get('people_mentioned') or result.get('entities_json') or []
         if isinstance(people, list) and people and isinstance(people[0], dict):
-            people = [p.get('person', '') for p in people]
+            people = [p.get('person', '') or p.get('uid', '') for p in people]
         people = [p for p in people if p]
         return {
             "global_summary": result.get('global_summary', ''),
-            "events": events,
+            "events": norm_events,
             "people_mentioned": people,
         }
 

fam-edge/src/fam_edge/oracle_db.py

@@ -71,6 +71,7 @@ class OracleDB:
             ts TEXT,
             description TEXT,
             person_list_json TEXT,
+            person_appearances_json TEXT,
             is_attention_event INTEGER DEFAULT 0,
             FOREIGN KEY(video_id) REFERENCES videos(id)
         );
@@ -81,6 +82,8 @@ class OracleDB:
             first_seen TEXT,
             appearances INTEGER DEFAULT 0,
             source TEXT DEFAULT 'llm',
+            features_json TEXT,
+            display_uid TEXT,
             updated_at TEXT
         );
         CREATE TABLE IF NOT EXISTS sync_cursor (
@@ -122,6 +125,18 @@ class OracleDB:
         ]:
             if col not in cols:
                 c.execute(ddl)
+        # 兼容旧库：events 表补 person_appearances_json（新架构 v3 加）
+        ev_cols = [r[1] for r in c.execute("PRAGMA table_info(events)").fetchall()]
+        if 'person_appearances_json' not in ev_cols:
+            c.execute("ALTER TABLE events ADD COLUMN person_appearances_json TEXT")
+        # 兼容旧库：people 表补 features_json / display_uid（新架构 v3 加）
+        pe_cols = [r[1] for r in c.execute("PRAGMA table_info(people)").fetchall()]
+        for col, ddl in [
+            ('features_json', "ALTER TABLE people ADD COLUMN features_json TEXT"),
+            ('display_uid', "ALTER TABLE people ADD COLUMN display_uid TEXT"),
+        ]:
+            if col not in pe_cols:
+                c.execute(ddl)
         self._conn.commit()
 
     # ------------------------------------------------------------------
@@ -238,7 +253,11 @@ class OracleDB:
 
     def mark_video_processed(self, video_id: int, summary: str, events: List[dict],
                             people: List[str], compute_provider: str) -> List[int]:
-        """落库视频结果；返回新插入事件的 id 列表（与 events 参数一一对应，供事件截图用）"""
+        """落库视频结果；返回新插入事件的 id 列表（与 events 参数一一对应）。
+
+        events 内每条可含 person_appearances（[{uid, features, action}]），
+        原样存到 events.person_appearances_json，供 person_service 聚合特征。
+        """
         with self._write_lock:
             now = _now_iso()
             self._conn.execute(
@@ -250,11 +269,13 @@ class OracleDB:
             self._conn.execute("DELETE FROM events WHERE video_id=?", (video_id,))
             event_ids: List[int] = []
             for ev in events:
+                pa = ev.get('person_appearances')
                 cur = self._conn.execute(
                     "INSERT INTO events (video_id, ts, description, person_list_json, "
-                    "is_attention_event) VALUES (?,?,?,?,?)",
+                    "person_appearances_json, is_attention_event) VALUES (?,?,?,?,?,?)",
                     (video_id, ev.get('timestamp', ''), ev.get('description', ''),
                      json.dumps(ev.get('people', []), ensure_ascii=False),
+                     json.dumps(pa, ensure_ascii=False) if pa else None,
                      1 if ev.get('is_attention_event') else 0))
                 event_ids.append(cur.lastrowid)
             self._conn.commit()
@@ -276,27 +297,71 @@ class OracleDB:
     # people
     # ------------------------------------------------------------------
     def upsert_person(self, label: str, canonical_name: str = '', source: str = 'llm',
-                     first_seen: str = ''):
+                     first_seen: str = '', features: dict = None,
+                     display_uid: str = ''):
+        """登记/更新人物。
+
+        features: 该人物的结构化特征 dict（gender/age_band/build/hair/clothing/face/
+                  distinguishing）。与已有 features_json 合并（已有非 unknown 字段不被
+                  覆盖，新非 unknown 字段补齐）。None 时不更新特征列。
+        display_uid: 大模型给的人物 UID（如 "人物A"）。label 本身就是 UID 时可省略。
+        """
         now = _now_iso()
         row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone()
+        # 特征合并（在已有 features_json 基础上）
+        merged_features = self._merge_features(
+            row['features_json'] if row else None, features) if row else (
+            self._merge_features(None, features))
         if row:
             # manual 覆盖 llm；llm 不覆盖 manual
             if source == 'manual' or row['source'] != 'manual':
                 self._conn.execute(
                     "UPDATE people SET canonical_name=?, source=?, appearances=appearances+1, "
-                    "updated_at=? WHERE label=?",
-                    (canonical_name or row['canonical_name'], source, now, label))
+                    "features_json=?, display_uid=?, updated_at=? WHERE label=?",
+                    (canonical_name or row['canonical_name'], source,
+                     merged_features, display_uid or row['display_uid'] or label, now, label))
             else:
                 self._conn.execute(
-                    "UPDATE people SET appearances=appearances+1, updated_at=? WHERE label=?",
-                    (now, label))
+                    "UPDATE people SET appearances=appearances+1, features_json=?, "
+                    "display_uid=?, updated_at=? WHERE label=?",
+                    (merged_features, display_uid or row['display_uid'] or label, now, label))
         else:
             self._conn.execute(
                 "INSERT INTO people (label, canonical_name, first_seen, appearances, "
-                "source, updated_at) VALUES (?,?,?,1,?,?)",
-                (label, canonical_name, first_seen or now, source, now))
+                "source, features_json, display_uid, updated_at) "
+                "VALUES (?,?,?,1,?,?,?,?)",
+                (label, canonical_name, first_seen or now, source,
+                 merged_features, display_uid or label, now))
         self._conn.commit()
 
+    @staticmethod
+    def _merge_features(old_json: Optional[str], new_features: Optional[dict]) -> Optional[str]:
+        """合并人物特征：已有非 unknown 字段不被覆盖；新字段在 old 为空/unknown 时补齐。
+
+        - old_json 为 None / 空 -> 直接用 new_features
+        - new_features 为 None / 空 -> 不变
+        - 字段级：new 值非 'unknown' 且非空时覆盖 old（old 为 unknown/空）；
+          new 值为 'unknown' 时保留 old（哪怕 old 也是 unknown）
+        """
+        if not new_features:
+            return old_json
+        try:
+            old = json.loads(old_json) if old_json else {}
+        except (ValueError, TypeError):
+            old = {}
+        if not isinstance(old, dict):
+            old = {}
+        merged = dict(old)
+        for k, v in new_features.items():
+            v_str = str(v).strip() if v is not None else ''
+            if v_str and v_str.lower() != 'unknown':
+                # 新值是有效特征，覆盖（无论 old 是什么）
+                merged[k] = v_str
+            elif k not in merged:
+                # 新值 unknown 且 old 没该字段，至少把字段占位（写 unknown）
+                merged[k] = v_str or 'unknown'
+        return json.dumps(merged, ensure_ascii=False)
+
     def set_canonical(self, label: str, canonical_name: str, source: str = 'manual'):
         """手动命名：设置规范名（label 可视为别名）。"""
         self.upsert_person(label, canonical_name, source='manual')

fam-edge/src/fam_edge/person_service.py

@@ -43,7 +43,7 @@ class PersonService:
 
     # ------------------------------------------------------------------
     def reconcile(self):
-        """汇总 + LLM 合并一次。可由定时或手动触发。"""
+        """汇总 + 聚合特征 + LLM 合并一次。可由定时或手动触发。"""
         # 1. 统计每个人物标签出现过的视频数（去重），校准 appearances（防每次 reconcile 累加膨胀）
         label_videos: Dict[str, set] = {}
         for v in self.db.get_all_videos():
@@ -57,21 +57,30 @@ class PersonService:
         for label, vids in label_videos.items():
             self.db.set_person_appearances(label, len(vids), source='llm')
 
-        # 2. 收集未命名（无 canonical 或 canonical==label）的标签 + 描述样本
+        # 2. 从 events.person_appearances_json 聚合每个 uid 的特征，写 people.features_json
+        uid_features = self._aggregate_features()
+        for uid, feats in uid_features.items():
+            # 用 upsert_person 合并特征（已有非 unknown 字段不被覆盖）
+            # appearances 已在第 1 步校准，这里不再 +1（传 features 即可，display_uid=uid）
+            self.db.upsert_person(uid, source='llm', features=feats, display_uid=uid)
+
+        # 3. 收集未命名（无 canonical 或 canonical==label）的标签 + 特征文本
         rows = self.db.get_people()
         manual = {r['label']: r['canonical_name'] for r in rows if r['source'] == 'manual' and r['canonical_name']}
         unnamed = [r for r in rows if not r['canonical_name'] or r['canonical_name'] == r['label']]
         if not unnamed:
             logger.info("PersonService: 无待合并人物，跳过 LLM 合并")
-            self.db.record_activity('person', 'merge_skip', f"已校准 {len(label_videos)} 个标签，无待合并")
+            self.db.record_activity(
+                'person', 'merge_skip',
+                f"已校准 {len(label_videos)} 个标签 + 聚合 {len(uid_features)} 个特征，无待合并")
             return
 
-        samples = self._collect_descriptions([r['label'] for r in unnamed])
-        mapping = self._llm_merge(unnamed, samples)
+        features_lines = self._collect_features_text([r['label'] for r in unnamed])
+        mapping = self._llm_merge(unnamed, features_lines)
         if not mapping:
             return
 
-        # 3. 落库：canonical 若是另一个 label（target），解析为其已有 canonical，保证同一身份统一显示名
+        # 4. 落库：canonical 若是另一个 label（target），解析为其已有 canonical，保证同一身份统一显示名
         label_to_canonical = {r['label']: (r['canonical_name'] or r['label']) for r in rows}
         updated = 0
         for label, canonical in mapping.items():
@@ -83,35 +92,84 @@ class PersonService:
                 updated += 1
         self.db.record_activity(
             'person', 'merge_done',
-            f"校准 {len(label_videos)} 标签，LLM 合并更新 {updated} 条（{', '.join(list(mapping)[:6])}）")
+            f"校准 {len(label_videos)} 标签 + 聚合 {len(uid_features)} 特征，"
+            f"LLM 合并更新 {updated} 条（{', '.join(list(mapping)[:6])}）")
         logger.info(f"PersonService: LLM 合并完成，更新 {updated} 条")
 
-    def _collect_descriptions(self, labels: List[str]) -> Dict[str, List[str]]:
-        """从 events 表收集每个标签出现时的描述样本。"""
-        samples: Dict[str, List[str]] = {l: [] for l in labels}
+    def _aggregate_features(self) -> Dict[str, Dict]:
+        """从 events.person_appearances_json 聚合每个 uid 的合并特征。
+
+        遍历所有事件的 person_appearances，按 uid 收集 features dict，
+        合并规则：首次非 unknown 值优先（与 oracle_db._merge_features 一致）。
+        """
+        uid_features: Dict[str, Dict] = {}
         rows = self.db._conn.execute(
-            "SELECT description, person_list_json FROM events").fetchall()
+            "SELECT person_appearances_json FROM events "
+            "WHERE person_appearances_json IS NOT NULL").fetchall()
         for r in rows:
             try:
-                plist = json.loads(r['person_list_json'] or '[]')
+                appearances = json.loads(r['person_appearances_json'] or '[]')
             except (ValueError, TypeError):
-                plist = []
-            for p in plist:
-                if p in samples and len(samples[p]) < 3 and r['description']:
-                    samples[p].append(r['description'])
-        return samples
+                continue
+            if not isinstance(appearances, list):
+                continue
+            for pa in appearances:
+                if not isinstance(pa, dict):
+                    continue
+                uid = (pa.get('uid') or '').strip()
+                if not uid or uid in ('无人', '无'):
+                    continue
+                feats = pa.get('features') or {}
+                if not isinstance(feats, dict):
+                    continue
+                if uid not in uid_features:
+                    uid_features[uid] = dict(feats)
+                else:
+                    merged = dict(uid_features[uid])
+                    for k, v in feats.items():
+                        v_str = str(v).strip() if v is not None else ''
+                        if v_str and v_str.lower() != 'unknown':
+                            merged[k] = v_str
+                        elif k not in merged:
+                            merged[k] = v_str or 'unknown'
+                    uid_features[uid] = merged
+        return uid_features
 
-    def _llm_merge(self, unnamed: List, samples: Dict[str, List[str]]) -> Dict[str, str]:
-        """请 LLM 把标签合并为规范名。返回 {label: canonical}。"""
+    def _collect_features_text(self, labels: List[str]) -> str:
+        """渲染每个 label 的特征为文本行，供 LLM 合并 prompt 用。
+
+        格式：- 人物A：性别=男; 年龄=中年; 身材=瘦高; 发型=短发黑色; 衣着=红色卫衣; 面部=蓄须; 辨识=左手戴手表
+        无 features_json 的 label 用"（无特征）"占位。
+        """
+        rows = self.db.get_people()
+        feat_map = {r['label']: r['features_json'] for r in rows}
+        lines = []
+        feat_keys = ['gender', 'age_band', 'build', 'hair', 'clothing', 'face', 'distinguishing']
+        feat_labels = {'gender': '性别', 'age_band': '年龄', 'build': '身材',
+                       'hair': '发型', 'clothing': '衣着', 'face': '面部',
+                       'distinguishing': '辨识'}
+        for label in labels:
+            raw = feat_map.get(label) or '{}'
+            try:
+                feats = json.loads(raw) if raw else {}
+            except (ValueError, TypeError):
+                feats = {}
+            if not feats:
+                lines.append(f"- {label}：（无特征）")
+                continue
+            parts = []
+            for k in feat_keys:
+                if k in feats:
+                    parts.append(f"{feat_labels.get(k, k)}={feats[k]}")
+            lines.append(f"- {label}：" + '; '.join(parts) if parts else f"- {label}：（无特征）")
+        return chr(10).join(lines)
+
+    def _llm_merge(self, unnamed: List, features_lines: str) -> Dict[str, str]:
+        """请 LLM 把标签合并为规范名（基于特征文本）。返回 {label: canonical}。"""
         if self._llm is None:
             logger.warning("PersonService: 无可用的 LLM 适配器，跳过合并")
             return {}
-        lines = []
-        for r in unnamed:
-            label = r['label']
-            desc = '；'.join(samples.get(label, [])) or '（无描述）'
-            lines.append(f"- {label}：出现场景 {desc}")
-        prompt = build_person_merge_prompt(chr(10).join(lines))
+        prompt = build_person_merge_prompt(features_lines)
         try:
             text = self._llm.chat(prompt, max_tokens=1024)
         except Exception as e:

fam-edge/src/fam_edge/video_processor.py

@@ -4,13 +4,17 @@ VideoProcessor - 整视频分析编排
 流程（不再切片/抽帧）:
 1. 从 OracleDB 取当前 known_members_context（已命名/合并的人物）
 2. 按 vision_order 依次调适配器的 analyze_video（Gemini 整视频 -> NVIDIA 整视频）
-3. 首个成功结果 -> 归一化 -> 写 OracleDB（videos + events 表）
-4. 把本视频 people_mentioned 更新进 people 表（供 person_service 后续合并）
+3. 首个成功结果 -> 归一化 -> 写 OracleDB（videos + events 表，含 person_appearances）
+4. 把本视频 people_mentioned 更新进 people 表（带 features 特征，供 person_service 合并）
 
 降级: 全部视觉模型失败 -> 标记视频 failed（不再本地融合）
+
+注: 不依赖 OpenCV/cv2。视频文件校验用 ffprobe（subprocess）；不再生成帧 jpg。
 """
 import os
 import re
+import json
+import subprocess
 from datetime import datetime, timedelta, timezone
 from typing import Dict, List, Optional
 
@@ -86,42 +90,82 @@ def _parse_event_ts(ts: str, start_dt):
     return ts, 0.0
 
 
+def _ffprobe_available() -> bool:
+    """ffprobe 是否可用（ffmpeg 套件自带）。"""
+    try:
+        r = subprocess.run(
+            ['ffprobe', '-version'],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5)
+        return r.returncode == 0
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return False
+    except Exception:
+        return False
+
+
 def validate_video(path: str) -> tuple:
     """校验视频文件是否为正常可解码视频。
 
     返回 (ok: bool, error: str, meta: dict|None)
       - meta: {fps, frames, duration_sec, width, height}
-    用 OpenCV 打开并读取至少 1 帧（不校验会导致空/半成品文件浪费云端配额）。
+
+    用 ffprobe（subprocess）查 stream 信息。无 ffprobe 时仅做大小检查
+    （与旧 cv2 缺失时行为一致，跳过深度校验）。
+    不校验会导致空/半成品文件浪费云端配额。
     """
-    meta = None
     try:
         if not path or not os.path.isfile(path):
             return False, "file_missing", None
         if os.path.getsize(path) == 0:
             return False, "file_empty", None
+        if not _ffprobe_available():
+            return True, "", None  # 无 ffprobe 时跳过深度校验（仅大小检查）
+        # -v error: 只报错；-show_entries: 只取需要的字段；-of json: JSON 输出
+        r = subprocess.run(
+            ['ffprobe', '-v', 'error', '-show_entries',
+             'stream=codec_type,avg_frame_rate,nb_frames,duration,width,height',
+             '-of', 'json', path],
+            capture_output=True, text=True, timeout=30)
+        if r.returncode != 0:
+            return False, f"ffprobe_error: {r.stderr[:200]}", None
         try:
-            import cv2
-        except ImportError:
-            return True, "", None  # 无 cv2 时跳过深度校验（仅大小检查）
-        cap = cv2.VideoCapture(path)
+            data = json.loads(r.stdout or '{}')
+        except ValueError:
+            return False, "ffprobe_bad_json", None
+        streams = data.get('streams') or []
+        vstream = next((s for s in streams if s.get('codec_type') == 'video'), None)
+        if not vstream:
+            return False, "no_video_stream", None
+        # fps: avg_frame_rate 形如 "25/1" -> 25.0
+        fps = 0.0
+        avg_rate = vstream.get('avg_frame_rate', '0/1')
         try:
-            if not cap.isOpened():
-                return False, "cannot_open", None
-            ok, frame = cap.read()
-            if not ok or frame is None:
-                return False, "no_decodable_frame", None
-            fps = float(cap.get(cv2.CAP_PROP_FPS) or 0)
-            frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
-            meta = {
-                "fps": round(fps, 2),
-                "frames": frames,
-                "duration_sec": round(frames / max(fps, 0.01), 1),
-                "width": int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0),
-                "height": int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0),
-            }
-        finally:
-            cap.release()
+            num, den = avg_rate.split('/')
+            den_f = float(den or '1')
+            fps = float(num) / den_f if den_f else 0.0
+        except (ValueError, ZeroDivisionError):
+            fps = 0.0
+        frames = 0
+        try:
+            frames = int(vstream.get('nb_frames') or 0)
+        except (ValueError, TypeError):
+            frames = 0
+        duration = 0.0
+        try:
+            duration = float(vstream.get('duration') or 0)
+        except (ValueError, TypeError):
+            duration = 0.0
+        meta = {
+            "fps": round(fps, 2),
+            "frames": frames,
+            "duration_sec": round(duration, 1) if duration else (
+                round(frames / max(fps, 0.01), 1) if frames and fps else 0),
+            "width": int(vstream.get('width') or 0),
+            "height": int(vstream.get('height') or 0),
+        }
         return True, "", meta
+    except subprocess.TimeoutExpired:
+        return False, "ffprobe_timeout", None
     except Exception as e:
         return False, f"validate_exc: {e}", None
 
@@ -248,94 +292,69 @@ class VideoProcessor:
                 pass
 
         norm_events = []
-        offsets = []
         for ev in events:
-            abs_ts, off = _parse_event_ts(ev.get('timestamp'), start_dt)
+            abs_ts, _ = _parse_event_ts(ev.get('timestamp'), start_dt)
+            ev_people = [_clean_person(str(p)) for p in ev.get('people', []) if p]
+            # 透传 person_appearances（含 uid/features/action），清洗 uid 字符串
+            appearances = ev.get('person_appearances') or []
+            norm_appearances = []
+            for pa in appearances:
+                if not isinstance(pa, dict):
+                    continue
+                uid = _clean_person(str(pa.get('uid', '')))
+                if not uid:
+                    continue
+                feats = pa.get('features') or {}
+                if not isinstance(feats, dict):
+                    feats = {}
+                norm_appearances.append({
+                    "uid": uid,
+                    "features": feats,
+                    "action": str(pa.get('action', '')),
+                })
             norm_events.append({
                 "timestamp": abs_ts,
                 "description": str(ev.get('description', '')),
-                "people": [_clean_person(str(p)) for p in ev.get('people', []) if p],
+                "people": ev_people,
+                "person_appearances": norm_appearances,
                 "is_attention_event": bool(ev.get('is_attention_event', False)),
             })
-            offsets.append(off)
 
         # 清洗 people_mentioned（去掉括号注释串，防污染人物表/合并）
         people = [_clean_person(str(p)) for p in people if p]
         people = [p for p in people if p and p not in ('无人', '无')]
 
+        # mark_video_processed 会把 norm_events 里的 person_appearances 落到
+        # events.person_appearances_json，供 person_service 聚合特征
         event_ids = self.db.mark_video_processed(video_id, summary, norm_events, people, provider)
-        # 缩略图 + 每个事件对应时间点的画面截图（用相对偏移直接定位，避免模型绝对时间误差）
-        if vrow and vrow['local_path']:
-            self._generate_thumb(video_id, vrow['local_path'])
-            self._generate_event_thumbs(video_id, vrow['local_path'], offsets, event_ids)
 
-        # 更新 people 表（标签级，待 person_service 合并）
+        # 更新 people 表（标签级 + 特征：从该视频所有 person_appearances 收集每个 uid 的特征）
+        uid_features = {}
+        for ev in norm_events:
+            for pa in ev.get('person_appearances', []):
+                uid = pa.get('uid')
+                if not uid or uid in ('无人', '无'):
+                    continue
+                feats = pa.get('features') or {}
+                if uid not in uid_features:
+                    uid_features[uid] = feats
+                else:
+                    # 同一 uid 多次出现：合并非 unknown 字段（与 upsert_person 的合并一致）
+                    merged = dict(uid_features[uid])
+                    for k, v in feats.items():
+                        v_str = str(v).strip() if v is not None else ''
+                        if v_str and v_str.lower() != 'unknown':
+                            merged[k] = v_str
+                        elif k not in merged:
+                            merged[k] = v_str or 'unknown'
+                    uid_features[uid] = merged
         for p in people:
             if p and p not in ('无人', '无'):
-                self.db.upsert_person(p, source='llm')
+                feats = uid_features.get(p)
+                if feats:
+                    self.db.upsert_person(p, source='llm', features=feats, display_uid=p)
+                else:
+                    self.db.upsert_person(p, source='llm')
         logger.info(f"[video_id={video_id}] 已落库: summary={len(summary)}字, "
-                    f"events={len(norm_events)}, people={people}")
-
-    def _thumbs_dir(self) -> str:
-        db_path = self.config.get('oracle_db', {}).get(
-            'path', '/opt/fam-edge/data/oracle.db')
-        d = os.path.abspath(os.path.join(os.path.dirname(db_path), '..', 'thumbs'))
-        os.makedirs(d, exist_ok=True)
-        return d
-
-    def _generate_thumb(self, video_id: int, video_path: str) -> bool:
-        """抽视频首帧生成 JPEG 缩略图（/opt/fam-edge/thumbs/{video_id}.jpg）"""
-        try:
-            import cv2
-            out = os.path.join(self._thumbs_dir(), f"{video_id}.jpg")
-            cap = cv2.VideoCapture(video_path)
-            try:
-                ok, frame = cap.read()
-            finally:
-                cap.release()
-            if not ok or frame is None:
-                logger.warning(f"抽帧失败 video_id={video_id}: 无法读取首帧")
-                return False
-            h, w = frame.shape[:2]
-            if w > 640:
-                frame = cv2.resize(frame, (640, int(h * 640 / w)))
-            cv2.imwrite(out, frame, [cv2.IMWRITE_JPEG_QUALITY, 65])
-            logger.info(f"缩略图已生成: {out}")
-            return True
-        except Exception as e:
-            logger.warning(f"抽帧异常 video_id={video_id}: {e}")
-            return False
-
-    def _generate_event_thumbs(self, video_id: int, video_path: str,
-                               offsets: List[float], event_ids: List[int]):
-        """按事件在视频内的偏移秒定位帧，生成事件画面截图 ev_{event_id}.jpg"""
-        try:
-            import cv2
-        except Exception as e:
-            logger.warning(f"事件截图依赖缺失 video_id={video_id}: {e}")
-            return
-        try:
-            thumbs = self._thumbs_dir()
-            cap = cv2.VideoCapture(video_path)
-            try:
-                for off, eid in zip(offsets, event_ids):
-                    if off < 0:
-                        off = 0.0
-                    cap.set(cv2.CAP_PROP_POS_MSEC, int(off * 1000))
-                    ok, frame = cap.read()
-                    if not ok or frame is None:
-                        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
-                        ok, frame = cap.read()
-                    if not ok or frame is None:
-                        logger.warning(f"事件截图失败 ev_{eid}: 无法读取 offset={off:.0f}s")
-                        continue
-                    h, w = frame.shape[:2]
-                    if w > 640:
-                        frame = cv2.resize(frame, (640, int(h * 640 / w)))
-                    out = os.path.join(thumbs, f"ev_{eid}.jpg")
-                    cv2.imwrite(out, frame, [cv2.IMWRITE_JPEG_QUALITY, 65])
-                    logger.info(f"事件截图已生成 ev_{eid}.jpg (offset={off:.0f}s)")
-            finally:
-                cap.release()
-        except Exception as e:
-            logger.warning(f"事件截图异常 video_id={video_id}: {e}")
+                    f"events={len(norm_events)}, people={people}, "
+                    f"with_features={len(uid_features)}")