fix: 必剪转写实例复用导致的上传状态残留与并发竞争

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

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

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
pumpkinperson996
2026-07-09 15:44:47 -05:00
co-authored by Claude Fable 5
parent 6d67e5a76a
commit 944985fc94
2 changed files with 92 additions and 0 deletions
+18
View File
@@ -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}")
+74
View File
@@ -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"