From ff01d14c79aba224b6865d2997ab39cdf75d279b Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Fri, 21 Aug 2026 18:06:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(v3):=20=E4=BA=BA=E7=89=A9=E6=A8=A1?= =?UTF-8?q?=E5=9D=97=E9=87=8D=E6=9E=84=20-=20=E5=A4=A7=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E5=8C=96=E7=89=B9=E5=BE=81=E5=80=BC=E6=9B=BF?= =?UTF-8?q?=E4=BB=A3=20OpenCV=20=E5=B8=A7=E5=AE=9A=E4=BD=8D=E3=80=82prompt?= =?UTF-8?q?=20=E5=A2=9E=E5=8A=A0=20person=5Fappearances(uid+7=E7=89=B9?= =?UTF-8?q?=E5=BE=81+action)=E5=B9=B6=E9=87=8D=E5=86=99=E5=90=88=E5=B9=B6?= =?UTF-8?q?=20prompt(=E7=A8=B3=E5=AE=9A=E7=89=B9=E5=BE=81=E4=BC=98?= =?UTF-8?q?=E5=85=88=E6=AF=94=E5=AF=B9)=EF=BC=9Bgemini=20=E9=80=8F?= =?UTF-8?q?=E4=BC=A0=E7=89=B9=E5=BE=81=E5=AD=97=E6=AE=B5=EF=BC=9Boracle=5F?= =?UTF-8?q?db=20events/people=20=E5=8A=A0=20person=5Fappearances=5Fjson/fe?= =?UTF-8?q?atures=5Fjson/display=5Fuid=20+=20=5Fmerge=5Ffeatures=EF=BC=9Bv?= =?UTF-8?q?ideo=5Fprocessor=20=E5=8E=BB=20cv2=20=E6=94=B9=20ffprobe=20?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C=E3=80=81=5Fstore=5Fresult=20=E8=81=9A?= =?UTF-8?q?=E5=90=88=20uid=20=E7=89=B9=E5=BE=81=E8=90=BD=20people=E3=80=81?= =?UTF-8?q?=E5=88=A0=E7=BC=A9=E7=95=A5=E5=9B=BE/=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E6=88=AA=E5=9B=BE=EF=BC=9Bperson=5Fservice=20=E8=81=9A?= =?UTF-8?q?=E5=90=88=E7=89=B9=E5=BE=81=E5=90=8E=E5=9F=BA=E4=BA=8E=E7=89=B9?= =?UTF-8?q?=E5=BE=81=E6=96=87=E6=9C=AC=20LLM=20=E5=90=88=E5=B9=B6=EF=BC=9B?= =?UTF-8?q?api=5Fgateway=20=E5=88=A0=203=20=E4=B8=AA=E5=9B=BE=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=9BNAS=20=E9=95=9C=E5=83=8F+DDL=20=E5=8A=A0?= =?UTF-8?q?=E6=96=B0=E5=AD=97=E6=AE=B5=EF=BC=9Bfam-ui=20=E7=89=B9=E5=BE=81?= =?UTF-8?q?=E5=8D=A1=E6=9B=BF=E4=BB=A3=E5=A4=B4=E5=83=8F=E3=80=81=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E6=97=B6=E9=97=B4=E7=BA=BF=E4=BA=BA=E7=89=A9=E7=89=B9?= =?UTF-8?q?=E5=BE=81=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fam-core/src/fam_core/db_layer.py | 21 +- fam-edge/config/config.yaml | 2 +- fam-edge/requirements.txt | 3 +- .../src/fam_edge/ai_orchestrator/prompts.py | 54 ++++- .../src/fam_edge/api_gateway/api_gateway.py | 64 +----- .../fam_edge/model_adapters/base_adapter.py | 8 +- .../fam_edge/model_adapters/gemini_adapter.py | 18 +- fam-edge/src/fam_edge/oracle_db.py | 83 ++++++- fam-edge/src/fam_edge/person_service.py | 108 +++++++-- fam-edge/src/fam_edge/video_processor.py | 217 ++++++++++-------- fam-ui/src/app.py | 139 +++++++---- scripts/ddl.sql | 3 + 12 files changed, 463 insertions(+), 257 deletions(-) diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py index 33b9187..7c9d7c6 100644 --- a/fam-core/src/fam_core/db_layer.py +++ b/fam-core/src/fam_core/db_layer.py @@ -165,18 +165,21 @@ def upsert_sync_events(rows: List[Dict]) -> int: cur.execute( """INSERT INTO sync_events (id, video_id, ts, description, person_list_json, - is_attention_event, updated_at, synced_at) - VALUES (%s,%s,%s,%s,%s,%s,%s, NOW()) + person_appearances_json, is_attention_event, + updated_at, synced_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s, NOW()) ON DUPLICATE KEY UPDATE video_id=VALUES(video_id), ts=VALUES(ts), description=VALUES(description), person_list_json=VALUES(person_list_json), + person_appearances_json=VALUES(person_appearances_json), is_attention_event=VALUES(is_attention_event), updated_at=VALUES(updated_at), synced_at=NOW()""", (r.get('id'), r.get('video_id'), r.get('ts'), r.get('description'), - r.get('person_list_json'), 1 if r.get('is_attention_event') else 0, + r.get('person_list_json'), r.get('person_appearances_json'), + 1 if r.get('is_attention_event') else 0, r.get('updated_at')) ) n += 1 @@ -242,19 +245,22 @@ def upsert_sync_people(rows: List[Dict]) -> int: cur.execute( """INSERT INTO sync_people (id, label, canonical_name, first_seen, appearances, - source, updated_at, synced_at) - VALUES (%s,%s,%s,%s,%s,%s,%s, NOW()) + source, features_json, display_uid, updated_at, synced_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s, NOW()) ON DUPLICATE KEY UPDATE label=VALUES(label), canonical_name=VALUES(canonical_name), first_seen=VALUES(first_seen), appearances=VALUES(appearances), source=VALUES(source), + features_json=VALUES(features_json), + display_uid=VALUES(display_uid), updated_at=VALUES(updated_at), synced_at=NOW()""", (r.get('id'), r.get('label'), r.get('canonical_name'), r.get('first_seen'), r.get('appearances') or 0, - r.get('source'), r.get('updated_at'))) + r.get('source'), r.get('features_json'), + r.get('display_uid'), r.get('updated_at'))) n += 1 conn.commit() return n @@ -267,7 +273,8 @@ def get_sync_people() -> List[Dict]: try: cur = conn.cursor(pymysql.cursors.DictCursor) cur.execute( - "SELECT id, label, canonical_name, first_seen, appearances, source, updated_at " + "SELECT id, label, canonical_name, first_seen, appearances, source, " + "features_json, display_uid, updated_at " "FROM sync_people ORDER BY id ASC") return cur.fetchall() finally: diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index fce0faf..b479e80 100644 --- a/fam-edge/config/config.yaml +++ b/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"] diff --git a/fam-edge/requirements.txt b/fam-edge/requirements.txt index 41a82f7..f883577 100644 --- a/fam-edge/requirements.txt +++ b/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 包） diff --git a/fam-edge/src/fam_edge/ai_orchestrator/prompts.py b/fam-edge/src/fam_edge/ai_orchestrator/prompts.py index 10e9ffb..8a45190 100644 --- a/fam-edge/src/fam_edge/ai_orchestrator/prompts.py +++ b/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}""" diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py index 8050f22..5df3c48 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/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//thumb, /api/oracle/event//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//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//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 校验）。 diff --git a/fam-edge/src/fam_edge/model_adapters/base_adapter.py b/fam-edge/src/fam_edge/model_adapters/base_adapter.py index 8a60d72..d67c33a 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/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], # 本视频出现的人物标识/真名 } diff --git a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py index f5cbd59..5cefb8e 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/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, } diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index dc48ddf..789f05f 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/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') diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index 0acead0..77b0b08 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/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: diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index a62e91a..dcb8b74 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/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)}") diff --git a/fam-ui/src/app.py b/fam-ui/src/app.py index 7d9febf..4241eca 100644 --- a/fam-ui/src/app.py +++ b/fam-ui/src/app.py @@ -279,7 +279,7 @@ def page_header(icon: str, title: str, sub: str = ''): def render_event_list(events: list): - """渲染事件时间线：左相对时间 + 右事件卡（事件画面截图 + 人物/关注徽章 + 描述）""" + """渲染事件时间线：左相对时间 + 右事件卡（人物徽章 + 人物特征摘要 + 描述）""" items = [] for e in events: ts = parse_ts(e.get('ts')) @@ -288,16 +288,41 @@ def render_event_list(events: list): persons = parse_persons(e.get('person_list_json')) attention = bool(e.get('is_attention_event')) desc = e.get('description') or '（无描述）' - eid = e.get('id') - # 事件对应时间点的画面截图（Oracle 带 token 接口；加载失败自动隐藏） - img_html = '' - if _oracle_url and eid: - img_html = ( - f'') + # 人物出现明细：从 person_appearances_json 渲染 uid + 简短特征 + action + appearances_html = '' + pa_raw = e.get('person_appearances_json') + appearances = [] + if pa_raw: + try: + appearances = json.loads(pa_raw) if isinstance(pa_raw, str) else pa_raw + except (ValueError, TypeError): + appearances = [] + if appearances and isinstance(appearances, list): + cells = [] + for pa in appearances: + if not isinstance(pa, dict): + continue + uid = str(pa.get('uid') or '').strip() + if not uid: + continue + feats = pa.get('features') or {} + feat_bits = [] + for fk in ('gender', 'clothing', 'face'): + fv = (feats.get(fk) or '').strip() if isinstance(feats, dict) else '' + if fv and fv.lower() != 'unknown': + feat_bits.append(fv) + feat_str = ' · '.join(feat_bits) if feat_bits else '无特征' + action = str(pa.get('action') or '').strip() + cells.append( + f'
' + f'{esc(uid)}' + f'{esc(feat_str)}' + f'{f"
{esc(action)}
" if action else ""}' + f'
') + if cells: + appearances_html = '
' + ''.join(cells) + '
' badges = ''.join( f'{esc(p)}' for p in sorted(persons)) @@ -308,7 +333,7 @@ def render_event_list(events: list): f'
' f'
{esc(time_label)}' f'{f"{esc(camera)}" if camera else ""}
' - f'
{img_html}{badges}' + f'
{badges}{appearances_html}' f'
{esc(desc)}
' f'
' ) @@ -535,18 +560,10 @@ if page == "🕒 事件时间轴": f'{esc(p)}' for p in str(provider).split(',')) summary = vid.get('summary_json') or '暂无全局摘要' - # 视频缩略图（Oracle 带 token 接口，无图时优雅降级） - thumb_html = '' - if _oracle_url: - thumb_html = ( - f'') + # 不再展示视频首帧缩略图（Oracle 端不再生成 jpg）； + # 摘要文本 + 下方事件时间线已足够定位画面内容 st.markdown( f'
' - f'{thumb_html}' f'
' f'{esc(vid.get("camera_name") or "未知摄像头")}' f'会话 #{vid["id"]}' @@ -562,7 +579,8 @@ if page == "🕒 事件时间轴": cursor = conn.cursor() cursor.execute( """SELECT e.id, e.ts, e.description, e.person_list_json, - e.is_attention_event, v.camera_name + e.person_appearances_json, e.is_attention_event, + v.camera_name FROM sync_events e JOIN sync_videos v ON e.video_id = v.id WHERE e.video_id=%s ORDER BY e.ts ASC""", @@ -731,7 +749,8 @@ elif page == "👤 人物管理": try: cursor = conn.cursor() cursor.execute( - "SELECT id, label, canonical_name, first_seen, appearances, source " + "SELECT id, label, canonical_name, first_seen, appearances, source, " + "features_json, display_uid " "FROM sync_people ORDER BY id ASC") people = cursor.fetchall() finally: @@ -745,13 +764,19 @@ elif page == "👤 人物管理": for p in people: key = p['canonical_name'] or p['label'] groups.setdefault(key, {'display': key, 'is_named': bool(p['canonical_name']), - 'labels': [], 'appearances': 0, 'first_seen': None}) + 'labels': [], 'appearances': 0, 'first_seen': None, + 'features_json': None, 'display_uid': None}) g = groups[key] g['labels'].append(p['label']) g['appearances'] += (p.get('appearances') or 0) fs = p.get('first_seen') if fs and (g['first_seen'] is None or str(fs) < str(g['first_seen'])): g['first_seen'] = fs + # 特征：取该聚合下任一 label 的 features_json（非空优先） + if not g['features_json'] and p.get('features_json'): + g['features_json'] = p['features_json'] + if not g['display_uid'] and p.get('display_uid'): + g['display_uid'] = p['display_uid'] if not groups: st.markdown( @@ -807,35 +832,57 @@ elif page == "👤 人物管理": f'border:1px solid #713f12;padding:1px 8px;border-radius:10px;' f'margin-left:8px;">未命名') label_str = ' · '.join(g['labels']) - # 人物代表画面：Oracle 端在该人物出现的所有事件中挑第一个有截图的事件画面 - # （该人物在多个视频多次出现，肯定能找到出现时刻的画面，不用视频首帧冒充） - rep_img = '' - if _oracle_url: - for lb in g['labels']: - try: - rp = requests.get( - f"{_oracle_url}/api/oracle/person/avatar", - params={"label": lb, "token": _oracle_token}, - timeout=10) - if rp.status_code == 200: - import urllib.parse - rep_img = ( - f'') - break - except Exception: + # 人物特征卡：从 sync_people.features_json 渲染结构化特征 + # （不依赖任何 jpg 文件；大模型每次分析视频时落库的特征值） + feats_html = '' + feats_raw = g.get('features_json') or '{}' + try: + feats = json.loads(feats_raw) if feats_raw else {} + except (ValueError, TypeError): + feats = {} + if feats: + # 结构化特征网格 + feat_rows = [ + ('性别', feats.get('gender')), + ('年龄段', feats.get('age_band')), + ('身材', feats.get('build')), + ('发型', feats.get('hair')), + ('衣着', feats.get('clothing')), + ('面部', feats.get('face')), + ('辨识点', feats.get('distinguishing')), + ] + feat_cells = [] + for label_name, val in feat_rows: + v = (val or '').strip() if val else '' + if not v: continue + cls = '' if v.lower() != 'unknown' else 'style="color:#475569;"' + feat_cells.append( + f'' + f'{esc(label_name)} ' + f'{esc(v)}') + if feat_cells: + feats_html = ( + f'
' + + ''.join(feat_cells) + '
') + else: + feats_html = ( + '
' + '特征待大模型补充（下段视频分析时由 VLM 落库）
') + uid_html = '' + if g.get('display_uid') and g['display_uid'] != key: + uid_html = (f'UID: {esc(g["display_uid"])}') st.markdown( f'
' - f'{esc(key)}{tag}
' + f'{esc(key)}{tag}{uid_html}
' f'
' f'标识: {esc(label_str)}
' f'
' f'出现 {g["appearances"]} 次 · 首次 {esc(first_str)}
' - f'{rep_img}', + f'{feats_html}', unsafe_allow_html=True) if not is_named: # 未命名身份：每个 label 都给一个命名框 diff --git a/scripts/ddl.sql b/scripts/ddl.sql index b4b9644..0e7c100 100644 --- a/scripts/ddl.sql +++ b/scripts/ddl.sql @@ -155,6 +155,7 @@ CREATE TABLE IF NOT EXISTS sync_events ( ts VARCHAR(32) COMMENT '事件时间点（文本）', description TEXT COMMENT '事件描述', person_list_json LONGTEXT COMMENT '涉及人物 JSON 数组（字符串或标签）', + person_appearances_json LONGTEXT COMMENT '该时刻人物结构化特征 JSON（[{uid,features,action}]）', is_attention_event TINYINT(1) DEFAULT 0 COMMENT 'AI 判断是否为关注事件', updated_at VARCHAR(32), synced_at DATETIME DEFAULT CURRENT_TIMESTAMP, @@ -170,6 +171,8 @@ CREATE TABLE IF NOT EXISTS sync_people ( first_seen VARCHAR(32) COMMENT '首次出现时间', appearances INT DEFAULT 0 COMMENT '出现次数', source VARCHAR(20) DEFAULT 'llm' COMMENT 'llm / manual（manual 优先不被覆盖）', + features_json LONGTEXT COMMENT '人物结构化特征 JSON（性别/年龄/身形/发型/衣着/面部/辨识）', + display_uid VARCHAR(100) COMMENT '大模型给的人物 UID（与 label 一致时冗余）', updated_at VARCHAR(32), synced_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_label (label),