baiyue-logistics-chat/logistics_chat/speech_service.py

"""Bounded, in-memory recording transcription; separate from text chat credentials."""

import base64
import json
import math
import hashlib
import os
from pathlib import Path
import shutil
import subprocess
import threading
import time
from urllib.parse import urlsplit

import requests
from flask import current_app

from .config import get_setting


MAX_AUDIO_BYTES = 5 * 1024 * 1024
MAX_DURATION_SECONDS = 60
UNAVAILABLE = '语音输入暂不可用，请使用文字输入。'


class SpeechFailure(Exception):
    def __init__(self, message, status=502):
        self.message, self.status = message, status


def ffmpeg_binary():
    configured = get_setting('FFMPEG_BIN')
    return str(configured) if configured and Path(configured).is_file() else shutil.which('ffmpeg')


def speech_enabled():
    provider = get_setting('ASR_PROVIDER')
    if provider == 'baidu':
        return bool(get_setting('BAIDU_ASR_API_KEY') and get_setting('BAIDU_ASR_SECRET_KEY') and ffmpeg_binary())
    if provider != 'dashscope':
        return False
    try:
        url = urlsplit(get_setting('ASR_URL'))
    except ValueError:
        return False
    return bool(get_setting('ASR_API_KEY') and get_setting('ASR_MODEL')
                and url.scheme == 'https' and url.hostname and not url.username and not url.password)


def post_json(url, **options):
    response = requests.post(url, timeout=(5, 30), stream=True, allow_redirects=False, **options)
    try:
        if response.status_code != 200:
            raise SpeechFailure('语音转文字失败，请稍后重试。')
        data = bytearray()
        for chunk in response.iter_content(4096):
            data.extend(chunk)
            if len(data) > 64 * 1024:
                raise SpeechFailure('语音识别结果无效，请重试。')
        return json.loads(data)
    finally:
        response.close()


def baidu_token():
    key, secret = get_setting('BAIDU_ASR_API_KEY'), get_setting('BAIDU_ASR_SECRET_KEY')
    identity = hashlib.sha256((key + '\0' + secret).encode()).hexdigest()
    lock = current_app.extensions.setdefault('speech_token_lock', threading.Lock())
    with lock:
        cached = current_app.extensions.get('speech_token')
        if cached and cached['identity'] == identity and cached['expires'] > time.time():
            return cached['token']
        data = post_json('https://aip.baidubce.com/oauth/2.0/token', data={
            'grant_type':'client_credentials', 'client_id':key, 'client_secret':secret})
        token = data.get('access_token')
        if not isinstance(token, str) or not token:
            raise SpeechFailure(UNAVAILABLE, 503)
        current_app.extensions['speech_token'] = {'identity':identity, 'token':token,
            'expires':time.time() + max(0, min(float(data.get('expires_in', 0)), 2592000) - 60)}
        return token


def to_pcm(content):
    """Decode browser WebM/MP4 in pipes to mono 16 kHz, without temporary audio files."""
    binary = ffmpeg_binary()
    if not binary:
        raise SpeechFailure(UNAVAILABLE, 503)
    result = subprocess.run([binary, '-hide_banner', '-loglevel', 'error', '-protocol_whitelist', 'pipe',
        '-i', 'pipe:0', '-map', '0:a:0', '-vn', '-sn', '-dn', '-ac', '1', '-ar', '16000',
        '-t', '61', '-f', 's16le', 'pipe:1'], input=content, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        timeout=10, creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0)
    if result.returncode != 0 or len(result.stdout) < 12800:
        raise SpeechFailure('录音无法识别，请重新录制。', 422)
    if len(result.stdout) > 16000 * 2 * MAX_DURATION_SECONDS:
        raise SpeechFailure('录音超过 60 秒，请缩短后重试。', 400)
    return result.stdout


