deepvoice_client/lib/data/services/volcano_asr_service.dart

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:uuid/uuid.dart';
import 'package:get/get.dart';
import 'package:flutter/foundation.dart';
import 'package:record/record.dart';

/// 自定义异常类
class AsrServiceException implements Exception {
  final String message;
  AsrServiceException(this.message);
  
  @override
  String toString() => message;
}

/// Volcano ASR 服务，负责音频采集、数据打包、WebSocket 连接和数据传输
class VolcanoAsrService extends GetxService {
  // 服务地址
  static const String _baseUrl = 'wss://openspeech.bytedance.com/api/v3/sauc/bigmodel';
  
  // 配置参数
  final String appKey;
  final String accessKey;
  
  // WebSocket 与录音相关
  WebSocket? _webSocket;
  AudioRecorder? _audioRecorder;
  StreamSubscription<List<int>>? _audioStreamSubscription;
  
  // 状态管理
  bool _isConnected = false;
  bool _isRecording = false;
  int _currentSequence = 1;
  int _serverSequence = 0;  // 添加服务器序列号跟踪
  bool _isInitialized = false;
  bool _waitingForInitResponse = false;
  Completer<void>? _initCompleter;
  
  // 回调
  Function(String)? onResult;
  Function(String)? onError;
  Function(String)? onConnectionStatusChanged;
  
  VolcanoAsrService({
    required this.appKey,
    required this.accessKey,
    this.onResult,
    this.onError,
    this.onConnectionStatusChanged,
  });
  
  Future<void> _connectWebSocket() async {
    if (_webSocket != null) {
      debugPrint('⚠️ ASR: WebSocket already exists');
      return;
    }
    
    final connectId = const Uuid().v4();
    final headers = {
      'X-Api-App-Key': appKey,
      'X-Api-Access-Key': accessKey,
      'X-Api-Resource-Id': 'volc.bigasr.sauc.duration',
      'X-Api-Connect-Id': connectId,
      'Connection': 'Upgrade',  // 添加 WebSocket 升级头
      'Upgrade': 'websocket',
    };
    
    try {
      _webSocket = await WebSocket.connect(
        _baseUrl,
        headers: headers,
        protocols: ['binary'],  // 指定二进制协议
      );
      debugPrint('ℹ️ ASR: WebSocket connected');
      
      _webSocket!.listen(
        _handleWebSocketMessage,
        onError: (error) {
          debugPrint('🔴 ASR: WebSocket error: $error');
          onError?.call('WebSocket error: $error');
          _handleDisconnect();
        },
        onDone: () {
          debugPrint('ℹ️ ASR: WebSocket closed');
          if (_waitingForInitResponse) {
            _initCompleter?.completeError('WebSocket closed during initialization');
            _initCompleter = null;
          }
          _handleDisconnect();
        },
        cancelOnError: false,  // 不要在错误时自动取消
      );
      
      _isConnected = true;
      _currentSequence = 1;  // 重置序列号
      _isInitialized = false;  // 重置初始化状态
      onConnectionStatusChanged?.call('connected');
      
      // 发送初始化请求前等待一下
      await Future.delayed(const Duration(milliseconds: 100));
      await _sendInitRequest(connectId);
      
    } catch (e) {
      debugPrint('🔴 ASR: Failed to connect: $e');
      rethrow;
    }
  }
  
