Node.js调用Qwen3-ASR API实现批量语音文件处理

你是不是也遇到过这样的场景？手头有一堆会议录音、访谈音频或者视频素材，需要把它们全部转成文字。一个个手动上传、等待、复制结果，不仅效率低下，还容易出错。特别是当文件数量多、时长长的时候，这种重复劳动简直让人崩溃。

今天我就来分享一个实用的解决方案：用Node.js写一个自动化脚本，批量处理语音文件，通过Qwen3-ASR服务进行转写。这个方案不仅支持批量处理，还内置了错误重试和进度监控，让你一次设置，轻松搞定所有音频转文字的工作。

1. 环境准备与项目搭建

1.1 Node.js安装与检查

首先确保你的电脑上已经安装了Node.js。如果你还没安装，可以去Node.js官网下载最新版本。安装完成后，打开终端或命令行工具，检查一下版本：

node --version
npm --version

我用的Node.js版本是18.0.0以上，npm版本是8.0.0以上。如果你的版本比较旧，建议升级一下，避免兼容性问题。

1.2 创建项目目录

找个合适的位置，创建一个新的项目文件夹：

mkdir qwen3-asr-batch-processor
cd qwen3-asr-batch-processor

1.3 初始化项目并安装依赖

初始化Node.js项目，然后安装我们需要的几个包：

npm init -y
npm install axios form-data fs-extra progress dotenv

简单说一下这几个包的作用：

• axios：用来发送HTTP请求，调用Qwen3-ASR的API
• form-data：处理文件上传，把音频文件包装成表单数据
• fs-extra：增强版的文件系统操作，比Node.js自带的fs更好用
• progress：在命令行显示处理进度条，看着更直观
• dotenv：管理环境变量，把API密钥这些敏感信息放在.env文件里

2. 获取Qwen3-ASR API访问权限

2.1 了解Qwen3-ASR的能力

在开始写代码之前，先简单了解一下Qwen3-ASR。这是阿里开源的一个语音识别模型，有几个很实用的特点：

• 支持52种语言和方言：包括中文、英文、日语、韩语等主流语言，还有各种中国方言
• 识别准确率高：在嘈杂环境、老人儿童语音、甚至歌声识别上都有不错的表现
• 处理速度快：0.6B版本在128并发下，10秒能处理5小时音频
• 支持长时间音频：单次可以处理长达20分钟的音频文件

2.2 获取API密钥

要使用Qwen3-ASR的API服务，你需要先获取API密钥。通常可以在阿里云的相关服务页面申请。拿到密钥后，我们创建一个.env文件来保存它：

# 创建.env文件
touch .env

在.env文件里添加你的API密钥：

QWEN3_ASR_API_KEY=你的API密钥
QWEN3_ASR_API_ENDPOINT=https://api.example.com/v1/audio/transcriptions

注意：.env文件包含敏感信息，千万不要提交到GitHub等公开仓库。记得在.gitignore文件里加上.env。

3. 核心脚本编写

3.1 基础配置文件

先创建一个config.js文件，存放一些配置信息：

// config.js
require('dotenv').config();

module.exports = {
  apiKey: process.env.QWEN3_ASR_API_KEY,
  apiEndpoint: process.env.QWEN3_ASR_API_ENDPOINT,
  
  // 支持的文件格式
  supportedFormats: ['.mp3', '.wav', '.m4a', '.flac', '.ogg'],
  
  // 重试配置
  retryConfig: {
    maxRetries: 3,           // 最大重试次数
    retryDelay: 1000,        // 重试延迟（毫秒）
    timeout: 30000           // 请求超时时间（毫秒）
  },
  
  // 输出配置
  outputConfig: {
    format: 'txt',           // 输出格式：txt或json
    includeTimestamps: false, // 是否包含时间戳
    language: 'auto'         // 语言设置：auto自动检测，或指定如zh、en
  }
};

3.2 单个文件处理函数

接下来写一个处理单个音频文件的函数。这是整个脚本的核心：

// transcribeSingleFile.js
const axios = require('axios');
const FormData = require('form-data');
const fs = require('fs-extra');
const path = require('path');
const config = require('./config');

