negodiy/aiform_dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚀 MVP: FastAPI + React форма с SMS верификацией

Browse Source 
✅ Инфраструктура: PostgreSQL, Redis, RabbitMQ, S3
✅ Backend: SMS сервис + API endpoints
✅ Frontend: React форма (3 шага) + SMS верификация
This commit is contained in:
AI Assistant
2025-10-24 16:19:58 +03:00
parent 8af23e90fa
commit 0f82eef08d
42 changed files with 2902 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
backend/Dockerfile Normal file
View File

@@ -0,0 +1,21 @@
# Python FastAPI Backend Dockerfile
FROM python:3.10-slim
# Устанавливаем рабочую директорию
WORKDIR /app
# Копируем requirements.txt
COPY requirements.txt .
# Устанавливаем зависимости
RUN pip install --no-cache-dir -r requirements.txt
# Копируем исходный код
COPY . .
# Открываем порт
EXPOSE 8100
# Запускаем приложение
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8100"]

4
backend/app/api/__init__.py
View File

@@ -0,0 +1,4 @@
"""
API Routes
"""

51
backend/app/api/claims.py Normal file
View File

@@ -0,0 +1,51 @@
"""
Claims API Routes - Обработка заявок
"""
from fastapi import APIRouter, HTTPException
from .models import ClaimCreateRequest, ClaimResponse
import uuid
from datetime import datetime
router = APIRouter(prefix="/api/v1/claims", tags=["Claims"])
@router.post("/create", response_model=ClaimResponse)
async def create_claim(claim: ClaimCreateRequest):
"""
Создать новую заявку
Принимает данные формы и создает заявку в системе
"""
try:
# Генерируем ID и номер заявки
claim_id = str(uuid.uuid4())
claim_number = f"ERV-{datetime.now().strftime('%Y%m%d')}-{claim_id[:8].upper()}"
# TODO: Сохранить в PostgreSQL
# TODO: Отправить в очередь RabbitMQ для обработки
# TODO: Интеграция с CRM
return ClaimResponse(
success=True,
claim_id=claim_id,
claim_number=claim_number,
message=f"Заявка {claim_number} успешно создана"
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Ошибка при создании заявки: {str(e)}"
)
@router.get("/{claim_id}")
async def get_claim(claim_id: str):
"""Получить информацию о заявке по ID"""
# TODO: Получить из БД
return {
"claim_id": claim_id,
"status": "processing",
"message": "Заявка в обработке"
}

64
backend/app/api/models.py Normal file
View File

@@ -0,0 +1,64 @@
"""
Pydantic модели для API
"""
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
from datetime import date
class SMSSendRequest(BaseModel):
"""Запрос на отправку SMS кода"""
phone: str = Field(..., description="Номер телефона в формате +79001234567")
@field_validator('phone')
@classmethod
def validate_phone(cls, v: str) -> str:
# Убираем все кроме цифр и +
clean = ''.join(c for c in v if c.isdigit() or c == '+')
if not clean.startswith('+'):
clean = '+' + clean
if len(clean) != 12: # +7 + 10 цифр
raise ValueError('Неверный формат телефона')
return clean
class SMSVerifyRequest(BaseModel):
"""Запрос на проверку SMS кода"""
phone: str = Field(..., description="Номер телефона")
code: str = Field(..., min_length=6, max_length=6, description="6-значный код")
class ClaimCreateRequest(BaseModel):
"""Запрос на создание заявки"""
# Шаг 1: Основная информация
phone: str
email: Optional[str] = None
inn: Optional[str] = None
policy_number: str
policy_series: Optional[str] = None
# Шаг 2: Данные о происшествии
incident_date: Optional[str] = None
incident_description: Optional[str] = None
transport_type: Optional[str] = None # "air", "train", "bus", etc.
# Шаг 3: Данные для выплаты
payment_method: str = "sbp" # "sbp", "card", "bank_transfer"
bank_name: Optional[str] = None
card_number: Optional[str] = None
account_number: Optional[str] = None
# Файлы (UUID после загрузки)
uploaded_files: Optional[List[str]] = []
# Метаданные
source: str = "web_form"
class ClaimResponse(BaseModel):
"""Ответ после создания заявки"""
success: bool
claim_id: Optional[str] = None
claim_number: Optional[str] = None
message: str

53
backend/app/api/sms.py Normal file
View File

@@ -0,0 +1,53 @@
"""
SMS API Routes
"""
from fastapi import APIRouter, HTTPException
from ..services.sms_service import sms_service
from .models import SMSSendRequest, SMSVerifyRequest
router = APIRouter(prefix="/api/v1/sms", tags=["SMS"])
@router.post("/send")
async def send_sms_code(request: SMSSendRequest):
"""
Отправить SMS код верификации
- **phone**: Номер телефона в формате +79001234567
"""
code = await sms_service.send_verification_code(request.phone)
if code:
return {
"success": True,
"message": "Код отправлен на указанный номер",
"debug_code": code if sms_service.enabled else None # Показываем код только в dev
}
else:
raise HTTPException(
status_code=429,
detail="Слишком много запросов. Попробуйте через минуту."
)
@router.post("/verify")
async def verify_sms_code(request: SMSVerifyRequest):
"""
Проверить SMS код
- **phone**: Номер телефона
- **code**: 6-значный код из SMS
"""
is_valid = await sms_service.verify_code(request.phone, request.code)
if is_valid:
return {
"success": True,
"message": "Код подтвержден"
}
else:
raise HTTPException(
status_code=400,
detail="Неверный код или код истек"
)

125
backend/app/config.py
View File

