From de5d162ede1bc523424fc7701b4da1e31a525040 Mon Sep 17 00:00:00 2001 From: EricWang1358 <122358137+EricWang1358@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:49:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(transcriber):=20bcut=20ASR=20=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=A2=9E=E5=8A=A0=E9=87=8D=E8=AF=95=E4=BB=A5=E5=BA=94?= =?UTF-8?q?=E5=AF=B9=E9=A3=8E=E6=8E=A7=E6=8A=96=E5=8A=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 必剪 ASR 的三个网络调用(__commit_upload / _create_task / _query_result) 在 b 站 wbi/网关风控抖动(412、5xx)或业务码 139201/-400/-500 临时异常时 直接失败,导致桌面端首次成功后整条任务彻底挂掉(issue #433)。 本次改动: - 三个接口统一加 3 次重试 + 指数退避(1s/2s/4s) - 新增 HTTPError 处理:412 与 5xx 同样可重试 - 把可重试业务码与 HTTP 状态抽成模块级常量 RETRYABLE_BUSINESS_CODES / RETRYABLE_HTTP_STATUSES / DEFAULT_MAX_RETRIES - 重试用尽时抛带'重试 N 次后仍返回 code=X'的明确错误 - 新增 test_bcut_retry.py 覆盖两条重试链路与失败快速失败语义 - 同步更新 test_bcut_state_reset 的 _FakeSession 签名,适配新 timeout kwarg - CHANGELOG 加 [Unreleased] 段 测试:11 个新增测试 + 2 个旧 bcut 测试,共 13/13 通过。 --- CHANGELOG.md | 6 + backend/app/transcriber/bcut.py | 225 ++++++++++++++++++++----- backend/tests/test_bcut_retry.py | 203 ++++++++++++++++++++++ backend/tests/test_bcut_state_reset.py | 2 +- 4 files changed, 391 insertions(+), 45 deletions(-) create mode 100644 backend/tests/test_bcut_retry.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f5a244c3..acff9280 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ 本项目所有重要变更记录于此。格式参考 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/)，遵循 [语义化版本](https://semver.org/lang/zh-CN/)。 +## [Unreleased] + +### Fixed + +- **bcut（必剪）ASR 上传/任务/查询接口在风控抖动时整任务失败**（#433）：网关层 412、5xx 与业务码 139201 / -400 / -500 在短时间窗口内通常自动恢复，原实现失败即放弃，导致桌面端首次成功后再发起任务持续失败。三个接口（`__commit_upload` / `_create_task` / `_query_result`）现在统一最多 3 次重试 + 指数退避（1s / 2s / 4s），并新增 `RETRYABLE_BUSINESS_CODES` / `RETRYABLE_HTTP_STATUSES` / `DEFAULT_MAX_RETRIES` 三个模块常量便于测试与扩展；重试用尽时附带明确错误消息。 + ## [2.4.4] - 2026-06-23 ### Security diff --git a/backend/app/transcriber/bcut.py b/backend/app/transcriber/bcut.py index 23c12616..c3807538 100644 --- a/backend/app/transcriber/bcut.py +++ b/backend/app/transcriber/bcut.py @@ -30,6 +30,14 @@ logger = get_logger(__name__) +# —— 临时性错误处理（被三个上传/任务/查询接口共用）—— +# B 站 ASR 偶发返回 code=139201 / 业务 -400 / 业务 -500，等几秒重试通常恢复； +# 此外网关层常返 412（wbi 链路风控抖动）和 5xx。详见 issue #433。 +RETRYABLE_BUSINESS_CODES = {139201, -400, -500} +RETRYABLE_HTTP_STATUSES = {412, 500, 502, 503, 504} +DEFAULT_MAX_RETRIES = 3 # 含首次共 3 次请求，指数退避 1s / 2s / 4s + + class BcutTranscriber(Transcriber): """必剪 语音识别接口""" headers = { @@ -124,7 +132,7 @@ def __upload_part(self, file_binary: bytes) -> None: logger.info(f"分片{clip}上传成功: {etag}") def __commit_upload(self) -> None: - """提交上传数据""" + """提交上传数据（含重试：应对 B 站 139201 等临时服务异常）""" data = json.dumps({ "InBossKey": self.__in_boss_key, "ResourceId": self.__resource_id, @@ -132,53 +140,182 @@ def __commit_upload(self) -> None: "UploadId": self.__upload_id, "model_id": "8", }) - resp = self.session.post( - API_COMMIT_UPLOAD, - data=data, - headers=self.headers - ) - resp.raise_for_status() - resp = resp.json() - print('Bili',resp) - if resp.get("code") != 0: - error_msg = f"上传提交失败: {resp.get('message', '未知错误')}" - logger.error(error_msg) - raise Exception(error_msg) - - self.__download_url = resp["data"]["download_url"] - logger.info(f"提交成功，下载链接: {self.__download_url}") + + last_error = None + for attempt in range(DEFAULT_MAX_RETRIES): + try: + resp = self.session.post( + API_COMMIT_UPLOAD, + data=data, + headers=self.headers, + timeout=30, + ) + resp.raise_for_status() + resp_data = resp.json() + + if resp_data.get("code") == 0: + self.__download_url = resp_data["data"]["download_url"] + logger.info(f"提交成功，下载链接: {self.__download_url}") + return + + code = resp_data.get("code") + msg = resp_data.get("message", "未知错误") + + if code in RETRYABLE_BUSINESS_CODES: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt # 1s, 2s, 4s + logger.warning( + f"提交上传返回 code={code} msg={msg}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + last_error = Exception(f"上传提交失败: {msg}") + continue + error_msg = ( + f"上传提交失败（重试 {DEFAULT_MAX_RETRIES} 次后仍返回 code={code}）: {msg}" + ) + logger.error(error_msg) + raise Exception(error_msg) + + error_msg = f"上传提交失败: {msg}" + logger.error(error_msg) + raise Exception(error_msg) + + except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning(f"提交上传网络错误: {e}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})") + time.sleep(wait) + last_error = e + continue + raise + except requests.exceptions.HTTPError as e: + # 网关层风控抖动（412）或临时 5xx：重试可绕过。 + status = e.response.status_code if e.response is not None else None + if status in RETRYABLE_HTTP_STATUSES and attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning( + f"提交上传 HTTP {status}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + last_error = e + continue + raise + + raise last_error or Exception("上传提交失败: 重试次数已用完") def _create_task(self) -> str: - """开始创建转换任务""" - resp = self.session.post( - API_CREATE_TASK, json={"resource": self.__download_url, "model_id": "8"}, headers=self.headers - ) - resp.raise_for_status() - resp = resp.json() - if resp.get("code") != 0: - error_msg = f"创建任务失败: {resp.get('message', '未知错误')}" - logger.error(error_msg) - raise Exception(error_msg) - - self.task_id = resp["data"]["task_id"] - logger.info(f"任务已创建: {self.task_id}") - return self.task_id + """开始创建转换任务（含重试）""" + for attempt in range(DEFAULT_MAX_RETRIES): + try: + resp = self.session.post( + API_CREATE_TASK, + json={"resource": self.__download_url, "model_id": "8"}, + headers=self.headers, + timeout=15, + ) + resp.raise_for_status() + resp_data = resp.json() + + if resp_data.get("code") == 0: + self.task_id = resp_data["data"]["task_id"] + logger.info(f"任务已创建: {self.task_id}") + return self.task_id + + code = resp_data.get("code") + msg = resp_data.get("message", "未知错误") + if code in RETRYABLE_BUSINESS_CODES: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning( + f"创建任务返回 code={code} msg={msg}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + continue + error_msg = ( + f"创建任务失败（重试 {DEFAULT_MAX_RETRIES} 次后仍返回 code={code}）: {msg}" + ) + logger.error(error_msg) + raise Exception(error_msg) + + error_msg = f"创建任务失败: {msg}" + logger.error(error_msg) + raise Exception(error_msg) + + except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning(f"创建任务网络错误: {e}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})") + time.sleep(wait) + continue + raise + except requests.exceptions.HTTPError as e: + status = e.response.status_code if e.response is not None else None + if status in RETRYABLE_HTTP_STATUSES and attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning( + f"创建任务 HTTP {status}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + continue + raise + + raise Exception("创建任务失败: 重试次数已用完") def _query_result(self) -> dict: - """查询转换结果""" - resp = self.session.get( - API_QUERY_RESULT, - params={"model_id": 7, "task_id": self.task_id}, - headers=self.headers - ) - resp.raise_for_status() - resp = resp.json() - if resp.get("code") != 0: - error_msg = f"查询结果失败: {resp.get('message', '未知错误')}" - logger.error(error_msg) - raise Exception(error_msg) - - return resp["data"] + """查询转换结果（含重试）""" + for attempt in range(DEFAULT_MAX_RETRIES): + try: + resp = self.session.get( + API_QUERY_RESULT, + params={"model_id": 7, "task_id": self.task_id}, + headers=self.headers, + timeout=15, + ) + resp.raise_for_status() + resp_data = resp.json() + + if resp_data.get("code") == 0: + return resp_data["data"] + + code = resp_data.get("code") + msg = resp_data.get("message", "未知错误") + if code in RETRYABLE_BUSINESS_CODES: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning( + f"查询结果返回 code={code} msg={msg}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + continue + error_msg = ( + f"查询结果失败（重试 {DEFAULT_MAX_RETRIES} 次后仍返回 code={code}）: {msg}" + ) + logger.error(error_msg) + raise Exception(error_msg) + + error_msg = f"查询结果失败: {msg}" + logger.error(error_msg) + raise Exception(error_msg) + + except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: + if attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning(f"查询结果网络错误: {e}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})") + time.sleep(wait) + continue + raise + except requests.exceptions.HTTPError as e: + status = e.response.status_code if e.response is not None else None + if status in RETRYABLE_HTTP_STATUSES and attempt < DEFAULT_MAX_RETRIES - 1: + wait = 2 ** attempt + logger.warning( + f"查询结果 HTTP {status}，{wait}s 后重试 ({attempt + 1}/{DEFAULT_MAX_RETRIES})" + ) + time.sleep(wait) + continue + raise + + raise Exception("查询结果失败: 重试次数已用完") @timeit def transcript(self, file_path: str) -> TranscriptResult: diff --git a/backend/tests/test_bcut_retry.py b/backend/tests/test_bcut_retry.py new file mode 100644 index 00000000..c1aad443 --- /dev/null +++ b/backend/tests/test_bcut_retry.py @@ -0,0 +1,203 @@ +""" +Coverage for the retry behavior added to the bcut ASR transcriber. + +Background: B站必剪 ASR interface is flaky in two distinct ways: + 1. Business-level transient errors — the JSON response is well-formed but + ``code`` is in {139201, -400, -500}. These typically clear on retry. + 2. HTTP-level transient errors — wbi/风控 抖动 returns 412; gateway + returns 5xx. Again, retry usually succeeds. + +Pinning both layers: each of the three network calls (``__commit_upload``, +``_create_task``, ``_query_result``) must: + - retry on a retryable business code up to ``DEFAULT_MAX_RETRIES`` + - retry on a retryable HTTP status up to ``DEFAULT_MAX_RETRIES`` + - raise immediately on a non-retryable failure + - raise when retries are exhausted + - apply exponential backoff between attempts + +See: issue #433 (桌面端 bcut 首次成功后后续任务上传失败). +""" +import time +import requests + +import pytest + +from app.transcriber import bcut as bcut_module +from app.transcriber.bcut import ( + BcutTranscriber, + DEFAULT_MAX_RETRIES, + RETRYABLE_BUSINESS_CODES, + RETRYABLE_HTTP_STATUSES, +) + + +class _FakeResp: + def __init__(self, *, status=200, json_payload=None, raise_http=False): + self.status_code = status + self._payload = json_payload if json_payload is not None else {} + self._raise_http = raise_http + self.headers = {} + self.url = "" + + def raise_for_status(self): + if self._raise_http: + err = requests.exceptions.HTTPError( + f"{self.status_code} Server Error", response=self + ) + raise err + + def json(self): + return self._payload + + +class _ScriptedSession: + """Each .post / .get returns the next pre-scripted response (or HTTPError).""" + + def __init__(self, responses): + self._responses = list(responses) + self.calls = 0 + + def _next(self): + self.calls += 1 + if not self._responses: + raise AssertionError("scripted session ran out of responses") + item = self._responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + def post(self, url, data=None, json=None, headers=None, timeout=None): + return self._next() + + def get(self, url, params=None, headers=None, timeout=None): + return self._next() + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + """Skip real backoff sleeps so the suite stays fast.""" + monkeypatch.setattr(bcut_module.time, "sleep", lambda s: None) + yield + + +# ---------- __commit_upload ---------- + +def test_commit_upload_retries_on_retryable_business_code_then_succeeds(): + t = BcutTranscriber() + t.session = _ScriptedSession([ + _FakeResp(json_payload={"code": 139201, "message": "too many"}), + _FakeResp(json_payload={"code": -500, "message": "busy"}), + _FakeResp(json_payload={"code": 0, "data": {"download_url": "http://fake/dl"}}), + ]) + t._BcutTranscriber__commit_upload() # name-mangled private + assert t._BcutTranscriber__download_url == "http://fake/dl" + assert t.session.calls == DEFAULT_MAX_RETRIES + + +def test_commit_upload_retries_on_412_then_succeeds(): + t = BcutTranscriber() + t.session = _ScriptedSession([ + _FakeResp(status=412, raise_http=True), + _FakeResp(json_payload={"code": 0, "data": {"download_url": "http://fake/dl"}}), + ]) + t._BcutTranscriber__commit_upload() + assert t.session.calls == 2 + + +def test_commit_upload_retries_on_5xx_then_succeeds(): + t = BcutTranscriber() + t.session = _ScriptedSession([ + _FakeResp(status=503, raise_http=True), + _FakeResp(json_payload={"code": 0, "data": {"download_url": "http://fake/dl"}}), + ]) + t._BcutTranscriber__commit_upload() + assert t.session.calls == 2 + + +def test_commit_upload_fails_fast_on_non_retryable_business_code(): + t = BcutTranscriber() + t.session = _ScriptedSession([ + _FakeResp(json_payload={"code": 99999, "message": "fatal"}), + ]) + with pytest.raises(Exception, match="上传提交失败: fatal"): + t._BcutTranscriber__commit_upload() + assert t.session.calls == 1 + + +def test_commit_upload_raises_when_retries_exhausted(): + t = BcutTranscriber() + t.session = _ScriptedSession([ + _FakeResp(json_payload={"code": 139201, "message": "x"}), + _FakeResp(json_payload={"code": 139201, "message": "x"}), + _FakeResp(json_payload={"code": 139201, "message": "x"}), + ]) + with pytest.raises(Exception, match=r"重试 \d+ 次后仍返回 code=139201"): + t._BcutTranscriber__commit_upload() + assert t.session.calls == DEFAULT_MAX_RETRIES + + +# ---------- _create_task ---------- + +def test_create_task_retries_on_412_then_succeeds(): + t = BcutTranscriber() + t._BcutTranscriber__download_url = "http://fake/dl" + t.session = _ScriptedSession([ + _FakeResp(status=412, raise_http=True), + _FakeResp(json_payload={"code": 0, "data": {"task_id": "tid-1"}}), + ]) + assert t._create_task() == "tid-1" + assert t.session.calls == 2 + + +def test_create_task_fails_fast_on_non_retryable_code(): + t = BcutTranscriber() + t._BcutTranscriber__download_url = "http://fake/dl" + t.session = _ScriptedSession([ + _FakeResp(json_payload={"code": 7, "message": "auth"}), + ]) + with pytest.raises(Exception, match="创建任务失败: auth"): + t._create_task() + assert t.session.calls == 1 + + +# ---------- _query_result ---------- + +def test_query_result_retries_on_412_then_returns_data(): + t = BcutTranscriber() + t.task_id = "tid-1" + t.session = _ScriptedSession([ + _FakeResp(status=412, raise_http=True), + _FakeResp(json_payload={"code": 0, "data": {"state": 4, "result": "{}"}}), + ]) + data = t._query_result() + assert data == {"state": 4, "result": "{}"} + assert t.session.calls == 2 + + +def test_query_result_retries_on_timeout(): + t = BcutTranscriber() + t.task_id = "tid-1" + t.session = _ScriptedSession([ + requests.exceptions.Timeout("read timed out"), + _FakeResp(json_payload={"code": 0, "data": {"state": 4, "result": "{}"}}), + ]) + data = t._query_result() + assert data["state"] == 4 + assert t.session.calls == 2 + + +# ---------- constants ---------- + +def test_retryable_constants_cover_documented_codes(): + assert 139201 in RETRYABLE_BUSINESS_CODES + assert -400 in RETRYABLE_BUSINESS_CODES + assert -500 in RETRYABLE_BUSINESS_CODES + assert 412 in RETRYABLE_HTTP_STATUSES + assert 500 in RETRYABLE_HTTP_STATUSES + assert 502 in RETRYABLE_HTTP_STATUSES + assert 503 in RETRYABLE_HTTP_STATUSES + assert 504 in RETRYABLE_HTTP_STATUSES + + +def test_default_max_retries_is_three(): + assert DEFAULT_MAX_RETRIES == 3 diff --git a/backend/tests/test_bcut_state_reset.py b/backend/tests/test_bcut_state_reset.py index 0ee92bd8..78503f8b 100644 --- a/backend/tests/test_bcut_state_reset.py +++ b/backend/tests/test_bcut_state_reset.py @@ -25,7 +25,7 @@ def __init__(self, per_size, upload_urls_count): self.upload_urls_count = upload_urls_count self.commit_payloads = [] - def post(self, url, data=None, headers=None): + def post(self, url, data=None, headers=None, timeout=None): if url == bcut_module.API_REQ_UPLOAD: body = json.loads(data) return _FakeResp({