yuming
68847167dd
之前用固定步长204采样y坐标，与实际行位置不匹配， 只检测到5个角标（实际16个），导致等待逻辑永远超时。 改为每次截图后重新 detect_card_rows 检测实际行位置。
1048 lines
57 KiB
Python
1048 lines
57 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
萤石录像自动下载脚本
|
||
通过 ADB 自动化模拟器中的萤石 App，下载前一天的所有报警录像。
|
||
"""
|
||
import subprocess, time, os, shutil, struct, zlib, io, hashlib, math
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
from PIL import Image
|
||
|
||
ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb")
|
||
DEVICE = "emulator-5554"
|
||
OUTPUT_DIR = Path("./recordings")
|
||
PKG = "com.videogo"
|
||
|
||
# 屏幕坐标（Pixel 6 模拟器 1080x2400，2026-07-24 App 更新到 7.8.1 后重新实测校准）
|
||
COORD = {
|
||
"home_camera_card": (540, 500), # 首页摄像头卡片
|
||
"live_video_btn": (270, 1340), # 摄像头详情页-实时视频按钮
|
||
"record_tab": (540, 2290), # 底部录像 Tab
|
||
"prev_day_arrow": (77, 1124), # < 上一天箭头
|
||
"next_day_arrow": (370, 1124), # > 下一天箭头（用于归位到今日）
|
||
"download_btn": (406, 935), # 录像页工具栏-下载按钮
|
||
"batch_download": (540, 1904), # 弹出菜单"批量下载"
|
||
"reload_btn": (540, 1960), # 加载失败页的"重新加载"按钮
|
||
"today_thumb": (198, 1388), # "今日"默认列表第一张缩略图（用于切换工具栏到下载模式）
|
||
}
|
||
|
||
# 批量下载界面缩略图布局（7.8.1 实测；此版本自绘 UI，uiautomator dump 拿不到 bounds，
|
||
# 改用固定网格坐标点击）
|
||
# 3 列 x 中心：201, 540, 878；3 行 y 中心：1148, 1351, 1554
|
||
THUMB_X = [201, 540, 878]
|
||
THUMB_Y = [1148, 1351, 1554]
|
||
# 每次点 3 整行 = 9 个（不触发 10 并发限制），然后滚动 3 行（609 像素）
|
||
|
||
# 打勾（已下载完成）蓝色角标的检测参数。已打勾时卡片左下角有一个高饱和蓝色圆角
|
||
# 方块（约 RGB 100,143,252）；未下载/下载中没有。
|
||
# 关键教训：这个下载列表滚动时【不会 snap 对齐到固定行】，同一张卡片在不同滚动
|
||
# 状态下会停在不同的垂直位置（实测两张截图里蓝勾 y 偏移分别是 +62 和 +94，漂移
|
||
# 达 32px）。所以绝不能用"固定单点采样"判打勾——单点极易错过漂移后的蓝勾，导致
|
||
# 打勾永远检测不到、每屏 9 格全部重点，这才是"重复下载"的真正总根源。
|
||
# 正确做法：在卡片左下角开一个足够大的窗口（96x100），统计其中"强蓝"像素的数量，
|
||
# 数量超过阈值即判已打勾。窗口大到能覆盖 y 漂移；强蓝在灰/黑/彩色画面里几乎不出现，
|
||
# 所以打勾格实测蓝像素≈2500、未勾格≈0，区分极清晰。
|
||
# BADGE_OFFSET 现在表示"窗口中心"相对九宫格点击中心的偏移。
|
||
BADGE_OFFSET = (-121, 78)
|
||
BLUE_WIN = (96, 100) # 蓝勾搜索窗口宽、高
|
||
BLUE_THRESHOLD = 500 # 窗口内强蓝像素数 >= 此值判为已打勾（打勾≈2500，未勾≈0，余量充足）
|
||
|
||
# 动态行定位参数（替代写死的 THUMB_Y）。原理见 detect_card_rows 文档字符串。
|
||
# 列表滚动时卡片会因"整点分组标签"整行平移（实测漂移达 174px），固定 THUMB_Y 追不上，
|
||
# 导致点击/打勾检测全部错位、把已下载卡片误判为未下载重复点击。改为每屏用亮度标准差
|
||
# (std) 垂直剖面实测出真实的卡片行中心 Y：卡片缩略图有纹理 std 高，卡片间隙/分组标签/
|
||
# 灰占位是纯色 std≈0，据此切出"纹理带"，每个带的竖直中心就是一行卡片的点击中心。
|
||
ROW_Y0, ROW_Y1 = 1000, 2380 # 列表纵向扫描范围（避开顶部状态栏/预览/标题栏、底部手势条）
|
||
ROW_STD_X0, ROW_STD_W = 50, 980 # 横向 std 采样区（避开左右圆角/阴影）
|
||
ROW_STD_COLS = 200 # 降采样列数（Pillow 一次折成每行 175 个样本算 std，避免逐行上百次采样）
|
||
ROW_STD_LOW = 18 # std 阈值：>= 判为卡片纹理，< 判为间隙/标签/占位（实测卡片≥46、非卡片=0，余量足）
|
||
ROW_MIN_H = 90 # 纹理带最小高度（滤掉分组标签、半截卡片、噪声）
|
||
ROW_MAX_H = 260 # 纹理带最大高度（超过视为两行粘连，按行距二次切分）
|
||
ROW_PITCH = 204 # 相邻卡片行的竖直间距（用于粘连带二次切分）
|
||
ROW_EDGE_MARGIN = 12 # 纹理带触及扫描区上/下边缘视为被截断的半行，丢弃
|
||
THUMB_LOAD_WAIT = 1.0 # 每轮截图前等待缩略图从"灰占位"加载出真实画面，避免假间隙漏检
|
||
|
||
|
||
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 ensure_not_anr(max_wait=90):
|
||
"""确保 App 当前没有卡死（ANR），反复检测+点 Wait 直到恢复正常。
|
||
|
||
根本原因（实测复现）：批量下载循环里长时间连续点击/截图会让 App 偶尔卡死，
|
||
卡死期间截图读到的是冻结前的旧画面，打勾识别会把"其实已经点过/下载中"的卡片
|
||
误判为"未下载"，导致反复重新触发下载——单日 220 个下载里只有 11 种真实内容，
|
||
最严重的一组被重复下载了 28 次，就是这个机制造成的。
|
||
之前 dismiss_anr_if_any() 只在导航阶段被调用，批量下载循环里完全没有调用，
|
||
是这个问题的直接缺口。
|
||
|
||
卡死超过 max_wait 仍未恢复则直接报错停止，而不是继续在冻结画面上盲点，
|
||
避免制造更多重复下载。
|
||
"""
|
||
waited = 0
|
||
while dismiss_anr_if_any():
|
||
waited += 4
|
||
if waited >= max_wait:
|
||
raise RuntimeError(f"App 卡死(ANR)超过 {max_wait}s 仍未恢复，停止自动化，避免继续误触造成更多重复下载")
|
||
return waited
|
||
|
||
|
||
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。用 Pillow 实现，不依赖系统 ImageMagick。"""
|
||
adb("shell", "screencap", "-p", "/sdcard/_pixchk.png")
|
||
adb("pull", "/sdcard/_pixchk.png", png_path)
|
||
return sample_rgb_from_file(png_path, x, y)
|
||
|
||
|
||
def screenshot(path="/tmp/ezviz_screen.png"):
|
||
# 用 exec-out 直接把 PNG 输出到本地，省掉"先写 /sdcard 再 pull"的两步往返，
|
||
# 每次截图快约 0.5~1s（批量下载循环每轮要截多次图，累计提速明显）。
|
||
with open(path, "wb") as f:
|
||
subprocess.run([ADB, "-s", DEVICE, "exec-out", "screencap", "-p"],
|
||
stdout=f)
|
||
return path
|
||
|
||
|
||
def sample_rgb_from_file(png_path, x, y):
|
||
"""从已经拉取到本地的截图文件采样指定坐标 RGB（不重新截图，供批量取点复用）。
|
||
用 Pillow 实现，不依赖系统 ImageMagick。"""
|
||
try:
|
||
with Image.open(png_path) as img:
|
||
img = img.convert("RGB")
|
||
r, g, b = img.getpixel((x, y))
|
||
return (r, g, b)
|
||
except Exception:
|
||
return (0, 0, 0)
|
||
|
||
|
||
def count_blue_pixels(png_path, cx, cy):
|
||
"""统计以 (cx,cy) 为中心、大小 BLUE_WIN 的窗口内"强蓝"像素数量。
|
||
强蓝判据：b>150 且 b-r>80（蓝勾角标的高饱和蓝，灰/黑/彩色画面几乎不出现）。
|
||
用窗口统计而非单点采样，是为了吸收卡片垂直位置的漂移（见 BADGE_OFFSET 说明）。
|
||
用 Pillow 实现，不依赖系统 ImageMagick。"""
|
||
w, h = BLUE_WIN
|
||
x0, y0 = cx - w // 2, cy - h // 2
|
||
try:
|
||
with Image.open(png_path) as img:
|
||
img = img.convert("RGB")
|
||
crop = img.crop((x0, y0, x0 + w, y0 + h))
|
||
cnt = 0
|
||
for r, g, b in crop.getdata():
|
||
if b > 150 and b - r > 80:
|
||
cnt += 1
|
||
return cnt
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
def get_checked_cells(png_path):
|
||
"""【已废弃，仅保留备查】基于写死 THUMB_Y 的打勾检测。分组标签平移会使采样错位，
|
||
现由 get_checked_cells_dynamic 取代。"""
|
||
checked = {}
|
||
for gy in THUMB_Y:
|
||
for gx in THUMB_X:
|
||
cx, cy = gx + BADGE_OFFSET[0], gy + BADGE_OFFSET[1]
|
||
checked[(gx, gy)] = count_blue_pixels(png_path, cx, cy) >= BLUE_THRESHOLD
|
||
return checked
|
||
|
||
|
||
def std_profile(png_path):
|
||
"""计算截图列表区每一行的"亮度标准差"垂直剖面，返回 [(y, std), ...]（y 从 ROW_Y0 起）。
|
||
|
||
卡片缩略图有画面纹理 → 一行内各像素亮度差异大 → std 高（实测 ≥46）；
|
||
卡片间水平间隙、整点分组标签、灰色占位图都是纯色 → std≈0。据此可把"卡片行"
|
||
从"间隙/标签/占位"里分出来。
|
||
|
||
性能：不逐行采样（一屏上百次太慢），而是用 Pillow 把 ROW_STD_W 宽的采样区灰度图
|
||
横向降采样到 ROW_STD_COLS 列（每列是原若干像素的平均），再在 Python 里按行算这
|
||
ROW_STD_COLS 个样本的标准差。用 Pillow 实现，不依赖系统 ImageMagick。"""
|
||
h = ROW_Y1 - ROW_Y0
|
||
try:
|
||
with Image.open(png_path) as img:
|
||
img = img.convert("L")
|
||
crop = img.crop((ROW_STD_X0, ROW_Y0, ROW_STD_X0 + ROW_STD_W, ROW_Y0 + h))
|
||
small = crop.resize((ROW_STD_COLS, h), Image.Resampling.NEAREST)
|
||
pixels = list(small.getdata())
|
||
prof = []
|
||
for rr in range(h):
|
||
row_vals = pixels[rr * ROW_STD_COLS:(rr + 1) * ROW_STD_COLS]
|
||
if not row_vals:
|
||
continue
|
||
mean = sum(row_vals) / len(row_vals)
|
||
var = sum((v - mean) ** 2 for v in row_vals) / len(row_vals)
|
||
prof.append((ROW_Y0 + rr, math.sqrt(var)))
|
||
return prof
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def detect_card_rows(png_path):
|
||
"""动态检测当前屏可见的完整卡片行，返回各行中心 y 的升序列表。
|
||
|
||
列表滚动时卡片会因整点分组标签整行平移，写死的 THUMB_Y 追不上（这是"重复下载"的
|
||
总根源）。本函数每屏用 std_profile 实测卡片行的真实位置：把 std >= ROW_STD_LOW 的
|
||
连续区间当作一条"卡片纹理带"，过滤掉太矮（标签/半截/噪声）、太高（两行粘连，按
|
||
ROW_PITCH 二次切分）、以及触及扫描区上下边缘（被屏幕截断的半行）的带，剩下每条带的
|
||
竖直中心就是一行卡片的点击中心 y。配合固定的 3 列 THUMB_X 即可定位 9 宫格。
|
||
|
||
实测（3 张不同滚动位置的真实截图）：检出的行中心精准落在卡片竖直中心，分组标签处
|
||
不产生行——彻底消除固定网格遇标签错位的问题。"""
|
||
prof = std_profile(png_path)
|
||
textured = [(y, s >= ROW_STD_LOW) for y, s in prof]
|
||
runs = []
|
||
i, n = 0, len(textured)
|
||
while i < n:
|
||
if textured[i][1]:
|
||
j = i
|
||
while j < n and textured[j][1]:
|
||
j += 1
|
||
runs.append((textured[i][0], textured[j - 1][0]))
|
||
i = j
|
||
else:
|
||
i += 1
|
||
rows = []
|
||
for top, bot in runs:
|
||
height = bot - top
|
||
if height < ROW_MIN_H:
|
||
continue # 分组标签、半截卡片、纹理噪声
|
||
if top <= ROW_Y0 + ROW_EDGE_MARGIN or bot >= ROW_Y1 - ROW_EDGE_MARGIN:
|
||
continue # 触边=被屏幕上/下缘截断的半行（相邻屏会完整出现，不漏）
|
||
if height > ROW_MAX_H:
|
||
k = max(1, round(height / ROW_PITCH)) # 两行粘连，按行距切分
|
||
for m in range(k):
|
||
rows.append(top + int(height * (m + 0.5) / k))
|
||
else:
|
||
rows.append((top + bot) // 2)
|
||
return sorted(rows)
|
||
|
||
|
||
def get_checked_cells_dynamic(png_path, row_centers):
|
||
"""对动态检测出的每一行 × 固定 3 列，判断是否已打勾（下载完成）。
|
||
返回 {(gx, cy): True/False}，key 的 cy 是动态行中心。判据同 count_blue_pixels。
|
||
横向 3 列 THUMB_X 不随滚动漂移，故列坐标仍固定；只有行 y 是动态的。"""
|
||
checked = {}
|
||
for cy in row_centers:
|
||
for gx in THUMB_X:
|
||
bx, by = gx + BADGE_OFFSET[0], cy + BADGE_OFFSET[1]
|
||
checked[(gx, cy)] = count_blue_pixels(png_path, bx, by) >= BLUE_THRESHOLD
|
||
return checked
|
||
|
||
|
||
def count_checked_badges(png_path):
|
||
"""统计整屏已打勾（蓝色角标）的卡片数量。
|
||
先用 detect_card_rows 检测实际行位置，再逐行逐列检测打勾，
|
||
避免固定步长与实际行位置不匹配导致漏检。"""
|
||
try:
|
||
rows = detect_card_rows(png_path)
|
||
checked = get_checked_cells_dynamic(png_path, rows)
|
||
return sum(1 for v in checked.values() if v)
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
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 date_label_hash(png_path="/tmp/_datechk.png"):
|
||
"""截图并裁剪出日期条区域（"< 07-23 筛选 >"整条），返回内容 md5。
|
||
只框住日期条，避开上方一直在播放的视频画面，避免视频帧变化造成误判。
|
||
"""
|
||
import hashlib
|
||
adb("shell", "screencap", "-p", "/sdcard/_datechk.png")
|
||
adb("pull", "/sdcard/_datechk.png", png_path)
|
||
with Image.open(png_path) as img:
|
||
crop = img.crop((40, 1080, 440, 1170)) # 400x90+40+1080
|
||
buf = io.BytesIO()
|
||
crop.save(buf, format="PNG")
|
||
return hashlib.md5(buf.getvalue()).hexdigest()
|
||
|
||
|
||
def tap_and_check_date_changed(arrow_coord, timeout=12, interval=1.0):
|
||
"""点一次日期箭头，轮询日期条区域的截图 md5 判断是否变化。
|
||
返回 True=翻页成功，False=日期条未变化（已到边界，如已在今日不能再往后翻）。
|
||
"""
|
||
baseline = date_label_hash()
|
||
adb("shell", "input", "tap", str(arrow_coord[0]), str(arrow_coord[1]))
|
||
for _ in range(int(timeout / interval)):
|
||
time.sleep(interval)
|
||
cur = date_label_hash()
|
||
if cur != baseline:
|
||
return True
|
||
return False
|
||
|
||
|
||
def navigate_to_yesterday_batch(days_back=1):
|
||
"""导航到"今天往前第 days_back 天"的 SD 卡批量下载模式。
|
||
days_back=0 是今天，1 是昨天，2 是前天……（默认 1 保持原行为）。
|
||
关键节点用 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("→ 点击今日缩略图，切换工具栏到下载模式（新装/重启后默认是延时摄影等非下载工具栏）")
|
||
# 这个工具栏是自绘 UI，find_by_text 拿不到"下载"文字节点，不做强校验；
|
||
# 后面点下载按钮那步本身有"未弹出菜单则重试"的兜底。
|
||
assert_activity("VideoPlayActivity", "未在实时视频/录像页")
|
||
adb("shell", "input", "tap", str(COORD["today_thumb"][0]), str(COORD["today_thumb"][1]))
|
||
time.sleep(2)
|
||
|
||
print("→ 归位到今日（App 重启后录像页会停留在上次浏览的日期，不一定是今日，"
|
||
"所以先连续点'下一天'直到到头，再退一天到昨天）")
|
||
assert_activity("VideoPlayActivity", "未在实时视频/录像页")
|
||
for i in range(8):
|
||
if not tap_and_check_date_changed(COORD["next_day_arrow"]):
|
||
print(f" 第 {i+1} 次点'下一天'后日期未变化，已到今日")
|
||
break
|
||
print(f" 第 {i+1} 次点'下一天'，日期已翻页")
|
||
else:
|
||
print(" ⚠ 连续 8 次翻页仍在变化，按当前位置继续")
|
||
|
||
print(f"→ 从今日往前退 {days_back} 天")
|
||
for step in range(days_back):
|
||
if not tap_and_check_date_changed(COORD["prev_day_arrow"]):
|
||
print(f" ! 第 {step+1} 次点'上一天'未检测到日期变化，重试一次")
|
||
if not tap_and_check_date_changed(COORD["prev_day_arrow"]):
|
||
raise RuntimeError(f"点击'上一天'箭头后日期未变化，无法退到第 {days_back} 天")
|
||
# 等日期切换后的录像列表稳定加载
|
||
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 screenshot_hash(png_path="/tmp/_scrollchk.png"):
|
||
"""截图并返回缩略图网格区域内容的 md5，用于判断滚动前后画面是否变化（到顶/到底检测）。
|
||
|
||
不能对整屏截图直接哈希：顶部状态栏时钟每分钟跳动、视频预览区域的"正在下载"提示文字
|
||
等都会随时间变化，跟"列表是否真的滚动了"无关，会导致哈希永远对不上，滚到底/滚到顶
|
||
后无法被正确识别，造成无限重复滚动+点击同一批卡片（实测触发过这个问题）。
|
||
改成只裁剪缩略图网格区域（避开顶部状态栏+视频预览+标题栏），滚不动时该区域内容才会
|
||
真正保持一致。
|
||
"""
|
||
import hashlib
|
||
adb("shell", "screencap", "-p", "/sdcard/_scrollchk.png")
|
||
adb("pull", "/sdcard/_scrollchk.png", png_path)
|
||
with Image.open(png_path) as img:
|
||
crop = img.crop((0, 1000, 1080, 2400)) # 1080x1400+0+1000
|
||
buf = io.BytesIO()
|
||
crop.save(buf, format="PNG")
|
||
return hashlib.md5(buf.getvalue()).hexdigest()
|
||
|
||
|
||
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 get_remote_new_file_count(baseline_files):
|
||
"""统计远程 SD 卡目录里相对 baseline 新增的文件数，用于判断这一批是否真的下载完。"""
|
||
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")
|
||
all_files = set(l for l in result.stdout.strip().splitlines() if l)
|
||
return len(all_files - baseline_files)
|
||
|
||
|
||
def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_round=120):
|
||
"""循环：点击固定网格 9 个卡片（点击即下载）→ 轮询确认下载数量稳定 → 滚动 2 行。
|
||
新版 UI（7.8.1）自绘，dump 拿不到缩略图 bounds，改用固定网格坐标 THUMB_X x THUMB_Y。
|
||
首次点击卡片会弹"使用手机网络下载将消耗流量"确认框，点确定后补点一次该卡片。
|
||
|
||
之前用固定 sleep(20s) 判断"这批下载完了"，如果视频较大/网速慢，20s 内没下完，
|
||
下一轮又新点 9 个，同时在跑的下载数就可能超过 App 的并发上限（10个），超出的点击
|
||
会被静默忽略；等脚本继续往下滚，这些"点了但没真下载"的卡片就再也够不到了。
|
||
现在改成轮询远程目录里新增文件数直到连续两次不再增长，确认上一批真的落盘完成/
|
||
趋于稳定后再滚动，从根上避免超并发。max_wait_per_round 是保险上限，避免个别
|
||
文件异常卡住导致整个流程无限等待。
|
||
|
||
滚动距离也从"整 3 行"改成"2 行"，故意留 1 行重叠：即使滑动手势的实际像素
|
||
距离和行高没有精确对齐，重叠的这一行也保证不会有卡片"从没出现在网格里就被划走"。
|
||
|
||
第二次教训（比上一版更根本）：不能跨"轮"比较截图来判断是否到底。因为点击本身会把
|
||
卡片从"未下载"变成"已下载(打勾)"，这个状态变化本身就会让下一轮开始时的截图跟上一轮
|
||
不一样——即使列表根本没有真的往下滚。正确做法是只在同一轮内、紧贴着"滚动"这一个
|
||
动作的前后各截一次图来判断是否到底，避免掺杂点击造成的状态变化。
|
||
|
||
第三次教训（比前两次更根本，本质原因）：即使改成"同轮内滚动前后比较"，只要画面里有
|
||
一个仍在真实下载的大文件（如几百分钟的超大录像），它的下载进度环/百分比会持续跳动，
|
||
导致任意两次截图之间必然存在像素差异——"是否到底"的信号被这张大文件自身的进度跳动
|
||
淹没，永远检测不到"画面无变化"，脚本因此不断在同一屏重新点击，包括点到已经打勾✓
|
||
完成的卡片，把它们的下载重新触发一遍，造成"重复下载"。
|
||
根本修复：不能只依赖"整屏/局部截图是否变化"这一间接信号，必须直接识别每张卡片
|
||
本身是否已经打勾（已下载完成），点击前逐个跳过已打勾的卡片，只点真正还没下载完的。
|
||
这样无论"是否到底"这个滚动前后哈希判断能不能被大文件干扰失效，都不会再对已完成的
|
||
卡片重复触发下载——滚动哈希判断继续保留作为"提前结束循环"的优化，但不再是防止
|
||
重复下载的唯一防线。
|
||
|
||
第四次教训：打勾检测本身也有盲区——它只能识别"已经 100% 完成"的卡片，识别不了
|
||
"已经点过、正在下载但还没下完"的卡片（这两种状态在打勾位置的像素上是一样的，都不是
|
||
蓝色打勾）。而滚动距离是故意设计成"2 行"、留 1 行重叠的（见上面的说明），这意味着
|
||
上一轮网格里的第 3 行（第 7、8、9 个卡片）滚动后必然变成这一轮的第 1 行——如果这几张
|
||
卡片是大文件、90 秒轮询等不到它下完，打勾检测看到的仍是"未完成"，会被当成全新卡片
|
||
重新点击一次，导致刚点过还在下载中的卡片被重复触发（实测复现："第一次下载了 9 个，
|
||
上滑后又从第 7 个重新下载一遍"）。
|
||
这个重叠关系是滚动距离决定的确定性事实，不需要靠像素判断去猜——所以改成无条件跳过
|
||
每一轮的第 1 行（除了最开始的第 1 轮），因为它就是上一轮已经点过的第 3 行，不管有没有
|
||
下完都不需要再点第二次。
|
||
|
||
第五次教训：以上几种修复都假设"打勾检测读到的是当前真实画面"，但长时间连续点击/
|
||
截图会偶尔把 App 拖到卡死(ANR)，卡死期间截图读到的是冻结前的旧画面——打勾检测会把
|
||
"其实已经点过/正在下载"的卡片误判为"未下载"，导致反复重新触发下载。这不是某一行
|
||
的重叠问题，是全局性的：实测单日 220 个下载里只有 11 种真实内容，最严重一组被重复
|
||
下载了 28 次。dismiss_anr_if_any() 之前只在导航阶段被调用，批量下载循环里完全没
|
||
调用，是这个问题的直接缺口。修复：在每轮点击前、以及轮询下载数量期间都调用
|
||
ensure_not_anr()，确认 App 没卡死再继续；卡死超时仍未恢复则直接报错停止，而不是
|
||
在冻结画面上继续盲点制造更多重复。
|
||
|
||
第六次教训：第四次教训里"无条件跳过每轮第1行"这个修复本身还有个漏洞——它假设
|
||
"每一行永远是等高的3张卡片"，但实际列表里穿插着"18:00""19:00"这类整点分组标签，
|
||
不属于卡片网格、却会额外占竖直空间。一旦某次滚动跨越了这种标签，实际卡片内容
|
||
相对固定网格坐标就会错位，"这一轮第1行 = 上一轮第3行"这个前提就不成立了——但
|
||
脚本还是死板地把第1行整行跳过，结果是把一整行从没点过的新卡片当成重叠行漏掉
|
||
（实测复现：截图里一整行卡片既没打勾也没在下载）。
|
||
根本修复：不能再靠"位置固定跳过第1行"这种盲猜，改成 cell_fingerprint() 逐张
|
||
卡片内容比对——只有指纹跟上一轮保存的最后一行指纹完全一致的卡片才是真重叠，
|
||
跳过；不一致就说明是错位带来的新内容，必须正常处理（判断打勾/点击），不能再
|
||
整行无脑跳过。
|
||
|
||
第七次教训：第六次教训里指纹比对方案刚实现时，取"上一轮最后一行"指纹用的是
|
||
本轮点击前的截图（grid_shot），而不是点击+等待下载稳定之后的截图。同一张卡片
|
||
点击前是"未下载"的鲜艳缩略图，点击后会变成"下载中"的暗色调，两种状态的指纹
|
||
永远对不上，导致"跳过真重叠"实测恒为 0（round 12~31 全部是 0），等于指纹比对
|
||
完全没生效，每轮都把上一轮点过的最后一行又点了一遍，比修复前更频繁地重复下载。
|
||
修复：把指纹计算挪到本轮点击+轮询结束、滚动前的那一刻，用 pre_scroll_hash 已经
|
||
截好的同一张图算指纹，确保两轮比较的都是同一种视觉状态（都是"这一刻画面"，
|
||
而不是"点击前 vs 点击后"两种状态）。
|
||
|
||
第八次教训：第六/七次教训修好时序后仍有残留重复——实测相册里65个新文件有8对
|
||
(16个)内容重复，重复文件的时间间隔都稳定在34~45秒（正好是"一轮点击+等待轮询"
|
||
的周期），说明还是"重叠行"没识别出来。根因：指纹比对只查"这一轮第1行"是否
|
||
等于"上一轮最后一行"，隐含假设"每次滚动固定重叠1行"；但 input swipe 模拟的
|
||
是惯性滚动，实际滚动距离会随惯性、是否跨过整点分组标签而浮动，重叠有时是
|
||
1行、有时是2行。于是又改成本轮全部9格跟上一轮全部9格做指纹比对——但这版依然
|
||
"跳过真重叠恒为0"，实测53个新文件仍有13对重复。
|
||
|
||
第九次教训（推翻第六~八次的整条指纹路线，最终方案）：像素指纹从根上就走不通。
|
||
真正原因是下载列表里卡片缩略图的视觉状态会变：未下载时显示的是"灰色占位图"，
|
||
下载完成后才替换成"彩色实际画面"（截图实证）。所以同一张卡片"滚动前看到是灰
|
||
占位、滚动回来时已下完变成彩色"，缩略图整块都不同，任何基于像素的 MD5 指纹都
|
||
必然对不上 → 恒为0。左上角时间戳文字虽稳定，但白字压在亮背景上 OCR 不可靠，且
|
||
"同分钟同时长"不唯一。
|
||
彻底换思路：重复的唯一来源，是滚动留的重叠行滚上来时"上一轮刚点的卡片还没下完、
|
||
没显示打勾"，导致 badge 检测（本能跳过打勾卡片）拦不住、被当新卡片重点。那就
|
||
从源头堵：每轮点击后，等到"这一轮点击的卡片全部显示打勾（下载完成）"再滚动。
|
||
这样滚动后的重叠行必然是打勾状态，下一轮开头 get_checked_cells 天然跳过它们——
|
||
不需要指纹、不需要 OCR，纯确定性。个别卡片下载失败/超时一直不打勾，则设上限
|
||
超时后照常滚动（那几张可能重复一次，靠下游 analyze.py 的 MD5 查重兜底）。
|
||
|
||
第十次教训（找到贯穿始终的总根源，本次最终方案）：第九次"等打勾再滚"实测重复率
|
||
仍约 30%，且用户亲眼看到"已打勾的又下一遍"。深挖发现：打勾检测和点击都建立在
|
||
写死的网格坐标 THUMB_Y=[1148,1351,1554] 上，但这个列表滚动时【不 snap 对齐】，
|
||
而且每隔一段插入"整点分组标签"行（如"21:00"），它占竖直空间却不属于卡片网格。
|
||
一旦滚动跨过标签，所有卡片相对固定 THUMB_Y 整行平移（实测漂移达 174px，接近一整
|
||
行）。平移后：点击落在标签/错位卡片上、蓝勾采样点对不准（把已下载误读成未下载），
|
||
于是重复点击。前面 ANR/指纹/时机/等打勾所有修复，都建在这个"坐标会错位"的地基
|
||
上打转，治标不治本。
|
||
根本修复：不再写死行坐标，改成每屏用亮度标准差(std)垂直剖面【动态实测】卡片行的
|
||
真实中心 y（detect_card_rows）——卡片缩略图有纹理 std 高、间隙/分组标签/灰占位是
|
||
纯色 std≈0，据此切出纹理带，带中心即行中心。横向 3 列 THUMB_X 不漂移予以保留。
|
||
坐标动态对准后，第九次教训的"等打勾跳重叠"机制才真正生效。三张不同滚动位置的真实
|
||
截图验证：检出行中心精准落在卡片中心、分组标签处不产生行。THUMB_Y 与旧
|
||
get_checked_cells 就此废弃（仅保留备查）。
|
||
|
||
第十一次教训（第十次动态定位落地后暴露的两个到底部边界 bug）：
|
||
(1) 每轮限点前 3 行以避免超并发，但列表【到底】时滚不动，被限制漏在第 4+ 行的
|
||
底部未下载卡片就永远轮不到处理，漏下最后几个（实测底部 00:xx 那几个未下载）。
|
||
(2) 到底检测原用整屏像素哈希，但底部未下载卡片是"灰色占位图"，滚动时会微微
|
||
重渲染/回弹，哈希每次都不同，永远判不出"滚不动"，一直空滚（实测空转到 216 轮）。
|
||
修复：到底检测改用"卡片行有没有整体平移"（滚不动=行中位数位移<40px，抗底部占位
|
||
抖动，比像素哈希可靠）；判到底后不直接结束，先【清扫】当前屏所有检出行、分批
|
||
（每批≤9 不超并发）点未下载的并等打勾，把被限 3 行漏下的底部卡片补下完再 break。
|
||
实测：底部 6 个漏下卡片被清扫补下（相册 +6），到底判定位移=0 正确 break 不再空滚。
|
||
"""
|
||
print("→ 先滚动到列表顶部（起始滚动位置继承自上次播放位置，不一定是最新记录）")
|
||
top_hash = None
|
||
for i in range(15):
|
||
adb("shell", "input", "swipe", "540", "1150", "540", "2350", "600")
|
||
time.sleep(1.3)
|
||
cur_hash = screenshot_hash()
|
||
if cur_hash == top_hash:
|
||
print(f" 已到顶部（第 {i+1} 次滑动后画面不再变化）")
|
||
break
|
||
top_hash = cur_hash
|
||
else:
|
||
print(" ⚠ 滑动 15 次仍未稳定，按当前位置继续（可能列表过长）")
|
||
|
||
print(f"→ 开始批量下载循环（最多 {max_rounds} 轮）")
|
||
dialog_dismissed = False
|
||
|
||
for r in range(max_rounds):
|
||
# 点击前先确认 App 没有卡死：卡死时截图读到的是冻结前的旧画面，
|
||
# 打勾识别会失真，是"重复下载"的真正根因（见 ensure_not_anr 文档字符串）。
|
||
ensure_not_anr()
|
||
|
||
# 截图前等一下，让"灰色占位"缩略图加载出真实画面——否则纯色占位 std≈0
|
||
# 会被 detect_card_rows 当成间隙，导致整行漏检（见 THUMB_LOAD_WAIT 说明）。
|
||
time.sleep(THUMB_LOAD_WAIT)
|
||
|
||
# 每屏动态实测卡片行中心 y（消除整点分组标签平移造成的错位，这是"重复下载"
|
||
# 的总根源），再对每行 × 固定 3 列逐格判打勾，只点还没下载完的。
|
||
grid_shot = screenshot("/tmp/_gridchk.png")
|
||
full_rows = detect_card_rows(grid_shot)
|
||
# 检测不到行时重试最多3次：缩略图加载慢、ANR刚恢复、画面切换中都会导致
|
||
# 临时检测不到行。之前只重试1次就放弃，会误判"全部已打勾"直接滚动，
|
||
# 漏掉当前屏所有卡片（这是漏下载的重要原因之一）。
|
||
retry = 0
|
||
while not full_rows and retry < 3:
|
||
retry += 1
|
||
time.sleep(2.0 + retry * 1.0) # 逐次加长等待，给缩略图加载留时间
|
||
grid_shot = screenshot("/tmp/_gridchk.png")
|
||
full_rows = detect_card_rows(grid_shot)
|
||
if not full_rows:
|
||
print(f" 轮 {r+1}: ⚠ 连续 {retry+1} 次未检出卡片行，跳过本轮滚动（不盲点）")
|
||
continue
|
||
# 每轮最多处理最上面 3 行（9 格）。萤石同时下载上限约 10 个，一屏可见 5~6 行，
|
||
# 若把检出的所有行都点了（12~14 个）会超并发：多点的卡片不进下载队列、滚上来
|
||
# 仍未打勾，被反复重点，实测会让"跳过已打勾"退化成 0、重复率回升。多出的行
|
||
# 下一轮滚上来再处理（滚动量≈2 行，正好衔接），不会漏。full_rows 保留全部行，
|
||
# 供到底判定和到底清扫复用（省一次滚动前的重复截图+检测）。
|
||
row_centers = full_rows # 处理所有检测到的行，下面按未打勾数截断到9个
|
||
checked = get_checked_cells_dynamic(grid_shot, row_centers)
|
||
|
||
# 跳过已经打勾（下载完成）的卡片。滚动留的 1~2 行重叠滚上来时必然已是打勾
|
||
# 状态（本轮末尾"等全部点击的卡片打勾再滚动"保证），所以重叠行天然被跳过，
|
||
# 不会被当成新卡片重复触发下载——现在行中心是动态对准的，这个跳过才可靠。
|
||
badge_skipped = sum(1 for cy in row_centers for gx in THUMB_X if checked.get((gx, cy)))
|
||
# 收集所有未打勾格子（从上到下），最多9个避免超并发
|
||
todo_cells = [(gx, cy) for cy in row_centers for gx in THUMB_X if not checked.get((gx, cy))]
|
||
tapped_cells = todo_cells[:6] # 萤石并发上限10，留余量点6个
|
||
for (gx, cy) in tapped_cells:
|
||
adb("shell", "input", "tap", str(gx), str(cy))
|
||
time.sleep(0.5)
|
||
if not dialog_dismissed and dismiss_dialog_if_any():
|
||
dialog_dismissed = True
|
||
adb("shell", "input", "tap", str(gx), str(cy))
|
||
time.sleep(0.5)
|
||
|
||
print(f" 轮 {r+1}: 检出 {len(row_centers)} 行，跳过已打勾 {badge_skipped} 个，点击待下载 {len(tapped_cells)} 个")
|
||
|
||
if tapped_cells:
|
||
# 关键：等这一轮点击的卡片全部显示打勾（下载完成）再滚动。这样滚动
|
||
# 留下的重叠行必然是打勾状态，下一轮开头 get_checked_cells 会跳过它们，
|
||
# 从根上消除"重叠行还没下完就被重复点击"这个唯一的重复来源。
|
||
baseline_checked = count_checked_badges(grid_shot)
|
||
target_checked = baseline_checked + len(tapped_cells)
|
||
print(f" 轮 {r+1}: 等待刚点击的 {len(tapped_cells)} 个卡片全部打勾（基线 {baseline_checked}，目标 {target_checked}）...")
|
||
waited = 0
|
||
done = 0
|
||
while waited < max_wait_per_round:
|
||
time.sleep(poll_interval)
|
||
waited += poll_interval
|
||
ensure_not_anr()
|
||
wait_shot = screenshot("/tmp/_waitchk.png")
|
||
current_checked = count_checked_badges(wait_shot)
|
||
done = current_checked - baseline_checked
|
||
if done >= len(tapped_cells):
|
||
break
|
||
pending = len(tapped_cells) - done
|
||
total = get_remote_new_file_count(baseline)
|
||
if pending:
|
||
# 第二轮等待：不重试点击，让正在下载的完成
|
||
print(f" 轮 {r+1}: 等待 {waited}s 后 {done}/{len(tapped_cells)} 个已打勾，继续等待...")
|
||
wait2_waited = 0
|
||
while wait2_waited < max_wait_per_round:
|
||
time.sleep(poll_interval)
|
||
wait2_waited += poll_interval
|
||
ensure_not_anr()
|
||
w2_shot = screenshot("/tmp/_wait2.png")
|
||
current2 = count_checked_badges(w2_shot)
|
||
done = current2 - baseline_checked
|
||
if done >= len(tapped_cells):
|
||
break
|
||
pending = len(tapped_cells) - done
|
||
if pending:
|
||
# 两轮等待后仍未完成，重试点击所有未完成的卡片
|
||
print(f" 轮 {r+1}: 再等 {wait2_waited}s 后仍有 {pending} 个未完成，重试点击...")
|
||
for (gx, cy) in tapped_cells:
|
||
adb("shell", "input", "tap", str(gx), str(cy))
|
||
time.sleep(0.5)
|
||
retry_waited = 0
|
||
while retry_waited < max_wait_per_round:
|
||
time.sleep(poll_interval)
|
||
retry_waited += poll_interval
|
||
ensure_not_anr()
|
||
retry_shot = screenshot("/tmp/_retrywait.png")
|
||
current3 = count_checked_badges(retry_shot)
|
||
done = current3 - baseline_checked
|
||
if done >= len(tapped_cells):
|
||
break
|
||
pending = len(tapped_cells) - done
|
||
if pending:
|
||
print(f" 轮 {r+1}: 三轮等待后仍有 {pending} 个未完成，照常滚动（靠下游去重兜底）")
|
||
else:
|
||
print(f" 轮 {r+1}: 全部打勾完成")
|
||
else:
|
||
print(f" 轮 {r+1}: {len(tapped_cells)} 个全部打勾完成（等待 {waited}s），累计新下载 {total} 个，开始滚动")
|
||
else:
|
||
print(f" 轮 {r+1}: 这一屏全部已打勾，无需等待，直接滚动")
|
||
|
||
# 滚动前的行位置直接复用本轮开头检出的 full_rows（点击+等打勾期间列表没滚动，
|
||
# 行位置不变），省掉一次滚动前的重复截图+检测。
|
||
# 到底检测不能用整屏像素哈希：列表底部有未下载的"灰色占位图"，滚动时会
|
||
# 微微重渲染/回弹，像素哈希每次都不同，永远判不出"滚不动"，导致一直空滚
|
||
# （实测跑到 216 轮还在底部空转）。改用"卡片行有没有整体平移"：真滚动了行会
|
||
# 整体位移几百像素，滚不动（到底）时行位置几乎不变。
|
||
pre_rows = full_rows
|
||
|
||
# 向下滚动 2 行（约 406 像素，保留 1 行重叠避免漏跳）
|
||
adb("shell", "input", "swipe", "540", "1900", "540", "700", "500")
|
||
time.sleep(1.0)
|
||
|
||
post_shot = screenshot("/tmp/_postscroll.png")
|
||
post_rows = detect_card_rows(post_shot)
|
||
|
||
def _median(vals):
|
||
return sorted(vals)[len(vals) // 2] if vals else -1
|
||
def _moved(a, b):
|
||
return bool(a and b and abs(_median(a) - _median(b)) >= 40)
|
||
|
||
if not _moved(pre_rows, post_rows):
|
||
# 疑似到底，但一次没滚动也可能是偶发滚动手势失效 / 下载中大文件卡顿吸收了
|
||
# 手势（实测 26、27 号就因单次误判、提前结束、漏下几百个）。再滚一次确认：
|
||
# 只有连续两次都没推进才判真到底。
|
||
time.sleep(2)
|
||
adb("shell", "input", "swipe", "540", "1900", "540", "700", "500")
|
||
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}: 连续两次滚动都未推进，确认到达列表底部，清扫当前屏剩余")
|
||
for sweep in range(6): # 最多 6 批，防意外死循环
|
||
ensure_not_anr()
|
||
time.sleep(THUMB_LOAD_WAIT)
|
||
sshot = screenshot("/tmp/_sweep.png")
|
||
srows = detect_card_rows(sshot)
|
||
# 清扫时检测不到行要重试：底部缩略图可能还在加载，直接判"无未下载"
|
||
# 会提前结束、漏掉底部最后几个卡片。
|
||
sweep_retry = 0
|
||
while not srows and sweep_retry < 3:
|
||
sweep_retry += 1
|
||
time.sleep(2.0)
|
||
sshot = screenshot("/tmp/_sweep.png")
|
||
srows = detect_card_rows(sshot)
|
||
if not srows:
|
||
print(f" → 清扫第 {sweep+1} 批：连续 {sweep_retry+1} 次未检出卡片行，视为底部已无内容")
|
||
break
|
||
schk = get_checked_cells_dynamic(sshot, srows)
|
||
todo = [(gx, cy) for cy in srows for gx in THUMB_X if not schk.get((gx, cy))]
|
||
if not todo:
|
||
print(" → 底部已无未下载卡片，清扫完成")
|
||
break
|
||
batch = todo[:9]
|
||
print(f" → 清扫第 {sweep+1} 批：点击 {len(batch)} 个未下载卡片")
|
||
for (gx, cy) in batch:
|
||
adb("shell", "input", "tap", str(gx), str(cy))
|
||
time.sleep(0.5)
|
||
if not dialog_dismissed and dismiss_dialog_if_any():
|
||
dialog_dismissed = True
|
||
adb("shell", "input", "tap", str(gx), str(cy))
|
||
time.sleep(0.5)
|
||
# 等这批全部打勾（同主循环机制）
|
||
waited = 0
|
||
pend = list(batch)
|
||
while waited < max_wait_per_round:
|
||
time.sleep(poll_interval)
|
||
waited += poll_interval
|
||
ensure_not_anr()
|
||
wshot = screenshot("/tmp/_sweepwait.png")
|
||
wchk = get_checked_cells_dynamic(wshot, srows)
|
||
pend = [c for c in batch if not wchk.get(c)]
|
||
if not pend:
|
||
break
|
||
if pend:
|
||
print(f" → 清扫批 {sweep+1} 有 {len(pend)} 个超时未完成，靠下游去重兜底")
|
||
break
|
||
|
||
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, date_str=None):
|
||
"""把下载好的文件 adb pull 到本地。
|
||
date_str 为录像日期（YYYYMMDD），归档到 recordings/BH5455783/{date_str}/。
|
||
不传则默认昨天（保持原行为）。"""
|
||
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)
|
||
|
||
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 get_all_remote_files():
|
||
"""获取设备上今天下载目录里的全部录像文件（不做 baseline 差集）。
|
||
|
||
最终拉取阶段必须用这个函数，不能只拉"相对 baseline 新增"的文件：如果之前
|
||
某次运行中途跑崩/被 kill，它已经下载到设备本地的文件会被下一次运行的新
|
||
baseline 当成"运行前就存在"，永远不会被视为新增，导致这些早就下载好的文件
|
||
永远拉不回 Mac。pull_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 只用于判断"这一轮点击触发的下载是否已经稳定"，不再限制最终拉取范围
|
||
baseline = get_baseline_files()
|
||
print(f"→ 基准文件数: {len(baseline)}")
|
||
navigate_to_yesterday_batch()
|
||
tap_all_thumbnails(baseline) # 默认 max_rounds=300，跑到滚到底为止
|
||
wait_for_downloads(baseline)
|
||
|
||
all_files = get_all_remote_files()
|
||
print(f"→ 设备上共有 {len(all_files)} 个录像文件（含之前跑崩残留的历史文件），开始全部拉取")
|
||
if all_files:
|
||
pull_files(list(all_files))
|
||
else:
|
||
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} 天）=====")
|
||
for attempt in range(2): # 每天最多重试1次（共2次尝试），避免偶发ANR/网络问题导致整天漏下载
|
||
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() # 拉完即时清空，释放空间给下一天
|
||
break # 成功则跳出重试循环
|
||
except Exception as e:
|
||
print(f"⚠ {rec_date} 第 {attempt+1} 次尝试失败：{e}")
|
||
if attempt == 0:
|
||
print(f" 重试一次...")
|
||
time.sleep(5)
|
||
else:
|
||
print(f" {rec_date} 两次尝试均失败，跳过该天")
|
||
print(f"\n===== 最近 {n_days} 天下载完成 =====")
|
||
|
||
|
||
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()
|