refactor: 全量迁云——fam-core 直读甲骨文 SQLite，NAS 只剩推送进程
前一天刚把登录迁到甲骨文，隔天 NAS 上的 fam-core 又挂了导致数据接口 502。 盘点后确认：甲骨文的 SQLite 才是权威数据源（videos 3113 / events 16159 / people 60 / model_calls 9876），NAS 的 MariaDB 全是它的镜像——前端读的数据 本来就产自甲骨文，绕了一圈回家又绕回来。 改动： - db_layer.py 从 725 行重写成 377 行：MySQL 镜像查询改为直读 fam-edge 的 SQLite。5 个 upsert_sync_*（约 300 行去重逻辑，8/29 和 9/3 两次 1062 事故的 发源地）连同 oracle_sync.py 整个删除。SQL 方言：JSON_CONTAINS -> json_each （前置 json_valid，历史脏数据不会把查询搞崩）、LEFT() -> substr()、%s -> ?。 函数名 get_sync_* 一并改掉——已经没有 sync 这回事了 - 新增 edge_client.py：写操作（改名/删除）、帧图头像、服务状态都打给同机 fam-edge，全走 127.0.0.1 - 新增 fam-notifier/：motion_notifier 从 fam-core 拆出独立成服务，游标从 MariaDB 换成本地 JSON 文件。NAS 上从此没有 Flask、没有数据库、没有监听端口 - fam-core 移到甲骨文 /opt/fam-core（systemd，gunicorn -w 2，只绑 127.0.0.1:5401——5400 被 chat-relay 占了）。Caddy 的 /api/* 从"frp 隧道 回源 NAS"改成同机反代，forward_auth 闸门不变 - 前端删掉侧边栏同步面板、统计页同步状态、服务状态页的"NAS 同步"卡片与 "立即同步"按钮（背后的镜像层已不存在）；换成"NAS 运动推送"卡片，读 fam-edge activity 新增的 motion 段（心跳年龄 + 最近事件） - 顺带修掉一个隐蔽 bug：镜像表为保外键稳定用的是 NAS 本地自增 id，而帧图接口 要的是甲骨文的 id，两边在 9/3 那次 id 重排后就对不上了。现在只有一套 id 测试：fam-core 21（新增 12 个 db_layer 用例：脏 JSON 不崩、人物精确匹配不误伤 "人物B"、日期过滤、统计口径、chat_history 懒建表）、fam-notifier 6、 fam-edge 157，全绿。 生产验证：甲骨文 /api/ui/stats 返回 videos 2965 / events 16159 / people 59； NAS 侧 fam-notifier 已推送成功（事件 33070-33072 落库，心跳新鲜）； chat_history 19 条经 scripts/import_chat_history.py 迁移完成。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
27
fam-notifier/config/config.yaml.example
Normal file
27
fam-notifier/config/config.yaml.example
Normal file
@@ -0,0 +1,27 @@
|
||||
# fam-notifier 配置（NAS 上唯一保留的服务）
|
||||
# 复制为 config.yaml 后填实际值；${VAR} 会从环境变量解析（start_notifier.sh 会 source .env）
|
||||
|
||||
motion_notifier:
|
||||
enabled: true
|
||||
poll_enabled: true # 轮询主路径（默认开启）
|
||||
dsm_host: "127.0.0.1" # Surveillance Station 就在本机（NAS），走回环即可
|
||||
dsm_port: 5000
|
||||
dsm_account: "${DSM_ACCOUNT}"
|
||||
dsm_password: "${DSM_PASSWORD}"
|
||||
camera_ids: [2] # 轮询关注的摄像头（Generic_ONVIF-001）
|
||||
|
||||
# 推送目标：甲骨文 fam-edge。这是本服务唯一的出口，单向。
|
||||
oracle_base_url: "http://129.146.26.249:5000"
|
||||
oracle_token: "${ORACLE_SYNC_TOKEN}"
|
||||
timeout_sec: 10 # 单次 SS 请求超时
|
||||
|
||||
# 心跳：跟轮询 SS 无关，只是定期空 POST 一下甲骨文的 /api/ss/motion，证明
|
||||
# NAS->甲骨文这条推送链路本身还活着（enabled=true 就跑，不受 poll_enabled 影响）。
|
||||
# 甲骨文侧 dsm_motion_prefilter.max_heartbeat_age_sec（默认 900s）据此判断"无运动"
|
||||
# 结论是否可信——这个心跳间隔要明显小于那个阈值，否则会被误判成链路已死。
|
||||
heartbeat_interval_sec: 300
|
||||
|
||||
# 轮询参数
|
||||
poll_interval_sec: 60 # 轮询间隔
|
||||
poll_window_hours: 2 # 每轮回看窗口（小时），覆盖轮询间隔内的新事件
|
||||
batch_size: 100 # 单批推送上限
|
||||
3
fam-notifier/requirements.txt
Normal file
3
fam-notifier/requirements.txt
Normal file
@@ -0,0 +1,3 @@
|
||||
# NAS 上唯一要装的依赖：一个 HTTP 客户端 + YAML 配置解析
|
||||
requests>=2.31.0
|
||||
PyYAML>=6.0
|
||||
36
fam-notifier/scripts/start_notifier.sh
Executable file
36
fam-notifier/scripts/start_notifier.sh
Executable file
@@ -0,0 +1,36 @@
|
||||
#!/bin/bash
|
||||
# fam-notifier 启动脚本（NAS 端）
|
||||
# 用法: bash scripts/start_notifier.sh
|
||||
#
|
||||
# NAS 上没有 systemd，挂了不会自启——重启方式见 docs/DEPLOY.md §1.2。
|
||||
set -e
|
||||
|
||||
APP_DIR="$(cd "$(dirname "$0")/.." && pwd)"
|
||||
CONFIG_FILE="$APP_DIR/config/config.yaml"
|
||||
|
||||
if [ ! -f "$CONFIG_FILE" ]; then
|
||||
echo "错误: 配置文件不存在: $CONFIG_FILE"
|
||||
echo "请复制 config/config.yaml.example 为 config.yaml 并填入实际值"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
cd "$APP_DIR"
|
||||
|
||||
# 加载 DSM_ACCOUNT / DSM_PASSWORD / ORACLE_SYNC_TOKEN。
|
||||
# set -a 包裹：.env 里是裸赋值，不 export 的话子进程读不到（2026-09-01 踩过）。
|
||||
ENV_FILE="$APP_DIR/../.env"
|
||||
if [ -f "$ENV_FILE" ]; then
|
||||
set -a
|
||||
source "$ENV_FILE"
|
||||
set +a
|
||||
fi
|
||||
|
||||
if [ -d "venv" ]; then
|
||||
source venv/bin/activate
|
||||
fi
|
||||
|
||||
# 包在 src/ 下，不加这行 `python -m fam_notifier` 找不到模块
|
||||
export PYTHONPATH="$APP_DIR/src${PYTHONPATH:+:$PYTHONPATH}"
|
||||
|
||||
echo "启动 fam-notifier（轮询 Surveillance Station -> 推送甲骨文）..."
|
||||
exec python -m fam_notifier
|
||||
0
fam-notifier/src/fam_notifier/__init__.py
Normal file
0
fam-notifier/src/fam_notifier/__init__.py
Normal file
34
fam-notifier/src/fam_notifier/__main__.py
Normal file
34
fam-notifier/src/fam_notifier/__main__.py
Normal file
@@ -0,0 +1,34 @@
|
||||
"""
|
||||
fam-notifier 入口：`python -m fam_notifier`
|
||||
|
||||
没有 Flask，没有对外端口——这个进程只做一件事：轮询 Surveillance Station，
|
||||
把运动事件单向推给甲骨文的 fam-edge。状态查看去网站的「服务状态」页，
|
||||
心跳和事件都记在甲骨文那边（service_activity / ss_motion_events）。
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
|
||||
from .logger import setup_logger
|
||||
from .notifier import MotionNotifier
|
||||
|
||||
logger = setup_logger('fam-notifier')
|
||||
|
||||
|
||||
def main():
|
||||
notifier = MotionNotifier()
|
||||
if not notifier.enabled:
|
||||
logger.error("motion_notifier.enabled=false，没什么可做的，退出")
|
||||
return 1
|
||||
notifier.start()
|
||||
logger.info("fam-notifier 已启动（Ctrl+C 退出）")
|
||||
try:
|
||||
while notifier.is_alive():
|
||||
time.sleep(5)
|
||||
except KeyboardInterrupt:
|
||||
logger.info("收到中断，停止轮询")
|
||||
notifier.stop()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
32
fam-notifier/src/fam_notifier/config_loader.py
Normal file
32
fam-notifier/src/fam_notifier/config_loader.py
Normal file
@@ -0,0 +1,32 @@
|
||||
"""
|
||||
配置加载器 - 从 config.yaml 读取配置
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import yaml
|
||||
|
||||
|
||||
def _resolve_env_vars(value):
|
||||
"""递归解析字符串中的 ${ENV_VAR} 引用"""
|
||||
if isinstance(value, str):
|
||||
def replace_env(match):
|
||||
env_name = match.group(1)
|
||||
return os.environ.get(env_name, match.group(0))
|
||||
return re.sub(r'\$\{(\w+)\}', replace_env, value)
|
||||
elif isinstance(value, dict):
|
||||
return {k: _resolve_env_vars(v) for k, v in value.items()}
|
||||
elif isinstance(value, list):
|
||||
return [_resolve_env_vars(item) for item in value]
|
||||
return value
|
||||
|
||||
|
||||
def load_config(config_path=None):
|
||||
"""加载 YAML 配置文件，自动解析 ${ENV_VAR} 引用"""
|
||||
if config_path is None:
|
||||
config_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
||||
'config', 'config.yaml'
|
||||
)
|
||||
with open(config_path, 'r', encoding='utf-8') as f:
|
||||
raw = yaml.safe_load(f)
|
||||
return _resolve_env_vars(raw)
|
||||
34
fam-notifier/src/fam_notifier/logger.py
Normal file
34
fam-notifier/src/fam_notifier/logger.py
Normal file
@@ -0,0 +1,34 @@
|
||||
"""
|
||||
日志工具 - 统一格式（stdout + 文件双写）
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 日志目录：fam-notifier/logs/（相对 src 的上一级），失败则退化为仅 stdout
|
||||
_LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'logs')
|
||||
|
||||
|
||||
def setup_logger(name='fam-notifier', level=logging.INFO):
|
||||
"""配置并返回 logger（stdout + 文件双写）"""
|
||||
logger = logging.getLogger(name)
|
||||
if logger.handlers:
|
||||
return logger
|
||||
logger.setLevel(level)
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s [%(name)s] %(levelname)s %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
handler = logging.StreamHandler(sys.stdout)
|
||||
handler.setFormatter(formatter)
|
||||
logger.addHandler(handler)
|
||||
# 文件输出（daemon 模式下 stdout 不可见，文件是唯一可追溯日志）
|
||||
try:
|
||||
os.makedirs(_LOG_DIR, exist_ok=True)
|
||||
file_handler = logging.FileHandler(
|
||||
os.path.join(_LOG_DIR, 'fam-notifier.log'), encoding='utf-8')
|
||||
file_handler.setFormatter(formatter)
|
||||
logger.addHandler(file_handler)
|
||||
except OSError:
|
||||
pass # 目录不可写时退化为仅 stdout
|
||||
return logger
|
||||
400
fam-notifier/src/fam_notifier/notifier.py
Normal file
400
fam-notifier/src/fam_notifier/notifier.py
Normal file
@@ -0,0 +1,400 @@
|
||||
"""
|
||||
MotionNotifier - NAS 端运动监测通知服务（轮询主路径，2026-08-22 定型；
|
||||
2026-09-13 从 fam-core 拆出，成为 NAS 上唯一保留的服务）
|
||||
|
||||
拆出来的原因：摄像头插在 NAS 上（Surveillance Station 就是 NAS 本身），这部分
|
||||
搬不走；而其余所有东西——数据库、查询接口、AI 问答、登录——都已经迁到甲骨文。
|
||||
拆完之后 NAS 侧没有 Flask、没有 MariaDB、没有对外端口，只有这一个进程单向往
|
||||
甲骨文推事件，挂了重启即可，不影响网站。
|
||||
|
||||
职责:
|
||||
在 NAS 本机**轮询**群晖 Surveillance Station 的运动侦测事件
|
||||
（SYNO.SurveillanceStation.EventCenter.Event method=List，参数名下划线风格
|
||||
camera_ids/start_time/end_time，event_type=10 即运动），增量推送到甲骨文
|
||||
FAM-Edge 的 /api/ss/motion 接口。轮询能拿到真实 event_id/start_time/duration。
|
||||
|
||||
数据方向（关键约束）: NAS -> Oracle，单向。甲骨文不再反向访问 NAS。
|
||||
- 原 fam-edge 的 dsm_motion_client（甲骨文主动查 SS API）已停用；
|
||||
- 推送后由甲骨文本地落库 ss_motion_events，供 video_processor 本地预过滤使用。
|
||||
|
||||
可靠性设计:
|
||||
- 游标持久化在本地 JSON 文件（拆出前存 NAS MariaDB 的 sync_cursor 表，
|
||||
现在 NAS 上已经没有数据库了）；重启优先续用游标，
|
||||
停机期间的事件由窗口回看补推；仅首次部署时才初始化为 SS 当前最大 id。
|
||||
- 推送失败的批次不推进游标，下一轮窗口回看重试（不丢事件）。
|
||||
- 心跳线程（enabled 即跑，与轮询无关）：定期空 POST /api/ss/motion，证明
|
||||
NAS->Oracle 推送链路存活，供甲骨文侧判断"无运动"结论是否可信。
|
||||
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import threading
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import requests
|
||||
|
||||
from .logger import setup_logger
|
||||
from .config_loader import load_config
|
||||
|
||||
logger = setup_logger('fam-notifier')
|
||||
|
||||
# 游标文件：拆出前这是 MariaDB 里的一行，现在 NAS 上没有数据库了。
|
||||
# 丢了也不致命——窗口回看会把最近的事件补推一遍，甲骨文侧按 event_id 幂等落库。
|
||||
_CURSOR_PATH = os.environ.get('FAM_NOTIFIER_CURSOR') or os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
||||
'data', 'cursor.json')
|
||||
|
||||
|
||||
def _read_cursor() -> str:
|
||||
try:
|
||||
with open(_CURSOR_PATH, encoding='utf-8') as f:
|
||||
return str(json.load(f).get('motion_last_event_id') or '')
|
||||
except (OSError, ValueError):
|
||||
return ''
|
||||
|
||||
|
||||
def _write_cursor(value: str):
|
||||
try:
|
||||
os.makedirs(os.path.dirname(_CURSOR_PATH), exist_ok=True)
|
||||
with open(_CURSOR_PATH, 'w', encoding='utf-8') as f:
|
||||
json.dump({'motion_last_event_id': str(value)}, f)
|
||||
except OSError as e:
|
||||
logger.warning(f"游标写入失败（下轮窗口回看会补推，不丢事件）: {e}")
|
||||
|
||||
_MOTION_NOTIFIER = None
|
||||
|
||||
|
||||
def get_motion_notifier():
|
||||
"""模块级单例（app.py 启动时创建并 start，状态接口经此获取）。"""
|
||||
global _MOTION_NOTIFIER
|
||||
if _MOTION_NOTIFIER is None:
|
||||
_MOTION_NOTIFIER = MotionNotifier()
|
||||
return _MOTION_NOTIFIER
|
||||
|
||||
|
||||
class MotionNotifier:
|
||||
def __init__(self):
|
||||
cfg = load_config().get('motion_notifier', {})
|
||||
self.enabled = bool(cfg.get('enabled', False))
|
||||
self.dsm_host = cfg.get('dsm_host', '192.168.50.64')
|
||||
self.dsm_port = int(cfg.get('dsm_port', 5000))
|
||||
self.dsm_account = self._resolve(cfg.get('dsm_account', ''))
|
||||
self.dsm_password = self._resolve(cfg.get('dsm_password', ''))
|
||||
self.camera_ids = cfg.get('camera_ids', [2])
|
||||
self.oracle_base_url = cfg.get('oracle_base_url',
|
||||
'http://129.146.26.249:5000').rstrip('/')
|
||||
self.oracle_token = self._resolve(cfg.get('oracle_token', '${ORACLE_SYNC_TOKEN}'))
|
||||
self.poll_interval_sec = int(cfg.get('poll_interval_sec', 60))
|
||||
self.poll_window_hours = int(cfg.get('poll_window_hours', 2))
|
||||
self.batch_size = int(cfg.get('batch_size', 100))
|
||||
self.timeout = int(cfg.get('timeout_sec', 10))
|
||||
# 是否启用轮询（默认开启）：轮询 EventCenter.Event.List 为主数据源，拿到
|
||||
# 真实 event_id/start_time/duration；Webhook 仅作为可选的低延迟补充
|
||||
# （SS 行动规则未配置时不会触发，不影响主路径）
|
||||
self.poll_enabled = bool(cfg.get('poll_enabled', True))
|
||||
# 心跳：跟"轮询 SS"是两回事——不查 SS，只是定期空 POST 一下甲骨文，证明
|
||||
# NAS->Oracle 这条推送链路本身还活着。enabled=true 时始终跑（不受
|
||||
# poll_enabled 影响），供甲骨文侧 has_motion_in_range_local() 判断
|
||||
# "这段时间没收到运动事件"是真的没运动，还是推送链路已经挂了。
|
||||
self.heartbeat_interval_sec = int(cfg.get('heartbeat_interval_sec', 300))
|
||||
# 轮询路径（poll_enabled=true 时）关注的摄像头
|
||||
self.camera_ids = cfg.get('camera_ids', [2])
|
||||
self._base = f"http://{self.dsm_host}:{self.dsm_port}/webapi"
|
||||
self._sid = None
|
||||
self._running = False
|
||||
self._thread = None
|
||||
self._last_event_id = None
|
||||
# 推送过但 duration<=0（动作进行中，SS 尚未回填真实时长）的事件 id。
|
||||
# 下轮窗口回查时若已结束（duration>0）补推覆盖 Oracle，避免分割永远跳过。
|
||||
self._zero_dur_ids = set()
|
||||
self._last_poll_at = None
|
||||
self._last_error = None
|
||||
self._pushed_total = 0
|
||||
self._hb_running = False
|
||||
self._hb_thread = None
|
||||
self._last_heartbeat_at = None
|
||||
self._last_heartbeat_error = None
|
||||
|
||||
@staticmethod
|
||||
def _resolve(v):
|
||||
"""解析 ${ENV} 引用；非字符串或不含 ${...} 原样返回。"""
|
||||
if isinstance(v, str) and v.startswith('${') and v.endswith('}'):
|
||||
return os.environ.get(v[2:-1], '')
|
||||
return v
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Surveillance Station 登录 / 事件查询
|
||||
# ------------------------------------------------------------------
|
||||
def _login(self):
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"{self._base}/auth.cgi",
|
||||
params={"api": "SYNO.API.Auth", "version": 6, "method": "login",
|
||||
"account": self.dsm_account, "passwd": self.dsm_password,
|
||||
"session": "SurveillanceStation", "format": "sid"},
|
||||
timeout=self.timeout)
|
||||
data = resp.json()
|
||||
if data.get('success'):
|
||||
return data['data']['sid']
|
||||
logger.warning(f"SS 登录失败: {data.get('error')}")
|
||||
except Exception as e:
|
||||
logger.warning(f"SS 登录异常: {e}")
|
||||
return None
|
||||
|
||||
def _fetch_events(self, start_ts: int, end_ts: int):
|
||||
"""查询 [start_ts, end_ts] 窗口内的 SS 事件。返回事件列表或 None（查询失败）。"""
|
||||
if self._sid is None:
|
||||
self._sid = self._login()
|
||||
if self._sid is None:
|
||||
return None
|
||||
for attempt in range(2):
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"{self._base}/entry.cgi",
|
||||
params={"api": "SYNO.SurveillanceStation.EventCenter.Event",
|
||||
"version": 1, "method": "List",
|
||||
"camera_ids": ",".join(str(c) for c in self.camera_ids),
|
||||
"start_time": start_ts, "end_time": end_ts,
|
||||
"limit": 1000, "_sid": self._sid},
|
||||
timeout=self.timeout)
|
||||
data = resp.json()
|
||||
except Exception as e:
|
||||
logger.warning(f"SS 事件查询异常: {e}")
|
||||
return None
|
||||
if not data.get('success'):
|
||||
code = (data.get('error') or {}).get('code')
|
||||
if code in (106, 107, 119) and attempt == 0:
|
||||
# session 过期/被顶掉，重新登录重试一次
|
||||
self._sid = self._login()
|
||||
if self._sid is None:
|
||||
return None
|
||||
continue
|
||||
logger.warning(f"SS 事件查询失败: {data.get('error')}")
|
||||
return None
|
||||
# 响应按 ds_id（CMS 多机场景的服务器 id，单机固定 "0"）分组，拉平
|
||||
events = [e for grp in (data.get('data') or {}).values()
|
||||
for e in (grp or [])]
|
||||
return events
|
||||
return None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 推送到甲骨文
|
||||
# ------------------------------------------------------------------
|
||||
def push_events_to_oracle(self, events) -> int:
|
||||
"""把标准化后的事件列表推送到 Oracle /api/ss/motion。返回成功推送条数。"""
|
||||
if not events:
|
||||
return 0
|
||||
norm = []
|
||||
for e in events:
|
||||
eid = e.get('id') or e.get('event_id')
|
||||
if eid is None:
|
||||
continue
|
||||
norm.append({
|
||||
"event_id": int(eid),
|
||||
"camera_id": e.get('camera_id'),
|
||||
"event_type": e.get('event_type'),
|
||||
"start_time": e.get('start_time'),
|
||||
"duration": e.get('duration'),
|
||||
"thumbnail_url": e.get('thumbnail_url') or e.get('thumbnail_dir'),
|
||||
})
|
||||
if not norm:
|
||||
return 0
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{self.oracle_base_url}/api/ss/motion",
|
||||
json={"token": self.oracle_token, "events": norm},
|
||||
timeout=(10, 30))
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"推送运动事件到 Oracle 失败: {e}")
|
||||
self._last_error = str(e)
|
||||
return 0
|
||||
if resp.status_code != 200:
|
||||
logger.error(f"推送运动事件到 Oracle 返回 {resp.status_code}: {resp.text[:200]}")
|
||||
self._last_error = f"HTTP {resp.status_code}"
|
||||
return 0
|
||||
try:
|
||||
stored = resp.json().get('stored', 0)
|
||||
except ValueError:
|
||||
stored = len(norm)
|
||||
self._pushed_total += stored
|
||||
self._last_error = None
|
||||
logger.info(f"运动事件推送成功: {len(norm)} 条 -> Oracle 存储 {stored}")
|
||||
return stored
|
||||
|
||||
def send_heartbeat(self) -> bool:
|
||||
"""空 events 调一次 /api/ss/motion，只为证明 NAS->Oracle 推送链路还活着。
|
||||
|
||||
跟 push_events_to_oracle 分开一个方法，是因为那个方法 events 为空时直接
|
||||
return 0（不发请求）——心跳恰恰就是要在没有真实事件时也发一次请求。
|
||||
"""
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{self.oracle_base_url}/api/ss/motion",
|
||||
json={"token": self.oracle_token, "events": []},
|
||||
timeout=(10, 30))
|
||||
except requests.RequestException as e:
|
||||
logger.warning(f"运动心跳推送失败: {e}")
|
||||
self._last_heartbeat_error = str(e)
|
||||
return False
|
||||
if resp.status_code != 200:
|
||||
logger.warning(f"运动心跳推送返回 {resp.status_code}: {resp.text[:200]}")
|
||||
self._last_heartbeat_error = f"HTTP {resp.status_code}"
|
||||
return False
|
||||
self._last_heartbeat_at = datetime.now()
|
||||
self._last_heartbeat_error = None
|
||||
return True
|
||||
|
||||
def _heartbeat_run(self):
|
||||
logger.info(f"运动心跳线程启动，间隔 {self.heartbeat_interval_sec}s")
|
||||
while self._hb_running:
|
||||
try:
|
||||
self.send_heartbeat()
|
||||
except Exception as e:
|
||||
self._last_heartbeat_error = str(e)
|
||||
logger.error(f"运动心跳异常: {e}", exc_info=True)
|
||||
for _ in range(self.heartbeat_interval_sec):
|
||||
if not self._hb_running:
|
||||
break
|
||||
time.sleep(1)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 增量轮询主循环
|
||||
# ------------------------------------------------------------------
|
||||
def _init_cursor(self):
|
||||
"""启动时初始化 last_event_id。
|
||||
|
||||
优先使用文件里保存的游标：服务停机/重启期间产生的新事件，会在重启后
|
||||
被下一轮窗口回看捞到并补推（不丢事件）。
|
||||
仅当没有有效游标（首次部署）时，取 SS 当前最大事件 id 作为起点，
|
||||
避免把历史事件全部回灌一遍。
|
||||
"""
|
||||
saved = _read_cursor()
|
||||
if saved and int(saved) > 0:
|
||||
self._last_event_id = int(saved)
|
||||
logger.info(f"运动通知游标初始化（本地文件）: last_event_id={self._last_event_id}")
|
||||
return
|
||||
now = int(datetime.now(timezone.utc).timestamp())
|
||||
events = self._fetch_events(now - 3600, now) # 最近 1h 用于定位最大 id
|
||||
if events:
|
||||
self._last_event_id = max(
|
||||
int(e.get('id', 0)) for e in events if e.get('id'))
|
||||
else:
|
||||
self._last_event_id = 0
|
||||
logger.info(f"运动通知游标初始化（SS 当前最大）: last_event_id={self._last_event_id}")
|
||||
|
||||
def _poll_once(self):
|
||||
now = int(datetime.now(timezone.utc).timestamp())
|
||||
start_ts = now - int(self.poll_window_hours * 3600)
|
||||
events = self._fetch_events(start_ts, now)
|
||||
if events is None:
|
||||
return # 查询失败，下一轮重试
|
||||
by_id = {int(e.get('id')): e for e in events if e.get('id')}
|
||||
|
||||
# 1) 补推：先前推送时 duration<=0（动作进行中）的事件，现在若已结束
|
||||
# （SS 回填真实 duration>0），重推覆盖 Oracle（ON CONFLICT UPDATE）。
|
||||
# 重启兜底：游标 DB 续用 → 窗口回看会把历史事件整体重推，届时同样覆盖。
|
||||
if self._zero_dur_ids:
|
||||
recheck = []
|
||||
for eid in list(self._zero_dur_ids):
|
||||
e = by_id.get(eid)
|
||||
if e and int(e.get('duration') or 0) > 0:
|
||||
recheck.append(e)
|
||||
self._zero_dur_ids.discard(eid)
|
||||
if recheck:
|
||||
self.push_events_to_oracle(recheck)
|
||||
logger.info(f"补推 {len(recheck)} 条已结束事件的真实 duration")
|
||||
|
||||
# 2) 增量推送新事件（id 大于游标）
|
||||
new = [e for e in events
|
||||
if e.get('id') and int(e.get('id')) > (self._last_event_id or 0)]
|
||||
if not new:
|
||||
return
|
||||
new.sort(key=lambda e: int(e.get('id', 0)))
|
||||
cursor = self._last_event_id
|
||||
for i in range(0, len(new), self.batch_size):
|
||||
batch = new[i:i + self.batch_size]
|
||||
stored = self.push_events_to_oracle(batch)
|
||||
if stored > 0:
|
||||
# 只有推送成功的批次才推进游标；失败批次保持原地，
|
||||
# 下一轮窗口回看会重新捞到并重试（不丢事件）
|
||||
cursor = max(int(e.get('id', 0)) for e in batch)
|
||||
# 记录推送时仍在进行中的事件（duration<=0），待下轮补推真实时长
|
||||
for e in batch:
|
||||
if int(e.get('duration') or 0) <= 0:
|
||||
self._zero_dur_ids.add(int(e.get('id')))
|
||||
self._last_event_id = cursor
|
||||
_write_cursor(str(self._last_event_id))
|
||||
|
||||
def _run(self):
|
||||
logger.info(f"MotionNotifier 启动，轮询间隔 {self.poll_interval_sec}s，"
|
||||
f"目标 SS {self.dsm_host}:{self.dsm_port}")
|
||||
try:
|
||||
self._init_cursor()
|
||||
except Exception as e:
|
||||
logger.warning(f"运动通知游标初始化失败（从 0 开始）: {e}")
|
||||
self._last_event_id = 0
|
||||
while self._running:
|
||||
try:
|
||||
self._poll_once()
|
||||
self._last_poll_at = datetime.now()
|
||||
except Exception as e:
|
||||
self._last_error = str(e)
|
||||
logger.error(f"运动通知轮询异常: {e}", exc_info=True)
|
||||
# 分段休眠，便于 stop 快速唤醒
|
||||
for _ in range(self.poll_interval_sec):
|
||||
if not self._running:
|
||||
break
|
||||
time.sleep(1)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def start(self):
|
||||
if not self.enabled:
|
||||
logger.info("MotionNotifier 未启用（motion_notifier.enabled=false）")
|
||||
return
|
||||
# 心跳线程跟轮询是否开启无关：只要 MotionNotifier 整体 enabled，就该
|
||||
# 持续证明推送链路活着，哪怕当前正好没有真实运动事件可推送。
|
||||
if not self._hb_running:
|
||||
self._hb_running = True
|
||||
self._hb_thread = threading.Thread(
|
||||
target=self._heartbeat_run, daemon=True, name='motion-heartbeat')
|
||||
self._hb_thread.start()
|
||||
if not self.poll_enabled:
|
||||
logger.info("MotionNotifier 轮询已禁用（poll_enabled=false），仅作为 "
|
||||
"Webhook 推送客户端 + 心跳 + 摄像头名映射使用")
|
||||
return
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self._thread = threading.Thread(target=self._run, daemon=True,
|
||||
name='motion-notifier')
|
||||
self._thread.start()
|
||||
|
||||
def is_alive(self):
|
||||
return self._thread is not None and self._thread.is_alive()
|
||||
|
||||
def stop(self):
|
||||
self._running = False
|
||||
if self._thread:
|
||||
self._thread.join(timeout=5)
|
||||
self._hb_running = False
|
||||
if self._hb_thread:
|
||||
self._hb_thread.join(timeout=5)
|
||||
|
||||
def status(self) -> dict:
|
||||
return {
|
||||
"heartbeat_running": self._hb_thread is not None and self._hb_thread.is_alive(),
|
||||
"last_heartbeat_at": self._last_heartbeat_at.isoformat() if self._last_heartbeat_at else None,
|
||||
"last_heartbeat_error": self._last_heartbeat_error,
|
||||
"running": self.is_alive(),
|
||||
"enabled": self.enabled,
|
||||
"poll_enabled": self.poll_enabled,
|
||||
"last_event_id": self._last_event_id,
|
||||
"zero_dur_pending": len(self._zero_dur_ids),
|
||||
"last_poll_at": self._last_poll_at.isoformat() if self._last_poll_at else None,
|
||||
"last_error": self._last_error,
|
||||
"pushed_total": self._pushed_total,
|
||||
"oracle": self.oracle_base_url,
|
||||
"dsm": f"{self.dsm_host}:{self.dsm_port}",
|
||||
}
|
||||
7
fam-notifier/tests/conftest.py
Normal file
7
fam-notifier/tests/conftest.py
Normal file
@@ -0,0 +1,7 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 让测试能直接 `from fam_notifier.xxx import yyy`，无需先 pip install -e .
|
||||
_SRC = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'src')
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
114
fam-notifier/tests/test_notifier.py
Normal file
114
fam-notifier/tests/test_notifier.py
Normal file
@@ -0,0 +1,114 @@
|
||||
import time
|
||||
|
||||
from fam_notifier.notifier import MotionNotifier
|
||||
|
||||
|
||||
def _cfg(**overrides):
|
||||
base = {
|
||||
"enabled": True,
|
||||
"poll_enabled": False,
|
||||
"dsm_host": "192.168.50.64",
|
||||
"dsm_port": 5000,
|
||||
"dsm_account": "ericwyuan",
|
||||
"dsm_password": "secret",
|
||||
"oracle_base_url": "http://oracle.example:5000",
|
||||
"oracle_token": "tok123",
|
||||
"heartbeat_interval_sec": 300,
|
||||
"timeout_sec": 5,
|
||||
}
|
||||
base.update(overrides)
|
||||
return {"motion_notifier": base}
|
||||
|
||||
|
||||
def _make_notifier(monkeypatch, **cfg_overrides):
|
||||
monkeypatch.setattr(
|
||||
"fam_notifier.notifier.load_config",
|
||||
lambda: _cfg(**cfg_overrides))
|
||||
return MotionNotifier()
|
||||
|
||||
|
||||
class _FakeResp:
|
||||
def __init__(self, status_code=200, payload=None, text=""):
|
||||
self.status_code = status_code
|
||||
self._payload = payload or {}
|
||||
self.text = text
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# 心跳（跟轮询无关，enabled=true 就该跑）
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def test_send_heartbeat_success_updates_state(monkeypatch):
|
||||
n = _make_notifier(monkeypatch)
|
||||
calls = []
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
calls.append((url, json))
|
||||
return _FakeResp(200, {"status": "ok", "received": 0, "stored": 0})
|
||||
monkeypatch.setattr("fam_notifier.notifier.requests.post", fake_post)
|
||||
ok = n.send_heartbeat()
|
||||
assert ok is True
|
||||
assert n._last_heartbeat_at is not None
|
||||
assert n._last_heartbeat_error is None
|
||||
assert calls[0][0] == "http://oracle.example:5000/api/ss/motion"
|
||||
assert calls[0][1]["events"] == []
|
||||
assert calls[0][1]["token"] == "tok123"
|
||||
|
||||
|
||||
def test_send_heartbeat_http_error_records_failure(monkeypatch):
|
||||
n = _make_notifier(monkeypatch)
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
return _FakeResp(500, {}, text="boom")
|
||||
monkeypatch.setattr("fam_notifier.notifier.requests.post", fake_post)
|
||||
ok = n.send_heartbeat()
|
||||
assert ok is False
|
||||
assert n._last_heartbeat_error == "HTTP 500"
|
||||
|
||||
|
||||
def test_send_heartbeat_network_exception_records_failure(monkeypatch):
|
||||
import requests as _requests
|
||||
n = _make_notifier(monkeypatch)
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
raise _requests.RequestException("connection refused")
|
||||
monkeypatch.setattr("fam_notifier.notifier.requests.post", fake_post)
|
||||
ok = n.send_heartbeat()
|
||||
assert ok is False
|
||||
assert "connection refused" in n._last_heartbeat_error
|
||||
|
||||
|
||||
def test_start_runs_heartbeat_thread_even_when_poll_disabled(monkeypatch):
|
||||
"""核心诉求: 心跳跟"是否轮询 SS"是两回事——poll_enabled=false 时轮询线程
|
||||
不应该启动，但心跳线程必须照样跑，否则甲骨文永远收不到心跳，
|
||||
has_motion_in_range_local() 会一直 fail-open，省配额的效果就没了。"""
|
||||
n = _make_notifier(monkeypatch, poll_enabled=False, heartbeat_interval_sec=3600)
|
||||
monkeypatch.setattr(
|
||||
"fam_notifier.notifier.requests.post",
|
||||
lambda url, json=None, timeout=None: _FakeResp(200, {"stored": 0}))
|
||||
try:
|
||||
n.start()
|
||||
time.sleep(0.2)
|
||||
assert n.is_alive() is False # 轮询线程未启动
|
||||
assert n._hb_thread is not None and n._hb_thread.is_alive()
|
||||
finally:
|
||||
n.stop()
|
||||
|
||||
|
||||
def test_start_does_nothing_when_disabled(monkeypatch):
|
||||
n = _make_notifier(monkeypatch, enabled=False)
|
||||
n.start()
|
||||
time.sleep(0.1)
|
||||
assert n._hb_thread is None
|
||||
assert n.is_alive() is False
|
||||
|
||||
|
||||
def test_status_includes_heartbeat_fields(monkeypatch):
|
||||
n = _make_notifier(monkeypatch)
|
||||
st = n.status()
|
||||
assert "heartbeat_running" in st
|
||||
assert "last_heartbeat_at" in st
|
||||
assert "last_heartbeat_error" in st
|
||||
Reference in New Issue
Block a user