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,找到这个音频链接。

下面是一个使用requestsBeautifulSoup的基础示例:

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星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