飞书语音识别与多维表格自动化：构建本地音频处理流水线-编程实验室

1. 项目概述：当飞书遇上语音，一个被低估的效率工具

最近在折腾一个挺有意思的小项目，叫feishu-voice。乍一看名字，你可能觉得这又是一个飞书API的简单封装，或者一个语音转文字的玩具。但实际深入之后，我发现它远不止于此。这个项目本质上是一个桥梁，一个将飞书多维表格（Bitable）与本地语音文件处理能力连接起来的自动化工具。它的核心场景非常聚焦：批量、自动地将本地音频文件上传到飞书，并利用飞书强大的语音识别能力，将音频内容转录为结构化的文本，最终存入多维表格中，形成一个可查询、可分析、可协作的语音知识库。

想象一下这些场景：你是一个内容创作者，每天需要整理大量的访谈录音；你是一个项目经理，每周有几十个会议纪要需要处理；或者你是一个学生，需要整理课堂录音。传统流程是：听录音 -> 手动打字 -> 整理到文档或表格。这个过程耗时、费力，且容易出错。feishu-voice瞄准的就是这个痛点。它通过脚本自动化了“上传-识别-入库”的全流程，让你从重复劳动中解放出来，把精力集中在内容本身的分析和决策上。

我之所以对这个项目感兴趣，是因为它完美地结合了两个趋势：一是企业级协同工具（如飞书）开放其核心能力（如语音识别、多维表格）给开发者；二是本地化、轻量级的自动化脚本需求日益增长。它不需要复杂的服务器部署，几行命令就能跑起来，特别适合个人或小团队提升信息处理效率。接下来，我就带你彻底拆解这个项目，从设计思路到每一行代码的实操，分享我踩过的坑和总结出的最佳实践。

2. 核心设计思路与架构拆解

2.1 为什么选择飞书作为平台？

在动手之前，首先要理解为什么这个项目基于飞书构建，而不是其他云服务或开源方案。这背后有几个关键考量：

第一，识别质量与成本。飞书内置的语音识别（ASR）引擎，经过海量办公场景数据的训练，在中文普通话、带口音的普通话、以及中英文混杂场景下的表现，尤其是对会议、访谈等场景的噪音抑制和说话人区分，比许多开源方案（如Vosk）或通用云服务（在某些细分场景下）要更胜一筹。更重要的是，对于飞书用户而言，这部分识别能力往往包含在已有的套餐中，边际成本极低，甚至为零。自己搭建或调用其他商用API，则涉及持续的token费用。

第二，生态闭环。识别出的文本直接存入飞书多维表格，这意味着数据生来就在协作环境里。你可以立刻基于这些文本进行任务分配（@同事）、评论讨论、关联其他文档或表格，甚至通过飞书机器人推送识别结果。这个“识别 -> 存储 -> 协作 -> 通知”的闭环，如果自己从零搭建，需要组合存储服务、通知服务、权限系统等，复杂度陡增。

第三，安全与合规。对于企业用户，数据留在飞书生态内，通常比上传到另一个第三方云服务更容易通过内部的安全审计。飞书本身提供了完备的权限管控、审计日志等功能。

所以，项目的核心设计思路很清晰：以飞书多维表格为数据枢纽和呈现层，以飞书云文档的语音识别能力为引擎，通过本地脚本驱动自动化流程。脚本的角色是“粘合剂”和“控制器”。

2.2 项目架构与工作流解析

典型的feishu-voice工作流遵循以下步骤，这也构成了项目的主要模块：

