diff --git a/backend/app/transcriber/bcut.py b/backend/app/transcriber/bcut.py index 8859291..23c1261 100644 --- a/backend/app/transcriber/bcut.py +++ b/backend/app/transcriber/bcut.py @@ -1,5 +1,6 @@ import json import logging +import threading import time from typing import Optional, List, Dict, Union @@ -40,6 +41,10 @@ class BcutTranscriber(Transcriber): self.session = requests.Session() self.task_id = None self.__etags = [] + # 实例被 transcriber_provider 缓存复用,并发任务会交错读写上传会话状态 + # (etags/upload_id/task_id),必须整段串行化 + # ponytail: 全局实例锁,并发转写会排队;如需吞吐量应改为每任务独立实例 + self._lock = threading.Lock() self.__in_boss_key: Optional[str] = None self.__resource_id: Optional[str] = None @@ -59,6 +64,15 @@ class BcutTranscriber(Transcriber): def _upload(self, file_path: str) -> None: """申请上传""" + # 实例被 transcriber_provider 缓存复用,必须清掉上一次上传的会话状态, + # 否则 __etags 跨上传残留,提交的 etag 数与本次分片数不符,B 站会拒绝合并 + self.__etags = [] + self.__in_boss_key = None + self.__resource_id = None + self.__upload_id = None + self.__upload_urls = [] + self.__download_url = None + file_binary = self._load_file(file_path) if not file_binary: raise ValueError("无法读取文件数据") @@ -169,6 +183,10 @@ class BcutTranscriber(Transcriber): @timeit def transcript(self, file_path: str) -> TranscriptResult: """执行识别过程,符合 Transcriber 接口""" + with self._lock: + return self._transcript_locked(file_path) + + def _transcript_locked(self, file_path: str) -> TranscriptResult: try: logger.info(f"开始处理文件: {file_path}") diff --git a/backend/tests/test_bcut_state_reset.py b/backend/tests/test_bcut_state_reset.py new file mode 100644 index 0000000..0ee92bd --- /dev/null +++ b/backend/tests/test_bcut_state_reset.py @@ -0,0 +1,74 @@ +import json + +from app.transcriber import bcut as bcut_module +from app.transcriber.bcut import BcutTranscriber + + +class _FakeResp: + def __init__(self, payload=None, etag=""): + self._payload = payload or {} + self.headers = {"Etag": etag} + self.url = "" + + def raise_for_status(self): + pass + + def json(self): + return self._payload + + +class _FakeSession: + """模拟必剪上传三步接口:申请上传 / PUT 分片 / 提交合并""" + + def __init__(self, per_size, upload_urls_count): + self.per_size = per_size + self.upload_urls_count = upload_urls_count + self.commit_payloads = [] + + def post(self, url, data=None, headers=None): + if url == bcut_module.API_REQ_UPLOAD: + body = json.loads(data) + return _FakeResp({ + "data": { + "in_boss_key": "boss", + "resource_id": "res", + "upload_id": "up", + "upload_urls": [f"http://fake/{i}" for i in range(self.upload_urls_count)], + "per_size": self.per_size, + "size": body["size"], + } + }) + if url == bcut_module.API_COMMIT_UPLOAD: + self.commit_payloads.append(json.loads(data)) + return _FakeResp({"code": 0, "data": {"download_url": "http://fake/dl"}}) + raise AssertionError(f"unexpected post: {url}") + + def put(self, url, data=None, headers=None): + return _FakeResp(etag=f"etag-{url[-1]}") + + +def _upload_file(transcriber, tmp_path, name, content, chunks): + f = tmp_path / name + f.write_bytes(content) + transcriber.session = _FakeSession(per_size=5, upload_urls_count=chunks) + transcriber._upload(str(f)) + return transcriber.session.commit_payloads + + +def test_second_upload_commits_only_its_own_etags(tmp_path): + t = BcutTranscriber() + + # 第一次:3 字节,1 分片 + commits = _upload_file(t, tmp_path, "a.mp3", b"aaa", chunks=1) + assert len(commits[0]["Etags"].split(",")) == 1 + + # 第二次(同一实例):8 字节,2 分片——修复前会提交 3 个 etag + commits = _upload_file(t, tmp_path, "b.mp3", b"aaaaaaaa", chunks=2) + assert len(commits[0]["Etags"].split(",")) == 2 + + +def test_first_upload_unchanged(tmp_path): + t = BcutTranscriber() + commits = _upload_file(t, tmp_path, "a.mp3", b"aaa", chunks=1) + assert commits[0]["Etags"] == "etag-0" + assert commits[0]["UploadId"] == "up"