diff --git a/.workbuddy/memory/2026-08-20.md b/.workbuddy/memory/2026-08-20.md new file mode 100644 index 0000000..0aee92e --- /dev/null +++ b/.workbuddy/memory/2026-08-20.md @@ -0,0 +1,31 @@ +# 2026-08-20 工作记录 + +## NAS 瘦身 + 甲骨文同步（架构重构收尾，任务 #49/#50/#57/#59） + +完成 NAS 端从「视频处理节点」到「纯管理后台」的改造： + +### fam-core（NAS）改动 +- 删除：scheduler / dispatcher / poller / event_receiver / video_server 五个模块 + tools/ 下 backfill 脚本（不再切片/抽帧/上传/提供视频下载） +- 新增 `oracle_sync.py`：唯一后台线程，每 30 分钟 `GET /api/oracle/sync?since=&token=` 拉增量 → upsert 到本地 MariaDB `sync_*` 镜像表 → 推进 `sync_cursor`；`push_name_correct` 回推命名校正 +- `db_layer.py` 重写：移除 process_tasks/event_details/monitor_events/family_members 旧逻辑，新增 sync_videos/sync_events/sync_people/sync_cursor 的 upsert/查询、`get_sync_stats`、`query_sync_events_for_person_date`、`get_sync_known_members_context` +- `member_manager.py`：命名/合并改回推 Oracle `/api/oracle/people/correct` + 即时 trigger_now 拉回 +- `chat_handler.py`：上下文改查 sync_events（按 person_list_json + 视频日期过滤），仍走 Oracle `/api/edge/chat/ask` +- `app.py`：仅启动 OracleSync 线程，移除看门狗 +- config：删除 scheduler/dispatcher/poller/video_server/storage，新增 `oracle_sync` 段（token 用 `${ORACLE_SYNC_TOKEN}` 环境变量） + +### fam-ui（NAS）改动 +- 改写读 sync_videos/sync_events/sync_people；事件时间轴改为「视频会话列表 + 事件时间线」（无帧图）；人物管理移除照片逻辑（仅标签/规范名/命名/合并）；统计改自 sync 表；侧边栏任务队列状态改为同步状态 + +### DDL / 文档 +- `scripts/ddl.sql` 新增 sync_videos / sync_events / sync_people / sync_cursor 四张镜像表（时间字段用 VARCHAR 规避 MariaDB 严格模式） +- README §1.1/§2.2/§3.1/§3.2/§3.3/§4/§5/§8.3 全面更新到新数据流（Google 硬盘→rclone→甲骨文→同步→NAS 镜像） + +### 提交 +- `[阶段2]` fam-edge 重构（整视频分析+同步接口+人物服务）已 commit+push +- `[阶段3]` fam-core 瘦身+Oracle-Sync+UI 改读 已 commit+push + +### 待办（#51 部署验证，未做） +- Oracle：rclone 配 Google 服务账号 JSON（orcLenas@gen-lang-client-0523399799.iam.gserviceaccount.com）→ systemd timer 定时同步到 /opt/fam-edge/gdrive_videos；设 ORACLE_SYNC_TOKEN 环境变量；重启 fam-edge +- NAS：MariaDB 跑新 DDL 建 sync_* 表；设 ORACLE_SYNC_TOKEN；重启 fam-core + fam-ui；卸载 NAS venv 无用包（opencv/numpy 等） +- 端到端验证：Google 硬盘新视频 → 甲骨文处理 → NAS 30 分钟同步可见 +- 服务账号私钥未入库，部署时需单独落到 Oracle 服务器本地 diff --git a/.workbuddy/memory/2026-08-21.md b/.workbuddy/memory/2026-08-21.md new file mode 100644 index 0000000..61b6d37 --- /dev/null +++ b/.workbuddy/memory/2026-08-21.md @@ -0,0 +1,180 @@ +# 2026-08-21 工作记录 + +## #51 部署验证 - rclone 同步 Google 硬盘 + 全链路跑通 + +### Oracle（fam-edge）部署完成 +- rclone v1.75.0 装到 /usr/local/bin（aarch64，zip 解压） +- 服务账号 JSON 存 /opt/fam-edge/gdrive-sa.json（chmod 600，用户提供） +- rclone remote `gdrive:`（drive scope + service_account_file） +- 同步脚本 /opt/fam-edge/rclone_sync.sh + systemd timer（每 5 分钟）： + `rclone sync --drive-shared-with-me gdrive:SS/Generic_ONVIF-001 /opt/fam-edge/gdrive_videos` +- 首次全量同步 20 个视频 7.52GB（~2 分钟完成） +- 共享目录是 `SS`（群晖监控站），摄像头 `Generic_ONVIF-001` 按 YYYYMMDDAM/PM 分目录 + +### 踩坑与修复（3 个 commit） +1. `watch_processor._scan_files` 改 os.walk 递归（原 listdir 只看根目录，视频在子目录） +2. `.env` 的 GEMINI/NVIDIA key 是旧无效值 → 用 env.sh 的真实 key 修复（config_loader 只读 .env） +3. **Gemini 上传**：必须用 `/upload/v1beta/files` 端点（/v1beta/files 是元数据端点）+ resumable 协议，370MB 整视频 ~55s 上传成功 +4. **NVIDIA**：base64 塞 payload 超 25MB 上限 → 改 Assets API（POST 拿 assetId+uploadUrl，PUT 上传）；PUT 必须**全小写 content-type 且值=POST contentType(video/mp4)**，否则 S3 预签名 SignatureDoesNotMatch +5. `parse_vlm_json` schema 从旧架构（entities_json/frame_details）改为新架构（global_summary/events/people_mentioned），兼容旧结构转换 + +### 端到端验证通过 +- 21 视频全部登记，video 1/2 已 done（provider=gemini，凌晨视频无事件=分析正确） +- NAS 重启 fam-core 触发立即同步：sync_videos=21，cursor 推进正确 +- 剩余 19 个视频串行处理中（Gemini 免费配额 429 会降级 flash-lite，之后可考虑 NVIDIA） + +### 待观察 +- Gemini 免费 key 配额限制（~20 req/day），后续 19 个视频可能大量 429 +- rclone timer 每 5 分钟增量同步验证 +- NAS 30 分钟自动同步已生效 + +## 生产-消费队列 + 模型超时×2（新需求，commit ce27e4a） +- 新增 `fam-edge/src/fam_edge/video_queue.py`：VideoQueue 生产-消费队列 + - 生产者线程：30s 轮询 rclone 落地目录，新文件登记 pending 并入队；启动时补入队 DB 未处理完的（重启恢复，实测 14 个） + - 消费者线程：max_concurrent=1，从队列取 video_id → VideoProcessor.process_video(timeout_multiplier=2) + - 防重复入队：_queued set；重试上限：max_retries=2（oracle_db.videos 加 retry_count 列，mark_video_failed 递增） +- `video_processor.process_video` 加 timeout_multiplier：遍历 adapter 时临时 adapter.timeout = 原配置 ×倍数，finally 恢复（gemini 600→1200s 生效日志确认） +- config.yaml video_processing 加 timeout_multiplier: 2 / max_retries: 2 +- 删除旧 watch_processor.py，app.py 改启动 VideoQueue +- oracle_db：retry_count 列（建表+ALTER 兼容旧库）、get_video_by_id、PRAGMA busy_timeout=10000 +- 部署验证：启动恢复入队 14 个、video_id=8 gemini 600→1200s、生产者登记 id=22 新视频、done 持续增长 + +## 模型调用统计界面 + lite 超时实测×4（commit b87b38b + 3e94c2a） +- Oracle 端：oracle_db 新增 model_calls 表（provider/model/video_id/started_at/duration_sec/success/error/created_at）；BaseModelAdapter 加 model_call_hook + _emit_model_call，gemini(_generate_video 每模型每attempt)/nvidia(analyze_video) 每次请求记录；get_sync_delta 下发 model_calls（created_at >= since + NAS 幂等 upsert 防漏） +- gemini 支持模型级 model_timeouts（最终值不参与 ×2）：实测 lite 370MB 视频耗时 22.6s（生产 27~34s），lite 限制 = 22.6×4 ≈ 90s；flash 仍 1200s（600×2）。日志确认：flash 本轮 1200s / lite 本轮 90s +- NAS 端：sync_model_calls 镜像表（ddl 8.1）；db_layer upsert_sync_model_calls/get_sync_model_calls/get_sync_model_calls_stats；oracle_sync 拉 model_calls（last_count 变 4 元组） +- fam-ui 新增 "🤖 模型统计" 页：按模型聚合（成功/失败/成功率/平均耗时/最后调用）+ 最近 100 条调用明细（请求时间/模型/耗时/状态/失败原因/视频） +- 验证：Oracle model_calls 正常记录（flash 429_quota 失败、lite 成功 27~34s）；NAS 同步 4 条，fam-ui 200 + +## 整体排查 + lite 超时 8 分钟（commit 83eefc1） +- **NVIDIA 400 排查结论**：`nvidia/nemotron-nano-12b-v2-vl` 对整视频分析**稳定复现 400/500 服务端内部错误**（"not enough values to unpack (expected 2, got 1)"），所有视频、有无 num_frames 均复现 → NVIDIA 兜底在当前模型/端点**不可用**（asset 上传 OK，chat.completions 必失败）。建议从 vision_order 移除或换模型 +- gemini-flash-lite 超时 90s → **480s（8 分钟）**，config model_timeouts 已改并部署（日志确认 480s） +- 处理进度：done 33 / failed 1（video24 待重试）/ pending 12 + +## NVIDIA 多模型降级链 + 实测结论（commit b7b5fe6） +- nvidia_adapter 改造：model_chain（asset 上传一次，逐个模型 video_url 引用尝试）+ switch_interval_sec 切换间隔（默认 5s）+ model_timeouts 每模型独立超时 + compute_provider 带模型名（nvidia:{model}） +- **实测全部 NVIDIA 云端视频候选不可用**：omni 500（asset_id 引用失败）、12b 400、llama-3.2-11b-vision 400（不支持视频）、cosmos-reason2/phi-3-vision/gemma-3/kosmos-2/fuyu-8b/paligemma 404；base64 超 25MB；public URL 方案被用户否决（不暴露视频） +- config：model_name=omni + fallback 12b/llama-11b，链机制保留，未来可用模型出现只需改配置 +- 手动验证：模型链 [1/3]→[2/3]→[3/3] 逐个尝试+5s 间隔+失败原因记录，全部失败返回 None +- ⚠️ 安全提醒：Oracle 曾短暂起 http.server:8000 暴露视频目录做 POC，已按用户要求关闭 + +## 前端修复：SQL 1054 + 视频缩略图（commit 4148507） +- **SQL 1054 修复**：fam-ui 事件查询 `SELECT ... camera_name FROM sync_events`（该列在 sync_videos）→ 改 JOIN sync_videos 取 v.camera_name +- **视频缩略图**：Oracle 装 opencv-python-headless(5.0)；video_processor 处理成功后抽首帧(宽≤640, JPEG q65)存 /opt/fam-edge/thumbs/{video_id}.jpg；api_gateway 新增 GET /api/oracle/video/{id}/thumb（token 校验，无 token 401）；已 done 46 视频 backfill 46/46 +- **fam-ui**：config 加 oracle_url(129.146.203.203:5000)+oracle_token(${ORACLE_SYNC_TOKEN})；事件时间轴视频会话头显示缩略图（img onerror 优雅降级）；start_ui.sh 补 source 项目 .env 注入 token +- 验证：NAS→Oracle thumb HTTP 200/40KB；fam-ui 进程 token env 就绪；UI 8501 正常 +- 人物管理无照片：架构局限（LLM 仅输出人物名，无图像锚点），如需人物照片需从视频定位+裁剪，待后续 + +## 每个事件对应时间点画面截图（commit 05f727a） +- oracle_db.mark_video_processed 返回 event_ids（与 events 一一对应） +- video_processor._generate_event_thumbs：事件 ts - 视频 event_start_time = 偏移秒 → cv2 跳帧(CAP_PROP_POS_MSEC)截图，存 /opt/fam-edge/thumbs/ev_{event_id}.jpg（宽≤640, q65；解析失败/负偏移取首帧） +- api_gateway 新增 GET /api/oracle/event/{event_id}/thumb（token 校验） +- fam-ui 事件查询加 e.id；render_event_list 每条事件显示对应截图（onerror 隐藏降级） +- backfill：已 done 视频 102 个事件截图全部生成；验证 NAS→Oracle HTTP 200/51KB、无 token 401 + +## 事件截图时间对不上修复（commit 5233296 + 3fcd00d） +- **根因**：模型输出"绝对北京时间"靠自身推算，1 小时视频内误差可达分钟级 → 截图按不准的时间定位帧必然图文不符 +- **修复**：① prompt 改为要求输出"视频内相对时间 HH:MM:SS"（模型对相对位置判断准）；② 后端 _parse_event_ts 解析相对时间 → 绝对时间 = start + offset 精确落库（兼容旧绝对格式）；③ 截图直接用 offset 跳帧 +- **连带 bug**：_parse_event_start_from_filename 只认带分隔符日期（2026-08-21），监控文件名是纯数字 20260820-140416 → event_start_time 空 → 新增纯数字格式解析 +- 启发式：相对时间 >6h 视为模型误输出绝对时间，不强行定位 +- 验证：重置 video 46 重分析 → event_start=14:04:16 ✓，事件 ts 14:05:37/14:06:21/14:13:02，截图 offset 36/48/81/125/526s 精确 ✓ +- 旧视频（除重分析的）仍用旧时间戳截图；如需全部修正需批量重分析（成本高，用户确认后再做） + +## 全部旧视频重分析 + 人物管理图片（commit 92ef9fe） +- **全量重分析**：清空 events(104) + 旧事件截图(108) + 重置全部 45+1 个视频 pending → 队列后台串行重跑（新相对时间逻辑，预计 1.5-2.5h，Gemini flash 429 → lite 兜底） +- **人物管理图片**：fam-ui 每个人物身份取其一 label 出现事件的截图作头像（查 sync_events.person_list_json LIKE → 显示 Oracle event thumb） +- 修正：video 46 起初被排除重跑但 events 被清 → 一并重置统一重跑 + +## 队列健壮性三项（commit 6afdda5，用户需求清单） +1. **失败重试间隔 30s→1h**：videos 表加 last_fail_at（mark_video_failed 记录）；video_queue._retry_allowed 对 failed 要求 retry_count0/可打开/可读帧/元数据 fps-frames-duration-分辨率）；新文件先过 mtime 稳定窗口(stable_window_sec=60 防 rclone 半成品)再校验，失败登记 status='invalid' 不入队（可追溯，_retry_allowed 排除）；process_video 处理前二次确认（失败标 failed:invalid_file:xxx） +- 验证：正常视频 ok+meta(30min/2880x1620/14.28fps)、损坏 cannot_open、空文件 file_empty；部署后日志"重试上限 10" +- 重分析 42 个 pending 继续后台跑 + +## 代码审查 18 项处理（commit 034dca9 + 4e85a98） +- **已解决（新增）**：#18 密钥明文直配 config.yaml（token/gemini/nvidia，弃 .env 依赖，验证 token_ok/key 长度正常）；#5 appearances 改 distinct 视频数覆盖校准（set_person_appearances，防 reconcile 累加膨胀）；#9 oracle_sync _pull_lock 防 trigger_now 与后台并发双拉；#11 LLM 合并命名解析 target 已有 canonical（统一显示名）；#12 熔断器状态转换加锁；#10 问答人名匹配改 JSON_CONTAINS 精确匹配（MariaDB 语法验证 OK）；#4 OracleDB _write_lock 复合写串行化；#6 每消费者独立 VideoProcessor（消 adapter.timeout 共享竞争）；#15 done 视频文件被覆盖 mtime>processed_at 自动重置重分析 +- **累计已解决**：#1/#2/#3/#4/#5/#6/#9/#10/#11/#12/#15/#18；部分缓解 #7/#14/#17 +- **未解决（说明）**：#8 删除传播（tombstone 大工程，暂缓）；#13 Gemini 文件/NVIDIA asset 残留清理（暂缓）；#16 gunicorn 线程数（运维配置，暂缓） +- 重分析进度：done 13 / pending 33（约剩 1h） + +## Prompt 集中化重构（commit a72b286，用户提供详细设计文档） +- 新增 fam-edge/src/fam_edge/ai_orchestrator/prompts.py：build_video_prompt（含 3 秒密度抽取/7 维度描述/people_mentioned 一致性强制/输出硬约束/边界情况/camera_name 注入）、build_chat_prompt、build_person_merge_prompt（唯一性+保守不合并）；__init__ 导出 +- gemini/nvidia adapter._build_video_prompt 改调共享函数（消除两份发散），注入 camera_name +- person_service._llm_merge 改调 build_person_merge_prompt +- fam-core chat_handler 新增 prompts.py（跨模块独立维护，风格一致），删除内联 CHAT_SYSTEM_PROMPT，chat_ask 改调 build_chat_prompt +- 部署验证：prompt 生成正常（3 秒密度/camera/唯一性均含）；fam-edge active；fam-core health OK +- 注意：3 秒密度可能让有人时段 events 数百条，Gemini maxOutputTokens=4096 可能截断——若出现 JSON 解析失败需提高 max_tokens + +## 最新代码调试 + 人物标识清洗（commit 0d73c23） +- 调试结果：服务全正常（fam-edge active、NAS core/ui 200、rclone timer active）；重分析 done 26/pending 20；新 prompt 正常——有人视频 5-16 事件、无人时段(晚 19:31/凌晨 5 点)events=0 合理、无 JSON 解析失败 +- 发现并修复：模型输出 people 含 known_members 上下文格式串"人物A（别名/标识：人物B）"污染人物表 → prompts.py 明确"people 只填标识本身不带括号注释" + video_processor._clean_person 后端清洗（events.people 与 people_mentioned 均清洗）+ 清理存量脏 label（人物A（别名/标识：人物B）→ 人物A） +- people 表现状：A/B/C/D 四标签，LLM 合并 B/C/D → canonical 人物A + +## 前端日期标签与视频文件对不上（commit da76319） + +**用户反馈**：前端显示"2026-08-21 · 画面静止/无人员活动"，和下面的视频文件对不上。 + +**根因链**（排查发现）： +1. 前端事件时间轴/统计按 `processed_at`（**分析处理时间**）做日期分组——全量重分析都在 8/21 完成，导致 8/15~8/20 录制的视频全部堆在"2026-08-21"标签下，与文件名（录制时间）错位 +2. Oracle videos 表 8/21 11:29~12:59 被整表重建（created 全在此区间），重分析串行进行 +3. 20 个视频 event_start_time 为空：video 30-45 在 13:00~13:25 处理时服务器跑的是旧文件名解析代码（无纯数字正则 YYYYMMDD-HHMMSS），且被后续部署重启中断（pending）；28/29 从未处理 +4. NAS 镜像状态与 Oracle 错位（35-45 NAS 显示 done、Oracle 实为 pending）——Oracle 重建后 NAS 增量未对齐 + +**修复**： +1. **日期维度统一改视频实际录制时间**：fam-ui（事件时间轴统计/列表/关注事件统计）与 db_layer（get_sync_videos/get_sync_stats/query_sync_events_for_person_date）全部改用 `COALESCE(NULLIF(event_start_time,''), processed_at)`（录制时间优先，回退处理时间） +2. **Oracle backfill**：对 12 个缺失 start 的视频按文件名解析回填（30/33-42/45）→ Oracle 46 个视频全部有 event_start_time +3. **NAS 强制全量重同步**：删 sync_cursor last_since + 重启 fam-core → since='' 全量拉取 → EMPTY_START 20→0、状态与 Oracle 对齐（done 28/failed 4/pending 14） + +**验证**：done 视频按录制日期分布 8/15:10、8/16:8、8/17:2、8/18:2、8/19:2、8/20:1、8/21:3；8/15/8/20 筛选正确；fam-ui 200 无报错。 + +**遗留**：Oracle 端 pending 14 个在队列继续串行处理（video 28 处理中）；failed 4 个（24/25/26 等，Gemini 429）等 1h 重试间隔自动重试。 + +## Google 硬盘删除联动 + DB 摘要保留（用户需求确认，无代码改动） +- **用户需求**：①谷歌删了甲骨文也删（文件层）；②Oracle/NAS 数据库生成的摘要不能删（DB 层） +- **验证结论**： + - 文件层：rclone sync 本就是镜像语义（远程删→本地删），实测放临时文件→sync→Deleted:1 确认生效 + - DB 层：video_queue 生产者只扫描"目录存在的文件"，文件消失不影响 Oracle videos/events/people 记录；NAS 镜像照常同步，前端摘要保留；截图接口 404 由前端 onerror 降级 +- **加固**：rclone_sync.sh 加 `--max-delete 200`（防 Google API 临时故障级联误删本地，单次最多删 200 个），脚本验证 exit=0、75 文件不受影响 +- 注意：当前是**单向镜像**（Google 为源→本地），不建议反向传播（本地删→谷歌删会误删原始监控），如需 rclone bisync 真双向需用户确认风险 + +## 实时服务状态界面 + 7 天活动记录（commit 0d0a7f6 + fec9a3e） +- **Oracle**：service_activity 表（service/action/detail/ts，写入时清 7 天前）；record_activity/get_recent_activities/get_queue_status；VideoQueue 打点(register/reanalyze/process_start/process_done/process_fail)+_current 当前处理跟踪+status()；PersonService 打点(merge_done/merge_skip)；api_gateway 新增 GET /api/oracle/activity（token 鉴权，返回 queue/db.by_status/rclone/person/model_calls/最近50条活动） +- **rclone_sync.sh**：同步结果写 activity 表（service=rclone, transferred/deleted/exit）；TRANS 提取正则修过一次（输出带 B 单位） +- **fam-ui**：新增"🖥 服务状态"页（导航第7项）：状态卡（队列运行/排队/完成/待处理/失败 + 当前处理视频 + rclone/人物/NAS同步/模型 4 张服务卡）+ 最近活动时间流（服务徽章），直连 Oracle /api/oracle/activity（复用 oracle_url+token） +- **踩坑**：video_queue.status() 用 Dict 注解未导入 → worker 启动 NameError → 补 typing.Dict +- 验证：接口数据正常（queue running/queued 19/当前 video 56；rclone sync_done/person merge_done 打点生效；401 鉴权；NAS→Oracle HTTP 200）；fam-ui 200 + +## 模型调用时间差 8 小时修复（commit 02fc80b） +- **根因**：gemini/nvidia adapter 的 model_calls.started_at 用 `datetime.now()`（Oracle 服务器 UTC），而 created_at 用 _now_iso()（北京时间）→ 前端模型统计/服务状态页时间差 8h +- **修复**：两个 adapter started_at 改 `datetime.now(timezone(timedelta(hours=8)))`；Oracle 历史 215 条 +8h；NAS 镜像 sync_model_calls 210 条 DATE_ADD +8h +- 验证：最新记录 started 16:05:02 与 created 对齐（北京时间）✓ + +## 人物管理无照片修复（commit 27bfd6a） +- **根因**：NAS sync_events 残留重分析前旧事件（223 条 vs Oracle 115 条，id 1-108 旧事件未删除——#8 删除传播未做的副作用）。人物页头像查 `ORDER BY e.id LIMIT 1` 取到旧事件 id=1 → Oracle ev_1.jpg 不存在（重分析后事件从 109 起）→ 404 → onerror 隐藏 → 无照片 +- **修复**：① NAS 清空 sync_events + 重置 cursor 全量重拉 → 精确 115 条镜像；② api_gateway event_thumb 兜底：ev 缺失时查所属 video 返回 thumbs/{video_id}.jpg（视频首帧）；③ 清理 Oracle people 脏 label（人物A（别名/标识：人物B）→人物A） +- 验证：NAS→Oracle ev109 HTTP 200/51KB ✓；人物页查询到的都是新事件 id → 截图存在 + +## 人物头像改为"出现事件画面"（commit 2ba3478，用户否决视频首帧方案） +- **用户意见**：不能用视频首帧当头像（首帧可能无人/非本人）——该人物在众多视频中多次出现，肯定能在他出现的事件里找到画面 +- **新方案**：Oracle 新增 GET /api/oracle/person/avatar?label=X（token 鉴权）：events.person_list_json 按 `%"label"%` JSON 数组精确匹配，按 id DESC 遍历返回第一个 ev_{id}.jpg 存在的截图（人物出现事件中最新的有截图画面）；撤销 event_thumb 的视频首帧兜底（改回 404） +- **fam-ui 人物页**：头像改调 avatar 接口（先 requests 探测 200 再渲染 img，label 用 urllib.parse.quote 编码） +- 验证：人物A/B avatar 均 200（39-41KB 真实事件画面）、无 token 401、NAS→Oracle 200、UI 200 + +## Oracle 磁盘扩容（用户控制台扩盘 + 重启生效） +- **背景**：rclone 持续同步视频导致 45G 盘写满（剩 46M，gdrive_videos 31G/75 个视频）。用户说已扩容但服务器 lsblk 一直 46.6G（在线未生效） +- **处理**：growpart/resize2fs 均 NOCHANGE（块设备没变）；临时清理 /tmp 残留+apt+journal 释放 ~900M；禁用 rclone timer 防写满；按用户要求重启服务器 +- **重启后扩容生效**：sda 46.6G→**150G**，Ubuntu cloud-init 开机自动扩展分区+文件系统 → df 146G 可用 90G（39%） +- **恢复**：enable --now rclone-sync.timer；同步恢复（视频 75→121 个持续下载中）；fam-edge active +- 经验：Oracle 在线扩容偶尔不立即生效，重启可触发（Ubuntu 自动 growfs）；扩容后无需手动 growpart +- 提醒：视频持续增长（121 个已占 56G），150G 约可再装 240 个视频，长期需考虑清理策略（如只保留分析完的摘要+删本地视频，需改 rclone 策略避免重新拉回）或再扩盘 + +## 人物模块重构 v3 - 大模型特征值替代 OpenCV（commit ff01d14 + 68337f8 + 677c5bd，用户提供设计文档） +- **核心**：prompt 增加 person_appearances(uid+7特征+action)，VLM 直接产出结构化特征，跨视频靠特征合并，彻底移除 cv2 +- **改动**（12 文件）：prompts.py(6原则+schema+规则8条+合并prompt重写)、gemini _normalize 透传、oracle_db(events.person_appearances_json/people.features_json/display_uid + _merge_features + upsert_person/mark_video_processed)、video_processor(validate_video 改 ffprobe 去 cv2、_store_result 聚合 uid 特征落 people、删 thumb/event_thumbs)、person_service(_aggregate_features/_collect_features_text 特征文本合并)、api_gateway 删 3 图接口、NAS db_layer+ddl 加字段、fam-ui 特征卡替代头像+事件人物特征块 +- **踩坑 2 个**： + 1. max_tokens 4096→16384：3秒密度+特征使 JSON 巨大被截断解析失败 + 2. **json_parser.validate_schema 白名单丢弃 person_appearances**（在 _normalize 之前执行）→ 补透传 +- **验证**：video 46 重分析 → lite 输出 16 事件含完整特征（性别男/中年/中等/短发/蓝色POLO衫/无辨识）；events 带特征 6、people.features_json 落库；NAS 同步 6 条+人物A特征卡数据 ✓ +- 部署：Oracle 卸载 opencv/numpy（ffprobe 已有）；NAS ALTER 加 3 列 +- 遗留：NVIDIA 模型链仍不可用（gemini 429 时 lite 兜底）；历史视频无特征（前端显示"特征待大模型补充"，下段分析自动补） diff --git a/.workbuddy/memory/2026-08-22.md b/.workbuddy/memory/2026-08-22.md new file mode 100644 index 0000000..a753df0 --- /dev/null +++ b/.workbuddy/memory/2026-08-22.md @@ -0,0 +1,214 @@ +# 2026-08-22 + +## 核查 SS Webhook 配置字段（sentinel-home-ai） +- 本机 SS 版本 9.3.0-12139（≥9.1.1），Webhook 动作设备可用。 +- 关键发现：**SS Webhook 与 motion_bp.py 接收端 schema 不匹配**。 + - SS Webhook 是「用户自定义参数名 + 模板变量」机制，Motion Detection 支持： + %EVENT_TIME%（字符串，非 epoch）、%DEVICE_NAME%（摄像头名文本，非 camera_id）、 + %EVENT_NAME%、%SERVER_NAME%、%THUMBNAIL_URL%。每事件单次 POST，body 为 form/JSON 的 key=value。 + - 接收端 motion_bp.py 期望 EventCenter.Event.List 风格：events:[{event_id(int 主键), + camera_id, event_type=10, start_time(epoch), duration, thumbnail_url}]。 + - 因 Webhook 无法提供 event_id，`record_motion_events` 中 `if eid is None: continue` + 导致真实 webhook 事件 100% 被静默丢弃。 +- 结论：轮询路径（MotionNotifier._fetch_events 走 EventCenter.Event.List）是唯一可靠数据源， + 已端到端验证。Webhook 如需启用，须改接收端做字段映射 + 与轮询去重（待用户拍板）。 + +## Webhook 驱动改造（去轮询）—— 已完成并端到端验证 +- 用户决策：**不要轮询，改用 SS Webhook 作为唯一数据源**。 +- 改造内容（commit ca6d425 / e52d8e9，已 push Gitea + 部署 NAS）： + - `motion_notifier.py`：`poll_enabled=false`（默认关），关闭轮询线程；新增 + `refresh_camera_map()`（启动时一次性从 SS Camera API 拉 name→id 映射）、 + `resolve_camera_id()`、`parse_ss_time_to_epoch()`（EVENT_TIME 字符串→epoch，按 NAS +8）、 + `_synth_event_id()`（device_name+event_time+thumbnail_url 哈希成稳定 int，幂等去重）、 + `build_event_from_webhook()`。 + - `motion_bp.py`：`/api/ss/webhook` 解析 JSON/表单、单条/数组，用 build_event_from_webhook + 映射后推送甲骨文；SS 端需按固定参数名配置（event_time/device_name/event_name/server_name/thumbnail_url）。 + - `config.yaml`：加 `camera_name_to_id: {"Generic_ONVIF-001": 2}` 兜底，删原 camera_ids 轮询项。 + - `app.py`：启动加载摄像头映射；`start_core.sh` 修正 APP_DIR 路径 bug 并显式 source .env。 +- 验证：模拟 JSON + 表单 + 重复推送，均 pushed=1；甲骨文 ss_motion_events 落库正确 + （camera_id=2, event_type=10, start_time epoch 北京时区对齐）；`has_motion_in_range_local` + 窗口命中 True / 远离 False；测试数据已清理。 +- **待用户操作**：在 SS「行動規則 → 事件=偵測到動作 → 動作=Webhook」配置，URL 填 + `http://127.0.0.1:8000/api/ss/webhook`，参数名严格用 event_time/device_name/event_name/ + server_name/thumbnail_url（值分别对应 %EVENT_TIME%/%DEVICE_NAME%/%EVENT_NAME%/%SERVER_NAME%/%THUMBNAIL_URL%）。 +- 注意：甲骨文 ss_motion_events 现存 21 条为历史轮询真实事件（2026-08-22 上午），轮询已停，不再新增。 + +## 「webhook→回查 SS 事件列表补全」链路测试（不写生产代码） +- 用户要求先测清整条链路再决定是否实现。测试结论：**思路完全可行，各段均真机验证通过**。 +- **链路 A（SS 事件列表真实字段）**：NAS 上 `SYNO.SurveillanceStation.EventCenter.Event` method=List + （camera_ids=2, event_types=10）真实返回 id/start_time/duration/thumbnail_url/thumbnail_dir。 + 实测 e.g. id=25536 dur=63s、id=25535 dur=95s、id=25538 dur=0。即 duration 真实存在（短动作 0~几秒，长动作数十秒）。 +- **链路 A2（webhook 触发时刻→补全匹配）**：模拟 webhook 触发时刻=事件 start_time（epoch）， + 用 ±120s 窗口找时间最接近事件，diff=0 精准匹配，补全出真实 event_id/start_time/duration/thumbnail。 + （注：真实 %EVENT_TIME% 是分钟级字符串，解析为 epoch 与真实 start_time 差 ≤ 数秒，120s 窗口足够覆盖。） +- **链路 B（补全结果推甲骨文落库）**：POST /api/ss/motion（body 含 token + events[{ + event_id,camera_id,event_type,start_time,duration,thumbnail_url}]）均 stored=1； + 甲骨文库 ss_motion_events 字段核对正确（event_id=25535 dur=95 / 25536 dur=63）。 +- **链路 C（预过滤）**：has_motion_in_range_local 窗口含事件=True、远离时段=False，逻辑正确。 +- 全部测试数据已清理，库恢复 21 条历史真实事件。 +- **未真机验证的一环**：SS 后台尚未配 Webhook，故「真实 webhook 触发」这步是模拟（用真实 start_time 当 trigger）。 + 其余 SS 查列表 / 推送甲骨文 / 落库 / 预过滤均为真机。 +- 下一步若实现：在 motion_notifier 加 enrich_event_from_ss()，webhook 收到后本机回查 SS 列表补全真实字段。 + +## 「能否输出对应运动视频」调研——可行，直接读录像文件 + ffmpeg 裁剪 +- 用户问运动事件能否输出对应视频片段。结论：**完全可行，且已真机裁剪出 22s MP4 演示**。 +- **SS API 导出路径全部不可用（SS 9.3.0-12139）**： + - `Recording.Export` 仅剩向导方法（CamEnum/CheckName/Save/Load/Delete，EventArchive.js），Save 需复杂参数（431）， + 非旧文档的 RangeExport（9.3 已无此方法，实测 103）。 + - `ThirdParty/Recording/Download/v1`（camId/startTime/endTime ISO 字符串）返回 401（需第三方授权，不可用）。 +- **关键发现：录像文件明文直读**（无需任何 API）： + - 路径：`/volume1/surveillance/<摄像头名>//<摄像头名>-YYYYMMDD-HHMMSS-<起始epoch毫秒>-.mp4` + - 每 30 分钟一个片段（约 377MB），h264 2880x1620@14.28fps + pcm_alaw(8kHz) 音频； + 文件名含起始 epoch 毫秒，可用 (camera_id, start_time) 直接定位（camera_id→目录名需 SS Camera API 映射）。 + - recording_encrypt.db 存在但当前录像为明文（未启用加密）。 +- **裁剪验证（真机）**：事件 id=25535（start=1787369601）落在 + `Generic_ONVIF-001-20260822-110859-1787368139533-7.mp4`（11:08:59 起 30min），偏移=1787369601-1787368139=1462s。 + `ffmpeg -ss 1462 -t 20 -i 片段 -an -c:v copy` → 22.13s / 4.4MB MP4（copy 模式从关键帧起，±秒级偏差；需精确可重编码）。 + **必须 -an**：pcm_alaw 音频无法封装进 mp4 容器，-c copy 会报 "Could not find tag for codec pcm_alaw"。 +- 注意：NAS ffmpeg 为老版本（/usr/bin/ffmpeg，不支持 -show_entries，用 `-i 2>&1 | grep Duration` 查时长）。 +- 下一步若实现：motion 事件补全 start_time/duration 后，NAS 端按 (camera, start) 定位片段 + ffmpeg 裁剪输出。 + +## Webhook vs 事件列表一致性核查（重要认知） +- 用户质疑：Webhook 与事件列表是否 1:1？事件列表有的会不会 webhook 没推？ +- 实测（NAS）：**ActionRule List = 0 条**（SS 后台尚未配任何行动规则 → 真实 webhook 零推送）； + 今日 camera2 运动事件 = 141 条（id 25405~25545），事件稀疏突发（35 个 >120s 断档，最长 4h 无动作）。 +- 机制结论：**Webhook ⊂ 事件列表，不是 1:1**。EventCenter 是全量事件记录；行动规则引擎按 + 「摄像头范围/时间段/去抖合并/开关/规则数」过滤后才触发 Webhook。故**事件列表有、webhook 没推 + 是完全可能的**，纯 Webhook 驱动存在漏事件 → 预过滤误杀风险。 +- 反方向（webhook 有、列表无）：罕见，仅事件落库毫秒级时序，±120s 窗口 + fallback 已兜底。 +- 对策待用户拍板：A. Webhook 实时 + 每 5 分钟增量对账补推（推荐，对账非事件轮询，只补差异）； + B. 接受漏风险（规则配全量摄像头+关去抖）；C. 回轮询（用户已否决）。 + +## Webhook 实现审查 + 用户最终决策：恢复轮询主路径（commit 7adc324，已部署验证） +- Webhook 实现审查发现 P0/P1/P2 问题：①合成 event_id 碰撞（同分钟多条事件若 thumbnail 为空 → 同 id 被 + UNIQUE 吞掉）；②推 Oracle 失败无重试直接丢；③无对账兜底（Webhook⊂事件列表）；④SS Webhook 请求格式 + 从未真机验证（行动规则 0 条）；⑤时间解析失败伪造为 now(UTC) 混时区；⑥camera_id=None 照常入库。 +- **用户拍板：放弃 Webhook，恢复轮询（简单稳定）**。改动： + - motion_notifier.py：poll_enabled 默认 true；_fetch_events limit 100→1000； + _init_cursor 优先续用 DB 游标（重启补推停机期间事件），DB 空才初始化为 SS 当前最大 id； + _poll_once 仅推送成功批次才前进游标（失败批次下轮窗口回看重试，不丢事件）； + docstring 改为轮询主路径说明。 + - config.yaml：poll_enabled: true，加 camera_ids: [2]，注释同步。 + - motion_bp.py：docstring 改为"Webhook 可选补充，非主路径"（端点保留）。 +- **端到端验证（真机）**：重启后游标续用 DB=25491 → 第一轮补推 55 条停机期间事件（25492~25546）、 + 第二轮增量 2 条（25547/25548）；Oracle ss_motion_events 总数 21+55+2=**78**，event_id 全为真实 SS id + （25471~25548），duration 真实分布（0~31s）；/api/ss/status：poll_enabled=true running=true + heartbeat_running=true pushed_total=57。 +- 结论：轮询路径现在简单稳定、不丢事件（重启补推 + 失败重试 + 真实字段），Webhook 降级为可选。 + +## 运动事件驱动架构 v3（commit a1523b4/8d6cfad/ffcc292，已部署验证） +- 用户决策：**不再处理整段视频**；timeline 事件轴/人物管理/统计全部用"运动时间处理后的数据"。前端不改。 +- 澄清：rclone 整段素材**必须保留**（在甲骨文按运动时间分割）；音频**保留转码**（aac）。 +- **实测纠正认知**：SS 事件 duration=0 只出现在动作进行中（查询时未结束），结束必为正数 + （id=25538 11:38 查 dur=0，12:13 查 dur=2；近 1h 43 条无一条已结束且 dur=0）。 + 故分割只处理已结束事件（start+duration ≤ now+grace 10s），进行中的下轮再分割。 +- **Oracle 端改动**：oracle_db 加 motion_event_id/camera_id 列 + get_motion_events_in_range/ + has_unfinished_motion_in_range/get_video_by_motion_event_id；video_processor 素材→分割/片段→分析 + 双分支（ffmpeg -c:v copy -c:a aac 保留音频，motion_event_id 幂等）；video_queue 片段入队； + config 加 motion_segment 块（clips_dir=/opt/fam-edge/motion_clips）。 +- **E2E 真机验证**：素材 327（11:08:59）→ 分割 35 段（event_id 25504-25538，dur 1-132s）→ 片段只分析 + （summary 真实：媳妇/爷爷/汤圆）→ events 绝对时间 ts → NAS 同步 → /api/ui/videos 显示 motion_ 片段 + （前端契约字段全在）→ /api/ui/videos/ 详情 events+people 正常 → /api/proxy/frame 帧图 200。 +- **踩坑**：① sqlite3.Row 无 .get()（vrow 访问用下标）；② ffmpeg subprocess args 首元素必须是可执行 + 文件（漏了 ffmpeg 路径 → "No such file or directory: '-y'"）；③ **Oracle fam-edge 由 systemd + fam-edge.service 守护（Restart=always）**——部署后必须 `sudo systemctl restart fam-edge`，手动 + setsid 启动会和守护打架（端口冲突 Connection in use）。 +- 旧整段分析的历史数据（319 个 done 素材）保留展示；新素材按新逻辑分割。 + +## 清理旧数据重提取 + 人物管理按运动视频重设计（commits 4cf4fc4/6a29e88/9664459） +- **用户决策**：甲骨文+NAS 后台数据全删（整段视频提取的旧结果 + 人物数据），按新框架运动视频重新提取。 +- **清理**：Oracle videos(364)/events(1687)/people(44)/model_calls/service_activity 全清 + motion_clips 目录清空 + （保留 ss_motion_events 93 条 + 素材文件 + 配置；先备份 oracle.db.bak）；NAS sync_videos/sync_events/sync_people/sync_cursor 全清。 +- **重提取**：systemctl restart fam-edge → producer 重新登记全部素材 → 分割（8/15-21 素材无运动事件→0 段； + 8/22 素材→运动片段）→ 片段分析 → 重新聚合人物（3 身份：爷爷/汤圆/媳妇）。 +- **问题**：历史素材分割 0 段会以"空会话"占满时间轴 → db_layer.get_sync_videos/get_sync_stats 加内容过滤 + （LEFT(filename,7)='motion_' OR EXISTS 有事件），前端契约不变。 +- **人物管理重设计**：PersonCard.vue 新增「运动片段」区块（缩略图/时间/摘要/事件数，点击跳 /timeline?video=）； + 后端新增 GET /api/ui/people/clips?label=（db_layer.get_sync_people_clips 按 label/canonical_name 匹配 + person_list_json → 关联运动片段，含 first_ts/clip_events）；Timeline.vue 支持 ?video= 定位。 +- **踩坑**：pymysql execute 用 % 做参数占位符，SQL 字面量含 'motion_%' 的 % 会报 + "unsupported format character" 500 → 改用 LEFT(filename,7)='motion_'。 +- 验证：/api/ui/videos 全为 motion_ 片段（15 条，含重新提取的 12:34 等新片段）；people 3 身份； + people/clips 返回媳妇 3 个片段。前端 dist 已构建部署（fam-ui/dist 不入库，构建产物单独 tar 部署）。 + +## 代码-文档一致性整理（commits 354ff18/01a8ac2，已 push） +- README 全量同步 v3：2.1 网络要点（轮询主路径/Webhook 可选）；2.2 拓扑图（Streamlit→Vue3、Video-Queue/分割）； + 3.1-3.3 模块表（MotionNotifier poll 主路径+游标/心跳、UI-API、people clips、Person-Service）；4.1 表清单 + （videos 加 motion_event_id/camera_id、ss_motion_events、sync_cursor 双游标）；5.1-5.2 API 表（/api/ss/motion、 + /api/oracle/frame|avatar、/api/ui/people-clips、/api/ui/* 全表、/api/proxy/*）；5.3 片段直传；6.1 历史标注； + 8.1-8.4 部署（systemd fam-edge、start_core.sh source .env、npm build、motion_segment config、数据清理指引）； + 9 快速开始；12 进度。 +- docs/DEPLOY.md 重写为 v3（Vue3 构建/部署、systemd、tar 管道、验证清单含 motion_clips）。 +- 删除过时 scripts/start_ui.sh（Streamlit）；start_edge.sh 注明 systemd 为准。 +- 补提交 fam-core/tests/test_motion_notifier.py。 +- 残留扫描：README 仅剩"不再依赖 Streamlit/从 Streamlit 迁移"正确表述。git 工作区干净（仅 .workbuddy/ 记忆）。 + +## 服务状态页「视频分割」状态卡（commit cbcc5d0，已部署） +- 用户需求：服务状态界面要能看到视频分割状态（前端口误选方向后澄清）。 +- Oracle：oracle_db.get_segment_status()（motion_event_id 片段 total/done/pending/failed + ss_motion_events 事件数 + + 最近 segment 活动）；api_gateway /api/oracle/activity 加 segment 字段 + clips_files（motion_clips 目录文件数）。 +- NAS：/api/ui/service-status 代理自动透传 segment（前端零后端改动）。 +- 前端：ServiceStatus.vue 新增「视频分割」ServiceCard（运动片段 done/total、文件数、待处理/失败、运动事件数、最近分割活动）。 +- 验证：Oracle activity.segment 返回 {total:75, done:54, pending:21, failed:0, clips_files:75, motion_events:99}； + NAS 代理 service-status 透传正常；前端 dist 已构建部署。 + +## 事件-分割一致性保障 + 历史事件回灌（commit 177c2a8） +- **用户质疑**：Oracle 的 ss_motion_events 是否=SS 全部事件？为何没有 8/22 之前的？ +- **实测**：SS 事件中心保留 8/15 起全部运动事件（按天 191/47/256/419/334/256/312/196 ≈ 2000 条）； + Oracle 只有 8/22 起的——NAS 游标初始化故意"不回灌历史"，属设计缺陷。 +- **修复 1（一致性缺口）**：SS 事件在动作进行中 duration=0，NAS 首次推送后游标前进不再重推 → + Oracle duration 永久 0 → 分割过滤跳过 → 事件丢失（此前被"重启重推"掩盖）。MotionNotifier 加 + `_zero_dur_ids`：推送过 duration<=0 的事件记入，下轮窗口回查已结束(duration>0)补推覆盖。 +- **修复 2（历史回灌）**：NAS 一次性脚本拉取 8/15 00:00~8/22 00:00 的 SS 运动事件（分 6h 段）→ + POST /api/ss/motion 幂等落库，**1815 条全部入库**（与按天统计吻合）；Oracle ss_motion_events 达 + 1947 条（min event_id=1）；重置 304 个 8/15-21 已 done 素材为 pending → 重启 Oracle 重新分割， + 历史运动片段持续生成（60s 内已 404 片段）。 +- **一致性保障清单**：①事件全覆盖（SS 全量→Oracle 幂等，回灌兜底）；②duration 真实（0→补推）； + ③游标续用/失败批次不前进（重启补推、不丢）；④分割按 event_id 幂等（motion_event_id 关联）； + ⑤start_time 对齐（SS epoch ↔ 素材文件名解析整秒，±0.5s 素材毫秒偏差可接受）。 +- **注意**：历史 1800+ 片段分析需大量 Gemini 配额（max_concurrent=1 + 429 重试），会排队跑很久。 + +## 事件-片段一致性对账（commit 1dd9df3，已部署） +- 用户重复追问"如何保证 SS 事件与分割视频一致（含 start/duration）"→ 补可验证的对账机制。 +- oracle_db.get_segment_consistency()：event_total（SS 全量 1947）/ finished（已结束可分割 1922）/ + segmented（已分割去重 1793）/ gap_count（缺口 129，动态收敛中）/ gaps 明细（event_id/start_time/duration）/ + material_range（素材覆盖 8/15 10:31~8/22 12:39）。 +- api_gateway activity.segment.consistency；前端服务状态页分割卡展示"一致性：事件/已结束/已分割/缺口"。 +- **缺口解读**：129 缺口主要是素材仍在队列处理（今天 13:0x 事件 + 少量历史 1222-1226），随分割收敛； + 真正无法分割的（素材窗口外/素材失败）会持续显示，可据此排查。 +- 一致性三层：数值（SS 原样+DO UPDATE+duration=0 补推）/数量（motion_event_id 幂等 1:1+对账缺口）/ + 文件（-ss offset -t dur，±0.5s 毫秒取整、±1-2s 关键帧对齐）。 + +## 人物管理数据彻底清除（用户要求，无代码改动） +- 用户："人物管理全是错的（历史数据原因），人物相关数据都删除，相关的都做下删除"。 +- **删除范围**：Oracle people 表(9)、videos.people_json(465)、events.person_list_json(683)/person_appearances_json(640)； + NAS sync_people(7)、sync_videos.people_json(397)、sync_events 两字段(424/416)。备份 oracle.db.bak.20260822_055139。 + 保留 videos(2124)/events(683)/ss_motion_events(1947)/chat_history。 +- **关键发现**：重启 Oracle 后 producer 会重新分析 **1646 个历史(8/15-21) pending 片段**并立即产生新人物 + （爷爷/人物A/人物B…），删了还会长 → 必须同时停掉历史片段分析：把 event_start_time < '2026-08-22 00:00:00' + 的 pending 片段标记 **skipped_reset**（不再分析、不产生人物），只保留 8/22 当天新运动片段分析提取人物。 +- 重启后 Oracle 正以 known_members=无 分析今日新片段 motion_25579（13:10 事件）→ 人物管理从今日数据重新提取 + （当前仅 1 个未命名"人物A"）。验证：/api/ui/people=1 group、named-members=空、视频详情 people_json=None。 + +## 外网访问 8000 + 登录校验（commits 9c51c65/aca1a67，已部署） +- 需求：NAS :8000 参考 3000 端口（Gitea）通过外网访问，且先登录才能访问（账号 ericwyuan / 密码 iLoveJava5）。 +- **外网机制（已查明）**：NAS 跑 frpc（/etc/frp/frpc.toml，S99frpc.sh 守护）→ 甲骨文 129.146.203.203:7000（frps）。 + 已有 gitea(外3000→NAS3000)、wordpress(外8500→NAS8088)。新增 fam-core proxy：外网 **8000→NAS 8000**， + S99frpc.sh restart 生效（proxy added: [fam-core gitea wordpress]）。甲骨文安全组 8000 已放行（实测 200）。 +- **登录校验（新模块 auth.py）**：POST /api/login（凭据 FAM_AUTH_USER/FAM_AUTH_PASS，NAS .env 默认 ericwyuan/iLoveJava5）、 + /api/logout、/api/auth/check、GET /login 内置深色登录页（SPA 零改动）；init_auth(app) 全局 before_request： + 页面未登录 302 /login、/api/* 未登录 401；白名单免登录：/login /api/login /api/logout /api/auth/check /health + /api/ss/webhook /assets/* /favicon.ico。登录态=进程内 token+HttpOnly cookie（7 天），重启 fam-core 需重新登录。 +- **验证（真机）**：内网+外网全链路：未登录 302→/login、登录页 200、API 401、错误密码 401、正确登录 200+cookie、 + 带 cookie 访问 /api/ui/videos 200（15 条运动片段）、/api/auth/check authed true/false、SS webhook 免登录 200。 + 外网 http://129.146.203.203:8000 完整流程通过。测试注入的假 webhook 事件已清理（ss_motion_events 1947）。 +- **注意**：SS Webhook（/api/ss/webhook）是唯一免登录入站端点（SS 无法带登录态），外网暴露后有被滥发假事件的 + 风险，如介意可后续给 webhook 加独立 token 或 frp secretKey；HTTP 明文传输，HTTPS 需另配。 + +## 恢复全部历史片段分析（用户决策，无代码改动） +- 用户："所有分割后没有处理的运动视频都需要处理" → 把之前删人物数据时标记的 **1646 个 skipped_reset 历史片段 + （8/15-21）全部改回 pending**（清空分析结果字段，文件齐全 0 缺失）。 +- 机制：video_queue.start() 的 `_enqueue_existing()`（get_pending_videos limit=10000）重启时把 pending 全部补入队； + 片段走 process_video 片段分支直接分析（motion_event_id 非空，local_path→motion_clips）。 +- 结果：systemctl restart fam-edge → 日志"启动恢复入队 1646 个待处理视频"，queued=1645 + current=1（motion_25148， + 8/21 09:33 事件），从 8/21 往 8/15 顺序处理。单并发 + Gemini 429 重试，预计跑数小时~十数小时。 diff --git a/.workbuddy/memory/MEMORY.md b/.workbuddy/memory/MEMORY.md new file mode 100644 index 0000000..0e14d8c --- /dev/null +++ b/.workbuddy/memory/MEMORY.md @@ -0,0 +1,30 @@ +# sentinel-home-ai 项目长期记忆 + +## 项目概况 +- 家庭多模态智能监控系统，仓库 `http://192.168.50.64:3000/ericwyuan/sentinel-home-ai`（Gitea 内网） +- 本地 `/Users/ericwyuan/Desktop/Work/sentinel-home-ai`（monorepo：fam-core NAS 端 + fam-edge Oracle 端 + fam-ui Vue3 前端） + +## 架构 v3（2026-08-22 定稿：运动事件驱动） +- **不再分析整段视频**。Google Drive --rclone--> Oracle gdrive_videos（整段素材保留）； + Oracle 按 NAS 推送的 `ss_motion_events`（start_time/duration，Unix epoch）ffmpeg 分割运动片段 + （`-c:v copy -c:a aac` 保留音频，产物 `/opt/fam-edge/motion_clips/`），**只分析运动片段**。 +- NAS MotionNotifier 轮询 SS `EventCenter.Event.List`（60s，camera_ids=2, event_types=10）推送 + /api/ss/motion；游标 MariaDB 续用 + 失败批次不前进 + 心跳。Webhook 端点保留为可选补充。 +- **SS duration=0 = 动作进行中**（结束才回填真实时长），分割只处理已结束事件。 +- 前端（fam-ui Vue3）不改：/api/ui/videos|stats|people|attention-events|... 契约保持； + 事件帧图 /api/proxy/frame 用绝对 ts − event_start_time 偏移取帧，运动片段天然兼容。 + +## 服务器与部署 +- NAS (192.168.50.64)：SSH 2222, ericwyuan/iLoveJava5；fam-core 代码 `/volume1/web/sentinel-home-ai`， + venv `fam-core/venv`，`start_core.sh` source .env；重启 = kill gunicorn + `setsid bash start_core.sh` + （NAS 无 systemd 守护）。 +- **Oracle (129.146.203.203)**：SSH key `~/.ssh/oracle_sentinel`；fam-edge 由 **systemd + `fam-edge.service` 守护（Restart=always）**——部署代码后必须 `sudo systemctl restart fam-edge`， + 手动 setsid 会端口冲突；venv `/opt/fam-edge/venv`；库 `/opt/fam-edge/data/oracle.db`； + 素材 `/opt/fam-edge/gdrive_videos`；logs 目录 ubuntu 可写。 +- 部署：本地 git 提交 push Gitea → tar 管道（NAS 直接解包；Oracle `--strip-components=1` 到临时目录再 cp，避免动 data/venv）。 + +## 提交规范（强制） +- 格式 `[阶段X.Y子任务号] 子任务名称 - 完成内容简述`；Bug 修复 `fix(模块): 问题简述` +- 一任务一 commit；开工先 `git pull --rebase`；阻塞先 commit 加 `[WIP]` +- 禁止 `git push --force` 和 `--no-verify` diff --git a/fam-core/config/config.yaml b/fam-core/config/config.yaml index 2347964..9ae52ac 100644 --- a/fam-core/config/config.yaml +++ b/fam-core/config/config.yaml @@ -31,6 +31,7 @@ oracle_sync: chat_handler: # 智能问答统一走 FAM-Edge 编排端点（Gemini → NVIDIA → 本地 Ollama 兜底） qa_url: "http://129.146.203.203:5000/api/edge/chat/ask" + qa_stream_url: "http://129.146.203.203:5000/api/edge/chat/ask/stream" # 流式版 timeout: 120 # 运动监测通知服务（2026-08-22 定稿：轮询主路径） diff --git a/fam-core/src/fam_core/app.py b/fam-core/src/fam_core/app.py index 62d835c..7820868 100644 --- a/fam-core/src/fam_core/app.py +++ b/fam-core/src/fam_core/app.py @@ -90,6 +90,21 @@ def status(): }), 200 +@app.route('/api/sync/trigger', methods=['POST']) +def sync_trigger(): + """手动立即触发一次甲骨文增量同步（服务状态页"立即同步"按钮）。 + + 正常情况下后台线程每 30 分钟自动拉一次；这个接口给用户想立刻看到最新数据 + 时用，跟后台线程共用同一把拉取锁（oracle_sync._pull_lock），不会并发重复拉。 + """ + if not _sync: + return jsonify({"error": "同步服务未初始化"}), 503 + ok = _sync.trigger_now() + if not ok: + return jsonify({"status": "failed", "error": _sync.status().get("last_error")}), 502 + return jsonify({"status": "ok", **_sync.status()}), 200 + + if __name__ == '__main__': cfg = load_config() port = cfg.get('server', {}).get('port', 8000) diff --git a/fam-core/src/fam_core/chat_handler/chat_handler.py b/fam-core/src/fam_core/chat_handler/chat_handler.py index c9b8a51..25587a2 100644 --- a/fam-core/src/fam_core/chat_handler/chat_handler.py +++ b/fam-core/src/fam_core/chat_handler/chat_handler.py @@ -13,7 +13,7 @@ Chat-Handler - Flask 蓝图，接收用户问答（新架构 v2） """ import json import requests -from flask import Blueprint, request, jsonify +from flask import Blueprint, request, jsonify, Response, stream_with_context from ..logger import setup_logger from ..config_loader import load_config @@ -115,6 +115,95 @@ def chat_ask(): }), 200 +@chat_bp.route('/api/chat/ask/stream', methods=['POST']) +def chat_ask_stream(): + """流式问答：SSE 逐块推送，边生成边显示。 + + 先立即推一条 context 事件（用了哪些 sync_events，本地查询很快，不用等 + 大模型）；再把甲骨文 /api/edge/chat/ask/stream 的分块原样转发给前端； + 最后一次性把拼好的完整回答写进 chat_history（跟非流式版一致）。 + """ + data = request.get_json(silent=True) + if not data: + return jsonify({"error": "Invalid JSON"}), 400 + + question = data.get('question', '') + queried_person = data.get('queried_person', '') + queried_date = data.get('queried_date', '') + if not question or not queried_person or not queried_date: + return jsonify({"error": "缺少必填字段: question, queried_person, queried_date"}), 400 + + logger.info(f"Chat(stream): person={queried_person}, date={queried_date}, question={question}") + rows = db_layer.query_sync_events_for_person_date(queried_person, queried_date) + + def sse(obj): + return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n" + + def generate(): + if len(rows) == 0: + answer = f"今天没有观察到{queried_person}。" + context_summary = "查询 sync_events 0 条" + yield sse({"type": "context", "count": 0, "summary": context_summary}) + yield sse({"type": "chunk", "provider": None, "text": answer}) + yield sse({"type": "done", "provider": None}) + db_layer.insert_chat_history( + user_question=question, ai_answer=answer, + context_summary=context_summary, + queried_date=queried_date, queried_person=queried_person) + return + + context = _format_events(rows) + context_summary = f"查询 sync_events {len(rows)} 条" + yield sse({"type": "context", "count": len(rows), "summary": context_summary, + "preview": context[:800]}) + + members = db_layer.get_sync_known_members_context() + prompt = build_chat_prompt( + context=context, members=members or queried_person, + question=question, queried_person=queried_person) + + cfg = load_config() + stream_url = cfg.get('chat_handler', {}).get( + 'qa_stream_url', 'http://129.146.203.203:5000/api/edge/chat/ask/stream') + timeout = cfg.get('chat_handler', {}).get('timeout', 120) + + full_answer = [] + provider_used = None + try: + resp = requests.post(stream_url, json={"prompt": prompt}, + timeout=(10, timeout), stream=True) + if resp.status_code != 200: + raise Exception(f"HTTP {resp.status_code}") + for line in resp.iter_lines(decode_unicode=True): + if not line or not line.startswith('data: '): + continue + yield line + '\n\n' # 原样转发给前端（已经是同样的 SSE 格式） + try: + obj = json.loads(line[len('data: '):]) + except ValueError: + continue + if obj.get('type') == 'chunk' and obj.get('text'): + full_answer.append(obj['text']) + provider_used = obj.get('provider') or provider_used + elif obj.get('type') == 'done': + provider_used = obj.get('provider') or provider_used + except Exception as e: + logger.error(f"流式问答代理失败: {e}") + if not full_answer: + yield sse({"type": "error", "message": "AI 服务暂时不可用，请稍后重试"}) + return + + answer = ''.join(full_answer).strip() or "抱歉，暂时无法生成回答。" + logger.info(f"流式问答由 {provider_used} 提供回答（长度={len(answer)}）") + db_layer.insert_chat_history( + user_question=question, ai_answer=answer, + context_summary=context_summary, + queried_date=queried_date, queried_person=queried_person) + + return Response(stream_with_context(generate()), mimetype='text/event-stream', + headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}) + + @chat_bp.route('/api/chat/history', methods=['GET']) def chat_history(): """获取对话历史""" diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py index 1560651..98441ee 100644 --- a/fam-core/src/fam_core/db_layer.py +++ b/fam-core/src/fam_core/db_layer.py @@ -370,6 +370,39 @@ def upsert_sync_model_calls(rows: List[Dict]) -> int: conn.close() +def upsert_sync_identity_map(rows: List[Dict]) -> int: + """批量 upsert Oracle 传来的 person_identity_map 增量（幂等，重复覆盖）。 + + 甲骨文不稳定，提取出的有效数据（含闭集人物识别结果/纠错记录）都要同步到 + NAS 防丢失——这张表跟 sync_people/sync_model_calls 走一样的镜像模式。 + """ + if not rows: + return 0 + conn = get_conn() + n = 0 + try: + cur = conn.cursor() + for r in rows: + cur.execute( + """INSERT INTO sync_identity_map + (id, video_id, raw_uid, canonical_name, source, updated_at, synced_at) + VALUES (%s,%s,%s,%s,%s,%s, NOW()) + ON DUPLICATE KEY UPDATE + video_id=VALUES(video_id), + raw_uid=VALUES(raw_uid), + canonical_name=VALUES(canonical_name), + source=VALUES(source), + updated_at=VALUES(updated_at), + synced_at=NOW()""", + (r.get('id'), r.get('video_id'), r.get('raw_uid'), + r.get('canonical_name'), r.get('source'), r.get('updated_at'))) + n += 1 + conn.commit() + return n + finally: + conn.close() + + def get_sync_model_calls(limit: int = 200) -> List[Dict]: """最近模型调用记录（前端统计展示）。""" conn = get_conn() diff --git a/fam-core/src/fam_core/member_manager/member_manager.py b/fam-core/src/fam_core/member_manager/member_manager.py index 954cbb0..58f8daf 100644 --- a/fam-core/src/fam_core/member_manager/member_manager.py +++ b/fam-core/src/fam_core/member_manager/member_manager.py @@ -142,3 +142,38 @@ def merge_member(): "display_name": m.get('canonical_name') or m['label'], } for m in members] }), 200 + + +@member_bp.route('/api/member/identity-correct', methods=['POST']) +def identity_correct(): + """事件时间轴"这个人识别错了"纠错入口（比 /api/member/name 粒度更细）。 + + 请求: {"video_id": 123, "current_name": "爷爷", "new_name": "爸爸"} + 只改这一段视频里被错误识别的那个人，不影响同名字符串在其他视频里的映射 + （人物 uid 只在单次视频分析内稳定，同一字符串在不同视频里可能是不同真人， + 不能像 /api/member/name 那样按全局 label 改）。 + """ + data = request.get_json(silent=True) + if not data: + return jsonify({"error": "Invalid JSON"}), 400 + + video_id = data.get('video_id') + current_name = (data.get('current_name') or '').strip() + new_name = (data.get('new_name') or '').strip() + if not video_id or not current_name or not new_name: + return jsonify({"error": "缺少必填字段: video_id, current_name, new_name"}), 400 + + logger.info(f"人物纠错: video_id={video_id} {current_name} -> {new_name}（回推 Oracle）") + ok, err = get_sync().push_identity_correct(video_id, current_name, new_name) + if not ok: + return jsonify({"error": f"回推 Oracle 失败: {err}"}), 502 + + try: + get_sync().trigger_now() + except Exception as e: + logger.warning(f"纠错后即时拉回失败（下一个周期会自动同步）: {e}") + + return jsonify({ + "status": "ok", "video_id": video_id, + "current_name": current_name, "new_name": new_name, + }), 200 diff --git a/fam-core/src/fam_core/oracle_sync.py b/fam-core/src/fam_core/oracle_sync.py index 4e921df..1007884 100644 --- a/fam-core/src/fam_core/oracle_sync.py +++ b/fam-core/src/fam_core/oracle_sync.py @@ -90,22 +90,25 @@ class OracleSync: events = data.get('events', []) or [] people = data.get('people', []) or [] model_calls = data.get('model_calls', []) or [] + identity_map = data.get('identity_map', []) or [] server_time = data.get('server_time', '') or '' n_videos = db_layer.upsert_sync_videos(videos) n_events = db_layer.upsert_sync_events(events) n_people = db_layer.upsert_sync_people(people) n_calls = db_layer.upsert_sync_model_calls(model_calls) + n_identity = db_layer.upsert_sync_identity_map(identity_map) if server_time: db_layer.set_sync_cursor(server_time) self._last_sync_at = datetime.now() self._last_error = None - self._last_count = (n_videos, n_events, n_people, n_calls) + self._last_count = (n_videos, n_events, n_people, n_calls, n_identity) logger.info( f"同步完成: videos+{n_videos} events+{n_events} people+{n_people} " - f"model_calls+{n_calls} since={since!r} -> server_time={server_time}") + f"model_calls+{n_calls} identity_map+{n_identity} since={since!r} " + f"-> server_time={server_time}") return True # ------------------------------------------------------------------ @@ -133,6 +136,39 @@ class OracleSync: logger.error(f"命名校正回推失败: {msg}") return False, msg + # ------------------------------------------------------------------ + def push_identity_correct(self, video_id: int, current_name: str, new_name: str): + """回推事件时间轴/人物管理"这个人识别错了"纠错到 Oracle。 + + 跟 push_name_correct 的区别：这个按 (video_id, 当前展示名) 定位，只改这 + 一段视频里错认的那个人，不影响同名字符串在其他视频里的映射（人物 uid + 只在单次视频分析内稳定，跨视频复用同一字符串完全可能是不同真人）。 + + 返回 (success: bool, error: str) + """ + try: + video_id = int(video_id) + except (TypeError, ValueError): + return False, "video_id 必须是数字" + current_name = (current_name or '').strip() + new_name = (new_name or '').strip() + if not current_name or not new_name: + return False, "缺少 current_name / new_name" + try: + resp = requests.post( + f"{self.base_url}/api/oracle/identity/correct", + json={"video_id": video_id, "current_name": current_name, + "new_name": new_name, "token": self.token}, + timeout=(10, 30)) + except requests.RequestException as e: + logger.error(f"人物纠错回推失败: {e}") + return False, str(e) + if resp.status_code == 200: + return True, "" + msg = f"HTTP {resp.status_code}: {resp.text[:200]}" + logger.error(f"人物纠错回推失败: {msg}") + return False, msg + # ------------------------------------------------------------------ def _run(self): logger.info(f"OracleSync 线程启动，间隔 {self.interval_sec}s，目标 {self.base_url}") diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index e83768a..5b5b3ad 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -66,6 +66,38 @@ motion_segment: min_duration_sec: 1 # 短于该时长的事件不分割 unfinished_grace_sec: 10 # start_time+duration 距当前 ≤ 该秒视为"已结束"容差 +# 闭集人物识别（2026-08-22 新增，家里固定 4 人：爷爷/爸爸/媳妇/汤圆）： +# 原来靠大模型自己编的"人物A/B/C"临时 uid + 文字特征描述跨视频合并，验证下来 +# 不可靠（用户原话"现在的识别全是错的"）。人脸向量方案也验证过，家庭监控这种 +# 大广角/远距离画面下同人内部相似度经常比不同人还低，此路不通。 +# 现在改成：汤圆(幼儿/儿童特征)/媳妇(唯一成年女性) 直接用 Gemini 已产出的 +# 性别/年龄字段判断，免费且验证下来接近 100% 准；爷爷/爸爸(两个成年男性，纯 +# 外观规则/人脸向量都区分不开) 改用视觉大模型"看参考图比对"——NVIDIA 实测 +# 6/6 全对且配额与主分析链路完全独立，设为优先，Gemini flash-lite(7/8) 兜底。 +# 参考图目录结构：{ref_dir}/爷爷/*.jpg、{ref_dir}/爸爸/*.jpg（已用确认过身份的 +# 历史截图种好，见 PROGRESS.md 记录）。 +person_identifier: + enabled: true + ref_dir: "/opt/fam-edge/data/person_refs" + max_ref_per_person: 6 + # 连续两次分类调用之间的最小间隔（不分 provider 统一限速）：正常处理新片段时 + # 调用本来就稀疏，这个主要是给历史数据批量回填用的，避免短时间内密集调用打爆配额 + min_call_interval_sec: 2 + nvidia: + api_key: "${NVIDIA_API_KEY}" + model_name: "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning" + fallback_models: [] # 目前只验证过这一个能用的 NVIDIA 视觉模型，先留好扩展位 + timeout: 60 + max_retries: 3 # 每个模型对瞬时故障（429/5xx/超时）最多重试几次（含首次） + retry_backoff_sec: 3 # 重试前基础等待秒数，指数退避（3s -> 6s -> 12s） + gemini: + api_key: "${GEMINI_API_KEY}" + extra_api_keys: ["${GEMINI_API_KEY_2}", "${GEMINI_API_KEY_3}", "${GEMINI_API_KEY_4}"] + model_name: "gemini-flash-lite-latest" + timeout: 60 + max_retries: 2 + retry_backoff_sec: 3 + # 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama models: - provider: "gemini" @@ -90,6 +122,9 @@ models: - "智能摄像头-3" - "智能摄像头-4" timeout: 600 + # 问答专用超时（跟上面视频分析的 timeout 分开）：用户在等交互式回答，一个 + # key/模型卡住不该等 10 分钟，超时要短，快速降级到下一个 key/模型/provider + chat_timeout: 20 # 模型级独立超时（最终值，不参与编排层 ×2 放大） # gemini-flash-lite 实测 ~22-34s，按用户要求放宽至 8 分钟（480s），避免大视频/排队时过早切断 model_timeouts: @@ -119,6 +154,7 @@ models: base_url: "https://integrate.api.nvidia.com/v1" api_key: "${NVIDIA_API_KEY}" timeout: 600 + chat_timeout: 20 # 问答专用超时，跟视频分析的 timeout 分开 max_base64_mb: 20 # 超过此大小直接跳过 NVIDIA，不做注定失败的编码+上传 switch_interval_sec: 5 # 模型切换间隔：一个失败后等待再试下一个（未来加模型时用） model_timeouts: # 模型级独立超时（最终值，不参与 ×2） 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 96b91f8..f488328 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -14,6 +14,7 @@ API-Gateway - Flask 蓝图（新架构 v3） 一次 Gemini 调用一并产出（见 ai_orchestrator/prompts.py），frame_service 不再 额外调用任何模型。NAS 经 core 代理读取，不在 NAS 做图像计算。 """ +import json import os from flask import Blueprint, request, jsonify, Response @@ -102,6 +103,39 @@ def people_correct(): return jsonify({"status": "ok", "label": label, "canonical_name": canonical}), 200 +@api_bp.route('/api/oracle/identity/correct', methods=['POST']) +def identity_correct(): + """事件时间轴/人物管理"这个人识别错了"纠错入口（比 people/correct 粒度更细）。 + + 请求: {"video_id": 123, "current_name": "爷爷", "new_name": "爸爸", "token": "..."} + 只改这一段视频里被错误识别的那个人，不影响同名字符串在其他视频里的映射—— + 人物 uid 只在单次视频分析内稳定，同一个"人物A"字符串在不同视频里可能是不同 + 真人，纠错必须落到 (video_id, 当前展示名) 这一粒度，不能按全局 label 改。 + 写 manual 来源，受保护不会被后续自动识别覆盖回去；立即重写这段视频的展示数据。 + """ + if not _check_token(): + return jsonify({"error": "unauthorized"}), 401 + data = request.get_json(silent=True) + if not data: + return jsonify({"error": "Invalid JSON"}), 400 + video_id = data.get('video_id') + current_name = (data.get('current_name') or '').strip() + new_name = (data.get('new_name') or '').strip() + if not video_id or not current_name or not new_name: + return jsonify({"error": "缺少 video_id / current_name / new_name"}), 400 + try: + video_id = int(video_id) + except (TypeError, ValueError): + return jsonify({"error": "video_id 必须是数字"}), 400 + try: + state.get_db().correct_video_identity(video_id, current_name, new_name) + except Exception as e: + logger.error(f"identity_correct 异常: {e}") + return jsonify({"error": str(e)}), 500 + return jsonify({"status": "ok", "video_id": video_id, + "current_name": current_name, "new_name": new_name}), 200 + + @api_bp.route('/api/edge/chat/ask', methods=['POST']) def chat_ask(): """智能问答编排：Gemini → NVIDIA → 本地 Ollama（两云端都失败才用本地兜底） @@ -125,6 +159,28 @@ def chat_ask(): return jsonify({"answer": answer, "provider": provider}), 200 +@api_bp.route('/api/edge/chat/ask/stream', methods=['POST']) +def chat_ask_stream(): + """智能问答编排（流式版）：SSE 逐块推送，边生成边显示，不用等全量回答。 + + 请求同 /api/edge/chat/ask。响应 Content-Type: text/event-stream， + 每行 `data: \\n\\n`，json 结构见 qa.QAOrchestrator.run_qa_stream 注释。 + """ + data = request.get_json(silent=True) + if not data or 'prompt' not in data: + return jsonify({"error": "缺少必填字段: prompt"}), 400 + + prompt = data['prompt'] + max_tokens = int(data.get('max_tokens', 1024)) + + def generate(): + for event in get_qa().run_qa_stream(prompt, max_tokens=max_tokens): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + + return Response(generate(), mimetype='text/event-stream', + headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}) + + @api_bp.route('/api/oracle/activity', methods=['GET']) def activity(): """实时服务状态 + 最近活动流（token 校验）。 diff --git a/fam-edge/src/fam_edge/frame_service.py b/fam-edge/src/fam_edge/frame_service.py index 8dafedc..adb9f76 100644 --- a/fam-edge/src/fam_edge/frame_service.py +++ b/fam-edge/src/fam_edge/frame_service.py @@ -91,7 +91,10 @@ def extract_frame(db, video_id: int, ts: str, width: int = FRAME_W) -> bytes: offset = 0.0 _ensure_dir() - cache = os.path.join(CACHE_DIR, f"frame_{video_id}_{int(offset)}.jpg") + # 缓存 key 必须带 width：同一 (video_id, offset) 不同调用方可能要不同分辨率 + # （时间轴缩略图 400px / 头像 600px / 人物识别裁人脸要接近原始分辨率 2880px）， + # 不带 width 会导致后来的高分辨率请求悄悄拿到早先缓存的低分辨率帧。 + cache = os.path.join(CACHE_DIR, f"frame_{video_id}_{int(offset)}_{width}.jpg") if os.path.isfile(cache) and os.path.getsize(cache) > 0: with open(cache, 'rb') as f: return f.read() 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 d67c33a..0af2860 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/base_adapter.py @@ -80,6 +80,15 @@ class BaseModelAdapter(ABC): raise NotImplementedError( f"{self.provider_name} 适配器未实现 chat()（不参与智能问答）") + def chat_stream(self, prompt: str, max_tokens: int = 512): + """流式问答：逐块 yield 文本增量。默认实现退化为"等 chat() 整段返回后 + 一次性当一个大 chunk 吐出"——子类没有真流式 API（或懒得接）时这样也能 + 用，只是没有逐字显示的效果；Gemini 有原生 SSE 流式接口，重写了这个方法。 + """ + result = self.chat(prompt, max_tokens=max_tokens) + if result: + yield result + @abstractmethod def get_timeout(self) -> int: """该模型的调用超时秒数""" 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 d18c4ed..0b26208 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py @@ -74,6 +74,10 @@ class GeminiAdapter(BaseModelAdapter): self.key_labels.append(str(label)) self.api_key = self.api_keys[0] if self.api_keys else '' # 向后兼容单 key 用法 self.timeout = config.get('timeout', 600) + # 问答（chat）专用超时——跟视频分析的 timeout 分开，不能共用 600s： + # 智能问答是同步等待用户看结果的交互场景，一个 key/模型卡住不该让用户等 + # 10 分钟，超时应该短、快速降级到下一个 key/模型/provider + self.chat_timeout = config.get('chat_timeout', 20) # 模型级独立超时（最终值，不参与编排层 ×N 放大）: {model_name: seconds} # 例: {"gemini-flash-lite-latest": 90}（按实测耗时 ×4 配置） self.model_timeouts = { @@ -431,10 +435,10 @@ class GeminiAdapter(BaseModelAdapter): "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens}}, - timeout=self.timeout + timeout=self.chat_timeout ) except requests.Timeout: - logger.warning(f"Gemini {key_label} [{model}] 问答超时") + logger.warning(f"Gemini {key_label} [{model}] 问答超时({self.chat_timeout}s)") continue except Exception as e: logger.error(f"Gemini {key_label} [{model}] 问答异常: {e}") @@ -452,6 +456,69 @@ class GeminiAdapter(BaseModelAdapter): continue return None + def chat_stream(self, prompt: str, max_tokens: int = 512): + """流式问答：逐块 yield 文本增量。用于聊天界面边生成边显示，不用等全量 + 返回再展示——之前整段等待是"卡住没反馈"体验差的根源之一。 + + 按 key 轮换 × 模型链依次尝试，但只在"这次尝试还没吐出任何文本"时才允许 + 换下一个 key/模型；一旦已经开始吐字给用户看了，中途出错就直接结束这次 + 生成（不再悄悄换 provider 接着写，否则会出现两段风格/内容不连贯的回答 + 拼在一起，比直接告知"生成中断"更让人困惑）。 + """ + if self._cb.is_open(): + logger.warning("Gemini 熔断器 OPEN，跳过问答（流式）") + return + if not self.api_keys: + logger.warning("Gemini API Key 未配置，跳过问答（流式）") + return + got_any = False + for idx, api_key in self._rotated_keys(): + key_label = self.key_labels[idx] + for model in self.model_chain: + try: + resp = requests.post( + f"{self._base_url}/models/{model}:streamGenerateContent" + f"?alt=sse&key={api_key}", + json={"contents": [{"parts": [{"text": prompt}]}], + "generationConfig": { + "temperature": 0.3, "maxOutputTokens": max_tokens}}, + timeout=self.chat_timeout, stream=True, + ) + except requests.Timeout: + logger.warning(f"Gemini {key_label} [{model}] 流式问答超时({self.chat_timeout}s)") + continue + except Exception as e: + logger.error(f"Gemini {key_label} [{model}] 流式问答异常: {e}") + continue + if resp.status_code != 200: + logger.warning(f"Gemini {key_label} [{model}] 流式问答 HTTP {resp.status_code}") + resp.close() + continue + try: + for line in resp.iter_lines(decode_unicode=True): + if not line or not line.startswith('data: '): + continue + chunk = line[len('data: '):] + try: + obj = json.loads(chunk) + except ValueError: + continue + cands = obj.get('candidates', []) + text = ''.join( + p.get('text', '') + for p in (cands[0].get('content', {}) if cands else {}).get('parts', [])) + if text: + got_any = True + yield text + except Exception as e: + logger.warning(f"Gemini {key_label} [{model}] 流式读取中断: {e}") + finally: + resp.close() + if got_any: + self._cb.record_success() + return # 已经开始吐字，不管这次是否读完都不再换 provider + self._cb.record_failure() + def get_timeout(self) -> int: return self.timeout diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py index 6be8c8a..9166c97 100644 --- a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py @@ -62,6 +62,8 @@ class NvidiaVisionAdapter(BaseModelAdapter): self.api_key = self._resolve_key(config.get('api_key', '')) self.base_url = config.get('base_url', 'https://integrate.api.nvidia.com/v1') self.timeout = config.get('timeout', 600) + # 问答专用超时，跟视频分析分开——交互式问答不该等到跟视频分析一样久 + self.chat_timeout = config.get('chat_timeout', 20) self.max_base64_mb = float(config.get('max_base64_mb', 20)) # 模型级独立超时（最终值，不参与编排层 ×N 放大）: {model_name: seconds} self.model_timeouts = { @@ -209,7 +211,7 @@ class NvidiaVisionAdapter(BaseModelAdapter): messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=max_tokens, - timeout=self.timeout + timeout=self.chat_timeout ) content = resp.choices[0].message.content if content: diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index e8c049d..19094fc 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -120,11 +120,21 @@ class OracleDB: thumbnail_url TEXT, received_at TEXT ); + CREATE TABLE IF NOT EXISTS person_identity_map ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + video_id INTEGER, + raw_uid TEXT, + canonical_name TEXT, + source TEXT, + updated_at TEXT, + UNIQUE(video_id, raw_uid) + ); CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at); CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id); CREATE INDEX IF NOT EXISTS idx_model_calls_created ON model_calls(created_at); CREATE INDEX IF NOT EXISTS idx_activity_ts ON service_activity(ts); CREATE INDEX IF NOT EXISTS idx_motion_window ON ss_motion_events(start_time, event_type); + CREATE INDEX IF NOT EXISTS idx_identity_map_video ON person_identity_map(video_id); """) # 兼容旧库：补 retry_count / file_valid / media 等列（生产-消费队列用） cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] @@ -599,6 +609,109 @@ class OracleDB: 'features_text': features_text, 'event_start_time': best['event_start_time']} + # ------------------------------------------------------------------ + # 人物对应关系表（2026-08-22 新增）：记录"某个视频里 Gemini 给的原始 uid" + # 与"闭集识别解析出的规范名"之间的映射，作为可追溯、可纠错的中间层。 + # + # 设计动机：events.person_appearances_json 里的 uid 只在单次视频分析内稳定， + # 同一字符串在不同视频里完全可能指向不同真人——不能靠"改 label 的 + # canonical_name"来纠错（一个 label 撞了多个真人，改一次就把另一个人也带歪 + # 了）。所以纠错必须落到 (video_id, raw_uid) 这一粒度，而不是全局 label。 + # + # events/videos 表里实际展示用的 person_list_json / person_appearances_json / + # people_json 会在识别（或纠正）时被直接重写成规范名（rewrite_event_person_names）， + # 保持"读的时候不用现查表拼接"的简单模型；这张表只作为"这次重写是怎么来的"的 + # 记录 + 纠错操作的定位依据，不参与展示时的实时查询。 + # ------------------------------------------------------------------ + def set_identity_mapping(self, video_id: int, raw_uid: str, + canonical_name: str, source: str = 'auto_id') -> bool: + """记录/更新 (video_id, raw_uid) -> canonical_name。manual 来源受保护， + 不会被后续自动识别结果（rule/auto_id）覆盖。返回是否真的发生了变化 + （调用方据此决定要不要顺带重写 events 展示数据）。""" + now = _now_iso() + row = self._conn.execute( + "SELECT canonical_name, source FROM person_identity_map " + "WHERE video_id=? AND raw_uid=?", (video_id, raw_uid)).fetchone() + if row: + if row['source'] == 'manual' and source != 'manual': + return False + if row['canonical_name'] == canonical_name and row['source'] == source: + return False + self._conn.execute( + "UPDATE person_identity_map SET canonical_name=?, source=?, updated_at=? " + "WHERE video_id=? AND raw_uid=?", + (canonical_name, source, now, video_id, raw_uid)) + else: + self._conn.execute( + "INSERT INTO person_identity_map " + "(video_id, raw_uid, canonical_name, source, updated_at) VALUES (?,?,?,?,?)", + (video_id, raw_uid, canonical_name, source, now)) + self._conn.commit() + return True + + def get_identity_map_for_video(self, video_id: int) -> Dict[str, str]: + rows = self._conn.execute( + "SELECT raw_uid, canonical_name FROM person_identity_map WHERE video_id=?", + (video_id,)).fetchall() + return {r['raw_uid']: r['canonical_name'] for r in rows if r['canonical_name']} + + def rewrite_event_person_names(self, video_id: int, rename_map: Dict[str, str]): + """按 {当前展示名: 新名} 把该视频全部 events 的 person_list_json / + person_appearances_json[].uid，以及 videos.people_json 里的名字替换掉。 + rename_map 的 key 是"事件数据里当前显示的名字"（可能是原始 uid，也可能是 + 上一轮已经替换过的规范名——纠错场景下就是这种情况）。 + """ + if not rename_map: + return + now = _now_iso() + with self._write_lock: + rows = self._conn.execute( + "SELECT id, person_list_json, person_appearances_json FROM events " + "WHERE video_id=?", (video_id,)).fetchall() + for r in rows: + changed = False + plist = json.loads(r['person_list_json'] or '[]') + new_plist = [rename_map.get(x, x) for x in plist] + if new_plist != plist: + changed = True + pa = json.loads(r['person_appearances_json']) if r['person_appearances_json'] else None + if pa: + for p in pa: + if isinstance(p, dict) and p.get('uid') in rename_map: + p['uid'] = rename_map[p['uid']] + changed = True + if changed: + self._conn.execute( + "UPDATE events SET person_list_json=?, person_appearances_json=? " + "WHERE id=?", + (json.dumps(new_plist, ensure_ascii=False), + json.dumps(pa, ensure_ascii=False) if pa is not None + else r['person_appearances_json'], + r['id'])) + vrow = self._conn.execute( + "SELECT people_json FROM videos WHERE id=?", (video_id,)).fetchone() + if vrow and vrow['people_json']: + plist = json.loads(vrow['people_json']) + new_plist = [rename_map.get(x, x) for x in plist] + if new_plist != plist: + self._conn.execute( + "UPDATE videos SET people_json=?, updated_at=? WHERE id=?", + (json.dumps(new_plist, ensure_ascii=False), now, video_id)) + self._conn.commit() + + def correct_video_identity(self, video_id: int, current_name: str, new_name: str): + """纠错入口（人物管理页 / 事件时间轴"修正"按钮都走这个）：把某视频里当前 + 展示为 current_name 的人物改成 new_name。写 manual 来源，受保护不会被后续 + 自动识别覆盖回去；同时立即重写这段视频的展示数据，不用等下一轮识别。""" + row = self._conn.execute( + "SELECT raw_uid FROM person_identity_map WHERE video_id=? AND canonical_name=?", + (video_id, current_name)).fetchone() + # 没有映射记录（比如这条数据是老流水线时代产出的，从没跑过闭集识别） + # 就把 current_name 本身当 raw_uid 存一条新映射 + raw_uid = row['raw_uid'] if row else current_name + self.set_identity_mapping(video_id, raw_uid, new_name, source='manual') + self.rewrite_event_person_names(video_id, {current_name: new_name}) + def get_events_for_label(self, label: str, limit: int = 6): """该人物（canonical_name 或 UID label）出现的候选事件，按时间倒序（最近优先）。 @@ -745,8 +858,11 @@ class OracleDB: 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': + # manual/auto_id 覆盖 llm；llm 不覆盖 manual/auto_id（auto_id 是闭集人物 + # 识别的确定性结论——比 llm 的文字特征合并猜测可靠得多，同样需要保护， + # 不能被后续 person_service 的 llm 合并跑批悄悄覆盖回去） + _protected = ('manual', 'auto_id', 'rule') + if source in _protected or row['source'] not in _protected: self._conn.execute( "UPDATE people SET canonical_name=?, source=?, appearances=appearances+1, " "features_json=?, display_uid=?, updated_at=? WHERE label=?", @@ -804,7 +920,8 @@ class OracleDB: now = _now_iso() row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone() if row: - if source == 'manual' or row['source'] != 'manual': + _protected = ('manual', 'auto_id', 'rule') + if source in _protected or row['source'] not in _protected: self._conn.execute( "UPDATE people SET appearances=?, source=?, updated_at=? WHERE label=?", (int(count), source, now, label)) @@ -853,6 +970,11 @@ class OracleDB: model_calls = self._conn.execute( "SELECT * FROM model_calls WHERE created_at >= ? ORDER BY id ASC", (since_iso,)).fetchall() + # 人物对应关系表（甲骨文不稳定，提取出的有效数据都要同步到 NAS 防丢失； + # 这张表是识别结果的可追溯记录 + 纠错依据，同样纳入增量同步） + identity_map = self._conn.execute( + "SELECT * FROM person_identity_map WHERE updated_at > ? ORDER BY id ASC", + (since_iso,)).fetchall() def _ser(row): d = dict(row) @@ -863,6 +985,7 @@ class OracleDB: "events": [_ser(e) for e in events], "people": [_ser(p) for p in people], "model_calls": [_ser(m) for m in model_calls], + "identity_map": [_ser(m) for m in identity_map], "server_time": _now_iso(), } diff --git a/fam-edge/src/fam_edge/person_identifier.py b/fam-edge/src/fam_edge/person_identifier.py new file mode 100644 index 0000000..8e4168a --- /dev/null +++ b/fam-edge/src/fam_edge/person_identifier.py @@ -0,0 +1,256 @@ +""" +PersonIdentifier - 闭集人物识别（家里固定 4 个人：爷爷/爸爸/媳妇/汤圆） + +背景（2026-08-22）: 原来靠大模型每次视频分析自己编的"人物A/B/C"临时 uid + 一段 +性别/年龄/衣着文字描述做跨视频合并，原理上就不可靠——文字描述会因光线/角度/换衣服 +对不上，反复出现张冠李戴（用户原话："现在的识别全是错的"）。人脸向量方案也验证 +过，家庭监控这种大广角/远距离/糊画面下同人内部相似度经常比不同人还低，此路不通。 + +现在改成基于已知这个家庭只有 4 个固定成员的闭集规则： + - 汤圆（幼儿/儿童）、媳妇（唯一成年女性）：Gemini 每次分析已经会标性别/年龄段， + 这两条命中率验证下来接近 100%，直接用，不需要额外模型调用。 + - 爷爷、爸爸（两个成年男性，纯外观规则/人脸向量都区分不开）：改用视觉大模型 + "看图比对"——给几张已确认身份的参考图 + 待判断的截图，直接问模型这是谁。 + 实测 NVIDIA nemotron-omni 在留出测试集上 6/6 全对，Gemini flash-lite 7/8， + NVIDIA 配额与 Gemini 完全独立、不跟主分析链路抢配额，设为优先。 + +调用粒度：每个运动片段（视频行）只调一次（不是每个事件都调）——同一段视频里 +人不会中途换衣服，取片段内最大 bbox 的成年男性外观代表整段。 + +健壮性（2026-08-22 补，用于支撑历史数据批量回填）: + - NVIDIA/Gemini 各自支持模型链（model_chain，fallback_models 可再加型号）+ + 每个模型独立重试（429/503/超时/连接错误这类瞬时故障，指数退避），非瞬时错误 + （400 参数错误等）不重试、直接换下一个模型/provider。 + - min_call_interval_sec 控制连续两次分类调用之间的最小间隔（不分 provider 统一 + 限速）——批量回填时会短时间内密集调用，需要限速避免打爆配额/被限流。 +""" +import base64 +import os +import re +import time +from typing import Optional + +import requests + +from .logger import setup_logger + +try: + from openai import OpenAI +except ImportError: + OpenAI = None + +logger = setup_logger('fam-edge.person_identifier') + +REF_DIR_DEFAULT = '/opt/fam-edge/data/person_refs' +ADULT_MALE_CANDIDATES = ('爷爷', '爸爸') +# HTTP 状态码：值得重试的瞬时故障（配额限流/服务过载），其余（400 参数错误/401 鉴权等）不重试 +_RETRYABLE_STATUS = (429, 500, 502, 503, 504) + + +class PersonIdentifier: + def __init__(self, config: dict): + self.enabled = bool(config.get('enabled', True)) + self.ref_dir = config.get('ref_dir', REF_DIR_DEFAULT) + self.max_ref_per_person = int(config.get('max_ref_per_person', 6)) + self.min_call_interval_sec = float(config.get('min_call_interval_sec', 2)) + self._last_call_at = 0.0 + + nv = config.get('nvidia', {}) + self.nvidia_model_chain = [nv.get('model_name', 'nvidia/nemotron-3-nano-omni-30b-a3b-reasoning')] + [ + m for m in nv.get('fallback_models', []) or [] if m] + self.nvidia_base_url = nv.get('base_url', 'https://integrate.api.nvidia.com/v1') + self.nvidia_api_key = self._resolve(nv.get('api_key', '${NVIDIA_API_KEY}')) + self.nvidia_timeout = int(nv.get('timeout', 60)) + self.nvidia_max_retries = int(nv.get('max_retries', 3)) + self.nvidia_retry_backoff = float(nv.get('retry_backoff_sec', 3)) + + gm = config.get('gemini', {}) + self.gemini_model = gm.get('model_name', 'gemini-flash-lite-latest') + raw_keys = [gm.get('api_key', '${GEMINI_API_KEY}')] + list(gm.get('extra_api_keys', []) or []) + self.gemini_api_keys = [k for k in (self._resolve(r) for r in raw_keys) if k] + self.gemini_timeout = int(gm.get('timeout', 60)) + self.gemini_max_retries = int(gm.get('max_retries', 2)) + self.gemini_retry_backoff = float(gm.get('retry_backoff_sec', 3)) + + self._refs = None # lazy: {person: [base64_str, ...]} + + @staticmethod + def _resolve(raw: str) -> str: + if isinstance(raw, str) and raw.startswith('${') and raw.endswith('}'): + return os.environ.get(raw[2:-1], '') + return raw + + def _load_refs(self): + if self._refs is not None: + return self._refs + refs = {} + for person in ADULT_MALE_CANDIDATES: + d = os.path.join(self.ref_dir, person) + files = [] + if os.path.isdir(d): + files = sorted(f for f in os.listdir(d) if f.lower().endswith(('.jpg', '.jpeg', '.png'))) + imgs = [] + for f in files[:self.max_ref_per_person]: + try: + with open(os.path.join(d, f), 'rb') as fh: + imgs.append(base64.b64encode(fh.read()).decode('ascii')) + except OSError: + continue + refs[person] = imgs + self._refs = refs + return refs + + def has_references(self) -> bool: + refs = self._load_refs() + return all(refs.get(p) for p in ADULT_MALE_CANDIDATES) + + def _pace(self): + """连续两次分类调用之间强制最小间隔，批量回填时避免短时间内打爆配额。""" + if self.min_call_interval_sec <= 0: + return + wait = self.min_call_interval_sec - (time.time() - self._last_call_at) + if wait > 0: + time.sleep(wait) + + def classify_adult_male(self, crop_bytes: bytes) -> Optional[str]: + """给一张成年男性截图，返回 '爷爷' / '爸爸'，判断不了返回 None（调用方保持原样不动）。 + + NVIDIA 优先（配额独立、实测更准，模型链+重试），失败/未配置则退回 Gemini + flash-lite（多 key 轮换+重试）。两边都失败返回 None——绝不瞎猜，宁可这次不 + 设置 canonical_name，留给下次（或人工在人物管理页确认）。 + """ + if not self.enabled or not self.has_references(): + return None + self._pace() + self._last_call_at = time.time() + result = self._classify_nvidia(crop_bytes) + if result: + return result + return self._classify_gemini(crop_bytes) + + def _build_prompt_and_images(self, crop_bytes: bytes): + refs = self._load_refs() + query_b64 = base64.b64encode(crop_bytes).decode('ascii') + images = [] # list of (b64, caption) + idx = 1 + for person in ADULT_MALE_CANDIDATES: + for b64 in refs.get(person, []): + images.append((b64, f'（上图是参考图{idx}，此人是：{person}）')) + idx += 1 + images.append((query_b64, '（上图是待判断的截图，请判断这是「爷爷」还是「爸爸」）')) + prefix = ('下面先给你几张参考图，每张图后面标了这个人是谁' + '（这户人家只有这两个成年男性，一个是爷爷，一个是爸爸）：') + suffix = ('只根据外观线索（体型/发型/衣着/姿态等）判断，用 JSON 回答，' + '格式：{"person":"爷爷或爸爸"}，不要输出其他内容。') + return prefix, images, suffix + + def _extract_json_person(self, text: str) -> Optional[str]: + text = (text or '').strip() + for cand in ('爷爷', '爸爸'): + if cand in text: + # 两个都出现时（比如复述了参考图说明）不采信，避免误判 + if '爷爷' in text and '爸爸' in text: + # 优先信 JSON 里 "person" 字段紧跟的那个 + m = re.search(r'"person"\s*:\s*"(爷爷|爸爸)"', text) + if m: + return m.group(1) + return None + return cand + return None + + # ------------------------------------------------------------------ + # NVIDIA：模型链 × 每个模型独立重试（瞬时故障退避重试，非瞬时故障直接换模型） + # ------------------------------------------------------------------ + def _classify_nvidia(self, crop_bytes: bytes) -> Optional[str]: + if not self.nvidia_api_key or OpenAI is None: + return None + prefix, images, suffix = self._build_prompt_and_images(crop_bytes) + if len(images) > 12: + images = images[-12:] # NVIDIA 单请求最多 12 张图，优先保留最新的参考+待判断图 + content = [{'type': 'text', 'text': prefix}] + for b64, caption in images: + content.append({'type': 'image_url', 'image_url': {'url': f'data:image/jpeg;base64,{b64}'}}) + content.append({'type': 'text', 'text': caption}) + content.append({'type': 'text', 'text': suffix}) + + client = OpenAI(base_url=self.nvidia_base_url, api_key=self.nvidia_api_key) + for model in self.nvidia_model_chain: + for attempt in range(self.nvidia_max_retries): + try: + resp = client.chat.completions.create( + model=model, + messages=[{'role': 'user', 'content': content}], + temperature=0.1, max_tokens=200, + timeout=self.nvidia_timeout) + text = resp.choices[0].message.content + person = self._extract_json_person(text) + if person: + logger.info(f"NVIDIA[{model}] 人物识别: {person}") + return person + logger.warning(f"NVIDIA[{model}] 返回结果无法解析出人物: {text[:100] if text else text}") + break # 解析不出人物是内容问题，不是瞬时故障，重试没用，换下一个模型 + except Exception as e: + status = getattr(getattr(e, 'response', None), 'status_code', None) + retryable = status in _RETRYABLE_STATUS or status is None + if retryable and attempt < self.nvidia_max_retries - 1: + backoff = self.nvidia_retry_backoff * (2 ** attempt) + logger.warning( + f"NVIDIA[{model}] 第 {attempt+1}/{self.nvidia_max_retries} 次失败" + f"（status={status}），{backoff:.1f}s 后重试: {e}") + time.sleep(backoff) + continue + logger.warning(f"NVIDIA[{model}] 失败（status={status}），换下一个模型: {e}") + break + return None + + # ------------------------------------------------------------------ + # Gemini：多 key 轮换 × 每个 key 独立重试 + # ------------------------------------------------------------------ + def _classify_gemini(self, crop_bytes: bytes) -> Optional[str]: + prefix, images, suffix = self._build_prompt_and_images(crop_bytes) + parts = [{'text': prefix}] + for b64, caption in images: + parts.append({'inline_data': {'mime_type': 'image/jpeg', 'data': b64}}) + parts.append({'text': caption}) + parts.append({'text': suffix}) + + for key in self.gemini_api_keys: + for attempt in range(self.gemini_max_retries): + try: + resp = requests.post( + f'https://generativelanguage.googleapis.com/v1beta/models/' + f'{self.gemini_model}:generateContent?key={key}', + json={'contents': [{'parts': parts}], + 'generationConfig': {'temperature': 0.1, 'maxOutputTokens': 200}}, + timeout=self.gemini_timeout) + data = resp.json() + if not data.get('candidates'): + err = data.get('error') or {} + status = resp.status_code + if status in _RETRYABLE_STATUS and attempt < self.gemini_max_retries - 1: + backoff = self.gemini_retry_backoff * (2 ** attempt) + logger.warning( + f"Gemini key 第 {attempt+1}/{self.gemini_max_retries} 次失败" + f"（status={status}），{backoff:.1f}s 后重试: {err}") + time.sleep(backoff) + continue + logger.warning(f"Gemini 人物识别失败（status={status}）: {err}") + break # 这个 key 不行了，换下一个 key + text = ''.join( + p.get('text', '') + for p in data['candidates'][0].get('content', {}).get('parts', [])) + person = self._extract_json_person(text) + if person: + logger.info(f"Gemini 人物识别: {person}") + return person + except requests.RequestException as e: + if attempt < self.gemini_max_retries - 1: + backoff = self.gemini_retry_backoff * (2 ** attempt) + logger.warning( + f"Gemini 网络异常，{backoff:.1f}s 后重试（第 {attempt+1}/" + f"{self.gemini_max_retries} 次）: {e}") + time.sleep(backoff) + continue + logger.warning(f"Gemini 人物识别异常: {e}") + break + return None diff --git a/fam-edge/src/fam_edge/qa.py b/fam-edge/src/fam_edge/qa.py index 9076b50..7cf8eb1 100644 --- a/fam-edge/src/fam_edge/qa.py +++ b/fam-edge/src/fam_edge/qa.py @@ -33,3 +33,34 @@ class QAOrchestrator: return answer, adapter.provider_name logger.info(f"QA {adapter.provider_name} 无返回，降级下一模型") return None, None + + def run_qa_stream(self, prompt: str, max_tokens: int = 1024): + """流式版：依次尝试各适配器的 chat_stream()，yield 结构化事件字典。 + + 事件类型: + {"type":"provider_trying","provider":p} 开始尝试这个 provider + {"type":"chunk","provider":p,"text":t} 这个 provider 吐出的文本增量 + {"type":"provider_failed","provider":p} 这个 provider 一个字都没吐出就失败，换下一个 + {"type":"done","provider":p} 成功结束（这个 provider 至少吐出过一块） + {"type":"all_failed"} 所有 provider 都失败 + + 跟 run_qa 一样"仅在还没吐出任何文本时才允许换下一个 provider"——一旦 + 开始给用户看字了，中途失败就结束这次生成，不再悄悄换源接着写。 + """ + for adapter in self.adapters: + yield {"type": "provider_trying", "provider": adapter.provider_name} + got_any = False + try: + for chunk in adapter.chat_stream(prompt, max_tokens=max_tokens): + if chunk: + got_any = True + yield {"type": "chunk", "provider": adapter.provider_name, "text": chunk} + except Exception as e: + logger.warning(f"QA {adapter.provider_name} 流式异常: {e}") + if got_any: + logger.info(f"QA 流式命中 provider={adapter.provider_name}") + yield {"type": "done", "provider": adapter.provider_name} + return + logger.info(f"QA {adapter.provider_name} 流式无返回，降级下一模型") + yield {"type": "provider_failed", "provider": adapter.provider_name} + yield {"type": "all_failed"} diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index a227598..ca410e4 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/fam-edge/src/fam_edge/video_processor.py @@ -15,6 +15,7 @@ import os import re import json import subprocess +import tempfile from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional @@ -22,6 +23,8 @@ from .logger import setup_logger from .config_loader import load_config from .model_adapters.adapter_factory import build_adapters from .model_adapters.base_adapter import BaseModelAdapter +from .person_identifier import PersonIdentifier +from . import frame_service from . import oracle_db logger = setup_logger('fam-edge.video_processor') @@ -200,6 +203,9 @@ class VideoProcessor: self.motion_keep_audio = bool(seg.get('keep_audio', True)) self.motion_min_duration = float(seg.get('min_duration_sec', 1)) self.motion_grace_sec = int(seg.get('unfinished_grace_sec', 10)) + # 闭集人物识别（家里固定 4 人）：汤圆/媳妇 用性别年龄规则；爷爷/爸爸 用 + # person_identifier 视觉大模型比对，每片段每个 uid 只调一次 + self.person_identifier = PersonIdentifier(self.config.get('person_identifier', {})) adapters = build_adapters(self.config.get('models', [])) self.vision_adapters: Dict[str, BaseModelAdapter] = { a.provider_name: a for a in adapters if a.get_role() == 'vision'} @@ -477,13 +483,111 @@ class VideoProcessor: elif k not in merged: merged[k] = v_str or 'unknown' uid_features[uid] = merged + # 闭集人物识别（家里固定 4 人）：汤圆/媳妇 用性别年龄规则免费识别（source=rule）； + # 爷爷/爸爸 每个 uid（同一片段内视为同一人，不逐事件重复调用）用视觉大模型 + # 比对一次（source=auto_id）。resolved: {原始 uid: (规范名, 来源)} + resolved = self._resolve_closed_set_identities(video_id, uid_features, norm_events) + + # 人物对应关系表：记录 (video_id, raw_uid) -> canonical_name，并把这段视频 + # 展示用的 events/videos 数据直接重写成规范名（读的时候不用现查表拼接）。 + # manual 纠正过的映射受保护，这里不会覆盖。 + rename_map = {} + for uid, (canonical, source) in resolved.items(): + if self.db.set_identity_mapping(video_id, uid, canonical, source=source): + rename_map[uid] = canonical + if rename_map: + self.db.rewrite_event_person_names(video_id, rename_map) + for p in people: if p and p not in ('无人', '无'): feats = uid_features.get(p) + canonical, source = resolved.get(p, ('', 'llm')) + # 已解析的人物直接用规范名作为 people 表的 label，跨视频天然汇总到 + # 同一行；解析不了的沿用原始 uid（跟旧行为一致，留给下次/人工确认） + label = canonical or p if feats: - self.db.upsert_person(p, source='llm', features=feats, display_uid=p) + self.db.upsert_person(label, canonical_name=canonical, source=source, + features=feats, display_uid=p) else: - self.db.upsert_person(p, source='llm') + self.db.upsert_person(label, canonical_name=canonical, source=source) logger.info(f"[video_id={video_id}] 已落库: summary={len(summary)}字, " f"events={len(norm_events)}, people={people}, " - f"with_features={len(uid_features)}") + f"with_features={len(uid_features)}, 闭集识别={resolved}") + + def _resolve_closed_set_identities(self, video_id: int, uid_features: Dict, + norm_events: List[Dict]) -> Dict[str, tuple]: + """闭集人物识别：返回 {uid: (canonical_name, source)}。 + + 汤圆（幼儿/儿童特征）、媳妇（唯一成年女性）靠 Gemini 已经产出的性别/年龄 + 字段直接判断（source='rule'），验证过命中率接近 100%，不需要额外模型调用。 + 爷爷/爸爸两个成年男性外观规则/人脸向量都区分不开（验证过），改用视觉大模型 + 比对参考图（source='auto_id'），每个 uid 只取本片段内最大 bbox 的一次出现 + 判断一次，不逐事件重复调用。 + """ + resolved: Dict[str, tuple] = {} + for uid, feats in uid_features.items(): + gender = str(feats.get('gender', '') or '').strip() + age_band = str(feats.get('age_band', '') or '').strip() + if age_band in ('幼儿', '儿童'): + resolved[uid] = ('汤圆', 'rule') + elif gender == '女': + resolved[uid] = ('媳妇', 'rule') + elif gender == '男': + crop = self._best_crop_for_uid(video_id, uid, norm_events) + if crop: + person = self.person_identifier.classify_adult_male(crop) + if person: + resolved[uid] = (person, 'auto_id') + return resolved + + def _best_crop_for_uid(self, video_id: int, uid: str, norm_events: List[Dict]) -> Optional[bytes]: + """取该 uid 在本片段里最大 bbox 的一次出现，裁剪成一张人物截图（jpeg bytes）。 + + bbox 缺失时（实测偶发：某些云端响应——尤其 flash-lite 兜底——没有带 + person_appearances.bbox 字段）回退到该 uid 第一次出现时刻的整帧居中裁剪， + 跟 build_avatar() 已有的兜底逻辑一致，好过直接放弃识别这个人。 + """ + best_ts, best_bbox, best_area = None, None, 0 + first_ts = None + for ev in norm_events: + for pa in ev.get('person_appearances', []): + if pa.get('uid') != uid: + continue + if first_ts is None: + first_ts = ev.get('timestamp') + bbox = pa.get('bbox') + if not bbox or len(bbox) != 4: + continue + ymin, xmin, ymax, xmax = bbox + area = max(0, ymax - ymin) * max(0, xmax - xmin) + if area > best_area: + best_area, best_ts, best_bbox = area, ev.get('timestamp'), bbox + if best_ts is None and first_ts is None: + return None + frame = frame_service.extract_frame(self.db, video_id, best_ts or first_ts, width=2880) + if frame is None: + return None + frame_service._ensure_dir() + fd, tmp = tempfile.mkstemp(suffix='.jpg', dir=frame_service.CACHE_DIR) + os.close(fd) + try: + with open(tmp, 'wb') as f: + f.write(frame) + size = frame_service._out_size(tmp) + if not size: + return None + if best_bbox is not None: + px = frame_service._bbox_to_pixels(best_bbox, *size) + ok = frame_service._crop_ffmpeg(tmp, px, 300) + else: + ok = frame_service._center_square_ffmpeg(tmp, 300) + if not ok: + return None + with open(tmp, 'rb') as f: + return f.read() + finally: + if os.path.exists(tmp): + try: + os.remove(tmp) + except OSError: + pass diff --git a/fam-edge/tests/test_frame_service.py b/fam-edge/tests/test_frame_service.py index ab1ef50..eb1f927 100644 --- a/fam-edge/tests/test_frame_service.py +++ b/fam-edge/tests/test_frame_service.py @@ -1,3 +1,6 @@ +import os + +from fam_edge import frame_service from fam_edge.frame_service import _bbox_to_pixels @@ -20,3 +23,52 @@ def test_bbox_to_pixels_full_frame(): def test_bbox_to_pixels_zero_area(): x1, y1, x2, y2 = _bbox_to_pixels([500, 500, 500, 500], 400, 300) assert (x1, y1) == (x2, y2) + + +class _FakeRow(dict): + """支持 row['key'] 访问的假 sqlite3.Row。""" + def __getitem__(self, k): + return dict.get(self, k) + + +class _FakeDb: + def __init__(self, local_path, event_start_time): + self._row = _FakeRow(local_path=local_path, event_start_time=event_start_time) + + def get_video_by_id(self, video_id): + return self._row + + +def test_extract_frame_cache_key_includes_width(tmp_path, monkeypatch): + """核心诉求: 同一 (video_id, ts) 不同调用方要不同分辨率（时间轴缩略图/头像/ + 人物识别裁人脸），缓存 key 不带 width 会导致后来的高分辨率请求悄悄拿到早先 + 缓存的低分辨率帧——这里验证两次不同 width 请求各自落到独立的缓存文件。""" + monkeypatch.setattr(frame_service, "CACHE_DIR", str(tmp_path)) + video_path = tmp_path / "fake_video.mp4" + video_path.write_bytes(b"not a real video, ffmpeg call is mocked") + db = _FakeDb(str(video_path), "2026-08-22 10:00:00") + + written_widths = [] + + def fake_run_ffmpeg(args, timeout=60): + # 把请求的 -vf scale=WIDTH:-2 记下来，往输出路径写点假数据模拟成功 + out_path = args[-1] + vf = next((a for a in args if a.startswith('scale=')), '') + written_widths.append(vf) + with open(out_path, 'wb') as f: + f.write(b'\xff\xd8fakejpeg') + return True + + monkeypatch.setattr(frame_service, "_run_ffmpeg", fake_run_ffmpeg) + + data_small = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=400) + data_large = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=2880) + + assert data_small is not None and data_large is not None + cache_files = sorted(os.listdir(tmp_path)) + frame_caches = [f for f in cache_files if f.startswith('frame_42_5_')] + assert len(frame_caches) == 2, f"expected 2 distinct cache files, got {frame_caches}" + assert 'frame_42_5_400.jpg' in frame_caches + assert 'frame_42_5_2880.jpg' in frame_caches + # 两次都真的各自调用了 ffmpeg（第二次没有因为撞到第一次的缓存而被跳过） + assert len(written_widths) == 2 diff --git a/fam-edge/tests/test_gemini_adapter.py b/fam-edge/tests/test_gemini_adapter.py index 60171e9..6c86410 100644 --- a/fam-edge/tests/test_gemini_adapter.py +++ b/fam-edge/tests/test_gemini_adapter.py @@ -108,3 +108,17 @@ def test_rotated_keys_single_key_never_errors(): a = GeminiAdapter(_cfg()) for _ in range(3): assert a._rotated_keys() == [(0, "key-primary")] + + +def test_chat_timeout_defaults_short_not_shared_with_video_timeout(): + """核心诉求: 问答是交互场景，不能沿用视频分析的 600s 超时——否则一个卡住 + 的 key/模型会让用户在聊天界面一直等，这正是"一直卡着"这个 bug 的根因。""" + a = GeminiAdapter(_cfg(timeout=600)) + assert a.timeout == 600 + assert a.chat_timeout == 20 + assert a.chat_timeout != a.timeout + + +def test_chat_timeout_configurable(): + a = GeminiAdapter(_cfg(chat_timeout=8)) + assert a.chat_timeout == 8 diff --git a/fam-edge/tests/test_oracle_db.py b/fam-edge/tests/test_oracle_db.py index 8d47046..1e2f005 100644 --- a/fam-edge/tests/test_oracle_db.py +++ b/fam-edge/tests/test_oracle_db.py @@ -181,3 +181,133 @@ def test_has_motion_in_range_local_filters_by_camera_id(tmp_path): [{"event_id": 1, "camera_id": 99, "event_type": 10, "start_time": 2500, "duration": 5}]) assert db.has_motion_in_range_local(2000, 3000, camera_id=2) is False assert db.has_motion_in_range_local(2000, 3000, camera_id=99) is True + + +# ---------------------------------------------------------------------- +# 人物对应关系表（video_id, raw_uid) -> canonical_name +# ---------------------------------------------------------------------- + +def _seed_video_with_events(db, filename="motion_1_1000.mp4"): + vid = db.ensure_video(filename, f"/tmp/{filename}", event_start_time="2026-08-22 10:00:00") + events = [ + {"timestamp": "10:00:01", "description": "在客厅走动", "people": ["人物A"], + "person_appearances": [{"uid": "人物A", "features": {"gender": "男"}, "action": "走动"}]}, + {"timestamp": "10:00:05", "description": "坐下", "people": ["人物A", "人物B"], + "person_appearances": [ + {"uid": "人物A", "features": {"gender": "男"}, "action": "坐下"}, + {"uid": "人物B", "features": {"gender": "女"}, "action": "站立"}]}, + ] + db.mark_video_processed(vid, "摘要", events, ["人物A", "人物B"], "gemini") + return vid + + +def test_set_identity_mapping_inserts_new_row(tmp_path): + db = _db(tmp_path) + assert db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") is True + assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} + + +def test_set_identity_mapping_updates_existing_non_manual_row(tmp_path): + db = _db(tmp_path) + db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") + assert db.set_identity_mapping(1, "人物A", "爸爸", source="auto_id") is True + assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"} + + +def test_set_identity_mapping_manual_protected_from_auto_overwrite(tmp_path): + """核心诉求: 人工纠正过的映射不能被后续自动识别悄悄改回去。""" + db = _db(tmp_path) + db.set_identity_mapping(1, "人物A", "爸爸", source="manual") + changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") + assert changed is False + assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"} + + +def test_set_identity_mapping_manual_can_override_manual(tmp_path): + db = _db(tmp_path) + db.set_identity_mapping(1, "人物A", "爸爸", source="manual") + changed = db.set_identity_mapping(1, "人物A", "爷爷", source="manual") + assert changed is True + assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} + + +def test_set_identity_mapping_no_change_returns_false(tmp_path): + db = _db(tmp_path) + db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") + changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") + assert changed is False + + +def test_get_identity_map_for_video_scoped_per_video(tmp_path): + """核心诉求: 同一个 raw_uid 字符串在不同视频里可能是不同真人，映射必须按 + video_id 隔离，不能串。""" + db = _db(tmp_path) + db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") + db.set_identity_mapping(2, "人物A", "爸爸", source="auto_id") + assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"} + assert db.get_identity_map_for_video(2) == {"人物A": "爸爸"} + + +def test_rewrite_event_person_names_updates_events_and_video(tmp_path): + db = _db(tmp_path) + vid = _seed_video_with_events(db) + db.rewrite_event_person_names(vid, {"人物A": "爷爷", "人物B": "媳妇"}) + + rows = db._conn.execute( + "SELECT person_list_json, person_appearances_json FROM events " + "WHERE video_id=? ORDER BY id", (vid,)).fetchall() + assert json.loads(rows[0]["person_list_json"]) == ["爷爷"] + pa0 = json.loads(rows[0]["person_appearances_json"]) + assert pa0[0]["uid"] == "爷爷" + assert json.loads(rows[1]["person_list_json"]) == ["爷爷", "媳妇"] + pa1 = json.loads(rows[1]["person_appearances_json"]) + assert {p["uid"] for p in pa1} == {"爷爷", "媳妇"} + + vrow = db._conn.execute("SELECT people_json FROM videos WHERE id=?", (vid,)).fetchone() + assert set(json.loads(vrow["people_json"])) == {"爷爷", "媳妇"} + + +def test_rewrite_event_person_names_noop_on_empty_map(tmp_path): + db = _db(tmp_path) + vid = _seed_video_with_events(db) + before = db._conn.execute( + "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() + db.rewrite_event_person_names(vid, {}) + after = db._conn.execute( + "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() + assert [r["person_list_json"] for r in before] == [r["person_list_json"] for r in after] + + +def test_correct_video_identity_end_to_end(tmp_path): + """核心诉求: 纠错入口应该找到当前展示名对应的映射行，改写映射 + 立即重写 + 展示数据，且标记为 manual（受保护）。""" + db = _db(tmp_path) + vid = _seed_video_with_events(db) + db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id") + db.rewrite_event_person_names(vid, {"人物A": "爷爷"}) + + db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸") + + assert db.get_identity_map_for_video(vid) == {"人物A": "爸爸"} + rows = db._conn.execute( + "SELECT person_list_json FROM events WHERE video_id=? ORDER BY id", (vid,)).fetchall() + assert json.loads(rows[0]["person_list_json"]) == ["爸爸"] + # manual 之后不能被自动识别覆盖回去 + changed = db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id") + assert changed is False + + +def test_correct_video_identity_without_prior_mapping_uses_current_name_as_raw_uid(tmp_path): + """核心诉求: 老流水线时代产出的数据从没跑过闭集识别，映射表里没有记录—— + 纠错依然要能生效，把 current_name 本身当 raw_uid 存一条新映射。""" + db = _db(tmp_path) + vid = db.ensure_video("motion_2_2000.mp4", "/tmp/x.mp4", event_start_time="2026-08-22 10:00:00") + events = [{"timestamp": "10:00:01", "description": "走动", "people": ["爷爷"], + "person_appearances": [{"uid": "爷爷", "features": {"gender": "男"}, "action": "走动"}]}] + db.mark_video_processed(vid, "摘要", events, ["爷爷"], "gemini") + + db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸") + assert db.get_identity_map_for_video(vid) == {"爷爷": "爸爸"} + rows = db._conn.execute( + "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall() + assert json.loads(rows[0]["person_list_json"]) == ["爸爸"] diff --git a/fam-edge/tests/test_person_identifier.py b/fam-edge/tests/test_person_identifier.py new file mode 100644 index 0000000..12bf42b --- /dev/null +++ b/fam-edge/tests/test_person_identifier.py @@ -0,0 +1,347 @@ +import os + +import pytest + +from fam_edge.person_identifier import PersonIdentifier + + +def _cfg(ref_dir, **overrides): + base = { + "enabled": True, + "ref_dir": ref_dir, + "max_ref_per_person": 6, + "min_call_interval_sec": 0, # 测试不需要真实限速，避免拖慢用例 + "nvidia": {"api_key": "nvkey", "model_name": "nvidia/test", "timeout": 30, + "max_retries": 2, "retry_backoff_sec": 0.01}, + "gemini": {"api_key": "gkey1", "model_name": "gemini-flash-lite-latest", "timeout": 30, + "max_retries": 2, "retry_backoff_sec": 0.01}, + } + base.update(overrides) + return base + + +def _write_refs(tmp_path, grandpa=2, dad=2): + for person, n in (("爷爷", grandpa), ("爸爸", dad)): + d = tmp_path / person + d.mkdir(parents=True, exist_ok=True) + for i in range(n): + (d / f"{i:02d}.jpg").write_bytes(b"fakejpegbytes") + + +class _FakeResp: + def __init__(self, status_code=200, payload=None): + self.status_code = status_code + self._payload = payload or {} + + def json(self): + return self._payload + + +@pytest.fixture(autouse=True) +def _no_real_sleep(monkeypatch): + """全部用例都不需要真的睡（限速/退避都测计数和结果，不测真实耗时）。""" + monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: None) + + +def test_has_references_false_when_dirs_missing(tmp_path): + pi = PersonIdentifier(_cfg(str(tmp_path / "refs"))) + assert pi.has_references() is False + + +def test_has_references_true_when_both_present(tmp_path): + _write_refs(tmp_path, grandpa=3, dad=2) + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.has_references() is True + + +def test_has_references_false_when_only_one_person_has_refs(tmp_path): + (tmp_path / "爷爷").mkdir(parents=True) + (tmp_path / "爷爷" / "01.jpg").write_bytes(b"x") + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.has_references() is False + + +def test_extract_json_person_clean_json(): + pi = PersonIdentifier(_cfg("/nonexistent")) + assert pi._extract_json_person('{"person":"爷爷"}') == '爷爷' + assert pi._extract_json_person('{"person":"爸爸"}') == '爸爸' + + +def test_extract_json_person_no_match_returns_none(): + pi = PersonIdentifier(_cfg("/nonexistent")) + assert pi._extract_json_person('不知道是谁') is None + assert pi._extract_json_person('') is None + + +def test_extract_json_person_both_mentioned_uses_json_field(): + """核心诉求: 模型有时会把参考图说明也复述一遍，回复里两个名字都出现—— + 这时候不能瞎猜，要从 JSON 的 person 字段里精确取，取不到就返回 None。""" + pi = PersonIdentifier(_cfg("/nonexistent")) + text = '参考图1是爷爷，参考图5是爸爸。{"person":"爸爸"}' + assert pi._extract_json_person(text) == '爸爸' + + +def test_extract_json_person_both_mentioned_no_json_field_returns_none(): + pi = PersonIdentifier(_cfg("/nonexistent")) + text = '这个人可能是爷爷，也可能是爸爸，不太确定' + assert pi._extract_json_person(text) is None + + +def test_classify_returns_none_when_disabled(tmp_path): + _write_refs(tmp_path) + pi = PersonIdentifier(_cfg(str(tmp_path), enabled=False)) + assert pi.classify_adult_male(b"crop") is None + + +def test_classify_returns_none_when_no_references(tmp_path): + pi = PersonIdentifier(_cfg(str(tmp_path / "empty"))) + assert pi.classify_adult_male(b"crop") is None + + +def test_classify_falls_back_to_gemini_when_nvidia_unavailable(tmp_path, monkeypatch): + """openai SDK 未安装时 NVIDIA 路径应该静默跳过（不报错），落到 Gemini。""" + _write_refs(tmp_path) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None) + + def fake_post(url, json=None, timeout=None): + return _FakeResp(200, { + "candidates": [{"content": {"parts": [{"text": '{"person":"爸爸"}'}]}}] + }) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.classify_adult_male(b"crop") == '爸爸' + + +def _fake_openai_factory(reply_text=None, exc=None, fail_times=0): + """构造一个假 OpenAI 客户端：先失败 fail_times 次再成功，或者一直抛 exc。""" + state = {"calls": 0} + + class FakeMessage: + content = reply_text + + class FakeChoice: + message = FakeMessage() + + class FakeChatResp: + choices = [FakeChoice()] + + class FakeCompletions: + def create(self, **kwargs): + state["calls"] += 1 + if state["calls"] <= fail_times: + raise (exc or RuntimeError("boom")) + if exc and fail_times == 0: + raise exc + return FakeChatResp() + + class FakeChat: + completions = FakeCompletions() + + class FakeOpenAI: + def __init__(self, base_url=None, api_key=None): + pass + chat = FakeChat() + + return FakeOpenAI, state + + +def test_classify_nvidia_success_skips_gemini(tmp_path, monkeypatch): + _write_refs(tmp_path) + FakeOpenAI, state = _fake_openai_factory(reply_text='{"person":"爷爷"}') + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI) + + gemini_called = {"n": 0} + def fake_post(url, json=None, timeout=None): + gemini_called["n"] += 1 + return _FakeResp(200, {}) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.classify_adult_male(b"crop") == '爷爷' + assert gemini_called["n"] == 0 + assert state["calls"] == 1 + + +class _FakeHTTPError(Exception): + def __init__(self, status_code): + self.response = type("R", (), {"status_code": status_code})() + + +def test_nvidia_retries_transient_error_then_succeeds(tmp_path, monkeypatch): + """核心诉求: 429/503 这类瞬时故障要退避重试，不是第一次失败就放弃换 provider。""" + _write_refs(tmp_path) + FakeOpenAI, state = _fake_openai_factory( + reply_text='{"person":"爸爸"}', exc=_FakeHTTPError(503), fail_times=1) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI) + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.classify_adult_male(b"crop") == '爸爸' + assert state["calls"] == 2 # 第一次 503 失败重试一次后成功 + + +def test_nvidia_gives_up_after_max_retries_falls_back_to_gemini(tmp_path, monkeypatch): + _write_refs(tmp_path) + FakeOpenAI, state = _fake_openai_factory(exc=_FakeHTTPError(503), fail_times=99) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI) + + def fake_post(url, json=None, timeout=None): + return _FakeResp(200, { + "candidates": [{"content": {"parts": [{"text": '{"person":"汤圆"}'}]}}] + }) + # 用一个不属于爷爷/爸爸的返回值只是为了确认真的调用到了 gemini 分支 + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + + cfg = _cfg(str(tmp_path)) + pi = PersonIdentifier(cfg) + pi.classify_adult_male(b"crop") + assert state["calls"] == pi.nvidia_max_retries # 重试到上限就放弃，不会无限重试 + + +def test_nvidia_non_retryable_error_gives_up_immediately(tmp_path, monkeypatch): + """核心诉求: 400 参数错误这类非瞬时故障，重试没有意义，应该立刻换下一个模型/provider， + 不要浪费时间重试一个注定失败的请求。""" + _write_refs(tmp_path) + FakeOpenAI, state = _fake_openai_factory(exc=_FakeHTTPError(400), fail_times=99) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", + lambda *a, **k: _FakeResp(500, {"error": "down"})) + pi = PersonIdentifier(_cfg(str(tmp_path))) + pi.classify_adult_male(b"crop") + assert state["calls"] == 1 # 400 不重试，一次就放弃这个模型 + + +def test_nvidia_falls_through_model_chain(tmp_path, monkeypatch): + """核心诉求: 第一个模型重试耗尽后，应该换模型链里的下一个型号再试，而不是 + 直接放弃整个 NVIDIA provider。""" + _write_refs(tmp_path) + calls = [] + + class FakeMessage: + def __init__(self, content): + self.content = content + + class FakeChoice: + def __init__(self, content): + self.message = FakeMessage(content) + + class FakeChatResp: + def __init__(self, content): + self.choices = [FakeChoice(content)] + + class FakeCompletions: + def create(self, model, **kwargs): + calls.append(model) + if model == 'nvidia/model-a': + raise _FakeHTTPError(503) + return FakeChatResp('{"person":"爷爷"}') + + class FakeChat: + completions = FakeCompletions() + + class FakeOpenAI: + def __init__(self, base_url=None, api_key=None): + pass + chat = FakeChat() + + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI) + cfg = _cfg(str(tmp_path), nvidia={ + "api_key": "nvkey", "model_name": "nvidia/model-a", + "fallback_models": ["nvidia/model-b"], "timeout": 30, + "max_retries": 2, "retry_backoff_sec": 0.01, + }) + pi = PersonIdentifier(cfg) + assert pi.classify_adult_male(b"crop") == '爷爷' + assert calls == ['nvidia/model-a', 'nvidia/model-a', 'nvidia/model-b'] + + +def test_gemini_retries_transient_error_on_same_key(tmp_path, monkeypatch): + _write_refs(tmp_path) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None) + calls = [] + def fake_post(url, json=None, timeout=None): + calls.append(url.split('key=')[-1]) + if len(calls) == 1: + return _FakeResp(503, {"error": {"code": 503}}) + return _FakeResp(200, { + "candidates": [{"content": {"parts": [{"text": '{"person":"媳妇"}'}]}}] + }) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + pi = PersonIdentifier(_cfg(str(tmp_path))) + pi.classify_adult_male(b"crop") + assert calls == ['gkey1', 'gkey1'] # 同一个 key 重试，不是立刻跳到下一个 key + + +def test_classify_gemini_rotates_across_keys_after_retries_exhausted(tmp_path, monkeypatch): + _write_refs(tmp_path) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None) + + calls = [] + def fake_post(url, json=None, timeout=None): + key = url.split('key=')[-1] + calls.append(key) + if key == 'gkey1': + return _FakeResp(429, {"error": {"code": 429}}) + return _FakeResp(200, { + "candidates": [{"content": {"parts": [{"text": '{"person":"爷爷"}'}]}}] + }) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + + cfg = _cfg(str(tmp_path), gemini={ + "api_key": "gkey1", "extra_api_keys": ["gkey2"], + "model_name": "gemini-flash-lite-latest", "timeout": 30, + "max_retries": 2, "retry_backoff_sec": 0.01, + }) + pi = PersonIdentifier(cfg) + assert pi.classify_adult_male(b"crop") == '爷爷' + assert calls == ['gkey1', 'gkey1', 'gkey2'] # gkey1 重试用尽才换 gkey2 + + +def test_classify_both_providers_fail_returns_none(tmp_path, monkeypatch): + """核心诉求: NVIDIA 和 Gemini 都失败时绝不能瞎猜，必须返回 None。""" + _write_refs(tmp_path) + monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None) + + def fake_post(url, json=None, timeout=None): + return _FakeResp(500, {"error": "boom"}) + monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post) + pi = PersonIdentifier(_cfg(str(tmp_path))) + assert pi.classify_adult_male(b"crop") is None + + +def test_env_var_credentials_resolved(tmp_path): + os.environ["TEST_NVIDIA_KEY_XYZ"] = "realkey" + try: + cfg = _cfg(str(tmp_path), nvidia={"api_key": "${TEST_NVIDIA_KEY_XYZ}"}) + pi = PersonIdentifier(cfg) + assert pi.nvidia_api_key == "realkey" + finally: + del os.environ["TEST_NVIDIA_KEY_XYZ"] + + +def test_max_ref_per_person_limits_loaded_refs(tmp_path): + _write_refs(tmp_path, grandpa=10, dad=10) + pi = PersonIdentifier(_cfg(str(tmp_path), max_ref_per_person=3)) + refs = pi._load_refs() + assert len(refs['爷爷']) == 3 + assert len(refs['爸爸']) == 3 + + +def test_pace_sleeps_when_called_too_soon(tmp_path, monkeypatch): + """核心诉求: 批量回填会短时间内密集调用，min_call_interval_sec 要真的限速， + 不能形同虚设。""" + _write_refs(tmp_path) + slept = [] + monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: slept.append(s)) + pi = PersonIdentifier(_cfg(str(tmp_path), min_call_interval_sec=5)) + pi._last_call_at = __import__("time").time() # 刚刚调用过 + pi._pace() + assert slept and slept[0] > 0 + + +def test_pace_no_sleep_when_interval_already_elapsed(tmp_path, monkeypatch): + _write_refs(tmp_path) + slept = [] + monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: slept.append(s)) + pi = PersonIdentifier(_cfg(str(tmp_path), min_call_interval_sec=5)) + pi._last_call_at = 0 # 很久以前 + pi._pace() + assert slept == [] diff --git a/fam-edge/tests/test_qa.py b/fam-edge/tests/test_qa.py new file mode 100644 index 0000000..56e005d --- /dev/null +++ b/fam-edge/tests/test_qa.py @@ -0,0 +1,81 @@ +from fam_edge.qa import QAOrchestrator + + +class _FakeAdapter: + def __init__(self, provider_name, chunks=None, raises=False): + self.provider_name = provider_name + self._chunks = chunks or [] + self._raises = raises + + def chat_stream(self, prompt, max_tokens=512): + if self._raises: + raise RuntimeError("boom") + for c in self._chunks: + yield c + + def chat(self, prompt, max_tokens=512): + return ''.join(self._chunks) or None + + +def _orchestrator(adapters): + qa = QAOrchestrator.__new__(QAOrchestrator) # 跳过 __init__（不需要真实 config/adapters） + qa.adapters = adapters + return qa + + +def test_run_qa_stream_first_provider_success(): + qa = _orchestrator([_FakeAdapter("gemini", chunks=["你", "好"])]) + events = list(qa.run_qa_stream("hi")) + types = [e["type"] for e in events] + assert types == ["provider_trying", "chunk", "chunk", "done"] + assert events[1]["text"] == "你" + assert events[2]["text"] == "好" + assert events[-1]["provider"] == "gemini" + + +def test_run_qa_stream_falls_back_when_first_yields_nothing(): + """核心诉求: 第一个 provider 一个字都没吐出来才允许换下一个——不是失败就切， + 是"完全没有产出"才切。""" + qa = _orchestrator([ + _FakeAdapter("gemini", chunks=[]), + _FakeAdapter("nvidia", chunks=["答案"]), + ]) + events = list(qa.run_qa_stream("hi")) + types = [e["type"] for e in events] + assert types == ["provider_trying", "provider_failed", "provider_trying", "chunk", "done"] + assert events[-1]["provider"] == "nvidia" + + +def test_run_qa_stream_does_not_switch_after_partial_output(): + """核心诉求: 已经开始吐字之后中途失败，不能悄悄换下一个 provider 接着写 + （会出现两段风格/内容不连贯的回答拼在一起）——直接结束这次生成。""" + class _PartialThenRaise: + provider_name = "gemini" + def chat_stream(self, prompt, max_tokens=512): + yield "先吐" + raise RuntimeError("connection reset") + + qa = _orchestrator([_PartialThenRaise(), _FakeAdapter("nvidia", chunks=["不该被用到"])]) + events = list(qa.run_qa_stream("hi")) + types = [e["type"] for e in events] + assert types == ["provider_trying", "chunk", "done"] + assert events[1]["text"] == "先吐" + assert events[-1]["provider"] == "gemini" + + +def test_run_qa_stream_all_providers_fail(): + qa = _orchestrator([ + _FakeAdapter("gemini", chunks=[]), + _FakeAdapter("nvidia", chunks=[], raises=True), + ]) + events = list(qa.run_qa_stream("hi")) + assert events[-1]["type"] == "all_failed" + assert "provider_failed" in [e["type"] for e in events] + + +def test_run_qa_stream_exception_treated_as_no_output(): + qa = _orchestrator([_FakeAdapter("gemini", raises=True), _FakeAdapter("nvidia", chunks=["ok"])]) + events = list(qa.run_qa_stream("hi")) + assert events[0] == {"type": "provider_trying", "provider": "gemini"} + assert events[1] == {"type": "provider_failed", "provider": "gemini"} + assert events[-1]["provider"] == "nvidia" diff --git a/fam-ui/src/api.js b/fam-ui/src/api.js index 2cf4050..73655b8 100644 --- a/fam-ui/src/api.js +++ b/fam-ui/src/api.js @@ -51,6 +51,13 @@ export const api = { request('/api/member/merge', { method: 'POST', body: JSON.stringify({ source, target }) }), status: () => request('/api/status'), + syncTrigger: () => request('/api/sync/trigger', { method: 'POST' }), + + identityCorrect: (video_id, current_name, new_name) => + request('/api/member/identity-correct', { + method: 'POST', + body: JSON.stringify({ video_id, current_name, new_name }), + }), } /** 剥离全角/半角括号备注（如 '人物A（别名：人物B）' -> '人物A'），与后端归一化一致 */ diff --git a/fam-ui/src/components/EventItem.vue b/fam-ui/src/components/EventItem.vue index f83daf1..b395b47 100644 --- a/fam-ui/src/components/EventItem.vue +++ b/fam-ui/src/components/EventItem.vue @@ -1,12 +1,39 @@