✅ Интеграция SSE + Redis Pub/Sub для real-time OCR результатов

🎯 Основные изменения: Backend: - Реализован SSE endpoint /events/{task_id} для real-time стриминга событий - Интеграция Redis Pub/Sub для получения событий от n8n - Исправлен путь к .env файлу (абсолютный путь) - Убран префикс /api/v1 для events router - Добавлено подробное логирование событий Frontend: - Переключён на Vite dev mode для работы proxy - Настроен proxy /events -> backend:8100 - Реализована модалка с крутилкой при загрузке файла - SSE клиент для получения OCR результатов в real-time - Отображение результатов AI анализа в модалке Docker: - Frontend: изменён на npm run dev (Vite dev server) - Добавлен host.docker.internal для доступа к backend - Настроен proxy в docker-compose Утилиты: - monitor_redis_direct.py - мониторинг Redis Pub/Sub - test_redis_publish_direct.py - тестирование публикации в Redis 🚀 Полная цепочка работает: Frontend → Backend SSE → Redis Pub/Sub ← n8n → OCR/AI → Result
2025-10-27 19:37:41 +03:00
parent d8508aa89d
commit f72aa59476
9 changed files with 331 additions and 64 deletions
--- a/backend/app/api/events.py
+++ b/backend/app/api/events.py
@@ -79,13 +79,14 @@ async def stream_events(task_id: str):
    Returns:
        StreamingResponse с событиями
    """
+    logger.info(f"🚀 SSE connection requested for task_id: {task_id}")
    
    async def event_generator():
        """Генератор событий из Redis Pub/Sub"""
        channel = f"ocr_events:{task_id}"
        
        # Подписываемся на канал Redis
-        pubsub = redis_service.redis.pubsub()
+        pubsub = redis_service.client.pubsub()
        await pubsub.subscribe(channel)
        
        logger.info(f"📡 Client subscribed to {channel}")
@@ -96,19 +97,37 @@ async def stream_events(task_id: str):
        try:
            # Слушаем события
            while True:
+                logger.info(f"⏳ Waiting for message on {channel}...")
                message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=30.0)
                
-                if message and message['type'] == 'message':
-                    event_data = message['data'].decode('utf-8')
+                if message:
+                    logger.info(f"📥 Received message type: {message['type']}")
+                    if message['type'] == 'message':
+                        event_data = message['data']  # Уже строка (decode_responses=True)
+                        logger.info(f"📦 Raw event data: {event_data[:200]}...")
+                        event = json.loads(event_data)
                        
-                    # Отправляем событие клиенту
-                    yield f"data: {event_data}\n\n"
+                        # Обработка формата от n8n Redis ноды (вложенный)
+                        # Формат: {"claim_id": "...", "event": {...}}
+                        if 'event' in event and isinstance(event['event'], dict):
+                            # Извлекаем вложенное событие
+                            actual_event = event['event']
+                            logger.info(f"📦 Unwrapped n8n Redis format for {task_id}")
+                        else:
+                            # Формат уже плоский (от backend API или старых источников)
+                            actual_event = event
                        
-                    # Если обработка завершена - закрываем соединение
-                    event = json.loads(event_data)
-                    if event.get('status') in ['completed', 'error']:
-                        logger.info(f"✅ Task {task_id} finished, closing SSE")
-                        break
+                        # Отправляем событие клиенту (плоский формат)
+                        event_json = json.dumps(actual_event, ensure_ascii=False)
+                        logger.info(f"📤 Sending event to client: {actual_event.get('status', 'unknown')}")
+                        yield f"data: {event_json}\n\n"
+                        
+                        # Если обработка завершена - закрываем соединение
+                        if actual_event.get('status') in ['completed', 'error', 'success']:
+                            logger.info(f"✅ Task {task_id} finished, closing SSE")
+                            break
+                else:
+                    logger.info(f"⏰ Timeout waiting for message on {channel}")
                
                # Пинг каждые 30 сек чтобы соединение не закрылось
                await asyncio.sleep(0.1)
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -163,7 +163,7 @@ class Settings(BaseSettings):
    log_file: str = "/app/logs/erv_platform.log"
    
    class Config:
-        env_file = "../.env"
+        env_file = "/var/www/fastuser/data/www/crm.clientright.ru/erv_platform/.env"
        case_sensitive = False
        extra = "ignore"  # Игнорируем лишние поля из .env

--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -98,7 +98,7 @@ app.include_router(claims.router)
 app.include_router(policy.router)
 app.include_router(upload.router)
 app.include_router(draft.router)
-app.include_router(events.router, prefix="/api/v1")
+app.include_router(events.router)


@app.get("/")
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -8,6 +8,8 @@ services:
      - "5173:3000"
    environment:
      - REACT_APP_API_URL=http://147.45.146.17:8100
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
    networks:
      - erv-network
    restart: unless-stopped
--- a/frontend/Dockerfile
+++ b/frontend/Dockerfile
@@ -1,4 +1,4 @@
-# React Frontend Dockerfile
+# React Frontend Dockerfile (DEV MODE для Vite Proxy)
 FROM node:18-alpine

 # Устанавливаем рабочую директорию
@@ -13,14 +13,8 @@ RUN npm install
 # Копируем исходный код
 COPY . .

-# Собираем приложение
-RUN npm run build
-
-# Устанавливаем serve для статических файлов
-RUN npm install -g serve
-
-# Открываем порт
+# Открываем порт (Vite dev server на 5173, но внутри контейнера на 3000)
 EXPOSE 3000

-# Запускаем приложение
-CMD ["serve", "-s", "dist", "-l", "3000"]
+# Запускаем Vite dev server с proxy (изменяем порт на 3000)
+CMD ["npm", "run", "dev", "--", "--host", "0.0.0.0", "--port", "3000"]
--- a/frontend/src/components/form/Step1Policy.tsx
+++ b/frontend/src/components/form/Step1Policy.tsx
@@ -1,6 +1,6 @@
 import { useState, useEffect, useRef } from 'react';
 import { Form, Input, Button, message, Upload, Spin, Alert, Modal } from 'antd';
-import { FileProtectOutlined, UploadOutlined, LoadingOutlined, CheckCircleOutlined, CloseCircleOutlined } from '@ant-design/icons';
+import { FileProtectOutlined, UploadOutlined, LoadingOutlined } from '@ant-design/icons';
 import type { UploadFile } from 'antd/es/upload/interface';
 import { convertToPDF } from '../../utils/pdfConverter';

@@ -57,57 +57,86 @@ export default function Step1Policy({ formData, updateFormData, onNext, addDebug
  const [policyNotFound, setPolicyNotFound] = useState(false);
  const [fileList, setFileList] = useState<UploadFile[]>([]);
  const [uploading, setUploading] = useState(false);
+  const [waitingForOcr, setWaitingForOcr] = useState(false);  // ⬅️ НОВЫЙ state для ожидания SSE!
  const [uploadProgress, setUploadProgress] = useState('');
-  const [ocrResult, setOcrResult] = useState<any>(null);
+  const [, setOcrResult] = useState<any>(null);
+  const [ocrModalVisible, setOcrModalVisible] = useState(false);  // ⬅️ Видимость модалки
+  const [ocrModalContent, setOcrModalContent] = useState<any>(null);  // ⬅️ Контент модалки
  const eventSourceRef = useRef<EventSource | null>(null);
  
  // SSE подключение для получения результатов OCR/Vision
  useEffect(() => {
    const claimId = formData.claim_id;
-    if (!claimId || !uploading) return;
+    if (!claimId || !waitingForOcr) {
+      console.log('🔍 SSE useEffect: условие не выполнено', { claimId, waitingForOcr });
+      return;
+    }

-    // Подключаемся к SSE для получения результатов OCR
-    const eventSource = new EventSource(`http://147.45.189.234:8000/events/${claimId}`);
+    console.log('🔌 SSE: Открываю соединение к', `/events/${claimId}`);
+    
+    // Открываем модалку с крутилкой
+    setOcrModalVisible(true);
+    setOcrModalContent('loading');
+    
+    // Подключаемся к SSE для получения результатов OCR (через Vite proxy)
+    const eventSource = new EventSource(`/events/${claimId}`);
    eventSourceRef.current = eventSource;
    
+    console.log('✅ SSE: EventSource создан');
+
    eventSource.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        console.log('📨 SSE event received:', data);

        if (data.event_type === 'ocr_completed') {
-          setUploadProgress('');  // Убираем крутилку
+          console.log('✅ SSE: Получил событие ocr_completed!', data);
+          
+          setUploadProgress('');
+          setUploading(false);
+          setWaitingForOcr(false);  // Останавливаем ожидание
          setOcrResult(data);

-          if (data.status === 'success' && data.data?.is_valid_document) {
-            // ✅ Полис распознан успешно
-            message.success(data.message || '✅ Полис успешно распознан!');
-            addDebugEvent?.('ocr', 'success', data.message, data.data);
+          // Обрабатываем формат от n8n: data.output.is_policy или data.is_valid_document
+          const aiOutput = data.data?.output || data.data;
+          const isValidPolicy = aiOutput?.is_policy === 'yes' || data.data?.is_valid_document === true;
+          
+          // Обновляем содержимое модалки на результат (вместо крутилки)
+          setOcrModalContent({ success: isValidPolicy, data: aiOutput, message: data.message });
+          
+          if (data.status === 'completed' || data.status === 'success') {
+            const policyNumber = aiOutput?.policy_number || 'неизвестно';
+            const holderName = aiOutput?.policyholder_full_name || '';
+            const insuredPersons = aiOutput?.insured_persons || [];
+            
+            if (isValidPolicy) {
+              // ✅ Полис распознан - логируем в Debug Panel
+              addDebugEvent?.('ocr_ai_result', 'success', `✅ AI анализ завершён`, {
+                policy_number: policyNumber,
+                holder: holderName,
+                insured_persons: insuredPersons,
+                policy_period: aiOutput?.policy_period,
+                program_name: aiOutput?.program_name,
+                full_ai_output: aiOutput
+              });
+              
+              // Сохраняем извлечённые AI данные
+              updateFormData({ 
+                policyAiData: aiOutput,
+                policyNumber: policyNumber,
+                holderName: holderName
+              });
+            } else {
+              // ❌ Не полис
+              addDebugEvent?.('ocr', 'error', '❌ Документ не является полисом ERV', aiOutput);
+              setFileList([]);
+              setPolicyNotFound(true);
+            }
          } else {
-            // ❌ Документ не распознан или это не полис
-            const warnings = data.data?.ai_analysis?.warnings || ['Документ не распознан'];
-            
-            Modal.error({
-              title: '❌ Документ не распознан',
-              content: (
-                <div>
-                  <p>{data.message}</p>
-                  {warnings.length > 0 && (
-                    <ul>
-                      {warnings.map((w: string, i: number) => (
-                        <li key={i}>{w}</li>
-                      ))}
-                    </ul>
-                  )}
-                  <p style={{ marginTop: 12, color: '#666' }}>
-                    Пожалуйста, загрузите скан страхового полиса ERV.
-                  </p>
-                </div>
-              ),
-            });
-            
-            addDebugEvent?.('ocr', 'error', data.message, data.data);
-            setFileList([]);  // Очищаем список файлов
+            // Ошибка обработки
+            addDebugEvent?.('ocr', 'error', data.message || 'Ошибка OCR', data.data);
+            setFileList([]);
+            setPolicyNotFound(true);
          }
        }
      } catch (error) {
@@ -116,17 +145,24 @@ export default function Step1Policy({ formData, updateFormData, onNext, addDebug
    };

    eventSource.onerror = (error) => {
-      console.error('SSE connection error:', error);
+      console.error('❌ SSE connection error:', error);
+      console.error('SSE readyState:', eventSource.readyState);
+      setOcrModalContent({ success: false, data: null, message: 'Ошибка подключения к серверу' });
+      setWaitingForOcr(false);
      eventSource.close();
    };
    
+    eventSource.onopen = () => {
+      console.log('✅ SSE: Соединение открыто!');
+    };
+
    return () => {
      if (eventSourceRef.current) {
        eventSourceRef.current.close();
        eventSourceRef.current = null;
      }
    };
-  }, [formData.claim_id, uploading]);
+  }, [formData.claim_id, waitingForOcr]);

  // Обработчик изменения поля полиса с автозаменой и маской
  const handleVoucherChange = (e: React.ChangeEvent<HTMLInputElement>) => {
@@ -321,8 +357,15 @@ export default function Step1Policy({ formData, updateFormData, onNext, addDebug
          policyValidationWarning: '' // Silent validation
        });

-        message.success(`Загружено файлов: ${uploadResult.uploaded_count}`);
-        onNext();
+        // ⏳ Включаем режим ожидания SSE результата!
+        console.log('🔄 Устанавливаю waitingForOcr=true для claim_id:', claimId);
+        setWaitingForOcr(true);  // ⬅️ Это откроет SSE соединение в useEffect!
+        setUploadProgress('⏳ Ждём результат распознавания полиса...');
+        message.info('Файл загружен. Ожидаем результат OCR и AI анализа...');
+        console.log('📡 waitingForOcr установлен в true, useEffect должен сработать!');
+        
+        // SSE событие обработается в useEffect и покажет модалку
+        // НЕ вызываем onNext() здесь!
      } else {
        addDebugEvent?.('upload', 'error', `❌ Ошибка загрузки файлов`, { error: 'Upload failed' });
        message.error('Ошибка загрузки файлов');
@@ -490,6 +533,70 @@ export default function Step1Policy({ formData, updateFormData, onNext, addDebug
          </p>
        </div>
      )}
+
+      {/* Модальное окно ожидания OCR результата */}
+      <Modal
+        open={ocrModalVisible}
+        closable={ocrModalContent !== 'loading'}
+        maskClosable={false}
+        footer={ocrModalContent === 'loading' ? null : [
+          <Button key="close" type="primary" onClick={() => setOcrModalVisible(false)}>
+            Закрыть
+          </Button>
+        ]}
+        width={700}
+        centered
+      >
+        {ocrModalContent === 'loading' ? (
+          <div style={{ textAlign: 'center', padding: '40px 20px' }}>
+            <Spin indicator={<LoadingOutlined style={{ fontSize: 48 }} spin />} />
+            <h3 style={{ marginTop: 24, marginBottom: 12 }}>⏳ Обрабатываем документ</h3>
+            <p style={{ color: '#666', marginBottom: 8 }}>OCR распознавание текста...</p>
+            <p style={{ color: '#666', marginBottom: 8 }}>AI анализ содержимого...</p>
+            <p style={{ color: '#666' }}>Проверка валидности полиса...</p>
+            <p style={{ color: '#999', fontSize: 12, marginTop: 20 }}>
+              Это может занять 20-30 секунд. Пожалуйста, подождите...
+            </p>
+          </div>
+        ) : ocrModalContent ? (
+          <div>
+            <h3 style={{ marginBottom: 16 }}>
+              {ocrModalContent.success ? '✅ Результат распознавания' : '❌ Ошибка распознавания'}
+            </h3>
+            {ocrModalContent.success ? (
+              <div>
+                <p><strong>Номер полиса:</strong> {ocrModalContent.data?.policy_number || 'н/д'}</p>
+                <p><strong>Владелец:</strong> {ocrModalContent.data?.policyholder_full_name || 'н/д'}</p>
+                {ocrModalContent.data?.insured_persons?.length > 0 && (
+                  <>
+                    <p><strong>Застрахованные лица:</strong></p>
+                    <ul>
+                      {ocrModalContent.data.insured_persons.map((person: any, i: number) => (
+                        <li key={i}>{person.full_name} (ДР: {person.birth_date || 'н/д'})</li>
+                      ))}
+                    </ul>
+                  </>
+                )}
+                {ocrModalContent.data?.policy_period && (
+                  <p><strong>Период:</strong> {ocrModalContent.data.policy_period.insured_from} - {ocrModalContent.data.policy_period.insured_to}</p>
+                )}
+                <p style={{ marginTop: 16 }}><strong>Полный ответ AI:</strong></p>
+                <pre style={{ background: '#f5f5f5', padding: 12, borderRadius: 4, fontSize: 12, maxHeight: 400, overflow: 'auto' }}>
+                  {JSON.stringify(ocrModalContent.data, null, 2)}
+                </pre>
+              </div>
+            ) : (
+              <div>
+                <p>{ocrModalContent.message || 'Документ не распознан'}</p>
+                <p style={{ marginTop: 16 }}><strong>Полный ответ:</strong></p>
+                <pre style={{ background: '#fff3f3', padding: 12, borderRadius: 4, fontSize: 12, maxHeight: 400, overflow: 'auto' }}>
+                  {JSON.stringify(ocrModalContent.data, null, 2)}
+                </pre>
+              </div>
+            )}
+          </div>
+        ) : null}
+      </Modal>
    </Form>
  );
 }
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -5,10 +5,14 @@ export default defineConfig({
  plugins: [react()],
  server: {
    host: '0.0.0.0',
-    port: 5173,
+    port: 3000,
    proxy: {
      '/api': {
-        target: 'http://localhost:8100',
+        target: 'http://host.docker.internal:8100',
+        changeOrigin: true
+      },
+      '/events': {
+        target: 'http://host.docker.internal:8100',
        changeOrigin: true
      }
    }
--- a/monitor_redis_direct.py
+++ b/monitor_redis_direct.py
@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""
+Мониторинг Redis Pub/Sub для проверки прямой публикации из n8n
+"""
+import redis
+import json
+from datetime import datetime
+
+print("=" * 60)
+print("🎧 МОНИТОРИНГ REDIS PUB/SUB")
+print("=" * 60)
+
+# Подключение к Redis
+r = redis.Redis(
+    host='crm.clientright.ru',
+    port=6379,
+    password='CRM_Redis_Pass_2025_Secure!',
+    decode_responses=True
+)
+
+# Проверка подключения
+try:
+    r.ping()
+    print("✅ Redis подключен!")
+except Exception as e:
+    print(f"❌ Ошибка подключения: {e}")
+    exit(1)
+
+# Подписка на все каналы ocr_events:*
+pubsub = r.pubsub()
+pubsub.psubscribe('ocr_events:*')
+
+print(f"📡 Слушаем каналы: ocr_events:*")
+print(f"⏰ Запущено: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+print("-" * 60)
+print("\n⏳ Ожидаю события... (Ctrl+C для выхода)\n")
+
+# Счётчик событий
+event_count = 0
+
+try:
+    for message in pubsub.listen():
+        if message['type'] == 'pmessage':
+            event_count += 1
+            print(f"\n{'='*60}")
+            print(f"📢 СОБЫТИЕ #{event_count}")
+            print(f"⏰ Время: {datetime.now().strftime('%H:%M:%S')}")
+            print(f"📺 Канал: {message['channel']}")
+            print(f"📦 Данные:")
+            print("-" * 60)
+            
+            try:
+                # Пытаемся распарсить как JSON
+                data = json.loads(message['data'])
+                print(json.dumps(data, indent=2, ensure_ascii=False))
+            except json.JSONDecodeError:
+                # Если не JSON - показываем как есть
+                print(message['data'])
+            
+            print(f"{'='*60}\n")
+            
+except KeyboardInterrupt:
+    print(f"\n\n✅ Остановлено. Получено событий: {event_count}")
+    print(f"⏰ Завершено: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+finally:
+    pubsub.close()
+    r.close()
+
--- a/test_redis_publish_direct.py
+++ b/test_redis_publish_direct.py
@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+"""
+Тест прямой публикации в Redis (имитация n8n Redis ноды)
+"""
+import redis
+import json
+from datetime import datetime
+
+print("=" * 60)
+print("🧪 ТЕСТ ПРЯМОЙ ПУБЛИКАЦИИ В REDIS")
+print("=" * 60)
+
+# Подключение к Redis
+r = redis.Redis(
+    host='crm.clientright.ru',
+    port=6379,
+    password='CRM_Redis_Pass_2025_Secure!',
+    decode_responses=True
+)
+
+# Проверка подключения
+try:
+    r.ping()
+    print("✅ Redis подключен!")
+except Exception as e:
+    print(f"❌ Ошибка подключения: {e}")
+    exit(1)
+
+# Тестовые данные
+claim_id = "CLM-TEST-DIRECT-123"
+channel = f"ocr_events:{claim_id}"
+
+event_data = {
+    "event_type": "ocr_completed",
+    "status": "success",
+    "message": "✅ Тест прямой публикации из Python (имитация n8n)",
+    "data": {
+        "file_id": "test-file-123",
+        "is_valid_document": True,
+        "test_mode": True,
+        "source": "direct_redis_publish"
+    },
+    "timestamp": datetime.now().isoformat()
+}
+
+message = json.dumps(event_data, ensure_ascii=False)
+
+print(f"\n📺 Канал: {channel}")
+print(f"📦 Сообщение:")
+print(json.dumps(event_data, indent=2, ensure_ascii=False))
+print("\n" + "=" * 60)
+
+# Публикация
+try:
+    num_subscribers = r.publish(channel, message)
+    print(f"\n✅ Сообщение опубликовано!")
+    print(f"👥 Количество подписчиков: {num_subscribers}")
+    
+    if num_subscribers == 0:
+        print("\n⚠️  ВНИМАНИЕ: Нет активных подписчиков!")
+        print("   Это нормально, если никто не слушает канал.")
+        print("   Запусти monitor_redis_direct.py в другом терминале.")
+    else:
+        print(f"\n🎉 {num_subscribers} подписчик(ов) получили сообщение!")
+        
+except Exception as e:
+    print(f"\n❌ Ошибка публикации: {e}")
+finally:
+    r.close()
+
+print("\n" + "=" * 60)
+print(f"⏰ Завершено: {datetime.now().strftime('%H:%M:%S')}")
+