@@ -3,10 +3,13 @@
"""
from pydantic_settings import BaseSettings
from functools import lru_cache
from typing import List
class Settings(BaseSettings):
# App
# ============================================
# APPLICATION
# ============================================
app_name: str = "ERV Insurance Platform"
app_env: str = "development"
debug: bool = True
@@ -16,47 +19,144 @@ class Settings(BaseSettings):
backend_url: str = "http://localhost:8100"
frontend_url: str = "http://localhost:5173"
# PostgreSQL
# ============================================
# DATABASE (PostgreSQL)
# ============================================
postgres_host: str = "147.45.189.234"
postgres_port: int = 5432
postgres_db: str = "default_db"
postgres_user: str = "gen_user"
postgres_password: str = "2~~9_^kVsU?2\\S"
# Redis
@property
def database_url(self) -> str:
"""Формирует URL для подключения к PostgreSQL"""
return f"postgresql+asyncpg://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
# ============================================
# REDIS
# ============================================
redis_host: str = "localhost"
redis_port: int = 6379
redis_password: str = "CRM_Redis_Pass_2025_Secure!"
redis_db: int = 0
redis_prefix: str = "erv:"
# RabbitMQ
@property
def redis_url(self) -> str:
"""Формирует URL для подключения к Redis"""
if self.redis_password:
return f"redis://:{self.redis_password}@{self.redis_host}:{self.redis_port}/{self.redis_db}"
return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"
# ============================================
# RABBITMQ
# ============================================
rabbitmq_host: str = "185.197.75.249"
rabbitmq_port: int = 5672
rabbitmq_user: str = "admin"
rabbitmq_password: str = "tyejvtej"
rabbitmq_vhost: str = "/"
# OCR Service
ocr_api_url: str = "http://147.45.146.17:8001"
@property
def rabbitmq_url(self) -> str:
"""Формирует URL для подключения к RabbitMQ"""
return f"amqp://{self.rabbitmq_user}:{self.rabbitmq_password}@{self.rabbitmq_host}:{self.rabbitmq_port}{self.rabbitmq_vhost}"
# OpenRouter AI
# ============================================
# S3 STORAGE (Timeweb Cloud Storage)
# ============================================
s3_endpoint: str = "https://s3.timeweb.com"
s3_bucket: str = "erv-platform-files"
s3_access_key: str = "your_access_key_here"
s3_secret_key: str = "your_secret_key_here"
s3_region: str = "ru-1"
# ============================================
# OCR SERVICE
# ============================================
ocr_api_url: str = "http://147.45.146.17:8001"
ocr_api_key: str = ""
# ============================================
# AI SERVICE (OpenRouter)
# ============================================
openrouter_api_key: str = "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"
openrouter_base_url: str = "https://openrouter.ai/api/v1"
openrouter_model: str = "google/gemini-2.0-flash-001"
# ============================================
# FLIGHT APIs
# ============================================
# FlightAware
flightaware_api_key: str = "Puz0cdxAHzAEqMRZwtdeqBUSm9naJfwK"
flightaware_base_url: str = "https://aeroapi.flightaware.com/aeroapi"
# AviationStack (резервный)
aviationstack_api_key: str = ""
aviationstack_base_url: str = "http://api.aviationstack.com/v1"
# ============================================
# NSPK BANKS API
# ============================================
nspk_banks_api_url: str = "https://qr.nspk.ru/proxyapp/c2bmembers.json"
# ============================================
# SMS SERVICE (SigmaSMS)
# ============================================
sms_api_url: str = "https://online.sigmasms.ru/api/"
sms_login: str = ""
sms_password: str = ""
sms_token: str = ""
sms_sender: str = "lexpriority"
sms_enabled: bool = True
# ============================================
# VTIGER CRM (PHP Bridge)
# ============================================
crm_webservice_url: str = "http://crm.clientright.ru/webservice.php"
crm_webform_url: str = "https://crm.clientright.ru/modules/Webforms/capture.php"
crm_token: str = ""
# ============================================
# RATE LIMITING
# ============================================
rate_limit_per_minute: int = 60
rate_limit_per_hour: int = 1000
# ============================================
# FILE UPLOAD
# ============================================
max_upload_size_mb: int = 50
allowed_file_extensions: str = "pdf,jpg,jpeg,png,heic,heif,webp"
@property
def allowed_extensions_list(self) -> List[str]:
"""Список разрешенных расширений файлов"""
return [ext.strip() for ext in self.allowed_file_extensions.split(",")]
# ============================================
# CORS
cors_origins: list = [
"http://localhost:5173",
"http://147.45.146.17:5173",
"https://erv-claims.clientright.ru"
]
# ============================================
cors_origins: str = "http://localhost:5173,http://147.45.146.17:5173,https://erv-claims.clientright.ru,http://crm.clientright.ru"
@property
def cors_origins_list(self) -> List[str]:
"""Список CORS origins"""
if isinstance(self.cors_origins, str):
return [origin.strip() for origin in self.cors_origins.split(",")]
return self.cors_origins
# ============================================
# LOGGING
# ============================================
log_level: str = "INFO"
log_file: str = "/app/logs/erv_platform.log"
class Config:
env_file = "../.env"
case_sensitive = False
extra = "ignore" # Игнорируем лишние поля из .env
@lru_cache()
@@ -66,3 +166,4 @@ def get_settings() -> Settings:
settings = get_settings()

246
backend/app/main.py
View File

@@ -3,33 +3,84 @@ ERV Insurance Platform - FastAPI Backend
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.config import settings
import redis
import asyncpg
from contextlib import asynccontextmanager
import logging
from .config import settings
from .services.database import db
from .services.redis_service import redis_service
from .services.rabbitmq_service import rabbitmq_service
from .api import sms, claims
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Lifecycle events: startup and shutdown
"""
# STARTUP
logger.info("🚀 Starting ERV Platform...")
try:
# Подключаем PostgreSQL
await db.connect()
except Exception as e:
logger.warning(f"⚠️ PostgreSQL not available: {e}")
try:
# Подключаем Redis
await redis_service.connect()
except Exception as e:
logger.warning(f"⚠️ Redis not available: {e}")
try:
# Подключаем RabbitMQ
await rabbitmq_service.connect()
except Exception as e:
logger.warning(f"⚠️ RabbitMQ not available: {e}")
logger.info("✅ ERV Platform started successfully!")
yield
# SHUTDOWN
logger.info("🛑 Shutting down ERV Platform...")
await db.disconnect()
await redis_service.disconnect()
await rabbitmq_service.disconnect()
logger.info("👋 ERV Platform stopped")
# Создаём FastAPI приложение
app = FastAPI(
title="ERV Insurance Platform API",
description="API для обработки страховых обращений с OCR, AI и интеграциями",
description="API для обработки страховых обращений",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
lifespan=lifespan
)
# CORS middleware
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# API Routes
app.include_router(sms.router)
app.include_router(claims.router)
# ============================================
# HEALTH CHECKS
# ============================================
@app.get("/")
async def root():
@@ -37,140 +88,111 @@ async def root():
return {
"message": "🚀 ERV Insurance Platform API",
"version": "1.0.0",
"docs": f"{settings.backend_url}/docs",
"status": "running"
"status": "running",
"docs": "http://147.45.146.17:8100/docs"
}
@app.get("/health")
async def health_check():
"""Проверка здоровья сервисов"""
async def health():
"""Health check - проверка всех сервисов"""
health_status = {
"api": "ok",
"redis": "checking",
"postgres": "checking",
"ocr": "checking"
"status": "ok",
"message": "API работает!",
"services": {}
}
# Проверка Redis
try:
r = redis.Redis(
host=settings.redis_host,
port=settings.redis_port,
password=settings.redis_password,
decode_responses=True
)
r.ping()
health_status["redis"] = "ok"
except Exception as e:
health_status["redis"] = f"error: {str(e)}"
# Проверка PostgreSQL
try:
conn = await asyncpg.connect(
host=settings.postgres_host,
port=settings.postgres_port,
database=settings.postgres_db,
user=settings.postgres_user,
password=settings.postgres_password
)
await conn.execute("SELECT 1")
await conn.close()
health_status["postgres"] = "ok"
except Exception as e:
health_status["postgres"] = f"error: {str(e)}"
# Проверка OCR
import httpx
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{settings.ocr_api_url}/", timeout=5.0)
health_status["ocr"] = "ok" if response.status_code in [200, 404] else "unreachable"
except Exception as e:
health_status["ocr"] = f"error: {str(e)}"
all_ok = all(v == "ok" for v in health_status.values())
return JSONResponse(
status_code=200 if all_ok else 503,
content={
"status": "healthy" if all_ok else "degraded",
"services": health_status
pg_healthy = await db.health_check()
health_status["services"]["postgresql"] = {
"status": "✅ healthy" if pg_healthy else "❌ unhealthy",
"connected": pg_healthy
}
except:
health_status["services"]["postgresql"] = {
"status": "❌ unavailable",
"connected": False
}
# Проверка Redis
try:
redis_healthy = await redis_service.health_check()
health_status["services"]["redis"] = {
"status": "✅ healthy" if redis_healthy else "❌ unhealthy",
"connected": redis_healthy
}
except:
health_status["services"]["redis"] = {
"status": "❌ unavailable",
"connected": False
}
# Проверка RabbitMQ
try:
rabbitmq_healthy = await rabbitmq_service.health_check()
health_status["services"]["rabbitmq"] = {
"status": "✅ healthy" if rabbitmq_healthy else "❌ unhealthy",
"connected": rabbitmq_healthy
}
except:
health_status["services"]["rabbitmq"] = {
"status": "❌ unavailable",
"connected": False
}
# Общий статус
all_healthy = all(
service.get("connected", False)
for service in health_status["services"].values()
)
if not all_healthy:
health_status["status"] = "degraded"
health_status["message"] = "⚠️ Некоторые сервисы недоступны"
return health_status
# ============================================
# API V1 ENDPOINTS
# ============================================
@app.get("/api/v1/test")
async def test_endpoint():
async def test():
"""Тестовый endpoint"""
return {
"message": "✅ API работает!",
"env": settings.app_env,
"debug": settings.debug,
"success": True,
"message": "✅ Backend API работает!",
"services": {
"redis": f"{settings.redis_host}:{settings.redis_port}",
"postgres": f"{settings.postgres_host}:{settings.postgres_port}",
"rabbitmq": f"{settings.rabbitmq_host}:{settings.rabbitmq_port}",
"ocr": settings.ocr_api_url
"redis": "localhost:6379",
"postgres": "147.45.189.234:5432",
"ocr": "147.45.146.17:8001",
"rabbitmq": "185.197.75.249:5672"
}
}
@app.get("/api/v1/info")
async def get_info():
async def info():
"""Информация о платформе"""
return {
"platform": "ERV Insurance Claims",
"version": "1.0.0",
"features": [
"OCR документов (паспорт, билеты)",
"AI автозаполнение (Gemini Vision)",
"Проверка рейсов (FlightAware)",
"СБП выплаты",
"Интеграция с CRM"
],
"tech_stack": {
"backend": "Python FastAPI",
"frontend": "React TypeScript",
"database": "PostgreSQL + MySQL",
"cache": "Redis",
"queue": "RabbitMQ",
"storage": "S3 Timeweb",
"ocr": "Internal Service",
"ai": "OpenRouter Gemini 2.0"
}
"storage": "S3 Timeweb"
},
"features": [
"OCR документов (паспорт, билеты)",
"AI автозаполнение (Gemini Vision)",
"Проверка рейсов (FlightAware)",
"СБП выплаты",
"Интеграция с CRM Vtiger"
]
}
# ============================================
# STARTUP/SHUTDOWN
# ============================================
@app.on_event("startup")
async def startup_event():
"""При старте приложения"""
print("🚀 ERV Insurance Platform запускается...")
print(f"📍 Backend URL: {settings.backend_url}")
print(f"📍 API Docs: {settings.backend_url}/docs")
print(f"🔗 Frontend URL: {settings.frontend_url}")
@app.on_event("shutdown")
async def shutdown_event():
"""При остановке приложения"""
print("👋 ERV Insurance Platform остановлен")
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8100,
reload=True
)
uvicorn.run(app, host="0.0.0.0", port=8100)

