backend/app/transcriber/bcut.py

import json
import logging
import time
from typing import Optional, List, Dict, Union

import requests

from app.decorators.timeit import timeit
from app.models.transcriber_model import TranscriptSegment, TranscriptResult
from app.transcriber.base import Transcriber
from app.utils.logger import get_logger
from events import transcription_finished

__version__ = "0.0.3"

API_BASE_URL = "https://member.bilibili.com/x/bcut/rubick-interface"

# 申请上传
API_REQ_UPLOAD = API_BASE_URL + "/resource/create"

# 提交上传
API_COMMIT_UPLOAD = API_BASE_URL + "/resource/create/complete"

# 创建任务
API_CREATE_TASK = API_BASE_URL + "/task"

# 查询结果
API_QUERY_RESULT = API_BASE_URL + "/task/result"

logger = get_logger(__name__)


def _bilibili_cookie() -> Optional[str]:
    """读取「下载配置」里保存的 B 站 Cookie（没有则返回 None）。"""
    try:
        from app.services.cookie_manager import CookieConfigManager
        return CookieConfigManager().get("bilibili")
    except Exception:
        return None


def _with_cookie_hint(msg: str) -> str:
    """bcut 是 B 站的接口，未带 B 站 Cookie 时容易被风控拒绝（如「第三方服务异常」）。

    未配置 Cookie 时在报错后面追加可行动的提示；已配置则原样返回。
    """
    if "下载配置" in msg:  # 已带提示，保持幂等
        return msg
    if _bilibili_cookie():
        return msg
    return (
        f"{msg}。bcut（必剪）转写走的是 B 站接口，未配置 B 站 Cookie 时容易被风控拒绝："
        "请在「设置 → 下载配置」中填写 B 站 Cookie 后重试，"
        "或在「设置 → 音频转写配置」中切换为本地转写引擎（fast-whisper / mlx-whisper）。"
    )


