FireRedASR-AED-L模型Python爬虫实战:自动采集与识别网络公开课音频
FireRedASR-AED-L模型Python爬虫实战:自动采集与识别网络公开课音频
你是不是也遇到过这种情况?在网上找到一个非常棒的公开课系列,但只有音频,没有文字稿。想边听边记笔记,或者想快速查找某个知识点,都得从头到尾听一遍,效率实在太低了。
今天,我们就来解决这个问题。我将带你手把手搭建一个自动化工具,它能自动从公开课网站上抓取音频,然后用强大的语音识别模型,把音频内容转写成文字稿。整个过程完全自动化,你只需要提供一个网址,剩下的交给程序。
这个工具的核心是两个部分:一是用Python爬虫技术抓取和下载音频文件;二是用FireRedASR-AED-L模型进行高精度的语音转文字。听起来有点复杂?别担心,我会用最直白的方式,一步步拆解给你看。即使你之前没怎么接触过爬虫或者语音识别,跟着做也能搞定。
1. 准备工作:环境与工具
在开始写代码之前,我们需要先把“厨房”收拾好,把必要的“厨具”准备好。整个过程很简单,主要是安装几个Python库。
首先,确保你的电脑上已经安装了Python,建议使用Python 3.8或以上的版本。打开你的命令行工具(Windows上是CMD或PowerShell,Mac或Linux上是终端),我们来安装几个核心的库。
核心库安装:
- Requests / Scrapy: 用于从网页上抓取数据。Requests更简单直接,Scrapy功能更强大,适合复杂的爬取任务。我们先从Requests开始,它更容易上手。
- BeautifulSoup4: 用来解析HTML网页,从一堆代码里找到我们需要的音频链接。
- FireRedASR-AED-L模型相关库: 我们需要一个能调用这个语音识别模型的工具。这里假设你已经通过CSDN星图镜像广场或其他方式部署好了该模型的服务,并获得了API接口地址和调用方式。我们将使用
requests库来调用这个API。
在命令行里,一次性安装它们:
pip install requests beautifulsoup4
如果后续你想用Scrapy,可以再单独安装。至于语音识别模型的调用,我们完全基于HTTP API,所以不需要额外安装复杂的机器学习框架,这对新手非常友好。
安装完成后,我们新建一个项目文件夹,比如叫做audio_crawler,在里面开始我们的代码之旅。
2. 第一步:编写音频爬虫
我们的目标是找到网页里的音频文件,并把它下载到本地。我们以某个假设的公开课网站为例,讲解核心思路。请注意,在实际操作中,你必须严格遵守目标网站的robots.txt协议,尊重版权,仅对允许爬取或公开授权的资源进行操作。
2.1 分析网页结构,找到音频链接
首先,我们得知道音频文件藏在哪里。用浏览器打开一个公开课页面,按F12打开开发者工具,切换到“网络”(Network)标签页,然后播放音频。你会看到一个新的网络请求出现,它的类型可能是media,它的地址(URL)就是音频文件的真实链接。
我们的任务就是用Python代码模拟这个过程:访问网页,解析HTML,找到这个音频链接。
下面是一个使用requests和BeautifulSoup的基础示例:
import requests
from bs4 import BeautifulSoup
import re
def find_audio_url(page_url):
"""
根据课程页面URL,尝试查找并返回音频文件的直接链接。
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
try:
response = requests.get(page_url, headers=headers, timeout=10)
response.raise_for_status() # 检查请求是否成功
except requests.RequestException as e:
print(f"请求页面失败: {e}")
return None
soup = BeautifulSoup(response.text, 'html.parser')
# 方法1: 查找常见的音频标签 <audio src="...">
audio_tag = soup.find('audio')
if audio_tag and audio_tag.get('src'):
audio_url = audio_tag['src']
# 处理可能的相对路径
if not audio_url.startswith('http'):
from urllib.parse import urljoin
audio_url = urljoin(page_url, audio_url)
print(f"通过<audio>标签找到音频: {audio_url}")
return audio_url
# 方法2: 查找包含音频链接的脚本或特定class的div
# 很多网站用JavaScript加载音频,链接可能藏在<script>标签或data属性中
for script in soup.find_all('script'):
if script.string:
# 使用正则表达式寻找常见的音频文件格式链接
matches = re.findall(r'https?://[^\s"\']+?\.(mp3|m4a|wav|aac)[^\s"\']*', script.string, re.IGNORECASE)
if matches:
print(f"在脚本中找到音频链接: {matches[0]}")
return matches[0]
# 方法3: 查找所有链接,过滤出音频文件
all_links = soup.find_all('a', href=True)
for link in all_links:
href = link['href']
if re.search(r'\.(mp3|m4a|wav|aac)(\?.*)?$', href, re.IGNORECASE):
if not href.startswith('http'):
from urllib.parse import urljoin
href = urljoin(page_url, href)
print(f"在<a>标签中找到音频: {href}")
return href
print("未能在页面中找到音频链接。")
return None
# 测试一下
if __name__ == '__main__':
test_url = "https://example.com/lecture/123" # 请替换为实际测试地址
audio_url = find_audio_url(test_url)
print(f"最终音频地址: {audio_url}")
这段代码提供了三种寻找音频链接的策略,基本能覆盖大部分简单网站。关键点在于观察和适配,你需要根据目标网站的实际结构,调整查找逻辑。
2.2 下载音频文件
找到链接后,下载就很简单了,依然是使用requests库。
import os
def download_audio(audio_url, save_dir='downloads'):
"""
下载音频文件到本地指定目录。
"""
if not audio_url:
print("音频链接无效,跳过下载。")
return None
# 创建保存目录
os.makedirs(save_dir, exist_ok=True)
# 从URL中提取文件名
filename = os.path.basename(audio_url).split('?')[0] # 去掉URL参数
if not filename:
filename = 'audio_' + hashlib.md5(audio_url.encode()).hexdigest()[:8] + '.mp3'
save_path = os.path.join(save_dir, filename)
headers = {'User-Agent': 'Mozilla/5.0'}
try:
print(f"正在下载: {audio_url}")
response = requests.get(audio_url, headers=headers, stream=True, timeout=30)
response.raise_for_status()
# 以二进制流形式写入文件
with open(save_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"下载完成,保存至: {save_path}")
return save_path
except requests.RequestException as e:
print(f"下载音频失败: {e}")
return None
# 接续上面的代码
if audio_url:
downloaded_file = download_audio(audio_url)
现在,单个页面的音频下载功能就完成了。但公开课通常是一个系列,有多个页面。所以我们需要一个“列表页”爬虫,先抓取所有课程页面的链接,再对每个页面执行上面的“找链接-下载”流程。
3. 第二步:批量处理与任务队列
当你要处理几十上百个音频时,顺序下载和转写会非常慢。这里我们引入两个提升效率的概念:异步处理和任务队列。我们用一个简单直观的方式来实现。
3.1 组织爬取任务
假设我们有一个列表页,里面列出了所有课程的链接。我们先爬取这个列表。
def crawl_course_list(list_url):
"""
从课程列表页抓取所有课程详情页的链接。
"""
course_links = []
headers = {'User-Agent': 'Mozilla/5.0'}
try:
resp = requests.get(list_url, headers=headers)
soup = BeautifulSoup(resp.text, 'html.parser')
# 假设每个课程链接都在class为‘course-item’的<a>标签里
for item in soup.find_all('a', class_='course-item', href=True):
link = item['href']
if not link.startswith('http'):
from urllib.parse import urljoin
link = urljoin(list_url, link)
course_links.append(link)
print(f"从列表页共找到 {len(course_links)} 门课程。")
except Exception as e:
print(f"爬取列表页失败: {e}")
return course_links
3.2 简单的并行下载
我们可以使用Python内置的concurrent.futures模块中的ThreadPoolExecutor来并发下载,加快速度。
from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_download_audio(course_urls, max_workers=3):
"""
并发下载多个课程页面的音频。
max_workers控制同时下载的线程数,不要设置太高,避免对目标网站造成压力。
"""
downloaded_files = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交任务
future_to_url = {executor.submit(process_single_course, url): url for url in course_urls}
# 处理完成的任务
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
if result:
downloaded_files.append(result)
except Exception as exc:
print(f'课程页面 {url} 处理过程中产生异常: {exc}')
return downloaded_files
def process_single_course(course_url):
"""处理单个课程页面的完整流程:找链接 -> 下载"""
audio_url = find_audio_url(course_url)
if audio_url:
return download_audio(audio_url)
return None
这样,我们就有了一个能批量抓取音频的小系统。下载好的音频文件都存放在downloads文件夹里。
4. 第三步:调用FireRedASR-AED-L模型进行转写
音频下载好了,接下来就是重头戏:把声音变成文字。我们假设你已经部署好了FireRedASR-AED-L模型服务,它提供了一个HTTP API接口,比如 http://your-model-server:port/asr。
4.1 准备调用API
语音识别API通常接受一个音频文件,返回转写后的文本。我们需要将本地的音频文件发送过去。
import json
def transcribe_audio(file_path, api_url, api_key=None):
"""
调用语音识别API转写单个音频文件。
"""
if not os.path.exists(file_path):
print(f"文件不存在: {file_path}")
return None
headers = {}
if api_key:
headers['Authorization'] = f'Bearer {api_key}'
try:
with open(file_path, 'rb') as audio_file:
files = {'file': (os.path.basename(file_path), audio_file, 'audio/mpeg')}
data = {'model': 'FireRedASR-AED-L'} # 根据API要求可能需要其他参数
print(f"正在转写: {file_path}")
response = requests.post(api_url, files=files, data=data, headers=headers, timeout=60)
response.raise_for_status()
result = response.json()
# 假设API返回格式为 {'text': '识别出的文字...'}
transcribed_text = result.get('text', '')
print(f"转写完成: {file_path}")
return transcribed_text
except requests.exceptions.RequestException as e:
print(f"API请求失败 ({file_path}): {e}")
except json.JSONDecodeError as e:
print(f"解析API响应失败 ({file_path}): {e}")
return None
4.2 批量转写与结果保存
和下载一样,转写也可以批量进行。我们将转写好的文本按课程名称保存为.txt文件。
def batch_transcribe(audio_file_paths, api_url, api_key=None, max_workers=2):
"""
并发转写多个音频文件。
注意:语音识别是计算密集型任务,服务器压力大,并发数不宜过高。
"""
transcripts = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {executor.submit(transcribe_audio, file_path, api_url, api_key): file_path for file_path in audio_file_paths}
for future in as_completed(future_to_file):
file_path = future_to_file[future]
try:
text = future.result()
if text:
transcripts[file_path] = text
except Exception as exc:
print(f'文件 {file_path} 转写过程中产生异常: {exc}')
# 保存结果
for file_path, text in transcripts.items():
txt_filename = os.path.splitext(os.path.basename(file_path))[0] + '.txt'
txt_path = os.path.join('transcripts', txt_filename)
os.makedirs('transcripts', exist_ok=True)
with open(txt_path, 'w', encoding='utf-8') as f:
f.write(text)
print(f"文字稿已保存: {txt_path}")
return transcripts
5. 整合与运行:打造完整工作流
现在,我们把所有零件组装起来,形成一个完整的脚本。你可以通过修改main函数中的参数来运行它。
import hashlib
# 前面定义的所有函数都放在这里
def main():
# 1. 配置参数
COURSE_LIST_URL = "https://example.com/courses" # 课程列表页
ASR_API_URL = "http://your-model-server:port/asr" # 你的语音识别API地址
ASR_API_KEY = "your-api-key-here" # 如果需要
MAX_DOWNLOAD_WORKERS = 3 # 同时下载的线程数
MAX_TRANSCRIBE_WORKERS = 2 # 同时转写的线程数
# 2. 爬取课程链接
print("步骤1: 爬取课程列表...")
course_urls = crawl_course_list(COURSE_LIST_URL)
if not course_urls:
print("未找到课程链接,程序退出。")
return
# 3. 批量下载音频
print(f"\n步骤2: 开始批量下载音频 (共{len(course_urls)}门课程)...")
downloaded_files = batch_download_audio(course_urls, max_workers=MAX_DOWNLOAD_WORKERS)
print(f"音频下载完成,共成功下载 {len(downloaded_files)} 个文件。")
# 4. 批量转写音频
if downloaded_files:
print(f"\n步骤3: 开始批量语音转写...")
transcripts = batch_transcribe(downloaded_files, ASR_API_URL, ASR_API_KEY, max_workers=MAX_TRANSCRIBE_WORKERS)
print(f"转写完成,共生成 {len(transcripts)} 份文字稿。")
else:
print("没有可转写的音频文件。")
print("\n所有任务已完成!请查看 'downloads' 文件夹下的音频和 'transcripts' 文件夹下的文字稿。")
if __name__ == '__main__':
main()
运行这个脚本,泡杯咖啡,回来的时候,音频和文字稿就都准备好了。
6. 总结
走完这一趟,你会发现,把公开课音频变成可搜索、可编辑的文字稿,并没有想象中那么难。核心就是两步:用爬虫拿到资源,用AI模型转换格式。我们用的工具(Requests, BeautifulSoup)和思路(并发、队列)都是非常通用的,你可以把这个框架应用到很多类似的信息自动化收集场景里。
在实际操作中,有几点需要特别注意:一是遵守规则,确认你要爬取的网站是否允许自动化访问,控制请求频率,做个有道德的爬虫工程师;二是处理异常,网络请求、文件读写、API调用都可能出错,完善的错误处理和日志记录能让你的脚本更健壮;三是管理好资源,批量处理时注意内存和磁盘空间。
这个工具的价值在于,它把你从重复的“找资源-听录音-记笔记”的循环中解放出来,让你能更专注于内容本身的学习和消化。你可以基于生成的文字稿做摘要、划重点、甚至训练自己的知识库。技术的意义,不就是帮我们节省时间,去做更有创造力的事情吗?希望这个实战教程能给你带来启发,动手试试看,定制一个属于你自己的学习资源自动化管道吧。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐
所有评论(0)