4
backend/app/services/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""
ERV Platform Services
"""

76
backend/app/services/database.py Normal file
View File

@@ -0,0 +1,76 @@
"""
PostgreSQL Database Service
"""
import asyncpg
from typing import Optional, Dict, Any, List
from ..config import settings
import logging
logger = logging.getLogger(__name__)
class DatabaseService:
"""Сервис для работы с PostgreSQL"""
def __init__(self):
self.pool: Optional[asyncpg.Pool] = None
async def connect(self):
"""Создает пул подключений к PostgreSQL"""
try:
self.pool = await asyncpg.create_pool(
host=settings.postgres_host,
port=settings.postgres_port,
database=settings.postgres_db,
user=settings.postgres_user,
password=settings.postgres_password,
min_size=5,
max_size=20,
command_timeout=60
)
logger.info(f"✅ PostgreSQL connected: {settings.postgres_host}:{settings.postgres_port}/{settings.postgres_db}")
except Exception as e:
logger.error(f"❌ PostgreSQL connection error: {e}")
raise
async def disconnect(self):
"""Закрывает пул подключений"""
if self.pool:
await self.pool.close()
logger.info("PostgreSQL pool closed")
async def execute(self, query: str, *args) -> str:
"""Выполняет SQL запрос без возврата данных"""
async with self.pool.acquire() as conn:
return await conn.execute(query, *args)
async def fetch_one(self, query: str, *args) -> Optional[Dict[str, Any]]:
"""Возвращает одну запись"""
async with self.pool.acquire() as conn:
row = await conn.fetchrow(query, *args)
return dict(row) if row else None
async def fetch_all(self, query: str, *args) -> List[Dict[str, Any]]:
"""Возвращает все записи"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(query, *args)
return [dict(row) for row in rows]
async def fetch_val(self, query: str, *args):
"""Возвращает одно значение"""
async with self.pool.acquire() as conn:
return await conn.fetchval(query, *args)
async def health_check(self) -> bool:
"""Проверка здоровья БД"""
try:
result = await self.fetch_val("SELECT 1")
return result == 1
except Exception as e:
logger.error(f"Database health check failed: {e}")
return False
# Глобальный экземпляр
db = DatabaseService()