class BcutTranscriber(Transcriber):
    """必剪 语音识别接口"""
    headers = {
        'User-Agent': 'Bilibili/1.0.0 (https://www.bilibili.com)',
        'Content-Type': 'application/json'
    }

    def __init__(self):
        self.session = requests.Session()
        # 带上「下载配置」里的 B 站 Cookie（如有），降低被 B 站风控拒绝的概率
        cookie = _bilibili_cookie()
        if cookie:
            self.headers = {**self.headers, 'Cookie': cookie}
        self.task_id = None
        self.__etags = []

        self.__in_boss_key: Optional[str] = None
        self.__resource_id: Optional[str] = None
        self.__upload_id: Optional[str] = None
        self.__upload_urls: List[str] = []
        self.__per_size: Optional[int] = None
        self.__clips: Optional[int] = None

        self.__etags: List[str] = []
        self.__download_url: Optional[str] = None
        self.task_id: Optional[str] = None
        
    def _load_file(self, file_path: str) -> bytes:
        """读取文件内容"""
        with open(file_path, 'rb') as f:
            return f.read()

    def _upload(self, file_path: str) -> None:
        """申请上传"""
        file_binary = self._load_file(file_path)
        if not file_binary:
            raise ValueError("无法读取文件数据")
            
        payload = json.dumps({
            "type": 2,
            "name": "audio.mp3",
            "size": len(file_binary),
            "ResourceFileType": "mp3",
            "model_id": "8",
        })

        resp = self.session.post(
            API_REQ_UPLOAD,
            data=payload,
            headers=self.headers
        )
        resp.raise_for_status()
        resp = resp.json()
        resp_data = resp["data"]

        self.__in_boss_key = resp_data["in_boss_key"]
        self.__resource_id = resp_data["resource_id"]
        self.__upload_id = resp_data["upload_id"]
        self.__upload_urls = resp_data["upload_urls"]
        self.__per_size = resp_data["per_size"]
        self.__clips = len(resp_data["upload_urls"])

        logger.info(
            f"申请上传成功, 总计大小{resp_data['size'] // 1024}KB, {self.__clips}分片, 分片大小{resp_data['per_size'] // 1024}KB: {self.__in_boss_key}"
        )
        self.__upload_part(file_binary)
        self.__commit_upload()

    def __upload_part(self, file_binary: bytes) -> None:
        """上传音频数据"""
        for clip in range(self.__clips):
            start_range = clip * self.__per_size
            end_range = min((clip + 1) * self.__per_size, len(file_binary))
            logger.info(f"开始上传分片{clip}: {start_range}-{end_range}")
            resp = self.session.put(
                self.__upload_urls[clip],
                data=file_binary[start_range:end_range],
                headers={'Content-Type': 'application/octet-stream'}
            )
            resp.raise_for_status()
            etag = resp.headers.get("Etag", "").strip('"')
            self.__etags.append(etag)
            logger.info(f"分片{clip}上传成功: {etag}")

    def __commit_upload(self) -> None:
        """提交上传数据"""
        data = json.dumps({
            "InBossKey": self.__in_boss_key,
            "ResourceId": self.__resource_id,
            "Etags": ",".join(self.__etags),
            "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}")

    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

    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"]

    @timeit
    def transcript(self, file_path: str) -> TranscriptResult:
        """执行识别过程，符合 Transcriber 接口"""
        try:
            logger.info(f"开始处理文件: {file_path}")
            
            # 上传文件
            logger.info("正在上传文件...")
            self._upload(file_path)
            
            # 创建任务
            logger.info("提交转录任务...")
            self._create_task()
            
            # 轮询检查任务状态
            logger.info("等待转录结果...")
            task_resp = None
            max_retries = 500
            for i in range(max_retries):
                task_resp = self._query_result()
                
                if task_resp["state"] == 4:  # 完成状态
                    break
                elif task_resp["state"] == 3:  # 失败状态
                    error_msg = f"B站ASR任务失败，状态码: {task_resp['state']}"
                    logger.error(error_msg)
                    raise Exception(error_msg)
                    
                # 每隔一段时间打印进度
                if i % 10 == 0:
                    logger.info(f"转录进行中... {i}/{max_retries}")
                    
                time.sleep(1)
                
            if not task_resp or task_resp["state"] != 4:
                error_msg = f"B站ASR任务未能完成，状态: {task_resp.get('state') if task_resp else 'Unknown'}"
                logger.error(error_msg)
                raise Exception(error_msg)
                
            # 解析结果
            logger.info("转录成功，处理结果...")
            result_json = json.loads(task_resp["result"])
            
            # 提取分段数据
            segments = []
            full_text = ""
            
            for u in result_json.get("utterances", []):
                text = u.get("transcript", "").strip()
                # B站ASR返回的时间戳是毫秒，需要转换为秒
                start_time = float(u.get("start_time", 0)) / 1000.0
                end_time = float(u.get("end_time", 0)) / 1000.0
                
                full_text += text + " "
                segments.append(TranscriptSegment(
                    start=start_time,
                    end=end_time,
                    text=text
                ))
            
            # 创建结果对象
            result = TranscriptResult(
                language=result_json.get("language", "zh"),
                full_text=full_text.strip(),
                segments=segments,
                raw=result_json
            )
            
            # 触发完成事件
            # self.on_finish(file_path, result)
            
            return result
            
        except Exception as e:
            logger.error(f"B站ASR处理失败: {str(e)}")
            # 未配置 B 站 Cookie 时附加「去下载配置填 Cookie / 换本地转写引擎」的提示
            raise Exception(_with_cookie_hint(str(e))) from e

    def on_finish(self, video_path: str, result: TranscriptResult) -> None:
        """转录完成的回调"""
        logger.info(f"B站ASR转写完成: {video_path}")
        transcription_finished.send({
            "file_path": video_path,
        })