async function transcribeSingleFile(filePath, options = {}) {
  const {
    language = config.outputConfig.language,
    includeTimestamps = config.outputConfig.includeTimestamps,
    task = 'transcribe'
  } = options;

  // 检查文件是否存在
  if (!await fs.pathExists(filePath)) {
    throw new Error(`文件不存在: ${filePath}`);
  }

  // 检查文件格式
  const ext = path.extname(filePath).toLowerCase();
  if (!config.supportedFormats.includes(ext)) {
    throw new Error(`不支持的文件格式: ${ext}。支持格式: ${config.supportedFormats.join(', ')}`);
  }

  // 创建表单数据
  const formData = new FormData();
  formData.append('file', fs.createReadStream(filePath));
  formData.append('model', 'qwen3-asr-1.7b'); // 使用1.7B模型，准确率更高
  formData.append('language', language);
  formData.append('response_format', includeTimestamps ? 'verbose_json' : 'text');
  formData.append('task', task);

  // 准备请求头
  const headers = {
    'Authorization': `Bearer ${config.apiKey}`,
    ...formData.getHeaders()
  };

  // 带重试机制的请求
  let lastError;
  for (let attempt = 1; attempt <= config.retryConfig.maxRetries; attempt++) {
    try {
      console.log(`处理文件: ${path.basename(filePath)} (尝试 ${attempt}/${config.retryConfig.maxRetries})`);
      
      const response = await axios.post(config.apiEndpoint, formData, {
        headers,
        timeout: config.retryConfig.timeout,
        maxContentLength: Infinity,
        maxBodyLength: Infinity
      });

      // 根据返回格式处理结果
      if (includeTimestamps) {
        return {
          text: response.data.text,
          segments: response.data.segments,
          language: response.data.language
        };
      } else {
        return response.data;
      }

    } catch (error) {
      lastError = error;
      
      // 如果是最后一次尝试，直接抛出错误
      if (attempt === config.retryConfig.maxRetries) {
        break;
      }
      
      // 等待一段时间后重试
      console.log(`请求失败，${config.retryConfig.retryDelay/1000}秒后重试...`);
      await new Promise(resolve => setTimeout(resolve, config.retryConfig.retryDelay));
    }
  }

  throw new Error(`处理文件失败: ${path.basename(filePath)}。错误: ${lastError.message}`);
}

module.exports = transcribeSingleFile;

3.3 批量处理与进度监控

单个文件处理搞定后，我们来写批量处理的逻辑，加上进度条让过程更直观：

// batchProcessor.js
const fs = require('fs-extra');
const path = require('path');
const ProgressBar = require('progress');
const transcribeSingleFile = require('./transcribeSingleFile');
const config = require('./config');

class BatchProcessor {
  constructor(inputDir, outputDir) {
    this.inputDir = inputDir;
    this.outputDir = outputDir;
    this.results = [];
    this.errors = [];
  }

  // 扫描目录获取所有音频文件
  async scanAudioFiles() {
    console.log(`扫描目录: ${this.inputDir}`);
    
    const files = await fs.readdir(this.inputDir);
    const audioFiles = files.filter(file => {
      const ext = path.extname(file).toLowerCase();
      return config.supportedFormats.includes(ext);
    });

    console.log(`找到 ${audioFiles.length} 个音频文件`);
    return audioFiles.map(file => path.join(this.inputDir, file));
  }

  // 确保输出目录存在
  async ensureOutputDir() {
    await fs.ensureDir(this.outputDir);
  }

  // 保存单个文件的结果
  async saveResult(filePath, result, options = {}) {
    const fileName = path.basename(filePath, path.extname(filePath));
    const outputPath = path.join(this.outputDir, `${fileName}.${options.format || 'txt'}`);
    
    if (options.format === 'json') {
      await fs.writeJson(outputPath, result, { spaces: 2 });
    } else {
      await fs.writeFile(outputPath, typeof result === 'string' ? result : result.text);
    }
    
    return outputPath;
  }