226
backend/app/services/rabbitmq_service.py Normal file
View File

@@ -0,0 +1,226 @@
"""
RabbitMQ Service для асинхронной обработки задач
"""
import aio_pika
from aio_pika import Connection, Channel, Queue, Exchange, Message
from aio_pika.pool import Pool
from typing import Optional, Callable, Dict, Any
import json
import logging
from ..config import settings
logger = logging.getLogger(__name__)
class RabbitMQService:
"""Сервис для работы с RabbitMQ"""
# Названия очередей
QUEUE_OCR_PROCESSING = "erv_ocr_processing"
QUEUE_AI_EXTRACTION = "erv_ai_extraction"
QUEUE_FLIGHT_CHECK = "erv_flight_check"
QUEUE_CRM_INTEGRATION = "erv_crm_integration"
QUEUE_NOTIFICATIONS = "erv_notifications"
def __init__(self):
self.connection: Optional[Connection] = None
self.channel: Optional[Channel] = None
self.queues: Dict[str, Queue] = {}
async def connect(self):
"""Подключение к RabbitMQ"""
try:
self.connection = await aio_pika.connect_robust(
settings.rabbitmq_url,
timeout=30
)
self.channel = await self.connection.channel()
await self.channel.set_qos(prefetch_count=10)
logger.info(f"✅ RabbitMQ connected: {settings.rabbitmq_host}:{settings.rabbitmq_port}")
# Объявляем очереди
await self._declare_queues()
except Exception as e:
logger.error(f"❌ RabbitMQ connection error: {e}")
raise
async def disconnect(self):
"""Отключение от RabbitMQ"""
if self.connection:
await self.connection.close()
logger.info("RabbitMQ connection closed")
async def _declare_queues(self):
"""Объявляем все рабочие очереди"""
queue_names = [
self.QUEUE_OCR_PROCESSING,
self.QUEUE_AI_EXTRACTION,
self.QUEUE_FLIGHT_CHECK,
self.QUEUE_CRM_INTEGRATION,
self.QUEUE_NOTIFICATIONS,
]
for queue_name in queue_names:
queue = await self.channel.declare_queue(
queue_name,
durable=True, # Очередь переживет перезапуск
arguments={
"x-message-ttl": 3600000, # TTL сообщений 1 час
"x-max-length": 10000, # Максимум сообщений в очереди
}
)
self.queues[queue_name] = queue
logger.info(f"✅ Queue declared: {queue_name}")
async def publish(
self,
queue_name: str,
message: Dict[str, Any],
priority: int = 5,
headers: Optional[Dict[str, Any]] = None
):
"""
Публикация сообщения в очередь
Args:
queue_name: Название очереди
message: Данные сообщения (dict)
priority: Приоритет (0-10, где 10 - максимальный)
headers: Дополнительные заголовки
"""
try:
msg_body = json.dumps(message).encode()
msg = Message(
body=msg_body,
priority=priority,
headers=headers or {},
content_type="application/json",
delivery_mode=aio_pika.DeliveryMode.PERSISTENT # Сохранять на диск
)
# Публикуем в default exchange с routing_key = queue_name
await self.channel.default_exchange.publish(
msg,
routing_key=queue_name
)
logger.debug(f"📤 Message published to {queue_name}: {message.get('task_id', 'unknown')}")
except Exception as e:
logger.error(f"❌ Failed to publish message to {queue_name}: {e}")
raise
async def consume(
self,
queue_name: str,
callback: Callable,
prefetch_count: int = 1
):
"""
Подписка на сообщения из очереди
Args:
queue_name: Название очереди
callback: Асинхронная функция-обработчик
prefetch_count: Количество сообщений для одновременной обработки
"""
try:
queue = self.queues.get(queue_name)
if not queue:
logger.error(f"Queue {queue_name} not found")
return
await self.channel.set_qos(prefetch_count=prefetch_count)
await queue.consume(callback)
logger.info(f"👂 Consuming from {queue_name}")
except Exception as e:
logger.error(f"❌ Failed to consume from {queue_name}: {e}")
raise
async def health_check(self) -> bool:
"""Проверка здоровья RabbitMQ"""
try:
if self.connection and not self.connection.is_closed:
return True
return False
except Exception as e:
logger.error(f"RabbitMQ health check failed: {e}")
return False
# ============================================
# ВСПОМОГАТЕЛЬНЫЕ МЕТОДЫ ДЛЯ ЗАДАЧ
# ============================================
async def publish_ocr_task(self, claim_id: str, file_id: str, file_path: str):
"""Отправка задачи на OCR обработку"""
await self.publish(
self.QUEUE_OCR_PROCESSING,
{
"task_type": "ocr_processing",
"claim_id": claim_id,
"file_id": file_id,
"file_path": file_path
},
priority=8
)
async def publish_ai_extraction_task(self, claim_id: str, file_id: str, ocr_text: str):
"""Отправка задачи на AI извлечение данных"""
await self.publish(
self.QUEUE_AI_EXTRACTION,
{
"task_type": "ai_extraction",
"claim_id": claim_id,
"file_id": file_id,
"ocr_text": ocr_text
},
priority=7
)
async def publish_flight_check_task(self, claim_id: str, flight_number: str, flight_date: str):
"""Отправка задачи на проверку рейса"""
await self.publish(
self.QUEUE_FLIGHT_CHECK,
{
"task_type": "flight_check",
"claim_id": claim_id,
"flight_number": flight_number,
"flight_date": flight_date
},
priority=6
)
async def publish_crm_integration_task(self, claim_id: str, form_data: Dict[str, Any]):
"""Отправка задачи на интеграцию с CRM"""
await self.publish(
self.QUEUE_CRM_INTEGRATION,
{
"task_type": "crm_integration",
"claim_id": claim_id,
"form_data": form_data
},
priority=9 # Высокий приоритет
)
async def publish_notification_task(self, claim_id: str, notification_type: str, data: Dict[str, Any]):
"""Отправка задачи на отправку уведомления"""
await self.publish(
self.QUEUE_NOTIFICATIONS,
{
"task_type": "notification",
"claim_id": claim_id,
"notification_type": notification_type,
"data": data
},
priority=5
)
# Глобальный экземпляр
rabbitmq_service = RabbitMQService()

