feat: 萤石 CB60 录像每日同步初始版本
通过 ADB 自动化模拟器中的萤石 App，每天把 SD 卡里昨天的录像批量 下载到本地。云 API 方案 (sync.py/probe.py) 是备选，需先在萤石开放 平台关闭设备码流加密。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
yuming
+22
@@ -0,0 +1,22 @@
|
||||
# 敏感配置和缓存
|
||||
config.py
|
||||
token_cache.json
|
||||
sync.log
|
||||
|
||||
# 下载的录像（每天几百 MB）
|
||||
recordings/
|
||||
|
||||
# Python
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.venv/
|
||||
*.egg-info/
|
||||
|
||||
# IDE / 系统
|
||||
.DS_Store
|
||||
.idea/
|
||||
.vscode/
|
||||
|
||||
# 临时
|
||||
*.tmp
|
||||
/tmp/
|
||||
+488
@@ -0,0 +1,488 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
萤石录像自动下载脚本
|
||||
通过 ADB 自动化模拟器中的萤石 App，下载前一天的所有报警录像。
|
||||
"""
|
||||
import subprocess, time, os, shutil, struct, zlib
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
PYTHON = "/usr/local/bin/python3.12"
|
||||
|
||||
ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb")
|
||||
DEVICE = "emulator-5554"
|
||||
OUTPUT_DIR = Path("./recordings")
|
||||
PKG = "com.videogo"
|
||||
|
||||
# 屏幕坐标（Pixel 6 模拟器 1080x2400，2026-06-15 实测校准）
|
||||
COORD = {
|
||||
"home_camera_card": (540, 500), # 首页摄像头卡片
|
||||
"live_video_btn": (270, 1340), # 摄像头详情页-实时视频按钮
|
||||
"record_tab": (540, 2290), # 底部录像 Tab
|
||||
"prev_day_arrow": (80, 1130), # < 昨天箭头
|
||||
"download_btn": (426, 880), # 录像页工具栏-下载按钮
|
||||
"batch_download": (540, 1904), # 弹出菜单"批量下载"（UIAutomator dump 实测）
|
||||
"reload_btn": (540, 1960), # 加载失败页的"重新加载"按钮
|
||||
}
|
||||
|
||||
# 批量下载界面缩略图布局（UIAutomator dump 实测）
|
||||
# 3 列 x 中心：201, 540, 878
|
||||
# 行高 203 像素；y 中心从 1240 开始
|
||||
THUMB_X = [201, 540, 878]
|
||||
THUMB_ROW_H = 203
|
||||
# 每次点 3 整行 = 9 个（不触发 10 并发限制），然后滚动 3 行
|
||||
|
||||
|
||||
def adb(*args):
|
||||
return subprocess.run([ADB, "-s", DEVICE] + list(args),
|
||||
capture_output=True, text=True)
|
||||
|
||||
|
||||
def tap(x, y, delay=0.8):
|
||||
adb("shell", "input", "tap", str(x), str(y))
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
def find_by_text(target_text, retries=4, delay=2.0):
|
||||
"""通过 uiautomator dump 查找指定 text 节点的中心坐标。
|
||||
返回 (cx, cy) 或 None。重试次数受限，每次间隔 >= 2s，避免 ANR。
|
||||
"""
|
||||
import re
|
||||
# 用 [^<>]*? 限制在同一 <node /> 标签内，并跨过含 / 的属性
|
||||
# （如 resource-id="com.videogo:id/tv_download_all"）
|
||||
pat = re.compile(
|
||||
r'text="' + re.escape(target_text) + r'"[^<>]*?bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"',
|
||||
)
|
||||
for i in range(retries):
|
||||
adb("shell", "uiautomator", "dump", "/sdcard/uib.xml")
|
||||
adb("pull", "/sdcard/uib.xml", "/tmp/uib.xml")
|
||||
try:
|
||||
with open("/tmp/uib.xml") as f:
|
||||
content = f.read()
|
||||
except FileNotFoundError:
|
||||
content = ""
|
||||
m = pat.search(content)
|
||||
if m:
|
||||
x1, y1, x2, y2 = map(int, m.groups())
|
||||
return ((x1 + x2) // 2, (y1 + y2) // 2)
|
||||
if i < retries - 1:
|
||||
time.sleep(delay)
|
||||
return None
|
||||
|
||||
|
||||
def tap_text(text, after_delay=1.5, retries=8):
|
||||
"""查找文字元素并点击。找不到抛 RuntimeError，避免误触别的页面。"""
|
||||
pos = find_by_text(text, retries=retries)
|
||||
if pos is None:
|
||||
raise RuntimeError(f"找不到目标文字：'{text}'，停止以避免误触")
|
||||
cx, cy = pos
|
||||
print(f" → tap '{text}' at ({cx}, {cy})")
|
||||
adb("shell", "input", "tap", str(cx), str(cy))
|
||||
time.sleep(after_delay)
|
||||
|
||||
|
||||
def tap_until_text(x, y, expected_text, max_tries=4, after_tap_wait=4.0):
|
||||
"""点击 (x,y) 并等待 expected_text 出现，验证页面切换。
|
||||
没切换则重试点击。tap 后等 4 秒再 dump，避免 ANR（dump 太密集会让 App 卡死）。
|
||||
"""
|
||||
for trial in range(max_tries):
|
||||
adb("shell", "input", "tap", str(x), str(y))
|
||||
time.sleep(after_tap_wait)
|
||||
if find_by_text(expected_text, retries=1):
|
||||
print(f" ✓ tap ({x},{y}) 后看到 '{expected_text}'")
|
||||
return True
|
||||
print(f" ! tap ({x},{y}) 后未出现 '{expected_text}'，重试 {trial+1}/{max_tries}")
|
||||
# ANR 风险点：连续失败时给 App 喘息时间
|
||||
time.sleep(2)
|
||||
raise RuntimeError(f"点击 ({x},{y}) 重试 {max_tries} 次仍未跳转到含 '{expected_text}' 的页面")
|
||||
|
||||
|
||||
def dismiss_anr_if_any():
|
||||
"""检测系统 ANR 弹窗，存在则点 Wait 让 App 继续运行（不杀进程）。"""
|
||||
result = adb("shell", "dumpsys", "activity", "activities")
|
||||
if "isn't responding" in result.stdout or "AppErrorDialog" in result.stdout:
|
||||
print(" ⚠ 检测到 ANR 弹窗，点 Wait")
|
||||
# Wait 按钮 bounds=[70,1299][1010,1425]，中心 (540,1362)
|
||||
adb("shell", "input", "tap", "540", "1362")
|
||||
time.sleep(4)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def tap_until_activity(x, y, expected_activity_keyword, max_tries=6, after_tap_wait=6.0):
|
||||
"""点击 (x,y)，通过 dumpsys 检查 Activity 是否切换到含 keyword 的页面。
|
||||
每次尝试前都检查 ANR，存在则先点 Wait。间隔放宽到 6s 给模拟器+萤石喘息时间。
|
||||
"""
|
||||
for trial in range(max_tries):
|
||||
dismiss_anr_if_any()
|
||||
adb("shell", "input", "tap", str(x), str(y))
|
||||
time.sleep(after_tap_wait)
|
||||
cur = current_activity()
|
||||
if cur and expected_activity_keyword in cur:
|
||||
print(f" ✓ tap ({x},{y}) 后进入 '{cur}'")
|
||||
return True
|
||||
print(f" ! tap ({x},{y}) 后 Activity='{cur}' 不含 '{expected_activity_keyword}'，重试 {trial+1}/{max_tries}")
|
||||
# 失败时给 App 更长的喘息时间（避免连环 ANR）
|
||||
time.sleep(5)
|
||||
raise RuntimeError(
|
||||
f"点击 ({x},{y}) 重试 {max_tries} 次仍未跳到含 '{expected_activity_keyword}' 的 Activity"
|
||||
)
|
||||
|
||||
|
||||
def current_activity():
|
||||
"""返回当前 topResumedActivity 的简短名（如 'MainTabActivity'）。"""
|
||||
import re
|
||||
result = adb("shell", "dumpsys", "activity", "activities")
|
||||
m = re.search(r'topResumedActivity=ActivityRecord\{[^}]*\s+com\.videogo/\.?([^\s}]+)', result.stdout)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def assert_activity(keyword, hint=""):
|
||||
"""断言当前 Activity 名含 keyword。不含则报错。避免在错页面盲点。"""
|
||||
cur = current_activity()
|
||||
if not cur or keyword not in cur:
|
||||
raise RuntimeError(f"期望 Activity 含 '{keyword}'，实际为 '{cur}'。{hint}")
|
||||
return cur
|
||||
|
||||
|
||||
def tap_safe(x, y, expect_activity_keyword, hint="", delay=1.5):
|
||||
"""先断言当前 Activity 匹配 keyword，再点击 (x,y)，避免误触别的页面。"""
|
||||
cur = assert_activity(expect_activity_keyword, hint)
|
||||
print(f" → tap ({x},{y}) [当前 Activity: {cur}]")
|
||||
adb("shell", "input", "tap", str(x), str(y))
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
def sample_pixel_rgb(x, y, png_path="/tmp/_pixchk.png"):
|
||||
"""截图并采样指定坐标 RGB。需要系统装了 ImageMagick (`magick`)。"""
|
||||
import re
|
||||
adb("shell", "screencap", "-p", "/sdcard/_pixchk.png")
|
||||
adb("pull", "/sdcard/_pixchk.png", png_path)
|
||||
out = subprocess.run(
|
||||
["magick", png_path, "-format", "%[pixel:p{" + str(x) + "," + str(y) + "}]", "info:"],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
m = re.match(r"srgba?\((\d+),(\d+),(\d+)", out.stdout.strip())
|
||||
if not m:
|
||||
return (0, 0, 0)
|
||||
return tuple(map(int, m.groups()))
|
||||
|
||||
|
||||
def tap_until_pixel_changed(x, y, sample_x, sample_y, condition_fn,
|
||||
max_tries=5, timeout_per_try=8, interval=1.0):
|
||||
"""点 (x,y)，然后轮询 (sample_x, sample_y) 像素，直到 condition_fn(rgb) 为真。
|
||||
超时则重试点击。用于无文字可定位、但有视觉信号的场景（如切日期后工具栏变化）。
|
||||
"""
|
||||
for trial in range(max_tries):
|
||||
adb("shell", "input", "tap", str(x), str(y))
|
||||
for sec in range(int(timeout_per_try / interval)):
|
||||
time.sleep(interval)
|
||||
rgb = sample_pixel_rgb(sample_x, sample_y)
|
||||
if condition_fn(rgb):
|
||||
print(f" ✓ tap ({x},{y}) 生效，{sec+1}s 后检测到像素变化 rgb={rgb}")
|
||||
return True
|
||||
print(f" ! tap ({x},{y}) 后像素无变化（{timeout_per_try}s），重试 {trial+1}/{max_tries}")
|
||||
raise RuntimeError(f"点击 ({x},{y}) 重试 {max_tries} 次后像素仍未变化")
|
||||
|
||||
|
||||
def screenshot(path="/tmp/ezviz_screen.png"):
|
||||
adb("shell", "screencap", "-p", "/sdcard/ezviz_tmp.png")
|
||||
adb("pull", "/sdcard/ezviz_tmp.png", path)
|
||||
return path
|
||||
|
||||
|
||||
def launch_app():
|
||||
"""从头启动 App，等待 618 开屏广告自动关闭后回到首页。
|
||||
用 am start -S 强制重启主 Activity（比 monkey 更可靠：避免之前残留的
|
||||
RN 业务页占据栈顶导致 monkey 不进入首页）。
|
||||
"""
|
||||
adb("shell", "am", "force-stop", PKG)
|
||||
time.sleep(2)
|
||||
adb("shell", "am", "start", "-S", "-n", f"{PKG}/.LauncherActivity")
|
||||
# 等开屏广告（618 全屏广告自动播放 5~8 秒）+ App 启动稳定
|
||||
time.sleep(15)
|
||||
|
||||
|
||||
def is_error_page():
|
||||
"""检测当前是否为'加载失败'白屏页面
|
||||
取屏幕中央像素，白色/浅灰 = 错误页，深色 = 有视频内容。
|
||||
"""
|
||||
tmp = "/tmp/ezviz_check.png"
|
||||
adb("shell", "screencap", "-p", "/sdcard/ezviz_chk.png")
|
||||
adb("pull", "/sdcard/ezviz_chk.png", tmp)
|
||||
try:
|
||||
with open(tmp, "rb") as f:
|
||||
data = f.read()
|
||||
# 解析 PNG IDAT 得到原始像素
|
||||
pos, idat = 8, b""
|
||||
width = height = 0
|
||||
while pos < len(data):
|
||||
length = struct.unpack(">I", data[pos:pos+4])[0]
|
||||
ctype = data[pos+4:pos+8]
|
||||
cdata = data[pos+8:pos+8+length]
|
||||
if ctype == b"IHDR":
|
||||
width, height = struct.unpack(">II", cdata[:8])
|
||||
elif ctype == b"IDAT":
|
||||
idat += cdata
|
||||
elif ctype == b"IEND":
|
||||
break
|
||||
pos += 12 + length
|
||||
raw = zlib.decompress(idat)
|
||||
# screencap 输出 RGBA，每行有 1 字节 filter 前缀
|
||||
bpp = 4
|
||||
stride = width * bpp + 1
|
||||
cx, cy = width // 2, height // 3 # 取屏幕上 1/3 处中央（视频区域）
|
||||
row_start = cy * stride + 1
|
||||
r = raw[row_start + cx * bpp]
|
||||
g = raw[row_start + cx * bpp + 1]
|
||||
b = raw[row_start + cx * bpp + 2]
|
||||
# 亮度 > 220 认为是白屏（错误页）
|
||||
return (r + g + b) // 3 > 220
|
||||
except Exception:
|
||||
return False # 解析失败时不做干预
|
||||
|
||||
|
||||
def navigate_to_yesterday_batch():
|
||||
"""导航到昨天的 SD 卡批量下载模式。
|
||||
关键节点用 tap_text 通过 UI dump 定位，避免坐标错位时误触首页底部
|
||||
Tab（萤石/对话/商城/社区/我的，"商城"中心 x=540 跟录像 Tab 撞车）。
|
||||
"""
|
||||
print("→ 点击首页摄像头卡片（用 Activity 切换验证，避免频繁 dump 触发 ANR）")
|
||||
# 详情页是 RN 实现的 EZReactBizActivity
|
||||
tap_until_activity(*COORD["home_camera_card"], expected_activity_keyword="EZReactBizActivity")
|
||||
|
||||
print("→ 点击实时视频（详情页有文字，可以 dump 定位）")
|
||||
tap_text("实时视频", after_delay=5)
|
||||
|
||||
for attempt in range(3):
|
||||
if not is_error_page():
|
||||
print(" 实时视频加载成功")
|
||||
break
|
||||
print(f" 检测到加载失败，重试 {attempt+1}/3 ...")
|
||||
tap(*COORD["reload_btn"], delay=6)
|
||||
else:
|
||||
print(" 三次失败，返回重试")
|
||||
adb("shell", "input", "keyevent", "KEYCODE_BACK")
|
||||
time.sleep(2)
|
||||
tap(*COORD["home_camera_card"], delay=3)
|
||||
tap_text("实时视频", after_delay=8)
|
||||
|
||||
# 实时视频页全自绘，dump 不到底部 Tab 文字。
|
||||
# 但当前 Activity = VideoPlayActivityNewVersion，跟首页 MainTabActivity 不同。
|
||||
# 用 Activity 校验保护坐标点击，避免误触首页商城。
|
||||
print("→ 切换到录像 Tab（先校验在实时视频页）")
|
||||
tap_safe(*COORD["record_tab"], expect_activity_keyword="VideoPlayActivity",
|
||||
hint="未进入实时视频页，可能上一步点击实时视频按钮失败")
|
||||
|
||||
print("→ 切换到昨天（像素验证：切换后日期标签 (180,1130) 出现深色文字）")
|
||||
assert_activity("VideoPlayActivity", "未在实时视频/录像页")
|
||||
# "今日" 状态下 (180,1130) 是浅蓝灰背景 (244,246,252)
|
||||
# "06-XX" 状态下该位置是数字深色文字（R~44）
|
||||
tap_until_pixel_changed(
|
||||
*COORD["prev_day_arrow"],
|
||||
sample_x=180, sample_y=1130,
|
||||
condition_fn=lambda rgb: rgb[0] < 100,
|
||||
)
|
||||
# 日期切换后工具栏不会自动切换。需要点一个缩略图把"录像选中状态"打开，
|
||||
# 工具栏才从"延时摄影/清晰度/对讲/..." 切到"收藏/下载/剪辑/..."
|
||||
# 工具栏第 3 位 (540,900) 由浅蓝（对讲圆圈）变白即为切换完成
|
||||
print("→ 点一个缩略图触发工具栏切换")
|
||||
adb("shell", "input", "tap", "200", "1400") # 第一行第一列缩略图
|
||||
for _ in range(15):
|
||||
time.sleep(1)
|
||||
rgb = sample_pixel_rgb(540, 900)
|
||||
if rgb[0] > 240 and rgb[1] > 240 and rgb[2] > 240:
|
||||
print(f" ✓ 工具栏已切换到含'下载'按钮（rgb={rgb}）")
|
||||
break
|
||||
else:
|
||||
raise RuntimeError("点缩略图后工具栏未切换到下载模式")
|
||||
# 等视频开始稳定播放再操作（否则 tap 会被加载过程吸收）
|
||||
time.sleep(3)
|
||||
|
||||
# 下载按钮可能因为视频还在加载而被吸收，循环重点击直到菜单弹出
|
||||
print("→ 点击下载按钮 + 等批量下载菜单出现（最多 4 次重试）")
|
||||
assert_activity("VideoPlayActivity", "未在录像页")
|
||||
for trial in range(4):
|
||||
adb("shell", "input", "tap", str(COORD["download_btn"][0]), str(COORD["download_btn"][1]))
|
||||
time.sleep(3)
|
||||
pos = find_by_text("批量下载", retries=3, delay=1.5)
|
||||
if pos:
|
||||
print(f" ✓ 菜单弹出，'批量下载' at {pos}")
|
||||
cx, cy = pos
|
||||
adb("shell", "input", "tap", str(cx), str(cy))
|
||||
time.sleep(3)
|
||||
break
|
||||
print(f" ! 第 {trial+1} 次点下载未弹出菜单，重试")
|
||||
else:
|
||||
raise RuntimeError("反复点下载按钮仍未弹出批量下载菜单")
|
||||
|
||||
|
||||
def dump_thumbnail_bounds():
|
||||
"""通过 uiautomator dump 拿到当前批量下载界面可见缩略图的 bounds 列表。
|
||||
返回 [(x1,y1,x2,y2), ...] 按 y 升序。
|
||||
过滤：宽 > 300，高 > 150，y1 在 [1000, 2300] 区间。
|
||||
"""
|
||||
import re
|
||||
adb("shell", "uiautomator", "dump", "/sdcard/uib.xml")
|
||||
adb("pull", "/sdcard/uib.xml", "/tmp/uib.xml")
|
||||
try:
|
||||
with open("/tmp/uib.xml") as f:
|
||||
content = f.read()
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
pat = re.compile(r'clickable="true"[^/]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"')
|
||||
out = []
|
||||
for m in pat.finditer(content):
|
||||
x1, y1, x2, y2 = map(int, m.groups())
|
||||
if 1000 <= y1 <= 2300 and (x2 - x1) > 300 and (y2 - y1) > 150:
|
||||
out.append((x1, y1, x2, y2))
|
||||
out.sort(key=lambda b: (b[1], b[0]))
|
||||
return out
|
||||
|
||||
|
||||
def dismiss_dialog_if_any():
|
||||
"""检测并关闭"使用手机网络下载将消耗流量"等萤石弹窗。
|
||||
点击 text="确定" 按钮。
|
||||
"""
|
||||
pos = find_by_text("确定", retries=1)
|
||||
if pos:
|
||||
print(f" ⚠ 检测到弹窗，点击'确定' at {pos}")
|
||||
adb("shell", "input", "tap", str(pos[0]), str(pos[1]))
|
||||
time.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def tap_all_thumbnails(max_rounds=100, batch_sleep=35):
|
||||
"""循环：dump → 点击 9 个 → 等下载 → 滚动 3 行。
|
||||
每轮点 3 整行（9 个）避免触发 10 并发限制。
|
||||
滚动距离 = 3 行 = ~609 像素，保证下轮可见行与本轮不重叠。
|
||||
检测到底：连续 2 轮 bounds 列表相同则结束。
|
||||
"""
|
||||
print(f"→ 开始批量下载循环（最多 {max_rounds} 轮）")
|
||||
# 进入批量模式后会有"使用手机网络流量"弹窗，先关掉
|
||||
dismiss_dialog_if_any()
|
||||
last_signature = None
|
||||
same_count = 0
|
||||
|
||||
for r in range(max_rounds):
|
||||
bounds = dump_thumbnail_bounds()
|
||||
if not bounds:
|
||||
print(f" 轮 {r+1}: 没拿到缩略图 bounds，结束")
|
||||
break
|
||||
|
||||
# 取完整可见行（高度 > 195）
|
||||
full = [b for b in bounds if (b[3] - b[1]) > 195]
|
||||
if len(full) < 3:
|
||||
print(f" 轮 {r+1}: 完整行不足 ({len(full)} < 3)，结束")
|
||||
break
|
||||
|
||||
# 检测到底：bounds 内容跟上次相同
|
||||
signature = tuple(full[:9])
|
||||
if signature == last_signature:
|
||||
same_count += 1
|
||||
if same_count >= 2:
|
||||
print(f" 轮 {r+1}: 连续 2 轮内容相同，已到底，结束")
|
||||
break
|
||||
else:
|
||||
same_count = 0
|
||||
last_signature = signature
|
||||
|
||||
to_click = full[:9]
|
||||
for x1, y1, x2, y2 in to_click:
|
||||
cx, cy = (x1 + x2) // 2, (y1 + y2) // 2
|
||||
adb("shell", "input", "tap", str(cx), str(cy))
|
||||
time.sleep(0.3)
|
||||
|
||||
print(f" 轮 {r+1}: 点击 {len(to_click)} 个，等 {batch_sleep}s 下载...")
|
||||
time.sleep(batch_sleep)
|
||||
|
||||
# 向下滚动 3 行（约 609 像素）
|
||||
adb("shell", "input", "swipe", "540", "1820", "540", "1211", "400")
|
||||
time.sleep(1.5)
|
||||
|
||||
print("→ 批量下载循环结束")
|
||||
|
||||
|
||||
def wait_for_downloads(baseline_files, timeout=600):
|
||||
"""等待下载目录新增文件稳定
|
||||
文件名使用下载时刻时间戳，固定存到今天的目录。
|
||||
baseline_files 为下载前已有的文件集合，用于计算新增数量。
|
||||
"""
|
||||
today = datetime.now().strftime("%Y%m%d")
|
||||
remote_dir = f"/sdcard/Android/data/{PKG}/files/{today}"
|
||||
print(f"→ 等待下载完成（目录 {today}，起始已有 {len(baseline_files)} 个）...")
|
||||
|
||||
last_new = -1
|
||||
stable_rounds = 0
|
||||
deadline = time.time() + timeout
|
||||
|
||||
while time.time() < deadline:
|
||||
result = adb("shell", "find", remote_dir, "-name", "*_BH*", "-type", "f")
|
||||
all_files = set(l for l in result.stdout.strip().splitlines() if l)
|
||||
new_files = all_files - baseline_files
|
||||
count = len(new_files)
|
||||
print(f" 新下载 {count} 个文件...")
|
||||
if count > 0 and count == last_new:
|
||||
stable_rounds += 1
|
||||
if stable_rounds >= 3:
|
||||
print(f"→ 下载稳定，共新增 {count} 个文件")
|
||||
return list(new_files)
|
||||
else:
|
||||
stable_rounds = 0
|
||||
last_new = count
|
||||
time.sleep(10)
|
||||
|
||||
print("警告：等待超时")
|
||||
# 返回已下载的新文件（即便未完全稳定）
|
||||
result = adb("shell", "find", remote_dir, "-name", "*_BH*", "-type", "f")
|
||||
all_files = set(l for l in result.stdout.strip().splitlines() if l)
|
||||
return list(all_files - baseline_files)
|
||||
|
||||
|
||||
def pull_files(remote_files):
|
||||
"""把下载好的文件 adb pull 到本地"""
|
||||
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
|
||||
# 本地目录按录像日期（昨天）命名，方便归档
|
||||
local_dir = OUTPUT_DIR / "BH5455783" / yesterday
|
||||
local_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
print(f"→ 拉取 {len(remote_files)} 个文件到 {local_dir}")
|
||||
for remote_path in remote_files:
|
||||
fname = Path(remote_path).name + ".mp4"
|
||||
local_path = local_dir / fname
|
||||
if local_path.exists():
|
||||
print(f" 跳过（已存在）: {fname}")
|
||||
continue
|
||||
adb("pull", remote_path, str(local_path))
|
||||
print(f" 已拉取: {fname}")
|
||||
|
||||
print(f"→ 完成，文件保存在 {local_dir}")
|
||||
|
||||
|
||||
def get_baseline_files():
|
||||
"""获取下载开始前已有的文件列表"""
|
||||
today = datetime.now().strftime("%Y%m%d")
|
||||
remote_dir = f"/sdcard/Android/data/{PKG}/files/{today}"
|
||||
result = adb("shell", "find", remote_dir, "-name", "*_BH*", "-type", "f")
|
||||
return set(l for l in result.stdout.strip().splitlines() if l)
|
||||
|
||||
|
||||
def sync():
|
||||
print("===== 萤石录像自动同步开始 =====")
|
||||
launch_app()
|
||||
# 记录下载前的基准文件集合
|
||||
baseline = get_baseline_files()
|
||||
print(f"→ 基准文件数: {len(baseline)}")
|
||||
navigate_to_yesterday_batch()
|
||||
tap_all_thumbnails() # 默认 max_rounds=100，跑到滚到底为止
|
||||
files = wait_for_downloads(baseline)
|
||||
if files:
|
||||
pull_files(files)
|
||||
else:
|
||||
print("未找到新下载文件")
|
||||
print("===== 同步完成 =====")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sync()
|
||||
@@ -0,0 +1,8 @@
|
||||
# 复制此文件为 config.py 后填入真实值。config.py 已加入 .gitignore。
|
||||
APP_KEY = "你的萤石开放平台 AppKey"
|
||||
APP_SECRET = "你的萤石开放平台 AppSecret"
|
||||
DEVICE_SERIAL = "BHxxxxxxx" # 设备序列号，在萤石 App 设备信息里
|
||||
VERIFY_CODE = "XXXXXX" # 设备验证码（贴在摄像头上）
|
||||
CHANNEL_NO = 1
|
||||
OUTPUT_DIR = "./recordings"
|
||||
TOKEN_CACHE = "./token_cache.json"
|
||||
@@ -0,0 +1,141 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
第一步：探测萤石API，确认录像列表和回放流接口的实际返回格式
|
||||
"""
|
||||
|
||||
import json, os, requests
|
||||
from datetime import datetime, timedelta
|
||||
from config import *
|
||||
|
||||
BASE = "https://open.ys7.com/api/lapp"
|
||||
|
||||
|
||||
def get_token():
|
||||
"""获取AccessToken，缓存到本地（有效期7天）"""
|
||||
if os.path.exists(TOKEN_CACHE):
|
||||
cache = json.load(open(TOKEN_CACHE))
|
||||
if datetime.now().timestamp() < cache["expire_ts"] - 3600:
|
||||
print(f"使用缓存Token，有效期至: {cache['expire_str']}")
|
||||
return cache["token"]
|
||||
|
||||
r = requests.post(f"{BASE}/token/get",
|
||||
data={"appKey": APP_KEY, "appSecret": APP_SECRET})
|
||||
d = r.json()
|
||||
if d["code"] != "200":
|
||||
raise Exception(f"Token获取失败: {d}")
|
||||
|
||||
token = d["data"]["accessToken"]
|
||||
expire_ts = d["data"]["expireTime"] / 1000
|
||||
expire_str = datetime.fromtimestamp(expire_ts).strftime("%Y-%m-%d %H:%M:%S")
|
||||
json.dump({"token": token, "expire_ts": expire_ts, "expire_str": expire_str},
|
||||
open(TOKEN_CACHE, "w"))
|
||||
print(f"Token已刷新，有效期至: {expire_str}")
|
||||
return token
|
||||
|
||||
|
||||
def probe_record_search(token, date_str):
|
||||
"""探测：查询SD卡录像列表（试几个端点）"""
|
||||
params = {
|
||||
"accessToken": token,
|
||||
"deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"startTime": f"{date_str} 00:00:00",
|
||||
"endTime": f"{date_str} 23:59:59",
|
||||
"recType": 0,
|
||||
"pageStart": 0,
|
||||
"pageSize": 10
|
||||
}
|
||||
endpoints = [
|
||||
"/record/search",
|
||||
"/storage/recording/day/times",
|
||||
"/device/record/list",
|
||||
"/video/rec/list",
|
||||
]
|
||||
for ep in endpoints:
|
||||
print(f"\n{'='*50}")
|
||||
print(f"【接口1变体】{ep} 日期={date_str}")
|
||||
r = requests.post(f"{BASE}{ep}", data=params)
|
||||
print(f"HTTP {r.status_code}")
|
||||
print(r.text[:500])
|
||||
return None
|
||||
|
||||
|
||||
def probe_video_by_time(token, date_str):
|
||||
"""探测：尝试多种时间格式"""
|
||||
from datetime import timezone
|
||||
dt_start = datetime.strptime(f"{date_str} 00:00:00", "%Y-%m-%d %H:%M:%S")
|
||||
dt_end = datetime.strptime(f"{date_str} 01:00:00", "%Y-%m-%d %H:%M:%S")
|
||||
ts_start = int(dt_start.timestamp() * 1000) # 毫秒时间戳
|
||||
ts_end = int(dt_end.timestamp() * 1000)
|
||||
|
||||
formats = [
|
||||
("yyyy-MM-dd HH:mm:ss", f"{date_str} 00:00:00", f"{date_str} 01:00:00"),
|
||||
("yyyyMMddHHmmss", date_str.replace("-","")+"000000", date_str.replace("-","")+"010000"),
|
||||
("毫秒时间戳", str(ts_start), str(ts_end)),
|
||||
("秒时间戳", str(ts_start//1000), str(ts_end//1000)),
|
||||
]
|
||||
for fmt_name, s, e in formats:
|
||||
print(f"\n{'='*50}")
|
||||
print(f"【接口2】video/by/time 时间格式={fmt_name} start={s}")
|
||||
r = requests.post(f"{BASE}/video/by/time", data={
|
||||
"accessToken": token,
|
||||
"deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"startTime": s,
|
||||
"endTime": e
|
||||
})
|
||||
print(f"HTTP {r.status_code} {r.text[:200]}")
|
||||
return None
|
||||
|
||||
|
||||
def probe_live_address(token):
|
||||
"""探测：获取直播流地址（顺带看看协议格式）"""
|
||||
print(f"\n{'='*50}")
|
||||
print("【接口3】直播流地址（探测协议格式）")
|
||||
r = requests.post(f"{BASE}/v2/live/address/get", data={
|
||||
"accessToken": token,
|
||||
"deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"protocol": 2, # 1=ezopen 2=HLS 3=RTMP 4=FLV
|
||||
"quality": 1
|
||||
})
|
||||
print(f"HTTP {r.status_code}")
|
||||
print(r.text[:500])
|
||||
|
||||
|
||||
def probe_playback_address(token):
|
||||
"""探测回放流地址（已知录像时间段的毫秒时间戳）"""
|
||||
rec_start = 1781179564000
|
||||
rec_end = 1781211539000
|
||||
endpoints = [
|
||||
("/v2/live/address/get", {"startTime": rec_start, "endTime": rec_end, "protocol": 2, "quality": 1}),
|
||||
("/record/playback/address", {"startTime": rec_start, "endTime": rec_end}),
|
||||
("/v3/live/address/get", {"startTime": rec_start, "endTime": rec_end, "protocol": 2}),
|
||||
("/video/playback", {"startTime": rec_start, "endTime": rec_end}),
|
||||
]
|
||||
for ep, extra in endpoints:
|
||||
params = {"accessToken": token, "deviceSerial": DEVICE_SERIAL, "channelNo": CHANNEL_NO}
|
||||
params.update(extra)
|
||||
print(f"\n{'='*50}")
|
||||
print(f"【回放】{ep}")
|
||||
r = requests.post(f"{BASE}{ep}", data=params)
|
||||
print(f"HTTP {r.status_code} {r.text[:300]}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
token = get_token()
|
||||
|
||||
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
y_start_ms = int(datetime.strptime(f"{yesterday} 00:00:00", "%Y-%m-%d %H:%M:%S").timestamp() * 1000)
|
||||
y_end_ms = int(datetime.strptime(f"{yesterday} 23:59:59", "%Y-%m-%d %H:%M:%S").timestamp() * 1000)
|
||||
|
||||
print("\n=== 1. 查录像列表 ===")
|
||||
r = requests.post(f"{BASE}/video/by/time", data={
|
||||
"accessToken": token, "deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"startTime": y_start_ms, "endTime": y_end_ms
|
||||
})
|
||||
print(r.text[:800])
|
||||
|
||||
print("\n=== 2. 探测回放流地址 ===")
|
||||
probe_playback_address(token)
|
||||
@@ -0,0 +1,155 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
萤石CB60 SD卡录像每日同步脚本
|
||||
每天凌晨运行，下载前24小时内的所有SD卡录像到本地目录。
|
||||
"""
|
||||
import json, os, re, subprocess, logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
import requests
|
||||
|
||||
from config import *
|
||||
|
||||
# ====== 路径配置 ======
|
||||
OUTPUT_DIR = Path(OUTPUT_DIR)
|
||||
TOKEN_CACHE = Path(TOKEN_CACHE)
|
||||
LOG_FILE = Path("./sync.log")
|
||||
|
||||
# NAS上部署时改为 "ffmpeg"（Docker镜像内）
|
||||
FFMPEG = "/opt/homebrew/bin/ffmpeg"
|
||||
|
||||
BASE = "https://open.ys7.com/api/lapp"
|
||||
|
||||
# ====== 日志 ======
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(LOG_FILE, encoding="utf-8"),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ====== Token 管理 ======
|
||||
def get_token():
|
||||
if TOKEN_CACHE.exists():
|
||||
cache = json.loads(TOKEN_CACHE.read_text())
|
||||
if datetime.now().timestamp() < cache["expire_ts"] - 3600:
|
||||
log.info(f"使用缓存 Token，有效期至 {cache['expire_str']}")
|
||||
return cache["token"]
|
||||
|
||||
r = requests.post(f"{BASE}/token/get",
|
||||
data={"appKey": APP_KEY, "appSecret": APP_SECRET})
|
||||
d = r.json()
|
||||
if d["code"] != "200":
|
||||
raise Exception(f"Token 获取失败: {d}")
|
||||
|
||||
token = d["data"]["accessToken"]
|
||||
expire_ts = d["data"]["expireTime"] / 1000
|
||||
expire_str = datetime.fromtimestamp(expire_ts).strftime("%Y-%m-%d %H:%M:%S")
|
||||
TOKEN_CACHE.write_text(json.dumps({
|
||||
"token": token, "expire_ts": expire_ts, "expire_str": expire_str
|
||||
}))
|
||||
log.info(f"Token 已刷新，有效期至 {expire_str}")
|
||||
return token
|
||||
|
||||
|
||||
# ====== 查询录像列表 ======
|
||||
def get_recording_list(token, start_ms, end_ms):
|
||||
r = requests.post(f"{BASE}/video/by/time", data={
|
||||
"accessToken": token,
|
||||
"deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"startTime": start_ms,
|
||||
"endTime": end_ms,
|
||||
})
|
||||
d = r.json()
|
||||
if d["code"] != "200":
|
||||
log.warning(f"录像列表查询失败: {d}")
|
||||
return []
|
||||
return d.get("data", [])
|
||||
|
||||
|
||||
# ====== 获取 FLV 回放流地址 ======
|
||||
def get_flv_url(token):
|
||||
"""
|
||||
type=2：SD卡回放流，覆盖过去24小时内所有录像片段（服务器拼接成一个 FLV）
|
||||
protocol=4：HTTP FLV，ffmpeg 可直接下载
|
||||
"""
|
||||
r = requests.post(f"{BASE}/v2/live/address/get", data={
|
||||
"accessToken": token,
|
||||
"deviceSerial": DEVICE_SERIAL,
|
||||
"channelNo": CHANNEL_NO,
|
||||
"quality": 1,
|
||||
"type": 2,
|
||||
"protocol": 4,
|
||||
})
|
||||
d = r.json()
|
||||
if d["code"] != "200":
|
||||
raise Exception(f"FLV 地址获取失败: {d}")
|
||||
return d["data"]["url"], d["data"]["expireTime"]
|
||||
|
||||
|
||||
# ====== ffmpeg 下载 ======
|
||||
def download_flv(url, output_path):
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
cmd = [
|
||||
FFMPEG,
|
||||
"-i", url,
|
||||
"-c", "copy",
|
||||
"-y", str(output_path),
|
||||
"-v", "warning",
|
||||
]
|
||||
log.info(f"开始下载 → {output_path}")
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
|
||||
if not output_path.exists() or output_path.stat().st_size < 1024:
|
||||
log.error(f"下载失败或文件过小\n{result.stderr[:400]}")
|
||||
return False
|
||||
|
||||
size_kb = output_path.stat().st_size / 1024
|
||||
log.info(f"下载完成，大小 {size_kb:.0f} KB")
|
||||
return True
|
||||
|
||||
|
||||
# ====== 主流程 ======
|
||||
def sync():
|
||||
log.info("===== 开始同步 =====")
|
||||
token = get_token()
|
||||
|
||||
# 1. 查过去24小时是否有录像
|
||||
now_ms = int(datetime.now().timestamp() * 1000)
|
||||
start_ms = now_ms - 86_400_000
|
||||
recordings = get_recording_list(token, start_ms, now_ms)
|
||||
|
||||
if not recordings:
|
||||
log.info("过去24小时无SD卡录像，退出")
|
||||
return
|
||||
|
||||
log.info(f"发现 {len(recordings)} 段录像")
|
||||
|
||||
# 2. 获取 FLV 流（服务器把24小时内所有录像拼接好）
|
||||
url, expire_time = get_flv_url(token)
|
||||
log.info(f"FLV URL 有效期至 {expire_time}")
|
||||
|
||||
# 3. 文件命名：按 begin 时间戳
|
||||
begin_match = re.search(r'begin=(\d{14})', url)
|
||||
begin_str = begin_match.group(1) if begin_match else datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
date_str = begin_str[:8] # e.g. 20260613
|
||||
|
||||
output_path = OUTPUT_DIR / DEVICE_SERIAL / date_str / f"{begin_str}.mp4"
|
||||
|
||||
# 4. 已下载则跳过
|
||||
if output_path.exists() and output_path.stat().st_size > 1024:
|
||||
log.info(f"已存在 {output_path.name}，跳过")
|
||||
return
|
||||
|
||||
# 5. 下载
|
||||
download_flv(url, output_path)
|
||||
log.info("===== 同步完成 =====")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sync()
|
||||
Reference in New Issue
Block a user