  // 批量处理主函数
  async processBatch(options = {}) {
    console.log('开始批量处理音频文件...\n');
    
    // 扫描文件
    const audioFiles = await this.scanAudioFiles();
    if (audioFiles.length === 0) {
      console.log('没有找到可处理的音频文件');
      return { success: 0, failed: 0, results: [], errors: [] };
    }

    // 准备进度条
    const bar = new ProgressBar('处理中 [:bar] :percent :etas', {
      complete: '=',
      incomplete: ' ',
      width: 30,
      total: audioFiles.length
    });

    // 创建输出目录
    await this.ensureOutputDir();

    // 逐个处理文件
    for (const filePath of audioFiles) {
      try {
        const result = await transcribeSingleFile(filePath, options);
        
        // 保存结果
        const savedPath = await this.saveResult(filePath, result, options);
        
        // 记录成功结果
        this.results.push({
          input: filePath,
          output: savedPath,
          success: true,
          timestamp: new Date().toISOString()
        });
        
        // 显示一些信息
        const textPreview = typeof result === 'string' 
          ? result.substring(0, 100) + (result.length > 100 ? '...' : '')
          : result.text.substring(0, 100) + (result.text.length > 100 ? '...' : '');
        
        console.log(`\n✓ ${path.basename(filePath)} 处理完成`);
        console.log(`  预览: ${textPreview}`);
        
      } catch (error) {
        console.error(`\n✗ ${path.basename(filePath)} 处理失败: ${error.message}`);
        
        // 记录错误
        this.errors.push({
          file: filePath,
          error: error.message,
          timestamp: new Date().toISOString()
        });
      }
      
      // 更新进度条
      bar.tick();
    }

    // 生成处理报告
    await this.generateReport();
    
    return {
      success: this.results.length,
      failed: this.errors.length,
      results: this.results,
      errors: this.errors
    };
  }

  // 生成处理报告
  async generateReport() {
    const report = {
      timestamp: new Date().toISOString(),
      totalProcessed: this.results.length + this.errors.length,
      success: this.results.length,
      failed: this.errors.length,
      results: this.results,
      errors: this.errors
    };

    const reportPath = path.join(this.outputDir, 'processing_report.json');
    await fs.writeJson(reportPath, report, { spaces: 2 });
    
    console.log(`\n处理报告已保存: ${reportPath}`);
  }
}

module.exports = BatchProcessor;

3.4 主程序入口

最后，我们创建一个主文件来组织整个流程：

// main.js
const BatchProcessor = require('./batchProcessor');
const config = require('./config');
const path = require('path');

async function main() {
  console.log('=== Qwen3-ASR 批量语音处理工具 ===\n');
  
  // 检查API密钥
  if (!config.apiKey || config.apiKey === '你的API密钥') {
    console.error('错误：请先在 .env 文件中设置正确的 QWEN3_ASR_API_KEY');
    console.error('参考 .env.example 文件格式');
    process.exit(1);
  }

  // 设置输入输出目录
  const inputDir = path.join(__dirname, 'input');
  const outputDir = path.join(__dirname, 'output');
  
  // 创建处理器实例
  const processor = new BatchProcessor(inputDir, outputDir);
  
  // 处理选项
  const options = {
    language: 'auto',           // 自动检测语言
    includeTimestamps: false,   // 不包含时间戳
    format: 'txt'              // 输出为文本格式
  };
  
  try {
    // 开始批量处理
    const result = await processor.processBatch(options);
    
    console.log('\n=== 处理完成 ===');
    console.log(`成功: ${result.success} 个文件`);
    console.log(`失败: ${result.failed} 个文件`);
    
    if (result.errors.length > 0) {
      console.log('\n失败的文件：');
      result.errors.forEach((error, index) => {
        console.log(`${index + 1}. ${path.basename(error.file)}: ${error.error}`);
      });
    }
    
    console.log(`\n输出文件保存在: ${outputDir}`);
    
  } catch (error) {
    console.error('处理过程中发生错误:', error.message);
    process.exit(1);
  }
}

// 运行主函数
if (require.main === module) {
  main().catch(console.error);
}

module.exports = main;

4. 使用示例与测试

4.1 准备测试文件

在项目根目录创建input文件夹，放几个测试用的音频文件：

mkdir input
# 把你的音频文件（mp3、wav等）复制到input文件夹

4.2 运行脚本

直接运行主程序：

node main.js

你会看到类似这样的输出：

=== Qwen3-ASR 批量语音处理工具 ===

扫描目录: /path/to/your/project/input
找到 5 个音频文件
开始批量处理音频文件...

处理中 [===============>           ] 60% 12s

✓ meeting_audio.mp3 处理完成
  预览: 今天我们讨论一下项目进度。目前前端开发已经完成80%，后端...

✓ interview.wav 处理完成  
  预览: 问：您如何看待当前的人工智能发展趋势？答：我认为AI正在...

处理报告已保存: /path/to/your/project/output/processing_report.json

=== 处理完成 ===
成功: 5 个文件
失败: 0 个文件

输出文件保存在: /path/to/your/project/output

4.3 查看结果

处理完成后，去output文件夹看看结果：