配置与认证 (config.py/.env)：这是起点。你需要准备飞书开放平台的开发者权限，创建应用，获取app_id和app_secret。脚本会使用这些凭证获取访问令牌 (tenant_access_token)，这是调用所有飞书API的钥匙。这里的一个最佳实践是将敏感信息存储在环境变量或配置文件中，切勿硬编码在脚本里。
表格准备 (bitable.py)：脚本需要知道把数据存到哪里。你需要提前在飞书创建一个多维表格，并定义好列。通常必要的列包括：音频文件名、语音识别文本、识别状态、上传时间、音频文件在飞书的key（用于后续播放或下载）。脚本会通过API获取这个表格的app_token和table_id。
本地音频扫描与预处理 (file_utils.py)：脚本会扫描你指定的本地目录（如./audio），找出支持的音频文件（如.mp3,.wav,.m4a）。预处理可能包括检查文件大小（飞书API有上限）、时长，或者简单的音频格式验证。这里可以加入去重逻辑，避免重复处理同一文件。
上传音频至飞书 (upload.py)：这是关键一步。飞书并非直接接受一个音频文件进行识别，而是需要先将文件上传到飞书的云存储空间。脚本会调用上传图片或文件接口，将本地音频文件以“素材”形式上传。成功后，你会获得一个唯一的file_key。这个file_key是后续所有操作的基石，必须妥善保存到表格中。
发起语音识别请求 (speech_to_text.py)：获得file_key后，脚本调用飞书云文档的识别语音文件接口，将file_key提交给飞书服务端进行识别。这是一个异步操作，接口会立即返回一个本次识别任务的block_id。此时，识别任务在飞书后端排队处理。
轮询识别结果与解析 (polling.py)：由于识别是异步的，脚本需要定期（例如每5秒）查询识别任务的状态。通过block_id去查询，直到状态变为“成功”或“失败”。成功后，响应里会包含结构化的识别结果，通常是一个段落数组，包含文本内容和可能的时间戳（如果识别引擎支持）。
结果写入多维表格 (bitable.py)：将最终识别出的完整文本，连同文件名、状态、file_key、时间等信息，作为一条新记录，写入之前准备好的飞书多维表格中。至此，一个音频文件的处理流程结束。
批量处理与容错 (main.py)：主程序会循环处理目录下的所有文件，并为每个文件实现上述流程。必须加入健壮的异常处理和日志记录。比如，某个文件上传失败，不应导致整个程序崩溃，而是记录错误后跳过，继续处理下一个。

注意：飞书语音识别接口通常有频率限制（QPS）和每日调用额度。在批量处理大量文件时，需要在代码中主动加入延迟（如time.sleep），避免触发限流。同时，要处理网络超时、服务端错误等异常，并设计重试机制。

整个架构是典型的事件驱动管道（Pipeline）模式，每个模块职责单一，通过清晰的数据（文件路径、file_key、block_id、识别文本）进行传递。这种设计使得代码易于测试、调试和扩展。

3. 环境准备与核心依赖详解

3.1 飞书开放平台应用创建与配置

这是整个项目的地基，一步错，步步错。我们从头开始捋一遍。

首先，访问飞书开放平台，用你的飞书账号登录。如果你是企业管理员，最好创建一个专门用于自动化工具的“自建应用”；如果是个人使用，也可以。

创建应用：在开发者后台点击“创建企业自建应用”。应用名称可以叫“语音识别助手”，描述按实填写。
获取凭证：创建成功后，在“凭证与基础信息”页面，你会看到App ID和App Secret。把它们像银行密码一样保管好。我们后续的脚本将通过这两个值来证明“我是谁”。
开通权限：在“权限管理”页面，找到并添加以下关键权限：
- bitable:app：读写多维表格的权限。
- bitable:table:readonly或bitable:table:full_access：根据你是只读还是需要创建表来选择。
- speech_to_text:speech_file：语音文件识别权限。这是核心权限，必须添加。
- drive:drive:readonly或drive:drive:full_access：上传文件到云空间需要驱动器的权限。
- 注意：添加权限后，必须点击“申请线上发布”或“版本管理与发布”创建一个新版本并申请发布。部分权限（尤其是敏感权限）可能需要企业管理员审核。在测试阶段，你可以将应用添加到“企业可用应用”或直接安装到你的测试环境中。
获取表格Token：在你飞书的工作台，手动创建一个多维表格。比如，表头设计为：文件名（文本）、识别状态（单选：处理中/成功/失败）、识别文本（多行文本）、文件Key（文本）、上传时间（日期时间）。创建好后，浏览器地址栏的URL中，你能找到两个关键参数：appToken=和tableId=。把它们记下来。appToken是表格的唯一标识，tableId是表格内具体页面的标识。

3.2 本地Python环境与依赖库

项目基于Python，因为它有丰富的HTTP库和易于编写的脚本。建议使用Python 3.8及以上版本。

创建一个独立的虚拟环境是良好的习惯，可以避免包冲突：

python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate

核心依赖库通常包括：

