别再手动处理字幕了!这个开源项目让字幕秒变可查询API,数据提取效率提升10倍
为什么这个项目值得关注
在日常开发和数据处理工作中,字幕文件几乎是每个程序员都会遇到的格式。无论是视频字幕、音频转录文本,还是会议记录,字幕文件的处理都让开发者头疼不已。传统的处理方式往往是手动解析、逐行读取、编写大量重复代码。而今天要介绍的这个开源项目 Wei-Shaw/sub2api,将彻底改变你处理字幕的方式。
这个项目解决了什么问题?
想象一下,你需要从一份包含数万行的字幕文件中提取特定时间段的对白,或者搜索某个关键词出现的所有位置。按照传统方法,你需要编写解析脚本、处理时间格式转换、处理编码问题——光是这些准备工作就可能耗费你数小时。而 sub2api 提供了另一种思路:它将字幕文件转换为一个可查询的 API 服务,你只需发送 HTTP 请求,就能完成所有这些操作。
为什么选择 sub2api?
首先,它基于 Python 开发,这意味着它拥有庞大且成熟的生态系统支持。其次,API 架构让它可以轻松集成到任何技术栈中,无论你是用 Node.js、Java 还是 Go,都可以通过 HTTP 请求与它交互。更重要的是,它的源码完全开放,你可以根据自身需求进行定制和扩展,这种灵活性是许多商业方案无法提供的。
对于需要处理字幕数据的开发者来说,这不仅仅是一个工具,更是一种全新的工作方式。它将复杂的字幕处理逻辑封装成简洁的 API 接口,让你可以专注于业务逻辑本身,而不是被数据格式转换的细节所困扰。
环境搭建:快速启动指南
前置条件
在开始使用 sub2api 之前,你需要确保开发环境中已经安装了以下组件。这些是运行项目的基础依赖,缺少任何一个都可能导致后续步骤出现问题。
Python 版本是首要确认的事项。建议使用 Python 3.8 或更高版本,因为项目依赖的一些核心库在新版本中得到了性能优化和 bug 修复。你可以通过在终端中执行 python –version 或 python3 –version 来检查当前安装的 Python 版本。如果显示的版本号低于 3.8,或者提示找不到 python 命令,你需要先下载并安装 Python。
Git 版本控制工具也是必需的。sub2api 的源码托管在 GitHub 上,你需要使用 git 来克隆仓库。如果你使用的是 macOS 或 Linux 系统,git 通常已经预装。Windows 用户可以在 Git 官网 下载安装包,安装过程中记得勾选添加到 PATH 环境变量的选项。
pip 是 Python 的包管理工具,用于安装项目依赖。在大多数 Python 安装包中,pip 已经包含在内。你可以通过 pip –version 验证它是否可用。
安装步骤详解
第一步是克隆项目仓库到本地。打开终端或命令行工具,执行以下命令。这个命令会将 GitHub 上的源码下载到当前目录下的 sub2api 文件夹中。
git clone https://github.com/Wei-Shaw/sub2api.git
克隆完成后,进入项目目录查看项目结构。这是了解一个新项目的标准流程,通过查看文件组织方式,你可以快速理解项目的架构设计。
cd sub2api
ls -la
你将看到项目包含几个核心文件和目录。main.py 或 app.py 是项目的入口文件,通常包含应用的启动逻辑。requirements.txt 文件列出了项目所需的所有 Python 依赖包及其版本号。src 或 sub2api 目录通常包含项目的核心源代码。tests 目录包含单元测试和集成测试代码。
接下来是安装项目依赖。这是非常关键的一步,因为如果依赖没有正确安装,项目将无法运行。推荐的做法是创建一个独立的虚拟环境,这样可以避免依赖冲突问题。
python -m venv venv
source venv/bin/activate # Linux 和 macOS
# venv\Scripts\activate # Windows
虚拟环境创建并激活后,安装依赖就变得安全且简单。pip 会读取 requirements.txt 文件,自动下载并安装所有列出的包。这个过程可能需要几分钟,取决于你的网络速度和系统性能。
pip install -r requirements.txt
安装完成后,验证安装是否成功是一个好习惯。你可以尝试导入项目的主要模块,如果没有报错,说明安装过程顺利完成。
python -c "import sub2api; print('安装成功')"
配置说明
sub2api 的配置通常通过 config.yaml 或 .env 文件管理。在首次使用时,你需要根据实际需求调整这些配置文件。配置文件控制着 API 的运行端口、字幕文件的默认编码、日志级别等参数。
# config.yaml 示例配置
api:
host: "0.0.0.0"
port: 8000
debug: false
subtitle:
default_encoding: "utf-8"
supported_formats:
- "srt"
- "vtt"
- "ass"
logging:
level: "INFO"
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
如果你计划处理中文或其他非拉丁语系的字幕文件,确保将 default_encoding 设置为合适的字符编码。常见的编码格式包括 utf-8、gbk、gb2312 等。错误地设置编码可能导致字幕内容显示为乱码,这在处理中文内容时尤其容易出现。
核心功能详解
字幕解析引擎
sub2api 的核心是一个强大的字幕解析引擎,它能够处理多种主流字幕格式。这个解析引擎的设计遵循了模块化原则,每种格式都有独立的解析器实现,但它们都遵循统一的接口规范。这种设计使得添加新格式支持变得非常便捷,只需实现统一的接口方法即可。
对于 SRT 格式(SubRip Text),这是最常见的字幕格式,文件扩展名为 .srt。其基本结构包含序号、时间范围和文本内容三部分。解析器需要正确处理时间格式的转换,将形如 “00:01:23,456 –> 00:01:25,789” 的时间码转换为可计算的秒数或时间戳对象。同时,多行文本的处理也是一个技术点,因为某些字幕条目可能包含多行对白。
VTT 格式(WebVTT)作为网页字幕的标准格式,有着自己的特性。它以 “WEBVTT” 开头,支持 CSS 样式注释,包含 cue 设置等高级特性。解析器需要识别并正确处理这些特性,同时忽略网页渲染不需要的元信息。
ASS/SSA 格式(Advanced SubStation Alpha)是最复杂的字幕格式之一,支持高级样式、动画效果、定位信息等。这种格式的文件通常用于动漫字幕或需要精细控制的场景。解析 ASS 格式需要处理样式定义、事件脚本、样式覆盖等多个部分。
# sub2api 内部字幕解析的简化示例
class SubtitleParser:
def __init__(self, encoding='utf-8'):
self.encoding = encoding
def parse_file(self, file_path):
"""解析字幕文件并返回结构化数据"""
with open(file_path, 'r', encoding=self.encoding) as f:
content = f.read()
file_ext = file_path.split('.')[-1].lower()
if file_ext == 'srt':
return self._parse_srt(content)
elif file_ext == 'vtt':
return self._parse_vtt(content)
elif file_ext == 'ass':
return self._parse_ass(content)
else:
raise ValueError(f"不支持的格式: {file_ext}")
def _parse_srt(self, content):
"""解析 SRT 格式"""
subtitles = []
blocks = content.strip().split('\n\n')
for block in blocks:
lines = block.strip().split('\n')
if len(lines) < 3:
continue
# 解析序号
index = int(lines[0])
# 解析时间范围
time_range = lines[1]
start_time, end_time = self._parse_srt_time(time_range)
# 解析文本内容
text = '\n'.join(lines[2:])
subtitles.append({
'index': index,
'start': start_time,
'end': end_time,
'text': text
})
return subtitles
def _parse_srt_time(self, time_str):
"""将 SRT 时间格式转换为秒数"""
# 格式: 00:01:23,456 --> 00:01:25,789
pattern = r'(\d{2}):(\d{2}):(\d{2}),(\d{3})\s*-->\s*(\d{2}):(\d{2}):(\d{2}),(\d{3})'
match = re.match(pattern, time_str)
if not match:
return None, None
# 解析开始时间
start_h, start_m, start_s, start_ms = map(int, match.groups()[:4])
start_seconds = start_h * 3600 + start_m * 60 + start_s + start_ms / 1000
# 解析结束时间
end_h, end_m, end_s, end_ms = map(int, match.groups()[4:])
end_seconds = end_h * 3600 + end_m * 60 + end_s + end_ms / 1000
return start_seconds, end_seconds
REST API 服务
sub2api 提供了一个完整的 RESTful API 服务层,这是项目最核心的功能之一。API 设计遵循标准的 HTTP 语义,使用 JSON 作为数据交换格式,让它可以与任何支持 HTTP 请求的客户端无缝集成。
API 服务基于 FastAPI 框架构建,这是一个现代、快速的 Python Web 框架。FastAPI 自动提供交互式 API 文档(Swagger UI),这意味着你无需额外的文档工具,就能直观地查看和测试所有可用端点。
基本的 API 端点包括上传字幕文件、查询字幕内容、搜索关键词、获取特定时间点的字幕等。每个端点都有明确的输入参数定义和返回数据格式说明。
上传字幕文件时,API 会自动检测文件格式、编码,解析字幕内容,并返回一个唯一的内容标识符。这个标识符在后续的查询操作中用于指定要查询的字幕文件。
# API 端点示例(FastAPI 风格)
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
app = FastAPI(title="sub2api", version="1.0.0")
class SubtitleQuery(BaseModel):
content_id: str
start_time: float = None
end_time: float = None
keyword: str = None
@app.post("/upload")
async def upload_subtitle(file: UploadFile = File(...)):
"""上传并解析字幕文件"""
try:
content = await file.read()
content_id = subtitle_service.add_subtitle(content, file.filename)
return {
"status": "success",
"content_id": content_id,
"filename": file.filename,
"subtitle_count": subtitle_service.get_count(content_id)
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.post("/query")
def query_subtitle(query: SubtitleQuery):
"""查询字幕内容"""
results = subtitle_service.query(
content_id=query.content_id,
start_time=query.start_time,
end_time=query.end_time,
keyword=query.keyword
)
return {
"status": "success",
"count": len(results),
"results": results
}
@app.get("/subtitle/{content_id}")
def get_subtitle(content_id: str):
"""获取字幕文件的所有内容"""
subtitle = subtitle_service.get_full_subtitle(content_id)
if not subtitle:
raise HTTPException(status_code=404, detail="字幕不存在")
return subtitle
查询与搜索功能
除了基础的字幕解析,sub2api 还提供了强大的查询和搜索功能。这些功能让你可以高效地从大量字幕数据中提取所需信息,而无需将整个文件加载到内存中处理。
时间范围查询是最常用的功能之一。你可以指定一个时间范围,API 将返回该时间段内的所有字幕条目。这对于需要获取视频特定片段对白的场景非常有用,比如制作台词本、生成时间戳索引等。
关键词搜索功能支持精确匹配和模糊匹配两种模式。精确匹配会找到包含完全相同关键词的字幕条目,而模糊匹配则会找到包含相近词汇的条目,容忍一定程度的拼写错误或词汇变体。
上下文提取功能允许你获取搜索结果的前后几条字幕。这在理解对话上下文时特别有用,特别是当你需要分析某句话的完整对话场景时。
# 查询功能的使用示例
import requests
# 基础查询设置
BASE_URL = "http://localhost:8000"
# 按时间范围查询
def query_by_time_range(content_id, start_seconds, end_seconds):
response = requests.post(
f"{BASE_URL}/query",
json={
"content_id": content_id,
"start_time": start_seconds,
"end_time": end_seconds
}
)
return response.json()
# 搜索关键词
def search_keyword(content_id, keyword, context_lines=2):
response = requests.post(
f"{BASE_URL}/query",
json={
"content_id": content_id,
"keyword": keyword,
"context_lines": context_lines
}
)
return response.json()
# 获取某个时间点前后的字幕
def get_context_at_time(content_id, timestamp, range_seconds=30):
response = requests.post(
f"{BASE_URL}/query",
json={
"content_id": content_id,
"start_time": timestamp - range_seconds,
"end_time": timestamp + range_seconds
}
)
return response.json()
实战教程:从入门到精通
第一步:启动 API 服务
在开始使用 sub2api 的各项功能之前,首先需要启动 API 服务。这是所有后续操作的基础,服务启动后才能接收和处理 HTTP 请求。
启动服务的命令非常简单。在项目根目录下,执行以下命令即可:
python -m uvicorn main:app --reload --host 0.0.0.0 --port 8000
这个命令使用 uvicorn 作为 ASGI 服务器运行 FastAPI 应用。–reload 参数启用热重载模式,代码修改后服务会自动重启,这在开发调试阶段非常方便。–host 0.0.0.0 表示服务监听所有网络接口,允许外部设备访问。–port 8000 指定服务运行在 8000 端口。
服务启动后,你会在终端看到类似以下的输出:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000
现在,打开浏览器访问 http://localhost:8000/docs,你将看到 FastAPI 自动生成的交互式文档页面。这个页面不仅展示了所有可用的 API 端点,还允许你直接在网页上测试各项功能。
第二步:上传并处理字幕文件
服务启动后,第一件事是上传字幕文件。sub2api 支持多种上传方式,包括通过 API 上传和命令行工具导入。
使用 API 上传是最直接的方式。你可以使用任何 HTTP 客户端发送请求。以下是使用 Python requests 库上传文件的示例代码:
import requests
# 上传 SRT 字幕文件
def upload_subtitle(file_path):
url = "http://localhost:8000/upload"
with open(file_path, 'rb') as f:
files = {'file': f}
response = requests.post(url, files=files)
if response.status_code == 200:
result = response.json()
print(f"上传成功!")
print(f"内容ID: {result['content_id']}")
print(f"字幕数量: {result['subtitle_count']}")
return result['content_id']
else:
print(f"上传失败: {response.text}")
return None
# 使用示例
content_id = upload_subtitle('path/to/your/subtitle.srt')
上传成功后,API 会返回一个 content_id,这个 ID 是后续所有操作的凭证。你需要妥善保存它,以便在查询时使用。
对于批量处理场景,命令行工具提供了更便捷的方式:
# 导入单个文件
sub2api import your_video.srt
# 批量导入目录下的所有字幕文件
sub2api import --batch ./subtitles_folder/
# 导入并指定标签
sub2api import your_video.srt --tag "movie" --tag "english"
第三步:掌握基础查询操作
上传字幕文件后,就可以通过 API 进行各种查询操作了。理解这些基础操作是高效使用 sub2api 的关键。
获取字幕全文是最简单的操作,它会返回字幕文件中的所有条目,按照时间顺序排列:
import requests
def get_all_subtitles(content_id):
url = f"http://localhost:8000/subtitle/{content_id}"
response = requests.get(url)
if response.status_code == 200:
data = response.json()
print(f"共有 {data['count']} 条字幕记录\n")
for item in data['subtitles'][:10]: # 只显示前10条
start = format_time(item['start'])
end = format_time(item['end'])
print(f"[{start} --> {end}]")
print(item['text'])
print("---")
def format_time(seconds):
"""将秒数转换为时:分:秒格式"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
# 获取前10条字幕预览
content_id = "abc123def456" # 替换为实际的内容ID
get_all_subtitles(content_id)
按时间范围查询允许你提取特定时间段的对白。这在制作视频剪辑参考、生成特定片段台词时特别有用:
def query_time_range(content_id, start_seconds, end_seconds):
"""
查询指定时间范围内的字幕
参数说明:
- content_id: 上传字幕时获得的唯一标识
- start_seconds: 开始时间(秒)
- end_seconds: 结束时间(秒)
"""
url = "http://localhost:8000/query"
payload = {
"content_id": content_id,
"start_time": start_seconds,
"end_time": end_seconds
}
response = requests.post(url, json=payload)
if response.status_code == 200:
data = response.json()
print(f"时间范围 {start_seconds}s - {end_seconds}s 内的字幕:\n")
for item in data['results']:
start = format_time(item['start'])
end = format_time(item['end'])
print(f"[{start} --> {end}] {item['text']}")
return data['results']
else:
print(f"查询失败: {response.text}")
return []
# 查询第5分钟到第6分钟的内容
query_time_range("abc123def456", 300, 360)
第四步:高级搜索技巧
除了基础的时间查询,sub2api 还提供了强大的搜索功能,让你能够快速定位目标内容。
关键词搜索是最常用的搜索方式。你可以直接搜索单词或短语,API 会返回所有包含该关键词的字幕条目:
def search_subtitles(content_id, keyword, max_results=50):
"""
搜索包含特定关键词的字幕
参数说明:
- content_id: 字幕内容的唯一标识
- keyword: 要搜索的关键词
- max_results: 最大返回结果数
"""
url = "http://localhost:8000/search"
payload = {
"content_id": content_id,
"keyword": keyword,
"max_results": max_results
}
response = requests.post(url, json=payload)
if response.status_code == 200:
data = response.json()
print(f"关键词 '{keyword}' 共找到 {data['count']} 处出现\n")
for i, item in enumerate(data['results'], 1):
start = format_time(item['start'])
print(f"{i}. [{start}] {item['text']}")
return data['results']
else:
print(f"搜索失败: {response.text}")
return []
# 搜索关键词 "hello"
search_subtitles("abc123def456", "hello")
带上下文的搜索可以让你看到关键词出现的前后内容,这对于理解对话场景至关重要:
def search_with_context(content_id, keyword, context_before=2, context_after=2):
"""
搜索关键词并返回上下文
参数说明:
- context_before: 结果前显示的字幕条数
- context_after: 结果后显示的字幕条数
"""
url = "http://localhost:8000/search"
payload = {
"content_id": content_id,
"keyword": keyword,
"context_before": context_before,
"context_after": context_after,
"include_context": True
}
response = requests.post(url, json=payload)
if response.status_code == 200:
data = response.json()
for result in data['results']:
start = format_time(result['start'])
print(f"\n=== 出现位置: {start} ===")
# 显示前置上下文
if 'context_before' in result:
for ctx in result['context_before']:
print(f" [前置] {ctx['text']}")
# 显示匹配结果,高亮显示
print(f">>> {result['text']} <<<")
# 显示后置上下文
if 'context_after' in result:
for ctx in result['context_after']:
print(f" [后置] {ctx['text']}")
return data['results']
else:
print(f"搜索失败: {response.text}")
return []
# 搜索并显示上下文
search_with_context("abc123def456", "important", context_before=3, context_after=2)
第五步:数据导出与转换
处理完字幕数据后,通常需要导出结果用于其他用途。sub2api 支持多种导出格式,满足不同场景的需求。
导出为纯文本格式是最简单的导出方式,适合用于制作台词本或快速阅读:
def export_to_text(content_id, output_path):
"""
导出字幕为纯文本格式
参数说明:
- content_id: 要导出的字幕ID
- output_path: 输出文件路径
"""
url = f"http://localhost:8000/export/text/{content_id}"
response = requests.get(url)
if response.status_code == 200:
with open(output_path, 'w', encoding='utf-8') as f:
f.write(response.text)
print(f"已导出到: {output_path}")
else:
print(f"导出失败: {response.text}")
# 导出为纯文本
export_to_text("abc123def456", "output.txt")
导出为带时间戳的格式,适合用于视频编辑软件的时间轴参考:
def export_with_timestamps(content_id, output_path):
"""
导出为带时间戳的文本格式
输出格式: [00:01:23] 字幕文本
"""
url = f"http://localhost:8000/export/timestamped/{content_id}"
response = requests.get(url)
if response.status_code == 200:
with open(output_path, 'w', encoding='utf-8') as f:
f.write(response.text)
print(f"已导出带时间戳版本到: {output_path}")
else:
print(f"导出失败: {response.text}")
# 导出带时间戳版本
export_with_timestamps("abc123def456", "output_timestamped.txt")
导出为 JSON 格式,便于程序进一步处理:
def export_to_json(content_id, output_path):
"""
导出为 JSON 格式,便于程序处理
"""
url = f"http://localhost:8000/export/json/{content_id}"
response = requests.get(url)
if response.status_code == 200:
import json
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(response.json(), f, ensure_ascii=False, indent=2)
print(f"已导出JSON到: {output_path}")
else:
print(f"导出失败: {response.text}")
# 导出为 JSON
export_to_json("abc123def456", "output.json")
第六步:Python SDK 高级用法
除了直接调用 API,sub2api 还提供了 Python SDK,让你在 Python 代码中更方便地使用各项功能。SDK 对 API 调用进行了封装,提供了更友好的接口。
from sub2api import SubtitleClient
# 初始化客户端
client = SubtitleClient(base_url="http://localhost:8000")
# 链式调用示例:上传 -> 搜索 -> 导出
def process_subtitle_workflow(file_path, search_keyword, output_path):
"""
完整的工作流程示例
包含:上传文件、搜索关键词、导出结果
"""
# 上传字幕文件
print("步骤1: 上传字幕文件...")
upload_result = client.upload(file_path)
content_id = upload_result['content_id']
print(f"上传成功,content_id: {content_id}")
# 搜索关键词
print(f"\n步骤2: 搜索关键词 '{search_keyword}'...")
search_results = client.search(
content_id=content_id,
keyword=search_keyword,
include_context=True,
context_lines=2
)
print(f"找到 {len(search_results)} 条匹配结果")
# 导出结果
print(f"\n步骤3: 导出结果...")
client.export_to_file(
content_id=content_id,
output_path=output_path,
format='text',
filter_keyword=search_keyword
)
print(f"结果已导出到: {output_path}")
return content_id, search_results
# 执行完整工作流
content_id, results = process_subtitle_workflow(
file_path='path/to/subtitle.srt',
search_keyword='重要',
output_path='search_results.txt'
)
SDK 还支持批量处理场景,适合需要处理大量字幕文件的用户:
def batch_process_subtitles(folder_path, keyword):
"""
批量处理文件夹中的所有字幕文件
参数说明:
- folder_path: 包含字幕文件的文件夹路径
- keyword: 要搜索的关键词
"""
import os
# 获取文件夹中所有字幕文件
subtitle_files = []
for file in os.listdir(folder_path):
if file.endswith(('.srt', '.vtt', '.ass')):
subtitle_files.append(os.path.join(folder_path, file))
print(f"找到 {len(subtitle_files)} 个字幕文件")
# 批量上传
uploaded = {}
for file_path in subtitle_files:
filename = os.path.basename(file_path)
print(f"上传: {filename}")
try:
result = client.upload(file_path)
uploaded[filename] = result['content_id']
except Exception as e:
print(f"上传失败 {filename}: {e}")
# 批量搜索
print("\n开始批量搜索...")
all_results = {}
for filename, content_id in uploaded.items():
print(f"搜索文件: {filename}")
try:
results = client.search(
content_id=content_id,
keyword=keyword
)
all_results[filename] = results
print(f" 找到 {len(results)} 条匹配")
except Exception as e:
print(f" 搜索失败 {filename}: {e}")
all_results[filename] = []
# 生成汇总报告
summary = []
summary.append(f"关键词 '{keyword}' 搜索汇总报告")
summary.append("=" * 50)
summary.append(f"总文件数: {len(subtitle_files)}")
summary.append(f"成功处理: {len(all_results)}")
summary.append(f"总匹配数: {sum(len(r) for r in all_results.values())}")
summary.append("=" * 50)
for filename, results in all_results.items():
summary.append(f"\n{filename}: {len(results)} 处匹配")
report = '\n'.join(summary)
print("\n" + report)
# 保存报告
with open('search_summary.txt', 'w', encoding='utf-8') as f:
f.write(report)
return all_results
# 执行批量处理
batch_process_subtitles('./subtitles', '关键台词')
典型应用场景
场景一:视频内容分析与索引
在需要对视频内容进行深入分析时,字幕文件是宝贵的信息来源。通过 sub2api,你可以快速构建视频内容的索引系统,实现关键词定位、时间戳关联等功能。
具体实现思路是先将所有字幕文件上传到系统中,然后对每条字幕记录进行分词处理,建立倒排索引。当用户搜索某个关键词时,系统可以快速返回所有出现位置,并支持跳转到对应的时间点。
def build_video_index(subtitle_files):
"""
构建视频字幕索引系统
返回:
- 关键词到时间戳的映射
- 每个视频的内容摘要
"""
index = {}
summaries = {}
for file_path in subtitle_files:
# 上传字幕
result = client.upload(file_path)
content_id = result['content_id']
# 获取所有字幕
subtitles = client.get_all(content_id)
# 构建索引
video_id = os.path.basename(file_path)
summaries[video_id] = {
'content_id': content_id,
'total_lines': len(subtitles),
'duration': subtitles[-1]['end'] if subtitles else 0
}
# 对每条字幕分词并建立索引
for sub in subtitles:
words = jieba.cut(sub['text']) # 使用 jieba 进行中文分词
for word in words:
if len(word) < 2: # 忽略单字
continue
if word not in index:
index[word] = []
index[word].append({
'video_id': video_id,
'timestamp': sub['start'],
'text': sub['text']
})
return index, summaries
# 使用示例
index, summaries = build_video_index(['video1.srt', 'video2.srt', 'video3.srt'])
# 搜索关键词
def search_across_videos(keyword):
"""在所有视频中搜索关键词"""
results = []
if keyword in index:
for item in index[keyword]:
video_id = item['video_id']
timestamp = item['timestamp']
text = item['text']
results.append({
'video': video_id,
'time': format_time(timestamp),
'text': text
})
return results
# 搜索 "人工智能"
ai_mentions = search_across_videos('人工智能')
for result in ai_mentions:
print(f"[{result['video']}] {result['time']} - {result['text']}")
场景二:对话分析与情绪识别
字幕文件记录了完整的对话内容,是进行对话分析和情绪识别研究的理想数据源。通过 sub2api,你可以快速提取对话片段,结合自然语言处理工具进行情绪分析。
from sub2api import SubtitleClient
# 初始化
client = SubtitleClient(base_url="http://localhost:8000")
def analyze_conversations(content_id):
"""
分析字幕中的对话情绪
使用规则-based 的情绪分析方法
"""
# 获取所有字幕
subtitles = client.get_all(content_id)
# 定义情绪关键词
positive_words = ['好', '棒', '优秀', '喜欢', '开心', '高兴', '完美', '赞']
negative_words = ['差', '糟糕', '讨厌', '难过', '失望', '生气', '可恶', '不好']
results = []
for sub in subtitles:
text = sub['text']
score = 0
emotion = 'neutral'
# 计算情绪得分
for word in positive_words:
if word in text:
score += 1
for word in negative_words:
if word in text:
score -= 1
# 判断情绪类别
if score > 0:
emotion = 'positive'
elif score < 0:
emotion = 'negative'
if score != 0:
results.append({
'timestamp': sub['start'],
'text': text,
'emotion': emotion,
'score': score
})
return results
def generate_emotion_timeline(content_id, output_path):
"""
生成情绪时间线图
输出 JSON 格式的情绪变化数据
"""
analysis = analyze_conversations(content_id)
timeline = {
'total_entries': len(analysis),
'positive_count': sum(1 for a in analysis if a['emotion'] == 'positive'),
'negative_count': sum(1 for a in analysis if a['emotion'] == 'negative'),
'timeline': analysis
}
import json
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(timeline, f, ensure_ascii=False, indent=2)
return timeline
# 生成情绪分析报告
timeline = generate_emotion_timeline("abc123def456", "emotion_report.json")
print(f"分析完成:正面 {timeline['positive_count']} 处,负面 {timeline['negative_count']} 处")
场景三:多语言字幕对照与翻译辅助
在处理翻译项目时,经常需要对照源语言和目标语言的字幕。通过 sub2api,你可以轻松实现双语字幕的时间对齐和对照查看。
def align_bilingual_subtitles(source_file, target_file, output_path):
"""
对照双语字幕文件
假设源文件和目标文件的字幕条目基本对应
通过时间相似度进行自动对齐
"""
import json
# 上传两个字幕文件
source_result = client.upload(source_file)
target_result = client.upload(target_file)
# 获取所有字幕
source_subtitles = client.get_all(source_result['content_id'])
target_subtitles = client.get_all(target_result['content_id'])
# 简单的对齐算法(基于时间差最小原则)
aligned = []
for target_sub in target_subtitles:
target_time = target_sub['start']
# 找到最接近的源字幕
best_match = None
min_diff = float('inf')
for source_sub in source_subtitles:
diff = abs(source_sub['start'] - target_time)
if diff < min_diff:
min_diff = diff
best_match = source_sub
if best_match and min_diff < 2.0: # 时间差小于2秒视为匹配
aligned.append({
'source_text': best_match['text'],
'target_text': target_sub['text'],
'time': format_time(target_sub['start'])
})
# 生成对照输出
output_lines = []
output_lines.append("双语对照字幕")
output_lines.append("=" * 60)
for item in aligned:
output_lines.append(f"[{item['time']}]")
output_lines.append(f" 原文: {item['source_text']}")
output_lines.append(f" 译文: {item['target_text']}")
output_lines.append("-" * 60)
output_content = '\n'.join(output_lines)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(output_content)
print(f"已生成双语对照文件: {output_path}")
print(f"共对齐 {len(aligned)} 个字幕条目")
return aligned
# 对照字幕文件
align_bilingual_subtitles('english.srt', 'chinese.srt', 'bilingual_comparison.txt')
技巧与最佳实践
性能优化建议
在处理大型字幕文件或进行批量操作时,性能是需要重点考虑的因素。以下是一些经过实践验证的优化技巧。
对于大型字幕文件,建议使用流式处理方式,避免将整个文件加载到内存中。sub2api 的 API 设计已经考虑了这一点,默认情况下使用生成器模式返回结果,而不是一次性返回所有数据。
# 使用流式处理处理大型文件
def process_large_subtitle_streaming(content_id, batch_size=100):
"""
流式处理大型字幕文件
每次处理 batch_size 条记录,减少内存占用
"""
offset = 0
while True:
# 使用分页获取
batch = client.get_paginated(
content_id=content_id,
offset=offset,
limit=batch_size
)
if not batch:
break
# 处理这批数据
for subtitle in batch:
process_subtitle(subtitle)
offset += batch_size
print(f"已处理 {offset} 条记录...")
print("处理完成")
def process_subtitle(subtitle):
"""处理单条字幕记录"""
# 在这里实现你的处理逻辑
pass
缓存是另一个重要的优化手段。如果你的应用需要频繁查询相同的字幕内容,可以使用本地缓存来减少 API 调用次数。
from functools import lru_cache
import json
# 简单的内存缓存实现
cache = {}
CACHE_TTL = 3600 # 缓存有效期(秒)
def cached_get_all(content_id):
"""带缓存的获取所有字幕"""
import time
cache_key = f"subtitles_{content_id}"
current_time = time.time()
# 检查缓存
if cache_key in cache:
cached_data, cached_time = cache[cache_key]
if current_time - cached_time < CACHE_TTL:
print("使用缓存数据")
return cached_data
# 获取新数据
print("从API获取数据")
subtitles = client.get_all(content_id)
# 更新缓存
cache[cache_key] = (subtitles, current_time)
return subtitles
错误处理与容错
在生产环境中,健壮的错误处理是必不可少的。以下是一些常见的错误场景及其处理方式:
import requests
from requests.exceptions import ConnectionError, Timeout, HTTPError
def robust_upload(file_path, max_retries=3):
"""
带重试机制的上传函数
遇到网络错误时自动重试
"""
for attempt in range(max_retries):
try:
with open(file_path, 'rb') as f:
files = {'file': f}
response = requests.post(
"http://localhost:8000/upload",
files=files,
timeout=30 # 设置超时
)
response.raise_for_status()
return response.json()
except ConnectionError as e:
print(f"连接失败(第 {attempt + 1} 次尝试): {e}")
if attempt < max_retries - 1:
import time
time.sleep(2 ** attempt) # 指数退避
except Timeout as e:
print(f"请求超时(第 {attempt + 1} 次尝试): {e}")
if attempt < max_retries - 1:
import time
time.sleep(2 ** attempt)
except HTTPError as e:
print(f"HTTP错误: {e}")
raise
except Exception as e:
print(f"未知错误: {e}")
raise
raise Exception(f"上传失败,已重试 {max_retries} 次")
def robust_query_with_fallback(content_id, keyword):
"""
查询失败时尝试备选方案
"""
try:
# 尝试精确搜索
return client.search(content_id, keyword, exact_match=True)
except Exception as e:
print(f"精确搜索失败: {e}")
try:
# 尝试模糊搜索
return client.search(content_id, keyword, fuzzy_match=True)
except Exception as e:
print(f"模糊搜索也失败: {e}")
return []
安全注意事项
在使用 sub2api 时,安全问题同样需要关注。以下是一些基本的安全建议。
如果你的 API 服务需要暴露给外部网络,务必配置适当的访问控制。可以使用 API 密钥认证、IP 白名单或 JWT token 等机制。
# 使用 API 密钥认证的客户端示例
class AuthenticatedClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.api_key = api_key
def _get_headers(self):
return {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
def upload(self, file_path):
with open(file_path, 'rb') as f:
files = {'file': f}
headers = {'Authorization': f'Bearer {self.api_key}'}
response = requests.post(
f"{self.base_url}/upload",
files=files,
headers=headers
)
return response.json()
def query(self, content_id, keyword):
response = requests.post(
f"{self.base_url}/query",
json={'content_id': content_id, 'keyword': keyword},
headers=self._get_headers()
)
return response.json()
# 使用认证客户端
auth_client = AuthenticatedClient(
base_url="http://localhost:8000",
api_key="your-secret-api-key"
)
对于上传的文件内容,建议进行基本的验证和清理,防止恶意文件导致安全问题。
import os
def validate_uploaded_file(file_path):
"""
验证上传的文件
"""
# 检查文件扩展名
allowed_extensions = ['.srt', '.vtt', '.ass', '.ssa']
ext = os.path.splitext(file_path)[1].lower()
if ext not in allowed_extensions:
raise ValueError(f"不支持的文件类型: {ext}")
# 检查文件大小(限制为100MB)
max_size = 100 * 1024 * 1024
file_size = os.path.getsize(file_path)
if file_size > max_size:
raise ValueError(f"文件过大: {file_size / (1024*1024):.2f}MB(最大 {max_size / (1024*1024)}MB)")
# 检查文件内容是否为空
if file_size == 0:
raise ValueError("文件为空")
return True
def sanitize_text(text):
"""
清理文本内容,防止XSS等问题
"""
# 移除潜在的恶意内容
import html
text = html.escape(text)
# 根据需要添加更多清理逻辑
return text
总结与扩展
通过本教程的学习,你应该已经掌握了 sub2api 的核心功能和使用方法。这个项目将字幕处理从繁琐的手动操作中解放出来,通过 API 的形式提供了优雅、高效的解决方案。
核心要点回顾
我们首先了解了为什么 sub2api 值得关注——它解决了字幕处理中的效率问题,将复杂的解析逻辑封装成简洁的 API 接口。接着,我们详细介绍了环境搭建过程,包括 Python 环境配置、依赖安装和配置文件设置。在核心功能部分,我们深入剖析了字幕解析引擎的原理和 REST API 的设计理念。实战教程部分通过六个步骤,带领读者从启动服务开始,逐步掌握上传、查询、搜索、导出等各项操作。最后,我们探讨了典型应用场景和最佳实践,帮助读者将知识应用到实际工作中。
相关项目推荐
如果你对字幕处理和自然语言工具感兴趣,以下开源项目也值得关注:
在字幕处理领域,FFmpeg 是最强大的多媒体处理工具,可以进行各种格式转换和编辑操作。Aeneas 是一个自动同步音频和文本的工具,特别适合将纯文本转换为带时间戳的字幕。WebVTT.js 提供了纯 JavaScript 实现的 WebVTT 格式解析器,适合在浏览器环境中使用。
在自然语言处理方面,jieba 是最流行的中文分词工具,本教程中的示例就使用了它。THULAC 是清华大学开发的高效中文词法分析工具。SnowNLP 专注于中文情感分析,适合进行文本情绪分析。
sub2api 的设计理念可以启发更多类似的工具开发。将传统的文件处理任务转换为 API 服务,是提升开发效率和系统集成性的有效方式。希望你能从本教程中获得启发,在自己的项目中灵活运用这些思想。
如果你在使用过程中遇到问题或有改进建议,欢迎访问项目的 GitHub 页面提交 Issue 或 Pull Request。开源社区的发展离不开每个人的贡献,你的反馈将使 sub2api 变得更好。
项目地址:https://github.com/Wei-Shaw/sub2api
文档地址:项目自述文件中包含详细的安装和使用说明
期待看到你使用 sub2api 创建的优秀应用!
评论区