AI-Health/health_app/lib/utils/sse_handler.dart

import 'dart:async';
import 'dart:convert';
import 'package:dio/dio.dart';
import '../core/api_client.dart';

/// 跨平台 SSE 流处理（基于 Dio 流式响应，支持 Android/iOS/Web）
class SseHandler {
  /// 连接 SSE 端点，返回事件流
  static Stream<Map<String, dynamic>> connect({
    required String agentType,
    required String message,
    String? conversationId,
    String? imageUrl,
    String? pdfUrl,
    required String token,
  }) {
    final params = <String, String>{'message': message};
    if (conversationId != null) {
      params['conversationId'] = conversationId;
    }
    if (imageUrl != null && imageUrl.isNotEmpty) {
      params['imageUrl'] = imageUrl;
    }
    if (pdfUrl != null && pdfUrl.isNotEmpty) {
      params['pdfUrl'] = pdfUrl;
    }
    final query = params.entries
        .map((e) => '${e.key}=${Uri.encodeComponent(e.value)}')
        .join('&');
    final url = '$baseUrl/api/ai/$agentType/chat?$query';

    final cancelToken = CancelToken();
    late final StreamController<Map<String, dynamic>> controller;
    controller = StreamController<Map<String, dynamic>>(
      onListen: () {
        _connect(controller, url, token, cancelToken);
      },
      onCancel: () {
        if (!cancelToken.isCancelled) {
          cancelToken.cancel('用户停止生成');
        }
      },
    );
    return controller.stream;
  }

  static Future<void> _connect(
    StreamController<Map<String, dynamic>> controller,
    String url,
    String token,
    CancelToken cancelToken,
  ) async {
    try {
      final dio = Dio(
        BaseOptions(
          connectTimeout: const Duration(seconds: 15),
          receiveTimeout: const Duration(minutes: 5),
        ),
      );

      final response = await dio.get(
        url,
        cancelToken: cancelToken,
        options: Options(
          responseType: ResponseType.stream,
          headers: {'Authorization': 'Bearer $token'},
        ),
      );

      final stream = response.data.stream as Stream<List<int>>;
      var buffer = '';

      await for (final chunk in stream) {
        if (controller.isClosed) break;
        final text = utf8.decode(chunk, allowMalformed: true);
        buffer += text;

        // 按行解析 SSE 数据
        while (buffer.contains('\n')) {
          final newlineIdx = buffer.indexOf('\n');
          var line = buffer.substring(0, newlineIdx).trim();
          buffer = buffer.substring(newlineIdx + 1);

          if (line.isEmpty || !line.startsWith('data: ')) continue;
          final data = line.substring(6);

          if (data == '[DONE]') {
            controller.close();
            return;
          }

          try {
            controller.add(jsonDecode(data) as Map<String, dynamic>);
          } catch (_) {
            // 跳过无法解析的行
          }
        }
      }
      controller.close();
    } on DioException catch (e) {
      if (CancelToken.isCancel(e)) {
        if (!controller.isClosed) await controller.close();
        return;
      }
      if (!controller.isClosed) {
        controller.add({'action': 'error', 'message': e.toString()});
        await controller.close();
      }
    } catch (e) {
      if (!controller.isClosed) {
        controller.add({'action': 'error', 'message': e.toString()});
        await controller.close();
      }
    }
  }
}