requests: 用于调用飞书的所有HTTP API。这是绝对的核心。
python-dotenv: 用于从.env文件加载环境变量，安全地管理App ID和App Secret。
pydantic/dataclasses: 用于定义清晰的数据结构，比如API请求体和响应体模型，让代码更健壮、易读。
tenacity或retrying: 用于为网络请求添加优雅的重试机制，应对网络波动或飞书API的瞬时故障。
rich或tqdm: 可选，用于在命令行中显示漂亮的进度条和日志，提升交互体验。

你可以创建一个requirements.txt文件：

requests>=2.28.0 python-dotenv>=0.21.0 pydantic>=1.10.0 tenacity>=8.2.0 rich>=13.0.0

然后通过pip install -r requirements.txt安装。

3.3 项目配置文件与安全实践

绝对不要将敏感信息写在代码里！我们使用.env文件来配置。

在项目根目录创建.env文件：

# 飞书应用凭证 FEISHU_APP_ID=cli_xxxxxx FEISHU_APP_SECRET=xxxxxxxxxxxxxxxxxxxxxxxx # 飞书多维表格标识 FEISHU_APP_TOKEN=xxxxxxxxxxxxxxxxxxxxxxxx FEISHU_TABLE_ID=tblxxxxxxxxxxxxxxxxxxxxxxxx # 音频文件目录 AUDIO_FILE_DIR=./audio_files # 轮询间隔（秒） POLLING_INTERVAL=5 # 最大轮询次数 MAX_POLLING_COUNT=60

然后在Python代码中，通过python-dotenv加载：

import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量到环境变量 APP_ID = os.getenv('FEISHU_APP_ID') APP_SECRET = os.getenv('FEISHU_APP_SECRET') APP_TOKEN = os.getenv('FEISHU_APP_TOKEN') TABLE_ID = os.getenv('FEISHU_TABLE_ID') AUDIO_DIR = os.getenv('AUDIO_FILE_DIR', './audio')

安全提示：务必在.gitignore文件中加入.env，确保这个包含秘密的文件不会被意外提交到Git仓库。将.env.example（仅包含键名，无真实值）提交到仓库，作为配置模板供其他协作者使用。

4. 核心代码模块实现与解析

4.1 飞书API客户端封装

与飞书的所有交互都通过其开放API完成。一个好的客户端封装能大幅提升代码的可维护性。我们封装一个FeishuClient类。

import requests import time from tenacity import retry, stop_after_attempt, wait_exponential class FeishuClient: def __init__(self, app_id, app_secret): self.app_id = app_id self.app_secret = app_secret self._tenant_access_token = None self._token_expire_time = 0 self.base_url = "https://open.feishu.cn/open-apis" def _get_tenant_access_token(self): """获取或刷新租户访问令牌。令牌有效期为2小时，需要缓存和刷新。""" now = time.time() if self._tenant_access_token and now < self._token_expire_time - 60: # 提前1分钟刷新 return self._tenant_access_token url = f"{self.base_url}/auth/v3/tenant_access_token/internal" payload = {"app_id": self.app_id, "app_secret": self.app_secret} resp = requests.post(url, json=payload) resp.raise_for_status() data = resp.json() if data.get("code") == 0: self._tenant_access_token = data["tenant_access_token"] self._token_expire_time = now + data["expire"] # expire单位是秒 return self._tenant_access_token else: raise Exception(f"Failed to get tenant access token: {data}") @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def _request(self, method, endpoint, **kwargs): """带重试和自动令牌管理的通用请求方法。""" token = self._get_tenant_access_token() headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json; charset=utf-8", } if 'headers' in kwargs: headers.update(kwargs.pop('headers')) url = f"{self.base_url}{endpoint}" response = requests.request(method, url, headers=headers, **kwargs) # 检查飞书业务码，0表示成功 if response.status_code == 200: json_data = response.json() if json_data.get('code') == 0: return json_data.get('data') else: # 处理业务逻辑错误，例如无权限、参数错误等 raise Exception(f"Feishu API error [{json_data.get('code')}]: {json_data.get('msg')}") response.raise_for_status() # 处理HTTP错误如404, 500等 # 具体的API方法 def upload_file(self, file_path): """上传文件到飞书云空间。""" # 注意：飞书的上传接口需要分两步：1. 准备上传 2. 上传文件块。这里简化示意。 # 实际需参考飞书最新文档：/open-apis/drive/v1/files/upload_all with open(file_path, 'rb') as f: files = {'file': (os.path.basename(file_path), f, 'audio/mpeg')} # 根据实际类型调整 # 上传接口的endpoint和参数可能不同，需调整 data = self._request('POST', '/drive/v1/files/upload_all', files=files) return data.get('file_token') # 即 file_key def speech_to_text(self, file_key): """发起语音识别请求。""" payload = { "file_key": file_key, "config": {"language": "zh"} # 可配置语言，zh-中文，en-英文等 } data = self._request('POST', '/speech_to_text/v1/speech/file_recognize', json=payload) return data.get('block_id') # 返回异步任务ID def get_speech_recognition_result(self, block_id): """根据block_id查询语音识别结果。""" params = {'block_id': block_id} data = self._request('GET', '/speech_to_text/v1/speech/file_recognize', params=params) # 返回结构复杂，包含状态和结果 return data def add_bitable_record(self, app_token, table_id, fields): """向多维表格添加一条记录。""" payload = { "fields": fields # fields是一个字典，键是字段名，值是字段值 } data = self._request('POST', f'/bitable/v1/apps/{app_token}/tables/{table_id}/records', json=payload) return data.get('record', {}).get('record_id')

