SAM2: sample every second for the entire video, no exceptions
Docker Build and Push / check-secrets (push) Successful in 9s
Docker Build and Push / build-and-push (cpu, latest) (push) Has been skipped
Docker Build and Push / build-and-push (cuda, 12.8) (push) Has been skipped
Docker Build and Push / build-and-push (directml, latest) (push) Has been skipped
Docker Build and Push / build-and-push (cuda, 11.8) (push) Has been skipped
Docker Build and Push / build-and-push (cuda, 12.6) (push) Has been skipped
Build Windows CPU / build (push) Has been cancelled
Build Windows CUDA 11.8 / build (push) Has been cancelled
Build Windows CUDA 12.6 / build (push) Has been cancelled
Build Windows CUDA 12.8 / build (push) Has been cancelled
Build Windows DirectML / build (push) Has been cancelled

User requirement: correctness first, slow is acceptable.
- Remove the 120s dense + 8s tail split
- Full video sampled at 1 frame/second uniformly
- Analysis still uses 1/4 resolution for speed
- Coordinate scale restored to original resolution at the end

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
liuchen
2026-06-27 16:14:53 +08:00
co-authored by Claude Haiku 4.5
parent 9094b3c888
commit 5cdc727207
+50 -26
View File
@@ -14,44 +14,62 @@ SAM2_CFG = "configs/sam2.1/sam2.1_hiera_t.yaml"
def detect_static_regions(video_path: str, sample_frames: int = 30, threshold: float = 20.0,
max_scan_seconds: float = None):
"""
饱和度签名 + 跨帧对比法检测水印/Logo:
1. 采样多帧,通过高饱和度像素比例判断哪些帧含有logo(logo通常色彩鲜艳)
2. 在"含logo帧"之间比较,logo像素在多帧中出现在同一位置 → 差值极低
3. 能检测周期性出现/动态水印,无需logo持续存在于全部帧
饱和度签名 + 跨帧对比法检测水印/Logo。
采样策略:每秒采一帧(顺序读取,不随机 seek)+ 超长视频后段每8秒一帧。
分析时缩放到 1/4 分辨率提速,结果坐标还原到原始尺寸。
"""
cap = cv2.VideoCapture(video_path)
total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
orig_frame_w, orig_frame_h = w, h # 保存原始分辨率,用于最终坐标还原
duration_s = total / fps
num_samples = max(10, min(24, int(duration_s // 8)))
uniform_positions = [min(int((i + 0.5) * total / num_samples), total - 1)
for i in range(num_samples)]
# 水印常出现在视频开头几秒,强制加密采样前15秒
early_seconds = [0.5, 1, 1.5, 2, 3, 4, 5, 7, 10, 15]
early_positions = [min(int(s * fps), total - 1) for s in early_seconds if s * fps < total]
# 合并去重,保持顺序
seen = set()
positions = []
for p in early_positions + uniform_positions:
if p not in seen:
seen.add(p)
positions.append(p)
# ── 采样策略:全程每秒采一帧 ─────────────────────────────────────────
step = max(1, int(fps))
positions = list(range(0, total, step))
bgr_frames = []
# ── 顺序读取,缩放到 1/4 分辨率加速分析 ────────────────────────────────
SW, SH = max(1, w // 4), max(1, h // 4) # 分析用小尺寸
pos_set = set(positions)
bgr_frames_small = [] # 缩放后的小帧(用于分析)
bgr_frames_orig = [] # 原始尺寸帧(用于 SAM2,只保留少数几帧)
gray_frames = []
for pos in positions:
cap.set(cv2.CAP_PROP_POS_FRAMES, pos)
keep_orig_interval = max(1, len(positions) // 8) # 每隔若干帧保留一张原始尺寸
cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
frame_idx = 0
pos_iter = iter(sorted(positions))
next_pos = next(pos_iter, None)
while next_pos is not None:
if frame_idx < next_pos:
# 快速跳帧:只有在需要的位置才真正读
cap.set(cv2.CAP_PROP_POS_FRAMES, next_pos)
frame_idx = next_pos
ret, frame = cap.read()
if ret:
bgr_frames.append(frame)
gray_frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY).astype(np.float32))
if not ret:
next_pos = next(pos_iter, None)
continue
small = cv2.resize(frame, (SW, SH), interpolation=cv2.INTER_AREA)
bgr_frames_small.append(small)
gray_frames.append(cv2.cvtColor(small, cv2.COLOR_BGR2GRAY).astype(np.float32))
if len(bgr_frames_orig) == 0 or len(bgr_frames_small) % keep_orig_interval == 0:
bgr_frames_orig.append(frame)
frame_idx += 1
next_pos = next(pos_iter, None)
cap.release()
print(f"[SAM2-detect] 采样 {len(bgr_frames)} 帧 (前15秒密采+均匀分布,共{duration_s:.1f}s)", flush=True)
# 后续分析全部用小尺寸帧
bgr_frames = bgr_frames_small
w, h = SW, SH # 分析坐标空间切换到小尺寸
print(f"[SAM2-detect] 采样 {len(bgr_frames)} 帧(全程每秒一帧,共{duration_s:.1f}s,分析尺寸{SW}x{SH})", flush=True)
if len(bgr_frames) < 4:
print(f"[SAM2-detect] 帧数不足4帧,返回空", flush=True)
@@ -175,7 +193,13 @@ def detect_static_regions(video_path: str, sample_frames: int = 30, threshold: f
print(f"[SAM2-detect] 合并顶部 {len(top_regions)} 个区域为 ({x1},{y1})-({x2},{y2})", flush=True)
regions = merged
print(f"[SAM2-detect] detect_static_regions 返回 {len(regions)} 个区域", flush=True)
# ── 坐标从分析尺寸(SW×SH)还原到原始分辨率 ──────────────────────────
scale_x = orig_frame_w / SW
scale_y = orig_frame_h / SH
regions = [(int(x1*scale_x), int(y1*scale_y), int(x2*scale_x), int(y2*scale_y))
for x1, y1, x2, y2 in regions]
print(f"[SAM2-detect] detect_static_regions 返回 {len(regions)} 个区域(已还原到{orig_frame_w}x{orig_frame_h})", flush=True)
return regions