output/
├── meeting_audio.txt
├── interview.txt
├── presentation.txt
├── discussion.txt
├── recording.txt
└── processing_report.json

每个音频文件都对应一个文本文件，processing_report.json里记录了详细的处理日志。

5. 高级功能扩展

5.1 支持时间戳输出

如果你需要知道每个句子在音频中的具体时间位置，可以启用时间戳功能。修改main.js中的选项：

const options = {
  language: 'zh',              // 指定中文
  includeTimestamps: true,     // 包含时间戳
  format: 'json'               // 输出为JSON格式
};

这样输出的JSON文件会包含分段信息：

{
  "text": "今天我们讨论一下项目进度。目前前端开发已经完成80%。",
  "segments": [
    {
      "text": "今天我们讨论一下项目进度。",
      "start": 0.0,
      "end": 2.5
    },
    {
      "text": "目前前端开发已经完成80%。",
      "start": 2.5,
      "end": 5.0
    }
  ],
  "language": "zh"
}

5.2 并发处理提高速度

如果需要处理大量文件，可以添加并发处理功能。创建一个新的处理函数：

// concurrentProcessor.js
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const transcribeSingleFile = require('./transcribeSingleFile');
const path = require('path');
const fs = require('fs-extra');

// 工作线程处理函数
async function workerProcess() {
  if (!isMainThread) {
    const { filePath, options } = workerData;
    try {
      const result = await transcribeSingleFile(filePath, options);
      parentPort.postMessage({ success: true, filePath, result });
    } catch (error) {
      parentPort.postMessage({ success: false, filePath, error: error.message });
    }
  }
}

// 主线程并发处理函数
async function processConcurrently(filePaths, options, maxConcurrent = 3) {
  const results = [];
  const errors = [];
  
  // 分批处理
  for (let i = 0; i < filePaths.length; i += maxConcurrent) {
    const batch = filePaths.slice(i, i + maxConcurrent);
    const promises = batch.map(filePath => {
      return new Promise((resolve) => {
        const worker = new Worker(__filename, {
          workerData: { filePath, options }
        });
        
        worker.on('message', (message) => {
          resolve(message);
          worker.terminate();
        });
        
        worker.on('error', (error) => {
          resolve({ success: false, filePath, error: error.message });
          worker.terminate();
        });
      });
    });
    
    const batchResults = await Promise.all(promises);
    batchResults.forEach(result => {
      if (result.success) {
        results.push(result);
      } else {
        errors.push(result);
      }
    });
    
    console.log(`已处理 ${i + batch.length}/${filePaths.length} 个文件`);
  }
  
  return { results, errors };
}

// 根据运行环境选择执行路径
if (isMainThread) {
  module.exports = processConcurrently;
} else {
  workerProcess();
}

然后在主程序中使用：

// 在主程序中使用并发处理
const processConcurrently = require('./concurrentProcessor');

async function processWithConcurrency() {
  const filePaths = await getAudioFiles(); // 获取文件列表
  const options = { language: 'auto' };
  
  console.log(`开始并发处理 ${filePaths.length} 个文件...`);
  const { results, errors } = await processConcurrently(filePaths, options, 3);
  
  console.log(`处理完成：成功 ${results.length}，失败 ${errors.length}`);
}

5.3 错误处理与日志

为了更好的调试和监控，我们可以添加更详细的日志系统：

// logger.js
const fs = require('fs-extra');
const path = require('path');
const { format } = require('date-fns');

class Logger {
  constructor(logDir = 'logs') {
    this.logDir = logDir;
    this.logFile = path.join(logDir, `process_${format(new Date(), 'yyyy-MM-dd')}.log`);
    this.initialize();
  }

  async initialize() {
    await fs.ensureDir(this.logDir);
  }

  async log(level, message, data = {}) {
    const timestamp = new Date().toISOString();
    const logEntry = {
      timestamp,
      level,
      message,
      ...data
    };

    // 控制台输出
    console.log(`[${timestamp}] ${level.toUpperCase()}: ${message}`);
    
    // 文件输出
    const logLine = JSON.stringify(logEntry) + '\n';
    await fs.appendFile(this.logFile, logLine);
  }

  async info(message, data) {
    await this.log('info', message, data);
  }

  async error(message, data) {
    await this.log('error', message, data);
  }

  async warn(message, data) {
    await this.log('warn', message, data);
  }
}

module.exports = Logger;

在批处理器中使用日志：

const Logger = require('./logger');
const logger = new Logger();

