From 38dea6ba2ee8f659ecc20cdf09d8df58cd2cbb50 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Sat, 22 Aug 2026 10:25:56 +0800 Subject: [PATCH] =?UTF-8?q?[=E9=87=8D=E6=9E=84]=20=E8=BF=90=E5=8A=A8?= =?UTF-8?q?=E9=A2=84=E8=BF=87=E6=BB=A4=E6=94=B9=E4=B8=BA=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6(NAS=E6=8E=A8=E9=80=81)=20-=20=E7=94=B2?= =?UTF-8?q?=E9=AA=A8=E6=96=87=E4=B8=8D=E5=86=8D=E5=8F=8D=E5=90=91=E8=AE=BF?= =?UTF-8?q?=E9=97=AENAS=EF=BC=9A=E6=96=B0=E5=A2=9E=20ss=5Fmotion=5Fevents?= =?UTF-8?q?=20=E8=A1=A8/record=5Fmotion=5Fevents/has=5Fmotion=5Fin=5Frange?= =?UTF-8?q?=5Flocal=EF=BC=8C/api/ss/motion=20=E6=8E=A5=E6=94=B6=E7=AB=AF?= =?UTF-8?q?=E7=82=B9=EF=BC=8Cvideo=5Fprocessor=20=E6=94=B9=E7=94=A8?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E8=BF=90=E5=8A=A8=E4=BA=8B=E4=BB=B6=E9=A2=84?= =?UTF-8?q?=E8=BF=87=E6=BB=A4=EF=BC=8C=E5=81=9C=E7=94=A8=20dsm=5Fmotion=5F?= =?UTF-8?q?client=20=E5=8F=8D=E5=90=91=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fam-edge/config/config.yaml | 23 +++--- .../src/fam_edge/api_gateway/api_gateway.py | 30 ++++++++ fam-edge/src/fam_edge/dsm_motion_client.py | 12 ++- fam-edge/src/fam_edge/oracle_db.py | 76 +++++++++++++++++++ fam-edge/src/fam_edge/video_processor.py | 17 +++-- 5 files changed, 136 insertions(+), 22 deletions(-) diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index f64deed..086cff6 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -47,22 +47,17 @@ video_processing: # 降级顺序：先 gemini 整视频，失败再 nvidia 整视频；两者都失败 -> 标记 failed vision_order: ["gemini", "nvidia"] -# DSM 运动侦测预过滤（2026-08-22 新增）：分析前先查一下群晖 Surveillance Station -# 自己记录的运动侦测事件（SYNO.SurveillanceStation.EventCenter.Event，未公开文档的 -# 内部接口，参数名是下划线风格 camera_ids/start_time/end_time），这段时间窗口一条 -# 运动事件都没有就跳过云端分析（标记 done，compute_provider=skipped_no_motion）， -# 省掉长期无人时段白白消耗的 Gemini/NVIDIA 配额。 -# 账号密码走 .env（DSM_ACCOUNT/DSM_PASSWORD），不明文入库；查询失败/未配置一律 -# fail-open（照常送云端分析），不会因为这层可选优化漏检真实事件。 +# 运动侦测预过滤（2026-08-22 重构）： +# - 甲骨文【不再反向访问 NAS】。原 dsm_motion_client（甲骨文主动查 SS API）已停用， +# 改由 NAS 端 fam-core 的 MotionNotifier 轮询/接收 SS 事件后，主动 POST 推送到 +# 甲骨文 /api/ss/motion，落库 ss_motion_events。 +# - video_processor 分析前调用 db.has_motion_in_range_local()（本地运动事件表）做 +# 预过滤：窗口内无运动事件则跳过云端分析（compute_provider=skipped_no_motion）。 +# - 本地运动事件表为空（冷启动/尚未收到推送）一律 fail-open（照常分析），不漏检。 +# 此 block 仅保留 min_motion_seconds 语义参考；host/account 等反向访问字段已弃用。 dsm_motion_prefilter: - enabled: true - host: "192.168.50.64" - port: 5000 - account: "${DSM_ACCOUNT}" - password: "${DSM_PASSWORD}" - camera_id: 2 # Surveillance Station 里 Generic_ONVIF-001 的 camera_id + enabled: false # 已停用：甲骨文不再主动访问 NAS SS min_motion_seconds: 0 # 窗口内运动事件总时长需 ≥ 此值才算"有运动"（0=有事件就算） - timeout_sec: 10 # 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama models: 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 f4dae42..4d24e96 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -197,6 +197,36 @@ def oracle_avatar(): return Response(data, mimetype='image/jpeg') +@api_bp.route('/api/ss/motion', methods=['POST']) +def ss_motion(): + """接收 NAS 推送的运动侦测事件（单向：NAS -> Oracle）。 + + 请求体: {"token": "...", "events": [ + {"event_id": 25349, "camera_id": 2, "event_type": 10, + "start_time": 1787318023, "duration": 149, "thumbnail_url": "107471,12075"} + ]} + start_time/duration 为 Unix epoch（与 SS 同源，时区无关）。 + 落库 ss_motion_events（按 event_id 幂等），供 video_processor 本地预过滤使用。 + """ + if not _check_token(): + return jsonify({"error": "unauthorized"}), 401 + data = request.get_json(silent=True) + if not data or 'events' not in data: + return jsonify({"error": "缺少 events"}), 400 + events = data.get('events') or [] + if not isinstance(events, list): + return jsonify({"error": "events 必须是数组"}), 400 + try: + stored = state.get_db().record_motion_events(events) + except Exception as e: + logger.error(f"ss_motion 落库异常: {e}") + return jsonify({"error": str(e)}), 500 + if stored: + state.get_db().record_activity( + 'motion', 'push', f"接收 NAS 运动事件 {stored} 条") + return jsonify({"status": "ok", "received": len(events), "stored": stored}), 200 + + @api_bp.route('/health', methods=['GET']) def health(): """健康检查：DB 连通性 + producer/consumer 线程存活状态。 diff --git a/fam-edge/src/fam_edge/dsm_motion_client.py b/fam-edge/src/fam_edge/dsm_motion_client.py index 9de698c..6833120 100644 --- a/fam-edge/src/fam_edge/dsm_motion_client.py +++ b/fam-edge/src/fam_edge/dsm_motion_client.py @@ -1,5 +1,15 @@ """ -DsmMotionClient - 查询群晖 Surveillance Station 的运动侦测事件（预过滤用） +[已弃用 / DEPRECATED] 本模块自 2026-08-22 起不再被调用。 + +架构约束：甲骨文 FAM-Edge 不得反向访问 NAS。原 video_processor 在此处主动查询 +NAS 的 Surveillance Station（甲骨文 -> NAS），违反该约束，已停用。 + +替代方案：由 NAS 端 fam-core 的 MotionNotifier 轮询/接收 SS 事件后，主动 POST +推送到甲骨文的 /api/ss/motion，落库 ss_motion_events；video_processor 改用 +db.has_motion_in_range_local() 做本地预过滤。本文件保留仅作参考，请勿再实例化。 + +--- +DsmMotionClient（旧实现，仅作历史参考） - 查询群晖 Surveillance Station 的运动侦测事件（预过滤用） 背景: fam-edge 原来不管这段 30 分钟录像里有没有人走动，一律整段送云端 VLM 分析， 配额/耗时都花在长期无人的空转时段上。DSM 的 Surveillance Station 用 diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index 2b87f8c..589074d 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -110,10 +110,21 @@ class OracleDB: detail TEXT, ts TEXT ); + CREATE TABLE IF NOT EXISTS ss_motion_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_id INTEGER UNIQUE, + camera_id INTEGER, + event_type INTEGER, + start_time INTEGER, + duration INTEGER, + thumbnail_url TEXT, + received_at TEXT + ); 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); """) # 兼容旧库：补 retry_count / file_valid / media 等列（生产-消费队列用） cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] @@ -193,6 +204,71 @@ class OracleDB: "ORDER BY id DESC LIMIT ?", (int(limit),)).fetchall() return [dict(r) for r in rows] + # ------------------------------------------------------------------ + # 运动侦测事件（NAS 推送，单向：NAS -> Oracle，不再反向访问 NAS） + # 由 fam-core 的 MotionNotifier 轮询/接收 SS 事件后 POST 到 /api/ss/motion。 + # ------------------------------------------------------------------ + def record_motion_events(self, events: List[Dict]) -> int: + """批量 upsert NAS 推送来的运动事件（按 event_id 幂等）。返回成功条数。""" + n = 0 + now = _now_iso() + for e in (events or []): + eid = e.get('event_id') + if eid is None: + continue + self._conn.execute( + """INSERT INTO ss_motion_events + (event_id, camera_id, event_type, start_time, duration, + thumbnail_url, received_at) + VALUES (?,?,?,?,?,?,?) + ON CONFLICT(event_id) DO UPDATE SET + camera_id=excluded.camera_id, + event_type=excluded.event_type, + start_time=excluded.start_time, + duration=excluded.duration, + thumbnail_url=excluded.thumbnail_url, + received_at=excluded.received_at""", + (int(eid), e.get('camera_id'), e.get('event_type'), + e.get('start_time'), e.get('duration'), + e.get('thumbnail_url'), now)) + n += 1 + if n: + self._conn.commit() + return n + + def get_recent_motion_events(self, limit: int = 50) -> List[Dict]: + """最近运动事件（按 start_time 倒序）。""" + rows = self._conn.execute( + "SELECT id, event_id, camera_id, event_type, start_time, duration, " + "thumbnail_url, received_at FROM ss_motion_events " + "ORDER BY start_time DESC LIMIT ?", (int(limit),)).fetchall() + return [dict(r) for r in rows] + + def has_motion_in_range_local(self, start_ts: int, end_ts: int, + camera_id: int = None) -> Optional[bool]: + """本地运动预过滤：判断 [start_ts, end_ts] 窗口内是否存在运动事件。 + + 替代原 dsm_motion_client 反向访问 NAS 的做法。返回: + - None : 本地运动事件表为空（冷启动，尚未收到 NAS 推送），调用方 + 必须 fail-open（照常送云端分析），不能当作"无运动"跳过。 + - True/False : 窗口内确有/确无运动事件。 + start_ts/end_ts 为 Unix epoch（与 SS 事件 start_time 同源，时区无关）。 + """ + total = self._conn.execute( + "SELECT COUNT(*) c FROM ss_motion_events").fetchone()['c'] + if total == 0: + return None + sql = ("SELECT COUNT(*) c FROM ss_motion_events " + "WHERE event_type = 10 " + "AND start_time <= ? " + "AND (start_time + COALESCE(duration,0)) >= ?") + params = [end_ts, start_ts] + if camera_id is not None: + sql += " AND camera_id = ?" + params.append(camera_id) + cnt = self._conn.execute(sql, params).fetchone()['c'] + return cnt > 0 + def get_queue_status(self) -> Dict: """实时队列/处理状态（前端服务状态卡用）。""" total = self._conn.execute("SELECT COUNT(*) c FROM videos").fetchone()['c'] diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index 30855f2..d7cf601 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/fam-edge/src/fam_edge/video_processor.py @@ -22,7 +22,6 @@ 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 .dsm_motion_client import DsmMotionClient from . import oracle_db logger = setup_logger('fam-edge.video_processor') @@ -194,7 +193,6 @@ class VideoProcessor: 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'} - self.dsm_motion = DsmMotionClient(self.config.get('dsm_motion_prefilter', {})) def _ordered_vision_adapters(self) -> List[BaseModelAdapter]: ordered = [] @@ -238,18 +236,23 @@ class VideoProcessor: if event_start: self.db.set_event_start_time(video_id, event_start) - # DSM 运动侦测预过滤：这段时间窗口里群晖自己记录的运动事件一条都没有， - # 就跳过云端分析（省配额）。查询本身失败/未配置一律 fail-open（照常分析）， + # 运动侦测预过滤（v2: 改用 NAS 推送来的本地运动事件，不再反向访问 NAS）： + # 这段时间窗口里没有任何运动事件，就跳过云端分析（省配额）。 + # 本地运动事件表为空（冷启动，尚未收到 NAS 推送）一律 fail-open（照常分析）， # 绝不能因为这层可选优化漏检真实事件。 duration_sec = (vmeta or {}).get('duration_sec') if self.file_validate else None if event_start and duration_sec: try: - start_dt = datetime.strptime(event_start, '%Y-%m-%d %H:%M:%S') - has_motion = self.dsm_motion.has_motion_in_range(start_dt, duration_sec) + # event_start 是北京时间墙钟，转成与 SS 事件同源的 Unix epoch + start_dt = datetime.strptime(event_start, '%Y-%m-%d %H:%M:%S').replace( + tzinfo=timezone(timedelta(hours=8))) + start_ts = int(start_dt.timestamp()) + end_ts = start_ts + int(duration_sec) + has_motion = self.db.has_motion_in_range_local(start_ts, end_ts) except ValueError: has_motion = None if has_motion is False: - logger.info(f"[video_id={video_id}] DSM 运动预过滤：该时段无运动，跳过云端分析") + logger.info(f"[video_id={video_id}] 运动预过滤：该时段无运动，跳过云端分析") self.db.mark_video_processed( video_id, "（自动跳过：该时段未检测到运动）", [], [], 'skipped_no_motion') return True