1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
import asyncio
import os
import yt_dlp
import whisper
import torch
async def download_audio(video_url):
"""从视频 URL 下载音频并返回文件路径。"""
print("正在下载音频...")
# 先提取视频信息以获取标题
ydl_info_opts = {
'quiet': True,
'no_warnings': True,
}
loop = asyncio.get_event_loop()
video_info = await loop.run_in_executor(
None,
lambda: yt_dlp.YoutubeDL(ydl_info_opts).extract_info(video_url, download=False)
)
# 根据标题创建安全的文件名
safe_title = "".join([c if c.isalnum() or c in " -_" else "_" for c in video_info.get('title', 'video')])
filename = f"{safe_title}.mp3"
ydl_opts = {
'format': 'bestaudio/best',
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '192',
}],
'outtmpl': f'{safe_title}.%(ext)s',
'progress_hooks': [lambda d: print(f"下载进度: {d.get('_percent_str', '正在下载...')}")],
'overwrites': True,
}
# 下载视频并提取音频
await loop.run_in_executor(
None,
lambda: yt_dlp.YoutubeDL(ydl_opts).download([video_url])
)
return filename
async def transcribe_video(video_url, language="en", output_file=None):
"""下载视频并使用 Whisper 进行转录。"""
try:
# 下载音频文件
audio_file = await download_audio(video_url)
# 如果有 GPU 则加载 Whisper 模型到 GPU
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"使用设备: {device}")
print("正在加载 Whisper 模型...")
# 使用 turbo 模型以获得更快的转录速度
model = whisper.load_model("turbo", device=device)
# 开始转录
print("正在转录...")
result = model.transcribe(
audio_file,
language=language,
without_timestamps=True,
fp16=(device == "cuda")
)
transcript = result["text"]
# 如果指定了输出文件，则写入文件
if output_file:
with open(output_file, "w", encoding="utf-8") as f:
f.write(transcript)
print(f"转录文本已保存到 {output_file}")
return transcript
except Exception as e:
print(f"转录过程中出错: {e}")
raise
finally:
# 如有需要可清理下载的文件
# 如果想在转录后删除音频文件，请取消注释下面的代码
# if 'audio_file' in locals() and os.path.exists(audio_file):
# os.remove(audio_file)
# print(f"已删除临时文件: {audio_file}")
pass
async def main():
url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" # 替换为你的视频链接
language = "en" # 可更改为所需语言代码（如中文为 "zh"）
output_file = "transcript.txt"
transcript = await transcribe_video(url, language, output_file)
print("\n转录结果：")
print(transcript)
if __name__ == "__main__":
asyncio.run(main())
|