// 在处理函数中添加日志
async function transcribeWithLogging(filePath, options) {
  await logger.info('开始处理文件', { file: path.basename(filePath) });
  
  try {
    const result = await transcribeSingleFile(filePath, options);
    await logger.info('文件处理成功', { 
      file: path.basename(filePath),
      textLength: result.text?.length || result.length
    });
    return result;
  } catch (error) {
    await logger.error('文件处理失败', {
      file: path.basename(filePath),
      error: error.message
    });
    throw error;
  }
}

6. 实际应用建议

6.1 文件命名规范

为了让处理结果更有条理，建议对音频文件采用规范的命名方式：

项目名_日期_内容简述.扩展名
示例：projectX_20240220_需求讨论.mp3

这样生成的文本文件也会有清晰的命名，方便后续查找。

6.2 处理长音频文件

如果遇到很长的音频文件（比如超过20分钟），可以考虑先分割再处理。可以用ffmpeg进行音频分割：

# 安装ffmpeg（如果还没安装）
# Ubuntu/Debian: sudo apt-get install ffmpeg
# macOS: brew install ffmpeg

# 用Node.js调用ffmpeg分割音频
const { exec } = require('child_process');
const util = require('util');
const execPromise = util.promisify(exec);

async function splitAudio(inputFile, segmentDuration = 600) { // 默认10分钟一段
  const outputPattern = path.join(path.dirname(inputFile), 'split', 'segment_%03d.mp3');
  
  await fs.ensureDir(path.dirname(outputPattern));
  
  const command = `ffmpeg -i "${inputFile}" -f segment -segment_time ${segmentDuration} -c copy "${outputPattern}"`;
  
  try {
    await execPromise(command);
    const files = await fs.readdir(path.dirname(outputPattern));
    return files.filter(f => f.startsWith('segment_')).map(f => path.join(path.dirname(outputPattern), f));
  } catch (error) {
    throw new Error(`音频分割失败: ${error.message}`);
  }
}

6.3 结果后处理

转写完成后，可能需要对文本进行一些后处理，比如：

// postProcessor.js
function cleanTranscript(text) {
  // 移除多余的空白字符
  let cleaned = text.replace(/\s+/g, ' ').trim();
  
  // 修复常见的识别错误
  const commonCorrections = {
    '喂喂': '微微',
    '在在': '正在',
    '的的': '的',
    // 添加更多常见修正...
  };
  
  Object.entries(commonCorrections).forEach(([wrong, correct]) => {
    cleaned = cleaned.replace(new RegExp(wrong, 'g'), correct);
  });
  
  // 添加标点（如果API返回的文本没有标点）
  cleaned = cleaned.replace(/([^。！？\.\!\?])(\n|$)/g, '$1。\n');
  
  return cleaned;
}

function formatWithSpeakers(text, speakerLabels = ['A', 'B']) {
  // 简单的对话格式处理
  const lines = text.split('\n').filter(line => line.trim());
  let currentSpeaker = 0;
  let formatted = [];
  
  for (const line of lines) {
    // 简单的启发式规则：如果行较短，可能是同一人继续说话
    if (line.length > 50 || formatted.length === 0) {
      formatted.push(`${speakerLabels[currentSpeaker]}: ${line}`);
      currentSpeaker = (currentSpeaker + 1) % speakerLabels.length;
    } else {
      // 短行，追加到上一行
      const lastIndex = formatted.length - 1;
      formatted[lastIndex] = formatted[lastIndex] + ' ' + line;
    }
  }
  
  return formatted.join('\n');
}

7. 总结

整体用下来，这套基于Node.js和Qwen3-ASR的批量语音处理方案还是挺实用的。部署起来不复杂，基本上按照步骤走就能跑起来。对于需要处理大量音频转文字的场景，比如会议记录整理、访谈内容转录、视频字幕生成等，能节省不少时间。

实际使用中，我发现有几点比较重要：一是文件命名要规范，这样输出结果也整齐；二是对于特别长的音频，先分割再处理效果更好；三是记得定期查看处理报告，了解哪些文件成功了，哪些失败了，方便排查问题。

如果你刚开始接触这个，建议先从少量文件试起，熟悉了整个流程再处理大批量文件。遇到问题可以看看日志文件，里面记录了详细的处理过程。后续如果想扩展功能，比如添加自动翻译、情感分析什么的，也可以在这个基础上继续开发。

---

获取更多AI镜像

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