From 944985fc946d41db8c6c20b026376ca218fb417a Mon Sep 17 00:00:00 2001
From: pumpkinperson996 <david996@bu.edu>
Date: Thu, 9 Jul 2026 15:44:47 -0500
Subject: [PATCH] =?UTF-8?q?fix:=20=E5=BF=85=E5=89=AA=E8=BD=AC=E5=86=99?=
 =?UTF-8?q?=E5=AE=9E=E4=BE=8B=E5=A4=8D=E7=94=A8=E5=AF=BC=E8=87=B4=E7=9A=84?=
 =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E7=8A=B6=E6=80=81=E6=AE=8B=E7=95=99=E4=B8=8E?=
 =?UTF-8?q?=E5=B9=B6=E5=8F=91=E7=AB=9E=E4=BA=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

BcutTranscriber 被 transcriber_provider 缓存复用，但 __etags 等上传
会话状态只在 __init__ 清空、每次上传只追加，导致容器启动后第二次
及以后的转写提交的 etag 数与分片数不符，B 站返回"第三方服务异常"。
并发提交多个视频时，多个后台任务还会在同一实例上交错上传，etag
互相混入。

- _upload() 开头重置全部上传会话状态，修串行残留
- transcript() 加实例锁，整个转写会话串行执行，修并发交错

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 backend/app/transcriber/bcut.py        | 18 +++++++
 backend/tests/test_bcut_state_reset.py | 74 ++++++++++++++++++++++++++
 2 files changed, 92 insertions(+)
 create mode 100644 backend/tests/test_bcut_state_reset.py

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"