model_script/client/FishClient.py

from io import BytesIO
from typing import Dict, Any, List

import requests


class FishClient(object):
    def __init__(self, base_url: str):
        self.base_url = base_url

    def get_all_references_id(self) -> List[str]:
        """
        获取已存在的参考音频

        Returns:
            参考音频ID列表

        Raises:
            requests.RequestException: 下载或上传请求失败
            ValueError: 服务端返回错误响应或下载内容为空
        """
        url = f"{self.base_url}/v1/references/list"
        headers = self.build_common_header()
        response = requests.get(url, headers=headers)
        return response.json().get("reference_ids", [])

    def add_reference_id_by_url(self, reference_id: str, reference_text: str, audio_url: str):
        """
        从 URL 下载音频并调用添加参考音频的接口

        Args:
            reference_id: 参考音频唯一标识
            audio_url: 音频文件的 URL（支持 http/https）
            reference_text: 音频对应的文本内容

        Raises:
            requests.RequestException: 下载或上传请求失败
            ValueError: 服务端返回错误响应或下载内容为空
        """
        try:
            resp = requests.get(audio_url, timeout=30)
            resp.raise_for_status()  # 检查 HTTP 错误
            audio_content = resp.content
            if not audio_content:
                raise ValueError("从 URL 下载的音频文件为空")
        except requests.exceptions.RequestException as e:
            raise requests.RequestException(f"下载音频失败: {e}") from e

        url = f"{self.base_url}/v1/references/add"
        payload = {
            "id": reference_id,
            "text": reference_text,
        }

        # 定义文件名
        file_name = audio_url.split('/')[-1] or "audio.wav"
        # 使用 BytesIO 包装音频内容
        files = {
            "audio": (file_name, BytesIO(audio_content), "audio/wav"),
        }

        headers = self.build_common_header()

        try:
            response = requests.post(url, data=payload, files=files, headers=headers, timeout=30)
            resp_json = response.json()
        except requests.exceptions.RequestException as e:
            raise requests.RequestException(f"上传请求失败: {e}") from e
        finally:
            files["audio"][1].close()

        if response.status_code != 200:
            raise ValueError(f"服务端返回错误 (HTTP {response.status_code}): {resp_json.get('message', '未知错误')}")

        if not resp_json.get("success", False):
            raise ValueError(f"业务失败: {resp_json.get('message', '未知错误')}")

        return resp_json

    def get_model_info_by_id(self, reference_id: str) -> Dict[str, Any]:
        url = f"{self.base_url}/model/{reference_id}"
        headers = {
            "Authorization": "Bearer 0891f3f93a2640428f9988e267aa57e1",
            "Content-Type": "application/json"
        }
        response = requests.get(url, headers=headers, timeout=(10, 1800))
        return response.json()

    def is_official_api(self):
        return "api.fish.audio" in self.base_url

    @staticmethod
    def build_common_header() -> Dict[str, Any]:
        return {
            "Accept": "application/json",
        }