def baidu_transcribe(content, language):
    pcm = to_pcm(content)
    result = post_json('https://vop.baidu.com/server_api', json={'format':'pcm', 'rate':16000, 'channel':1,
        'cuid':'baiyue-logistics-chat', 'token':baidu_token(), 'dev_pid':1737 if language == 'en' else 1537,
        'speech':base64.b64encode(pcm).decode('ascii'), 'len':len(pcm)})
    code = result.get('err_no')
    if code == 3301:
        raise SpeechFailure('没有识别到文字，请靠近麦克风重试。', 422)
    if code != 0:
        current_app.logger.warning('Baidu ASR request failed: code=%s', code)
        raise SpeechFailure(UNAVAILABLE if code == 3302 else '语音转文字失败，请稍后重试。', 503 if code == 3302 else 502)
    parts = result.get('result')
    if not isinstance(parts, list) or any(not isinstance(part, str) for part in parts):
        raise SpeechFailure('语音识别结果无效，请重试。')
    return ''.join(parts)


def audio_format(content, mime):
    """Check container signatures, never trust the uploaded filename."""
    mime = mime.split(';', 1)[0].lower()
    if mime == 'audio/webm' and content.startswith(b'\x1aE\xdf\xa3'):
        return 'audio/webm'
    if mime in ('audio/mp4', 'audio/x-m4a') and content[4:8] == b'ftyp':
        return 'audio/mp4'
    if mime in ('audio/ogg', 'audio/opus') and content.startswith(b'OggS'):
        return 'audio/ogg'
    if mime in ('audio/wav', 'audio/x-wav') and content[:4] == b'RIFF' and content[8:12] == b'WAVE':
        return 'audio/wav'
    return None


def transcribe_recording(upload, duration, language):
    if not speech_enabled():
        return {'ok': False, 'error': UNAVAILABLE}, 503
    if not upload:
        return {'ok': False, 'error': '没有收到录音，请重新按住说话。'}, 400
    try:
        duration = float(duration)
    except (TypeError, ValueError):
        duration = 0
    if not math.isfinite(duration) or not 0.4 <= duration <= MAX_DURATION_SECONDS + 1:
        return {'ok': False, 'error': '录音时长无效，请录制 1 至 60 秒语音。'}, 400
    content = upload.read(MAX_AUDIO_BYTES + 1)
    if len(content) > MAX_AUDIO_BYTES:
        return {'ok': False, 'error': '录音文件过大，请缩短后重试。'}, 413
    mime = audio_format(content, upload.mimetype)
    if len(content) < 16 or not mime:
        return {'ok': False, 'error': '录音格式不受支持，请使用文字输入或更换浏览器。'}, 415
    language = 'en' if language == 'en' else 'zh'
    try:
        if get_setting('ASR_PROVIDER') == 'baidu':
            text = baidu_transcribe(content, language)
        else:
            body = {'model': get_setting('ASR_MODEL'), 'stream': False,
                    'messages': [{'role': 'user', 'content': [{'type': 'input_audio', 'input_audio': {
                        'data': 'data:' + mime + ';base64,' + base64.b64encode(content).decode('ascii')}}]}],
                    'asr_options': {'language': language, 'enable_itn': True}}
            result = post_json(get_setting('ASR_URL'), json=body,
                               headers={'Authorization': 'Bearer ' + get_setting('ASR_API_KEY')})
            text = result['choices'][0]['message']['content']
        if not isinstance(text, str) or not text.strip():
            return {'ok': False, 'error': '没有识别到文字，请靠近麦克风重试。'}, 422
        text = text.strip()
        if len(text) > 2000:
            return {'ok': False, 'error': '识别文字过长，请缩短录音后重试。'}, 422
        return {'ok': True, 'text': text}, 200
    except requests.Timeout:
        return {'ok': False, 'error': '语音转文字超时，请重试。'}, 504
    except SpeechFailure as error:
        return {'ok': False, 'error': error.message}, error.status
    except subprocess.TimeoutExpired:
        return {'ok': False, 'error': '录音处理超时，请缩短后重试。'}, 504
    except OSError:
        return {'ok': False, 'error': UNAVAILABLE}, 503
    except (requests.RequestException, ValueError, KeyError, IndexError, TypeError):
        return {'ok': False, 'error': '语音转文字失败，请稍后重试。'}, 502