自动跳过片头曲：视频指纹与片段级重复检测的实战

import cv2
import numpy as np


def dhash(frame, hash_size=8):
    """计算一帧的 dHash（64 bit）。"""
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) if frame.ndim == 3 else frame
    # 缩放到 (hash_size+1) x hash_size，用 INTER_AREA 减少噪点影响
    small = cv2.resize(gray, (hash_size + 1, hash_size), interpolation=cv2.INTER_AREA)
    # 相邻像素比较
    diff = small[:, 1:] > small[:, :-1]
    # 打包成 64 位整数
    value = 0
    for bit in diff.flatten():
        value = (value << 1) | int(bit)
    return value


def hamming(a: int, b: int) -> int:
    """汉明距离（不同位的个数）。"""
    return bin(a ^ b).count('1')

视频 A 的指纹：
t=0s   → 0x3c8a...
t=1s   → 0x3c8a...
t=2s   → 0x91f2...     ← 场景切换
...

import subprocess
import numpy as np


def extract_fingerprint(path: str, fps_sample: float = 1.0, width: int = 160):
    """抽取视频指纹：返回 (时间戳列表, dHash 列表)。"""
    cmd = [
        'ffmpeg', '-i', path,
        '-vf', f'fps={fps_sample},scale={width}:-2',
        '-f', 'rawvideo', '-pix_fmt', 'bgr24', '-an', '-',
    ]
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                            stderr=subprocess.DEVNULL)
    # 先读一帧确定尺寸
    hashes, stamps = [], []
    idx = 0
    # 逐块读（宽度可能因为 -2 调整，先探测）
    buf = b''
    H = None
    while True:
        chunk = proc.stdout.read(1 << 20)
        if not chunk:
            break
        buf += chunk
        if H is None:
            # 从 ffmpeg 的 stderr 拿尺寸比较麻烦，这里用另一种方式：先 probe
            pass
    ...

import json
import subprocess


def probe_size(path):
    out = subprocess.run(
        ['ffprobe', '-v', 'error', '-select_streams', 'v:0',
         '-show_entries', 'stream=width,height', '-of', 'json', path],
        capture_output=True, text=True, check=True).stdout
    s = json.loads(out)['streams'][0]
    return s['width'], s['height']


def extract_fingerprint(path, sample_fps=1.0, width=160):
    w0, h0 = probe_size(path)
    h = int(round(width * h0 / w0))
    h -= h % 2                                  # 保证偶数
    frame_bytes = width * h * 3

    cmd = ['ffmpeg', '-v', 'error', '-i', path,
           '-vf', f'fps={sample_fps},scale={width}:{h}',
           '-f', 'rawvideo', '-pix_fmt', 'bgr24', '-an', '-']
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)

    hashes, stamps = [], []
    idx = 0
    while True:
        raw = proc.stdout.read(frame_bytes)
        if len(raw) != frame_bytes:
            break
        frame = np.frombuffer(raw, np.uint8).reshape(h, width, 3)
        hashes.append(dhash(frame))
        stamps.append(idx / sample_fps)
        idx += 1
    proc.stdout.close()
    proc.wait()
    return np.array(stamps), np.array(hashes, dtype=np.uint64)

def match_windows(a: np.ndarray, b: np.ndarray, window: int = 30,
                  max_dist: int = 8, max_bad_ratio: float = 0.2):
    """找出 A 和 B 之间重复长度 >= window 的片段。
    返回 [(a_idx, b_idx, length)]。"""
    na, nb = len(a), len(b)
    matches = []
    for i in range(na - window):
        for j in range(nb - window):
            dists = np.array([hamming(a[i + k], b[j + k]) for k in range(window)])
            bad = (dists > max_dist).mean()
            if bad <= max_bad_ratio:
                matches.append((i, j, window))
    return matches

def hamming_matrix(a: np.ndarray, b: np.ndarray) -> np.ndarray:
    """计算 a 和 b 所有两两组合的汉明距离矩阵（a: N, b: M）。"""
    # broadcast XOR，然后数每个数的 1 的个数
    xor = a[:, None] ^ b[None, :]
    # 用查表法数 bit（比 bin().count() 快得多）
    return popcount64(xor)


_POP16 = np.array([bin(i).count('1') for i in range(1 << 16)], dtype=np.uint8)


def popcount64(x: np.ndarray) -> np.ndarray:
    """64 位整数的 popcount（分 4 段 16 位查表）。"""
    out = np.zeros(x.shape, dtype=np.uint8)
    for shift in range(0, 64, 16):
        out += _POP16[(x >> shift) & 0xFFFF]
    return out

def find_candidates(a, b, max_dist=8):
    """找出所有 a[i] ≈ b[j] 的位置对（首帧匹配）。"""
    d = hamming_matrix(a, b)
    ii, jj = np.where(d <= max_dist)
    return list(zip(ii.tolist(), jj.tolist()))

