model_script/script/tts_test.py

from __future__ import print_function

from typing import List, Dict, Any

import pandas as pd
import requests
import volcenginesdkcore
import volcenginesdkspeechsaasprod
import volcenginesdkspeechsaasprod20250521
from volcenginesdkcore.rest import ApiException

from helper.MySQLHelper import MySQLHelper

mysql_helper = MySQLHelper(
    host="rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com",
    username="readonly",
    password="HdkZ4TDmeK6SQ3BRtJBk",
    database="aigc-admin-prod"
)

ak = "AKLTZWIxNWRkMzUyYjBmNGU2Yjk5MTFiYWVmNmNiY2Q1Njg"
sk = "WW1NM1l6TTJNRFZrT0dFMk5HSXhZamt5TnpFd1kyWTNPR0V6TURZd056Yw=="

configuration = volcenginesdkcore.Configuration()
configuration.ak = ak
configuration.sk = sk
configuration.region = "cn-beijing"
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)


def add_speaker(audio_url: str, speaker: str):
    url = "https://aigc-api.aiddit.com/aigc/resources/aiAccount/saveTts"

    payload = {
        "params": {
            "model": 34,
            "ttsName": speaker,
            "trainAudioUrl": audio_url
        },
        "baseInfo": {
            "token": "80ce2034892c4428ab5b6e39ec0a9e2d",
        }
    }

    response = requests.post(url, headers={}, json=payload)
    response.raise_for_status()  # 如果状态码不是 2xx，抛出异常
    return response.json()


def speaker_insert_db(speaker_id: str):
    sql = f'INSERT INTO `aigc-admin-prod`.volcengine_tts_speaker (speaker_id, type, status, create_time, update_time) VALUES ("{speaker_id}", 2, 0, "2026-05-13 21:08:00", "2026-05-13 21:08:00");'
    print(sql)


def get_all_speakers() -> List[Dict[str, Any]]:
    api_instance = volcenginesdkspeechsaasprod20250521.SPEECHSAASPROD20250521Api()
    batch_list_mega_tts_train_status_request = volcenginesdkspeechsaasprod20250521.BatchListMegaTTSTrainStatusRequest(
        state='Success',
        page_size=100,
        page_number=1,
        project_name="aiddit",
    )

    try:
        # 复制代码运行示例，请自行打印API返回值。
        response = api_instance.batch_list_mega_tts_train_status(batch_list_mega_tts_train_status_request)
        result = []
        for item in response.statuses:
            result.append({
                "speaker_id": item.speaker_id,
                "alias": item.alias,
                "instance_no": item.instance_no,
            })
        return result
    except ApiException as e:
        print("Exception when calling api: %s\n" % e)
        return []


def update_tts_alias(tts_id: str, alias: str):
    # use global default configuration
    api_instance = volcenginesdkspeechsaasprod.SPEECHSAASPRODApi()
    alias_resource_pack_request = volcenginesdkspeechsaasprod.AliasResourcePackRequest(
        alias=alias,
        instance_number="",
        project_name="aiddit",
        train_id=tts_id,
    )

    try:
        response = api_instance.alias_resource_pack(alias_resource_pack_request)
        # response.raise_for_status()  # 如果状态码不是 2xx，抛出异常
        return {}
    except ApiException as e:
        print("Exception when calling api: %s\n" % e)
        return {}


def get_volc_engine_tts_info(tts_name: str, mode: int) -> Dict[str, Any]:
    sql = f"select * from ai_model_tts where speaker = '{tts_name}' and model = {mode};"
    results = mysql_helper.execute_query(sql)
    if results:
        return results[0]
    else:
        return {}


def get_tts_info(tts_id: str) -> Dict[str, Any]:
    sql = f"select * from ai_model_tts where id = '{tts_id}'"
    results = mysql_helper.execute_query(sql)
    return results[0]


def read_tts_id() -> List[str]:
    df = pd.read_csv("/Users/zhao/Desktop/fish_tts.csv")
    return df['tts_id'].tolist()


def main():
    speaker_id_alias_map = {}
    for item in get_all_speakers():
        speaker_id = item['speaker_id']
        alias = item['alias']
        speaker_id_alias_map[speaker_id] = alias

    df = pd.read_csv("/Users/zhao/Desktop/aigc_admin_prod_ai_model_tts.csv")
    dict_list = df.to_dict(orient='records')
    for item in dict_list:
        speaker_id = item['speaker_id']
        alias = item['speaker']
        if speaker_id not in speaker_id_alias_map:
            print(f"{speaker_id} not in speaker_id_alias_map")
            continue
        volc_alias = speaker_id_alias_map[speaker_id]
        if volc_alias == alias:
            print(f"{volc_alias} == {alias}")
            continue
        print(f'更新 {speaker_id} 的别名为 {alias}')
        response = update_tts_alias(speaker_id, alias)
        print(f'{speaker_id} -> {alias} -> {response}')


if __name__ == '__main__':
    main()