File size: 3,727 Bytes
bc7bdb7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import os
import shutil
import ssl
import logging
import httpx
from gradio_client import Client, handle_file
import config

ssl._create_default_https_context = ssl._create_unverified_context
os.environ['PYTHONHTTPSVERIFY'] = '0'
os.environ['CURL_CA_BUNDLE'] = ''
os.environ['REQUESTS_CA_BUNDLE'] = ''

# 💡 猴子補丁 (Monkey Patch) httpx 客戶端
try:
    _orig_httpx_init = httpx.Client.__init__
    def _patched_httpx_init(self, *args, **kwargs):
        kwargs['verify'] = False
        _orig_httpx_init(self, *args, **kwargs)
    httpx.Client.__init__ = _patched_httpx_init

    _orig_httpx_async_init = httpx.AsyncClient.__init__
    def _patched_httpx_async_init(self, *args, **kwargs):
        kwargs['verify'] = False
        _orig_httpx_async_init(self, *args, **kwargs)
    httpx.AsyncClient.__init__ = _patched_httpx_async_init
except Exception as e:
    pass

logger = logging.getLogger(__name__)

asr_client = None
tts_client = None

def get_asr_client():
    global asr_client
    if asr_client is None:
        try:
            asr_client = Client("https://ai-labs.ilrdf.org.tw/sapolita-kaldi/")
        except Exception as e:
            logger.error(f"ASR 引擎初始化失敗: {e}")
            asr_client = None
    return asr_client

def get_tts_client():
    global tts_client
    if tts_client is None:
        try:
            tts_client = Client("https://ai-labs.ilrdf.org.tw/hnang-kari-ai-asi-sluhay/")
        except Exception as e:
            logger.error(f"TTS 引擎初始化失敗: {e}")
            tts_client = None
    return tts_client

def get_clean_value(res):
    """資料清洗器:確保從 API 拿回來的結果是純文字"""
    if isinstance(res, dict) and 'value' in res: 
        return res['value']
    if isinstance(res, list) and len(res) > 0: 
        return res[0]
    return res

def speech_to_text(audio_path, tribe_name):
    """
    耳朵模組:將音檔轉為文字 (ASR)
    供語音訊息與影片音軌辨識使用
    """
    client_inst = get_asr_client()
    if not client_inst:
        logger.error("ASR 服務未就緒")
        return None

    # 💡 從 config.py 的 TRIBE_MAP 取得該族語的 ASR 代碼 (如 formosan_ami)
    asr_code = config.TRIBE_MAP.get(tribe_name, {}).get("asr_code", "formosan_ami")
    
    try:
        # 呼叫原語會辨識 API
        result_raw = client_inst.predict(
            dialect_id=asr_code, 
            audio_data=handle_file(audio_path), 
            api_name="/automatic_speech_recognition"
        )
        return get_clean_value(result_raw)
    except Exception as e:
        logger.error(f"ASR 辨識失敗: {e}")
        return None

def text_to_speech(text, tribe_name, filename):
    """
    嘴巴模組:將文字轉為音檔 (TTS)
    """
    client_inst = get_tts_client()
    if not client_inst:
        logger.error("TTS 服務未就緒")
        return None

    os.makedirs("static", exist_ok=True)
    save_path = f"static/{filename}.wav"
    
    try:
        # 1. 取得對應的發音人代碼
        speaker = get_clean_value(client_inst.predict(ethnicity=tribe_name, api_name="/lambda"))
        
        # 2. 如果是阿美語,強制指定特定的女聲 (維持 3.0 傳統)
        if tribe_name == "阿美": 
            speaker = "阿美_秀姑巒_女聲1"
        
        # 3. 執行合成
        temp_file = client_inst.predict(
            ref=speaker, 
            gen_text_input=text, 
            api_name="/default_speaker_tts"
        )
        
        # 4. 將暫存檔搬移到 static 資料夾
        shutil.move(temp_file, save_path)
        return save_path
    except Exception as e:
        logger.error(f"TTS 合成失敗: {e}")
        return None