def extend_diagonal(a, b, i, j, max_dist=8, max_bad_ratio=0.2, min_len=15):
    """从 (i,j) 开始沿对角线延伸，返回匹配长度。"""
    length = 0
    bad = 0
    while i + length < len(a) and j + length < len(b):
        d = hamming(a[i + length], b[j + length])
        if d > max_dist:
            bad += 1
            if bad / (length + 1) > max_bad_ratio:
                break
        length += 1
    return length if length >= min_len else 0


def find_repeats(a, b, max_dist=8, min_len=15):
    """找所有重复片段，返回 [(a_start, b_start, length)]。"""
    results = []
    used_i = set()
    for i, j in find_candidates(a, b, max_dist):
        if i in used_i:
            continue
        L = extend_diagonal(a, b, i, j, max_dist=max_dist, min_len=min_len)
        if L:
            results.append((i, j, L))
            used_i.update(range(i, i + L))       # 避免重复报告
    return results

from collections import defaultdict


def build_index(fingerprints: dict):
    """fingerprints: {video_id: np.array(hashes)}"""
    index = defaultdict(list)
    for vid, hashes in fingerprints.items():
        for pos, h in enumerate(hashes):
            index[h].append((vid, pos))
    return index


def candidate_pairs(index, fingerprints, min_shared=5):
    """找出至少共享 min_shared 个哈希的视频对。"""
    counter = defaultdict(int)
    for h, entries in index.items():
        vids = {vid for vid, _ in entries}
        if len(vids) < 2:
            continue
        vids = sorted(vids)
        for i in range(len(vids)):
            for j in range(i + 1, len(vids)):
                counter[(vids[i], vids[j])] += 1
    return {pair: c for pair, c in counter.items() if c >= min_shared}

# 安装（Debian/Ubuntu）
apt-get install libchromaprint-tools

# 生成指纹（输出 JSON，含指纹字符串和时长）
fpcalc -json -length 120 input.mp4 > fp.json

{
  "duration": 120,
  "fingerprint": "AQAAcwmSRImSRYmSRYmSRYmSRImSRYmSRYmSRImSRYmSRYmSRYmSRIm..."
}

import base64
import struct


def decode_chromaprint(fp_b64: str):
    """把 chromaprint 的 base64 指纹解成整数列表。"""
    raw = base64.b64decode(fp_b64)
    n = len(raw) // 4
    return list(struct.unpack(f'<{n}I', raw[:n * 4]))


def find_audio_match(fp_a: list, fp_b: list, min_len: int = 20):
    """找 A 的指纹在 B 中的最长精确匹配。"""
    set_b = {}
    for j, v in enumerate(fp_b):
        set_b.setdefault(v, []).append(j)

    best = None
    for i, v in enumerate(fp_a):
        for j in set_b.get(v, []):
            # 沿对角线延伸
            k = 0
            while (i + k < len(fp_a) and j + k < len(fp_b)
                   and fp_a[i + k] == fp_b[j + k]):
                k += 1
            if k >= min_len and (best is None or k > best[2]):
                best = (i, j, k)
    return best

1. 优先用音频指纹（快、准）→ 定位片头曲
2. 音频指纹失败（静音片头、无音频）→ 回退到视频哈希
3. 两者都有结果时交叉验证（时间戳应该接近）

def detect_intro(results, total_episodes, min_episodes=3, max_start=180):
    """从所有比对结果里找出片头。
    results: [(vid_a, a_start, vid_b, b_start, length)]
    """
    # 统计"某个内容片段出现在多少集的开头附近"
    clusters = defaultdict(list)
    for vid_a, a_start, vid_b, b_start, length in results:
        if a_start <= max_start:
            clusters[vid_a].append((a_start, length))
        if b_start <= max_start:
            clusters[vid_b].append((b_start, length))

    intro = {}
    for vid, spans in clusters.items():
        if len(spans) + 1 < min_episodes:
            continue
        # 取出现次数最多的起点/长度（多数投票）
        starts = [s for s, _ in spans]
        lengths = [l for _, l in spans]
        intro[vid] = (median(starts), median(lengths))
    return intro

# 检测黑边
ffmpeg -i input.mp4 -vf cropdetect=24:16:0 -t 60 -f null - 2>&1 | grep -o 'crop=[0-9:]*' | tail -1
# crop=1920:1080:0:0

# 生成指纹时先裁
ffmpeg -i input.mp4 -vf "crop=1920:1080:0:0,fps=1,scale=160:-2" ...

class FingerprintIndex:
    def __init__(self):
        self.fps = {}          # video_id -> (stamps, hashes)

    def add(self, video_id, path):
        stamps, hashes = extract_fingerprint(path, sample_fps=1.0)
        self.fps[video_id] = (stamps, hashes)

    def find_repeats_between(self, a_id, b_id, **kw):
        (sa, ha), (sb, hb) = self.fps[a_id], self.fps[b_id]
        return find_repeats(ha, hb, **kw)

    def detect_intro_outro(self, min_episodes=3):
        # 遍历候选对 → 收集结果 → 聚类
        ...