这个客户端封装了令牌管理、重试逻辑和基础错误处理，后续所有操作都基于它进行。

4.2 音频文件处理与状态管理

我们需要一个模块来管理本地音频文件的状态，防止重复处理，并记录处理日志。

import os import json import hashlib from pathlib import Path from dataclasses import dataclass, asdict from typing import Optional @dataclass class AudioFile: path: Path # 文件路径 file_name: str # 文件名 file_size: int # 文件大小 status: str = 'pending' # pending, uploading, recognizing, completed, failed feishu_file_key: Optional[str] = None recognition_block_id: Optional[str] = None recognized_text: Optional[str] = None error_message: Optional[str] = None @property def md5(self): """计算文件MD5，用于去重标识。""" hash_md5 = hashlib.md5() with open(self.path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() class AudioFileManager: def __init__(self, audio_dir: str, state_file: str = 'processing_state.json'): self.audio_dir = Path(audio_dir) self.state_file = Path(state_file) self._state_cache = {} # key: file_md5, value: AudioFile (序列化后的dict) self._load_state() def scan_audio_files(self, extensions=('.mp3', '.wav', '.m4a', '.aac')): """扫描音频目录，返回待处理的AudioFile对象列表。""" pending_files = [] for ext in extensions: for file_path in self.audio_dir.rglob(f'*{ext}'): if not file_path.is_file(): continue audio_file = AudioFile( path=file_path, file_name=file_path.name, file_size=file_path.stat().st_size ) # 检查是否已处理过（根据MD5或路径） if audio_file.md5 in self._state_cache: cached = self._state_cache[audio_file.md5] if cached.get('status') == 'completed': print(f"[跳过] 文件已处理完成: {file_path.name}") continue pending_files.append(audio_file) return pending_files def update_file_status(self, audio_file: AudioFile): """更新文件状态并持久化到本地JSON文件。""" self._state_cache[audio_file.md5] = asdict(audio_file) self._save_state() def _load_state(self): """从JSON文件加载处理状态。""" if self.state_file.exists(): try: with open(self.state_file, 'r', encoding='utf-8') as f: self._state_cache = json.load(f) except json.JSONDecodeError: self._state_cache = {} def _save_state(self): """保存处理状态到JSON文件。""" with open(self.state_file, 'w', encoding='utf-8') as f: json.dump(self._state_cache, f, ensure_ascii=False, indent=2)

这个管理器通过MD5和本地状态文件，实现了简单的断点续传和去重功能。即使程序中途中断，重新运行也不会从头开始处理所有文件。

4.3 主流程串联与错误处理

最后，我们将所有模块串联起来，形成完整的主程序逻辑。这里要特别注意错误处理和流程控制。

