Chord视频理解工具批处理教程：自动化分析100+视频文件脚本

1. 引言

如果你手头有几十个甚至上百个视频文件需要分析，比如监控录像、短视频素材、或者产品演示视频，你会怎么做？一个视频一个视频地上传、等待、查看结果，这个过程不仅枯燥，而且效率极低。想象一下，光是上传100个视频，每个花1分钟，就要将近两个小时，更别提中间可能出现的各种操作失误了。

这就是我们今天要解决的问题。Chord视频理解工具本身已经非常强大，它能看懂视频里发生了什么，还能精确找到特定目标出现的时间和位置。但它的Web界面是为单次分析设计的，不适合批量处理。手动操作不仅慢，还容易出错，特别是当视频数量多的时候。

好消息是，我们可以通过编写一个简单的Python脚本，让Chord工具“自动化”起来。这个脚本能帮你：

• 批量处理：自动读取文件夹里的所有视频文件，一个一个交给Chord分析。
• 解放双手：你只需要设置好一次，就可以去喝杯咖啡，回来时所有结果都整理好了。
• 结果汇总：把每个视频的分析结果（文字描述或目标定位信息）自动保存到文件里，方便你后续查看和整理。

无论你是做视频内容审核、素材分析，还是学术研究，这个脚本都能帮你把重复劳动的时间节省下来，把精力用在更有价值的事情上。接下来，我就带你一步步实现这个自动化流程。

2. 环境准备与脚本基础

在开始写脚本之前，我们需要确保电脑上已经准备好了运行环境。这个过程很简单，就像给新手机安装必要的App一样。

2.1 确认Chord工具已正常运行

首先，你的Chord视频理解工具必须已经成功启动并可以正常访问。如果你还没安装，需要先完成基础部署：

1. 启动Chord服务：在你的命令行或终端里，导航到Chord工具所在的目录，运行启动命令。通常会是类似这样的命令：

   streamlit run app.py
   

   或者根据你的具体安装方式启动。

2. 验证服务状态：打开浏览器，访问工具显示的地址（通常是 http://localhost:8501）。你应该能看到Chord的标准操作界面，能够正常上传视频并进行分析。

重要提示：我们的批处理脚本不会重新部署Chord，而是“借用”你已经运行起来的服务。所以确保服务在后台稳定运行是第一步。

2.2 安装必要的Python库

我们的脚本主要用Python来写，需要用到几个额外的库来帮助我们自动化操作浏览器。打开你的命令行，运行以下安装命令：

pip install selenium webdriver-manager

简单解释一下这两个库是干什么的：

• selenium：这是一个自动化测试工具，但我们可以用它来模拟人在浏览器里的操作，比如点击按钮、输入文字、上传文件等。
• webdriver-manager：这个库能帮我们自动下载和管理不同浏览器的驱动程序，比如Chrome的驱动。你不需要手动去找驱动下载了。

安装完成后，我们就可以开始编写脚本的核心部分了。

3. 批处理脚本核心实现

现在我们来一步步构建这个自动化脚本。我会把完整的代码拆解成几个部分，每部分都有详细说明，确保你能理解每一行代码的作用。

3.1 脚本基础框架

我们先搭建脚本的基本结构，导入必要的库，并设置一些基础参数：

import os
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import logging

# 设置日志，方便查看脚本运行情况
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class ChordBatchProcessor:
    def __init__(self, chord_url="http://localhost:8501"):
        """
        初始化批处理器
        :param chord_url: Chord工具的访问地址，默认是本地8501端口
        """
        self.chord_url = chord_url
        self.driver = None
        self.results = []  # 用于存储所有视频的分析结果
        
    def setup_driver(self):
        """设置并启动Chrome浏览器驱动"""
        logger.info("正在启动浏览器...")
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')  # 无头模式，不显示浏览器窗口
        options.add_argument('--no-sandbox')
        options.add_argument('--disable-dev-shm-usage')
        
        # 自动下载和管理Chrome驱动
        service = Service(ChromeDriverManager().install())
        self.driver = webdriver.Chrome(service=service, options=options)
        self.driver.implicitly_wait(10)  # 设置隐式等待时间
        logger.info("浏览器启动成功")

这段代码做了几件事：

1. 导入了所有需要的Python库。
2. 创建了一个ChordBatchProcessor类，这是我们脚本的核心。
3. 在初始化时，我们设置了Chord的访问地址（默认是本地的8501端口）。
4. setup_driver方法会启动一个“看不见”的Chrome浏览器（无头模式），这样脚本运行时不会弹出浏览器窗口干扰你。

3.2 视频文件扫描与准备