  Future<void> _sendInitRequest(String requestId) async {
    _waitingForInitResponse = true;
    _initCompleter = Completer<void>();
    
    final initRequest = {
      'audio': {
        'format': 'raw',
        'sample_rate': 16000,
        'channels': 1,
        'encoding': 'pcm',
        'bit_depth': 16,
      },
      'request': {
        'request_id': requestId,
        'language': 'zh',
        'model_name': 'bigmodel',
        'enable_punc': true,
        'enable_itn': true,
        'enable_voice_detection': true,
      }
    };
    
    final jsonBytes = utf8.encode(json.encode(initRequest));
    debugPrint('ℹ️ ASR: Sending init request: ${json.encode(initRequest)}');
    
    final header = Uint8List(4)
      ..[0] = 0x11  // version=1, header_size=1
      ..[1] = 0x11  // message_type=1(FULL_CLIENT_REQUEST), flags=1(POS_SEQUENCE)
      ..[2] = 0x10  // serialization=1(JSON), compression=0
      ..[3] = 0x00; // reserved
      
    final seqBytes = ByteData(4)..setInt32(0, _currentSequence, Endian.big);
    final sizeBytes = ByteData(4)..setInt32(0, jsonBytes.length, Endian.big);
    
    final fullRequest = Uint8List(header.length + 8 + jsonBytes.length)
      ..setAll(0, header)
      ..setAll(4, seqBytes.buffer.asUint8List())
      ..setAll(8, sizeBytes.buffer.asUint8List())
      ..setAll(12, jsonBytes);
    
    _webSocket?.add(fullRequest);
    debugPrint('ℹ️ ASR: Init request sent with sequence: $_currentSequence');
    
    // 设置超时
    Future.delayed(const Duration(seconds: 10), () {
      if (_waitingForInitResponse) {
        debugPrint('⚠️ ASR: Init request timeout after 10 seconds');
        _waitingForInitResponse = false;
        _initCompleter?.completeError('Init request timeout');
        _initCompleter = null;
      }
    });
    
    try {
      await _initCompleter?.future;
    } catch (e) {
      debugPrint('🔴 ASR: Init request failed: $e');
      rethrow;
    }
  }
  
  void _handleWebSocketMessage(dynamic message) {
    debugPrint('ℹ️ ASR: Received message type: ${message.runtimeType}');
    
    if (message is String) {
      debugPrint('ℹ️ ASR: Received string message: $message');
      try {
        final data = json.decode(message);
        if (data is Map) {
          if (data['error'] != null) {
            debugPrint('🔴 ASR: Server error: ${data['error']}');
            if (_waitingForInitResponse) {
              _waitingForInitResponse = false;
              _initCompleter?.completeError(data['error']);
              _initCompleter = null;
            }
            onError?.call(data['error']);
            return;
          }
        }
      } catch (e) {
        debugPrint('⚠️ ASR: Failed to parse string message: $e');
      }
    }
    
    if (message is! Uint8List || message.length < 12) {
      debugPrint('⚠️ ASR: Invalid binary message format');
      return;
    }
    
    final version = message[0] >> 4;
    final headerSize = message[0] & 0x0F;
    final messageType = message[1] >> 4;
    final flags = message[1] & 0x0F;
    final serialization = message[2] >> 4;
    final compression = message[2] & 0x0F;
    
    debugPrint('ℹ️ ASR: Binary message: version=$version, headerSize=$headerSize, type=$messageType, flags=$flags, serialization=$serialization, compression=$compression');
    
    final sequence = ByteData.view(message.buffer, 4, 4).getInt32(0, Endian.big);
    debugPrint('ℹ️ ASR: Message sequence: $sequence');
    _serverSequence = sequence;
    
    final payloadSize = ByteData.view(message.buffer, 8, 4).getInt32(0, Endian.big);
    if (message.length < 12 + payloadSize) {
      debugPrint('⚠️ ASR: Incomplete message received');
      return;
    }
    
    if (messageType == 0x0F) { // INIT_RESPONSE
      if (_waitingForInitResponse) {
        debugPrint('ℹ️ ASR: Received init response (type 15)');
        if (payloadSize > 0) {
          final payload = message.sublist(12, 12 + payloadSize);
          final jsonStr = utf8.decode(payload);
          debugPrint('ℹ️ ASR: Init response payload: $jsonStr');
          try {
            final response = json.decode(jsonStr);
            if (response is Map) {
              if (response['error'] != null) {
                final error = response['error'];
                debugPrint('🔴 ASR: Init response error: $error');
                _waitingForInitResponse = false;
                _initCompleter?.completeError(error);
                _initCompleter = null;
                return;
              }
              debugPrint('ℹ️ ASR: Init response success');
              _waitingForInitResponse = false;
              _currentSequence = (_serverSequence + 1);
              _isInitialized = true;
              _initCompleter?.complete();
              _initCompleter = null;
              return;
            }
          } catch (e) {
            debugPrint('⚠️ ASR: Failed to parse init response: $e');
          }
        } else {
          debugPrint('ℹ️ ASR: Empty init response success');
          _waitingForInitResponse = false;
          _currentSequence = (_serverSequence + 1);
          _isInitialized = true;
          _initCompleter?.complete();
          _initCompleter = null;
          return;
        }
      }
    } else if (messageType == 0x09) { // FULL_SERVER_RESPONSE
      final payload = message.sublist(12, 12 + payloadSize);
      final jsonStr = utf8.decode(payload);
      debugPrint('ℹ️ ASR: Received binary payload: $jsonStr');
      
      try {
        final response = json.decode(jsonStr);
        if (response is Map && response['result'] != null) {
          final result = response['result'];
          if (result['text'] != null && result['text'].isNotEmpty) {
            debugPrint('ℹ️ ASR: Result: ${result['text']}');
            onResult?.call(result['text']);
          }
        }
      } catch (e) {
        debugPrint('⚠️ ASR: Failed to parse binary payload: $e');
      }
    }
  }
  