146
backend/app/services/redis_service.py Normal file
View File

@@ -0,0 +1,146 @@
"""
Redis Service для кеширования, rate limiting, сессий
"""
import redis.asyncio as redis
from typing import Optional, Any
import json
from ..config import settings
import logging
logger = logging.getLogger(__name__)
class RedisService:
"""Сервис для работы с Redis"""
def __init__(self):
self.client: Optional[redis.Redis] = None
async def connect(self):
"""Подключение к Redis"""
try:
self.client = await redis.from_url(
settings.redis_url,
encoding="utf-8",
decode_responses=True
)
await self.client.ping()
logger.info(f"✅ Redis connected: {settings.redis_host}:{settings.redis_port}")
except Exception as e:
logger.error(f"❌ Redis connection error: {e}")
raise
async def disconnect(self):
"""Отключение от Redis"""
if self.client:
await self.client.close()
logger.info("Redis connection closed")
async def get(self, key: str) -> Optional[str]:
"""Получить значение по ключу"""
full_key = f"{settings.redis_prefix}{key}"
return await self.client.get(full_key)
async def set(self, key: str, value: Any, expire: Optional[int] = None):
"""Установить значение с опциональным TTL (в секундах)"""
full_key = f"{settings.redis_prefix}{key}"
if isinstance(value, (dict, list)):
value = json.dumps(value)
if expire:
await self.client.setex(full_key, expire, value)
else:
await self.client.set(full_key, value)
async def delete(self, key: str) -> bool:
"""Удалить ключ"""
full_key = f"{settings.redis_prefix}{key}"
result = await self.client.delete(full_key)
return result > 0
async def exists(self, key: str) -> bool:
"""Проверить существование ключа"""
full_key = f"{settings.redis_prefix}{key}"
return await self.client.exists(full_key) > 0
async def increment(self, key: str, amount: int = 1) -> int:
"""Инкремент значения"""
full_key = f"{settings.redis_prefix}{key}"
return await self.client.incrby(full_key, amount)
async def expire(self, key: str, seconds: int):
"""Установить TTL для ключа"""
full_key = f"{settings.redis_prefix}{key}"
await self.client.expire(full_key, seconds)
async def get_json(self, key: str) -> Optional[dict]:
"""Получить JSON значение"""
value = await self.get(key)
if value:
try:
return json.loads(value)
except json.JSONDecodeError:
return None
return None
async def set_json(self, key: str, value: dict, expire: Optional[int] = None):
"""Установить JSON значение"""
await self.set(key, json.dumps(value), expire)
async def health_check(self) -> bool:
"""Проверка здоровья Redis"""
try:
return await self.client.ping()
except Exception as e:
logger.error(f"Redis health check failed: {e}")
return False
# ============================================
# RATE LIMITING
# ============================================
async def check_rate_limit(self, identifier: str, max_requests: int, window_seconds: int) -> tuple[bool, int]:
"""
Проверка rate limiting
Returns: (allowed: bool, remaining: int)
"""
key = f"ratelimit:{identifier}"
full_key = f"{settings.redis_prefix}{key}"
current = await self.client.get(full_key)
if current is None:
# Первый запрос в окне
await self.client.setex(full_key, window_seconds, 1)
return True, max_requests - 1
current_count = int(current)
if current_count >= max_requests:
# Лимит превышен
ttl = await self.client.ttl(full_key)
return False, 0
# Инкремент счетчика
new_count = await self.client.incr(full_key)
return True, max_requests - new_count
# ============================================
# CACHE
# ============================================
async def cache_get(self, cache_key: str) -> Optional[Any]:
"""Получить из кеша"""
return await self.get_json(f"cache:{cache_key}")
async def cache_set(self, cache_key: str, value: Any, ttl: int = 3600):
"""Сохранить в кеш (TTL по умолчанию 1 час)"""
await self.set_json(f"cache:{cache_key}", value, ttl)
async def cache_delete(self, cache_key: str):
"""Удалить из кеша"""
await self.delete(f"cache:{cache_key}")
# Глобальный экземпляр
redis_service = RedisService()

