File size: 3,955 Bytes
3e90852 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import json
import os
import numpy as np
from tqdm import tqdm
from concurrent.futures import ProcessPoolExecutor, as_completed
import decord
from decord import VideoReader, cpu
# 屏蔽底层烦人的 C++ 报错输出
# decord.bridge.set_bridge('torch')
import logging
decord.logging.set_level(logging.ERROR)
# 🌟 模拟你修复后的训练配置
class DataArgs:
def __init__(self):
self.video_fps = 1
self.frames_upbound = 16 # 保持和你训练时设置的上限一致
self.force_sample = True
data_args = DataArgs()
def check_single_sample(item):
"""
核心验证逻辑：完全复刻训练时的视频抽帧代码
只有通过了真实 get_batch 考验的视频，才算真正的“好数据”
"""
video_path = item.get("video_abs_path")
if not video_path or not os.path.exists(video_path):
return False, item, "文件不存在或路径为空"
try:
# 完全照搬你的 process_video_with_decord 逻辑
vr = VideoReader(video_path, ctx=cpu(0), num_threads=1)
total_frame_num = len(vr)
if total_frame_num == 0:
return False, item, "视频总帧数为 0"
avg_fps = round(vr.get_avg_fps() / data_args.video_fps)
if avg_fps <= 0:
avg_fps = 1
frame_idx = [i for i in range(0, total_frame_num, avg_fps)]
if data_args.frames_upbound > 0:
if len(frame_idx) > data_args.frames_upbound or data_args.force_sample:
uniform_sampled_frames = np.linspace(0, total_frame_num - 1, data_args.frames_upbound, dtype=int)
frame_idx = uniform_sampled_frames.tolist()
# 🚨 终极考验：尝试真实提取这批帧！坏视频在这里一定会原形毕露崩溃
video = vr.get_batch(frame_idx).asnumpy()
if video.shape[0] == 0:
return False, item, "提取到的帧矩阵为空"
return True, item, "OK"
except Exception as e:
# 捕获所有 decord 底层抛出的异常 (包括你遇到的 h264 错误)
return False, item, f"解码崩溃: {str(e)}"
def run_deep_clean(input_json, clean_json, bad_json, num_workers=16):
print(f"正在读取数据集: {input_json} ...")
with open(input_json, 'r') as f:
data_list = json.load(f)
# data_list = data_list[5000:5100]
clean_data = []
bad_data = []
print(f"\n🚀 开始深度清洗 {len(data_list)} 条数据 (启用 {num_workers} 个进程并发)...")
# 使用进程池极大加快处理速度
with ProcessPoolExecutor(max_workers=num_workers) as executor:
futures = [executor.submit(check_single_sample, item) for item in data_list]
for future in tqdm(as_completed(futures), total=len(data_list), desc="🔨 质检进度"):
is_valid, item, error_msg = future.result()
if is_valid:
clean_data.append(item)
else:
item['error_reason'] = error_msg
bad_data.append(item)
print("\n" + "="*50)
print(f"✅ 清洗彻底完成！")
print(f"🎉 纯净数据: {len(clean_data)} 条 -> 已保存至 {clean_json}")
print(f"☠️ 损毁数据: {len(bad_data)} 条 -> 已保存至 {bad_json}")
print("="*50)
with open(clean_json, 'w') as f:
json.dump(clean_data, f, indent=4)
with open(bad_json, 'w') as f:
json.dump(bad_data, f, indent=4)
if __name__ == "__main__":
# 替换成你当前的 JSON 路径
INPUT_FILE = "./data/video_reversal_sft_train_cleaned_10k.json"
CLEAN_FILE = "./data/video_reversal_sft_train_ultra_clean.json"
BAD_FILE = "./data/video_reversal_sft_train_bad_samples.json"
# 根据你机器的 CPU 核心数调整 num_workers，核心越多越快
run_deep_clean(INPUT_FILE, CLEAN_FILE, BAD_FILE, num_workers=16) |