  Future<void> _startRecording() async {
    _audioRecorder ??= AudioRecorder();
    
    try {
      if (!await _audioRecorder!.hasPermission()) {
        throw Exception('Microphone permission denied');
      }
      
      const config = RecordConfig(
        encoder: AudioEncoder.pcm16bits,
        bitRate: 256000,
        sampleRate: 16000,
        numChannels: 1,
      );
      
      // 启动录音流
      final stream = await _audioRecorder!.startStream(config);
      _isRecording = true;
      
      // 监听音频流
      _audioStreamSubscription = stream.listen(
        (data) {
          if (_isRecording && _isConnected) {
            _sendAudioData(data, false);
          }
        },
        onError: (error) {
          debugPrint('🔴 ASR: Stream error: $error');
          _stopRecording();
        },
      );
      
    } catch (e) {
      debugPrint('🔴 ASR: Recording error: $e');
      await _stopRecording();
      rethrow;
    }
  }
  
  void _sendAudioData(List<int> audioData, bool isLast) {
    if (!_isConnected || _webSocket == null || !_isInitialized) {
      debugPrint('⚠️ ASR: Cannot send audio data: not ready');
      return;
    }
    
    final header = Uint8List(4)
      ..[0] = 0x11  // version=1, header_size=1
      ..[1] = isLast ? 0x23 : 0x21  // message_type=2(AUDIO_ONLY), flags=3/1
      ..[2] = 0x00  // no serialization, no compression
      ..[3] = 0x00; // reserved
      
    final seqBytes = ByteData(4)..setInt32(0, _currentSequence, Endian.big);
    final sizeBytes = ByteData(4)..setInt32(0, audioData.length, Endian.big);
    
    final packet = Uint8List(12 + audioData.length)
      ..setAll(0, header)
      ..setAll(4, seqBytes.buffer.asUint8List())
      ..setAll(8, sizeBytes.buffer.asUint8List())
      ..setAll(12, audioData);
    
    _webSocket?.add(packet);
    debugPrint('🔍 ASR: Sent audio data with sequence: $_currentSequence');
    _currentSequence++;  // 递增序列号
  }
  
  void _handleDisconnect() {
    _webSocket = null;
    _isConnected = false;
    _isInitialized = false;  // 重置初始化状态
    _currentSequence = 1;  // 重置序列号
    onConnectionStatusChanged?.call('disconnected');
    _stopRecording();
  }
  
  Future<void> _stopRecording() async {
    _isRecording = false;
    
    try {
      await _audioStreamSubscription?.cancel();
      _audioStreamSubscription = null;
      
      await _audioRecorder?.stop();
      _audioRecorder = null;
    } catch (e) {
      debugPrint('⚠️ ASR: Error stopping recording: $e');
    }
  }
  
  Future<void> startRecognition() async {
    await _connectWebSocket();
    await _startRecording();
  }
  
  Future<void> stopRecognition() async {
    await _stopRecording();
    _webSocket?.close();
    _handleDisconnect();
  }
  
  @override
  void onClose() {
    _audioStreamSubscription?.cancel();
    stopRecognition();
    super.onClose();
  }
}