180
backend/app/services/sms_service.py Normal file
View File

@@ -0,0 +1,180 @@
"""
SMS Service для отправки кодов верификации (SigmaSMS)
"""
import httpx
import random
import logging
from typing import Optional
from ..config import settings
from .redis_service import redis_service
logger = logging.getLogger(__name__)
class SMSService:
"""Сервис для работы с SMS через SigmaSMS API"""
def __init__(self):
self.api_url = settings.sms_api_url
self.login = settings.sms_login
self.password = settings.sms_password
self.token = settings.sms_token
self.sender = settings.sms_sender
self.enabled = settings.sms_enabled
async def _get_token(self) -> Optional[str]:
"""Получить JWT токен для API"""
try:
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.api_url}login",
json={
"username": self.login,
"password": self.password
},
timeout=10.0
)
if response.status_code == 200:
data = response.json()
return data.get("token")
else:
logger.error(f"Failed to get SMS token: {response.status_code}")
return self.token # Используем токен из .env как fallback
except Exception as e:
logger.error(f"Error getting SMS token: {e}")
return self.token
def generate_code(self) -> str:
"""Генерировать 6-значный код"""
return str(random.randint(100000, 999999))
async def send_sms(self, phone: str, message: str) -> bool:
"""
Отправить SMS
Args:
phone: Номер телефона (формат: +79001234567)
message: Текст сообщения
Returns:
bool: True если отправлено успешно
"""
if not self.enabled:
logger.warning("SMS отправка отключена в конфигурации")
return False
try:
# Получаем актуальный токен
token = await self._get_token()
if not token:
logger.error("No SMS token available")
return False
# Очищаем номер телефона
clean_phone = phone.replace("+", "").replace("-", "").replace(" ", "")
# Отправляем SMS
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.api_url}sendings",
headers={
"Authorization": f"Bearer {token}"
},
json={
"recipient": clean_phone,
"type": "sms",
"payload": {
"sender": self.sender,
"text": message
}
},
timeout=15.0
)
if response.status_code in [200, 201]:
logger.info(f"✅ SMS sent to {phone}")
return True
else:
logger.error(f"Failed to send SMS: {response.status_code} - {response.text}")
return False
except Exception as e:
logger.error(f"Error sending SMS: {e}")
return False
async def send_verification_code(self, phone: str) -> Optional[str]:
"""
Отправить код верификации на телефон
Args:
phone: Номер телефона
Returns:
str: Код верификации (для отладки) или None при ошибке
"""
# Проверка rate limiting (не больше 1 SMS в минуту на номер)
rate_limit_key = f"sms_rate:{phone}"
if await redis_service.exists(rate_limit_key):
ttl = await redis_service.client.ttl(f"{settings.redis_prefix}{rate_limit_key}")
logger.warning(f"Rate limit for {phone}, retry in {ttl} seconds")
return None
# Генерируем код
code = self.generate_code()
# Сохраняем код в Redis на 10 минут
verification_key = f"sms_verify:{phone}"
await redis_service.set(verification_key, code, expire=600) # 10 минут
# Устанавливаем rate limit на 60 секунд
await redis_service.set(rate_limit_key, "1", expire=60)
# Формируем сообщение
message = f"Ваш код подтверждения: {code}. Действителен 10 минут."
# Отправляем SMS
success = await self.send_sms(phone, message)
if success:
logger.info(f"Verification code sent to {phone}")
return code # Возвращаем для отладки
else:
# Удаляем код если не удалось отправить
await redis_service.delete(verification_key)
return None
async def verify_code(self, phone: str, code: str) -> bool:
"""
Проверить код верификации
Args:
phone: Номер телефона
code: Код для проверки
Returns:
bool: True если код верный
"""
verification_key = f"sms_verify:{phone}"
stored_code = await redis_service.get(verification_key)
if not stored_code:
logger.warning(f"No verification code found for {phone}")
return False
if stored_code == code:
# Удаляем код после успешной проверки
await redis_service.delete(verification_key)
logger.info(f"✅ Code verified for {phone}")
return True
else:
logger.warning(f"Invalid code for {phone}")
return False
# Глобальный экземпляр
sms_service = SMSService()