脚本需要知道要处理哪些视频文件。我们添加一个方法来扫描指定文件夹里的视频：

    def scan_video_files(self, folder_path, extensions=['.mp4', '.avi', '.mov']):
        """
        扫描文件夹中的视频文件
        :param folder_path: 视频文件夹路径
        :param extensions: 支持的视频格式扩展名
        :return: 视频文件路径列表
        """
        if not os.path.exists(folder_path):
            logger.error(f"文件夹不存在: {folder_path}")
            return []
            
        video_files = []
        for file in os.listdir(folder_path):
            file_path = os.path.join(folder_path, file)
            if os.path.isfile(file_path) and any(file.lower().endswith(ext) for ext in extensions):
                video_files.append(file_path)
                
        logger.info(f"在文件夹 {folder_path} 中找到 {len(video_files)} 个视频文件")
        return video_files

这个方法会：

• 检查你提供的文件夹是否存在。
• 遍历文件夹里的所有文件。
• 只保留扩展名是.mp4、.avi或.mov的视频文件。
• 返回找到的所有视频文件的完整路径列表。

3.3 核心自动化操作

这是脚本最核心的部分，负责模拟人工操作Chord界面的每一步：

    def process_single_video(self, video_path, task_type="describe", query_text=""):
        """
        处理单个视频文件
        :param video_path: 视频文件路径
        :param task_type: 任务类型，'describe' 或 'grounding'
        :param query_text: 查询文本（描述需求或目标对象）
        :return: 分析结果文本
        """
        try:
            logger.info(f"开始处理视频: {os.path.basename(video_path)}")
            
            # 1. 访问Chord界面
            self.driver.get(self.chord_url)
            time.sleep(2)  # 等待页面加载
            
            # 2. 上传视频文件
            upload_input = self.driver.find_element(By.CSS_SELECTOR, "input[type='file']")
            upload_input.send_keys(video_path)
            logger.info("视频上传成功，等待处理...")
            time.sleep(3)  # 等待视频上传和处理
            
            # 3. 根据任务类型选择模式
            if task_type == "describe":
                # 选择普通描述模式
                describe_radio = WebDriverWait(self.driver, 10).until(
                    EC.element_to_be_clickable((By.XPATH, "//label[contains(text(), '普通描述')]/preceding-sibling::input"))
                )
                describe_radio.click()
                
                # 输入描述问题
                query_input = self.driver.find_element(By.XPATH, "//textarea[contains(@placeholder, '问题')]")
                query_input.clear()
                query_input.send_keys(query_text if query_text else "详细描述这个视频的内容")
                
            elif task_type == "grounding":
                # 选择视觉定位模式
                grounding_radio = WebDriverWait(self.driver, 10).until(
                    EC.element_to_be_clickable((By.XPATH, "//label[contains(text(), '视觉定位')]/preceding-sibling::input"))
                )
                grounding_radio.click()
                
                # 输入要定位的目标
                target_input = self.driver.find_element(By.XPATH, "//textarea[contains(@placeholder, '要定位的目标')]")
                target_input.clear()
                target_input.send_keys(query_text if query_text else "检测视频中的所有目标")
            
            # 4. 点击分析按钮（假设按钮文本包含"分析"）
            analyze_button = WebDriverWait(self.driver, 10).until(
                EC.element_to_be_clickable((By.XPATH, "//button[contains(text(), '分析')]"))
            )
            analyze_button.click()
            logger.info("开始分析视频，请等待...")
            
            # 5. 等待分析完成（根据视频长度调整等待时间）
            video_duration = self.get_video_duration(video_path)
            wait_time = min(60, max(10, video_duration * 2))  # 等待时间基于视频时长
            time.sleep(wait_time)
            
            # 6. 获取分析结果
            result_element = WebDriverWait(self.driver, 30).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, ".stMarkdown, .result-container, [data-testid='stMarkdownContainer']"))
            )
            result_text = result_element.text
            
            logger.info(f"视频分析完成: {os.path.basename(video_path)}")
            return result_text
            
        except Exception as e:
            logger.error(f"处理视频 {os.path.basename(video_path)} 时出错: {str(e)}")
            return f"错误: {str(e)}"
    
    def get_video_duration(self, video_path):
        """估算视频时长（简化版本）"""
        # 实际应用中可以使用moviepy或opencv获取准确时长
        # 这里返回一个默认值，你可以根据实际情况调整
        return 10  # 默认10秒

这段代码模拟了人工操作Chord的完整流程：

1. 打开Chord页面：访问你本地的Chord工具。
2. 上传视频：找到文件上传框，把视频文件的路径传给它。
3. 选择任务模式：根据你的设置，选择“普通描述”或“视觉定位”模式。
4. 输入查询文本：在对应的输入框里填入你的问题或要检测的目标。
5. 点击分析按钮：找到并点击开始分析的按钮。
6. 等待分析完成：根据视频长度等待足够的时间让Chord完成分析。
7. 获取结果：从页面上找到显示结果的区域，提取文字内容。

3.4 批量处理与结果保存

