From e56d362d732ead4698e0db9dfc6df2636ab2b981 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Sat, 22 Aug 2026 00:50:49 +0800 Subject: [PATCH] =?UTF-8?q?refactor(fam-core):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E7=AC=AC=E4=BA=8C=E9=98=B6=E6=AE=B5=20-=20img=5Fproxy=20?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=20+=20=E7=A8=B3=E5=AE=9A=E6=80=A7=E5=B0=8F?= =?UTF-8?q?=E4=BF=AE=20+=20=E5=B7=A5=E7=A8=8B=E8=B4=A8=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit img_proxy 收尾: 提交此前未提交的图片代理蓝图(app.py 注册)，改为复用 oracle_sync 已解析好的 base_url/token/timeout，不再独立读一份配置(避免配置改动时两处不同步)。 稳定性/正确性: chat_handler 问答失败时不再把 Oracle 内部 HTTP 状态码等细节透传给 客户端，改为通用错误信息(详细原因仍记服务端日志); 删除 logger.py 里旧任务架构 遗留的死函数 log_task; 更新 config.yaml.example 到当前 v2 架构(原文件还是重构前 的 scheduler/dispatcher/video_server 旧结构，当前代码完全不读这些字段)。 工程质量: 新增 fam-core/tests(9 个单元测试，覆盖 _format_events 上下文格式化和 config_loader 的 ${ENV_VAR} 解析)。 部署时发现并修复一个和这次改动无关的运维问题: NAS .env 文件缺 export 关键字， plain source 只在当前 shell 生效不会被子进程(gunicorn)继承，导致 Oracle-Sync token 校验失败；用 set -a/set +a 强制导出重启，非代码改动。 已部署 NAS 并验证：/health、/api/status、/api/proxy/frame、/api/proxy/avatar 全部通过；浏览器实测事件时间轴、人物管理页正常渲染。 Co-Authored-By: Claude Sonnet 5 --- fam-core/config/config.yaml.example | 42 +++++-------- fam-core/src/fam_core/app.py | 2 + .../src/fam_core/chat_handler/chat_handler.py | 2 +- fam-core/src/fam_core/img_proxy.py | 61 +++++++++++++++++++ fam-core/src/fam_core/logger.py | 12 +--- fam-core/tests/conftest.py | 7 +++ fam-core/tests/test_chat_handler.py | 55 +++++++++++++++++ fam-core/tests/test_config_loader.py | 25 ++++++++ 8 files changed, 168 insertions(+), 38 deletions(-) create mode 100644 fam-core/src/fam_core/img_proxy.py create mode 100644 fam-core/tests/conftest.py create mode 100644 fam-core/tests/test_chat_handler.py create mode 100644 fam-core/tests/test_config_loader.py diff --git a/fam-core/config/config.yaml.example b/fam-core/config/config.yaml.example index f1d2586..aeefc04 100644 --- a/fam-core/config/config.yaml.example +++ b/fam-core/config/config.yaml.example @@ -1,5 +1,9 @@ -# FAM-Core 配置文件 (NAS 端) +# FAM-Core 配置文件 (NAS 端) - 新架构 v2 # 复制此文件为 config.yaml 并修改实际值 +# +# NAS 仅作管理后台，不再处理视频。唯一后台线程 Oracle-Sync 每 30 分钟 +# 从甲骨文 FAM-Edge 拉取增量镜像到本地 MariaDB（sync_videos/events/people）。 +# 所有视频分析在 Oracle 完成。 server: host: "0.0.0.0" @@ -11,32 +15,18 @@ database: user: "root" password: "" database: "sentinel_home_ai" + unix_socket: "/run/mysqld/mysqld10.sock" -scheduler: - scan_interval: 60 # 扫描间隔（秒） - video_dir: "/volume1/surveillance" # 视频录制目录 - video_extensions: [".mp4", ".mkv", ".avi"] - file_stable_seconds: 60 # 文件稳定判定秒数 - camera_name: "默认摄像头" - -dispatcher: - poll_interval: 30 # 轮询间隔（秒） - edge_url: "http://100.x.x.20:5000/api/edge/video/enqueue" # 异步队列端点（上传入队，Poller 拉取结果） - max_retries: 5 # 文件级重试次数（分块级重试另计，每块3次） - stale_timeout: 1800 # PROCESSING 僵尸回收（秒），需大于压缩+上传+Edge队列积压总时长 - # ffmpeg_path: "/var/packages/CodecPack/target/bin/ffmpeg41" # 可选，默认自动探测（Synology 需 CodecPack 版，系统版无 h264 编码） - compress_timeout: 3600 # 单个视频预压缩超时（秒） - -video_server: - base_url: "http://100.x.x.10:8000/media" - token: "your-secret-token-here" # 视频访问 token - video_dir: "/volume1/surveillance" +# 甲骨文同步（每 30 分钟拉增量镜像） +oracle_sync: + # FAM-Edge 对外同步接口地址（端口同其 server.port=5000） + base_url: "http://:5000" + # 与 Oracle 端 sync_api.token 一致 + token: "" + interval_sec: 1800 # 拉取间隔（秒），默认 30 分钟 + timeout: 120 # 单次拉取超时（秒） chat_handler: - ollama_url: "http://100.x.x.20:11434/api/generate" - model_name: "llava-phi3" + # 智能问答统一走 FAM-Edge 编排端点（Gemini → NVIDIA → 本地 Ollama 兜底） + qa_url: "http://:5000/api/edge/chat/ask" timeout: 120 - -storage: - # 关键帧落盘目录（event_receiver 写入，fam-ui 读取展示时间轴） - frame_image_dir: "/volume1/web/sentinel-home-ai/fam-ui/static/frames" diff --git a/fam-core/src/fam_core/app.py b/fam-core/src/fam_core/app.py index 19b29b4..b5df50f 100644 --- a/fam-core/src/fam_core/app.py +++ b/fam-core/src/fam_core/app.py @@ -19,6 +19,7 @@ from .logger import setup_logger from .oracle_sync import get_sync from .chat_handler.chat_handler import chat_bp from .member_manager.member_manager import member_bp +from .img_proxy import img_bp logger = setup_logger('fam-core.app') @@ -27,6 +28,7 @@ app = Flask(__name__) # 注册蓝图 app.register_blueprint(chat_bp) app.register_blueprint(member_bp) +app.register_blueprint(img_bp) # 健康检查 @app.route('/health', methods=['GET']) 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 9dc4b3a..19cba30 100644 --- a/fam-core/src/fam_core/chat_handler/chat_handler.py +++ b/fam-core/src/fam_core/chat_handler/chat_handler.py @@ -98,7 +98,7 @@ def chat_ask(): answer = _call_edge_qa(prompt) except Exception as e: logger.error(f"问答编排调用失败: {e}") - return jsonify({"error": f"AI 调用失败: {e}"}), 503 + return jsonify({"error": "AI 服务暂时不可用，请稍后重试"}), 503 chat_id = db_layer.insert_chat_history( user_question=question, diff --git a/fam-core/src/fam_core/img_proxy.py b/fam-core/src/fam_core/img_proxy.py new file mode 100644 index 0000000..3edfd41 --- /dev/null +++ b/fam-core/src/fam_core/img_proxy.py @@ -0,0 +1,61 @@ +""" +Img-Proxy - NAS 端图片代理（图源在 Oracle，计算全部在 Oracle） + +浏览器不直连 Oracle（避免公网暴露 5000 端口与 token 外泄）， +而是访问 NAS fam-core 的 /api/proxy/*，由 NAS 出网到 Oracle 拉取 jpeg 回传。 + +Oracle 侧已有磁盘缓存 / VLM 人物定位裁剪，NAS 端仅透传，不做图像计算。 +""" +from flask import Blueprint, Response, request + +from .oracle_sync import get_sync +from .logger import setup_logger + +logger = setup_logger('fam-core.img_proxy') + +img_bp = Blueprint('img_proxy', __name__) + + +def _fetch(rel: str, params: dict): + import requests + sync = get_sync() # 复用 oracle_sync 已解析好的 base_url/token/timeout，避免两处配置各读一份 + params = dict(params) + if sync.token: + params['token'] = sync.token + try: + resp = requests.get(f"{sync.base_url}{rel}", params=params, + timeout=(10, sync.timeout)) + if resp.status_code == 200 and resp.content: + return resp.content + logger.warning(f"Oracle {rel} 返回 {resp.status_code}: {resp.text[:120]}") + return None + except Exception as e: + logger.error(f"Oracle {rel} 请求失败: {e}") + return None + + +@img_bp.route('/api/proxy/frame', methods=['GET']) +def proxy_frame(): + """事件缩略帧: video_id + ts -> jpeg""" + data = _fetch('/api/oracle/frame', { + 'video_id': request.args.get('video_id', type=int) or 0, + 'ts': request.args.get('ts', ''), + 'w': request.args.get('w', 400, type=int), + }) + if data is None: + return '', 404 + return Response(data, mimetype='image/jpeg', + headers={'Cache-Control': 'public, max-age=86400'}) + + +@img_bp.route('/api/proxy/avatar', methods=['GET']) +def proxy_avatar(): + """人物头像: label/canonical -> jpeg""" + data = _fetch('/api/oracle/avatar', { + 'label': request.args.get('label', ''), + 'w': request.args.get('w', 160, type=int), + }) + if data is None: + return '', 404 + return Response(data, mimetype='image/jpeg', + headers={'Cache-Control': 'public, max-age=86400'}) \ No newline at end of file diff --git a/fam-core/src/fam_core/logger.py b/fam-core/src/fam_core/logger.py index 9db3a88..e6922db 100644 --- a/fam-core/src/fam_core/logger.py +++ b/fam-core/src/fam_core/logger.py @@ -1,10 +1,9 @@ """ -日志工具 - 统一格式，带 task_id 作为 trace_id +日志工具 - 统一格式（stdout + 文件双写） """ import logging import os import sys -from datetime import datetime # 日志目录：fam-core/logs/（相对 src 的上一级），失败则退化为仅 stdout _LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'logs') @@ -33,12 +32,3 @@ def setup_logger(name='fam-core', level=logging.INFO): except OSError: pass # 目录不可写时退化为仅 stdout return logger - - -def log_task(logger, task_id, stage, message, level=logging.INFO, duration_ms=None): - """带 task_id 的结构化日志""" - parts = [f"[task_id={task_id}]", stage] - if duration_ms is not None: - parts.append(f"done in {duration_ms}ms") - parts.append(message) - logger.log(level, ' '.join(parts)) diff --git a/fam-core/tests/conftest.py b/fam-core/tests/conftest.py new file mode 100644 index 0000000..51ea454 --- /dev/null +++ b/fam-core/tests/conftest.py @@ -0,0 +1,7 @@ +import os +import sys + +# 让测试能直接 `from fam_core.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) diff --git a/fam-core/tests/test_chat_handler.py b/fam-core/tests/test_chat_handler.py new file mode 100644 index 0000000..81044c1 --- /dev/null +++ b/fam-core/tests/test_chat_handler.py @@ -0,0 +1,55 @@ +from fam_core.chat_handler.chat_handler import _format_events + + +def test_multiple_persons_and_attention(): + rows = [{ + "ts": "2026-08-21 16:11:25", "camera_name": "客厅", + "description": "两人在客厅玩耍", + "person_list_json": '["汤圆", "爷爷"]', + "is_attention_event": True, + }] + out = _format_events(rows) + assert out == "[2026-08-21 16:11 客厅] 汤圆,爷爷: 两人在客厅玩耍 [关注事件]" + + +def test_no_person_defaults_to_wu_ren(): + rows = [{ + "ts": "2026-08-21 16:11:25", "camera_name": "客厅", + "description": "空房间", "person_list_json": '[]', + "is_attention_event": False, + }] + out = _format_events(rows) + assert "无人:" in out + assert "[关注事件]" not in out + + +def test_person_list_already_decoded_list(): + """pymysql 某些驱动/字段类型可能已经把 JSON 解码成 list，不是字符串。""" + rows = [{ + "ts": "2026-08-21 16:11:25", "camera_name": "客厅", + "description": "在客厅", "person_list_json": ["汤圆"], + "is_attention_event": False, + }] + out = _format_events(rows) + assert "汤圆:" in out + + +def test_malformed_person_list_json_degrades_gracefully(): + rows = [{ + "ts": "2026-08-21 16:11:25", "camera_name": "客厅", + "description": "在客厅", "person_list_json": "not valid json", + "is_attention_event": False, + }] + out = _format_events(rows) + assert "无人:" in out + + +def test_multiple_rows_joined_by_newline(): + rows = [ + {"ts": "2026-08-21 16:11:25", "camera_name": "客厅", + "description": "第一条", "person_list_json": '["汤圆"]', "is_attention_event": False}, + {"ts": "2026-08-21 16:12:00", "camera_name": "客厅", + "description": "第二条", "person_list_json": '["汤圆"]', "is_attention_event": False}, + ] + out = _format_events(rows) + assert len(out.split("\n")) == 2 diff --git a/fam-core/tests/test_config_loader.py b/fam-core/tests/test_config_loader.py new file mode 100644 index 0000000..342d316 --- /dev/null +++ b/fam-core/tests/test_config_loader.py @@ -0,0 +1,25 @@ +import os + +from fam_core.config_loader import _resolve_env_vars + + +def test_resolves_string_env_var(monkeypatch): + monkeypatch.setenv("FAM_CORE_TEST_VAR", "hello") + assert _resolve_env_vars("${FAM_CORE_TEST_VAR}") == "hello" + + +def test_unset_env_var_left_as_literal(): + assert _resolve_env_vars("${SOME_TOTALLY_UNSET_VAR_XYZ}") == "${SOME_TOTALLY_UNSET_VAR_XYZ}" + + +def test_resolves_nested_dict(monkeypatch): + monkeypatch.setenv("FAM_CORE_TEST_VAR", "secret") + data = {"a": {"b": "${FAM_CORE_TEST_VAR}"}, "c": ["${FAM_CORE_TEST_VAR}", "plain"]} + out = _resolve_env_vars(data) + assert out == {"a": {"b": "secret"}, "c": ["secret", "plain"]} + + +def test_non_string_values_passed_through(): + assert _resolve_env_vars(42) == 42 + assert _resolve_env_vars(True) is True + assert _resolve_env_vars(None) is None