ai_chat_assistant/lib/services/chat_sse_service.dart

// 负责SSE通信
import 'dart:convert';
import 'dart:io';
import 'dart:math';

import 'package:ai_chat_assistant/utils/tts_util.dart';
import 'package:flutter/services.dart';

import '../utils/common_util.dart';

class ChatSseService {
  static const MethodChannel _channel = MethodChannel('com.example.ai_chat_assistant/tts');
  // 缓存用户ID和会话ID
  String? _cachedUserId;
  String? _cachedConversationId;

  HttpClient? _currentClient;
  HttpClientResponse? _currentResponse;
  bool _isAborted = true;
  bool _isFirstData = true;
  bool _isTtsStarted = false;

  String? get conversationId => _cachedConversationId;

  Future<void>? _startTtsFuture;

  void request({
    required String messageId,
    required String text,
    required bool isChinese,
    required Function(String, String, bool) onStreamResponse,
  }) async {
    print("----------------------SSE Start");
    _isAborted = false;
    if (_cachedUserId == null) {
      _cachedUserId = _generateRandomUserId(6);
      print('初始化用户ID: $_cachedUserId');
    }
    String responseText = '';
    String tempText = '';
    _currentClient = HttpClient();
    try {
      final chatUri = Uri.parse('http://143.64.185.20:18606/chat');
      final request = await _currentClient!.postUrl(chatUri);
      request.headers.set('Content-Type', 'application/json');
      request.headers.set('Accept', 'text/event-stream');
      final body = {
        'message': text,
        'user': _cachedUserId,
      };
      if (_cachedConversationId != null) {
        body['conversation_id'] = _cachedConversationId;
      }
      request.add(utf8.encode(json.encode(body)));
      _currentResponse = await request.close();
      if (_currentResponse!.statusCode == 200) {
        _startTtsFuture = startTts(isChinese);
        await for (final line in _currentResponse!
            .transform(utf8.decoder)
            .transform(const LineSplitter())) {
          if (_isAborted) {
            break;
          }
          if (line.startsWith('data:')) {
            final jsonStr = line.substring(5).trim();
            if (jsonStr == '[DONE]' || jsonStr.contains('message_end')) {
              onStreamResponse(messageId, responseText, true);
              break;
            }
            try {
              final jsonData = json.decode(jsonStr);
              if (jsonData.containsKey('conversation_id') &&
                  _cachedConversationId == null) {
                _cachedConversationId = jsonData['conversation_id'];
              }
              if (jsonData['event'].toString().contains('message')) {
                final textChunk =
                    jsonData.containsKey('answer') ? jsonData['answer'] : '';
                if (_isAborted) {
                  break;
                }
                tempText += textChunk;
                int endIndex = _getCompleteTextEndIndex(tempText);
                String completeText =  CommonUtil.cleanText(tempText.substring(0, endIndex).trim(), true);
                if (completeText.isNotEmpty) {
                  if (_isFirstData && _startTtsFuture != null) {
                    await _startTtsFuture;
                    _isFirstData = false;
                  }
                  print("----------------------Send text: " + completeText);
                  completeText += isChinese ? '，' : ',';
                  TtsUtil.send(completeText);
                }
                tempText = tempText.substring(endIndex).trim();
                responseText += textChunk;
                onStreamResponse(messageId, responseText, false);
              }
            } catch (e) {
              print('解析 SSE 数据出错: $e, 原始数据: $jsonStr');
            }
          }
        }
      } else {
        print('SSE 连接失败，状态码: ${_currentResponse!.statusCode}');
      }
    } catch (e) {
      // todo
    } finally {
      resetRequest();
    }
  }

  Future<void> startTts(bool isChinese) async {
    print("----------------------TTS Start");
    if (!_isTtsStarted) {
      if (await TtsUtil.start(isChinese) == true) {
        _isTtsStarted = true;
      }
    }
  }

  Future<void> abort() async {
    _isAborted = true;
    TtsUtil.stopTts();
    resetRequest();
  }

  void resetRequest() {
    _currentResponse?.detachSocket().then((socket) {
      socket.destroy();
    });
    _currentClient?.close(force: true);
    _currentClient = null;
    _currentResponse = null;
    _isFirstData = true;
    _isTtsStarted = false;
  }

  int _getCompleteTextEndIndex(String buffer) {
    // 支持句号、问号、感叹号和换行符作为分割依据
    final sentenceEnders = RegExp(r'[,!?:，。！？：\n]');
    final matches = sentenceEnders.allMatches(buffer);
    return matches.isEmpty ? 0 : matches.last.end;
  }

  // 新增：重置会话方法
  void resetSession() {
    _cachedUserId = null;
    _cachedConversationId = null;
    print('SSE会话已重置');
  }

  String _generateRandomUserId(int length) {
    const chars =
        'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789';
    final random = Random();
    return String.fromCharCodes(Iterable.generate(
        length, (_) => chars.codeUnitAt(random.nextInt(chars.length))));
  }
}