现在我们把单个视频的处理能力扩展到批量处理，并添加结果保存功能：

    def process_batch(self, video_folder, output_file="results.txt", task_type="describe", query_text=""):
        """
        批量处理文件夹中的所有视频
        :param video_folder: 视频文件夹路径
        :param output_file: 结果输出文件
        :param task_type: 任务类型
        :param query_text: 查询文本
        """
        # 扫描视频文件
        video_files = self.scan_video_files(video_folder)
        if not video_files:
            logger.error("未找到可处理的视频文件")
            return
            
        total_count = len(video_files)
        logger.info(f"开始批量处理 {total_count} 个视频文件")
        
        # 打开结果文件
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write("Chord视频分析批处理结果\n")
            f.write("=" * 50 + "\n\n")
            
            # 逐个处理视频
            for i, video_path in enumerate(video_files, 1):
                video_name = os.path.basename(video_path)
                logger.info(f"处理进度: {i}/{total_count} - {video_name}")
                
                # 处理单个视频
                result = self.process_single_video(video_path, task_type, query_text)
                
                # 保存结果
                f.write(f"视频文件: {video_name}\n")
                f.write(f"分析时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"任务类型: {'普通描述' if task_type == 'describe' else '视觉定位'}\n")
                f.write(f"查询内容: {query_text}\n")
                f.write("-" * 40 + "\n")
                f.write(f"分析结果:\n{result}\n")
                f.write("=" * 50 + "\n\n")
                
                # 记录到内存中
                self.results.append({
                    'video': video_name,
                    'result': result,
                    'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
                })
                
                # 处理间隔，避免请求过快
                time.sleep(2)
        
        logger.info(f"批量处理完成！结果已保存到: {output_file}")
        logger.info(f"成功处理 {len(self.results)}/{total_count} 个视频")
    
    def cleanup(self):
        """清理资源"""
        if self.driver:
            self.driver.quit()
            logger.info("浏览器已关闭")

这个process_batch方法是脚本的“指挥官”：

1. 先扫描文件夹，找到所有视频文件。
2. 创建一个文本文件来保存所有结果。
3. 逐个处理每个视频，把分析结果写入文件。
4. 同时在内存中也保存一份结果，方便后续使用。
5. 处理完所有视频后，会统计成功处理的数量。

4. 完整脚本与使用示例

现在我们把所有代码组合起来，形成一个完整的、可以直接运行的脚本：

#!/usr/bin/env python3
"""
Chord视频理解工具批处理脚本
作者：技术博客内容生成专家
功能：自动化批量处理视频文件，支持普通描述和视觉定位两种模式
"""

import os
import time
import argparse
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('chord_batch.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class ChordBatchProcessor:
    """Chord视频批处理处理器"""
    
    def __init__(self, chord_url="http://localhost:8501", headless=True):
        """
        初始化处理器
        :param chord_url: Chord服务地址
        :param headless: 是否使用无头模式（不显示浏览器窗口）
        """
        self.chord_url = chord_url
        self.headless = headless
        self.driver = None
        self.results = []
        
    def setup_driver(self):
        """设置浏览器驱动"""
        logger.info("正在启动Chrome浏览器...")
        options = webdriver.ChromeOptions()
        
        if self.headless:
            options.add_argument('--headless')  # 无头模式
        options.add_argument('--no-sandbox')
        options.add_argument('--disable-dev-shm-usage')
        options.add_argument('--disable-gpu')
        options.add_argument('--window-size=1920,1080')
        
        # 自动管理Chrome驱动
        try:
            service = Service(ChromeDriverManager().install())
            self.driver = webdriver.Chrome(service=service, options=options)
            self.driver.implicitly_wait(15)
            logger.info("浏览器驱动设置成功")
        except Exception as e:
            logger.error(f"浏览器驱动设置失败: {e}")
            raise
    
    def scan_video_files(self, folder_path, extensions=['.mp4', '.avi', '.mov', '.mkv']):
        """扫描视频文件"""
        if not os.path.exists(folder_path):
            raise FileNotFoundError(f"文件夹不存在: {folder_path}")
            
        video_files = []
        for root, dirs, files in os.walk(folder_path):
            for file in files:
                if any(file.lower().endswith(ext) for ext in extensions):
                    video_files.append(os.path.join(root, file))
        
        logger.info(f"在 {folder_path} 中找到 {len(video_files)} 个视频文件")
        return sorted(video_files)  # 按文件名排序
    
    def process_single_video(self, video_path, task_type="describe", query_text=""):
        """处理单个视频"""
        video_name = os.path.basename(video_path)
        logger.info(f"开始处理: {video_name}")
        
        try:
            # 访问Chord页面
            self.driver.get(self.chord_url)
            time.sleep(3)
            
            # 上传视频
            upload_input = WebDriverWait(self.driver, 20).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, "input[type='file'][accept*='video']"))
            )
            upload_input.send_keys(video_path)
            logger.info(f"已上传: {video_name}")
            time.sleep(5)  # 等待视频加载
            
            # 选择任务模式
            if task_type == "describe":
                # 普通描述模式
                try:
                    describe_radio = WebDriverWait(self.driver, 10).until(
                        EC.element_to_be_clickable((By.XPATH, "//*[contains(text(), '普通描述')]/ancestor::label/input"))
                    )
                    describe_radio.click()
                    
                    # 输入描述问题
                    query_inputs = self.driver.find_elements(By.TAG_NAME, "textarea")
                    for input_elem in query_inputs:
                        if input_elem.get_attribute("placeholder") and "问题" in input_elem.get_attribute("placeholder"):
                            input_elem.clear()
                            input_elem.send_keys(query_text or "详细描述视频内容")
                            break
                            
                except Exception as e:
                    logger.warning(f"选择描述模式时遇到问题: {e}")
                    # 尝试备用选择方式
                    try:
                        radios = self.driver.find_elements(By.CSS_SELECTOR, "input[type='radio']")
                        if len(radios) >= 1:
                            radios[0].click()  # 假设第一个是描述模式
                    except:
                        pass
                        
            elif task_type == "grounding":
                # 视觉定位模式
                try:
                    grounding_radio = WebDriverWait(self.driver, 10).until(
                        EC.element_to_be_clickable((By.XPATH, "//*[contains(text(), '视觉定位')]/ancestor::label/input"))
                    )
                    grounding_radio.click()
                    
                    # 输入定位目标
                    query_inputs = self.driver.find_elements(By.TAG_NAME, "textarea")
                    for input_elem in query_inputs:
                        if input_elem.get_attribute("placeholder") and "目标" in input_elem.get_attribute("placeholder"):
                            input_elem.clear()
                            input_elem.send_keys(query_text or "检测视频中的主要目标")
                            break
                            
                except Exception as e:
                    logger.warning(f"选择定位模式时遇到问题: {e}")
            
            # 查找并点击分析按钮
            analyze_buttons = self.driver.find_elements(By.TAG_NAME, "button")
            for button in analyze_buttons:
                if button.text and ("分析" in button.text or "Analyze" in button.text or "Run" in button.text):
                    button.click()
                    logger.info("已开始分析...")
                    break
            
            # 等待分析完成（根据文件大小估算等待时间）
            file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
            wait_time = min(120, max(15, file_size_mb * 2))  # 等待时间基于文件大小
            logger.info(f"预计等待 {wait_time:.1f} 秒...")
            time.sleep(wait_time)
            
            # 尝试获取结果
            result_text = "未找到结果"
            for _ in range(5):  # 尝试多次查找结果
                try:
                    # 尝试多种可能的结果容器选择器
                    selectors = [
                        ".stMarkdown",
                        "[data-testid='stMarkdownContainer']",
                        ".result-container",
                        ".output-area",
                        "div[class*='result']",
                        "div[class*='output']"
                    ]
                    
                    for selector in selectors:
                        try:
                            elements = self.driver.find_elements(By.CSS_SELECTOR, selector)
                            for elem in elements:
                                text = elem.text.strip()
                                if text and len(text) > 20:  # 确保是有效结果
                                    result_text = text
                                    break
                            if result_text != "未找到结果":
                                break
                        except:
                            continue
                    
                    if result_text != "未找到结果":
                        break
                        
                except Exception as e:
                    logger.debug(f"查找结果时出错: {e}")
                
                time.sleep(3)  # 等待3秒后重试
            
            logger.info(f"处理完成: {video_name}")
            return result_text
            
        except Exception as e:
            error_msg = f"处理失败: {str(e)}"
            logger.error(f"{video_name} {error_msg}")
            return error_msg
    
    def process_batch(self, video_folder, output_file="chord_results.txt", 
                     task_type="describe", query_text="", max_videos=None):
        """批量处理视频"""
        
        # 扫描视频文件
        video_files = self.scan_video_files(video_folder)
        if not video_files:
            logger.error("未找到视频文件，请检查文件夹路径")
            return False
        
        # 限制处理数量
        if max_videos and max_videos > 0:
            video_files = video_files[:max_videos]
            logger.info(f"限制处理前 {max_videos} 个视频")
        
        total = len(video_files)
        logger.info(f"开始批量处理 {total} 个视频")
        
        # 准备结果文件
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        if output_file == "chord_results.txt":
            output_file = f"chord_results_{timestamp}.txt"
        
        success_count = 0
        
        with open(output_file, 'w', encoding='utf-8') as f:
            # 写入文件头
            f.write(f"Chord视频批处理分析报告\n")
            f.write(f"生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"视频文件夹: {os.path.abspath(video_folder)}\n")
            f.write(f"任务类型: {'普通描述' if task_type == 'describe' else '视觉定位'}\n")
            f.write(f"查询内容: {query_text}\n")
            f.write("=" * 60 + "\n\n")
            
            # 处理每个视频
            for i, video_path in enumerate(video_files, 1):
                video_name = os.path.basename(video_path)
                logger.info(f"[{i}/{total}] 处理: {video_name}")
                
                # 处理视频
                result = self.process_single_video(video_path, task_type, query_text)
                
                # 记录结果
                self.results.append({
                    'index': i,
                    'video': video_name,
                    'path': video_path,
                    'result': result,
                    'success': "失败" if "失败" in result or "错误" in result else "成功",
                    'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
                })
                
                # 写入文件
                f.write(f"\n{'='*60}\n")
                f.write(f"视频 #{i}: {video_name}\n")
                f.write(f"处理时间: {self.results[-1]['timestamp']}\n")
                f.write(f"状态: {self.results[-1]['success']}\n")
                f.write(f"{'-'*40}\n")
                f.write(f"分析结果:\n{result}\n")
                
                if self.results[-1]['success'] == "成功":
                    success_count += 1
                
                # 进度间隔
                if i < total:
                    time.sleep(2)  # 视频间等待2秒
        
        # 写入统计信息
        with open(output_file, 'a', encoding='utf-8') as f:
            f.write(f"\n{'='*60}\n")
            f.write(f"处理统计:\n")
            f.write(f"总视频数: {total}\n")
            f.write(f"成功处理: {success_count}\n")
            f.write(f"失败数量: {total - success_count}\n")
            f.write(f"成功率: {success_count/total*100:.1f}%\n")
            f.write(f"完成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
        
        logger.info(f"批量处理完成！")
        logger.info(f"成功: {success_count}/{total} ({success_count/total*100:.1f}%)")
        logger.info(f"结果文件: {output_file}")
        
        return True
    
    def generate_summary(self, summary_file="processing_summary.csv"):
        """生成处理摘要CSV文件"""
        if not self.results:
            logger.warning("没有处理结果可汇总")
            return
        
        import csv
        
        with open(summary_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            # 写入标题行
            writer.writerow(['序号', '视频文件名', '文件路径', '处理状态', '处理时间', '结果摘要'])
            
            # 写入数据行
            for result in self.results:
                # 提取结果前100字符作为摘要
                result_preview = result['result'][:100] + "..." if len(result['result']) > 100 else result['result']
                writer.writerow([
                    result['index'],
                    result['video'],
                    result['path'],
                    result['success'],
                    result['timestamp'],
                    result_preview
                ])
        
        logger.info(f"处理摘要已保存到: {summary_file}")
    
    def cleanup(self):
        """清理资源"""
        if self.driver:
            try:
                self.driver.quit()
                logger.info("浏览器已关闭")
            except:
                pass

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='Chord视频理解工具批处理脚本')
    parser.add_argument('--folder', '-f', required=True, help='视频文件夹路径')
    parser.add_argument('--output', '-o', default='chord_results.txt', help='输出文件名')
    parser.add_argument('--task', '-t', choices=['describe', 'grounding'], default='describe', 
                       help='任务类型: describe(普通描述) 或 grounding(视觉定位)')
    parser.add_argument('--query', '-q', default='', help='查询文本（描述问题或定位目标）')
    parser.add_argument('--max', '-m', type=int, default=0, help='最大处理视频数（0表示全部）')
    parser.add_argument('--visible', '-v', action='store_true', help='显示浏览器窗口（默认无头模式）')
    parser.add_argument('--url', '-u', default='http://localhost:8501', help='Chord服务地址')
    
    args = parser.parse_args()
    
    # 设置查询文本默认值
    if not args.query:
        if args.task == 'describe':
            args.query = "详细描述这个视频的内容，包括场景、人物、动作和事件"
        else:
            args.query = "检测视频中的主要目标"
    
    processor = None
    try:
        # 创建处理器实例
        processor = ChordBatchProcessor(chord_url=args.url, headless=not args.visible)
        
        # 设置浏览器驱动
        processor.setup_driver()
        
        # 批量处理视频
        success = processor.process_batch(
            video_folder=args.folder,
            output_file=args.output,
            task_type=args.task,
            query_text=args.query,
            max_videos=args.max if args.max > 0 else None
        )
        
        if success:
            # 生成处理摘要
            processor.generate_summary()
            logger.info("所有任务完成！")
        else:
            logger.error("处理过程中出现问题")
            
    except Exception as e:
        logger.error(f"程序执行出错: {e}")
    finally:
        # 清理资源
        if processor:
            processor.cleanup()

if __name__ == "__main__":
    main()

4.1 如何使用这个脚本

保存上面的代码到一个文件，比如叫chord_batch.py。然后你可以通过命令行来使用它，非常方便：

基本用法：

# 处理文件夹中的所有视频，使用普通描述模式
python chord_batch.py --folder /path/to/your/videos

# 处理视频，使用视觉定位模式，检测"人"这个目标
python chord_batch.py --folder /path/to/videos --task grounding --query "人"

# 只处理前10个视频，并显示浏览器窗口（可以看到处理过程）
python chord_batch.py --folder /path/to/videos --max 10 --visible

# 指定输出文件名和查询内容
python chord_batch.py --folder /path/to/videos --output my_results.txt --query "描述视频中的主要活动和场景变化"

参数说明：

• --folder / -f：必须的，指定包含视频文件的文件夹路径
• --task / -t：任务类型，describe（普通描述）或grounding（视觉定位），默认是describe
• --query / -q：查询文本，不指定时使用默认值
• --max / -m：最大处理数量，0表示处理所有视频
• --visible / -v：显示浏览器窗口（默认不显示）
• --output / -o：输出文件名，默认是chord_results_时间戳.txt
• --url / -u：Chord服务地址，默认是http://localhost:8501

4.2 实际使用示例

假设你有一个视频文件夹~/videos_to_analyze，里面有50个监控视频，你想分析每个视频中是否有人出现，并记录出现的时间和位置。你可以这样运行：

python chord_batch.py \
  --folder ~/videos_to_analyze \
  --task grounding \
  --query "人" \
  --max 50 \
  --output surveillance_analysis.txt

脚本会：

1. 扫描~/videos_to_analyze文件夹里的所有视频文件
2. 只处理前50个视频（如果超过50个）
3. 对每个视频执行视觉定位任务，检测"人"这个目标
4. 把结果保存到surveillance_analysis.txt文件
5. 同时生成一个processing_summary.csv的摘要文件

处理过程中，你可以在终端看到实时进度：

2024-01-15 10:30:15 - INFO - 在 ~/videos_to_analyze 中找到 50 个视频文件
2024-01-15 10:30:20 - INFO - 开始批量处理 50 个视频
2024-01-15 10:30:25 - INFO - [1/50] 处理: camera1_20240115.mp4
2024-01-15 10:30:30 - INFO - 已上传: camera1_20240115.mp4
2024-01-15 10:30:35 - INFO - 已开始分析...
2024-01-15 10:31:15 - INFO - 处理完成: camera1_20240115.mp4
2024-01-15 10:31:17 - INFO - [2/50] 处理: camera2_20240115.mp4
...

处理完成后，打开结果文件，你会看到类似这样的内容：

Chord视频批处理分析报告
生成时间: 2024-01-15 11:45:30
视频文件夹: /home/user/videos_to_analyze
任务类型: 视觉定位
查询内容: 人
============================================================

视频 #1: camera1_20240115.mp4
处理时间: 2024-01-15 10:31:15
状态: 成功
----------------------------------------
分析结果:
检测到2个人出现在视频中：
1. 时间戳: 00:12-00:18，边界框: [0.45, 0.32, 0.58, 0.47]，描述: 男性，穿着蓝色上衣，从左侧走入画面
2. 时间戳: 00:25-00:31，边界框: [0.62, 0.28, 0.71, 0.42]，描述: 女性，背着背包，在画面中央停留

============================================================

视频 #2: camera2_20240115.mp4
处理时间: 2024-01-15 10:32:45
状态: 成功
----------------------------------------
分析结果:
检测到1个人出现在视频中：
时间戳: 00:05-00:15，边界框: [0.35, 0.40, 0.50, 0.60]，描述: 儿童，穿着红色外套，在画面中跑动

============================================================

处理统计:
总视频数: 50
成功处理: 48
失败数量: 2
成功率: 96.0%
完成时间: 2024-01-15 11:45:30

5. 高级技巧与优化建议

当你熟悉了基本用法后，这里有一些进阶技巧可以让你的批处理更加高效和稳定：

5.1 处理大量视频的优化策略

如果你有几百甚至上千个视频需要处理，可以考虑以下优化：

分批次处理：

# 你可以修改脚本，添加分批处理逻辑
def process_in_batches(self, video_folder, batch_size=20, delay_between_batches=60):
    """分批处理视频，每批处理完后休息一段时间"""
    all_videos = self.scan_video_files(video_folder)
    
    for batch_start in range(0, len(all_videos), batch_size):
        batch_end = min(batch_start + batch_size, len(all_videos))
        batch_videos = all_videos[batch_start:batch_end]
        
        logger.info(f"处理批次 {batch_start//batch_size + 1}/{(len(all_videos)+batch_size-1)//batch_size}")
        
        # 处理当前批次
        for video in batch_videos:
            self.process_single_video(video)
        
        # 批次间休息，避免过热或资源占用过高
        if batch_end < len(all_videos):
            logger.info(f"批次完成，休息 {delay_between_batches} 秒...")
            time.sleep(delay_between_batches)

并行处理（高级）： 对于有多个GPU或者强大CPU的系统，可以考虑使用多进程并行处理。但要注意Chord服务可能不支持并发请求，需要谨慎使用。

5.2 错误处理与重试机制

在实际使用中，可能会遇到各种问题。我们可以增强脚本的健壮性：

def process_single_video_with_retry(self, video_path, max_retries=3, task_type="describe", query_text=""):
    """带重试机制的单个视频处理"""
    for attempt in range(max_retries):
        try:
            result = self.process_single_video(video_path, task_type, query_text)
            if "错误" not in result and "失败" not in result:
                return result
            else:
                logger.warning(f"第{attempt+1}次尝试失败，结果包含错误信息")
        except Exception as e:
            logger.error(f"第{attempt+1}次尝试异常: {e}")
        
        if attempt < max_retries - 1:
            wait_time = (attempt + 1) * 10  # 重试等待时间递增
            logger.info(f"{wait_time}秒后重试...")
            time.sleep(wait_time)
    
    return f"经过{max_retries}次尝试后仍失败"

5.3 结果后处理与分析

批处理完成后，你可能需要对结果进行进一步分析。这里提供一个简单的结果分析脚本：

def analyze_results(self, result_file):
    """分析处理结果文件"""
    with open(result_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 分割各个视频的结果
    video_sections = content.split('=' * 60)
    
    analysis_stats = {
        'total_videos': 0,
        'successful': 0,
        'failed': 0,
        'avg_result_length': 0,
        'common_keywords': {}
    }
    
    keyword_counter = {}
    
    for section in video_sections:
        if '视频 #' in section:
            analysis_stats['total_videos'] += 1
            
            # 检查是否成功
            if '状态: 成功' in section:
                analysis_stats['successful'] += 1
                
                # 提取结果文本
                result_start = section.find('分析结果:')
                if result_start != -1:
                    result_text = section[result_start + 6:]  # 跳过"分析结果:"
                    
                    # 统计结果长度
                    analysis_stats['avg_result_length'] += len(result_text)
                    
                    # 简单关键词提取（实际中可以更复杂）
                    words = result_text.lower().split()
                    for word in words:
                        if len(word) > 3:  # 只统计长度大于3的词
                            keyword_counter[word] = keyword_counter.get(word, 0) + 1
            else:
                analysis_stats['failed'] += 1
    
    # 计算平均值
    if analysis_stats['successful'] > 0:
        analysis_stats['avg_result_length'] /= analysis_stats['successful']
    
    # 获取最常见的关键词
    from collections import Counter
    common_keywords = Counter(keyword_counter).most_common(10)
    analysis_stats['common_keywords'] = dict(common_keywords)
    
    return analysis_stats

5.4 实际应用场景建议

根据不同的使用场景，这里有一些实用建议：

场景1：内容审核

• 使用普通描述模式，查询文本设为"描述视频中的违规内容"
• 处理完成后，用文本搜索功能查找结果中的关键词（如"暴力"、"裸露"、"血腥"等）
• 将可疑视频单独标记出来进行人工复核

场景2：视频素材分析

• 使用普通描述模式，查询文本设为"详细描述视频的场景、人物、动作、色彩和氛围"
• 建立关键词索引，方便后续按场景、人物等条件搜索素材
• 根据描述长度和详细程度评估素材的质量和可用性

场景3：安防监控分析

• 使用视觉定位模式，查询文本设为"人 车 动物"（多个目标用空格分隔）
• 重点关注目标出现的时间戳和位置信息
• 可以结合时间信息分析人员流动规律

场景4：学术研究数据收集

• 两种模式结合使用，先进行普通描述了解视频内容
• 再针对特定目标进行视觉定位分析
• 将结果结构化存储到数据库，方便统计分析

6. 常见问题与解决方案

在实际使用过程中，你可能会遇到一些问题。这里列出了一些常见问题及其解决方法：

6.1 浏览器驱动问题

问题：脚本启动时提示找不到Chrome驱动或版本不匹配。 解决：

• 确保已安装最新版Chrome浏览器
• 脚本使用webdriver-manager自动管理驱动，但有时需要手动更新：

  pip install --upgrade webdriver-manager
  

• 或者指定Chrome路径：

  options.binary_location = "/path/to/chrome"
  

6.2 Chord服务连接问题

问题：脚本无法连接到Chord服务。 解决：

• 确认Chord服务正在运行：在浏览器中访问 http://localhost:8501 看是否能打开
• 检查防火墙设置，确保端口8501可访问
• 如果Chord运行在其他地址或端口，使用--url参数指定：

  python chord_batch.py --folder ./videos --url http://192.168.1.100:8501
  

6.3 视频上传失败

问题：视频文件上传后没有反应或报错。 解决：

• 检查视频格式是否支持（MP4、AVI、MOV）
• 检查视频文件是否损坏，尝试用播放器打开
• 对于大文件（>100MB），增加上传等待时间：

  # 在process_single_video方法中增加等待时间
  file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
  upload_wait = min(30, max(5, file_size_mb / 10))  # 每10MB等待1秒，最多30秒
  time.sleep(upload_wait)
  

6.4 分析结果获取失败

问题：视频分析完成但脚本无法获取结果。 解决：

• Chord界面可能更新，需要调整结果元素的选择器
• 增加等待时间和重试次数：

  # 增加等待时间
  wait_time = min(180, max(30, file_size_mb * 3))  # 最多等待3分钟
  
  # 增加重试次数
  for retry in range(10):  # 尝试10次
      try:
          # 尝试获取结果
          break
      except:
          time.sleep(5)  # 每次重试等待5秒
  

6.5 内存和性能问题

问题：处理大量视频时内存占用高或速度慢。 解决：

• 分批处理，每批20-30个视频
• 处理完一批后重启浏览器释放内存：

  def process_batch_with_restart(self, video_files, batch_size=20):
      for i in range(0, len(video_files), batch_size):
          batch = video_files[i:i+batch_size]
          # 处理当前批次
          for video in batch:
              self.process_single_video(video)
          # 重启浏览器释放内存
          self.driver.quit()
          time.sleep(2)
          self.setup_driver()
  

• 使用无头模式（默认）减少资源占用

6.6 结果文件过大

问题：处理大量视频后，结果文本文件非常大，难以查看。 解决：

• 将结果保存为结构化格式（如JSON或CSV）：

  import json
  
  def save_results_as_json(self, output_file="results.json"):
      results_data = {
          'metadata': {
              'processed_at': time.strftime('%Y-%m-%d %H:%M:%S'),
              'total_videos': len(self.results),
              'task_type': self.task_type
          },
          'videos': self.results
      }
      
      with open(output_file, 'w', encoding='utf-8') as f:
          json.dump(results_data, f, ensure_ascii=False, indent=2)
  

• 或者将每个视频的结果保存为单独文件：

  def save_results_individual(self, output_dir="results"):
      os.makedirs(output_dir, exist_ok=True)
      for result in self.results:
          video_name = os.path.splitext(result['video'])[0]
          output_file = os.path.join(output_dir, f"{video_name}_result.txt")
          with open(output_file, 'w', encoding='utf-8') as f:
              f.write(f"视频: {result['video']}\n")
              f.write(f"处理时间: {result['timestamp']}\n")
              f.write(f"分析结果:\n{result['result']}\n")
  

7. 总结

通过这个Chord视频理解工具的批处理脚本，我们成功将原本需要手动操作的重复性工作自动化了。现在让我们回顾一下这个方案的核心价值和使用要点：

7.1 主要收获

效率提升：这是最直接的收益。原本需要人工一个个上传、设置、等待、记录的视频分析任务，现在可以全自动完成。假设每个视频手动处理需要2分钟，100个视频就是200分钟（超过3小时）。使用脚本后，你可以设置好参数就去处理其他工作，大大节省了时间。

结果一致性：人工操作难免会有疏忽或错误，比如输错查询文本、忘记保存结果等。脚本确保了每个视频都按照相同的标准流程处理，结果格式统一，便于后续分析和整理。

可扩展性：这个脚本框架很容易扩展。你可以根据需要添加新的功能，比如：

• 支持更多视频格式
• 添加结果自动分类
• 集成到更大的工作流中
• 添加邮件或消息通知功能

易于使用：虽然脚本看起来有点复杂，但实际使用很简单。基本上就是安装依赖、运行命令、查看结果三个步骤。即使你不是专业的程序员，按照教程也能顺利使用。

7.2 使用建议

根据我这边的测试和使用经验，给你几个实用建议：

开始前：

1. 先小批量测试：不要一开始就处理几百个视频。先用3-5个视频测试，确保一切正常。
2. 检查视频格式：确保你的视频都是Chord支持的格式（MP4、AVI、MOV）。
3. 备份重要视频：虽然脚本很安全，但处理前备份重要视频总是一个好习惯。

运行时：

1. 使用无头模式：除非需要调试，否则保持无头模式（不显示浏览器窗口），这样资源占用更少，速度更快。
2. 合理设置等待时间：脚本中的等待时间是估算值。如果你的视频特别大或特别小，可以适当调整wait_time的计算公式。
3. 监控进度：脚本会输出详细的日志，定期查看日志文件可以了解处理进度和发现问题。

处理后：

1. 检查结果文件：处理完成后，快速浏览一下结果文件，确保没有大量失败的情况。
2. 使用摘要功能：脚本生成的CSV摘要文件很有用，可以用Excel打开，快速了解处理情况。
3. 结果后处理：根据你的具体需求，可以对结果进行进一步处理，比如提取关键信息、生成统计报告等。

7.3 可能的改进方向

如果你对这个脚本感兴趣，想要进一步定制或改进，这里有一些方向：

性能优化：

• 添加并行处理支持（如果Chord服务支持并发）
• 优化等待策略，根据视频长度动态调整
• 添加断点续传功能，处理中断后可以从上次停止的地方继续

功能增强：

• 支持更多任务类型和参数配置
• 添加结果自动分析和报告生成
• 集成到图形界面中，让非技术人员也能方便使用

稳定性提升：

• 更完善的错误处理和恢复机制
• 添加健康检查，自动重启异常的Chord服务
• 支持代理和网络异常处理

7.4 最后的话

技术工具的价值在于解决实际问题。Chord视频理解工具本身已经很强大了，但通过自动化脚本，我们让它变得更加实用。无论你是需要分析大量监控视频的安防人员，还是需要处理视频素材的内容创作者，或是进行视频数据分析的研究人员，这个脚本都能帮你节省大量时间。

记住，自动化不是要完全取代人工，而是要把人从重复性劳动中解放出来，让我们能专注于更有创造性和价值的工作。希望这个脚本能成为你视频分析工作的好帮手。

如果在使用过程中遇到问题，或者有改进建议，欢迎根据实际情况进行调整。技术的乐趣就在于不断优化和改进，让工具更好地为我们服务。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。