如何实现定时任务？Paraformer-large批处理脚本自动化教程

1. 理解需求：从手动识别到自动批处理

你已经成功部署了Paraformer-large语音识别离线版，并通过 Gradio 界面实现了音频文件的手动上传与转写。但如果你每天都要处理几十个录音文件——比如会议记录、课程录音或客服对话，每次都打开网页点击上传显然效率太低。

那有没有办法让系统“自己动手”，每天固定时间扫描某个文件夹，把新进来的音频全部自动识别一遍，并保存成文本？

当然可以！这就是我们今天要解决的核心问题：如何将 Paraformer-large 模型封装为一个可被定时触发的批处理脚本，实现全自动语音转文字流水线。

本文将带你一步步完成以下目标：

• 将语音识别功能从 Web 界面剥离，封装为独立运行的 Python 脚本
• 编写支持批量处理多个音频文件的逻辑
• 利用 Linux 的cron实现每日定时执行
• 输出结构化文本结果（按日期命名）
• 提供完整可运行代码和部署建议

整个过程无需修改原始模型，也不依赖 Gradio，真正做到“后台静默运行 + 定时自动处理”。

2. 准备工作：环境确认与路径规划

2.1 确认基础环境

本教程基于你已成功运行 Paraformer-large 镜像的前提，关键组件如下：

提示：若你在 AutoDL 或其他云平台使用该镜像，请确保实例处于“开机”状态，否则定时任务无法执行。

2.2 规划文件结构

为了便于管理和自动化，我们先定义清晰的目录结构：

/root/asr_pipeline/ ├── input/ # 待识别的音频文件存放处 ├── output/ # 自动输出的文字结果（按日期+文件名保存） ├── logs/ # 日志记录，方便排查问题 ├── batch_asr.py # 批处理主脚本 └── processed_files.txt # 记录已处理过的文件，避免重复识别

创建命令如下：

mkdir -p /root/asr_pipeline/{input,output,logs} touch /root/asr_pipeline/processed_files.txt

所有新的音频文件只需放入input/文件夹，剩下的交给脚本自动完成。

3. 核心实现：编写批处理识别脚本

3.1 安装必要依赖（如有需要）

虽然镜像已预装大部分库，但我们仍需确保funasr和pydub（用于格式兼容）可用：

pip install funasr pydub -q

pydub可帮助自动转换非标准格式（如 m4a、amr）为 wav，提升兼容性。

3.2 编写批处理脚本batch_asr.py

# batch_asr.py import os import time from datetime import datetime from funasr import AutoModel from pydub import AudioSegment # === 配置区 === INPUT_DIR = "/root/asr_pipeline/input" OUTPUT_DIR = "/root/asr_pipeline/output" LOG_FILE = "/root/asr_pipeline/logs/run.log" PROCESSED_FILE = "/root/asr_pipeline/processed_files.txt" MODEL_ID = "iic/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch" # 支持的音频格式 SUPPORTED_EXT = {'.wav', '.mp3', '.m4a', '.flac', '.ogg', '.wma', '.aac'} # 记录日志函数 def log(msg): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {msg}") with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(f"[{timestamp}] {msg}\n") # 转换任意音频为 16kHz wav（Paraformer 最佳输入） def convert_to_wav(audio_path): try: audio = AudioSegment.from_file(audio_path) # 统一采样率和声道 audio = audio.set_frame_rate(16000).set_channels(1) temp_wav = "/tmp/temp_convert.wav" audio.export(temp_wav, format="wav") return temp_wav except Exception as e: log(f"格式转换失败: {e}") return None # 加载模型（全局一次） log("正在加载 Paraformer-large 模型...") model = AutoModel( model=MODEL_ID, model_revision="v2.0.4", device="cuda:0" # 使用 GPU 加速 ) log("模型加载完成！") # 主处理函数 def process_audio_file(file_path): try: # 先尝试直接识别 if file_path.lower().endswith('.wav'): res = model.generate(input=file_path, batch_size_s=300) else: # 非 wav 文件先转换 converted = convert_to_wav(file_path) if not converted: return None res = model.generate(input=converted, batch_size_s=300) if res and len(res) > 0: return res[0]['text'] else: return "识别失败：返回结果为空" except Exception as e: return f"识别出错: {str(e)}" # 读取已处理文件列表 def load_processed(): if not os.path.exists(PROCESSED_FILE): return set() with open(PROCESSED_FILE, "r", encoding="utf-8") as f: return set(f.read().splitlines()) # 保存已处理文件 def save_processed(processed_set): with open(PROCESSED_FILE, "w", encoding="utf-8") as f: f.write("\n".join(processed_set)) # 主流程 if __name__ == "__main__": start_time = time.time() log("开始批量语音识别任务...") processed_files = load_processed() new_processed = processed_files.copy() audio_files = [] for root, _, files in os.walk(INPUT_DIR): for file in files: ext = os.path.splitext(file)[1].lower() if ext in SUPPORTED_EXT: full_path = os.path.join(root, file) if full_path not in processed_files: audio_files.append(full_path) if not audio_files: log("没有发现新的音频文件，任务结束。") else: log(f"共找到 {len(audio_files)} 个新文件待处理。") for audio_file in audio_files: filename = os.path.basename(audio_file) stem, _ = os.path.splitext(filename) output_path = os.path.join(OUTPUT_DIR, f"{stem}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt") log(f"正在识别: {filename}") text = process_audio_file(audio_file) if text: with open(output_path, "w", encoding="utf-8") as f: f.write(text) new_processed.add(audio_file) log(f"✅ 识别完成 → {output_path}") else: log(f"❌ 识别失败: {filename}") # 更新已处理列表 save_processed(new_processed) total_time = time.time() - start_time log(f"任务结束，共耗时 {total_time:.2f} 秒。\n")