import time from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn def main(): # 初始化客户端和文件管理器 client = FeishuClient(APP_ID, APP_SECRET) manager = AudioFileManager(AUDIO_DIR) # 扫描待处理文件 pending_files = manager.scan_audio_files() if not pending_files: print("没有发现新的待处理音频文件。") return print(f"发现 {len(pending_files)} 个待处理文件。") # 使用rich库创建进度条 with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeElapsedColumn(), ) as progress: main_task = progress.add_task("[cyan]处理音频文件中...", total=len(pending_files)) for audio_file in pending_files: progress.update(main_task, description=f"处理: {audio_file.file_name[:30]}...") try: # 步骤1: 更新状态为上传中 audio_file.status = 'uploading' manager.update_file_status(audio_file) # 步骤2: 上传文件到飞书 file_key = client.upload_file(str(audio_file.path)) audio_file.feishu_file_key = file_key audio_file.status = 'uploaded' manager.update_file_status(audio_file) progress.console.print(f" [green]✓[/green] 上传成功: {audio_file.file_name}") # 步骤3: 发起语音识别 block_id = client.speech_to_text(file_key) audio_file.recognition_block_id = block_id audio_file.status = 'recognizing' manager.update_file_status(audio_file) progress.console.print(f" [blue]i[/blue] 识别任务已提交，ID: {block_id[:8]}...") # 步骤4: 轮询识别结果 max_polls = int(os.getenv('MAX_POLLING_COUNT', 60)) poll_interval = int(os.getenv('POLLING_INTERVAL', 5)) for i in range(max_polls): time.sleep(poll_interval) result_data = client.get_speech_recognition_result(block_id) status = result_data.get('recognition_status') if status == 'success': # 解析识别文本 # 实际结构可能更复杂，需要根据飞书返回格式调整 full_text = "" for segment in result_data.get('recognition_text', []): full_text += segment.get('text', '') + "\n" audio_file.recognized_text = full_text.strip() audio_file.status = 'completed' manager.update_file_status(audio_file) progress.console.print(f" [green]✓[/green] 识别成功！") break elif status == 'failed': audio_file.status = 'failed' audio_file.error_message = result_data.get('error_msg', 'Unknown error') manager.update_file_status(audio_file) progress.console.print(f" [red]✗[/red] 识别失败: {audio_file.error_message}") break else: # 'running' or others if i % 5 == 0: # 每5次轮询打印一次状态 progress.console.print(f" [yellow]…[/yellow] 识别中... ({i+1}/{max_polls})") else: # 轮询超时 audio_file.status = 'failed' audio_file.error_message = 'Polling timeout' manager.update_file_status(audio_file) progress.console.print(f" [red]✗[/red] 轮询超时，识别未完成。") # 步骤5: 将结果写入飞书多维表格 (仅成功时) if audio_file.status == 'completed' and audio_file.recognized_text: fields = { "文件名": audio_file.file_name, "识别状态": "成功", "识别文本": audio_file.recognized_text, "文件Key": audio_file.feishu_file_key, "上传时间": int(time.time() * 1000) # 飞书时间戳是毫秒 } record_id = client.add_bitable_record(APP_TOKEN, TABLE_ID, fields) if record_id: progress.console.print(f" [green]✓[/green] 记录已写入表格，ID: {record_id}") else: progress.console.print(f" [yellow]![/yellow] 识别成功，但写入表格失败。") except Exception as e: # 捕获任何未处理的异常 audio_file.status = 'failed' audio_file.error_message = str(e) manager.update_file_status(audio_file) progress.console.print(f" [red]✗[/red] 处理过程异常: {e}") # 可以选择是否继续处理下一个文件 # continue finally: # 无论成功失败，更新进度条 progress.advance(main_task) # 建议在每次循环后加入短暂延迟，避免对飞书API造成过大压力 time.sleep(1) print("\n[bold green]所有文件处理完成！[/bold green]") # 可以在这里生成一个简单的处理报告 completed = sum(1 for f in manager._state_cache.values() if f.get('status') == 'completed') failed = sum(1 for f in manager._state_cache.values() if f.get('status') == 'failed') print(f"统计: 成功 {completed} 个，失败 {failed} 个。")

这个主流程包含了完整的异常捕获、状态更新、进度反馈和结果汇总，是一个健壮的生产级脚本雏形。

5. 高级功能扩展与优化思路

基础功能跑通后，我们可以根据实际需求，对这个工具进行深度定制和优化。

5.1 支持更多音频格式与预处理

飞书语音识别接口对音频格式、编码、采样率、文件大小可能有限制。我们可以在上传前加入一个预处理环节。