306
backend/db/init.sql Normal file
View File

@@ -0,0 +1,306 @@
-- ERV Platform Database Initialization Script
-- PostgreSQL 16+
-- Enable extensions
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS "pg_trgm";
-- ============================================
-- ТАБЛИЦА: claims (Основные заявки)
-- ============================================
CREATE TABLE IF NOT EXISTS claims (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
claim_number VARCHAR(50) UNIQUE NOT NULL,
-- Тип страхования
insurance_type VARCHAR(50) NOT NULL DEFAULT 'erv_travel',
case_type VARCHAR(100), -- например: 'flight_delay', 'medical', 'baggage_loss'
-- Данные клиента
client_phone VARCHAR(20) NOT NULL,
client_email VARCHAR(255),
client_inn VARCHAR(12),
client_full_name VARCHAR(500),
-- Данные полиса
policy_number VARCHAR(100),
policy_series VARCHAR(50),
-- Статус обработки
status VARCHAR(50) NOT NULL DEFAULT 'draft', -- draft, processing, crm_sent, completed, error
crm_id VARCHAR(100), -- ID в Vtiger CRM
-- Данные для аналитики
source VARCHAR(100), -- откуда пришла заявка: 'web_form', 'api', 'mobile_app'
user_agent TEXT,
ip_address INET,
-- Timestamps
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
submitted_at TIMESTAMP WITH TIME ZONE,
completed_at TIMESTAMP WITH TIME ZONE,
-- JSON поля для гибкости
form_data JSONB, -- все данные формы
metadata JSONB -- дополнительные метаданные
);
-- Индексы для claims
CREATE INDEX idx_claims_claim_number ON claims(claim_number);
CREATE INDEX idx_claims_status ON claims(status);
CREATE INDEX idx_claims_created_at ON claims(created_at DESC);
CREATE INDEX idx_claims_client_phone ON claims(client_phone);
CREATE INDEX idx_claims_policy_number ON claims(policy_number);
CREATE INDEX idx_claims_insurance_type ON claims(insurance_type);
CREATE INDEX idx_claims_form_data_gin ON claims USING gin(form_data);
-- ============================================
-- ТАБЛИЦА: claim_files (Файлы к заявкам)
-- ============================================
CREATE TABLE IF NOT EXISTS claim_files (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
claim_id UUID NOT NULL REFERENCES claims(id) ON DELETE CASCADE,
-- Файл
file_name VARCHAR(255) NOT NULL,
file_path VARCHAR(500) NOT NULL,
file_size BIGINT,
mime_type VARCHAR(100),
file_type VARCHAR(50), -- 'passport', 'ticket', 'receipt', 'medical_doc', etc.
-- S3 данные (если используется)
s3_bucket VARCHAR(255),
s3_key VARCHAR(500),
s3_url TEXT,
-- OCR/AI обработка
ocr_status VARCHAR(50), -- 'pending', 'processing', 'completed', 'error'
ocr_text TEXT,
ai_extracted_data JSONB,
-- Timestamps
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
processed_at TIMESTAMP WITH TIME ZONE
);
-- Индексы для claim_files
CREATE INDEX idx_claim_files_claim_id ON claim_files(claim_id);
CREATE INDEX idx_claim_files_file_type ON claim_files(file_type);
CREATE INDEX idx_claim_files_ocr_status ON claim_files(ocr_status);
-- ============================================
-- ТАБЛИЦА: processing_logs (Логи обработки)
-- ============================================
CREATE TABLE IF NOT EXISTS processing_logs (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
claim_id UUID REFERENCES claims(id) ON DELETE CASCADE,
-- Лог
level VARCHAR(20) NOT NULL, -- 'debug', 'info', 'warning', 'error', 'critical'
message TEXT NOT NULL,
context JSONB, -- дополнительный контекст
-- Источник
source VARCHAR(100), -- 'ocr_service', 'ai_service', 'flight_api', 'crm_integration', etc.
function_name VARCHAR(200),
-- Ошибки
error_type VARCHAR(100),
error_traceback TEXT,
-- Timestamp
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
-- Индексы для processing_logs
CREATE INDEX idx_processing_logs_claim_id ON processing_logs(claim_id);
CREATE INDEX idx_processing_logs_level ON processing_logs(level);
CREATE INDEX idx_processing_logs_created_at ON processing_logs(created_at DESC);
CREATE INDEX idx_processing_logs_source ON processing_logs(source);
-- ============================================
-- ТАБЛИЦА: api_calls (Логи внешних API вызовов)
-- ============================================
CREATE TABLE IF NOT EXISTS api_calls (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
claim_id UUID REFERENCES claims(id) ON DELETE SET NULL,
-- API детали
api_name VARCHAR(100) NOT NULL, -- 'ocr_service', 'openrouter_ai', 'flightaware', 'nspk_banks', etc.
endpoint VARCHAR(500),
method VARCHAR(10), -- 'GET', 'POST', etc.
-- Запрос
request_headers JSONB,
request_body JSONB,
-- Ответ
response_status INTEGER,
response_headers JSONB,
response_body JSONB,
-- Производительность
duration_ms INTEGER, -- длительность запроса в миллисекундах
-- Результат
success BOOLEAN DEFAULT FALSE,
error_message TEXT,
-- Timestamp
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
-- Индексы для api_calls
CREATE INDEX idx_api_calls_claim_id ON api_calls(claim_id);
CREATE INDEX idx_api_calls_api_name ON api_calls(api_name);
CREATE INDEX idx_api_calls_success ON api_calls(success);
CREATE INDEX idx_api_calls_created_at ON api_calls(created_at DESC);
-- ============================================
-- ТАБЛИЦА: queue_tasks (Задачи в очереди RabbitMQ)
-- ============================================
CREATE TABLE IF NOT EXISTS queue_tasks (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
claim_id UUID REFERENCES claims(id) ON DELETE CASCADE,
-- Задача
task_type VARCHAR(100) NOT NULL, -- 'ocr_processing', 'ai_extraction', 'flight_check', etc.
queue_name VARCHAR(100) NOT NULL,
priority INTEGER DEFAULT 5, -- 1-10, где 10 - максимальный приоритет
-- Данные
task_data JSONB,
-- Статус
status VARCHAR(50) NOT NULL DEFAULT 'pending', -- pending, processing, completed, failed, retry
attempts INTEGER DEFAULT 0,
max_attempts INTEGER DEFAULT 3,
-- Результат
result JSONB,
error_message TEXT,
-- Timestamps
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
started_at TIMESTAMP WITH TIME ZONE,
completed_at TIMESTAMP WITH TIME ZONE,
next_retry_at TIMESTAMP WITH TIME ZONE
);
-- Индексы для queue_tasks
CREATE INDEX idx_queue_tasks_claim_id ON queue_tasks(claim_id);
CREATE INDEX idx_queue_tasks_status ON queue_tasks(status);
CREATE INDEX idx_queue_tasks_task_type ON queue_tasks(task_type);
CREATE INDEX idx_queue_tasks_created_at ON queue_tasks(created_at DESC);
CREATE INDEX idx_queue_tasks_next_retry ON queue_tasks(next_retry_at) WHERE status = 'retry';
-- ============================================
-- ТАБЛИЦА: metrics (Метрики системы)
-- ============================================
CREATE TABLE IF NOT EXISTS metrics (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
-- Метрика
metric_name VARCHAR(100) NOT NULL,
metric_value NUMERIC(20, 4),
metric_unit VARCHAR(50), -- 'ms', 'count', 'bytes', '%', etc.
-- Теги для группировки
tags JSONB,
-- Timestamp
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
-- Индексы для metrics
CREATE INDEX idx_metrics_metric_name ON metrics(metric_name);
CREATE INDEX idx_metrics_created_at ON metrics(created_at DESC);
CREATE INDEX idx_metrics_tags_gin ON metrics USING gin(tags);
-- ============================================
-- ТАБЛИЦА: cache_entries (Кеш для редких запросов)
-- ============================================
CREATE TABLE IF NOT EXISTS cache_entries (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
-- Ключ-значение
cache_key VARCHAR(255) UNIQUE NOT NULL,
cache_value JSONB,
-- TTL
expires_at TIMESTAMP WITH TIME ZONE,
-- Timestamps
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
-- Индексы для cache_entries
CREATE INDEX idx_cache_entries_key ON cache_entries(cache_key);
CREATE INDEX idx_cache_entries_expires ON cache_entries(expires_at);
-- Автоматическая очистка устаревшего кеша
CREATE OR REPLACE FUNCTION cleanup_expired_cache()
RETURNS void AS $$
BEGIN
DELETE FROM cache_entries WHERE expires_at < CURRENT_TIMESTAMP;
END;
$$ LANGUAGE plpgsql;
-- ============================================
-- ТРИГГЕРЫ
-- ============================================
-- Автоматическое обновление updated_at для claims
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = CURRENT_TIMESTAMP;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER update_claims_updated_at
BEFORE UPDATE ON claims
FOR EACH ROW
EXECUTE FUNCTION update_updated_at_column();
-- ============================================
-- ПРЕДСТАВЛЕНИЯ (VIEWS)
-- ============================================
-- Статистика по заявкам
CREATE OR REPLACE VIEW claims_statistics AS
SELECT
DATE(created_at) as date,
insurance_type,
status,
COUNT(*) as count,
AVG(EXTRACT(EPOCH FROM (completed_at - created_at))) as avg_processing_time_seconds
FROM claims
GROUP BY DATE(created_at), insurance_type, status;
-- Статистика по API вызовам
CREATE OR REPLACE VIEW api_performance AS
SELECT
api_name,
DATE(created_at) as date,
COUNT(*) as total_calls,
SUM(CASE WHEN success THEN 1 ELSE 0 END) as successful_calls,
AVG(duration_ms) as avg_duration_ms,
MAX(duration_ms) as max_duration_ms
FROM api_calls
GROUP BY api_name, DATE(created_at);
-- ============================================
-- НАЧАЛЬНЫЕ ДАННЫЕ
-- ============================================
-- Можно добавить тестовые данные для разработки
-- INSERT INTO claims (claim_number, client_phone, insurance_type, status)
-- VALUES ('TEST-001', '+79001234567', 'erv_travel', 'draft');
-- Завершение
SELECT 'Database initialized successfully!' as message;