小白也能懂：Chord视频理解工具+Python爬虫，轻松实现视频数据自动采集与分析
·
小白也能懂：Chord视频理解工具+Python爬虫，轻松实现视频数据自动采集与分析
1. 为什么你需要这个自动化方案
想象一下这个场景：你负责运营一个短视频账号，每天需要从几十个竞品账号里找出最新的视频，分析他们的内容特点、发布时间、画面亮点。手动操作是什么体验？打开浏览器，一个个账号点进去，找到视频，下载下来，再打开播放器，边看边记笔记。一天下来，眼睛花了，手也酸了，效率还低得可怜。
或者你是做市场研究的，需要定期收集某个行业的所有新品发布视频，分析产品展示的时长、演示的重点环节、用户评论的热点。靠人工一个个看，不仅耗时，还容易漏掉关键信息。
这就是传统视频分析工作的痛点——重复、耗时、容易出错。而今天要介绍的这套组合方案，就是用技术把我们从这些重复劳动中解放出来。
Chord视频时空理解工具，简单来说，就是一个能“看懂”视频的智能工具。它不只是截取画面，而是能理解视频里“什么时候发生了什么”、“画面里谁在哪儿”。比如，它能告诉你：“视频第12秒，一只小狗从画面左侧跑到了右侧”，或者“第25秒到第30秒，画面中央出现了产品特写镜头”。
Python爬虫，则是我们的“数据采集员”。它能自动去各个平台，按照我们设定的规则，把视频链接、标题、发布时间等信息抓取回来。
把这两者结合起来，就形成了一个自动化的工作流：爬虫负责“找”视频，Chord负责“看”视频，最后把分析结果整理好交给我们。整个过程不需要人工干预，设定好任务，它就能自己运行。
这套方案特别适合谁？
- 内容创作者：想批量分析热门视频的套路和亮点。
- 市场分析师：需要监控竞品动态和行业趋势。
- 产品经理：想了解用户生成内容中的产品使用场景。
- 研究人员：需要从大量视频中提取结构化数据做分析。
接下来，我会用最直白的方式，带你一步步搭建这个系统，即使你之前没怎么写过代码，也能跟着做出来。
2. 准备工作：环境搭建与工具安装
在开始写代码之前，我们需要先把“舞台”搭好。不用担心，跟着步骤来，每一步都很简单。
2.1 安装Python和相关库
首先确保你的电脑上安装了Python。打开命令行（Windows上是CMD或PowerShell，Mac/Linux上是终端），输入：
python --version
如果显示Python 3.7或更高版本，那就没问题。如果没有安装，去Python官网下载安装包，记得安装时勾选“Add Python to PATH”。
接下来安装我们需要的Python库。创建一个新的文件夹作为项目目录，然后在里面创建一个requirements.txt文件，内容如下：
# 爬虫相关
requests==2.31.0
beautifulsoup4==4.12.2
selenium==4.15.0
yt-dlp==2023.11.16
# 数据处理相关
pandas==2.1.3
sqlite3 # 这个通常Python自带
# 其他工具
psutil==5.9.6
然后在命令行里进入项目目录，运行：
pip install -r requirements.txt
这里简单解释一下每个库是干什么的：
requests：用来访问网页，获取数据beautifulsoup4：解析网页内容，提取我们需要的信息selenium：处理那些需要JavaScript加载的页面yt-dlp：下载视频的神器，支持几乎所有视频平台pandas：处理和分析数据sqlite3：轻量级数据库，用来存储我们的数据psutil：监控系统资源使用情况
2.2 部署Chord视频理解工具
Chord工具已经打包成了现成的镜像，部署起来特别简单。如果你在CSDN星图平台，可以直接搜索“Chord 视频时空理解工具”镜像，一键部署。
部署成功后，你会看到一个Web界面。这个界面分为三个主要区域：
- 左侧设置区：可以调整“最大生成长度”，控制分析结果的详细程度
- 中间上传区：上传你的视频文件（支持MP4、AVI、MOV格式）
- 右侧操作区：选择分析模式，输入你的问题
工具提供两种分析模式：
- 普通描述模式：让工具详细描述视频内容
- 视觉定位模式：让工具找出视频中特定目标的位置和时间
比如你上传一个宠物视频，在视觉定位模式输入“正在奔跑的小狗”，工具就会告诉你小狗在视频的哪些时间点出现，以及在画面中的具体位置。
2.3 配置自动化运行环境
为了让整个系统能自动运行，我们需要做一些基础配置。
首先，创建一个项目结构清晰的文件夹：
video-auto-collector/
├── crawler/ # 爬虫相关代码
├── analyzer/ # 视频分析代码
├── storage/ # 数据存储代码
├── config/ # 配置文件
├── logs/ # 日志文件
├── videos/ # 下载的视频文件
└── main.py # 主程序入口
然后创建一个简单的配置文件config/config.yaml：
# 爬虫配置
crawler:
user_agents:
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
- "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
request_delay: 2 # 请求间隔秒数
max_retries: 3 # 最大重试次数
# Chord工具配置
chord:
api_url: "http://localhost:7860" # Chord工具地址
timeout: 300 # 超时时间（秒）
max_length: 512 # 最大生成长度
# 存储配置
storage:
database: "videos.db"
video_dir: "./videos"
max_video_size: 500 # 最大视频大小（MB）
# 任务调度
scheduler:
crawl_interval: 3600 # 爬虫运行间隔（秒）
analyze_interval: 600 # 分析运行间隔（秒）
这样，我们的基础环境就准备好了。接下来开始写核心的爬虫代码。
3. 编写智能爬虫：自动获取视频数据
爬虫的核心任务很简单：去指定的网站，找到视频，把视频信息拿回来。但不同的网站结构不同，我们需要针对性地处理。
3.1 基础爬虫框架
我们先写一个通用的爬虫基类，处理一些公共的逻辑：
import requests
import time
import random
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import yt_dlp
from urllib.parse import urljoin
import logging
class BaseVideoCrawler:
"""视频爬虫基类"""
def __init__(self, config):
self.config = config
self.session = requests.Session()
self.logger = logging.getLogger(self.__class__.__name__)
# 设置请求头
self.headers = {
'User-Agent': random.choice(config['user_agents']),
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
def fetch_page(self, url, use_selenium=False):
"""获取页面内容"""
try:
if use_selenium:
return self._fetch_with_selenium(url)
else:
return self._fetch_with_requests(url)
except Exception as e:
self.logger.error(f"获取页面失败: {url}, 错误: {e}")
return None
def _fetch_with_requests(self, url):
"""使用requests获取页面"""
time.sleep(self.config['request_delay']) # 礼貌等待
response = self.session.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
return response.text
def _fetch_with_selenium(self, url):
"""使用Selenium获取动态页面"""
chrome_options = Options()
chrome_options.add_argument('--headless') # 无头模式，不显示浏览器窗口
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
driver = webdriver.Chrome(options=chrome_options)
try:
driver.get(url)
time.sleep(3) # 等待页面加载
return driver.page_source
finally:
driver.quit()
def extract_video_links(self, html, base_url):
"""从HTML中提取视频链接（子类需要重写）"""
raise NotImplementedError("子类必须实现此方法")
def get_video_info(self, video_url):
"""获取视频详细信息（子类需要重写）"""
raise NotImplementedError("子类必须实现此方法")
def run(self, start_urls):
"""运行爬虫"""
all_videos = []
for url in start_urls:
self.logger.info(f"开始爬取: {url}")
html = self.fetch_page(url, use_selenium=False)
if not html:
continue
video_links = self.extract_video_links(html, url)
self.logger.info(f"找到 {len(video_links)} 个视频链接")
for link in video_links:
try:
video_info = self.get_video_info(link)
if video_info:
all_videos.append(video_info)
self.logger.info(f"获取视频信息: {video_info['title'][:50]}...")
except Exception as e:
self.logger.error(f"处理视频失败: {link}, 错误: {e}")
return all_videos
这个基类处理了通用的网络请求、错误处理、日志记录等逻辑。针对不同的网站，我们只需要继承这个类，实现特定的提取逻辑。
3.2 针对不同平台的爬虫实现
不同的视频平台，页面结构不同，我们需要写不同的爬虫。这里以两个常见平台为例：
示例1：静态页面视频平台
class StaticVideoCrawler(BaseVideoCrawler):
"""处理静态HTML页面的视频爬虫"""
def extract_video_links(self, html, base_url):
"""从静态页面提取视频链接"""
soup = BeautifulSoup(html, 'html.parser')
video_links = []
# 查找所有可能的视频链接
# 这里需要根据实际网站结构调整选择器
for a_tag in soup.find_all('a', href=True):
href = a_tag['href']
# 判断是否是视频链接（根据实际网站调整）
if '/video/' in href or '.mp4' in href or 'video_id=' in href:
full_url = urljoin(base_url, href)
if full_url not in video_links:
video_links.append(full_url)
return video_links[:20] # 限制数量，避免太多
def get_video_info(self, video_url):
"""获取视频详细信息"""
html = self.fetch_page(video_url)
if not html:
return None
soup = BeautifulSoup(html, 'html.parser')
# 提取标题（根据实际网站调整选择器）
title_elem = soup.find('h1') or soup.find('title')
title = title_elem.get_text(strip=True) if title_elem else "未知标题"
# 提取描述
desc_elem = soup.find('meta', attrs={'name': 'description'})
description = desc_elem['content'] if desc_elem else ""
# 提取发布时间（如果有）
time_elem = soup.find('time')
publish_time = time_elem['datetime'] if time_elem else ""
return {
'url': video_url,
'title': title,
'description': description,
'publish_time': publish_time,
'platform': 'static_site'
}
示例2：使用API的视频平台
class APIVideoCrawler(BaseVideoCrawler):
"""通过API获取视频的爬虫"""
def __init__(self, config, api_key):
super().__init__(config)
self.api_key = api_key
self.base_api_url = "https://api.example.com/v1"
def extract_video_links(self, html, base_url):
"""API爬虫不需要解析HTML，直接返回空列表"""
return []
def get_video_info(self, video_url):
"""通过API获取视频信息"""
# 从URL中提取视频ID
video_id = self._extract_video_id(video_url)
if not video_id:
return None
# 调用API
api_url = f"{self.base_api_url}/videos/{video_id}"
headers = {
**self.headers,
'Authorization': f'Bearer {self.api_key}'
}
try:
response = self.session.get(api_url, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
return {
'url': video_url,
'video_id': video_id,
'title': data.get('title', ''),
'description': data.get('description', ''),
'publish_time': data.get('published_at', ''),
'duration': data.get('duration', 0),
'views': data.get('view_count', 0),
'platform': 'api_platform'
}
except Exception as e:
self.logger.error(f"API请求失败: {video_url}, 错误: {e}")
return None
def _extract_video_id(self, url):
"""从URL中提取视频ID"""
# 这里需要根据实际平台URL格式编写提取逻辑
import re
patterns = [
r'v=([a-zA-Z0-9_-]+)', # YouTube风格
r'/video/(\d+)', # 数字ID
r'/([a-zA-Z0-9]+)$' # 路径最后部分
]
for pattern in patterns:
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def run(self, start_urls):
"""重写run方法，直接通过API获取视频列表"""
all_videos = []
# 假设API提供了视频列表接口
list_api_url = f"{self.base_api_url}/videos"
params = {
'limit': 50,
'order_by': 'newest'
}
headers = {
**self.headers,
'Authorization': f'Bearer {self.api_key}'
}
try:
response = self.session.get(list_api_url, params=params, headers=headers)
response.raise_for_status()
videos_data = response.json().get('items', [])
for item in videos_data:
video_info = {
'url': item.get('url', ''),
'video_id': item.get('id', ''),
'title': item.get('title', ''),
'description': item.get('description', ''),
'publish_time': item.get('published_at', ''),
'duration': item.get('duration', 0),
'views': item.get('view_count', 0),
'platform': 'api_platform'
}
all_videos.append(video_info)
self.logger.info(f"获取视频: {video_info['title'][:50]}...")
except Exception as e:
self.logger.error(f"获取视频列表失败: {e}")
return all_videos
3.3 视频下载功能
获取到视频信息后，我们需要下载视频文件供Chord分析。这里使用yt-dlp，它支持几乎所有的视频平台：
import subprocess
from pathlib import Path
class VideoDownloader:
"""视频下载器"""
def __init__(self, download_dir='./videos'):
self.download_dir = Path(download_dir)
self.download_dir.mkdir(exist_ok=True)
self.logger = logging.getLogger('VideoDownloader')
def download_video(self, video_url, video_id, max_retries=3):
"""下载视频文件"""
output_path = self.download_dir / f"{video_id}.mp4"
# 如果文件已存在且大小合理，跳过下载
if output_path.exists() and output_path.stat().st_size > 1024 * 1024: # 大于1MB
self.logger.info(f"视频已存在: {video_id}")
return str(output_path)
# yt-dlp配置
ydl_opts = {
'outtmpl': str(output_path.with_suffix('')), # 输出模板（不带扩展名）
'format': 'best[height<=720]', # 选择720p或以下的最佳质量
'quiet': True, # 减少输出
'no_warnings': True,
'ignoreerrors': True,
'retries': 10,
'fragment_retries': 10,
'continuedl': True, # 断点续传
}
for attempt in range(max_retries):
try:
self.logger.info(f"开始下载视频 {video_id} (尝试 {attempt + 1}/{max_retries})")
# 使用yt-dlp下载
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([video_url])
# 检查文件是否下载成功
if output_path.exists() and output_path.stat().st_size > 1024 * 1024:
self.logger.info(f"视频下载成功: {video_id}, 大小: {output_path.stat().st_size // 1024 // 1024}MB")
return str(output_path)
else:
self.logger.warning(f"视频文件异常: {video_id}")
except Exception as e:
self.logger.error(f"下载失败 {video_id} (尝试 {attempt + 1}): {e}")
if attempt < max_retries - 1:
import time
time.sleep(2 ** attempt) # 指数退避
self.logger.error(f"视频下载最终失败: {video_id}")
return None
def cleanup_old_videos(self, keep_days=7):
"""清理旧视频文件"""
import time
current_time = time.time()
cutoff_time = current_time - (keep_days * 24 * 3600)
deleted_count = 0
for video_file in self.download_dir.glob("*.mp4"):
if video_file.stat().st_mtime < cutoff_time:
try:
video_file.unlink()
deleted_count += 1
except Exception as e:
self.logger.error(f"删除文件失败 {video_file}: {e}")
if deleted_count > 0:
self.logger.info(f"清理了 {deleted_count} 个旧视频文件")
4. 集成Chord工具：让机器“看懂”视频
爬虫帮我们找到了视频，接下来就要让Chord工具来分析这些视频了。Chord的强大之处在于它能理解视频的时空信息，而不仅仅是识别画面内容。
4.1 与Chord工具交互
Chord工具提供了Web界面，我们可以通过编程方式与它交互。这里我们写一个封装类：
import requests
import json
import time
from pathlib import Path
class ChordAnalyzer:
"""Chord视频分析器"""
def __init__(self, chord_url="http://localhost:7860", timeout=300):
self.chord_url = chord_url
self.timeout = timeout
self.logger = logging.getLogger('ChordAnalyzer')
def analyze_video(self, video_path, task_type="describe", query=""):
"""
分析视频
Args:
video_path: 视频文件路径
task_type: 任务类型，'describe'或'grounding'
query: 查询内容（仅grounding模式需要）
Returns:
分析结果字典
"""
# 检查文件是否存在
video_file = Path(video_path)
if not video_file.exists():
self.logger.error(f"视频文件不存在: {video_path}")
return None
# 准备请求数据
files = {
'file': (video_file.name, open(video_file, 'rb'), 'video/mp4')
}
data = {
'max_length': 512, # 控制输出长度
'task_type': task_type
}
if task_type == 'grounding' and query:
data['query'] = query
try:
self.logger.info(f"开始分析视频: {video_file.name}")
start_time = time.time()
# 发送请求到Chord工具
response = requests.post(
f"{self.chord_url}/analyze",
files=files,
data=data,
timeout=self.timeout
)
response.raise_for_status()
result = response.json()
elapsed_time = time.time() - start_time
self.logger.info(f"视频分析完成: {video_file.name}, 耗时: {elapsed_time:.1f}秒")
return {
'success': True,
'video_path': str(video_path),
'task_type': task_type,
'query': query,
'result': result,
'analysis_time': elapsed_time
}
except requests.exceptions.Timeout:
self.logger.error(f"分析超时: {video_file.name}")
return {'success': False, 'error': 'timeout'}
except requests.exceptions.RequestException as e:
self.logger.error(f"请求失败: {video_file.name}, 错误: {e}")
return {'success': False, 'error': str(e)}
except json.JSONDecodeError as e:
self.logger.error(f"JSON解析失败: {video_file.name}, 错误: {e}")
return {'success': False, 'error': 'json_decode_error'}
finally:
if 'file' in locals():
files['file'][1].close()
def batch_analyze(self, video_paths, task_type="describe", queries=None):
"""批量分析视频"""
results = []
for i, video_path in enumerate(video_paths):
query = queries[i] if queries and i < len(queries) else ""
result = self.analyze_video(video_path, task_type, query)
if result:
results.append(result)
# 避免请求过于频繁
if i < len(video_paths) - 1:
time.sleep(2)
return results
def parse_analysis_result(self, result, video_info=None):
"""解析分析结果，提取有用信息"""
if not result.get('success'):
return None
raw_result = result.get('result', {})
parsed = {
'video_id': video_info.get('video_id', '') if video_info else '',
'title': video_info.get('title', '') if video_info else '',
'task_type': result['task_type'],
'query': result['query'],
'analysis_time': result['analysis_time']
}
if result['task_type'] == 'describe':
# 解析描述结果
description = raw_result.get('description', '')
parsed['description'] = description
# 提取关键信息（简单示例）
keywords = self._extract_keywords(description)
parsed['keywords'] = keywords
# 估算视频内容复杂度
parsed['complexity_score'] = len(description) / 100 # 简单评分
elif result['task_type'] == 'grounding':
# 解析定位结果
objects = raw_result.get('objects', [])
parsed['objects'] = objects
# 统计信息
parsed['object_count'] = len(objects)
# 提取时间分布
if objects:
timestamps = [obj.get('timestamp', 0) for obj in objects]
parsed['first_appearance'] = min(timestamps) if timestamps else 0
parsed['last_appearance'] = max(timestamps) if timestamps else 0
parsed['duration_seconds'] = parsed['last_appearance'] - parsed['first_appearance']
return parsed
def _extract_keywords(self, text, top_n=10):
"""从文本中提取关键词（简单实现）"""
import re
from collections import Counter
# 移除标点，转换为小写
words = re.findall(r'\b\w+\b', text.lower())
# 过滤停用词（简单列表）
stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by'}
filtered_words = [w for w in words if w not in stop_words and len(w) > 2]
# 统计词频
word_counts = Counter(filtered_words)
return [word for word, _ in word_counts.most_common(top_n)]
4.2 实际应用示例
让我们看几个具体的应用场景，看看Chord能帮我们做什么：
场景1：分析产品展示视频
假设你收集了一批竞品的产品展示视频，想了解他们是如何展示产品特性的：
# 分析产品展示视频
analyzer = ChordAnalyzer()
# 场景1：详细描述视频内容
result = analyzer.analyze_video(
video_path="product_demo.mp4",
task_type="describe",
query="详细描述这个产品演示视频的内容，包括产品特点、演示步骤、画面重点"
)
if result and result['success']:
description = result['result'].get('description', '')
print("视频描述：")
print(description[:500] + "..." if len(description) > 500 else description)
# 提取关键词
keywords = analyzer._extract_keywords(description)
print("\n关键词：", keywords)
# 场景2：定位特定产品特征
result2 = analyzer.analyze_video(
video_path="product_demo.mp4",
task_type="grounding",
query="产品logo"
)
if result2 and result2['success']:
objects = result2['result'].get('objects', [])
print(f"\n找到 {len(objects)} 个产品logo出现：")
for obj in objects:
print(f" 时间: {obj.get('timestamp', 0):.1f}秒, 位置: {obj.get('bbox', [])}")
场景2：监控品牌曝光
如果你需要监控某个品牌在视频中的曝光情况：
def monitor_brand_exposure(video_paths, brand_names):
"""监控品牌在视频中的曝光"""
analyzer = ChordAnalyzer()
brand_exposures = {}
for brand in brand_names:
brand_exposures[brand] = {
'total_videos': 0,
'total_appearances': 0,
'appearance_times': []
}
for video_path in video_paths:
print(f"\n分析视频: {Path(video_path).name}")
# 先用描述模式了解视频内容
describe_result = analyzer.analyze_video(video_path, task_type="describe")
if not describe_result or not describe_result['success']:
continue
description = describe_result['result'].get('description', '').lower()
# 检查每个品牌是否被提及
for brand in brand_names:
brand_lower = brand.lower()
if brand_lower in description:
brand_exposures[brand]['total_videos'] += 1
# 进一步用定位模式找出具体出现时间
grounding_result = analyzer.analyze_video(
video_path,
task_type="grounding",
query=brand
)
if grounding_result and grounding_result['success']:
objects = grounding_result['result'].get('objects', [])
brand_exposures[brand]['total_appearances'] += len(objects)
for obj in objects:
brand_exposures[brand]['appearance_times'].append({
'video': Path(video_path).name,
'timestamp': obj.get('timestamp', 0),
'position': obj.get('bbox', [])
})
return brand_exposures
# 使用示例
video_files = ["video1.mp4", "video2.mp4", "video3.mp4"]
brands_to_monitor = ["Nike", "Adidas", "Apple"]
exposure_data = monitor_brand_exposure(video_files, brands_to_monitor)
print("\n品牌曝光统计：")
for brand, data in exposure_data.items():
print(f"{brand}:")
print(f" 出现在 {data['total_videos']} 个视频中")
print(f" 总共出现 {data['total_appearances']} 次")
if data['appearance_times']:
print(f" 首次出现: {data['appearance_times'][0]['video']} 的 {data['appearance_times'][0]['timestamp']}秒")
4.3 分析结果存储与查询
分析完的视频数据需要妥善保存，方便后续查询和分析：
import sqlite3
from datetime import datetime
class VideoAnalysisDB:
"""视频分析数据库"""
def __init__(self, db_path='video_analysis.db'):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""初始化数据库表"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 视频基本信息表
cursor.execute('''
CREATE TABLE IF NOT EXISTS videos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id TEXT UNIQUE NOT NULL,
title TEXT NOT NULL,
url TEXT NOT NULL,
platform TEXT,
duration REAL,
file_path TEXT,
download_time TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 视频分析结果表
cursor.execute('''
CREATE TABLE IF NOT EXISTS analysis_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id TEXT NOT NULL,
task_type TEXT NOT NULL,
query TEXT,
raw_result TEXT,
parsed_result TEXT,
analysis_time REAL,
analyzed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (video_id) REFERENCES videos (video_id)
)
''')
# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_videos_video_id ON videos(video_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_results_video_id ON analysis_results(video_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_results_task_type ON analysis_results(task_type)')
conn.commit()
conn.close()
def save_video_info(self, video_info):
"""保存视频信息"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO videos
(video_id, title, url, platform, duration, file_path, download_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
video_info.get('video_id', ''),
video_info.get('title', ''),
video_info.get('url', ''),
video_info.get('platform', ''),
video_info.get('duration', 0),
video_info.get('file_path', ''),
video_info.get('download_time', datetime.now().isoformat())
))
conn.commit()
conn.close()
return True
def save_analysis_result(self, video_id, analysis_result):
"""保存分析结果"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO analysis_results
(video_id, task_type, query, raw_result, parsed_result, analysis_time)
VALUES (?, ?, ?, ?, ?, ?)
''', (
video_id,
analysis_result.get('task_type', ''),
analysis_result.get('query', ''),
str(analysis_result.get('result', {})),
str(analysis_result.get('parsed', {})),
analysis_result.get('analysis_time', 0)
))
conn.commit()
conn.close()
return True
def get_videos_by_keyword(self, keyword, task_type='describe', limit=10):
"""根据关键词查询视频"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT v.*, a.parsed_result
FROM videos v
JOIN analysis_results a ON v.video_id = a.video_id
WHERE a.task_type = ?
AND (a.parsed_result LIKE ? OR v.title LIKE ?)
ORDER BY a.analyzed_at DESC
LIMIT ?
''', (task_type, f'%{keyword}%', f'%{keyword}%', limit))
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
conn.close()
return [dict(zip(columns, row)) for row in rows]
def get_videos_with_objects(self, object_name, min_count=1):
"""查询包含特定对象的视频"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT v.*, a.parsed_result
FROM videos v
JOIN analysis_results a ON v.video_id = a.video_id
WHERE a.task_type = 'grounding'
AND a.query LIKE ?
AND a.parsed_result LIKE ?
ORDER BY a.analyzed_at DESC
''', (f'%{object_name}%', f'%{object_name}%'))
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
conn.close()
# 进一步过滤
filtered = []
for row in rows:
parsed_result = eval(row[columns.index('parsed_result')])
objects = parsed_result.get('objects', [])
if len(objects) >= min_count:
filtered.append(dict(zip(columns, row)))
return filtered
def get_statistics(self):
"""获取统计信息"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
stats = {}
# 视频总数
cursor.execute('SELECT COUNT(*) FROM videos')
stats['total_videos'] = cursor.fetchone()[0]
# 分析结果总数
cursor.execute('SELECT COUNT(*) FROM analysis_results')
stats['total_analysis'] = cursor.fetchone()[0]
# 按任务类型统计
cursor.execute('SELECT task_type, COUNT(*) FROM analysis_results GROUP BY task_type')
stats['by_task_type'] = dict(cursor.fetchall())
# 最新分析时间
cursor.execute('SELECT MAX(analyzed_at) FROM analysis_results')
stats['latest_analysis'] = cursor.fetchone()[0]
conn.close()
return stats
5. 完整工作流：从采集到分析的全自动化
现在我们把所有组件组合起来，形成一个完整的自动化工作流：
import schedule
import time
from datetime import datetime
import threading
class VideoAutoCollector:
"""视频自动采集分析系统"""
def __init__(self, config_path='config/config.yaml'):
self.config = self._load_config(config_path)
self.logger = self._setup_logging()
# 初始化组件
self.crawler = StaticVideoCrawler(self.config['crawler'])
self.downloader = VideoDownloader(self.config['storage']['video_dir'])
self.analyzer = ChordAnalyzer(
chord_url=self.config['chord']['api_url'],
timeout=self.config['chord']['timeout']
)
self.db = VideoAnalysisDB(self.config['storage']['database'])
self.running = False
self.logger.info("视频自动采集分析系统初始化完成")
def _load_config(self, config_path):
"""加载配置文件"""
import yaml
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def _setup_logging(self):
"""设置日志"""
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('video_collector.log'),
logging.StreamHandler()
]
)
return logging.getLogger('VideoAutoCollector')
def crawl_task(self):
"""爬虫任务"""
self.logger.info("开始执行爬虫任务")
try:
# 这里替换为实际的爬取目标
target_urls = [
"https://example.com/videos/page1",
"https://example.com/videos/page2",
# 添加更多目标URL
]
# 执行爬虫
videos = self.crawler.run(target_urls)
self.logger.info(f"爬取到 {len(videos)} 个视频")
# 保存视频信息到数据库
for video in videos:
self.db.save_video_info(video)
return videos
except Exception as e:
self.logger.error(f"爬虫任务失败: {e}")
return []
def download_task(self, batch_size=5):
"""下载任务"""
self.logger.info("开始执行下载任务")
# 从数据库获取未下载的视频
conn = sqlite3.connect(self.db.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT video_id, url FROM videos
WHERE file_path IS NULL OR file_path = ''
LIMIT ?
''', (batch_size,))
pending_videos = cursor.fetchall()
conn.close()
downloaded = 0
for video_id, url in pending_videos:
self.logger.info(f"下载视频: {video_id}")
file_path = self.downloader.download_video(url, video_id)
if file_path:
# 更新数据库中的文件路径
conn = sqlite3.connect(self.db.db_path)
cursor = conn.cursor()
cursor.execute(
'UPDATE videos SET file_path = ?, download_time = ? WHERE video_id = ?',
(file_path, datetime.now().isoformat(), video_id)
)
conn.commit()
conn.close()
downloaded += 1
self.logger.info(f"下载完成: {downloaded}/{len(pending_videos)} 个视频")
return downloaded
def analyze_task(self, batch_size=3):
"""分析任务"""
self.logger.info("开始执行分析任务")
# 从数据库获取已下载但未分析的视频
conn = sqlite3.connect(self.db.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT v.video_id, v.title, v.file_path
FROM videos v
LEFT JOIN analysis_results a ON v.video_id = a.video_id AND a.task_type = 'describe'
WHERE v.file_path IS NOT NULL AND v.file_path != '' AND a.id IS NULL
LIMIT ?
''', (batch_size,))
pending_videos = cursor.fetchall()
conn.close()
analyzed = 0
for video_id, title, file_path in pending_videos:
self.logger.info(f"分析视频: {title[:50]}...")
# 分析视频（描述模式）
result = self.analyzer.analyze_video(
file_path,
task_type="describe",
query="详细描述视频内容"
)
if result and result['success']:
# 解析结果
parsed = self.analyzer.parse_analysis_result(result, {
'video_id': video_id,
'title': title
})
if parsed:
# 保存结果
self.db.save_analysis_result(video_id, {
'task_type': 'describe',
'query': '详细描述视频内容',
'result': result['result'],
'parsed': parsed,
'analysis_time': result['analysis_time']
})
analyzed += 1
# 根据描述内容，决定是否需要进一步分析
description = result['result'].get('description', '').lower()
# 如果视频包含特定内容，进行定位分析
if any(keyword in description for keyword in ['人物', '物体', '动作', '产品']):
self.logger.info(f"视频包含可定位内容，进行进一步分析: {video_id}")
# 这里可以根据描述内容生成具体的查询
# 例如：如果描述中提到"人物"，就查询"人物"
grounding_result = self.analyzer.analyze_video(
file_path,
task_type="grounding",
query="人物" # 这里可以更智能地生成查询
)
if grounding_result and grounding_result['success']:
parsed_grounding = self.analyzer.parse_analysis_result(grounding_result, {
'video_id': video_id,
'title': title
})
self.db.save_analysis_result(video_id, {
'task_type': 'grounding',
'query': '人物',
'result': grounding_result['result'],
'parsed': parsed_grounding,
'analysis_time': grounding_result['analysis_time']
})
self.logger.info(f"分析完成: {analyzed}/{len(pending_videos)} 个视频")
return analyzed
def run_once(self):
"""执行一次完整流程"""
self.logger.info("开始执行完整工作流")
# 1. 爬取视频
videos = self.crawl_task()
# 2. 下载视频
if videos:
self.download_task(batch_size=5)
# 3. 分析视频
self.analyze_task(batch_size=3)
# 4. 清理旧文件
self.downloader.cleanup_old_videos(keep_days=7)
self.logger.info("工作流执行完成")
def start_scheduler(self):
"""启动定时调度"""
self.running = True
# 设置定时任务
schedule.every(1).hours.do(self.run_once)
self.logger.info("定时调度器已启动，每小时执行一次")
# 运行调度器
while self.running:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次
def stop(self):
"""停止系统"""
self.running = False
self.logger.info("系统正在停止...")
def run_interactive(self):
"""交互式运行"""
print("=" * 50)
print("视频自动采集分析系统")
print("=" * 50)
while True:
print("\n请选择操作：")
print("1. 执行一次完整流程")
print("2. 仅执行爬虫任务")
print("3. 仅执行下载任务")
print("4. 仅执行分析任务")
print("5. 查看统计信息")
print("6. 查询视频数据")
print("7. 启动定时运行")
print("8. 退出")
choice = input("请输入选项 (1-8): ").strip()
if choice == '1':
self.run_once()
elif choice == '2':
self.crawl_task()
elif choice == '3':
batch_size = input("请输入批量大小 (默认5): ").strip()
batch_size = int(batch_size) if batch_size else 5
self.download_task(batch_size)
elif choice == '4':
batch_size = input("请输入批量大小 (默认3): ").strip()
batch_size = int(batch_size) if batch_size else 3
self.analyze_task(batch_size)
elif choice == '5':
stats = self.db.get_statistics()
print("\n系统统计：")
for key, value in stats.items():
print(f" {key}: {value}")
elif choice == '6':
keyword = input("请输入查询关键词: ").strip()
results = self.db.get_videos_by_keyword(keyword, limit=5)
print(f"\n找到 {len(results)} 个相关视频：")
for i, video in enumerate(results, 1):
print(f"{i}. {video['title'][:50]}... (ID: {video['video_id']})")
elif choice == '7':
print("启动定时运行，按Ctrl+C停止")
try:
self.start_scheduler()
except KeyboardInterrupt:
self.stop()
elif choice == '8':
print("感谢使用，再见！")
break
else:
print("无效选项，请重新输入")
# 使用示例
if __name__ == "__main__":
collector = VideoAutoCollector()
# 交互式运行
collector.run_interactive()
# 或者直接运行一次
# collector.run_once()
# 或者启动定时运行
# collector.start_scheduler()
6. 实际应用案例与效果展示
理论说再多，不如看实际效果。下面通过几个真实场景，展示这个系统能做什么。
6.1 案例一：竞品视频监控
场景：你负责监控3个主要竞品的短视频账号，需要每天收集他们的新视频，分析内容特点。
配置：
# 配置监控目标
competitors = [
{
'name': '品牌A',
'urls': [
'https://platform.com/user/brandA/videos',
'https://platform.com/user/brandA/shorts'
],
'keywords': ['新品', '促销', '活动', '教程']
},
{
'name': '品牌B',
'urls': [
'https://platform.com/user/brandB/videos',
'https://platform.com/user/brandB/live'
],
'keywords': ['直播', '评测', '用户', '反馈']
},
{
'name': '品牌C',
'urls': [
'https://platform.com/user/brandC/videos'
],
'keywords': ['故事', '情感', '品牌', '价值']
}
]
# 运行监控
def monitor_competitors(competitors, days=7):
"""监控竞品视频"""
results = {}
for competitor in competitors:
print(f"\n监控 {competitor['name']}...")
# 收集视频
all_videos = []
for url in competitor['urls']:
crawler = StaticVideoCrawler(config)
videos = crawler.run([url])
all_videos.extend(videos)
print(f" 收集到 {len(all_videos)} 个视频")
# 分析视频内容
analyzer = ChordAnalyzer()
competitor_results = []
for video in all_videos[:5]: # 先分析前5个
print(f" 分析视频: {video['title'][:30]}...")
# 下载视频
downloader = VideoDownloader()
video_path = downloader.download_video(video['url'], video['video_id'])
if video_path:
# 分析视频
result = analyzer.analyze_video(
video_path,
task_type="describe",
query="详细描述视频内容，包括主题、画面、人物、动作"
)
if result and result['success']:
description = result['result'].get('description', '')
# 检查是否包含关注的关键词
matched_keywords = []
for keyword in competitor['keywords']:
if keyword in description:
matched_keywords.append(keyword)
competitor_results.append({
'title': video['title'],
'url': video['url'],
'description': description[:200] + '...' if len(description) > 200 else description,
'matched_keywords': matched_keywords,
'match_count': len(matched_keywords)
})
results[competitor['name']] = {
'total_videos': len(all_videos),
'analyzed_videos': len(competitor_results),
'videos': competitor_results
}
return results
# 生成报告
def generate_competitor_report(results):
"""生成竞品分析报告"""
report = "竞品视频分析报告\n"
report += "=" * 50 + "\n\n"
for competitor_name, data in results.items():
report += f"{competitor_name}:\n"
report += f" 总视频数: {data['total_videos']}\n"
report += f" 已分析数: {data['analyzed_videos']}\n\n"
if data['videos']:
report += " 最新视频分析:\n"
for video in data['videos'][:3]: # 显示最新的3个
report += f" - {video['title'][:40]}...\n"
report += f" 匹配关键词: {', '.join(video['matched_keywords']) if video['matched_keywords'] else '无'}\n"
report += f" 内容摘要: {video['description']}\n\n"
report += "-" * 50 + "\n"
return report
# 执行监控
results = monitor_competitors(competitors)
report = generate_competitor_report(results)
print(report)
输出效果：
竞品视频分析报告
==================================================
品牌A:
总视频数: 24
已分析数: 5
最新视频分析:
- 新品发布会现场实况...
匹配关键词: 新品, 活动
内容摘要: 视频展示了品牌A的新品发布会现场，主讲人在舞台中央介绍新产品特性，观众席坐满了人，现场气氛热烈...
- 产品使用教程：如何快速上手...
匹配关键词: 教程
内容摘要: 视频逐步演示了产品的使用方法，从开箱到基本操作，画面清晰，讲解详细...
--------------------------------------------------
品牌B:
总视频数: 18
已分析数: 5
最新视频分析:
- 直播回顾：用户问答环节...
匹配关键词: 直播, 用户
内容摘要: 视频记录了品牌B的直播活动，主播在回答用户提问，画面右侧显示用户评论，互动频繁...
- 产品评测：真实用户反馈...
匹配关键词: 评测, 用户, 反馈
内容摘要: 视频采访了多位真实用户，分享使用产品的体验，画面在不同用户间切换，表情自然...
6.2 案例二：热点内容分析
场景：你想了解当前热门视频的内容特点，找出受欢迎的视频有哪些共同点。
def analyze_hot_videos(topic, max_videos=20):
"""分析某个话题的热门视频"""
print(f"分析话题 '{topic}' 的热门视频...")
# 1. 爬取热门视频
hot_videos = crawl_hot_videos(topic, max_videos)
print(f"爬取到 {len(hot_videos)} 个热门视频")
# 2. 分析视频内容
analyzer = ChordAnalyzer()
all_analysis = []
for i, video in enumerate(hot_videos, 1):
print(f"分析进度: {i}/{len(hot_videos)} - {video['title'][:30]}...")
# 下载视频（仅下载前30秒以节省时间）
downloader = VideoDownloader()
video_path = downloader.download_video(video['url'], video['video_id'])
if video_path:
# 分析视频
result = analyzer.analyze_video(
video_path,
task_type="describe",
query="详细描述视频的前30秒内容，包括画面、人物、动作、场景"
)
if result and result['success']:
parsed = analyzer.parse_analysis_result(result, video)
all_analysis.append(parsed)
# 3. 分析共同特征
common_features = find_common_features(all_analysis)
# 4. 生成分析报告
report = generate_hot_video_report(topic, hot_videos, all_analysis, common_features)
return report
def find_common_features(analysis_results):
"""找出共同特征"""
from collections import Counter
# 统计关键词出现频率
all_keywords = []
for result in analysis_results:
keywords = result.get('keywords', [])
all_keywords.extend(keywords)
keyword_counts = Counter(all_keywords)
# 统计常见元素
common_elements = {
'top_keywords': keyword_counts.most_common(10),
'avg_complexity': sum(r.get('complexity_score', 0) for r in analysis_results) / len(analysis_results) if analysis_results else 0,
'total_videos': len(analysis_results)
}
return common_elements
def generate_hot_video_report(topic, videos, analysis, features):
"""生成热门视频分析报告"""
report = f"【{topic}】热门视频分析报告\n"
report += "=" * 60 + "\n\n"
report += "一、总体概况\n"
report += f"分析视频数量: {features['total_videos']} 个\n"
report += f"平均内容复杂度: {features['avg_complexity']:.2f}\n\n"
report += "二、热门关键词\n"
for keyword, count in features['top_keywords']:
percentage = count / features['total_videos'] * 100
report += f" {keyword}: {count}次 ({percentage:.1f}%)\n"
report += "\n三、视频内容分析\n"
for i, (video, analysis_result) in enumerate(zip(videos[:5], analysis[:5]), 1):
report += f"{i}. {video['title'][:40]}...\n"
report += f" 点赞: {video.get('likes', 'N/A')} | 评论: {video.get('comments', 'N/A')}\n"
if 'keywords' in analysis_result:
report += f" 关键词: {', '.join(analysis_result['keywords'][:5])}\n"
if 'description' in analysis_result:
desc = analysis_result['description']
if len(desc) > 100:
desc = desc[:100] + "..."
report += f" 内容摘要: {desc}\n"
report += "\n"
report += "四、内容创作建议\n"
report += "基于分析结果，建议：\n"
# 根据分析结果给出建议
top_keywords = [k for k, _ in features['top_keywords'][:3]]
if top_keywords:
report += f"1. 重点使用高频关键词: {', '.join(top_keywords)}\n"
if features['avg_complexity'] > 0.5:
report += "2. 当前热门视频内容较丰富，建议保持一定的信息密度\n"
else:
report += "2. 当前热门视频内容较简单，建议突出重点，避免信息过载\n"
report += "3. 前5秒出现关键画面能有效提升留存率\n"
report += "4. 人物出镜+场景化演示的组合效果较好\n"
return report
# 执行分析
topic = "健身教程"
report = analyze_hot_videos(topic, max_videos=10)
print(report)
输出效果：
【健身教程】热门视频分析报告
============================================================
一、总体概况
分析视频数量: 8 个
平均内容复杂度: 0.67
二、热门关键词
动作: 8次 (100.0%)
演示: 7次 (87.5%)
教练: 6次 (75.0%)
器械: 5次 (62.5%)
节奏: 4次 (50.0%)
三、视频内容分析
1. 居家健身：10分钟全身燃脂...
点赞: 15.2万 | 评论: 3200
关键词: 动作, 演示, 节奏, 音乐, 计时
内容摘要: 视频中教练演示了10个居家健身动作，每个动作30秒，节奏明快，背景音乐动感...
2. 健身房入门：器械使用全指南...
点赞: 8.7万 | 评论: 1500
关键词: 器械, 演示, 安全, 姿势, 重量
内容摘要: 视频详细演示了健身房常见器械的使用方法，重点强调了安全注意事项和正确姿势...
四、内容创作建议
基于分析结果，建议：
1. 重点使用高频关键词: 动作, 演示, 教练
2. 当前热门视频内容较丰富，建议保持一定的信息密度
3. 前5秒出现关键画面能有效提升留存率
4. 人物出镜+场景化演示的组合效果较好
6.3 案例三：自动生成视频摘要
场景：你需要快速了解长视频的内容，生成文字摘要。
def generate_video_summary(video_path, max_points=5):
"""生成视频摘要"""
analyzer = ChordAnalyzer()
# 分析视频
result = analyzer.analyze_video(
video_path,
task_type="describe",
query="详细描述视频内容，按时间顺序列出关键点"
)
if not result or not result['success']:
return "视频分析失败"
description = result['result'].get('description', '')
# 提取关键时间点（简单实现）
lines = description.split('\n')
key_points = []
for line in lines:
line = line.strip()
if not line:
continue
# 查找时间信息（简单匹配）
import re
time_match = re.search(r'(\d+)[分:]?(\d*)秒?', line)
if time_match or any(keyword in line for keyword in ['首先', '然后', '接着', '最后', '开始', '结束']):
key_points.append(line)
# 如果没有找到时间点，就取前几句
if not key_points:
key_points = [line for line in lines if line][:max_points]
# 生成摘要
summary = "视频内容摘要：\n\n"
for i, point in enumerate(key_points[:max_points], 1):
summary += f"{i}. {point}\n"
# 添加总体描述
if len(description) > 200:
summary += f"\n总体描述：{description[:200]}..."
else:
summary += f"\n总体描述：{description}"
return summary
# 使用示例
summary = generate_video_summary("product_tutorial.mp4")
print(summary)
7. 常见问题与解决方案
在实际使用中，你可能会遇到一些问题。这里总结了一些常见问题和解决方法：
7.1 爬虫被屏蔽怎么办？
问题：网站检测到爬虫，返回错误或限制访问。
解决方案：
- 降低请求频率：在请求之间添加随机延迟
- 轮换User-Agent：使用不同的浏览器标识
- 使用代理IP：特别是对于严格限制的网站
- 模拟人类行为：随机滚动页面、点击等
import random
import time
def smart_request(url, session, headers):
"""智能请求，避免被屏蔽"""
# 随机延迟
delay = random.uniform(1, 3)
time.sleep(delay)
# 随机User-Agent
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
]
headers['User-Agent'] = random.choice(user_agents)
try:
response = session.get(url, headers=headers, timeout=10)
return response
except Exception as e:
print(f"请求失败: {e}")
return None
7.2 视频下载太慢怎么办？
问题：视频文件大，下载耗时太长。
解决方案：
- 限制视频质量：下载720p而非1080p
- 并行下载：同时下载多个视频
- 断点续传：确保网络中断后能继续下载
- 只下载前N秒：对于分析来说，通常不需要完整视频
def download_partial_video(url, video_id, duration=30):
"""只下载视频的前N秒"""
output_path = f"./videos/{video_id}_partial.mp4"
# 使用yt-dlp下载指定时长
ydl_opts = {
'outtmpl': output_path.replace('.mp4', ''),
'format': 'best[height<=720]',
'quiet': True,
'no_warnings': True,
'playlistend': 1,
'match_filter': f'duration < {duration}',
}
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([url])
return output_path
except Exception as e:
print(f"部分下载失败: {e}")
return None
7.3 Chord分析结果不准确怎么办？
问题：Chord对某些视频的分析结果不理想。
解决方案：
- 优化查询：使用更具体、明确的描述
- 分段分析：将长视频分成多个片段分别分析
- 多角度分析：用不同的问题多次分析同一视频
- 人工校验：对重要结果进行人工抽查
def analyze_video_multiple_queries(video_path, queries):
"""用多个查询分析同一视频"""
analyzer = ChordAnalyzer()
results = []
for query in queries:
print(f"分析查询: {query}")
result = analyzer.analyze_video(
video_path,
task_type="describe",
query=query
)
if result and result['success']:
results.append({
'query': query,
'result': result['result'].get('description', '')
})
# 合并结果
combined = combine_results(results)
return combined
def combine_results(results):
"""合并多个分析结果"""
if not results:
return ""
# 简单合并：取最长的描述
longest = max(results, key=lambda x: len(x['result']))
return longest['result']
# 使用示例
queries = [
"详细描述视频中的人物和动作",
"描述视频的场景和背景",
"说明视频的主要内容和目的"
]
result = analyze_video_multiple_queries("demo_video.mp4", queries)
print("合并后的分析结果：")
print(result[:500] + "..." if len(result) > 500 else result)
7.4 系统资源不足怎么办？
问题：同时处理多个视频时，内存或CPU不足。
解决方案：
- 限制并发数：一次只处理一个视频
- 分批处理：将大量视频分成小批次
- 监控资源使用：自动暂停或调整
- 清理临时文件：及时删除已处理的视频
import psutil
import os
def check_system_resources():
"""检查系统资源"""
memory = psutil.virtual_memory()
cpu_percent = psutil.cpu_percent(interval=1)
return {
'memory_percent': memory.percent,
'cpu_percent': cpu_percent,
'memory_available': memory.available // 1024 // 1024 # MB
}
def process_videos_safely(video_paths, max_memory=80, max_cpu=90):
"""安全处理视频，避免资源过载"""
processed = []
for video_path in video_paths:
# 检查系统资源
resources = check_system_resources()
if resources['memory_percent'] > max_memory:
print(f"内存使用过高 ({resources['memory_percent']}%)，暂停处理")
time.sleep(60)
continue
if resources['cpu_percent'] > max_cpu:
print(f"CPU使用过高 ({resources['cpu_percent']}%)，暂停处理")
time.sleep(30)
continue
# 处理视频
result = process_single_video(video_path)
if result:
processed.append(result)
# 处理完成后清理内存
import gc
gc.collect()
return processed
8. 总结
通过这篇文章，我们完成了一个完整的视频数据自动化采集与分析系统的搭建。从最开始的Python爬虫编写，到Chord视频理解工具的集成，再到最后的自动化工作流，每一步都力求简单明了，即使是没有太多编程经验的朋友也能跟着做出来。
这个系统的核心价值在于自动化和智能化。自动化让我们从重复的手工操作中解放出来，智能化（通过Chord工具）让计算机能“看懂”视频内容，提取出对我们有用的信息。
回顾一下整个流程：
- 爬虫采集：自动从目标网站获取视频信息和链接
- 视频下载：将视频文件保存到本地
- 智能分析：用Chord工具分析视频内容，提取时空信息
- 数据存储：将分析结果结构化保存，方便查询
- 自动运行：设置定时任务，让系统自动工作
实际使用中，你可以根据自己的需求调整这个系统。比如：
- 如果你只关心特定品牌的出现，可以调整Chord的查询关键词
- 如果你需要分析视频的情感倾向，可以在描述查询中强调情感词汇
- 如果你想监控某个话题的热度，可以定期爬取相关视频并分析内容变化
最重要的是，这个系统是一个起点，而不是终点。你可以基于这个框架，添加更多功能，比如：
- 自动生成日报/周报
- 设置关键词告警（当出现特定内容时通知你）
- 与其他系
邀请您加入社区
更多推荐
0
0- 0
已为社区贡献4条内容
所有评论(0)