feat: 新增视频分析流水线、Web界面、孤儿清理工具;修复健壮性问题

- 新增 analysis/ 模块:MOG2活动检测、YOLO人物跟踪、MediaPipe摔倒检测、Ollama活动打标、缩略图抽取
- 新增 webapp/:Flask事件浏览界面,按需HEVC→H264转码缓存
- 新增 analyze.py:7阶段状态机流水线(scan→backfill→motion→person→fall→activity→thumbnail)
- 新增 cleanup_orphans.py:清理文件已删除的数据库孤儿记录
- 修复:motion/person/activity三阶段加单视频异常兜底,避免批次中断
- 修复:ollama_client加健康检查和3次重试,num_ctx从8192降到2048
- 修复:transcode_cache加fcntl文件锁,防止并发转码同一视频
- 修复:SQLite开WAL模式+busy_timeout,支持读写并发
- 修复:sync.py ffmpeg路径改为shutil.which查找,删除automate.py死代码PYTHON
- 修复:thumbnail失败加3次重试上限,避免无限重试损坏文件
- 新增:requirements.lock锁定版本、README文档
- 更新:.gitignore排除.playwright-mcp、截图、备份文件
This commit is contained in:
yuming
2026-09-18 15:52:49 +08:00
parent 203cca42a2
commit b3608f534e
31 changed files with 2792 additions and 27 deletions
+24 -1
View File
@@ -1,11 +1,21 @@
# 敏感配置和缓存 # 敏感配置和缓存
config.py /config.py
token_cache.json token_cache.json
sync.log sync.log
# 下载的录像(每天几百 MB) # 下载的录像(每天几百 MB)
recordings/ recordings/
# 本地视频分析产出(运行时生成)
analysis.db
analysis.db.bak.*
thumbnails/
transcoded/
# 模型权重(体积大,首次运行自动下载)
analysis/models/
yolo11n.pt
# Python # Python
__pycache__/ __pycache__/
*.pyc *.pyc
@@ -20,3 +30,16 @@ __pycache__/
# 临时 # 临时
*.tmp *.tmp
/tmp/ /tmp/
# 重复视频检测报告(运行时生成)
dedup_report.txt
# Playwright MCP 运行日志/快照
.playwright-mcp/
# 开发截图
webapp-check.png
webapp-index.png
# Frida
*.frida
+170
View File
@@ -0,0 +1,170 @@
# ezviz-sync — 萤石摄像头录像同步与智能分析
从萤石(EZVIZ)摄像头同步 SD 卡报警录像到本地,对录像做 AI 分析(人物检测、摔倒检测、活动语义打标),并提供 Web 界面浏览和回放。
目标设备:萤石 CB60,序列号 `BH5455783`。
## 目录结构
```
ezviz-sync/
├── automate.py # 主力下载脚本:ADB 自动化模拟器内萤石 App 批量下载
├── sync.py # [已废弃] 开放平台 API 下载方案,保留作备选
├── probe.py # API 探测调试脚本
├── analyze.py # 视频分析主入口(7 阶段流水线)
├── config.py # 萤石开放平台配置(密钥,已 gitignore)
├── config.example.py # 配置模板
├── requirements.txt # 直接依赖
├── requirements.lock # 锁定版本的完整依赖(pip freeze)
├── recordings/ # 下载的录像,按 设备号/日期/ 归档
├── analysis.db # SQLite 元数据库(运行时生成)
├── thumbnails/ # 事件缩略图(运行时生成)
├── transcoded/ # Web 播放转码缓存(运行时生成)
├── analysis/ # 分析模块
│ ├── db.py # SQLite 建表与 CRUD
│ ├── config.py # 分析参数配置(阈值、模型名、路径)
│ ├── video_scanner.py # 扫描录像文件入库 + MD5 查重
│ ├── motion_filter.py # MOG2 背景建模检测活动窗口
│ ├── person_tracker.py # YOLO11 + ByteTrack 人物跟踪
│ ├── fall_detector.py # MediaPipe Pose 摔倒检测
│ ├── activity_tagger.py # Ollama VLM 活动语义打标
│ ├── thumbnail.py # 事件缩略图抽取
│ ├── timestamp_reader.py # OSD 时间戳水印识别(模板匹配)
│ ├── backfill_start_time.py # 精确时间戳回填
│ ├── dedup_finder.py # 重复视频检测(生成报告)
│ ├── dedup_cleanup.py # 重复视频清理(默认 dry-run)
│ ├── cleanup_orphans.py # 清理文件已删除的数据库孤儿记录
│ ├── hashing.py # MD5 计算
│ ├── ollama_client.py # Ollama VLM 调用封装
│ ├── frida_spawn.py # Frida SSL pinning 绕过(抓包用)
│ ├── unpin.js # Frida 脱壳脚本
│ ├── models/ # MediaPipe pose 模型权重
│ └── templates/ # OSD 数字识别模板库
├── webapp/ # Flask Web 界面
│ ├── app.py # 路由:事件列表、统计、视频流、缩略图
│ ├── transcode_cache.py # 按需转码缓存(HEVC→H264)
│ ├── templates/index.html
│ └── static/ # app.js, style.css
└── yolo11n.pt # YOLO11 目标检测权重(首次运行自动下载)
```
## 环境依赖
| 工具 | 用途 | 安装 |
|---|---|---|
| Python 3.12 | 运行环境 | `pyenv install 3.12` 或系统包管理器 |
| ffmpeg / ffprobe | 视频处理、抽帧、转码 | `brew install ffmpeg` |
| ImageMagick (`magick`) | 截图像素分析(automate.py 蓝勾/行检测) | `brew install imagemagick` |
| Android SDK + ADB | 模拟器控制 | `brew install --cask android-platform-tools` |
| Android 模拟器 | 运行萤石 App | Android Studio AVD 或第三方模拟器 |
| Ollama | 本地 VLM 推理(活动打标) | `brew install ollama && ollama pull qwen2.5vl:7b` |
## 安装
```bash
# 1. 克隆项目
git clone <repo> && cd ezviz-sync
# 2. 创建虚拟环境
python3.12 -m venv .venv
source .venv/bin/activate
# 3. 安装依赖(用 lock 文件保证版本一致)
pip install -r requirements.lock
# 4. 配置萤石开放平台
cp config.example.py config.py
# 编辑 config.py 填入 AppKey / AppSecret / 设备序列号 / 验证码
# 5. 初始化数据库(首次运行 analyze.py 自动建表)
python analyze.py --only-stage scan
```
## 使用方法
### 1. 下载录像
```bash
# 下载昨天的录像(默认)
python automate.py
# 下载最近 7 天的录像
python automate.py 7
```
下载流程:启动模拟器内萤石 App → 导航到指定日期 → 批量下载 → adb pull 到本地 `recordings/设备号/日期/`。
### 2. 分析录像
```bash
# 跑完整流水线(scan → backfill → motion → person → fall → activity → thumbnail)
python analyze.py
# 只跑某个阶段
python analyze.py --only-stage motion
# 小批量测试
python analyze.py --only-stage person --limit 10
# 强制重跑某个阶段(回退已完成的视频状态)
python analyze.py --only-stage activity --force
```
流水线状态机:`pending → motion_done → person_done → fall_done → activity_done`,每个阶段只处理上一阶段完成的视频,重复运行幂等。
### 3. 启动 Web 界面
```bash
python -m webapp.app
# 打开 http://127.0.0.1:5050
```
功能:按事件类型/日期筛选、点击事件跳转播放对应时间点、转码缓存管理。
### 4. 维护工具
```bash
# 检测重复视频(只生成报告,不删除)
python -m analysis.dedup_finder
# 清理重复视频(先预览,确认后加 --execute)
python -m analysis.dedup_cleanup
python -m analysis.dedup_cleanup --execute
# 清理数据库中文件已不存在的孤儿记录
python -m analysis.cleanup_orphans # dry-run
python -m analysis.cleanup_orphans --execute
```
## 分析流水线详解
| 阶段 | 输入状态 | 输出状态 | 方法 | 说明 |
|---|---|---|---|---|
| scan | — | pending/corrupt/duplicate | ffprobe + MD5 | 扫描 recordings/ 入库,按 MD5 标记重复 |
| backfill | pending | pending(不变) | 模板匹配 OCR | 读取画面 OSD 时间戳,回填精确 start_time |
| motion | pending | motion_done | MOG2 背景建模 | 检测有活动的时间窗口,空场景跳过后续重模型 |
| person | motion_done | person_done | YOLO11n + ByteTrack | 只在活动窗口内跑人物检测跟踪 |
| fall | person_done | fall_done | MediaPipe Pose | 只在有人的时间段跑姿态估计,规则引擎判摔倒 |
| activity | fall_done | activity_done | Ollama Qwen2.5-VL | 活动窗口抽帧,VLM 分类为 走动/进食/坐下/站立/离开/其他 |
| thumbnail | — | — | ffmpeg 抽帧 | 给每条事件生成缩略图 |
## 关键设计决策
- **下载用 ADB 自动化而非开放平台 API**:开放平台 API 的 FLV 回放流把 24 小时录像拼接成一个文件,丢失了分段信息和精确时间戳;ADB 方案直接下载 App 内的分段报警录像,文件名含时间戳。
- **分析流水线分层过滤**:先用轻量 MOG2 预筛活动窗口,再在窗口内跑 YOLO,有人的时间段才跑姿态估计和 VLM,避免对空场景浪费算力。
- **摔倒检测用规则引擎而非 VLM**:安全信号需要可解释、可调阈值、宁可多报不漏报,MediaPipe 关键点 + "直立骤变横躺并持续"规则比模型黑盒更可靠。
- **Web 播放按需转码**:源文件是 HEVC 编码且 mp4 容器 sample entry 标签缺失,浏览器无法直接播放;第一次播放时用 VideoToolbox 硬件转码为 H264 并缓存,之后复用。
## 常见问题
**Q: automate.py 报错 `magick: command not found`**
A: 没装 ImageMagick。`brew install imagemagick`。automate.py 的蓝勾检测和卡片行定位都依赖 `magick` 命令。
**Q: analyze.py 跑 activity 阶段很慢/卡住**
A: 检查 Ollama 是否在运行:`ollama ps`。模型 `qwen2.5vl:7b` 需要约 6GB 显存/内存。
**Q: Web 端视频播放不了**
A: 首次播放会触发转码(HEVC→H264),大文件可能需要几十秒。转码缓存在 `transcoded/`,可在 Web 界面清空。
**Q: 数据库里有很多 duplicate 记录**
A: automate.py 早期版本有重复下载问题(已修复),同一段录像被下载成多份。scan 阶段按 MD5 自动标记 duplicate 并跳过分析。可用 `dedup_cleanup.py` 清理多余文件。
+121
View File
@@ -0,0 +1,121 @@
"""在 motion_filter 找出的活动窗口内按固定间隔抽帧,调用本地 Ollama VLM
做限定词表活动分类,连续同标签的采样点合并成一条 activity 事件。
"""
import json
from datetime import datetime, timedelta
import cv2
from analysis.config import ACTIVITY_SAMPLE_INTERVAL_SEC, ACTIVITY_MERGE_GAP_SEC, PROJECT_ROOT
from analysis.db import get_all_videos, insert_event, update_video_status
from analysis.ollama_client import classify_frame
def _sample_labels(file_path, windows, fps):
"""在活动窗口内按固定秒数间隔抽帧分类,返回 [(t, label, raw_text), ...]。"""
cap = cv2.VideoCapture(str(file_path))
if not cap.isOpened():
return []
results = []
for start_sec, end_sec in windows:
t = start_sec
while t <= end_sec:
cap.set(cv2.CAP_PROP_POS_FRAMES, int(t * fps))
ret, frame = cap.read()
if ret:
label, raw_text = classify_frame(frame)
results.append((t, label, raw_text))
t += ACTIVITY_SAMPLE_INTERVAL_SEC
cap.release()
return results
def _merge_labeled_samples(samples):
"""连续同标签的采样点合并成一段 {start, end, label, raw_texts}。"""
segments = []
for t, label, raw_text in samples:
if (segments and segments[-1]["label"] == label
and (t - segments[-1]["end"]) <= ACTIVITY_MERGE_GAP_SEC):
segments[-1]["end"] = t
segments[-1]["raw_texts"].append(raw_text)
else:
segments.append({"start": t, "end": t, "label": label, "raw_texts": [raw_text]})
return segments
def process_video(video):
"""返回 (segment_count, failed)。"""
try:
windows = json.loads(video["motion_windows_json"] or "[]")
except (json.JSONDecodeError, TypeError):
windows = []
if not windows:
update_video_status(video["id"], "activity_done")
return 0, False
file_path = PROJECT_ROOT / video["file_path"]
if not file_path.exists():
update_video_status(video["id"], "activity_failed", error_message="文件不存在")
return 0, True
try:
fps = video["fps"] or 15.0
samples = _sample_labels(file_path, windows, fps)
except Exception as e:
update_video_status(video["id"], "activity_failed", error_message=str(e))
return 0, True
segments = _merge_labeled_samples(samples)
start_time = None
if video["start_time"]:
try:
start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S")
except ValueError:
pass # 只精确到天的粗粒度值,无法算出绝对时间戳,timestamp_abs留空
for seg in segments:
timestamp_abs = (start_time + timedelta(seconds=seg["start"])).isoformat() if start_time else None
insert_event({
"video_id": video["id"],
"event_type": "activity",
"label": seg["label"],
"start_offset_sec": seg["start"],
"end_offset_sec": seg["end"],
"timestamp_abs": timestamp_abs,
"confidence": None,
"track_id": None,
"thumbnail_path": None,
"extra_json": json.dumps({"raw_texts": seg["raw_texts"]}, ensure_ascii=False),
"created_at": datetime.now().isoformat(),
})
update_video_status(video["id"], "activity_done")
return len(segments), False
def process_all(limit=None):
videos = get_all_videos(status="fall_done")
if limit:
videos = videos[:limit]
processed = 0
total_segments = 0
failed = 0
for video in videos:
segments, is_failed = process_video(video)
total_segments += segments
if is_failed:
failed += 1
processed += 1
return {"processed": processed, "total_segments": total_segments, "failed": failed}
if __name__ == "__main__":
print("===== 开始活动语义打标 =====")
result = process_all()
print(f"→ 处理 {result['processed']} 个视频,共标注 {result['total_segments']} 段活动")
print("===== 打标完成 =====")
+48
View File
@@ -0,0 +1,48 @@
"""把 OSD 时间戳水印识别出的精确到秒的时间,回填到 videos.start_time。
video_scanner 登记时 start_time 只精确到"天"(因为文件名时间戳是下载时刻而非
录制时刻)。这里用 timestamp_reader 重新读取每个视频首帧的画面水印,识别成功
则覆盖为精确到秒的时间;识别失败(分辨率未标定/置信度不足)则保留原有粗粒度
值不动,不强行覆盖成错误数据。
"""
from analysis.config import PROJECT_ROOT
from analysis.db import get_all_videos, update_start_time
from analysis.timestamp_reader import read_timestamp
# 720p 方案实测通过率仅约19%(模板库覆盖的时段太窄,泛化差),投入产出比太低,
# 暂不处理,720p视频保留 video_scanner 登记时"只精确到天"的粗粒度 start_time。
BACKFILL_RESOLUTIONS = {(2560, 1440)}
def process_all(limit=None):
videos = [
v for v in get_all_videos()
if v["status"] != "corrupt" and (v["width"], v["height"]) in BACKFILL_RESOLUTIONS
]
if limit:
videos = videos[:limit]
updated = 0
fallback = 0
for video in videos:
file_path = PROJECT_ROOT / video["file_path"]
precise_start_time = read_timestamp(file_path, capture_date=video["capture_date"])
if precise_start_time is None:
fallback += 1
continue
update_start_time(video["id"], precise_start_time)
updated += 1
return {"total": len(videos), "updated": updated, "fallback": fallback}
if __name__ == "__main__":
import sys
limit = int(sys.argv[1]) if len(sys.argv) > 1 else None
print("===== 开始回填精确时间戳 =====")
result = process_all(limit=limit)
print(f"→ 共处理 {result['total']} 条,成功回填 {result['updated']} 条,"
f"识别失败保留粗粒度值 {result['fallback']} 条")
print("===== 回填完成 =====")
+90
View File
@@ -0,0 +1,90 @@
"""清理数据库中文件已不存在的孤儿记录。
video_scanner 入库后,如果录像文件被手动删除或被 dedup_cleanup 清理,
数据库里的 videos/events 记录不会自动消失,导致:
- analyze.py 对不存在的文件跑分析,全部失败或产生空结果
- webapp 点击播放 404
- 统计数据虚高
本脚本扫描 videos 表,删除 file_path 不存在的记录及其关联 events,
并同步清理孤儿缩略图文件。执行前请确保已备份 analysis.db。
"""
import os
import sqlite3
from analysis.config import PROJECT_ROOT, THUMBNAILS_DIR, DB_PATH
def cleanup_orphans(dry_run=True):
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
videos = conn.execute("SELECT id, file_path, status FROM videos").fetchall()
to_delete = []
for v in videos:
full_path = PROJECT_ROOT / v["file_path"]
if not full_path.exists():
to_delete.append(v)
print(f"videos 总数: {len(videos)}")
print(f"文件不存在的孤儿记录: {len(to_delete)}")
# 按状态统计
by_status = {}
for v in to_delete:
by_status[v["status"]] = by_status.get(v["status"], 0) + 1
for status, cnt in sorted(by_status.items(), key=lambda x: -x[1]):
print(f" {status}: {cnt}")
if dry_run:
print("\n[dry-run] 未执行任何删除。加 --execute 实际执行。")
conn.close()
return
# 删除关联 events 和缩略图
deleted_events = 0
deleted_thumbs = 0
for v in to_delete:
events = conn.execute(
"SELECT id, thumbnail_path FROM events WHERE video_id = ?", (v["id"],)
).fetchall()
for e in events:
if e["thumbnail_path"]:
thumb_path = PROJECT_ROOT / e["thumbnail_path"]
if thumb_path.exists():
thumb_path.unlink()
deleted_thumbs += 1
conn.execute("DELETE FROM events WHERE video_id = ?", (v["id"],))
deleted_events += len(events)
# 删除 video 记录
video_ids = [v["id"] for v in to_delete]
if video_ids:
placeholders = ",".join("?" * len(video_ids))
conn.execute(f"DELETE FROM videos WHERE id IN ({placeholders})", video_ids)
conn.commit()
# 清理孤儿缩略图(events 已删但文件还在的)
event_thumbs = set(
r[0] for r in conn.execute("SELECT thumbnail_path FROM events WHERE thumbnail_path IS NOT NULL")
)
orphan_thumbs = 0
if THUMBNAILS_DIR.exists():
for f in THUMBNAILS_DIR.glob("*.jpg"):
rel = str(f.relative_to(PROJECT_ROOT))
if rel not in event_thumbs:
f.unlink()
orphan_thumbs += 1
conn.close()
print(f"\n已删除 videos 记录: {len(to_delete)}")
print(f"已删除 events 记录: {deleted_events}")
print(f"已清理缩略图文件: {deleted_thumbs + orphan_thumbs}")
if __name__ == "__main__":
import sys
dry_run = "--execute" not in sys.argv
cleanup_orphans(dry_run=dry_run)
+65
View File
@@ -0,0 +1,65 @@
"""本地视频分析模块的集中配置:路径、阈值、模型名、词表等所有可调参数。"""
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# 录像来源目录:recordings/{设备号}/{YYYYMMDD}/*.mp4
RECORDINGS_DIR = PROJECT_ROOT / "recordings"
# 分析结果落地位置
DB_PATH = PROJECT_ROOT / "analysis.db"
THUMBNAILS_DIR = PROJECT_ROOT / "thumbnails"
# 网页播放按需转码缓存(源文件mp4容器HEVC轨道sample entry编码标签缺失,
# 浏览器无法直接播放,第一次点击时转码为标准H264并缓存在此,避免重复转码)
TRANSCODED_DIR = PROJECT_ROOT / "transcoded"
# 活动语义打标的限定词表(供 VLM 分类 prompt 和 events.label 取值域共用)
ACTIVITY_LABELS = ["走动", "进食", "坐下", "站立", "离开", "其他"]
# Ollama 本地模型
OLLAMA_MODEL = "qwen2.5vl:7b"
# 活动窗口检测(MOG2 背景建模)参数
# 帧差法对"人物占画面小、动作幅度小"的场景(如老人在客厅整理东西)容易漏检,
# 改用 MOG2 自适应背景建模:背景随时间缓慢更新,前景掩码对小幅度/慢速运动更敏感。
MOTION_RESIZE_WIDTH = 320 # 缩放到此宽度再做背景建模,加速计算
MOTION_HISTORY = 120 # MOG2 背景模型参考的历史帧数
MOTION_VAR_THRESHOLD = 16 # 像素值与背景模型的马氏距离阈值,越小越敏感(越容易判定为前景)
MOTION_WARMUP_FRAMES = 15 # 背景模型尚未收敛的前N个采样帧不计入判断,避免开场误判为"全员活动"
MOTION_SHORT_CLIP_SEC = 2.0 # 短于此时长的视频(约占全量一半,中位时长仅0.56秒)帧数不够MOG2预热,
# 触发即录的告警片段本身就是一次完整活动,直接整段视为活动窗口,不跑MOG2
MOTION_PIXEL_RATIO = 0.015 # 一帧里前景像素(不含阴影)占比超过此值才算"有活动"
MOTION_MERGE_GAP_SEC = 1.5 # 两段活动间隔小于此秒数则合并为一段
MOTION_MIN_DURATION_SEC = 0.5 # 短于此秒数的活动段视为噪声,丢弃
MOTION_SAMPLE_STRIDE = 1 # 每隔几帧采样一次(1=逐帧,长视频可调大提速)
# 人物出现次数统计(YOLO11n 检测 + ByteTrack 跟踪)参数
# 只在 motion_filter 找出的活动窗口内跑,空场景直接跳过,省算力。
PERSON_MODEL_NAME = "yolo11n.pt" # ultralytics 官方COCO预训练权重,首次运行自动下载
PERSON_DEVICE = "mps" # M系列芯片走Metal加速
PERSON_CONF_THRESHOLD = 0.4 # 检测置信度阈值,低于此值的框直接丢弃
PERSON_SAMPLE_STRIDE = 2 # 每隔几帧检测一次(视频多为15fps,隔帧检测足够跟踪连续性)
PERSON_MIN_TRACK_DURATION_SEC = 0.3 # 持续时间短于此值的track视为误检噪声,丢弃
# 摔倒检测(MediaPipe Pose 关键点 + 规则引擎)参数
# 只在 person_tracker 已确认"有人"的时间段内跑姿态估计,空场景不跑。
# 安全信号宁可多报不漏报:判定摔倒 = 短时间内从"直立"骤变为"横躺",且横躺状态持续一段时间
# (躯干角度变水平这个单一条件区分不了"摔倒"和"弯腰/坐下",必须叠加"骤变"+"持续"两个条件)
FALL_MODEL_PATH = PROJECT_ROOT / "analysis" / "models" / "pose_landmarker_lite.task"
FALL_SAMPLE_STRIDE = 2 # 每隔几帧跑一次姿态估计
FALL_MIN_VISIBILITY = 0.5 # 关键点可见度低于此值不参与姿态计算,视为遮挡/未检出
FALL_ASPECT_RATIO_THRESH = 1.3 # 人体关键点bbox 宽/高 超过此值才可能是"横躺"
FALL_TORSO_ANGLE_UPRIGHT = 35 # 躯干(肩中点→髋中点连线)与垂直方向夹角(度),低于此值判定"直立"
FALL_TORSO_ANGLE_PRONE = 55 # 高于此值判定"趋于水平/横躺"
FALL_TRANSITION_MAX_SEC = 1.5 # "直立"变"横躺"必须在此时间内完成,排除缓慢弯腰/坐下/趟下
FALL_SUSTAIN_MIN_SEC = 1.0 # "横躺"状态至少持续这么久才确认为摔倒,排除弯腰瞬间的误判
FALL_GAP_TOLERANCE_SEC = 5.0 # 一次疑似摔倒事件期间,允许多久检测不到"横躺"样本仍视为
# 事件在继续(真实摔倒后姿态/角度常导致关键点长时间漏检,
# 不代表已经爬起来);超过此时长且中途也没等到"直立"信号,
# 才判定事件跟丢并结算
# 活动语义打标(本地 Ollama + Qwen2.5-VL 限定词表分类)参数
# 只在 motion_filter 找出的活动窗口内抽帧,VLM推理比传统CV慢得多,不能逐帧跑。
ACTIVITY_SAMPLE_INTERVAL_SEC = 2.0 # 每隔多少秒抽一帧送VLM分类
ACTIVITY_MERGE_GAP_SEC = 3.0 # 连续同标签采样点之间允许的最大间隔,超过则视为新的一段活动
+222
View File
@@ -0,0 +1,222 @@
"""analysis.db(SQLite)建表 + CRUD,唯一直接写 SQL 的地方。"""
import sqlite3
from contextlib import contextmanager
from analysis.config import DB_PATH
SCHEMA = """
CREATE TABLE IF NOT EXISTS videos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_serial TEXT NOT NULL,
capture_date TEXT NOT NULL,
file_path TEXT NOT NULL UNIQUE,
file_name TEXT NOT NULL,
file_size_bytes INTEGER,
duration_sec REAL,
fps REAL,
width INTEGER,
height INTEGER,
start_time TEXT,
status TEXT NOT NULL DEFAULT 'pending',
motion_windows_json TEXT,
error_message TEXT,
analyzed_at TEXT,
file_md5 TEXT,
duplicate_of INTEGER REFERENCES videos(id)
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id INTEGER NOT NULL REFERENCES videos(id),
event_type TEXT NOT NULL,
label TEXT,
start_offset_sec REAL,
end_offset_sec REAL,
timestamp_abs TEXT,
confidence REAL,
track_id INTEGER,
thumbnail_path TEXT,
extra_json TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_events_type_time ON events(event_type, timestamp_abs);
"""
@contextmanager
def get_connection():
conn = sqlite3.connect(DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row
# WAL 模式允许读写并发(analyze.py 写 + webapp 读不互锁),
# busy_timeout 避免 "database is locked" 直接报错
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db():
with get_connection() as conn:
conn.executescript(SCHEMA)
_migrate_add_dedup_columns(conn)
def _migrate_add_dedup_columns(conn):
"""老库没有 file_md5/duplicate_of 字段,CREATE TABLE IF NOT EXISTS 对已存在的表不生效,
这里用 ALTER TABLE 补齐(新建库这两列已经在 SCHEMA 里,ALTER 会因列已存在而报错,捕获忽略)。
"""
existing_columns = {row["name"] for row in conn.execute("PRAGMA table_info(videos)")}
if "file_md5" not in existing_columns:
conn.execute("ALTER TABLE videos ADD COLUMN file_md5 TEXT")
if "duplicate_of" not in existing_columns:
conn.execute("ALTER TABLE videos ADD COLUMN duplicate_of INTEGER REFERENCES videos(id)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_videos_md5 ON videos(file_md5)")
def video_exists(file_path):
with get_connection() as conn:
row = conn.execute(
"SELECT 1 FROM videos WHERE file_path = ?", (str(file_path),)
).fetchone()
return row is not None
def insert_video(video):
"""插入一条 video 记录,file_path 已存在则忽略(保证重复扫描幂等)。返回该记录的 id。"""
with get_connection() as conn:
conn.execute(
"""
INSERT OR IGNORE INTO videos (
device_serial, capture_date, file_path, file_name,
file_size_bytes, duration_sec, fps, width, height,
start_time, status, error_message, file_md5, duplicate_of
) VALUES (
:device_serial, :capture_date, :file_path, :file_name,
:file_size_bytes, :duration_sec, :fps, :width, :height,
:start_time, :status, :error_message, :file_md5, :duplicate_of
)
""",
video,
)
row = conn.execute(
"SELECT id FROM videos WHERE file_path = ?", (video["file_path"],)
).fetchone()
return row["id"]
def get_video_id_by_md5(file_md5):
with get_connection() as conn:
row = conn.execute(
"SELECT id FROM videos WHERE file_md5 = ?", (file_md5,)
).fetchone()
return row["id"] if row else None
def get_videos_missing_md5():
with get_connection() as conn:
rows = conn.execute(
"SELECT id, file_path FROM videos WHERE file_md5 IS NULL"
).fetchall()
return [dict(r) for r in rows]
def update_video_md5(video_id, file_md5):
with get_connection() as conn:
conn.execute(
"UPDATE videos SET file_md5 = ? WHERE id = ?", (file_md5, video_id)
)
def count_videos():
with get_connection() as conn:
row = conn.execute("SELECT COUNT(*) AS c FROM videos").fetchone()
return row["c"]
def get_all_videos(status=None):
with get_connection() as conn:
if status is None:
rows = conn.execute("SELECT * FROM videos ORDER BY capture_date, file_name").fetchall()
else:
rows = conn.execute(
"SELECT * FROM videos WHERE status = ? ORDER BY capture_date, file_name", (status,)
).fetchall()
return [dict(r) for r in rows]
def update_video_status(video_id, status, error_message=None):
with get_connection() as conn:
conn.execute(
"UPDATE videos SET status = ?, error_message = ? WHERE id = ?",
(status, error_message, video_id),
)
def update_start_time(video_id, start_time):
with get_connection() as conn:
conn.execute(
"UPDATE videos SET start_time = ? WHERE id = ?",
(start_time, video_id),
)
def update_motion_windows(video_id, motion_windows_json, status, analyzed_at):
with get_connection() as conn:
conn.execute(
"UPDATE videos SET motion_windows_json = ?, status = ?, analyzed_at = ? WHERE id = ?",
(motion_windows_json, status, analyzed_at, video_id),
)
def insert_event(event):
with get_connection() as conn:
conn.execute(
"""
INSERT INTO events (
video_id, event_type, label, start_offset_sec, end_offset_sec,
timestamp_abs, confidence, track_id, thumbnail_path, extra_json, created_at
) VALUES (
:video_id, :event_type, :label, :start_offset_sec, :end_offset_sec,
:timestamp_abs, :confidence, :track_id, :thumbnail_path, :extra_json, :created_at
)
""",
event,
)
def get_video_by_id(video_id):
with get_connection() as conn:
row = conn.execute("SELECT * FROM videos WHERE id = ?", (video_id,)).fetchone()
return dict(row) if row else None
def update_event_thumbnail(event_id, thumbnail_path):
with get_connection() as conn:
conn.execute(
"UPDATE events SET thumbnail_path = ? WHERE id = ?",
(thumbnail_path, event_id),
)
def update_event_extra(event_id, extra_json):
with get_connection() as conn:
conn.execute(
"UPDATE events SET extra_json = ? WHERE id = ?",
(extra_json, event_id),
)
def get_events(event_type=None):
with get_connection() as conn:
if event_type is None:
rows = conn.execute("SELECT * FROM events ORDER BY timestamp_abs").fetchall()
else:
rows = conn.execute(
"SELECT * FROM events WHERE event_type = ? ORDER BY timestamp_abs", (event_type,)
).fetchall()
return [dict(r) for r in rows]
+130
View File
@@ -0,0 +1,130 @@
"""基于 dedup_finder 检测出的字节级重复文件,生成清理计划 / 执行实际清理。
处理原则:同一组 MD5 完全相同的文件里,只保留文件名(时间戳)最小的一份,
其余文件从磁盘删除;这些多余视频如果已经跑过 Phase3-5 分析,产生的
person_presence/fall/activity 事件会让统计虚高,所以对应的 videos/events
数据库记录、缩略图、转码缓存也一并清理,避免留下孤儿数据。
默认只生成计划(不改动任何文件/数据库),确认无误后加 --execute 才真正执行,
且执行前还会要求在终端输入 yes 二次确认。
"""
import argparse
from analysis.config import PROJECT_ROOT, TRANSCODED_DIR
from analysis.db import get_connection
from analysis.dedup_finder import find_duplicates
def _video_id_for_path(conn, rel_path):
row = conn.execute(
"SELECT id FROM videos WHERE file_path = ?", (str(rel_path),)
).fetchone()
return row["id"] if row else None
def build_plan():
"""返回删除计划列表,每项对应一个要删除的多余文件。"""
dup_groups = find_duplicates()
plan = []
with get_connection() as conn:
for paths in dup_groups.values():
paths = sorted(paths, key=lambda p: p.name)
keep, remove = paths[0], paths[1:]
for path in remove:
rel_path = path.relative_to(PROJECT_ROOT)
video_id = _video_id_for_path(conn, rel_path)
thumbnails = []
event_count = 0
if video_id is not None:
rows = conn.execute(
"SELECT thumbnail_path FROM events WHERE video_id = ?", (video_id,)
).fetchall()
event_count = len(rows)
thumbnails = [r["thumbnail_path"] for r in rows if r["thumbnail_path"]]
plan.append({
"keep_path": str(keep.relative_to(PROJECT_ROOT)),
"remove_path": str(rel_path),
"video_id": video_id,
"event_count": event_count,
"thumbnails": thumbnails,
"size_bytes": path.stat().st_size,
})
return plan
def print_plan(plan):
total_bytes = sum(p["size_bytes"] for p in plan)
total_events = sum(p["event_count"] for p in plan)
orphan = [p for p in plan if p["video_id"] is None]
print(f"计划删除 {len(plan)} 个重复文件,释放 {total_bytes / 1024 / 1024:.1f} MB")
print(f"将同步清理 {total_events} 条关联 events 记录")
if orphan:
print(f"其中 {len(orphan)} 个文件在 analysis.db 中还没有对应记录(尚未扫描过),只删文件不动数据库")
preview = plan[:20]
for p in preview:
print(f" - {p['remove_path']} (保留 {p['keep_path']}, video_id={p['video_id']}, events={p['event_count']})")
if len(plan) > len(preview):
print(f" ...(其余 {len(plan) - len(preview)} 条略)")
def execute_plan(plan):
deleted_files = 0
deleted_events = 0
deleted_videos = 0
with get_connection() as conn:
for p in plan:
remove_path = PROJECT_ROOT / p["remove_path"]
if remove_path.exists():
remove_path.unlink()
deleted_files += 1
for thumb in p["thumbnails"]:
thumb_path = PROJECT_ROOT / thumb
if thumb_path.exists():
thumb_path.unlink()
if p["video_id"] is not None:
cache_path = TRANSCODED_DIR / f"{p['video_id']}.mp4"
if cache_path.exists():
cache_path.unlink()
cur = conn.execute("DELETE FROM events WHERE video_id = ?", (p["video_id"],))
deleted_events += cur.rowcount
conn.execute("DELETE FROM videos WHERE id = ?", (p["video_id"],))
deleted_videos += 1
return {
"deleted_files": deleted_files,
"deleted_events": deleted_events,
"deleted_videos": deleted_videos,
}
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="重复视频清理:默认只打印计划,加 --execute 才真正执行")
parser.add_argument("--execute", action="store_true", help="实际执行删除")
args = parser.parse_args()
plan = build_plan()
print_plan(plan)
if not args.execute:
print("\n(当前只是计划预览,未做任何改动;确认无误后加 --execute 参数执行)")
else:
answer = input(f"\n确认删除以上 {len(plan)} 个文件及关联数据库记录?输入 yes 继续:")
if answer.strip().lower() == "yes":
result = execute_plan(plan)
print(
f"\n完成:删除文件 {result['deleted_files']} 个,"
f"删除 videos 记录 {result['deleted_videos']} 条,"
f"删除 events 记录 {result['deleted_events']} 条"
)
else:
print("已取消")
+67
View File
@@ -0,0 +1,67 @@
"""检测 recordings/ 目录下内容完全相同(字节级重复)的录像文件。
根因:automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题,会把同一张
缩略图卡片的下载重复触发,导致同一段录像被存成多份文件名不同的拷贝。
先按文件大小分组(成本低,能筛掉绝大多数内容不同的文件),只对同大小的文件
再计算 MD5 确认是否真的内容相同,避免对全部文件逐个算 MD5 的开销。
只生成报告,不删除任何文件、不改动数据库。
"""
from collections import defaultdict
from analysis.config import PROJECT_ROOT, RECORDINGS_DIR
from analysis.hashing import compute_md5
REPORT_PATH = PROJECT_ROOT / "dedup_report.txt"
def find_duplicates():
"""返回 {md5: [文件路径, ...]},只包含重复数≥2的分组。"""
by_size = defaultdict(list)
for path in RECORDINGS_DIR.rglob("*.mp4"):
by_size[path.stat().st_size].append(path)
by_md5 = defaultdict(list)
for size, paths in by_size.items():
if len(paths) < 2:
continue
for path in paths:
by_md5[compute_md5(path)].append(path)
return {md5: paths for md5, paths in by_md5.items() if len(paths) > 1}
def build_report():
dup_groups = find_duplicates()
lines = []
total_extra_files = 0
total_wasted_bytes = 0
for md5, paths in sorted(dup_groups.items(), key=lambda kv: -len(kv[1])):
paths = sorted(paths, key=lambda p: p.name)
size = paths[0].stat().st_size
extra = len(paths) - 1
total_extra_files += extra
total_wasted_bytes += extra * size
lines.append(f"[{md5}] {len(paths)} 份重复,单份 {size / 1024:.1f} KB")
for p in paths:
lines.append(f" {p.relative_to(PROJECT_ROOT)}")
summary = (
f"重复分组数:{len(dup_groups)}\n"
f"多余拷贝数(可释放的文件数):{total_extra_files}\n"
f"可释放空间:{total_wasted_bytes / 1024 / 1024:.1f} MB\n"
)
return summary + "\n" + "\n".join(lines)
if __name__ == "__main__":
print("===== 开始检测重复视频 =====")
report = build_report()
REPORT_PATH.write_text(report, encoding="utf-8")
print(report.split("\n\n")[0])
print(f"→ 详细清单已写入 {REPORT_PATH.relative_to(PROJECT_ROOT)}")
print("===== 检测完成 =====")
+266
View File
@@ -0,0 +1,266 @@
"""在 person_tracker 已确认"有人"的时间段内跑 MediaPipe Pose 关键点估计,
用确定性规则引擎判断是否摔倒,写入 events 表(event_type='fall')。
不用VLM做摔倒检测:这是安全信号,需要可解释、可调阈值、宁可多报不漏报的确定性
规则,而不是模型的"感觉"。规则:短时间内从"直立"骤变为"横躺",且"横躺"状态
持续一段时间——单独判断"是否躺平"分不清摔倒和弯腰/坐下,必须叠加"骤变"和
"持续"两个时间条件。
"""
import json
import math
from datetime import datetime, timedelta
import cv2
import mediapipe as mp
import numpy as np
from mediapipe.tasks import python as mp_python
from mediapipe.tasks.python import vision
from analysis.config import (
FALL_MODEL_PATH,
FALL_SAMPLE_STRIDE,
FALL_MIN_VISIBILITY,
FALL_ASPECT_RATIO_THRESH,
FALL_TORSO_ANGLE_UPRIGHT,
FALL_TORSO_ANGLE_PRONE,
FALL_TRANSITION_MAX_SEC,
FALL_SUSTAIN_MIN_SEC,
FALL_GAP_TOLERANCE_SEC,
PROJECT_ROOT,
)
from analysis.db import get_all_videos, get_events, insert_event, update_video_status
LEFT_SHOULDER, RIGHT_SHOULDER = 11, 12
LEFT_HIP, RIGHT_HIP = 23, 24
def _make_landmarker():
base_options = mp_python.BaseOptions(model_asset_path=str(FALL_MODEL_PATH))
options = vision.PoseLandmarkerOptions(
base_options=base_options,
running_mode=vision.RunningMode.VIDEO,
num_poses=1,
)
return vision.PoseLandmarker.create_from_options(options)
def _classify_pose(landmarks, width, height):
"""返回 'upright' / 'prone' / 'unknown'(关键点不足或角度介于两者之间)。"""
visible = [lm for lm in landmarks if lm.visibility >= FALL_MIN_VISIBILITY]
if len(visible) < 4:
return "unknown"
xs = [lm.x * width for lm in visible]
ys = [lm.y * height for lm in visible]
bbox_w = max(xs) - min(xs)
bbox_h = max(ys) - min(ys)
if bbox_h <= 0:
return "unknown"
aspect_ratio = bbox_w / bbox_h
shoulders = [landmarks[i] for i in (LEFT_SHOULDER, RIGHT_SHOULDER) if landmarks[i].visibility >= FALL_MIN_VISIBILITY]
hips = [landmarks[i] for i in (LEFT_HIP, RIGHT_HIP) if landmarks[i].visibility >= FALL_MIN_VISIBILITY]
if not shoulders or not hips:
return "unknown"
shoulder_mid = (sum(lm.x for lm in shoulders) / len(shoulders) * width,
sum(lm.y for lm in shoulders) / len(shoulders) * height)
hip_mid = (sum(lm.x for lm in hips) / len(hips) * width,
sum(lm.y for lm in hips) / len(hips) * height)
dx = abs(hip_mid[0] - shoulder_mid[0])
dy = abs(hip_mid[1] - shoulder_mid[1])
torso_angle = math.degrees(math.atan2(dx, dy + 1e-6)) # 0°=垂直直立,90°=水平躺平
if aspect_ratio >= FALL_ASPECT_RATIO_THRESH and torso_angle >= FALL_TORSO_ANGLE_PRONE:
return "prone"
if aspect_ratio < 1.0 and torso_angle <= FALL_TORSO_ANGLE_UPRIGHT:
return "upright"
return "unknown"
def _extract_pose_states(landmarker, cap, fps, width, height, start_sec, end_sec):
"""在给定时间段内采样跑姿态估计,返回 [(t, state), ...]。"""
start_frame = int(start_sec * fps)
end_frame = int(end_sec * fps)
cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
states = []
frame_idx = start_frame
while frame_idx <= end_frame:
ret, frame = cap.read()
if not ret:
break
if (frame_idx - start_frame) % FALL_SAMPLE_STRIDE == 0:
t = frame_idx / fps
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=np.ascontiguousarray(rgb))
result = landmarker.detect_for_video(mp_image, int(t * 1000))
if result.pose_landmarks:
state = _classify_pose(result.pose_landmarks[0], width, height)
else:
state = "unknown"
states.append((t, state))
frame_idx += 1
return states
def _find_fall_windows(states):
"""从时序姿态状态里找出"直立→骤变→横躺并持续"的片段,返回 [(start_sec, end_sec), ...]。
用"疑似摔倒事件"(episode)建模:只有真正观察到"直立"样本才会结束当前事件的计时;
事件进行期间即使长时间检测不到姿态(真实摔倒后人体角度/遮挡经常导致关键点大面积
漏检,不代表已经爬起来),也继续算同一次事件,直到确认恢复直立,或者长时间
(超过 FALL_GAP_TOLERANCE_SEC)连"横躺"样本都等不到才判定事件已经结束/跟丢。
"""
windows = []
last_upright_t = None
episode_start = None
episode_last_prone = None
def close_episode():
nonlocal episode_start, episode_last_prone
if episode_start is not None and (episode_last_prone - episode_start) >= FALL_SUSTAIN_MIN_SEC:
windows.append((episode_start, episode_last_prone))
episode_start = None
episode_last_prone = None
for t, state in states:
if state == "upright":
close_episode()
last_upright_t = t
elif state == "prone":
if episode_start is None:
# 只有紧接在"直立"之后、且间隔够短,才算一次骤变
if last_upright_t is not None and (t - last_upright_t) <= FALL_TRANSITION_MAX_SEC:
episode_start = t
episode_last_prone = t
# 否则忽略:不是从直立状态骤变而来(比如视频开头就是横躺,或者是缓慢躺下)
elif (t - episode_last_prone) <= FALL_GAP_TOLERANCE_SEC:
episode_last_prone = t
else:
# 距离上一次确认"横躺"太久,且中途也没等到"直立"信号,视为跟丢,结算当前事件
close_episode()
# state == "unknown" 不改变状态机,只是跳过这一帧(容忍短暂遮挡/漏检)
close_episode()
return windows
def _merge_windows(events):
"""合并人物track的时间区间(有重叠或紧邻就合并成一段)。
同一个人有时会被 ByteTrack 拆成多个track_id(跟踪中途丢失又重新编号),
这些track的时间区间往往高度重叠——如果按track各自跑一遍姿态估计,
同一次摔倒会被重复检测、重复插入,所以要先合并再处理,保证同一段画面只处理一次。
"""
spans = sorted((e["start_offset_sec"], e["end_offset_sec"]) for e in events)
merged = []
for start, end in spans:
if merged and start <= merged[-1][1]:
merged[-1] = (merged[-1][0], max(merged[-1][1], end))
else:
merged.append((start, end))
return merged
def process_video(video, person_events):
if not person_events:
update_video_status(video["id"], "fall_done")
return 0
cap = cv2.VideoCapture(str(PROJECT_ROOT / video["file_path"]))
if not cap.isOpened():
update_video_status(video["id"], "fall_done")
return 0
fps = video["fps"] or 15.0
width, height = video["width"], video["height"]
start_time = None
if video["start_time"]:
try:
start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S")
except ValueError:
pass
fall_count = 0
# landmarker 实例按需复用:VIDEO 模式要求时间戳在同一实例内严格递增,只有当下一个
# 事件的起始时间会比当前实例已处理到的时间更早(多人同时/交替出现导致事件时间重叠)
# 时才换新实例。之前"每个事件都新建一个实例"的写法在批量跑千级视频时会持续新建
# GPU/Metal 上下文,资源迟迟未完全释放,最终在某个看似无关的帧读取操作上引发底层崩溃
landmarker = _make_landmarker()
last_t = -1.0
try:
# 先合并所有track的时间区间再处理:同一个人有时会被拆成多个重叠的track_id,
# 按原始track各自跑一遍会导致同一次摔倒被重复检测、重复插入
for merged_start, merged_end in _merge_windows(person_events):
start_frame_t = int(merged_start * fps) / fps
if start_frame_t <= last_t:
landmarker.close()
landmarker = _make_landmarker()
states = _extract_pose_states(
landmarker, cap, fps, width, height,
merged_start, merged_end,
)
if states:
last_t = states[-1][0]
for start_sec, end_sec in _find_fall_windows(states):
timestamp_abs = (start_time + timedelta(seconds=start_sec)).isoformat() if start_time else None
insert_event({
"video_id": video["id"],
"event_type": "fall",
"label": None,
"start_offset_sec": start_sec,
"end_offset_sec": end_sec,
"timestamp_abs": timestamp_abs,
"confidence": None,
"track_id": None,
"thumbnail_path": None,
"extra_json": None,
"created_at": datetime.now().isoformat(),
})
fall_count += 1
except Exception as e:
# 兜底:单条视频处理异常不应中断整个批次,标记失败并继续下一条。
# 异常本身可能就是底层解码/GPU上下文已经出问题,close()也可能再报错,不能让它盖过原始异常
try:
landmarker.close()
except Exception:
pass
cap.release()
update_video_status(video["id"], "fall_failed", error_message=str(e))
return fall_count
landmarker.close()
cap.release()
update_video_status(video["id"], "fall_done")
return fall_count
def process_all(limit=None):
videos = get_all_videos(status="person_done")
if limit:
videos = videos[:limit]
events_by_video = {}
for e in get_events("person_presence"):
events_by_video.setdefault(e["video_id"], []).append(e)
processed = 0
total_falls = 0
for video in videos:
total_falls += process_video(video, events_by_video.get(video["id"], []))
processed += 1
return {"processed": processed, "total_falls": total_falls}
if __name__ == "__main__":
print("===== 开始检测摔倒事件 =====")
result = process_all()
print(f"→ 处理 {result['processed']} 个视频,检测到 {result['total_falls']} 次疑似摔倒")
print("===== 检测完成 =====")
+16
View File
@@ -0,0 +1,16 @@
#!/usr/bin/env python3
"""spawn 萤石 app 并注入脱壳脚本,保持挂载。"""
import frida, sys, time
dev = frida.get_usb_device(timeout=10)
pid = dev.spawn(["com.videogo"])
print(f"spawned pid={pid}", flush=True)
session = dev.attach(pid)
with open("analysis/unpin.js") as f:
src = f.read()
script = session.create_script(src)
script.on("message", lambda m, d: print("[msg]", m.get("payload", m), flush=True))
script.load()
dev.resume(pid)
print("resumed, 保持挂载 1800s,请在模拟器里手动操作", flush=True)
time.sleep(1800)
+10
View File
@@ -0,0 +1,10 @@
"""通用文件 MD5 计算:供入库查重(video_scanner)和重复检测报告(dedup_finder)共用。"""
import hashlib
def compute_md5(path, chunk_size=1024 * 1024):
h = hashlib.md5()
with open(path, "rb") as f:
while chunk := f.read(chunk_size):
h.update(chunk)
return h.hexdigest()
+145
View File
@@ -0,0 +1,145 @@
"""MOG2背景建模检测每个视频里"有活动"的时间窗口,写回 videos.motion_windows_json。
长视频不能逐帧跑重模型,所以先用轻量背景建模预筛出活动窗口,后续 person_tracker /
fall_detector 只在窗口内跑,空场景直接跳过。
之前用固定阈值帧差法(相邻两帧像素差),对"人物占画面小、动作幅度小"的场景
(如老人在客厅整理东西,移动缓慢、遮挡画面比例低)容易漏检。改用 MOG2
自适应背景建模:背景是随时间缓慢更新的统计模型而非"上一帧",前景掩码对
慢速/小幅度运动更敏感,且能自动适应光照的缓慢变化。
"""
import json
from datetime import datetime
import cv2
from analysis.config import (
MOTION_RESIZE_WIDTH,
MOTION_HISTORY,
MOTION_VAR_THRESHOLD,
MOTION_WARMUP_FRAMES,
MOTION_SHORT_CLIP_SEC,
MOTION_PIXEL_RATIO,
MOTION_MERGE_GAP_SEC,
MOTION_MIN_DURATION_SEC,
MOTION_SAMPLE_STRIDE,
PROJECT_ROOT,
)
from analysis.db import get_all_videos, update_motion_windows
def detect_motion_windows(file_path, duration_sec=None):
"""返回该视频里的活动时间窗口列表 [[start_sec, end_sec], ...]。
duration_sec 由调用方传入(video_scanner用ffprobe测得,比cv2的帧数估算更准);
不传时退化为用 cv2 的帧数/帧率估算,供独立调用/测试使用。
"""
cap = cv2.VideoCapture(str(file_path))
if not cap.isOpened():
return []
fps = cap.get(cv2.CAP_PROP_FPS) or 15.0
if duration_sec is None:
frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT)
duration_sec = frame_count / fps if frame_count > 0 else None
# 触发即录的极短告警片段,帧数不够 MOG2 预热收敛,直接整段视为活动窗口
if duration_sec is not None and duration_sec < MOTION_SHORT_CLIP_SEC:
cap.release()
return [[0.0, duration_sec]]
subtractor = cv2.createBackgroundSubtractorMOG2(
history=MOTION_HISTORY, varThreshold=MOTION_VAR_THRESHOLD, detectShadows=True
)
motion_flags = []
frame_idx = 0
sample_idx = 0
while True:
ret, frame = cap.read()
if not ret:
break
if frame_idx % MOTION_SAMPLE_STRIDE != 0:
frame_idx += 1
continue
h, w = frame.shape[:2]
scale = MOTION_RESIZE_WIDTH / w
small = cv2.resize(frame, (MOTION_RESIZE_WIDTH, max(1, int(h * scale))))
small = cv2.GaussianBlur(small, (5, 5), 0)
fg_mask = subtractor.apply(small)
# MOG2 用 127 标记阴影、255 标记前景,阴影不算真实活动,只统计255
foreground_ratio = (fg_mask == 255).sum() / fg_mask.size
if sample_idx >= MOTION_WARMUP_FRAMES:
motion_flags.append((frame_idx / fps, foreground_ratio > MOTION_PIXEL_RATIO))
frame_idx += 1
sample_idx += 1
cap.release()
return merge_motion_windows(motion_flags)
def merge_motion_windows(motion_flags):
windows = []
window_start = None
window_last_motion = None
for t, is_motion in motion_flags:
if is_motion:
if window_start is None:
window_start = t
window_last_motion = t
elif window_start is not None and (t - window_last_motion) > MOTION_MERGE_GAP_SEC:
windows.append([window_start, window_last_motion])
window_start = None
window_last_motion = None
if window_start is not None:
windows.append([window_start, window_last_motion])
return [w for w in windows if (w[1] - w[0]) >= MOTION_MIN_DURATION_SEC]
def process_all(limit=None):
videos = get_all_videos(status="pending")
if limit:
videos = videos[:limit]
processed = 0
no_motion = 0
failed = 0
for video in videos:
try:
file_path = PROJECT_ROOT / video["file_path"]
if not file_path.exists():
update_video_status(video["id"], "motion_failed", error_message="文件不存在")
failed += 1
processed += 1
continue
windows = detect_motion_windows(file_path, duration_sec=video["duration_sec"])
update_motion_windows(
video["id"],
json.dumps(windows),
status="motion_done",
analyzed_at=datetime.now().isoformat(),
)
processed += 1
if not windows:
no_motion += 1
except Exception as e:
update_video_status(video["id"], "motion_failed", error_message=str(e))
failed += 1
processed += 1
return {"processed": processed, "no_motion": no_motion, "failed": failed}
if __name__ == "__main__":
print("===== 开始检测活动窗口 =====")
result = process_all()
print(f"→ 处理 {result['processed']} 个视频,其中 {result['no_motion']} 个未检测到活动")
print("===== 检测完成 =====")
+86
View File
@@ -0,0 +1,86 @@
"""封装本地 Ollama 视觉模型调用:给定一帧图片,从限定词表中分类出一个活动标签。
只做限定词表分类,不做开放式生成——分类结果直接落地为 events.label,取值域
必须严格落在词表内。VLM 输出格式不总是稳定(可能带标点/解释性文字),
所以要做防御性解析兜底,解析不出合法标签就归为"其他",不让脏数据污染打标结果。
"""
import time
import cv2
import ollama
from analysis.config import ACTIVITY_LABELS, OLLAMA_MODEL
_FALLBACK_LABEL = "其他"
_MAX_RETRIES = 3
_RETRY_DELAY = 5.0 # 秒,Ollama 偶尔忙/模型加载中,退避重试
def _build_prompt():
labels = "、".join(ACTIVITY_LABELS)
return (
"这是家中监控摄像头拍到的一帧画面。请判断画面里的人此刻最可能在做什么,"
f"只能从下面这些词里选一个作为回答,不要输出任何其他文字或标点:{labels}"
)
def _parse_label(raw_text):
stripped = raw_text.strip()
if stripped in ACTIVITY_LABELS:
return stripped
for label in ACTIVITY_LABELS:
if label in raw_text:
return label
return _FALLBACK_LABEL
_MAX_FRAME_WIDTH = 640 # 摄像头原始分辨率最高2560x1440,直接送入会因视觉token过多超出上下文,
# 而"走动/进食/坐下"这类粗粒度分类不需要高分辨率细节,缩小后再送入
def _resize_for_vlm(frame_bgr):
height, width = frame_bgr.shape[:2]
if width <= _MAX_FRAME_WIDTH:
return frame_bgr
scale = _MAX_FRAME_WIDTH / width
return cv2.resize(frame_bgr, (_MAX_FRAME_WIDTH, int(height * scale)))
def check_ollama_health():
"""检查 Ollama 服务是否可用、目标模型是否已拉取。返回 (ok, message)。"""
try:
models = ollama.list()
model_names = [m.get("name", "") for m in models.get("models", [])]
if not any(OLLAMA_MODEL in name for name in model_names):
return False, f"模型 {OLLAMA_MODEL} 未拉取,请先运行: ollama pull {OLLAMA_MODEL}"
return True, "ok"
except Exception as e:
return False, f"Ollama 服务不可用: {e}"
def classify_frame(frame_bgr):
"""frame_bgr: OpenCV BGR格式的numpy帧。返回 (标签, 原始回复文本)。"""
ok, buf = cv2.imencode(".jpg", _resize_for_vlm(frame_bgr))
if not ok:
return _FALLBACK_LABEL, ""
last_error = None
for attempt in range(_MAX_RETRIES):
try:
response = ollama.chat(
model=OLLAMA_MODEL,
messages=[{
"role": "user",
"content": _build_prompt(),
"images": [buf.tobytes()],
}],
options={"num_ctx": 2048}, # 单帧分类不需要8192上下文,省显存提速
)
raw_text = response["message"]["content"].strip()
return _parse_label(raw_text), raw_text
except Exception as e:
last_error = e
if attempt < _MAX_RETRIES - 1:
time.sleep(_RETRY_DELAY)
# 全部重试失败,归为"其他"而不是抛异常中断整个批次
return _FALLBACK_LABEL, f"[ollama_error] {last_error}"
+158
View File
@@ -0,0 +1,158 @@
"""在 motion_filter 找出的活动窗口内跑 YOLO11n 检测 + ByteTrack 跟踪,
统计人物出现次数,写入 events 表(event_type='person_presence')。
一个 track(同一个人从出现到消失的连续跟踪片段)对应一条 person_presence
事件。只在活动窗口内跑,空场景(motion_windows_json 为空)直接跳过,不跑模型。
"""
import json
from datetime import datetime, timedelta
import cv2
from analysis.config import (
PERSON_MODEL_NAME,
PERSON_DEVICE,
PERSON_CONF_THRESHOLD,
PERSON_SAMPLE_STRIDE,
PERSON_MIN_TRACK_DURATION_SEC,
PROJECT_ROOT,
)
from analysis.db import get_all_videos, insert_event, update_video_status
_model = None
def _get_model():
"""YOLO模型只在真正用到时才加载(避免只跑无活动视频时也触发权重下载/加载)。"""
global _model
if _model is None:
from ultralytics import YOLO
_model = YOLO(PERSON_MODEL_NAME)
return _model
def _track_windows(file_path, windows, fps):
"""在给定的活动窗口内跑检测+跟踪,返回 [(track_id, start_sec, end_sec, max_conf), ...]。"""
model = _get_model()
cap = cv2.VideoCapture(str(file_path))
if not cap.isOpened():
return []
spans = {} # (window_index, track_id) -> [min_t, max_t, max_conf]
for window_index, (start_sec, end_sec) in enumerate(windows):
start_frame = int(start_sec * fps)
end_frame = int(end_sec * fps)
cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
frame_idx = start_frame
first_call = True
while frame_idx <= end_frame:
ret, frame = cap.read()
if not ret:
break
if (frame_idx - start_frame) % PERSON_SAMPLE_STRIDE == 0:
# persist=False 强制在每个窗口开头重置跟踪器状态,跟踪器内部的track_id计数器
# 也会跟着从1重新开始——不同窗口间同一个数字的track_id并不代表同一个人,
# 所以下面的 spans key 必须带上 window_index,否则会把不同窗口的片段错误合并
results = model.track(
frame, persist=not first_call, classes=[0], conf=PERSON_CONF_THRESHOLD,
tracker="bytetrack.yaml", device=PERSON_DEVICE, verbose=False,
)
first_call = False
boxes = results[0].boxes
if boxes is not None and boxes.id is not None:
t = frame_idx / fps
for track_id, conf in zip(boxes.id.tolist(), boxes.conf.tolist()):
key = (window_index, int(track_id))
if key not in spans:
spans[key] = [t, t, conf]
else:
spans[key][1] = t
spans[key][2] = max(spans[key][2], conf)
frame_idx += 1
cap.release()
return [
(track_id, start, end, conf)
for (_window_index, track_id), (start, end, conf) in spans.items()
if (end - start) >= PERSON_MIN_TRACK_DURATION_SEC
]
def process_video(video):
"""返回 (track_count, failed)。failed=True 表示该视频处理异常已标记 person_failed。"""
try:
windows = json.loads(video["motion_windows_json"] or "[]")
except (json.JSONDecodeError, TypeError):
windows = []
if not windows:
update_video_status(video["id"], "person_done")
return 0, False
file_path = PROJECT_ROOT / video["file_path"]
if not file_path.exists():
update_video_status(video["id"], "person_failed", error_message="文件不存在")
return 0, True
try:
fps = video["fps"] or 15.0
tracks = _track_windows(file_path, windows, fps)
except Exception as e:
update_video_status(video["id"], "person_failed", error_message=str(e))
return 0, True
start_time = None
if video["start_time"]:
try:
start_time = datetime.strptime(video["start_time"], "%Y-%m-%d %H:%M:%S")
except ValueError:
pass # 只精确到天的粗粒度值("YYYY-MM-DD"),无法算出绝对时间戳,timestamp_abs留空
for track_id, start_sec, end_sec, conf in tracks:
timestamp_abs = (start_time + timedelta(seconds=start_sec)).isoformat() if start_time else None
insert_event({
"video_id": video["id"],
"event_type": "person_presence",
"label": None,
"start_offset_sec": start_sec,
"end_offset_sec": end_sec,
"timestamp_abs": timestamp_abs,
"confidence": conf,
"track_id": track_id,
"thumbnail_path": None,
"extra_json": None,
"created_at": datetime.now().isoformat(),
})
update_video_status(video["id"], "person_done")
return len(tracks), False
def process_all(limit=None):
videos = get_all_videos(status="motion_done")
if limit:
videos = videos[:limit]
processed = 0
total_tracks = 0
failed = 0
for video in videos:
tracks, is_failed = process_video(video)
total_tracks += tracks
if is_failed:
failed += 1
processed += 1
return {"processed": processed, "total_tracks": total_tracks, "failed": failed}
if __name__ == "__main__":
print("===== 开始统计人物出现次数 =====")
result = process_all()
print(f"→ 处理 {result['processed']} 个视频,共识别出 {result['total_tracks']} 段人物出现")
print("===== 统计完成 =====")
Binary file not shown.
Binary file not shown.
+103
View File
@@ -0,0 +1,103 @@
"""用 ffmpeg 在每条 events 记录的 start_offset_sec 处抽一帧,生成缩略图。
供网页事件列表展示预览图,点击缩略图再跳转播放对应视频的完整时间点。
失败重试上限:单个事件连续失败超过 MAX_ATTEMPTS 次后,在 extra_json 里标记
thumbnail_failed,后续运行直接跳过,避免对损坏文件无限重试。
"""
import json
import subprocess
from analysis.config import PROJECT_ROOT, THUMBNAILS_DIR
from analysis.db import get_events, get_video_by_id, update_event_thumbnail, update_event_extra
MAX_ATTEMPTS = 3
def _extract_frame(video_path, offset_sec, output_path):
# -ss 放在 -i 之后用精确seek:放在 -i 之前的快速seek在临近视频结尾处
# 会因缺少解码所需的参考帧而失败(HEVC "Could not find ref with POC")。
# 录像多为几秒到几十秒的短片,精确seek的性能代价可忽略。
result = subprocess.run(
[
"ffmpeg", "-i", str(video_path), "-ss", str(offset_sec),
"-frames:v", "1", "-q:v", "3", "-y", str(output_path),
],
capture_output=True, timeout=30,
)
return result.returncode == 0 and output_path.exists()
def _get_attempts(event):
try:
extra = json.loads(event["extra_json"] or "{}")
return extra.get("thumbnail_attempts", 0)
except (json.JSONDecodeError, TypeError):
return 0
def _increment_attempts(event):
try:
extra = json.loads(event["extra_json"] or "{}")
except (json.JSONDecodeError, TypeError):
extra = {}
extra["thumbnail_attempts"] = extra.get("thumbnail_attempts", 0) + 1
if extra["thumbnail_attempts"] >= MAX_ATTEMPTS:
extra["thumbnail_failed"] = True
update_event_extra(event["id"], json.dumps(extra, ensure_ascii=False))
def process_all(limit=None):
THUMBNAILS_DIR.mkdir(exist_ok=True)
events = []
for e in get_events():
if e["thumbnail_path"]:
continue
try:
extra = json.loads(e["extra_json"] or "{}")
if extra.get("thumbnail_failed"):
continue
except (json.JSONDecodeError, TypeError):
pass
events.append(e)
if limit:
events = events[:limit]
video_cache = {}
processed = 0
failed = 0
for event in events:
video_id = event["video_id"]
if video_id not in video_cache:
video_cache[video_id] = get_video_by_id(video_id)
video = video_cache[video_id]
if video is None:
_increment_attempts(event)
failed += 1
continue
video_path = PROJECT_ROOT / video["file_path"]
if not video_path.exists():
_increment_attempts(event)
failed += 1
continue
output_path = THUMBNAILS_DIR / f"{event['id']}.jpg"
offset_sec = event["start_offset_sec"] or 0.0
if _extract_frame(video_path, offset_sec, output_path):
update_event_thumbnail(event["id"], str(output_path.relative_to(PROJECT_ROOT)))
processed += 1
else:
_increment_attempts(event)
failed += 1
return {"processed": processed, "failed": failed}
if __name__ == "__main__":
print("===== 开始生成缩略图 =====")
result = process_all()
print(f"→ 生成 {result['processed']} 张缩略图,失败 {result['failed']} 条")
print("===== 生成完成 =====")
+158
View File
@@ -0,0 +1,158 @@
"""从摄像头画面自带的OSD时间戳水印读取精确到秒的录制时间。
原理:画面固定位置烧录 "YYYY-MM-DD HH:MM:SS" 字样。由于点阵字体渲染时字符间
不保证有空白列(会粘连),不用连通域/空白列分割单个字符,而是先定位"日期组"
"时间组"两个文本块的整体边界(这两组之间的空格间隙远大于组内字符间隙),再用
已知的固定字符数(日期10个字符,时间8个字符)做等宽硬切割。
日期组、时间组分别独立做 Otsu 二值化(而非整行统一阈值):两组背景光照条件
可能不同,统一阈值会让偏暗的一组文字被过度清除。
数字识别用多模板 IOU(交并比)相似度匹配,而非 cv2.matchTemplate 的
TM_CCOEFF_NORMED —— 二值图像中大量近空白/全黑的 cell 两两比较时,归一化相关
系数会退化出虚假的高分。模板库按分辨率持久化在 analysis/templates/<分辨率>/
digit_templates.npz,收集自多个已知真值样本中同一数字在所有出现位置的实例。
"""
from datetime import datetime, timedelta
import cv2
import numpy as np
from analysis.config import PROJECT_ROOT
TEMPLATES_DIR = PROJECT_ROOT / "analysis" / "templates"
PATTERN = ("digit digit digit digit dash digit digit dash digit digit "
"digit digit colon digit digit colon digit digit").split()
# 每种分辨率的裁剪坐标 + 日期组/时间组边界(相对裁剪区域的局部坐标)+ cell 尺寸。
# 边界均由列投影法(找字符间隙)在多个样本上标定校正,不是拍脑袋的估计值。
RESOLUTION_PROFILES = {
(2560, 1440): {
"crop_y": (1372, 1402),
"crop_x": (1750, 2560),
"date_group": (108, 373), # 10字符: Y Y Y Y - M M - D D
"time_group": (407, 613), # 8字符: H H : M M : S S
"cell_size": (20, 30), # (width, height),传给 cv2.resize
},
(1280, 720): {
"crop_y": (682, 705),
"crop_x": (870, 1280),
"date_group": (60, 192),
"time_group": (208, 312),
"cell_size": (20, 30),
},
}
# 识别置信度低于此值时判定为不可信,调用方应回退到"只精确到天"的粗粒度时间
MIN_CONFIDENCE = 0.5
_template_cache = {}
def _load_templates(profile_key):
"""按分辨率加载模板库,进程内缓存一次。"""
if profile_key not in _template_cache:
npz_path = TEMPLATES_DIR / f"{profile_key[1]}p" / "digit_templates.npz"
data = np.load(npz_path)
_template_cache[profile_key] = {digit: data[digit] for digit in data.files}
return _template_cache[profile_key]
def _cells_from_group(gray, group, n, cell_size):
s, e = group
sub = gray[:, s:e]
_, thresh = cv2.threshold(sub, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
w = (e - s) / n
cells = []
for i in range(n):
x0 = int(round(i * w))
x1 = int(round((i + 1) * w))
cells.append(cv2.resize(thresh[:, x0:x1], cell_size, interpolation=cv2.INTER_AREA))
return cells
def _iou(a, b):
a = a > 0
b = b > 0
inter = np.logical_and(a, b).sum()
union = np.logical_or(a, b).sum()
return inter / union if union > 0 else 0.0
def _match_digit(cell, templates):
best_ch, best_score = None, -1
for digit, tmpls in templates.items():
for tmpl in tmpls:
score = _iou(cell, tmpl)
if score > best_score:
best_score, best_ch = score, digit
return best_ch, best_score
def _decode(gray, profile, templates):
cells = (_cells_from_group(gray, profile["date_group"], 10, profile["cell_size"])
+ _cells_from_group(gray, profile["time_group"], 8, profile["cell_size"]))
out = []
scores = []
for cell, kind in zip(cells, PATTERN):
if kind == "digit":
ch, score = _match_digit(cell, templates)
out.append(ch)
scores.append(score)
elif kind == "dash":
out.append("-")
elif kind == "colon":
out.append(":")
s = "".join(out)
# out 顺序 = 日期组10个cell(已含2个'-') + 时间组8个cell(已含2个':')
formatted = f"{s[0:10]} {s[10:18]}"
return formatted, (min(scores) if scores else 0.0)
def read_timestamp(file_path, capture_date=None):
"""读取视频首帧的OSD时间戳水印,返回 "YYYY-MM-DD HH:MM:SS" 字符串。
分辨率不在已标定范围内、读帧失败、识别置信度过低、或识别结果未通过合法性
校验时返回 None,调用方应保留原有"只精确到天"的粗粒度 start_time 作为兜底。
单纯依赖 IOU 置信度阈值并不可靠:抽样发现即使分数达标,仍会出现"分钟81"
这类不合法数值、或年份被识别错一位数(如2026错认成2006)但恰好凑成一个
合法日期的情况。因此额外做两层校验:
1. 用 datetime 解析,过滤掉月/日/时/分/秒超出合法范围的结果
2. 若调用方提供 capture_date(文件所在日期文件夹,格式YYYYMMDD),解码出的
日期必须与之相差在1天以内(凌晨录制可能跨天),否则判定为误识别
"""
cap = cv2.VideoCapture(str(file_path))
ok, frame = cap.read()
cap.release()
if not ok:
return None
height, width = frame.shape[:2]
profile = RESOLUTION_PROFILES.get((width, height))
if profile is None:
return None
y0, y1 = profile["crop_y"]
x0, x1 = profile["crop_x"]
gray = cv2.cvtColor(frame[y0:y1, x0:x1], cv2.COLOR_BGR2GRAY)
templates = _load_templates((width, height))
formatted, min_score = _decode(gray, profile, templates)
if min_score < MIN_CONFIDENCE:
return None
try:
parsed = datetime.strptime(formatted, "%Y-%m-%d %H:%M:%S")
except ValueError:
return None
if capture_date is not None:
folder_date = datetime.strptime(capture_date, "%Y%m%d")
if abs((parsed.date() - folder_date.date())) > timedelta(days=1):
return None
return formatted
+53
View File
@@ -0,0 +1,53 @@
/*
* 通用 Android SSL Pinning 绕过脚本(Frida)
* 覆盖常见实现:系统 TrustManagerImpl、OkHttp3 CertificatePinner、
* SSLContext 自定义 TrustManager、Conscrypt。
* 用途:抓包萤石云 app (com.videogo) 的 HTTPS 私有接口。
*/
setTimeout(function () {
Java.perform(function () {
console.log('[unpin] 开始注入');
// 1. 系统 TrustManagerImpl.verifyChain —— 直接放行
try {
var TMI = Java.use('com.android.org.conscrypt.TrustManagerImpl');
TMI.verifyChain.implementation = function (untrustedChain, trustAnchorChain, host, clientAuth, ocspData, tlsSctData) {
console.log('[unpin] TrustManagerImpl.verifyChain 放行 host=' + host);
return untrustedChain;
};
} catch (e) { console.log('[unpin] TrustManagerImpl 跳过: ' + e); }
// 2. OkHttp3 CertificatePinner.check —— 直接返回
try {
var CP = Java.use('okhttp3.CertificatePinner');
CP.check.overload('java.lang.String', 'java.util.List').implementation = function (a, b) {
console.log('[unpin] okhttp3 CertificatePinner.check 放行 ' + a);
return;
};
} catch (e) { console.log('[unpin] okhttp3 CertificatePinner 跳过: ' + e); }
// 3. 自定义 X509TrustManager —— 替换成全信任
try {
var X509TM = Java.use('javax.net.ssl.X509TrustManager');
var SSLContext = Java.use('javax.net.ssl.SSLContext');
var TrustManager = Java.registerClass({
name: 'com.ez.FakeTM',
implements: [X509TM],
methods: {
checkClientTrusted: function () {},
checkServerTrusted: function () {},
getAcceptedIssuers: function () { return []; }
}
});
var TMs = [TrustManager.$new()];
var initOverload = SSLContext.init.overload(
'[Ljavax.net.ssl.KeyManager;', '[Ljavax.net.ssl.TrustManager;', 'java.security.SecureRandom');
initOverload.implementation = function (km, tm, sr) {
console.log('[unpin] SSLContext.init 替换 TrustManager');
initOverload.call(this, km, TMs, sr);
};
} catch (e) { console.log('[unpin] SSLContext 跳过: ' + e); }
console.log('[unpin] 注入完成');
});
}, 0);
+167
View File
@@ -0,0 +1,167 @@
"""扫描 recordings/{设备号}/{YYYYMMDD}/*.mp4,登记进 videos 表。
start_time 只精确到"天":文件名时间戳实测证实记录的是下载/导出时刻而非真实录制
时刻(容器是 MPEG-PS,没有 creation_time 元数据可交叉验证),因此放弃追求绝对
精确的录制时间,start_time 直接取自文件所在日期文件夹(capture_date)。
automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题,会把同一段录像重复
下载成多份文件名不同的拷贝(详见 dedup_finder.py)。这里在入库时按 MD5 查重:
新文件如果和已入库的某个视频内容完全相同,标记为 duplicate 状态,不再进入
Phase2~5 的分析流程,避免继续产生虚高的事件统计。
"""
import json
import subprocess
from datetime import datetime
from analysis.config import RECORDINGS_DIR, PROJECT_ROOT
from analysis.db import (
insert_video, init_db, count_videos, video_exists, get_video_id_by_md5,
get_videos_missing_md5, update_video_md5,
)
from analysis.hashing import compute_md5
def backfill_md5():
"""给这个字段上线之前已经入库、还没有 file_md5 的老记录补算一次,
否则新文件查重时比对不到旧文件,查重形同虚设。"""
for video in get_videos_missing_md5():
file_path = PROJECT_ROOT / video["file_path"]
if not file_path.exists():
continue # 文件已被删除(例如清理过的重复文件),跳过
update_video_md5(video["id"], compute_md5(file_path))
def probe_video(file_path):
"""用 ffprobe 取时长/帧率/分辨率。文件损坏或探测失败时返回 None。"""
try:
out = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(file_path)],
capture_output=True, text=True, timeout=30,
)
data = json.loads(out.stdout)
except (subprocess.TimeoutExpired, json.JSONDecodeError, OSError):
return None
fmt = data.get("format")
if not fmt or "duration" not in fmt:
return None
video_stream = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), None)
if not video_stream:
return None
fps = None
if video_stream.get("r_frame_rate"):
num, _, den = video_stream["r_frame_rate"].partition("/")
if num and den and int(den) != 0:
fps = int(num) / int(den)
return {
"duration_sec": float(fmt["duration"]),
"fps": fps,
"width": video_stream.get("width"),
"height": video_stream.get("height"),
}
def capture_date_to_start_time(capture_date):
return f"{capture_date[0:4]}-{capture_date[4:6]}-{capture_date[6:8]}"
def scan_all():
"""扫描全部设备目录下的全部录像文件,登记入库。返回 {已处理, 新增损坏, 新增重复} 统计。"""
init_db()
backfill_md5()
processed = 0
corrupt = 0
duplicates = 0
if not RECORDINGS_DIR.exists():
return {"processed": 0, "corrupt": 0, "duplicates": 0}
for device_dir in sorted(RECORDINGS_DIR.iterdir()):
if not device_dir.is_dir():
continue
device_serial = device_dir.name
for date_dir in sorted(device_dir.iterdir()):
if not date_dir.is_dir():
continue
capture_date = date_dir.name
for file_path in sorted(date_dir.glob("*.mp4")):
rel_path = str(file_path.relative_to(PROJECT_ROOT))
if video_exists(rel_path):
continue # 已登记过,跳过重复的探测/哈希计算
file_size_bytes = file_path.stat().st_size
file_md5 = compute_md5(file_path)
duplicate_of = get_video_id_by_md5(file_md5)
if duplicate_of is not None:
# 与已入库视频内容完全相同,标记为 duplicate,不进入后续分析流程
video = {
"device_serial": device_serial,
"capture_date": capture_date,
"file_path": rel_path,
"file_name": file_path.name,
"file_size_bytes": file_size_bytes,
"duration_sec": None,
"fps": None,
"width": None,
"height": None,
"start_time": capture_date_to_start_time(capture_date),
"status": "duplicate",
"error_message": f"与 video_id={duplicate_of} 内容完全相同(MD5一致),跳过分析",
"file_md5": file_md5,
"duplicate_of": duplicate_of,
}
insert_video(video)
duplicates += 1
processed += 1
continue
probe = probe_video(file_path)
if probe is None:
status = "corrupt"
error_message = "ffprobe 探测失败或文件损坏(可能是 0 字节空文件)"
duration_sec = fps = width = height = None
corrupt += 1
else:
status = "pending"
error_message = None
duration_sec = probe["duration_sec"]
fps = probe["fps"]
width = probe["width"]
height = probe["height"]
video = {
"device_serial": device_serial,
"capture_date": capture_date,
"file_path": rel_path,
"file_name": file_path.name,
"file_size_bytes": file_size_bytes,
"duration_sec": duration_sec,
"fps": fps,
"width": width,
"height": height,
"start_time": capture_date_to_start_time(capture_date),
"status": status,
"error_message": error_message,
"file_md5": file_md5,
"duplicate_of": None,
}
insert_video(video)
processed += 1
return {"processed": processed, "corrupt": corrupt, "duplicates": duplicates}
if __name__ == "__main__":
print("===== 开始扫描录像文件 =====")
result = scan_all()
print(f"→ 本次处理 {result['processed']} 个文件,其中损坏/空文件 {result['corrupt']} 个,"
f"重复文件 {result['duplicates']} 个(已跳过分析)")
print(f"→ 数据库当前共登记 {count_videos()} 条记录")
print("===== 扫描完成 =====")
+72
View File
@@ -0,0 +1,72 @@
"""本地视频分析主入口:串联 扫描→时间戳回填→活动窗口→人物统计→摔倒检测→活动打标→缩略图 全流程。
手动/事后触发,独立于 automate.py 的下载流程,不影响现有下载逻辑。
流水线按 videos.status 状态机推进,重复运行是幂等的:每个阶段只处理"上一阶段
已完成、本阶段还没跑过"的视频,天然支持"每天下载完后跑一次,只处理当天新增"。
"""
import argparse
from analysis import activity_tagger, backfill_start_time, fall_detector, motion_filter, person_tracker, thumbnail, video_scanner
from analysis.db import get_all_videos, update_video_status
STAGES = ["scan", "backfill", "motion", "person", "fall", "activity", "thumbnail"]
# 每个阶段完成后视频进入的状态,以及 --force 时应该回退到的"上一阶段完成"状态
_STAGE_STATUS = {
"motion": ("pending", "motion_done"),
"person": ("motion_done", "person_done"),
"fall": ("person_done", "fall_done"),
"activity": ("fall_done", "activity_done"),
}
def _force_reset(stage):
"""--force 时,把已经跑过该阶段的视频状态回退,让流水线重新处理一遍。"""
if stage not in _STAGE_STATUS:
return
reset_to, done_status = _STAGE_STATUS[stage]
for video in get_all_videos(status=done_status):
update_video_status(video["id"], reset_to)
def run_stage(stage, limit=None, force=False):
if force:
_force_reset(stage)
if stage == "scan":
result = video_scanner.scan_all()
print(f"[scan] 处理 {result['processed']} 个文件,损坏 {result['corrupt']} 个,重复 {result['duplicates']} 个")
elif stage == "backfill":
result = backfill_start_time.process_all(limit=limit)
print(f"[backfill] 处理 {result['total']} 条,成功回填精确时间 {result['updated']} 条,识别失败保留粗粒度 {result['fallback']} 条")
elif stage == "motion":
result = motion_filter.process_all(limit=limit)
print(f"[motion] 处理 {result['processed']} 个视频,{result['no_motion']} 个无活动,失败 {result.get('failed', 0)} 个")
elif stage == "person":
result = person_tracker.process_all(limit=limit)
print(f"[person] 处理 {result['processed']} 个视频,共 {result['total_tracks']} 段人物出现,失败 {result.get('failed', 0)} 个")
elif stage == "fall":
result = fall_detector.process_all(limit=limit)
print(f"[fall] 处理 {result['processed']} 个视频,检测到 {result['total_falls']} 次疑似摔倒")
elif stage == "activity":
result = activity_tagger.process_all(limit=limit)
print(f"[activity] 处理 {result['processed']} 个视频,标注 {result['total_segments']} 段活动,失败 {result.get('failed', 0)} 个")
elif stage == "thumbnail":
result = thumbnail.process_all(limit=limit)
print(f"[thumbnail] 生成 {result['processed']} 张缩略图,失败 {result['failed']} 条")
def main():
parser = argparse.ArgumentParser(description="本地视频分析主流程")
parser.add_argument("--only-stage", choices=STAGES, help="只跑指定阶段,不跑完整流程")
parser.add_argument("--force", action="store_true", help="强制重新跑该阶段(把已完成的视频状态回退)")
parser.add_argument("--limit", type=int, default=None, help="限制本次处理的视频数量,用于小批量测试")
args = parser.parse_args()
stages = [args.only_stage] if args.only_stage else STAGES
for stage in stages:
run_stage(stage, limit=args.limit, force=args.force)
if __name__ == "__main__":
main()
+84 -19
View File
@@ -7,8 +7,6 @@ import subprocess, time, os, shutil, struct, zlib
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
PYTHON = "/usr/local/bin/python3.12"
ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb") ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb")
DEVICE = "emulator-5554" DEVICE = "emulator-5554"
OUTPUT_DIR = Path("./recordings") OUTPUT_DIR = Path("./recordings")
@@ -435,8 +433,9 @@ def tap_and_check_date_changed(arrow_coord, timeout=12, interval=1.0):
return False return False
def navigate_to_yesterday_batch(): def navigate_to_yesterday_batch(days_back=1):
"""导航到昨天的 SD 卡批量下载模式。 """导航到"今天往前第 days_back 天"的 SD 卡批量下载模式。
days_back=0 是今天,1 是昨天,2 是前天……(默认 1 保持原行为)。
关键节点用 tap_text 通过 UI dump 定位,避免坐标错位时误触首页底部 关键节点用 tap_text 通过 UI dump 定位,避免坐标错位时误触首页底部
Tab(萤石/对话/商城/社区/我的,"商城"中心 x=540 跟录像 Tab 撞车)。 Tab(萤石/对话/商城/社区/我的,"商城"中心 x=540 跟录像 Tab 撞车)。
""" """
@@ -485,11 +484,12 @@ def navigate_to_yesterday_batch():
else: else:
print(" ⚠ 连续 8 次翻页仍在变化,按当前位置继续") print(" ⚠ 连续 8 次翻页仍在变化,按当前位置继续")
print("→ 从今日退一天到昨天") print(f"→ 从今日往前退 {days_back} 天")
for step in range(days_back):
if not tap_and_check_date_changed(COORD["prev_day_arrow"]): if not tap_and_check_date_changed(COORD["prev_day_arrow"]):
print(" ! 首次未检测到日期变化,重试一次") print(f" ! 第 {step+1} 次点'上一天'未检测到日期变化,重试一次")
if not tap_and_check_date_changed(COORD["prev_day_arrow"]): if not tap_and_check_date_changed(COORD["prev_day_arrow"]):
raise RuntimeError("点击'上一天'箭头后日期未变化,无法切换到昨天") raise RuntimeError(f"点击'上一天'箭头后日期未变化,无法退到第 {days_back} 天")
# 等日期切换后的录像列表稳定加载 # 等日期切换后的录像列表稳定加载
time.sleep(3) time.sleep(3)
@@ -777,15 +777,24 @@ def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_r
def _median(vals): def _median(vals):
return sorted(vals)[len(vals) // 2] if vals else -1 return sorted(vals)[len(vals) // 2] if vals else -1
scrolled = not (pre_rows and post_rows def _moved(a, b):
and abs(_median(pre_rows) - _median(post_rows)) < 40) return bool(a and b and abs(_median(a) - _median(b)) >= 40)
if not scrolled: if not _moved(pre_rows, post_rows):
# 到底了(滚不动)。此时当前屏可能还有被"每轮只处理前 3 行"限制漏下的 # 疑似到底,但一次没滚动也可能是偶发滚动手势失效 / 下载中大文件卡顿吸收了
# 底部未下载卡片(它们本该靠继续滚动移到前 3 行处理,但到底滚不动了)。 # 手势(实测 26、27 号就因单次误判、提前结束、漏下几百个)。再滚一次确认:
# 清扫:对当前屏【所有】检出行判打勾,分批(每批≤9 不超并发)点未下载的、 # 只有连续两次都没推进才判真到底。
time.sleep(2)
adb("shell", "input", "swipe", "540", "1820", "540", "1414", "400")
time.sleep(1.5)
post2_rows = detect_card_rows(screenshot("/tmp/_postscroll2.png"))
if _moved(pre_rows, post2_rows):
print(f" 轮 {r+1}: 首次滚动疑似卡顿未生效,二次滚动已推进,继续")
continue
# 连续两次滚动行位置都未变,确认到底。此时当前屏可能还有被"每轮只处理前
# 3 行"限制漏下的底部未下载卡片,清扫:对当前屏所有检出行分批点未下载的、
# 等打勾,直到当前屏没有未下载卡片,再结束循环。 # 等打勾,直到当前屏没有未下载卡片,再结束循环。
print(f" 轮 {r+1}: 滚动后行位置未变,已到达列表底部,清扫当前屏剩余未下载卡片") print(f" 轮 {r+1}: 连续两次滚动都未推进,确认到达列表底部,清扫当前屏剩余")
for sweep in range(6): # 最多 6 批,防意外死循环 for sweep in range(6): # 最多 6 批,防意外死循环
ensure_not_anr() ensure_not_anr()
time.sleep(THUMB_LOAD_WAIT) time.sleep(THUMB_LOAD_WAIT)
@@ -860,11 +869,14 @@ def wait_for_downloads(baseline_files, timeout=600):
return list(all_files - baseline_files) return list(all_files - baseline_files)
def pull_files(remote_files): def pull_files(remote_files, date_str=None):
"""把下载好的文件 adb pull 到本地""" """把下载好的文件 adb pull 到本地。
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") date_str 为录像日期(YYYYMMDD),归档到 recordings/BH5455783/{date_str}/。
# 本地目录按录像日期(昨天)命名,方便归档 不传则默认昨天(保持原行为)。"""
local_dir = OUTPUT_DIR / "BH5455783" / yesterday if date_str is None:
date_str = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
# 本地目录按录像日期命名,方便归档
local_dir = OUTPUT_DIR / "BH5455783" / date_str
local_dir.mkdir(parents=True, exist_ok=True) local_dir.mkdir(parents=True, exist_ok=True)
print(f"→ 拉取 {len(remote_files)} 个文件到 {local_dir}") print(f"→ 拉取 {len(remote_files)} 个文件到 {local_dir}")
@@ -922,5 +934,58 @@ def sync():
print("===== 同步完成 =====") print("===== 同步完成 =====")
def clear_device_recordings():
"""清空模拟器上萤石下载的录像文件 + 相册里的录像副本,释放存储空间。
只删录像(*_BH* 下载文件、YS 相册 *.mp4),不碰 App 的日志/播放器资源等其它文件。
多天连续下载时必须每天清,否则录像(含相册双份)会很快撑满模拟器 10G 存储。"""
today = datetime.now().strftime("%Y%m%d")
remote_dir = f"/sdcard/Android/data/{PKG}/files/{today}"
adb("shell", f"find {remote_dir} -name '*_BH*' -type f -delete")
adb("shell", "find /sdcard/DCIM/YS -iname '*.mp4' -type f -delete")
def sync_recent_days(n_days=7):
"""下载最近 n_days 天的录像(今天、昨天……往前 n_days-1 天),按录像日期归档。
每天独立重启 App 导航(比"退出批量下载页再切日期"稳妥)。关键:每天下载【前】
先清空模拟器上的录像+相册,让这一天从零开始——这样:①模拟器不会被多天累积撑满
(10G 存储,含相册双份,几天就满);②下完后设备上就是这一天的全部录像,直接
全量拉取到 recordings/{录像日期}/,不用再靠前后差集区分。拉完再清一次即时释放。
某天若无录像(点下载弹不出菜单)则跳过。
"""
print(f"===== 开始下载最近 {n_days} 天录像 =====")
for d in range(n_days):
rec_date = (datetime.now() - timedelta(days=d)).strftime("%Y%m%d")
print(f"\n===== [{d+1}/{n_days}] 录像日期 {rec_date}(今天往前第 {d} 天)=====")
try:
# 清空本地该天旧目录:重下的文件名是新的下载时刻,和旧文件名不同,
# 不清会产生"内容相同、文件名不同"的重复副本。SD 卡是真源,重下可恢复。
day_dir = OUTPUT_DIR / "BH5455783" / rec_date
if day_dir.exists():
shutil.rmtree(day_dir)
print(f" 已清空本地旧目录 {day_dir}(覆盖式重下)")
launch_app()
clear_device_recordings() # 下载前清空,保证有空间、且设备只剩这天的
navigate_to_yesterday_batch(days_back=d)
tap_all_thumbnails(set())
wait_for_downloads(set())
day_files = get_all_remote_files() # 这一天下载的全部录像
if day_files:
pull_files(list(day_files), date_str=rec_date)
else:
print(f"→ {rec_date} 没有下载到文件")
clear_device_recordings() # 拉完即时清空,释放空间给下一天
except Exception as e:
print(f"⚠ {rec_date} 处理失败,跳过:{e}")
continue
print(f"\n===== 最近 {n_days} 天下载完成 =====")
if __name__ == "__main__": if __name__ == "__main__":
import sys
# 用法:python automate.py → 只下昨天(原行为)
# python automate.py 7 → 下最近 7 天
if len(sys.argv) > 1:
sync_recent_days(int(sys.argv[1]))
else:
sync() sync()
+70
View File
@@ -0,0 +1,70 @@
absl-py==2.5.0
annotated-types==0.8.0
anyio==4.14.2
blinker==1.9.0
certifi==2026.7.22
cffi==2.1.0
charset-normalizer==3.4.9
click==8.4.2
colorama==0.4.6
contourpy==1.3.3
cycler==0.12.1
filelock==3.32.0
Flask==3.1.3
flatbuffers==25.12.19
fonttools==4.63.0
frida==16.6.6
frida-tools==12.5.1
fsspec==2026.6.0
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
idna==3.18
itsdangerous==2.2.0
Jinja2==3.1.6
kiwisolver==1.5.0
lap==0.5.13
MarkupSafe==3.0.3
matplotlib==3.11.1
mediapipe==0.10.35
mpmath==1.3.0
networkx==3.6.1
numpy==1.26.4
nvidia-ml-py==13.610.43
ollama==0.6.2
opencv-contrib-python==4.11.0.86
opencv-python==4.10.0.84
packaging==26.2
paho-mqtt==2.1.0
pillow==12.3.0
polars==1.43.0
polars-runtime-32==1.43.0
prompt_toolkit==3.0.52
psutil==7.2.2
pycparser==3.0
pycryptodome==3.23.0
pydantic==2.13.4
pydantic_core==2.46.4
pyezvizapi==1.0.5.0
Pygments==2.20.0
pyparsing==3.3.2
pytesseract==0.3.13
python-dateutil==2.9.0.post0
PyYAML==6.0.3
requests==2.34.2
setuptools==83.0.0
six==1.17.0
sounddevice==0.5.5
sympy==1.14.0
torch==2.13.0
torchvision==0.28.0
tqdm==4.69.1
typing-inspection==0.4.2
typing_extensions==4.16.0
ultralytics==8.4.105
ultralytics-thop==2.0.20
urllib3==2.7.0
wcwidth==0.8.2
websockets==13.1
Werkzeug==3.1.8
xmltodict==1.0.4
+10
View File
@@ -0,0 +1,10 @@
opencv-python==4.10.0.84
numpy>=1.26,<2.0
ultralytics>=8.3
lap>=0.5.12
mediapipe>=0.10.14
ollama>=0.4
pytesseract>=0.3
Flask>=3.0
Werkzeug>=3.0
tqdm>=4.66
+8 -5
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
萤石CB60 SD卡录像每日同步脚本 [已废弃,保留作为备选] 萤石CB60 SD卡录像每日同步脚本(开放平台 API 方案)
每天凌晨运行,下载前24小时内的所有SD卡录像到本地目录。
当前主力下载方案是 automate.py(ADB 自动化模拟器内萤石 App 批量下载)。
本脚本走萤石开放平台 API(open.ys7.com),需要先在开放平台关闭设备码流加密,
且 FLV 回放流拼接的录像不含精确时间戳,因此被 ADB 方案取代。保留仅作备选参考。
""" """
import json, os, re, subprocess, logging import json, os, re, shutil, subprocess, logging
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
import requests import requests
@@ -15,8 +18,8 @@ OUTPUT_DIR = Path(OUTPUT_DIR)
TOKEN_CACHE = Path(TOKEN_CACHE) TOKEN_CACHE = Path(TOKEN_CACHE)
LOG_FILE = Path("./sync.log") LOG_FILE = Path("./sync.log")
# NAS上部署时改为 "ffmpeg"(Docker镜像内) # ffmpeg 从 PATH 查找;NAS/Docker 部署时确保 ffmpeg 在 PATH 里即可,无需改代码
FFMPEG = "/opt/homebrew/bin/ffmpeg" FFMPEG = shutil.which("ffmpeg") or "ffmpeg"
BASE = "https://open.ys7.com/api/lapp" BASE = "https://open.ys7.com/api/lapp"
+106
View File
@@ -0,0 +1,106 @@
"""本地事件展示网页:纯只读查询 analysis.db + 播放原始录像文件,
不写入任何数据,可以在 analyze.py 跑到一半时随时启动查看已完成的部分。
"""
from flask import Flask, jsonify, render_template, request, send_file
from analysis.config import PROJECT_ROOT
from analysis.db import get_connection
from webapp import transcode_cache
app = Flask(__name__)
EVENT_TYPES = ["person_presence", "fall", "activity"]
@app.route("/")
def index():
return render_template("index.html")
@app.route("/api/events")
def api_events():
event_type = request.args.get("event_type")
capture_date = request.args.get("date") # YYYYMMDD
query = """
SELECT e.id, e.video_id, e.event_type, e.label, e.start_offset_sec,
e.end_offset_sec, e.timestamp_abs, e.confidence, e.thumbnail_path,
v.capture_date, v.file_name
FROM events e JOIN videos v ON v.id = e.video_id
WHERE 1=1
"""
params = []
if event_type in EVENT_TYPES:
query += " AND e.event_type = ?"
params.append(event_type)
if capture_date:
query += " AND v.capture_date = ?"
params.append(capture_date)
query += " ORDER BY e.timestamp_abs DESC, e.id DESC LIMIT 500"
with get_connection() as conn:
rows = conn.execute(query, params).fetchall()
events = [
{
"id": r["id"],
"video_id": r["video_id"],
"event_type": r["event_type"],
"label": r["label"],
"start_offset_sec": r["start_offset_sec"],
"end_offset_sec": r["end_offset_sec"],
"timestamp_abs": r["timestamp_abs"],
"confidence": r["confidence"],
"capture_date": r["capture_date"],
"file_name": r["file_name"],
"thumbnail_url": f"/media/thumb/{r['id']}" if r["thumbnail_path"] else None,
}
for r in rows
]
return jsonify(events)
@app.route("/api/stats")
def api_stats():
with get_connection() as conn:
rows = conn.execute(
"SELECT event_type, COUNT(*) AS c FROM events GROUP BY event_type"
).fetchall()
return jsonify({r["event_type"]: r["c"] for r in rows})
@app.route("/media/<int:video_id>/stream")
def media_stream(video_id):
with get_connection() as conn:
row = conn.execute("SELECT file_path FROM videos WHERE id = ?", (video_id,)).fetchone()
if row is None:
return "视频不存在", 404
try:
play_path = transcode_cache.ensure_transcoded(video_id, PROJECT_ROOT / row["file_path"])
except RuntimeError as e:
return f"转码失败:{e}", 500
return send_file(play_path, conditional=True)
@app.route("/api/cache/stats")
def api_cache_stats():
return jsonify(transcode_cache.cache_stats())
@app.route("/api/cache/clear", methods=["POST"])
def api_cache_clear():
return jsonify(transcode_cache.clear_cache())
@app.route("/media/thumb/<int:event_id>")
def media_thumb(event_id):
with get_connection() as conn:
row = conn.execute("SELECT thumbnail_path FROM events WHERE id = ?", (event_id,)).fetchone()
if row is None or not row["thumbnail_path"]:
return "缩略图不存在", 404
return send_file(PROJECT_ROOT / row["thumbnail_path"], conditional=True)
if __name__ == "__main__":
app.run(host="127.0.0.1", port=5050, debug=False)
+108
View File
@@ -0,0 +1,108 @@
const eventTypeFilter = document.getElementById("event-type-filter");
const dateFilter = document.getElementById("date-filter");
const eventList = document.getElementById("event-list");
const player = document.getElementById("player");
const playerInfo = document.getElementById("player-info");
const statsEl = document.getElementById("stats");
const cacheInfoEl = document.getElementById("cache-info");
const clearCacheBtn = document.getElementById("clear-cache-btn");
const EVENT_TYPE_LABELS = {
person_presence: "人物出现",
fall: "摔倒",
activity: "活动",
};
function formatEventLabel(event) {
const typeName = EVENT_TYPE_LABELS[event.event_type] || event.event_type;
return event.label ? `${typeName} · ${event.label}` : typeName;
}
async function loadStats() {
const res = await fetch("/api/stats");
const stats = await res.json();
const parts = Object.entries(stats).map(
([type, count]) => `${EVENT_TYPE_LABELS[type] || type}: ${count}`
);
statsEl.textContent = parts.join(" | ");
}
function formatBytes(bytes) {
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`;
if (bytes < 1024 * 1024 * 1024) return `${(bytes / 1024 / 1024).toFixed(1)} MB`;
return `${(bytes / 1024 / 1024 / 1024).toFixed(2)} GB`;
}
async function loadCacheStats() {
const res = await fetch("/api/cache/stats");
const stats = await res.json();
cacheInfoEl.textContent = `转码缓存:${stats.file_count} 个文件,共占用 ${formatBytes(stats.total_size_bytes)}`;
}
clearCacheBtn.addEventListener("click", async () => {
if (!confirm("确认清空所有转码缓存吗?清空后再次播放这些视频需要重新转码。")) return;
const res = await fetch("/api/cache/clear", { method: "POST" });
const result = await res.json();
alert(`已清理 ${result.deleted} 个文件,释放 ${formatBytes(result.freed_bytes)} 空间`);
loadCacheStats();
});
async function loadEvents() {
const params = new URLSearchParams();
if (eventTypeFilter.value) params.set("event_type", eventTypeFilter.value);
if (dateFilter.value) params.set("date", dateFilter.value.replaceAll("-", ""));
const res = await fetch(`/api/events?${params.toString()}`);
const events = await res.json();
renderEventList(events);
}
function renderEventList(events) {
eventList.innerHTML = "";
for (const event of events) {
const item = document.createElement("div");
item.className = "event-item";
const thumb = document.createElement("img");
thumb.className = "event-thumb";
thumb.alt = formatEventLabel(event);
if (event.thumbnail_url) thumb.src = event.thumbnail_url;
const info = document.createElement("div");
info.className = "event-info";
info.innerHTML = `
<div class="event-type">${formatEventLabel(event)}</div>
<div class="event-time">${event.timestamp_abs || event.capture_date}</div>
`;
item.appendChild(thumb);
item.appendChild(info);
item.addEventListener("click", () => playEvent(event));
eventList.appendChild(item);
}
}
function playEvent(event) {
const isSameVideo = player.dataset.videoId === String(event.video_id);
player.dataset.videoId = event.video_id;
const seekAndPlay = () => {
player.currentTime = event.start_offset_sec || 0;
player.play();
};
if (isSameVideo) {
seekAndPlay();
} else {
player.src = `/media/${event.video_id}/stream`;
player.addEventListener("loadedmetadata", () => { seekAndPlay(); loadCacheStats(); }, { once: true });
}
playerInfo.textContent = `${event.file_name} · ${formatEventLabel(event)}`;
}
eventTypeFilter.addEventListener("change", loadEvents);
dateFilter.addEventListener("change", loadEvents);
loadStats();
loadCacheStats();
loadEvents();
+96
View File
@@ -0,0 +1,96 @@
body {
font-family: -apple-system, "PingFang SC", sans-serif;
margin: 20px;
color: #222;
}
.filters {
margin-bottom: 16px;
display: flex;
gap: 16px;
align-items: center;
}
#stats {
color: #666;
font-size: 0.9em;
}
.cache-panel {
margin-bottom: 16px;
display: flex;
gap: 12px;
align-items: center;
color: #666;
font-size: 0.9em;
}
.cache-panel button {
cursor: pointer;
}
.main {
display: flex;
gap: 20px;
}
.player-panel {
flex: 1;
}
#player {
width: 100%;
max-width: 640px;
background: #000;
}
#player-info {
margin-top: 8px;
color: #555;
font-size: 0.9em;
}
.event-list {
flex: 1;
max-height: 80vh;
overflow-y: auto;
display: flex;
flex-direction: column;
gap: 8px;
}
.event-item {
display: flex;
gap: 10px;
cursor: pointer;
padding: 6px;
border-radius: 6px;
}
.event-item:hover {
background: #f0f0f0;
}
.event-thumb {
width: 120px;
height: 68px;
object-fit: cover;
background: #ddd;
border-radius: 4px;
flex-shrink: 0;
}
.event-info {
display: flex;
flex-direction: column;
justify-content: center;
}
.event-type {
font-weight: 600;
}
.event-time {
color: #666;
font-size: 0.9em;
}
+41
View File
@@ -0,0 +1,41 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>录像事件回放</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<h1>录像事件回放</h1>
<div class="filters">
<label>事件类型
<select id="event-type-filter">
<option value="">全部</option>
<option value="person_presence">人物出现</option>
<option value="fall">摔倒</option>
<option value="activity">活动</option>
</select>
</label>
<label>日期
<input type="date" id="date-filter">
</label>
<span id="stats"></span>
</div>
<div class="cache-panel">
<span id="cache-info"></span>
<button id="clear-cache-btn">清空转码缓存</button>
</div>
<div class="main">
<div class="player-panel">
<video id="player" controls></video>
<div id="player-info"></div>
</div>
<div id="event-list" class="event-list"></div>
</div>
<script src="{{ url_for('static', filename='app.js') }}"></script>
</body>
</html>
+96
View File
@@ -0,0 +1,96 @@
"""网页播放的按需转码缓存:源文件mp4容器HEVC轨道sample entry编码标签缺失,
浏览器<video>标签无法识别,第一次点击播放时用ffmpeg(VideoToolbox硬件加速)转码成
标准H264并缓存到 TRANSCODED_DIR,之后复用缓存不重复转码;不预先转码全部录像。
"""
import fcntl
import subprocess
from pathlib import Path
from analysis.config import TRANSCODED_DIR
def cached_path(video_id):
return TRANSCODED_DIR / f"{video_id}.mp4"
def _lock_path(video_id):
return TRANSCODED_DIR / f"{video_id}.lock"
def ensure_transcoded(video_id, source_path):
"""返回可直接播放的文件路径。已有缓存直接复用,否则先转码再缓存。
用 fcntl 文件锁防止两个请求同时转码同一个视频:先到的转码,后到的等锁释放后
直接读缓存。锁文件在转码完成后删除。
"""
output_path = cached_path(video_id)
if output_path.exists():
return output_path
TRANSCODED_DIR.mkdir(exist_ok=True)
lock_file = _lock_path(video_id)
lock_fd = open(lock_file, "w")
try:
# 阻塞等待锁,最多等 10 分钟(转码大文件可能慢)
fcntl.flock(lock_fd, fcntl.LOCK_EX)
# 拿到锁后再检查一次:可能前面的进程已经转码完成
if output_path.exists():
return output_path
tmp_path = output_path.with_suffix(".mp4.tmp")
result = subprocess.run(
[
"ffmpeg", "-i", str(source_path),
"-c:v", "h264_videotoolbox", "-b:v", "1500k",
"-c:a", "aac",
"-movflags", "+faststart",
"-f", "mp4", # 输出文件名带.tmp后缀,ffmpeg无法从扩展名猜出格式,需显式指定
"-y", str(tmp_path),
],
capture_output=True, timeout=600,
)
if result.returncode != 0 or not tmp_path.exists():
tmp_path.unlink(missing_ok=True)
raise RuntimeError(result.stderr.decode("utf-8", errors="ignore"))
tmp_path.rename(output_path) # 转码完成后原子改名,避免并发请求读到半成品文件
return output_path
finally:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
lock_file.unlink(missing_ok=True)
def cache_stats():
if not TRANSCODED_DIR.exists():
return {"file_count": 0, "total_size_bytes": 0, "files": []}
files = []
total = 0
for f in TRANSCODED_DIR.glob("*.mp4"):
size = f.stat().st_size
total += size
files.append({"video_id": int(f.stem), "size_bytes": size})
return {"file_count": len(files), "total_size_bytes": total, "files": files}
def clear_cache():
if not TRANSCODED_DIR.exists():
return {"deleted": 0, "freed_bytes": 0}
deleted = 0
freed = 0
for f in TRANSCODED_DIR.glob("*.mp4"):
freed += f.stat().st_size
f.unlink()
deleted += 1
# 清理残留的 .tmp 和 .lock 文件
for f in TRANSCODED_DIR.glob("*.tmp"):
f.unlink(missing_ok=True)
for f in TRANSCODED_DIR.glob("*.lock"):
f.unlink(missing_ok=True)
return {"deleted": deleted, "freed_bytes": freed}