尧图网站建设 
简介这是一份面向Python开发者的阿里云音频转字幕源码包基于阿里云智能语音服务的录音文件识别API实现视频、音频文件到SRT字幕的自动转写。代码覆盖音频上传、识别任务提交、结果查询与字幕文件生成等完整流程整个调用采用异步任务处理方式适合批量处理播客、课程录像、会议录音等场景也适合希望在不自行建模的前提下快速接入语音识别能力的开发者。资源共2000个文件以Python源码(.py)、字节码(.pyc)和类型标注(.pyi)为主另有动态库(.dll/.pyd)、wheel包、配置文件与说明文档等运行时依赖压缩包整体25.34MB结构相对紧凑便于下载后本地部署。已有290人学习。随包可获取可运行的完整源码、第三方依赖清单与阿里云API接入示例包内还包含CFFI头文件、JSON配置等SDK底层组件有助于读者理解录音文件识别接口的参数含义与异步回调时序也可据此做二次开发或改造成批量转写工具。1. 录音文件转 srt真正的难点在异步任务和时间戳对齐手里一批讲座录像、播客录音要出 srt 字幕。直觉是把 mp3 丢给阿里云录音文件识别 API等几秒拿回文本转成 srt 就完事。实际动手会发现坑全在流程里API 要的是公网可访问的文件 URL不是本地路径识别是异步的提交任务后得轮询状态返回结果是带毫秒时间戳的 JSONsrt 时间轴、序号、断句合并都要自己拼。下面按这个顺序从 Python 调通录音文件识别到 ffmpeg 提取音轨再到把 sentences 转成播放器能认的 srt 字幕每一步的参数和报错都写清楚。适合后端、自动化和做媒资批处理的工程师直接抄代码改改就能用。2. 阿里云录音文件识别的鉴权与轮询从 token 到 task_id2.1 为什么录音文件识别是「提交 轮询」而不是同步返回一句话识别可以同步返回因为音频就几秒钟服务端等得起。录音文件识别面对的是几十秒到几个小时的音频解码、断句、说话人分离都是耗时操作HTTP 长连接等不住所以 RESTful 接口设计成两步先提交文件 URL返回一个 task_id再拿 task_id 去轮询直到 Status 变成成功或失败。这个模型和语音合成不一样很多人第一次写就卡在这里。接口全景可以先记这张表动作请求方法接口关键入参返回获取 TokenPOSTnls-meta 的 token 服务AppKey、AccessKeyId、AccessKeySecretToken、ExpireTime提交识别任务POSTfiletrans 的 trans-requestappkey、token、file_linkTaskId查询任务结果POSTfiletrans 的 taskinfoappkey、token、task_idStatus、Result三个值要拎清楚AppKey 是智能语音交互项目标识在阿里云控制台创建AccessKeyId 和 AccessKeySecret 是账号密钥Token 是前两者换来的临时凭证有效期大约 24 小时脚本跑批时取一次就行不用每个文件都换。2.2 Python 获取 AccessToken 的最小代码import requests TOKEN_URL https://nls-meta.cn-shanghai.aliyuncs.com/esequence-api/token/v1 APP_KEY 你的AppKey AK_ID 你的AccessKeyId AK_SECRET 你的AccessKeySecret def get_token() - str: resp requests.post( TOKEN_URL, json{ AppKey: APP_KEY, AccessKeyId: AK_ID, AccessKeySecret: AK_SECRET, }, timeout10, ) resp.raise_for_status() data resp.json() # 返回结构因账号体系略有差异拿不到 Token 时打印 data 看实际字段 return data.get(Token) or data.get(Data, {}).get(Token)逻辑说明一次 POST 换 token超时设 10 秒足够。token 24 小时有效不要每次都调写个模块级缓存比每次重新换省时间。AccessKey 泄露风险比 token 大生产环境别把 AK 明文写进代码用环境变量注入。2.3 提交识别任务enable_words 与 file_link 的坑FILE_LINK https://your-bucket.oss-cn-shanghai.aliyuncs.com/audio/lecture-01.wav def submit_task(token: str, file_link: str) - str: resp requests.post( https://filetrans.cn-shanghai.aliyuncs.com/api/v1/rest/trans-request/v1, json{ appkey: APP_KEY, token: token, file_link: file_link, enable_words: True, max_sentence_silence: 500, }, timeout30, ) data resp.json() if data.get(StatusCode) SUCCESS: return data[TaskId] raise RuntimeError(fsubmit failed: {data})参数说明file_link 必须是 HTTP(S) 公网地址本地路径、内网地址都无效。enable_words 决定返回里有没有词级时间戳做普通 srt 用句子级就够开了会明显拉大返回体。max_sentence_silence 是断句静音阈值单位毫秒默认 500字幕场景后面会调到 700 到 1000。2.3.1 私有 OSS 文件怎么传 file_link录音文件识别服务端要主动拉文件所以私有的 OSS 文件直接传链接会报错。提交前生成签名 URLimport oss2 auth oss2.Auth(AK_ID, AK_SECRET) bucket oss2.Bucket(auth, https://oss-cn-shanghai.aliyuncs.com, your-bucket) signed_url bucket.sign_url(GET, audio/lecture-01.wav, 600)sign_url 的第三个参数是有效期秒数识别任务长就设 3600 以上签名 URL 过期时任务可能还在排队超长音频建议把过期时间放宽到一小时。2.4 查询结果先打印一次返回再适配 Status 字段def query_task(token: str, task_id: str) - dict: resp requests.post( https://filetrans.cn-shanghai.aliyuncs.com/api/v1/rest/trans-request/v1/taskinfo, json{appkey: APP_KEY, token: token, task_id: task_id}, timeout30, ) return resp.json() data query_task(token, task_id) print(data) # 先看真实字段再写适配逻辑常见返还有两种新版 Status 直接是字符串 RUNNING / SUCCESS / FAILED旧版 status 用数字 0 排队、1 识别中、2 成功、3 失败。不要凭记忆写死打印一次再定。任务成功时识别内容在 data[Result]注意它是字符串不是对象下一章就是处理它。3. 从 JSON 到 srtffmpeg 提取音频、时间轴格式化与句子合并3.1 ffmpeg 把视频转成 16k 单声道 wav录音文件识别支持 wav、mp3、m4a、ogg 等音频格式但不吃 mp4 这种视频容器。手里是视频时先拿 ffmpeg 把音轨剥出来ffmpeg -i lecture.mp4 -vn -acodec pcm_s16le -ar 16000 -ac 1 lecture.wav参数拆解-vn 丢掉视频流只处理音频-acodec pcm_s16le 是 PCM 16bit 编码wav 兼容性最好-ar 16000 把采样率重采样到 16k识别引擎主要在 8k/16k 上训练48k 原声不会提升效果只会变大文件-ac 1 转单声道双声道也能转写但声道选择、音量归一都会多出不可控因素进 API 前合并成单声道最省心。3.2 解析 sentencesResult 是字符串要先 json.loads从查询接口拿到的 data[Result] 在多数版本里是 JSON 字符串不是 dict。直接 data[Result][Sentences] 会报 TypeError。正确做法import json result json.loads(data[Result]) sentences result[Sentences] for s in sentences: print(s[Text], s[BeginTime], s[EndTime])常见字段不一定全有Text 是句子文本BeginTime / EndTime 是句子在音频里的起止毫秒ChannelId 是多声道任务的声道编号SpeakerId 在说话人分离开启后才有会议纪要场景按人聚合文本时用。不要按一套字段写死不同版本的识别引擎字段名可能是 Sentences 也可能是 sentences先打印 result.keys() 再写解析。3.3 毫秒转 srt 时间轴格式化函数怎么写srt 协议对时间轴的格式要求是 hh:mm:ss,mmm小时两位、毫秒三位毫秒位用逗号不是点。写一个纯函数def format_srt_time(ms: int) - str: ms max(0, int(ms)) hours ms // 3_600_000 minutes (ms % 3_600_000) // 60_000 seconds (ms % 60_000) // 1000 millis ms % 1000 return f{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}逻辑说明先除后模不会出现小时、分钟借位的进位错误毫秒直接截断而不是四舍五入因为四舍五入到 1000 会让结束时间等于下一秒 000破坏时间轴单调性。字幕播放器对毫秒位要求不严截断完全够用。3.4 按静音间隔合并句子避免字幕刷屏录音文件识别默认按 500ms 静音断句断句比人眼读字幕的节奏快直接一句一字幕会闪屏。常见做法是按句间空隙合并空隙小于阈值的句子合成一个字幕块间隙超过阈值的另起一块。GAP_MS 800 # 句间空隙超过 800ms 才拆成两条字幕 blocks [] current [] for s in sentences: if not current: current [s] continue if s[BeginTime] - current[-1][EndTime] GAP_MS: current.append(s) else: blocks.append(current) current [s] if current: blocks.append(current)然后写文件def write_srt(blocks, out_pathoutput.srt): lines [] for idx, block in enumerate(blocks, start1): start block[0][BeginTime] end block[-1][EndTime] text .join(x[Text] for x in block).strip() lines.append(str(idx)) lines.append(f{format_srt_time(start)} -- {format_srt_time(end)}) lines.append(text) lines.append() with open(out_path, w, encodingutf-8) as f: f.write(\n.join(lines)) write_srt(blocks)文本拼接用空字符串直接 join因为多数识别结果自带标点如果模型不带标点句间要自己补常见做法是判断上一句结尾字符不是标点就补逗号。字幕文件用 UTF-8 写老播放器乱码时把编码改成 utf-8-sig 即可。4. 长音频与多说话人的 4 个必调参数从 enable_words 到 400 排错4.1 4 个参数enable_words、speaker_diarization、max_sentence_silence、vocabulary_id提交任务的 JSON 里除了 appkey、token、file_link还有几个参数对字幕质量影响很大参数默认值作用推荐场景enable_wordsfalse返回词级时间戳逐字字幕、歌词滚动才需要max_sentence_silence500断句静音阈值单位 ms字幕合并想少就用 700 到 1000speaker_diarization-1-1 或 0 不分离1 句子级2 词级会议、客服通话录音设 1vocabulary_id无热词表 ID提升人名和专有名词识别错词反复出现时创建speaker_diarization 值得多说一句它只影响返回的 SpeakerId 字段文本本身不变开词级分离2会明显增加计算时间对话字幕一般句子级1就够。vocabulary_id 在控制台热词表里配置传表 ID注意热词生效有缓存延迟改完词表要等一两分钟再提任务。4.2 提交和查询阶段常见的 400 错误录音文件识别报错经常是 HTTP 200、body 里带错误码不要只盯状态码InvalidAppKeyAppKey 不存在或没开通录音文件识别能力去控制台核对该项目是否开通服务。InvalidTimeStamp.Expiredtoken 过期或本地时钟偏差换新 token容器里时钟漂移也会触发先 date 看一眼。400 content exists risk / ContentRiskDetected内容或文件名触发安全策略把文件名改成中性描述音频内容也检查一遍。TaskNotFoundExceptiontask_id 不存在或查询结果已过保留期说明重试窗口太长需要重新提交。排错第一步是打印完整返回def _check(data: dict) - dict: code data.get(Code) or data.get(StatusCode) if code not in (None, SUCCESS): raise RuntimeError(f{code}: {data.get(Message) or data.get(Msg)}) return data注意错误码字段名同样存在版本差异Code / StatusCode / ErrCode 都可能出现用打印而非猜。4.3 轮询频率与重试指数退避怎么写提交后的轮询建议 1 到 2 秒一次不要 0.2 秒疯狂扫。识别几分钟的任务每两秒查一次足够网络抖动时加指数退避import time def wait_result(token: str, task_id: str, timeout1800): deadline time.time() timeout interval 2 while time.time() deadline: data query_task(token, task_id) status data.get(Status) or data.get(status) if status in (SUCCESS, 2): return data if status in (FAILED, 3): raise RuntimeError(data.get(StatusText, data)) time.sleep(interval) interval min(interval * 1.5, 10) # 退避上限 10 秒 raise TimeoutError(ftask {task_id} timeout)逻辑说明每次等待时间翻 1.5 倍避免服务端把高频轮询当异常流量限流。注意这个退避只针对轮询任务本身失败或超时后重新提交task_id 是新的别拿旧 task_id 无限重试。5. 批量转写的断点续转用本地缓存避免重复计费5.1 用本地 JSON 缓存 task_id 和结果批量转写大量音频时最怕两件事跑到一半断网重启后又从第一个文件重新提交已经识别完的又重新计费。录音文件识别按次计费一个文件提交两次就是两份账单。我的做法是把 task_id 和最终 srt 路径写进本地 JSON作为断点续转的依据import json import pathlib CACHE_FILE trans_cache.json def load_cache(): if pathlib.Path(CACHE_FILE).exists(): return json.loads(pathlib.Path(CACHE_FILE).read_text(encodingutf-8)) return {} def save_cache(cache): pathlib.Path(CACHE_FILE).write_text( json.dumps(cache, ensure_asciiFalse, indent2), encodingutf-8 )主循环里先查缓存再提交cache load_cache() for wav in wav_files: srt_path wav.with_suffix(.srt) if str(wav) in cache and srt_path.exists(): continue # 已转写完成跳过 signed_url bucket.sign_url(GET, str(wav), 3600) task_id submit_task(token, signed_url) cache[str(wav)] {task_id: task_id, status: submitted} save_cache(cache)提交后崩了重启时按 task_id 继续查询就行不需要重新 submit任务失败时删掉缓存项再重新提交避免死循环。每次写入都落盘而不是最后统一写防止中途异常丢缓存。5.2 用 ffprobe 校验时长和字幕有没有丢尾句批量流程跑完用 ffprobe 拿音频时长和 srt 最后一条字幕的 EndTime 比对ffprobe -v error -show_entries formatduration -of csvp0 lecture.wav音频 3600 秒字幕结尾只有 3000 秒说明尾段识别失败或任务被截断。先切段重新识别切段时要带 500ms 前后重叠按重叠区文本对齐避免切点切掉半个词。时长对不上时问题通常出在转码时的采样率或声道参数把 ffmpeg 参数回到 16k 单声道 PCM 再试一次大部分情况能解决。本文还有配套的精品资源点击获取 录音文件转SRT字幕：阿里云语音识别异步处理与时间戳对齐实践
简介这是一份面向Python开发者的阿里云音频转字幕源码包基于阿里云智能语音服务的录音文件识别API实现视频、音频文件到SRT字幕的自动转写。代码覆盖音频上传、识别任务提交、结果查询与字幕文件生成等完整流程整个调用采用异步任务处理方式适合批量处理播客、课程录像、会议录音等场景也适合希望在不自行建模的前提下快速接入语音识别能力的开发者。资源共2000个文件以Python源码(.py)、字节码(.pyc)和类型标注(.pyi)为主另有动态库(.dll/.pyd)、wheel包、配置文件与说明文档等运行时依赖压缩包整体25.34MB结构相对紧凑便于下载后本地部署。已有290人学习。随包可获取可运行的完整源码、第三方依赖清单与阿里云API接入示例包内还包含CFFI头文件、JSON配置等SDK底层组件有助于读者理解录音文件识别接口的参数含义与异步回调时序也可据此做二次开发或改造成批量转写工具。1. 录音文件转 srt真正的难点在异步任务和时间戳对齐手里一批讲座录像、播客录音要出 srt 字幕。直觉是把 mp3 丢给阿里云录音文件识别 API等几秒拿回文本转成 srt 就完事。实际动手会发现坑全在流程里API 要的是公网可访问的文件 URL不是本地路径识别是异步的提交任务后得轮询状态返回结果是带毫秒时间戳的 JSONsrt 时间轴、序号、断句合并都要自己拼。下面按这个顺序从 Python 调通录音文件识别到 ffmpeg 提取音轨再到把 sentences 转成播放器能认的 srt 字幕每一步的参数和报错都写清楚。适合后端、自动化和做媒资批处理的工程师直接抄代码改改就能用。2. 阿里云录音文件识别的鉴权与轮询从 token 到 task_id2.1 为什么录音文件识别是「提交 轮询」而不是同步返回一句话识别可以同步返回因为音频就几秒钟服务端等得起。录音文件识别面对的是几十秒到几个小时的音频解码、断句、说话人分离都是耗时操作HTTP 长连接等不住所以 RESTful 接口设计成两步先提交文件 URL返回一个 task_id再拿 task_id 去轮询直到 Status 变成成功或失败。这个模型和语音合成不一样很多人第一次写就卡在这里。接口全景可以先记这张表动作请求方法接口关键入参返回获取 TokenPOSTnls-meta 的 token 服务AppKey、AccessKeyId、AccessKeySecretToken、ExpireTime提交识别任务POSTfiletrans 的 trans-requestappkey、token、file_linkTaskId查询任务结果POSTfiletrans 的 taskinfoappkey、token、task_idStatus、Result三个值要拎清楚AppKey 是智能语音交互项目标识在阿里云控制台创建AccessKeyId 和 AccessKeySecret 是账号密钥Token 是前两者换来的临时凭证有效期大约 24 小时脚本跑批时取一次就行不用每个文件都换。2.2 Python 获取 AccessToken 的最小代码import requests TOKEN_URL https://nls-meta.cn-shanghai.aliyuncs.com/esequence-api/token/v1 APP_KEY 你的AppKey AK_ID 你的AccessKeyId AK_SECRET 你的AccessKeySecret def get_token() - str: resp requests.post( TOKEN_URL, json{ AppKey: APP_KEY, AccessKeyId: AK_ID, AccessKeySecret: AK_SECRET, }, timeout10, ) resp.raise_for_status() data resp.json() # 返回结构因账号体系略有差异拿不到 Token 时打印 data 看实际字段 return data.get(Token) or data.get(Data, {}).get(Token)逻辑说明一次 POST 换 token超时设 10 秒足够。token 24 小时有效不要每次都调写个模块级缓存比每次重新换省时间。AccessKey 泄露风险比 token 大生产环境别把 AK 明文写进代码用环境变量注入。2.3 提交识别任务enable_words 与 file_link 的坑FILE_LINK https://your-bucket.oss-cn-shanghai.aliyuncs.com/audio/lecture-01.wav def submit_task(token: str, file_link: str) - str: resp requests.post( https://filetrans.cn-shanghai.aliyuncs.com/api/v1/rest/trans-request/v1, json{ appkey: APP_KEY, token: token, file_link: file_link, enable_words: True, max_sentence_silence: 500, }, timeout30, ) data resp.json() if data.get(StatusCode) SUCCESS: return data[TaskId] raise RuntimeError(fsubmit failed: {data})参数说明file_link 必须是 HTTP(S) 公网地址本地路径、内网地址都无效。enable_words 决定返回里有没有词级时间戳做普通 srt 用句子级就够开了会明显拉大返回体。max_sentence_silence 是断句静音阈值单位毫秒默认 500字幕场景后面会调到 700 到 1000。2.3.1 私有 OSS 文件怎么传 file_link录音文件识别服务端要主动拉文件所以私有的 OSS 文件直接传链接会报错。提交前生成签名 URLimport oss2 auth oss2.Auth(AK_ID, AK_SECRET) bucket oss2.Bucket(auth, https://oss-cn-shanghai.aliyuncs.com, your-bucket) signed_url bucket.sign_url(GET, audio/lecture-01.wav, 600)sign_url 的第三个参数是有效期秒数识别任务长就设 3600 以上签名 URL 过期时任务可能还在排队超长音频建议把过期时间放宽到一小时。2.4 查询结果先打印一次返回再适配 Status 字段def query_task(token: str, task_id: str) - dict: resp requests.post( https://filetrans.cn-shanghai.aliyuncs.com/api/v1/rest/trans-request/v1/taskinfo, json{appkey: APP_KEY, token: token, task_id: task_id}, timeout30, ) return resp.json() data query_task(token, task_id) print(data) # 先看真实字段再写适配逻辑常见返还有两种新版 Status 直接是字符串 RUNNING / SUCCESS / FAILED旧版 status 用数字 0 排队、1 识别中、2 成功、3 失败。不要凭记忆写死打印一次再定。任务成功时识别内容在 data[Result]注意它是字符串不是对象下一章就是处理它。3. 从 JSON 到 srtffmpeg 提取音频、时间轴格式化与句子合并3.1 ffmpeg 把视频转成 16k 单声道 wav录音文件识别支持 wav、mp3、m4a、ogg 等音频格式但不吃 mp4 这种视频容器。手里是视频时先拿 ffmpeg 把音轨剥出来ffmpeg -i lecture.mp4 -vn -acodec pcm_s16le -ar 16000 -ac 1 lecture.wav参数拆解-vn 丢掉视频流只处理音频-acodec pcm_s16le 是 PCM 16bit 编码wav 兼容性最好-ar 16000 把采样率重采样到 16k识别引擎主要在 8k/16k 上训练48k 原声不会提升效果只会变大文件-ac 1 转单声道双声道也能转写但声道选择、音量归一都会多出不可控因素进 API 前合并成单声道最省心。3.2 解析 sentencesResult 是字符串要先 json.loads从查询接口拿到的 data[Result] 在多数版本里是 JSON 字符串不是 dict。直接 data[Result][Sentences] 会报 TypeError。正确做法import json result json.loads(data[Result]) sentences result[Sentences] for s in sentences: print(s[Text], s[BeginTime], s[EndTime])常见字段不一定全有Text 是句子文本BeginTime / EndTime 是句子在音频里的起止毫秒ChannelId 是多声道任务的声道编号SpeakerId 在说话人分离开启后才有会议纪要场景按人聚合文本时用。不要按一套字段写死不同版本的识别引擎字段名可能是 Sentences 也可能是 sentences先打印 result.keys() 再写解析。3.3 毫秒转 srt 时间轴格式化函数怎么写srt 协议对时间轴的格式要求是 hh:mm:ss,mmm小时两位、毫秒三位毫秒位用逗号不是点。写一个纯函数def format_srt_time(ms: int) - str: ms max(0, int(ms)) hours ms // 3_600_000 minutes (ms % 3_600_000) // 60_000 seconds (ms % 60_000) // 1000 millis ms % 1000 return f{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}逻辑说明先除后模不会出现小时、分钟借位的进位错误毫秒直接截断而不是四舍五入因为四舍五入到 1000 会让结束时间等于下一秒 000破坏时间轴单调性。字幕播放器对毫秒位要求不严截断完全够用。3.4 按静音间隔合并句子避免字幕刷屏录音文件识别默认按 500ms 静音断句断句比人眼读字幕的节奏快直接一句一字幕会闪屏。常见做法是按句间空隙合并空隙小于阈值的句子合成一个字幕块间隙超过阈值的另起一块。GAP_MS 800 # 句间空隙超过 800ms 才拆成两条字幕 blocks [] current [] for s in sentences: if not current: current [s] continue if s[BeginTime] - current[-1][EndTime] GAP_MS: current.append(s) else: blocks.append(current) current [s] if current: blocks.append(current)然后写文件def write_srt(blocks, out_pathoutput.srt): lines [] for idx, block in enumerate(blocks, start1): start block[0][BeginTime] end block[-1][EndTime] text .join(x[Text] for x in block).strip() lines.append(str(idx)) lines.append(f{format_srt_time(start)} -- {format_srt_time(end)}) lines.append(text) lines.append() with open(out_path, w, encodingutf-8) as f: f.write(\n.join(lines)) write_srt(blocks)文本拼接用空字符串直接 join因为多数识别结果自带标点如果模型不带标点句间要自己补常见做法是判断上一句结尾字符不是标点就补逗号。字幕文件用 UTF-8 写老播放器乱码时把编码改成 utf-8-sig 即可。4. 长音频与多说话人的 4 个必调参数从 enable_words 到 400 排错4.1 4 个参数enable_words、speaker_diarization、max_sentence_silence、vocabulary_id提交任务的 JSON 里除了 appkey、token、file_link还有几个参数对字幕质量影响很大参数默认值作用推荐场景enable_wordsfalse返回词级时间戳逐字字幕、歌词滚动才需要max_sentence_silence500断句静音阈值单位 ms字幕合并想少就用 700 到 1000speaker_diarization-1-1 或 0 不分离1 句子级2 词级会议、客服通话录音设 1vocabulary_id无热词表 ID提升人名和专有名词识别错词反复出现时创建speaker_diarization 值得多说一句它只影响返回的 SpeakerId 字段文本本身不变开词级分离2会明显增加计算时间对话字幕一般句子级1就够。vocabulary_id 在控制台热词表里配置传表 ID注意热词生效有缓存延迟改完词表要等一两分钟再提任务。4.2 提交和查询阶段常见的 400 错误录音文件识别报错经常是 HTTP 200、body 里带错误码不要只盯状态码InvalidAppKeyAppKey 不存在或没开通录音文件识别能力去控制台核对该项目是否开通服务。InvalidTimeStamp.Expiredtoken 过期或本地时钟偏差换新 token容器里时钟漂移也会触发先 date 看一眼。400 content exists risk / ContentRiskDetected内容或文件名触发安全策略把文件名改成中性描述音频内容也检查一遍。TaskNotFoundExceptiontask_id 不存在或查询结果已过保留期说明重试窗口太长需要重新提交。排错第一步是打印完整返回def _check(data: dict) - dict: code data.get(Code) or data.get(StatusCode) if code not in (None, SUCCESS): raise RuntimeError(f{code}: {data.get(Message) or data.get(Msg)}) return data注意错误码字段名同样存在版本差异Code / StatusCode / ErrCode 都可能出现用打印而非猜。4.3 轮询频率与重试指数退避怎么写提交后的轮询建议 1 到 2 秒一次不要 0.2 秒疯狂扫。识别几分钟的任务每两秒查一次足够网络抖动时加指数退避import time def wait_result(token: str, task_id: str, timeout1800): deadline time.time() timeout interval 2 while time.time() deadline: data query_task(token, task_id) status data.get(Status) or data.get(status) if status in (SUCCESS, 2): return data if status in (FAILED, 3): raise RuntimeError(data.get(StatusText, data)) time.sleep(interval) interval min(interval * 1.5, 10) # 退避上限 10 秒 raise TimeoutError(ftask {task_id} timeout)逻辑说明每次等待时间翻 1.5 倍避免服务端把高频轮询当异常流量限流。注意这个退避只针对轮询任务本身失败或超时后重新提交task_id 是新的别拿旧 task_id 无限重试。5. 批量转写的断点续转用本地缓存避免重复计费5.1 用本地 JSON 缓存 task_id 和结果批量转写大量音频时最怕两件事跑到一半断网重启后又从第一个文件重新提交已经识别完的又重新计费。录音文件识别按次计费一个文件提交两次就是两份账单。我的做法是把 task_id 和最终 srt 路径写进本地 JSON作为断点续转的依据import json import pathlib CACHE_FILE trans_cache.json def load_cache(): if pathlib.Path(CACHE_FILE).exists(): return json.loads(pathlib.Path(CACHE_FILE).read_text(encodingutf-8)) return {} def save_cache(cache): pathlib.Path(CACHE_FILE).write_text( json.dumps(cache, ensure_asciiFalse, indent2), encodingutf-8 )主循环里先查缓存再提交cache load_cache() for wav in wav_files: srt_path wav.with_suffix(.srt) if str(wav) in cache and srt_path.exists(): continue # 已转写完成跳过 signed_url bucket.sign_url(GET, str(wav), 3600) task_id submit_task(token, signed_url) cache[str(wav)] {task_id: task_id, status: submitted} save_cache(cache)提交后崩了重启时按 task_id 继续查询就行不需要重新 submit任务失败时删掉缓存项再重新提交避免死循环。每次写入都落盘而不是最后统一写防止中途异常丢缓存。5.2 用 ffprobe 校验时长和字幕有没有丢尾句批量流程跑完用 ffprobe 拿音频时长和 srt 最后一条字幕的 EndTime 比对ffprobe -v error -show_entries formatduration -of csvp0 lecture.wav音频 3600 秒字幕结尾只有 3000 秒说明尾段识别失败或任务被截断。先切段重新识别切段时要带 500ms 前后重叠按重叠区文本对齐避免切点切掉半个词。时长对不上时问题通常出在转码时的采样率或声道参数把 ffmpeg 参数回到 16k 单声道 PCM 再试一次大部分情况能解决。本文还有配套的精品资源点击获取