格式转换：使用pydub或ffmpeg-python库，将不支持的格式（如.flac,.ogg）转换为支持的格式（如.mp3）。
音频分割：飞书接口可能有单文件时长限制（例如1小时）。对于超长录音，可以使用pydub按静音检测或固定时长进行分割，然后分批上传识别。
音量标准化与降噪：对于质量较差的录音，可以进行简单的预处理提升识别率。不过，飞书引擎本身抗噪能力较强，这一步通常非必需。

# 示例：使用pydub检查并转换格式 from pydub import AudioSegment def preprocess_audio(input_path, output_dir, target_format='mp3', target_bitrate='128k'): """预处理音频文件，转换格式和码率。""" audio = AudioSegment.from_file(input_path) # 简单处理：统一转换为单声道、16kHz采样率（如果原文件质量过高） # audio = audio.set_channels(1).set_frame_rate(16000) output_path = Path(output_dir) / (Path(input_path).stem + f'.{target_format}') audio.export(output_path, format=target_format, bitrate=target_bitrate) return output_path

5.2 识别结果后处理与结构化

飞书返回的可能是带时间戳的段落文本。我们可以进行后处理：

文本清洗：去除多余的换行、空格，处理识别错误的常见符号。
说话人分离：如果录音是多人对话，且飞书返回了说话人标签（部分高级API支持），可以按说话人整理文本。
关键信息提取：结合NLP库（如jieba分词，hanlp），提取摘要、关键词、时间点、任务项等。
结构化入库：不仅仅是存入一个“识别文本”字段。可以设计更丰富的多维表格列，如“发言人”、“时间点”、“关键动作”、“待办事项”，并尝试用规则或简单模型自动填充这些字段。

5.3 实现增量同步与监听模式

目前的脚本是“一次性”的。我们可以将其改造成一个常驻服务：

监听文件夹：使用watchdog库监听AUDIO_DIR，一旦有新的音频文件放入，自动触发处理流程。
增量同步：定期扫描文件夹，只处理新增或修改过的文件（通过对比文件MD5和修改时间）。
与日历/会议集成：通过飞书日历API，自动获取即将开始或已结束的会议，并尝试关联会议录音文件进行处理，实现从“会议预约”到“纪要生成”的全自动化。

5.4 构建图形化界面或Webhook服务

为了让非技术同事也能使用，可以：

构建简单的CLI界面：使用typer或click库，提供丰富的命令行参数，如指定目录、指定表格、重试失败文件等。
开发轻量级Web界面：使用Flask或FastAPI搭建一个内部网页，提供文件上传、处理状态查看、结果预览等功能。
提供Webhook回调：将脚本部署到服务器，并提供API端点。其他系统（如OA审批流、IM机器人）在产生音频文件后，可以调用这个Webhook来触发处理，并将结果回传。

6. 部署、监控与最佳实践

6.1 本地运行与服务器部署

本地运行：配置好.env文件后，直接python main.py即可。适合处理临时性、小批量的任务。

服务器部署（Docker化）：为了长期稳定运行，建议将项目Docker化。

# Dockerfile FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 安装ffmpeg用于音频预处理（如果用到pydub） RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/* CMD ["python", "main.py"]

可以结合cron或systemd timer定时运行容器，或者作为常驻服务运行监听模式的脚本。

6.2 日志、监控与告警

任何自动化脚本都必须有完善的日志。

结构化日志：使用logging模块，配置不同的级别（INFO, WARNING, ERROR），并输出到文件和控制台。可以使用JSON格式，便于后续用ELK等工具分析。

import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('feishu_voice.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__)

关键指标监控：记录处理文件数、成功率、平均处理时长、API调用次数等。
错误告警：可以将ERROR级别的日志通过飞书机器人webhook发送到指定的群聊，实现实时告警。

6.3 成本控制与API限额管理

飞书开放API有调用频率限制（流控）。务必在代码中做好管理：

阅读官方文档：仔细查看你所使用的每个接口（特别是语音识别和文件上传）的QPS（每秒查询率）和每日限额。
主动限速：在批量处理的循环中，使用time.sleep()主动加入延迟，确保不会触发流控。例如，在上传文件和发起识别请求之间加入1-2秒间隔。

处理流控错误：在_request方法中，捕获特定的流控错误码（如99991400），并实现指数退避重试。

# 在 _request 方法的异常处理部分 if json_data.get('code') == 99991400: # 假设这是流控错误码 wait_time = 60 # 等待60秒再重试 time.sleep(wait_time) return self._request(method, endpoint, **kwargs) # 递归重试