ef56ae5662
背景：Oracle 磁盘曾经被写满（gdrive_videos 持续下载新素材、旧文件迟迟没 清理），触发 rclone 的一个安全机制——同步遇到 IO 错误就整体拒绝执行删除， 形成"越满越删不掉，越删不掉越满"的死循环，最终连新视频都下载不了。这次 排查+手动清理已经解决了当次故障，但需要一道独立于 rclone 同步之外的兜底， 防止再次悄悄写满没人发现。 - oracle_db.py 新增 get_oldest_purgeable_material()：只挑最旧的、已完成 分割阶段（status='done'）的整段素材（非 motion_ 前缀），绝不碰运动片段 （事件时间轴/人物头像依赖它）和还在处理中的素材 - disk_guard.py 新增 DiskGuard 后台线程：5 分钟检查一次，剩余空间 <10GB 触发清理，删到 15GB 水位为止（留缓冲避免刚清完又立刻触发），复用已有的 delete_video() 完成实际删除 - app.py 启动这个后台服务；/api/oracle/activity 新增 disk 字段（实时剩余 空间 + 最近一次清理动作），供服务状态页展示 - 新增 9 个单元测试，全部通过（139/139） 已部署 Oracle 验证：DiskGuard 正常启动，配置生效。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
466 lines
22 KiB
Python
466 lines
22 KiB
Python
import json
|
||
|
||
from fam_edge.oracle_db import OracleDB
|
||
|
||
|
||
def _db(tmp_path):
|
||
return OracleDB(str(tmp_path / "oracle.db"))
|
||
|
||
|
||
def _set_heartbeat_age(db, age_sec):
|
||
"""把心跳时间戳直接改写成"距现在 age_sec 秒前"，用于测试新鲜度阈值边界。"""
|
||
from datetime import datetime, timedelta, timezone
|
||
ts = (datetime.now(timezone(timedelta(hours=8))) - timedelta(seconds=age_sec))
|
||
db.set_cursor('motion_heartbeat_at', ts.strftime('%Y-%m-%d %H:%M:%S'))
|
||
|
||
|
||
def test_upsert_person_same_gender_merges_into_one_row(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.upsert_person("人物A", features={"gender": "男", "hair": "短发黑色"})
|
||
db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色T恤"})
|
||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||
assert len(rows) == 1
|
||
assert rows[0]["appearances"] == 2
|
||
feats = json.loads(rows[0]["features_json"])
|
||
assert feats["hair"] == "短发黑色"
|
||
assert feats["clothing"] == "蓝色T恤"
|
||
|
||
|
||
def test_upsert_person_gender_conflict_splits_into_new_label(tmp_path):
|
||
"""核心诉求: 大模型给的 人物A/B/C 这类 uid 只在单次视频分析内稳定，不同视频
|
||
独立编号，同一字符串完全可能撞到不同真人（实测 "人物A" 混了男女两个人）。
|
||
性别冲突时不能直接合并覆盖，要拆成新 label，避免两个人的特征越merge越乱。"""
|
||
db = _db(tmp_path)
|
||
db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色Polo衫"})
|
||
db.upsert_person("人物A", features={"gender": "女", "clothing": "白色上衣"})
|
||
rows = {r["label"]: r for r in db._conn.execute("SELECT * FROM people").fetchall()}
|
||
assert set(rows.keys()) == {"人物A", "人物A#2"}
|
||
assert rows["人物A"]["appearances"] == 1
|
||
assert json.loads(rows["人物A"]["features_json"])["gender"] == "男"
|
||
assert rows["人物A#2"]["appearances"] == 1
|
||
assert json.loads(rows["人物A#2"]["features_json"])["gender"] == "女"
|
||
# 派生行的 display_uid 仍然记录原始大模型 uid，方便追溯来源
|
||
assert rows["人物A#2"]["display_uid"] == "人物A"
|
||
|
||
|
||
def test_upsert_person_gender_conflict_allocates_next_free_suffix(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.upsert_person("人物A", features={"gender": "男"})
|
||
db.upsert_person("人物A", features={"gender": "女"}) # -> 人物A#2
|
||
db.upsert_person("人物A", features={"gender": "unknown"}) # unknown 不冲突，合并回 人物A
|
||
db.upsert_person("人物A", features={"gender": "男", "hair": "光头"}) # 冲突，人物A 有 gender 男了不冲突；应仍合并
|
||
labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()}
|
||
assert labels == {"人物A", "人物A#2"}
|
||
# 再来一次性别冲突（对着 人物A#2，性别女）应该分配 人物A#3，而不是复用 人物A#2
|
||
db.upsert_person("人物A", features={"gender": "男"})
|
||
db.upsert_person("人物A", features={"gender": "女"})
|
||
# 这次新的女性冲突会先撞到 人物A（此时是男），分裂出下一个空闲后缀
|
||
labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()}
|
||
assert "人物A" in labels
|
||
assert len(labels) >= 2
|
||
|
||
|
||
def test_upsert_person_unknown_gender_never_triggers_split(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.upsert_person("人物A", features={"gender": "男"})
|
||
db.upsert_person("人物A", features={"gender": "unknown"})
|
||
db.upsert_person("人物A", features={"gender": "未知"})
|
||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||
assert len(rows) == 1
|
||
assert rows[0]["appearances"] == 3
|
||
|
||
|
||
def test_upsert_person_no_features_never_triggers_split(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.upsert_person("人物A", features={"gender": "男"})
|
||
db.upsert_person("人物A") # 无 features（source 更新等场景）
|
||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||
assert len(rows) == 1
|
||
assert rows[0]["appearances"] == 2
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 运动侦测事件（NAS 推送）
|
||
# ----------------------------------------------------------------------
|
||
|
||
def test_record_motion_events_upserts_by_event_id(tmp_path):
|
||
db = _db(tmp_path)
|
||
n = db.record_motion_events([
|
||
{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 5},
|
||
{"event_id": 2, "camera_id": 2, "event_type": 10, "start_time": 2000, "duration": 3},
|
||
])
|
||
assert n == 2
|
||
rows = db._conn.execute("SELECT * FROM ss_motion_events ORDER BY event_id").fetchall()
|
||
assert len(rows) == 2
|
||
# 重复推送同一个 event_id（幂等）应该更新而不是新增一行
|
||
db.record_motion_events(
|
||
[{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1000, "duration": 99}])
|
||
rows = db._conn.execute("SELECT * FROM ss_motion_events").fetchall()
|
||
assert len(rows) == 2
|
||
updated = db._conn.execute(
|
||
"SELECT duration FROM ss_motion_events WHERE event_id=1").fetchone()
|
||
assert updated['duration'] == 99
|
||
|
||
|
||
def test_record_motion_events_skips_missing_event_id(tmp_path):
|
||
db = _db(tmp_path)
|
||
n = db.record_motion_events([{"camera_id": 2, "start_time": 1000}])
|
||
assert n == 0
|
||
|
||
|
||
def test_heartbeat_age_none_when_never_recorded(tmp_path):
|
||
db = _db(tmp_path)
|
||
assert db.get_motion_heartbeat_age_sec() is None
|
||
|
||
|
||
def test_heartbeat_age_near_zero_right_after_recording(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.record_motion_heartbeat()
|
||
age = db.get_motion_heartbeat_age_sec()
|
||
assert age is not None and age < 5
|
||
|
||
|
||
def test_has_motion_in_range_local_fails_open_without_heartbeat(tmp_path):
|
||
"""核心诉求: 从未收到过心跳（冷启动，NAS 推送链路还没接上）必须 fail-open，
|
||
不能因为本地表是空的就悄悄跳过分析。"""
|
||
db = _db(tmp_path)
|
||
assert db.has_motion_in_range_local(1000, 2000) is None
|
||
|
||
|
||
def test_has_motion_in_range_local_fails_open_when_heartbeat_stale(tmp_path):
|
||
"""核心诉求: 表里有大量历史运动事件（曾经推送链路是健康的），但心跳已经
|
||
过期太久（NAS 服务挂了/网络断了/DSM Webhook 规则被误关）——这时候不能信任
|
||
"查询结果是 0 条 = 确认无运动"，必须当作链路已死，fail-open。"""
|
||
db = _db(tmp_path)
|
||
db.record_motion_events(
|
||
[{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 500, "duration": 10}])
|
||
_set_heartbeat_age(db, 1000) # 超过默认阈值 900s
|
||
assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None
|
||
|
||
|
||
def test_has_motion_in_range_local_trusts_result_when_heartbeat_fresh(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.record_motion_heartbeat()
|
||
assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is False
|
||
db.record_motion_events(
|
||
[{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 2500, "duration": 5}])
|
||
assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is True
|
||
|
||
|
||
def test_has_motion_in_range_local_respects_heartbeat_boundary(tmp_path):
|
||
db = _db(tmp_path)
|
||
_set_heartbeat_age(db, 899)
|
||
assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is not None
|
||
_set_heartbeat_age(db, 901)
|
||
assert db.has_motion_in_range_local(2000, 3000, max_heartbeat_age_sec=900) is None
|
||
|
||
|
||
def test_has_motion_in_range_local_overlap_semantics(tmp_path):
|
||
"""事件区间 [start_time, start_time+duration] 只要和查询窗口有重叠就算命中，
|
||
不要求事件完全落在窗口内部（也不要求窗口完全覆盖事件）。"""
|
||
db = _db(tmp_path)
|
||
db.record_motion_heartbeat()
|
||
# 事件在窗口开始之前就开始，但持续到窗口内 -> 应该命中
|
||
db.record_motion_events(
|
||
[{"event_id": 1, "camera_id": 2, "event_type": 10, "start_time": 1990, "duration": 20}])
|
||
assert db.has_motion_in_range_local(2000, 3000) is True
|
||
|
||
|
||
def test_has_motion_in_range_local_ignores_non_motion_event_type(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.record_motion_heartbeat()
|
||
db.record_motion_events(
|
||
[{"event_id": 1, "camera_id": 2, "event_type": 99, "start_time": 2500, "duration": 5}])
|
||
assert db.has_motion_in_range_local(2000, 3000) is False
|
||
|
||
|
||
def test_has_motion_in_range_local_filters_by_camera_id(tmp_path):
|
||
db = _db(tmp_path)
|
||
db.record_motion_heartbeat()
|
||
db.record_motion_events(
|
||
[{"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_rewrites_description_text(tmp_path):
|
||
"""核心诉求: description 是大模型写的自然语言句子，"人物A/人物B"这类 uid
|
||
会直接以文字形式嵌在句子里，只改 person_list_json/person_appearances_json
|
||
这些结构化字段的话，事件卡片上方徽章显示对了，描述文字里还是旧 uid，两处
|
||
对不上——description 也要做文本替换。"""
|
||
db = _db(tmp_path)
|
||
vid = db.ensure_video("motion_2_2000.mp4", "/tmp/motion_2_2000.mp4",
|
||
event_start_time="2026-08-22 13:00:00")
|
||
events = [
|
||
{"timestamp": "13:29:24",
|
||
"description": "人物B双手叉腰站在客厅中央；人物A在远处厨房；儿童已离开画面。",
|
||
"people": ["人物A", "人物B"],
|
||
"person_appearances": [
|
||
{"uid": "人物A", "features": {"gender": "男"}, "action": "站立"},
|
||
{"uid": "人物B", "features": {"gender": "女"}, "action": "叉腰"}]},
|
||
]
|
||
db.mark_video_processed(vid, "人物A和人物B都在客厅活动。", events,
|
||
["人物A", "人物B"], "gemini")
|
||
db.rewrite_event_person_names(vid, {"人物A": "爸爸", "人物B": "媳妇"})
|
||
|
||
ev_row = db._conn.execute(
|
||
"SELECT description FROM events WHERE video_id=?", (vid,)).fetchone()
|
||
assert ev_row["description"] == "媳妇双手叉腰站在客厅中央；爸爸在远处厨房；儿童已离开画面。"
|
||
|
||
v_row = db._conn.execute(
|
||
"SELECT summary_json FROM videos WHERE id=?", (vid,)).fetchone()
|
||
assert v_row["summary_json"] == "爸爸和媳妇都在客厅活动。"
|
||
|
||
|
||
def test_rewrite_event_person_names_longer_labels_replaced_before_shorter(tmp_path):
|
||
"""核心诉求: uid 可能带 "#2"/"#3" 这类同名冲突后缀，"人物A" 是 "人物A#2" 的
|
||
前缀——如果先替换短的 "人物A"，"人物A#2" 会被错误地部分命中变成"爷爷#2"，
|
||
而不是走它自己在 rename_map 里对应的正确目标。必须长的先替换。"""
|
||
db = _db(tmp_path)
|
||
vid = db.ensure_video("motion_3_3000.mp4", "/tmp/motion_3_3000.mp4",
|
||
event_start_time="2026-08-22 13:00:00")
|
||
events = [
|
||
{"timestamp": "13:00:01",
|
||
"description": "人物A和人物A#2一起在客厅。",
|
||
"people": ["人物A", "人物A#2"],
|
||
"person_appearances": [
|
||
{"uid": "人物A", "features": {"gender": "男"}, "action": "站立"},
|
||
{"uid": "人物A#2", "features": {"gender": "女"}, "action": "站立"}]},
|
||
]
|
||
db.mark_video_processed(vid, "摘要", events, ["人物A", "人物A#2"], "gemini")
|
||
db.rewrite_event_person_names(vid, {"人物A": "爷爷", "人物A#2": "媳妇"})
|
||
|
||
ev_row = db._conn.execute(
|
||
"SELECT description FROM events WHERE video_id=?", (vid,)).fetchone()
|
||
assert ev_row["description"] == "爷爷和媳妇一起在客厅。"
|
||
|
||
|
||
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"]) == ["爸爸"]
|
||
|
||
|
||
def test_delete_video_removes_video_and_events_rows(tmp_path):
|
||
db = _db(tmp_path)
|
||
vid = _seed_video_with_events(db)
|
||
|
||
local_path = db.delete_video(vid)
|
||
|
||
assert local_path == f"/tmp/motion_1_1000.mp4"
|
||
assert db._conn.execute("SELECT * FROM videos WHERE id=?", (vid,)).fetchone() is None
|
||
assert db._conn.execute("SELECT * FROM events WHERE video_id=?", (vid,)).fetchall() == []
|
||
|
||
|
||
def test_delete_video_removes_disk_file(tmp_path):
|
||
db = _db(tmp_path)
|
||
clip_path = tmp_path / "motion_9_2000.mp4"
|
||
clip_path.write_bytes(b"fake mp4 bytes")
|
||
vid = db.ensure_video("motion_9_2000.mp4", str(clip_path), event_start_time="2026-08-22 10:00:00")
|
||
db.mark_video_processed(vid, "摘要", [], [], "gemini")
|
||
|
||
db.delete_video(vid)
|
||
|
||
assert not clip_path.exists()
|
||
|
||
|
||
def test_delete_video_missing_file_on_disk_does_not_raise(tmp_path):
|
||
"""核心诉求: local_path 指向的文件已经不存在（比如手动清理过）时，删除记录
|
||
本身不能因为 os.remove 报错而失败——文件缺失不是数据库操作的错误。"""
|
||
db = _db(tmp_path)
|
||
vid = db.ensure_video("motion_9_2000.mp4", str(tmp_path / "already_gone.mp4"),
|
||
event_start_time="2026-08-22 10:00:00")
|
||
db.mark_video_processed(vid, "摘要", [], [], "gemini")
|
||
|
||
local_path = db.delete_video(vid)
|
||
|
||
assert local_path == str(tmp_path / "already_gone.mp4")
|
||
assert db._conn.execute("SELECT * FROM videos WHERE id=?", (vid,)).fetchone() is None
|
||
|
||
|
||
def test_delete_video_nonexistent_returns_none(tmp_path):
|
||
db = _db(tmp_path)
|
||
assert db.delete_video(99999) is None
|
||
|
||
|
||
def test_get_oldest_purgeable_material_none_when_empty(tmp_path):
|
||
db = _db(tmp_path)
|
||
assert db.get_oldest_purgeable_material() is None
|
||
|
||
|
||
def test_get_oldest_purgeable_material_ignores_motion_clips(tmp_path):
|
||
"""核心诉求: 运动片段（motion_ 前缀）是独立的分析产物，事件时间轴/人物
|
||
头像都依赖它，磁盘清理绝不能碰它，只能清理原始整段素材。"""
|
||
db = _db(tmp_path)
|
||
vid = db.ensure_video("motion_1_1000.mp4", "/tmp/motion_1_1000.mp4",
|
||
event_start_time="2026-08-22 10:00:00")
|
||
db.mark_video_processed(vid, "摘要", [], [], "gemini")
|
||
assert db.get_oldest_purgeable_material() is None
|
||
|
||
|
||
def test_get_oldest_purgeable_material_ignores_non_done_status(tmp_path):
|
||
"""核心诉求: 还在 pending/processing 的素材不能被清理，避免删掉还没
|
||
来得及处理的数据。"""
|
||
db = _db(tmp_path)
|
||
db.ensure_video("Generic_ONVIF-001-20260815-000000.mp4",
|
||
"/tmp/Generic_ONVIF-001-20260815-000000.mp4")
|
||
assert db.get_oldest_purgeable_material() is None
|
||
|
||
|
||
def test_get_oldest_purgeable_material_returns_oldest_done_material(tmp_path):
|
||
db = _db(tmp_path)
|
||
vid1 = db.ensure_video("Generic_ONVIF-001-20260815-000000.mp4",
|
||
"/tmp/Generic_ONVIF-001-20260815-000000.mp4")
|
||
db.mark_video_processed(vid1, "（整段素材已分割 0 段运动片段）", [], [], 'motion_segment')
|
||
vid2 = db.ensure_video("Generic_ONVIF-001-20260816-000000.mp4",
|
||
"/tmp/Generic_ONVIF-001-20260816-000000.mp4")
|
||
db.mark_video_processed(vid2, "（整段素材已分割 0 段运动片段）", [], [], 'motion_segment')
|
||
|
||
candidate = db.get_oldest_purgeable_material()
|
||
|
||
assert candidate["id"] == vid1
|
||
assert candidate["local_path"] == "/tmp/Generic_ONVIF-001-20260815-000000.mp4"
|
||
|
||
|
||
def test_delete_video_does_not_touch_ss_motion_events(tmp_path):
|
||
"""核心诉求: ss_motion_events 是运动侦测源事件，跟切出来的视频片段生命周期
|
||
独立，删视频不该连带删掉源事件（否则分割逻辑的幂等判断会被破坏）。"""
|
||
db = _db(tmp_path)
|
||
db.record_motion_events([
|
||
{"event_id": 555, "camera_id": 2, "event_type": 10,
|
||
"start_time": 1700000000, "duration": 10, "thumbnail_url": ""},
|
||
])
|
||
vid = db.ensure_video("motion_555_1700000000.mp4", "/tmp/motion_555_1700000000.mp4",
|
||
event_start_time="2026-08-22 10:00:00", motion_event_id=555)
|
||
db.mark_video_processed(vid, "摘要", [], [], "gemini")
|
||
|
||
db.delete_video(vid)
|
||
|
||
assert db.get_video_by_motion_event_id(555) is None
|
||
row = db._conn.execute("SELECT * FROM ss_motion_events WHERE event_id=?", (555,)).fetchone()
|
||
assert row is not None
|