3.3 脚本特点说明

4. 测试脚本：验证是否能正常运行

4.1 放入测试音频

将一个名为test.mp3的中文语音文件放入/root/asr_pipeline/input/。

4.2 手动运行一次脚本

cd /root/asr_pipeline python batch_asr.py

观察输出日志，你应该能看到类似：

[2025-04-05 10:23:15] 正在加载 Paraformer-large 模型... [2025-04-05 10:23:30] 模型加载完成！ [2025-04-05 10:23:30] 开始批量语音识别任务... [2025-04-05 10:23:30] 共找到 1 个新文件待处理。 [2025-04-05 10:23:30] 正在识别: test.mp3 [2025-04-05 10:23:45] ✅ 识别完成 → /root/asr_pipeline/output/test_20250405_102345.txt

检查output/目录下是否生成了对应文本文件，打开查看内容是否合理。

5. 设置定时任务：让系统每天自动执行

Linux 下最常用的定时工具是cron。我们将配置它每天早上 8 点自动扫描并处理新音频。

5.1 编辑 crontab

crontab -e

首次会提示选择编辑器，推荐选nano（按方向键移动，Ctrl+O 回车保存，Ctrl+X 退出）。

5.2 添加定时规则

在文件末尾添加一行：

0 8 * * * source /opt/miniconda3/bin/activate torch25 && cd /root/asr_pipeline && python batch_asr.py >> /root/asr_pipeline/logs/cron.log 2>&1

含义解析：

即：每天上午 8:00 执行脚本。

后面的命令解释：

• source ... activate torch25：激活 Conda 环境（必须！否则找不到包）
• cd ... && python：进入目录并运行脚本
• >> log 2>&1：把输出和错误都记录下来，方便排查

5.3 查看当前定时任务

crontab -l

应能看到刚添加的那一行。

5.4 （可选）立即测试 cron 是否生效

你可以临时改成每分钟执行来测试：

* * * * * source /opt/miniconda3/bin/activate torch25 && cd /root/asr_pipeline && python batch_asr.py >> /root/asr_pipeline/logs/cron.log 2>&1

等一分钟后查看logs/是否有新日志。测试完记得改回原设置！

6. 进阶优化建议

6.1 添加邮件通知（可选）

可通过mailx或 Python 发送简要报告：

# 示例片段：在脚本末尾添加 import smtplib from email.mime.text import MIMEText def send_report(content): msg = MIMEText(content) msg['Subject'] = 'ASR 批处理完成' msg['From'] = 'asr@local.com' msg['To'] = 'admin@example.com' s = smtplib.SMTP('localhost') s.send_message(msg) s.quit()

6.2 支持子目录递归扫描

当前脚本能自动遍历input/下所有子文件夹中的音频，无需额外设置。

6.3 控制资源占用

如果同时处理大量长音频，可能内存不足。可在model.generate(...)中加入参数限制：

res = model.generate( input=audio_path, batch_size_s=60, # 减小批次大小降低显存压力 max_single_segment_time=60000 # 每段最长 60 秒 )

6.4 清理旧文件（可另设定时任务）

定期清理input/中已处理的原始音频，释放空间：

# 比如每周日凌晨清空 input 文件夹 0 0 * * 0 rm -rf /root/asr_pipeline/input/* >> /root/asr_pipeline/logs/clean.log 2>&1

7. 总结：打造你的全自动语音转写流水线

通过本文，你已经完成了从“手动点击上传”到“全自动定时批处理”的跃迁。这套方案特别适合以下场景：

• 📌企业会议纪要自动生成
• 📌在线课程语音转文稿
• 📌客服录音内容分析
• 📌播客/访谈内容结构化

关键成果回顾

现在，只要你把音频放进input/文件夹，第二天就能在output/看到整理好的文字稿，真正实现“躺平式 AI 办公”。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。