commit 068d859f42ae22425027f9d126e4f7c10d66d4ae Author: Ernesto Herrera Date: Mon Mar 2 21:31:50 2026 -0700 first commit - MVE Incrementables Parser microservice with FastAPI, JWT, Celery, Redis diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..a0d71b8 --- /dev/null +++ b/.env.example @@ -0,0 +1,21 @@ +# Authentication +AUTH_USERNAME=admin +# Generate hash: python -c "from passlib.hash import bcrypt; print(bcrypt.hash('your_password'))" +AUTH_PASSWORD_HASH=$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyYqwMzYXhHO +JWT_SECRET=your-secret-key-change-this-in-production +JWT_EXPIRES_MINUTES=60 + +# File Upload +MAX_FILE_MB=10 + +# Logging +LOG_LEVEL=INFO + +# Redis/Celery +REDIS_URL=redis://redis:6379/0 +CELERY_BROKER_URL=redis://redis:6379/0 +CELERY_RESULT_BACKEND=redis://redis:6379/0 + +# Service +SERVICE_NAME=mve-incrementables-parser +SERVICE_VERSION=1.0.0 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..86bddb3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,48 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual Environment +venv/ +env/ +ENV/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# Environment +.env + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# Logs +*.log + +# OS +.DS_Store +Thumbs.db diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..fcd64a7 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,34 @@ +FROM python:3.12-slim + +# Set working directory +WORKDIR /app + +# Install system dependencies for PDF processing +RUN apt-get update && apt-get install -y \ + libmupdf-dev \ + mupdf-tools \ + && rm -rf /var/lib/apt/lists/* + +# Copy requirements +COPY requirements.txt . + +# Install Python dependencies +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY app/ ./app/ + +# Create non-root user +RUN useradd -m -u 1000 appuser && \ + chown -R appuser:appuser /app +USER appuser + +# Expose port +EXPOSE 9876 + +# Health check +HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \ + CMD python -c "import requests; requests.get('http://localhost:9876/health')" || exit 1 + +# Run the application +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "9876"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..b03dc41 --- /dev/null +++ b/README.md @@ -0,0 +1,511 @@ +# MVE Incrementables Parser + +Microservicio para extraer y parsear la sección de **Incrementables** de documentos PDF tipo "CARTA INS". Construido con **FastAPI** y **Python 3.12**. + +## 🚀 Características + +- ✅ **Autenticación JWT** con bcrypt +- ✅ **Extracción de texto** con PyMuPDF y pdfplumber (fallback) +- ✅ **Parsing robusto** de incrementables (fletes, seguros, almacenaje, regalías) +- ✅ **Validación de archivos** (tamaño, tipo MIME, PDF cifrado) +- ✅ **Logging estructurado** con correlation ID +- ✅ **Rate limiting** en login +- ✅ **Procesamiento asíncrono** con Celery y Redis +- ✅ **Cola de tareas** para procesamiento en background +- ✅ **Tests con pytest** +- ✅ **Docker & Docker Compose** + +## 📋 Requisitos + +- Python 3.12+ +- Docker & Docker Compose (opcional) + +## 🛠️ Instalación + +### Opción 1: Con Docker (Recomendado) + +1. **Clonar el repositorio** +```bash +cd mve-incrementables-parser +``` + +2. **Configurar variables de entorno** +```bash +cp .env.example .env +``` + +Edita `.env` y configura: +```env +AUTH_USERNAME=admin +AUTH_PASSWORD_HASH=$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/LewY5GyYqwMzYXhHO +JWT_SECRET=your-secret-key-change-this-in-production +JWT_EXPIRES_MINUTES=60 +MAX_FILE_MB=10 +LOG_LEVEL=INFO +``` + +**Generar hash de contraseña:** +```bash +python -c "from passlib.hash import bcrypt; print(bcrypt.hash('tu_contraseña'))" +``` + +3. **Levantar el servicio** +```bash +docker-compose up --build +``` + +El servicio estará disponible en `http://localhost:9876` + +**Servicios incluidos:** +- API FastAPI: puerto 9876 +- Redis: puerto 16379 +- Celery Worker: procesamiento en background + +### Opción 2: Sin Docker + +1. **Instalar dependencias del sistema** (para PyMuPDF) +```bash +# macOS +brew install mupdf-tools + +# Ubuntu/Debian +sudo apt-get install libmupdf-dev mupdf-tools +``` + +2. **Crear entorno virtual** +```bash +python3.12 -m venv venv +source venv/bin/activate # En Windows: venv\Scripts\activate +``` + +3. **Instalar dependencias Python** +```bash +pip install -r requirements.txt +``` + +4. **Configurar .env** (ver Opción 1, paso 2) + +5. **Ejecutar el servicio** +```bash +uvicorn app.main:app --host 0.0.0.0 --port 9876 --reload +``` + +## 📚 Uso + +### 1. Health Check + +```bash +curl http://localhost:9876/health +``` + +**Respuesta:** +```json +{ + "status": "ok", + "service": "mve-incrementables-parser", + "version": "1.0.0" +} +``` + +### 2. Autenticación (Login) + +```bash +curl -X POST http://localhost:9876/auth/login \ + -H "Content-Type: application/json" \ + -d '{ + "username": "admin", + "password": "tu_contraseña" + }' +``` + +**Respuesta:** +```json +{ + "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...", + "token_type": "bearer", + "expires_in": 3600 +} +``` + +⚠️ **Nota:** Guarda el `access_token` para usarlo en las siguientes peticiones. + +### 3. Parsear PDF (Síncrono) + +```bash +curl -X POST http://localhost:9876/v1/incrementables/parse \ + -H "Authorization: Bearer " \ + -F "file=@CARTA_INS_228718.pdf" \ + -F "document_ref=INS-228718" +``` + +**Con Correlation ID personalizado:** +```bash +curl -X POST http://localhost:9876/v1/incrementables/parse \ + -H "Authorization: Bearer " \ + -H "X-Correlation-Id: custom-id-12345" \ + -F "file=@CARTA_INS_228718.pdf" +``` + +**Respuesta exitosa (200):** +```json +{ + "correlation_id": "550e8400-e29b-41d4-a716-446655440000", + "document": { + "filename": "CARTA INS 228718.pdf", + "pages": 1, + "sha256": "a3d5f..." + }, + "incrementables": { + "currency": "USD", + "fletes": 1591.20, + "seguros": null, + "almacenaje_consolidacion": 0.00, + "regalias": null + }, + "extraction": { + "method": "text", + "anchors_found": ["AJUSTE DE INCREMENTABLES EN:"], + "warnings": [] + } +} +``` + +### 4. Parsear PDF (Asíncrono con Cola) + +Para archivos grandes o cuando no quieres esperar, usa el endpoint asíncrono: + +```bash +curl -X POST http://localhost:9876/v1/incrementables/parse/async \ + -H "Authorization: Bearer " \ + -F "file=@CARTA_INS_228718.pdf" \ + -F "document_ref=INS-228718" +``` + +**Respuesta (202):** +```json +{ + "task_id": "a8f2e9c1-5b3d-4e7f-9a1c-2d3e4f5a6b7c", + "correlation_id": "550e8400-e29b-41d4-a716-446655440000", + "status": "queued", + "message": "Task queued for processing. Use /v1/incrementables/status/{task_id} to check progress." +} +``` + +### 5. Consultar Estado de Tarea + +```bash +curl -X GET http://localhost:9876/v1/incrementables/status/a8f2e9c1-5b3d-4e7f-9a1c-2d3e4f5a6b7c \ + -H "Authorization: Bearer " +``` + +**Respuesta (en proceso):** +```json +{ + "task_id": "a8f2e9c1-5b3d-4e7f-9a1c-2d3e4f5a6b7c", + "status": "started", + "result": null, + "error": null +} +``` + +**Respuesta (completada):** +```json +{ + "task_id": "a8f2e9c1-5b3d-4e7f-9a1c-2d3e4f5a6b7c", + "status": "completed", + "result": { + "status": "completed", + "task_id": "a8f2e9c1-5b3d-4e7f-9a1c-2d3e4f5a6b7c", + "document": { + "filename": "CARTA INS 228718.pdf", + "pages": 1, + "sha256": "a3d5f...", + "document_ref": "INS-228718" + }, + "incrementables": { + "currency": "USD", + "fletes": 1591.20, + "seguros": null, + "almacenaje_consolidacion": 0.00, + "regalias": null + }, + "extraction": { + "method": "text", + "anchors_found": ["AJUSTE DE INCREMENTABLES EN:"], + "warnings": [] + } + }, + "error": null +} +``` + +**Errores comunes:** + +- **400**: Archivo no válido, no es PDF, excede tamaño máximo +- **401**: Token ausente, inválido o expirado +- **422**: No se encontró sección de incrementables o parsing falló +- **500**: Error interno del servidor + +## 🔄 Procesamiento Asíncrono + +El servicio soporta dos modos de procesamiento: + +### Modo Síncrono (`/parse`) +- Respuesta inmediata +- Ideal para PDFs pequeños +- Timeout en request HTTP + +### Modo Asíncrono (`/parse/async`) +- Respuesta inmediata con task_id +- Procesamiento en background con Celery +- Ideal para PDFs grandes o lotes +- Sin timeout (límite: 5 minutos por tarea) +- Consulta estado con `/status/{task_id}` + +### Arquitectura +``` +Cliente → FastAPI → Redis (Cola) → Celery Worker → Procesa PDF → Redis (Resultado) + ↓ +Cliente ← FastAPI ← Redis (Consulta) ←──────────────────────────────── +``` + +## 🧪 Tests + +```bash +# Ejecutar todos los tests +pytest + +# Con cobertura +pytest --cov=app --cov-report=html + +# Test específico +pytest tests/test_parser.py -v + +# Ver logs detallados +pytest -v -s +``` + +## 🏗️ Estructura del Proyecto + +``` +mve-incrementables-parser/ +├── app/ +│ ├── __init__.py +│ ├── main.py # FastAPI app principal +│ ├── schemas.py # Modelos Pydantic +│ ├── core/ +│ │ ├── config.py # Configuración (ENV) +│ │ ├── security.py # JWT, bcrypt, rate limiting +│ │ └── celery_app.py # Configuración de Celery +│ ├── api/ +│ │ ├── auth.py # Endpoint de login +│ │ └── v1/ +│ │ └── incrementables.py # Endpoints de parsing (sync/async) +│ ├── tasks/ +│ │ └── parse_tasks.py # Tareas de Celery +│ └── services/ +│ ├── pdf_text.py # Extracción de texto (PyMuPDF/pdfplumber) +│ └── parser.py # Parsing de incrementables +├── tests/ +│ ├── conftest.py # Fixtures de pytest +│ ├── test_parser.py # Tests del parser +│ └── test_api.py # Tests de endpoints +├── .env.example # Ejemplo de variables de entorno +├── .gitignore +├── Dockerfile +├── docker-compose.yml # API + Redis + Celery Worker +├── pytest.ini +├── requirements.txt +└── README.md +``` +│ │ └── v1/ +│ │ └── incrementables.py # Endpoint de parsing +│ └── services/ +│ ├── pdf_text.py # Extracción de texto (PyMuPDF/pdfplumber) +│ └── parser.py # Parsing de incrementables +├── tests/ +│ ├── conftest.py # Fixtures de pytest +│ ├── test_parser.py # Tests del parser +│ └── test_api.py # Tests de endpoints +├── .env.example # Ejemplo de variables de entorno +├── .gitignore +├── Dockerfile +├── docker-compose.yml +├── pytest.ini +├── requirements.txt +└── README.md +``` + +## 🔒 Seguridad + +### Autenticación JWT + +- **Login** con username/password → devuelve JWT +- **Protección** de endpoints `/v1/*` con Bearer token +- **Expiración** configurable (default: 60 minutos) +- **Rate limiting** básico en login (5 intentos / 5 minutos por IP) + +### Contraseñas + +- **Nunca** se loguean contraseñas en texto plano +- Uso de **bcrypt** para hashing +- Respuestas **uniformes** (401) en error de autenticación + +### Archivos + +- Validación de **tipo MIME** +- Validación de **extensión** `.pdf` +- **Tamaño máximo** configurable (default: 10 MB) +- Rechazo de **PDFs cifrados** +- Cálculo de **SHA256** para integridad + +## ⚙️ Configuración (Variables de Entorno) + +| Variable | Descripción | Default | +|----------|-------------|---------| +| `AUTH_USERNAME` | Usuario para login | `admin` | +| `AUTH_PASSWORD_HASH` | Hash bcrypt de contraseña | *requerido* | +| `JWT_SECRET` | Secret para firmar JWT | *requerido* | +| `JWT_EXPIRES_MINUTES` | Duración del token (minutos) | `60` | +| `MAX_FILE_MB` | Tamaño máximo de PDF (MB) | `10` | +| `LOG_LEVEL` | Nivel de logging | `INFO` | +| `REDIS_URL` | URL de conexión a Redis | `redis://redis:6379/0` | +| `CELERY_BROKER_URL` | URL del broker de Celery | `redis://redis:6379/0` | +| `CELERY_RESULT_BACKEND` | URL del backend de resultados | `redis://redis:6379/0` | +| `SERVICE_NAME` | Nombre del servicio | `mve-incrementables-parser` | +| `SERVICE_VERSION` | Versión del servicio | `1.0.0` | + +## 📖 API Documentation + +Una vez levantado el servicio, accede a la documentación interactiva: + +- **Swagger UI**: http://localhost:9876/docs +- **ReDoc**: http://localhost:9876/redoc + +## 🔍 Logging + +El servicio utiliza **logging estructurado en JSON**: + +```json +{ + "time": "2026-03-02T10:30:45", + "level": "INFO", + "name": "app.api.v1.incrementables", + "message": "Successfully parsed incrementables", + "correlation_id": "550e8400-e29b-41d4-a716-446655440000", + "currency": "USD", + "fletes": 1591.20 +} +``` + +### Correlation ID + +- **Automático**: Se genera UUID si no se proporciona +- **Manual**: Enviar header `X-Correlation-Id` +- **Propagación**: Aparece en logs y respuesta +- **Utilidad**: Tracking de requests en sistemas distribuidos + +## 🐛 Troubleshooting + +### Error: "PyMuPDF extraction failed" + +**Causa**: PDF corrupto o cifrado + +**Solución**: El servicio intentará con pdfplumber automáticamente. Si ambos fallan: +- Verificar que el PDF no esté cifrado +- Intentar con otro PDF +- Revisar logs para detalles + +### Error: "Incrementables section not found" + +**Causa**: El PDF no contiene la sección esperada o el formato es diferente + +**Solución**: +- Verificar que el PDF tenga el texto "AJUSTE DE INCREMENTABLES EN:" +- Revisar el formato del documento +- Si es un formato nuevo, actualizar los patrones en `app/services/parser.py` + +### Error: "Rate limit exceeded" + +**Causa**: Demasiados intentos de login desde la misma IP + +**Solución**: Esperar 5 minutos o reiniciar el servicio + +## 🚀 Despliegue en Producción + +### Recomendaciones + +1. **Cambiar credenciales**: Generar nueva contraseña y JWT_SECRET +2. **HTTPS**: Usar reverse proxy (nginx, traefik) +3. **Límites**: Ajustar `MAX_FILE_MB` según necesidad +4. **Logging**: Integrar con sistema de logs centralizado +5. **Monitoring**: Configurar health checks y alertas +6. **Recursos**: Ajustar memoria/CPU del contenedor + +### Ejemplo con nginx + +```nginx +server { + listen 443 ssl; + server_name api.mve.com; + + ssl_certificate /path/to/cert.pem; + ssl_certificate_key /path/to/key.pem; + + location / { + proxy_pass http://localhost:9876; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # Aumentar límites para PDFs grandes + client_max_body_size 20M; + } +} +``` + +## 📝 Ejemplos de Formato Esperado + +El servicio espera PDFs con formato similar a: + +``` +CARTA INSTRUCTIVO DE EMBARQUE No. 228718 + +AJUSTE DE INCREMENTABLES EN: + +Fletes: $1,591.20 USD +Seguros: USD +Almacenaje/Consolidación: $0.00 USD +Regalías: USD + +Observaciones... +``` + +### Campos Reconocidos + +- **Fletes**: Requerido, con o sin signo `$`, con comas +- **Seguros**: Opcional, puede estar vacío (solo "USD" → `null`) +- **Almacenaje/Consolidación**: Requerido, acepta variaciones de escritura +- **Regalías/Regalias**: Opcional, con o sin acento + +## 🤝 Contribuir + +1. Fork el proyecto +2. Crear rama feature (`git checkout -b feature/amazing-feature`) +3. Commit cambios (`git commit -m 'Add amazing feature'`) +4. Push a la rama (`git push origin feature/amazing-feature`) +5. Abrir Pull Request + +## 📄 Licencia + +Este proyecto es parte del sistema MVE y es de uso interno. + +## 📧 Contacto + +Para soporte o dudas, contactar al equipo de desarrollo MVE. + +--- + +**Versión:** 1.0.0 +**Última actualización:** Marzo 2026 diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..435c52f --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,2 @@ +# MVE Incrementables Parser Service +__version__ = "1.0.0" diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/auth.py b/app/api/auth.py new file mode 100644 index 0000000..23c6d6d --- /dev/null +++ b/app/api/auth.py @@ -0,0 +1,61 @@ +"""Authentication endpoints.""" +from fastapi import APIRouter, HTTPException, status, Request +from datetime import timedelta +import logging + +from app.core.config import get_settings +from app.core.security import authenticate_user, create_access_token, check_rate_limit +from app.schemas import LoginRequest, LoginResponse + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/auth", tags=["Authentication"]) + + +@router.post("/login", response_model=LoginResponse) +async def login(request: Request, credentials: LoginRequest): + """ + Authenticate user and return JWT token. + + - **username**: Username + - **password**: Password + + Returns JWT access token with expiration time. + """ + settings = get_settings() + + # Get client IP for rate limiting + client_ip = request.client.host if request.client else "unknown" + + # Check rate limit + if check_rate_limit(client_ip): + logger.warning(f"Rate limit exceeded for IP: {client_ip}") + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail="Too many login attempts. Please try again later." + ) + + # Authenticate (never log the password) + logger.info(f"Login attempt for user: {credentials.username} from IP: {client_ip}") + + if not authenticate_user(credentials.username, credentials.password): + logger.warning(f"Failed login attempt for user: {credentials.username}") + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + + # Create access token + access_token_expires = timedelta(minutes=settings.jwt_expires_minutes) + access_token = create_access_token( + data={"sub": credentials.username}, + expires_delta=access_token_expires + ) + + logger.info(f"Successful login for user: {credentials.username}") + + return LoginResponse( + access_token=access_token, + token_type="bearer", + expires_in=settings.jwt_expires_minutes * 60 # Convert to seconds + ) diff --git a/app/api/v1/__init__.py b/app/api/v1/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/v1/incrementables.py b/app/api/v1/incrementables.py new file mode 100644 index 0000000..16fa0be --- /dev/null +++ b/app/api/v1/incrementables.py @@ -0,0 +1,322 @@ +"""Incrementables parsing endpoints.""" +from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Header, status +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials +from celery.result import AsyncResult +import hashlib +import logging +import uuid +from typing import Optional + +from app.core.config import get_settings +from app.core.security import decode_access_token +from app.core.celery_app import celery_app +from app.tasks.parse_tasks import parse_pdf_task +from app.services.pdf_text import extract_text_from_pdf, PDFExtractionError +from app.services.parser import parse_incrementables, ParsingError +from app.schemas import ( + ParseResponse, ParseAsyncResponse, TaskStatusResponse, + DocumentInfo, IncrementablesData, ExtractionInfo +) + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/v1/incrementables", tags=["Incrementables"]) +security = HTTPBearer() + + +def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str: + """ + Dependency to validate JWT token and extract username. + """ + payload = decode_access_token(credentials.credentials) + username = payload.get("sub") + + if username is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid credentials" + ) + + return username + + +@router.post("/parse", response_model=ParseResponse) +async def parse_pdf( + file: UploadFile = File(..., description="PDF file to parse"), + document_ref: Optional[str] = Form(None, description="Optional document reference or folio"), + x_correlation_id: Optional[str] = Header(None), + current_user: str = Depends(get_current_user) +): + """ + Parse incrementables section from uploaded PDF. + + - **file**: PDF file (required) + - **document_ref**: Optional document reference or folio + + Returns parsed incrementables data with metadata. + + **Authentication required**: Bearer token in Authorization header. + """ + settings = get_settings() + + # Generate or use correlation ID + correlation_id = x_correlation_id or str(uuid.uuid4()) + + logger.info( + f"Parse request received", + extra={ + "correlation_id": correlation_id, + "pdf_filename": file.filename, + "document_ref": document_ref, + "user": current_user + } + ) + + # Validate file type + if not file.filename.lower().endswith('.pdf'): + logger.warning(f"Invalid file type: {file.filename}", extra={"correlation_id": correlation_id}) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Only PDF files are accepted" + ) + + # Validate content type + if file.content_type not in ["application/pdf", "application/x-pdf"]: + logger.warning( + f"Invalid content type: {file.content_type}", + extra={"correlation_id": correlation_id} + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid content type. Expected application/pdf, got {file.content_type}" + ) + + # Read file content + try: + pdf_bytes = await file.read() + except Exception as e: + logger.error(f"Failed to read file: {str(e)}", extra={"correlation_id": correlation_id}) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to read uploaded file" + ) + + # Validate file size + file_size_mb = len(pdf_bytes) / (1024 * 1024) + if file_size_mb > settings.max_file_mb: + logger.warning( + f"File too large: {file_size_mb:.2f} MB", + extra={"correlation_id": correlation_id} + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"File size exceeds maximum allowed size of {settings.max_file_mb} MB" + ) + + # Calculate SHA256 hash + file_hash = hashlib.sha256(pdf_bytes).hexdigest() + + # Extract text from PDF + try: + text, page_count, extraction_method = extract_text_from_pdf(pdf_bytes) + logger.info( + f"Text extracted: {len(text)} characters, {page_count} pages", + extra={"correlation_id": correlation_id} + ) + except PDFExtractionError as e: + logger.error( + f"PDF extraction failed: {str(e)}", + extra={"correlation_id": correlation_id} + ) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Failed to extract text from PDF: {str(e)}" + ) + + # Parse incrementables + try: + parsed_data = parse_incrementables(text) + logger.info( + f"Successfully parsed incrementables", + extra={ + "correlation_id": correlation_id, + "currency": parsed_data["currency"], + "fletes": parsed_data["fletes"] + } + ) + except ParsingError as e: + logger.error( + f"Parsing failed: {str(e)}", + extra={"correlation_id": correlation_id} + ) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Failed to parse incrementables section: {str(e)}" + ) + + # Build response + response = ParseResponse( + correlation_id=correlation_id, + document=DocumentInfo( + filename=file.filename, + pages=page_count, + sha256=file_hash + ), + incrementables=IncrementablesData( + currency=parsed_data["currency"], + fletes=parsed_data["fletes"], + seguros=parsed_data["seguros"], + almacenaje_consolidacion=parsed_data["almacenaje_consolidacion"], + regalias=parsed_data["regalias"] + ), + extraction=ExtractionInfo( + method=extraction_method, + anchors_found=parsed_data["anchors_found"], + warnings=parsed_data["warnings"] + ) + ) + + return response + + +@router.post("/parse/async", response_model=ParseAsyncResponse) +async def parse_pdf_async( + file: UploadFile = File(..., description="PDF file to parse"), + document_ref: Optional[str] = Form(None, description="Optional document reference or folio"), + x_correlation_id: Optional[str] = Header(None), + current_user: str = Depends(get_current_user) +): + """ + Parse incrementables section from uploaded PDF asynchronously using Celery. + + - **file**: PDF file (required) + - **document_ref**: Optional document reference or folio + + Returns task ID for checking status later. + + **Authentication required**: Bearer token in Authorization header. + """ + settings = get_settings() + + # Generate or use correlation ID + correlation_id = x_correlation_id or str(uuid.uuid4()) + + logger.info( + f"Async parse request received", + extra={ + "correlation_id": correlation_id, + "pdf_filename": file.filename, + "document_ref": document_ref, + "user": current_user + } + ) + + # Validate file type + if not file.filename.lower().endswith('.pdf'): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Only PDF files are accepted" + ) + + # Validate content type + if file.content_type not in ["application/pdf", "application/x-pdf"]: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid content type. Expected application/pdf, got {file.content_type}" + ) + + # Read file content + try: + pdf_bytes = await file.read() + except Exception as e: + logger.error(f"Failed to read file: {str(e)}", extra={"correlation_id": correlation_id}) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to read uploaded file" + ) + + # Validate file size + file_size_mb = len(pdf_bytes) / (1024 * 1024) + if file_size_mb > settings.max_file_mb: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"File size exceeds maximum allowed size of {settings.max_file_mb} MB" + ) + + # Convert bytes to hex for serialization + pdf_hex = pdf_bytes.hex() + + # Queue task + task = parse_pdf_task.delay(pdf_hex, file.filename, document_ref) + + logger.info( + f"Task queued: {task.id}", + extra={"correlation_id": correlation_id, "task_id": task.id} + ) + + return ParseAsyncResponse( + task_id=task.id, + correlation_id=correlation_id, + status="queued", + message="Task queued for processing. Use /v1/incrementables/status/{task_id} to check progress." + ) + + +@router.get("/status/{task_id}", response_model=TaskStatusResponse) +async def get_task_status( + task_id: str, + current_user: str = Depends(get_current_user) +): + """ + Get status of an async parsing task. + + - **task_id**: Task ID returned from /parse/async + + Returns task status and result if completed. + + **Authentication required**: Bearer token in Authorization header. + """ + task_result = AsyncResult(task_id, app=celery_app) + + if task_result.state == "PENDING": + return TaskStatusResponse( + task_id=task_id, + status="pending", + result=None, + error=None + ) + elif task_result.state == "STARTED": + return TaskStatusResponse( + task_id=task_id, + status="started", + result=None, + error=None + ) + elif task_result.state == "SUCCESS": + result_data = task_result.result + if result_data.get("status") == "failed": + return TaskStatusResponse( + task_id=task_id, + status="failed", + result=None, + error=result_data.get("message", "Unknown error") + ) + return TaskStatusResponse( + task_id=task_id, + status="completed", + result=result_data, + error=None + ) + elif task_result.state == "FAILURE": + return TaskStatusResponse( + task_id=task_id, + status="failed", + result=None, + error=str(task_result.info) + ) + else: + return TaskStatusResponse( + task_id=task_id, + status=task_result.state.lower(), + result=None, + error=None + ) diff --git a/app/core/__init__.py b/app/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/core/celery_app.py b/app/core/celery_app.py new file mode 100644 index 0000000..2492966 --- /dev/null +++ b/app/core/celery_app.py @@ -0,0 +1,25 @@ +"""Celery application configuration.""" +from celery import Celery +from .config import get_settings + +settings = get_settings() + +celery_app = Celery( + "mve_incrementables_parser", + broker=settings.celery_broker_url, + backend=settings.celery_result_backend, + include=["app.tasks.parse_tasks"] +) + +# Celery configuration +celery_app.conf.update( + task_serializer="json", + accept_content=["json"], + result_serializer="json", + timezone="UTC", + enable_utc=True, + task_track_started=True, + task_time_limit=300, # 5 minutes max + task_soft_time_limit=240, # 4 minutes soft limit + result_expires=3600, # Results expire after 1 hour +) diff --git a/app/core/config.py b/app/core/config.py new file mode 100644 index 0000000..8f734e6 --- /dev/null +++ b/app/core/config.py @@ -0,0 +1,39 @@ +"""Configuration management using Pydantic Settings.""" +from pydantic_settings import BaseSettings +from functools import lru_cache + + +class Settings(BaseSettings): + """Application settings loaded from environment variables.""" + + # Service Info + service_name: str = "mve-incrementables-parser" + service_version: str = "1.0.0" + + # Authentication + auth_username: str + auth_password_hash: str + jwt_secret: str + jwt_expires_minutes: int = 60 + jwt_algorithm: str = "HS256" + + # File Upload + max_file_mb: int = 10 + + # Logging + log_level: str = "INFO" + + # Redis/Celery + redis_url: str = "redis://localhost:6379/0" + celery_broker_url: str = "redis://localhost:6379/0" + celery_result_backend: str = "redis://localhost:6379/0" + + class Config: + env_file = ".env" + case_sensitive = False + + +@lru_cache() +def get_settings() -> Settings: + """Get cached settings instance.""" + return Settings() diff --git a/app/core/security.py b/app/core/security.py new file mode 100644 index 0000000..b11a18d --- /dev/null +++ b/app/core/security.py @@ -0,0 +1,99 @@ +"""Security utilities for authentication and JWT token management.""" +from datetime import datetime, timedelta +from typing import Optional +from jose import JWTError, jwt +import bcrypt +from fastapi import HTTPException, status +from .config import get_settings +import time +from collections import defaultdict + +# Simple in-memory rate limiter for login attempts +login_attempts = defaultdict(list) +MAX_LOGIN_ATTEMPTS = 5 +LOGIN_WINDOW_SECONDS = 300 # 5 minutes + + +def verify_password(plain_password: str, hashed_password: str) -> bool: + """Verify a password against its hash.""" + try: + return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) + except Exception as e: + return False + + +def get_password_hash(password: str) -> str: + """Generate password hash.""" + return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') + + +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: + """Create JWT access token.""" + settings = get_settings() + to_encode = data.copy() + + if expires_delta: + expire = datetime.utcnow() + expires_delta + else: + expire = datetime.utcnow() + timedelta(minutes=settings.jwt_expires_minutes) + + to_encode.update({"exp": expire, "iat": datetime.utcnow()}) + encoded_jwt = jwt.encode( + to_encode, + settings.jwt_secret, + algorithm=settings.jwt_algorithm + ) + return encoded_jwt + + +def decode_access_token(token: str) -> dict: + """Decode and verify JWT token.""" + settings = get_settings() + try: + payload = jwt.decode( + token, + settings.jwt_secret, + algorithms=[settings.jwt_algorithm] + ) + return payload + except JWTError: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + + +def check_rate_limit(identifier: str) -> bool: + """ + Simple rate limiter for login attempts. + Returns True if rate limit exceeded. + """ + current_time = time.time() + + # Clean old attempts + login_attempts[identifier] = [ + attempt_time for attempt_time in login_attempts[identifier] + if current_time - attempt_time < LOGIN_WINDOW_SECONDS + ] + + # Check if limit exceeded + if len(login_attempts[identifier]) >= MAX_LOGIN_ATTEMPTS: + return True + + # Record this attempt + login_attempts[identifier].append(current_time) + return False + + +def authenticate_user(username: str, password: str) -> bool: + """Authenticate user with username and password.""" + settings = get_settings() + + if username != settings.auth_username: + return False + + if not verify_password(password, settings.auth_password_hash): + return False + + return True diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..eff92ed --- /dev/null +++ b/app/main.py @@ -0,0 +1,90 @@ +"""Main FastAPI application.""" +import logging +import sys +from fastapi import FastAPI, Request, status +from fastapi.responses import JSONResponse +from fastapi.exceptions import RequestValidationError +from contextlib import asynccontextmanager + +from app.core.config import get_settings +from app.api import auth +from app.api.v1 import incrementables +from app.schemas import HealthResponse + +# Configure logging +settings = get_settings() +logging.basicConfig( + level=getattr(logging, settings.log_level.upper()), + format='{"time": "%(asctime)s", "level": "%(levelname)s", "name": "%(name)s", "message": "%(message)s"}', + stream=sys.stdout +) + +logger = logging.getLogger(__name__) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Application lifespan manager.""" + logger.info(f"Starting {settings.service_name} v{settings.service_version}") + yield + logger.info(f"Shutting down {settings.service_name}") + + +# Create FastAPI app +app = FastAPI( + title=settings.service_name, + version=settings.service_version, + description="MVE Incrementables Parser - Extracts incrementables data from PDFs", + lifespan=lifespan +) + + +# Exception handlers +@app.exception_handler(RequestValidationError) +async def validation_exception_handler(request: Request, exc: RequestValidationError): + """Handle validation errors with custom format.""" + logger.warning(f"Validation error: {exc.errors()}") + return JSONResponse( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + content={ + "detail": "Validation error", + "errors": exc.errors() + } + ) + + +@app.exception_handler(Exception) +async def general_exception_handler(request: Request, exc: Exception): + """Handle unexpected errors.""" + logger.error(f"Unexpected error: {str(exc)}", exc_info=True) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content={ + "detail": "Internal server error" + } + ) + + +# Health check endpoint +@app.get("/health", response_model=HealthResponse, tags=["Health"]) +async def health_check(): + """ + Health check endpoint. + + Returns service status and version information. + """ + return HealthResponse( + status="ok", + service=settings.service_name, + version=settings.service_version + ) + + +# Include routers +app.include_router(auth.router) +app.include_router(incrementables.router) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=9876) diff --git a/app/schemas.py b/app/schemas.py new file mode 100644 index 0000000..8c3f40a --- /dev/null +++ b/app/schemas.py @@ -0,0 +1,81 @@ +"""Pydantic schemas for request/response validation.""" +from pydantic import BaseModel, Field +from typing import Optional, List +from datetime import datetime + + +# Authentication Schemas +class LoginRequest(BaseModel): + """Login request schema.""" + username: str = Field(..., min_length=1) + password: str = Field(..., min_length=1) + + +class LoginResponse(BaseModel): + """Login response schema.""" + access_token: str + token_type: str = "bearer" + expires_in: int + + +# Health Check Schema +class HealthResponse(BaseModel): + """Health check response.""" + status: str = "ok" + service: str + version: str + + +# Incrementables Schemas +class DocumentInfo(BaseModel): + """PDF document metadata.""" + filename: str + pages: int + sha256: str + + +class IncrementablesData(BaseModel): + """Parsed incrementables data.""" + currency: str + fletes: float + seguros: Optional[float] = None + almacenaje_consolidacion: float + regalias: Optional[float] = None + + +class ExtractionInfo(BaseModel): + """Information about the extraction process.""" + method: str # "text", "ocr" (future) + anchors_found: List[str] + warnings: List[str] = [] + + +class ParseResponse(BaseModel): + """Parse endpoint response.""" + correlation_id: str + document: DocumentInfo + incrementables: IncrementablesData + extraction: ExtractionInfo + + +# Async/Queue schemas +class ParseAsyncResponse(BaseModel): + """Async parse endpoint response.""" + task_id: str + correlation_id: str + status: str = "queued" + message: str = "Task queued for processing" + + +class TaskStatusResponse(BaseModel): + """Task status response.""" + task_id: str + status: str # pending, started, completed, failed + result: Optional[dict] = None + error: Optional[str] = None + + +class ErrorResponse(BaseModel): + """Error response schema.""" + detail: str + correlation_id: Optional[str] = None diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/parser.py b/app/services/parser.py new file mode 100644 index 0000000..f9ec10e --- /dev/null +++ b/app/services/parser.py @@ -0,0 +1,211 @@ +"""Parser service for extracting incrementables data from PDF text.""" +import re +import logging +from typing import Dict, Optional, List, Tuple +from decimal import Decimal + +logger = logging.getLogger(__name__) + + +class ParsingError(Exception): + """Custom exception for parsing errors.""" + pass + + +class IncrementablesParser: + """Parser for extracting incrementables data from PDF text.""" + + # Anchor patterns to find the incrementables section + ANCHOR_PATTERNS = [ + r"AJUSTE\s+DE\s+INCREMENTABLES\s+EN:", + r"INCREMENTABLES\s+EN:", + r"AJUSTE\s+INCREMENTABLES:", + ] + + # Field patterns + FIELD_PATTERNS = { + "fletes": r"Fletes[:\s]*\$?\s*([\d,]+\.?\d*)\s*(USD|MXN|EUR)?", + "seguros": r"Seguros[:\s]*(?:\$?\s*([\d,]+\.?\d*)\s*)?(USD|MXN|EUR)?", + "almacenaje": r"(?:Almacenaje[/\s]*(?:Consolidaci[oó]n)?)[:\s]*\$?\s*([\d,]+\.?\d*)\s*(USD|MXN|EUR)?", + "regalias": r"(?:Regal[ií]as?)[:\s]*(?:\$?\s*([\d,]+\.?\d*)\s*)?(USD|MXN|EUR)?", + } + + def __init__(self, text: str): + """ + Initialize parser with PDF text. + + Args: + text: Extracted text from PDF + """ + self.text = text + self.warnings: List[str] = [] + self.anchors_found: List[str] = [] + + def _find_incrementables_section(self) -> Optional[str]: + """ + Find the incrementables section in the text. + + Returns: + Text snippet containing incrementables data, or None if not found + """ + for pattern in self.ANCHOR_PATTERNS: + match = re.search(pattern, self.text, re.IGNORECASE | re.MULTILINE) + if match: + anchor_text = match.group(0) + self.anchors_found.append(anchor_text) + logger.info(f"Found anchor: {anchor_text}") + + # Extract the next ~500 characters after the anchor + start_pos = match.end() + section = self.text[start_pos:start_pos + 500] + return section + + return None + + def _extract_currency(self, section: str) -> str: + """ + Extract currency from the section. + + Args: + section: Text section to search + + Returns: + Currency code (USD, MXN, EUR) or "USD" as default + """ + currency_pattern = r"\b(USD|MXN|EUR)\b" + match = re.search(currency_pattern, section) + + if match: + return match.group(1) + + # Default to USD but add warning + self.warnings.append("Currency not explicitly found, defaulting to USD") + return "USD" + + def _parse_amount(self, value: Optional[str]) -> Optional[float]: + """ + Parse monetary amount from string. + + Args: + value: String containing amount (e.g., "1,591.20" or "$1,591.20") + + Returns: + Float value or None if empty/invalid + """ + if not value or value.strip() == "": + return None + + try: + # Remove $ and commas + cleaned = value.replace("$", "").replace(",", "").strip() + if not cleaned: + return None + + # Convert to Decimal for precision, then to float for JSON + amount = float(Decimal(cleaned)) + return amount + except Exception as e: + logger.warning(f"Failed to parse amount '{value}': {str(e)}") + return None + + def _extract_field(self, section: str, field_name: str) -> Tuple[Optional[float], Optional[str]]: + """ + Extract a specific field from the section. + + Args: + section: Text section to search + field_name: Name of field (fletes, seguros, almacenaje, regalias) + + Returns: + Tuple of (amount, currency) or (None, None) + """ + pattern = self.FIELD_PATTERNS.get(field_name) + if not pattern: + return None, None + + match = re.search(pattern, section, re.IGNORECASE | re.MULTILINE) + if not match: + logger.warning(f"Field '{field_name}' not found in section") + return None, None + + groups = match.groups() + + # Extract amount (first group) + amount_str = groups[0] if len(groups) > 0 else None + amount = self._parse_amount(amount_str) + + # Extract currency (second group) + currency = groups[1] if len(groups) > 1 else None + + return amount, currency + + def parse(self) -> Dict: + """ + Parse incrementables data from text. + + Returns: + Dictionary with parsed data including: + - currency + - fletes + - seguros (may be None) + - almacenaje_consolidacion + - regalias (may be None) + - warnings + - anchors_found + + Raises: + ParsingError: If incrementables section not found or parsing fails + """ + # Find the section + section = self._find_incrementables_section() + if not section: + raise ParsingError( + "Incrementables section not found. Expected anchor like 'AJUSTE DE INCREMENTABLES EN:'" + ) + + logger.debug(f"Found section: {section[:200]}...") + + # Extract currency + currency = self._extract_currency(section) + + # Extract fields + fletes, _ = self._extract_field(section, "fletes") + seguros, _ = self._extract_field(section, "seguros") + almacenaje, _ = self._extract_field(section, "almacenaje") + regalias, _ = self._extract_field(section, "regalias") + + # Validate required fields + if fletes is None: + raise ParsingError("Required field 'fletes' not found or invalid") + + if almacenaje is None: + raise ParsingError("Required field 'almacenaje/consolidacion' not found or invalid") + + # seguros and regalias can be None (empty) + + return { + "currency": currency, + "fletes": fletes, + "seguros": seguros, + "almacenaje_consolidacion": almacenaje, + "regalias": regalias, + "warnings": self.warnings, + "anchors_found": self.anchors_found, + } + + +def parse_incrementables(text: str) -> Dict: + """ + Parse incrementables data from PDF text. + + Args: + text: Extracted text from PDF + + Returns: + Dictionary with parsed incrementables data + + Raises: + ParsingError: If parsing fails + """ + parser = IncrementablesParser(text) + return parser.parse() diff --git a/app/services/pdf_text.py b/app/services/pdf_text.py new file mode 100644 index 0000000..142e7fb --- /dev/null +++ b/app/services/pdf_text.py @@ -0,0 +1,112 @@ +"""PDF text extraction service using PyMuPDF and pdfplumber.""" +import fitz # PyMuPDF +import pdfplumber +import logging +from typing import Tuple, Optional +from io import BytesIO + +logger = logging.getLogger(__name__) + + +class PDFExtractionError(Exception): + """Custom exception for PDF extraction errors.""" + pass + + +def extract_text_with_pymupdf(pdf_bytes: bytes) -> Tuple[str, int]: + """ + Extract text from PDF using PyMuPDF (fitz). + + Args: + pdf_bytes: PDF file content as bytes + + Returns: + Tuple of (extracted_text, page_count) + + Raises: + PDFExtractionError: If extraction fails + """ + try: + doc = fitz.open(stream=pdf_bytes, filetype="pdf") + + if doc.is_encrypted: + raise PDFExtractionError("PDF is encrypted and cannot be read") + + page_count = len(doc) + text_parts = [] + + for page in doc: + text_parts.append(page.get_text()) + + doc.close() + full_text = "\n".join(text_parts) + + logger.info(f"Extracted {len(full_text)} characters using PyMuPDF from {page_count} pages") + return full_text, page_count + + except Exception as e: + logger.warning(f"PyMuPDF extraction failed: {str(e)}") + raise PDFExtractionError(f"PyMuPDF extraction failed: {str(e)}") + + +def extract_text_with_pdfplumber(pdf_bytes: bytes) -> Tuple[str, int]: + """ + Extract text from PDF using pdfplumber (fallback method). + + Args: + pdf_bytes: PDF file content as bytes + + Returns: + Tuple of (extracted_text, page_count) + + Raises: + PDFExtractionError: If extraction fails + """ + try: + with pdfplumber.open(BytesIO(pdf_bytes)) as pdf: + page_count = len(pdf.pages) + text_parts = [] + + for page in pdf.pages: + page_text = page.extract_text() + if page_text: + text_parts.append(page_text) + + full_text = "\n".join(text_parts) + + logger.info(f"Extracted {len(full_text)} characters using pdfplumber from {page_count} pages") + return full_text, page_count + + except Exception as e: + logger.error(f"pdfplumber extraction failed: {str(e)}") + raise PDFExtractionError(f"pdfplumber extraction failed: {str(e)}") + + +def extract_text_from_pdf(pdf_bytes: bytes) -> Tuple[str, int, str]: + """ + Extract text from PDF using available methods. + Tries PyMuPDF first, falls back to pdfplumber. + + Args: + pdf_bytes: PDF file content as bytes + + Returns: + Tuple of (extracted_text, page_count, method_used) + + Raises: + PDFExtractionError: If all extraction methods fail + """ + # Try PyMuPDF first + try: + text, pages = extract_text_with_pymupdf(pdf_bytes) + return text, pages, "text" + except PDFExtractionError as e: + logger.warning(f"PyMuPDF failed, trying pdfplumber: {str(e)}") + + # Fallback to pdfplumber + try: + text, pages = extract_text_with_pdfplumber(pdf_bytes) + return text, pages, "text" + except PDFExtractionError as e: + logger.error(f"All extraction methods failed: {str(e)}") + raise PDFExtractionError("Failed to extract text from PDF using all available methods") diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/tasks/parse_tasks.py b/app/tasks/parse_tasks.py new file mode 100644 index 0000000..ed787a2 --- /dev/null +++ b/app/tasks/parse_tasks.py @@ -0,0 +1,95 @@ +"""Celery tasks for PDF parsing.""" +import logging +import hashlib +from app.core.celery_app import celery_app +from app.services.pdf_text import extract_text_from_pdf, PDFExtractionError +from app.services.parser import parse_incrementables, ParsingError + +logger = logging.getLogger(__name__) + + +@celery_app.task(bind=True, name="parse_pdf_task") +def parse_pdf_task(self, pdf_bytes_hex: str, filename: str, document_ref: str = None): + """ + Celery task to parse PDF incrementables asynchronously. + + Args: + self: Celery task instance + pdf_bytes_hex: PDF content as hex string (to serialize) + filename: Original filename + document_ref: Optional document reference + + Returns: + Dictionary with parsed data or error information + """ + task_id = self.request.id + logger.info(f"Starting PDF parse task {task_id} for {filename}") + + try: + # Convert hex back to bytes + pdf_bytes = bytes.fromhex(pdf_bytes_hex) + + # Calculate SHA256 + file_hash = hashlib.sha256(pdf_bytes).hexdigest() + + # Extract text + try: + text, page_count, extraction_method = extract_text_from_pdf(pdf_bytes) + logger.info(f"Task {task_id}: Extracted {len(text)} chars from {page_count} pages") + except PDFExtractionError as e: + logger.error(f"Task {task_id}: Extraction failed - {str(e)}") + return { + "status": "failed", + "error": "extraction_failed", + "message": str(e), + "task_id": task_id + } + + # Parse incrementables + try: + parsed_data = parse_incrementables(text) + logger.info(f"Task {task_id}: Successfully parsed incrementables") + except ParsingError as e: + logger.error(f"Task {task_id}: Parsing failed - {str(e)}") + return { + "status": "failed", + "error": "parsing_failed", + "message": str(e), + "task_id": task_id + } + + # Build successful response + result = { + "status": "completed", + "task_id": task_id, + "document": { + "filename": filename, + "pages": page_count, + "sha256": file_hash, + "document_ref": document_ref + }, + "incrementables": { + "currency": parsed_data["currency"], + "fletes": parsed_data["fletes"], + "seguros": parsed_data["seguros"], + "almacenaje_consolidacion": parsed_data["almacenaje_consolidacion"], + "regalias": parsed_data["regalias"] + }, + "extraction": { + "method": extraction_method, + "anchors_found": parsed_data["anchors_found"], + "warnings": parsed_data["warnings"] + } + } + + logger.info(f"Task {task_id}: Completed successfully") + return result + + except Exception as e: + logger.error(f"Task {task_id}: Unexpected error - {str(e)}", exc_info=True) + return { + "status": "failed", + "error": "unexpected_error", + "message": str(e), + "task_id": task_id + } diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..9bb9f7c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,72 @@ +version: '3.8' + +services: + redis: + image: redis:7-alpine + container_name: mve-redis + ports: + - "16379:6379" + volumes: + - redis_data:/data + restart: unless-stopped + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 3s + retries: 3 + + mve-incrementables-parser: + build: . + container_name: mve-incrementables-parser + ports: + - "9876:9876" + environment: + - AUTH_USERNAME=${AUTH_USERNAME:-admin} + - AUTH_PASSWORD_HASH=${AUTH_PASSWORD_HASH} + - JWT_SECRET=${JWT_SECRET} + - JWT_EXPIRES_MINUTES=${JWT_EXPIRES_MINUTES:-60} + - MAX_FILE_MB=${MAX_FILE_MB:-10} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + - SERVICE_NAME=mve-incrementables-parser + - SERVICE_VERSION=1.0.0 + - REDIS_URL=redis://redis:6379/0 + - CELERY_BROKER_URL=redis://redis:6379/0 + - CELERY_RESULT_BACKEND=redis://redis:6379/0 + volumes: + - ./app:/app/app + depends_on: + redis: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9876/health"] + interval: 30s + timeout: 3s + retries: 3 + start_period: 5s + + celery-worker: + build: . + container_name: mve-celery-worker + command: celery -A app.core.celery_app worker --loglevel=info --concurrency=2 + environment: + - AUTH_USERNAME=${AUTH_USERNAME:-admin} + - AUTH_PASSWORD_HASH=${AUTH_PASSWORD_HASH} + - JWT_SECRET=${JWT_SECRET} + - JWT_EXPIRES_MINUTES=${JWT_EXPIRES_MINUTES:-60} + - MAX_FILE_MB=${MAX_FILE_MB:-10} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + - SERVICE_NAME=mve-incrementables-parser + - SERVICE_VERSION=1.0.0 + - REDIS_URL=redis://redis:6379/0 + - CELERY_BROKER_URL=redis://redis:6379/0 + - CELERY_RESULT_BACKEND=redis://redis:6379/0 + volumes: + - ./app:/app/app + depends_on: + redis: + condition: service_healthy + restart: unless-stopped + +volumes: + redis_data: diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..67c1eed --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* +asyncio_mode = auto diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..bfa7f47 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,15 @@ +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +python-multipart==0.0.6 +pydantic==2.5.3 +pydantic-settings==2.1.0 +python-jose[cryptography]==3.3.0 +bcrypt==4.1.2 +PyMuPDF==1.23.21 +pdfplumber==0.10.4 +python-dotenv==1.0.0 +pytest==7.4.4 +pytest-asyncio==0.23.3 +httpx==0.26.0 +celery==5.3.6 +redis==5.0.1 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..73b17fb --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,57 @@ +"""Pytest configuration and fixtures.""" +import pytest +from fastapi.testclient import TestClient +from app.main import app +from app.core.config import Settings, get_settings +import bcrypt + + +# Test settings +TEST_PASSWORD = "test_password_123" +TEST_PASSWORD_HASH = bcrypt.hashpw(TEST_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') + + +def get_test_settings(): + """Override settings for testing.""" + return Settings( + auth_username="testuser", + auth_password_hash=TEST_PASSWORD_HASH, + jwt_secret="test-secret-key-for-testing-only", + jwt_expires_minutes=60, + max_file_mb=10, + log_level="DEBUG", + service_name="mve-incrementables-parser", + service_version="1.0.0" + ) + + +@pytest.fixture +def test_settings(): + """Provide test settings.""" + return get_test_settings() + + +@pytest.fixture +def client(test_settings): + """Create a test client with overridden settings.""" + app.dependency_overrides[get_settings] = lambda: test_settings + with TestClient(app) as test_client: + yield test_client + app.dependency_overrides.clear() + + +@pytest.fixture +def auth_token(client): + """Get authentication token for testing.""" + response = client.post( + "/auth/login", + json={"username": "testuser", "password": TEST_PASSWORD} + ) + assert response.status_code == 200 + return response.json()["access_token"] + + +@pytest.fixture +def auth_headers(auth_token): + """Get authorization headers for testing.""" + return {"Authorization": f"Bearer {auth_token}"} diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..b8cc833 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,235 @@ +"""Tests for API endpoints.""" +import pytest +from io import BytesIO + + +class TestHealthEndpoint: + """Test health check endpoint.""" + + def test_health_check(self, client): + """Test health endpoint returns correct status.""" + response = client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "ok" + assert data["service"] == "mve-incrementables-parser" + assert data["version"] == "1.0.0" + + +class TestAuthEndpoint: + """Test authentication endpoints.""" + + def test_login_success(self, client): + """Test successful login.""" + response = client.post( + "/auth/login", + json={"username": "testuser", "password": "test_password_123"} + ) + + assert response.status_code == 200 + data = response.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + assert data["expires_in"] == 3600 + + def test_login_invalid_username(self, client): + """Test login with invalid username.""" + response = client.post( + "/auth/login", + json={"username": "wronguser", "password": "test_password_123"} + ) + + assert response.status_code == 401 + data = response.json() + assert data["detail"] == "Invalid credentials" + + def test_login_invalid_password(self, client): + """Test login with invalid password.""" + response = client.post( + "/auth/login", + json={"username": "testuser", "password": "wrongpassword"} + ) + + assert response.status_code == 401 + data = response.json() + assert data["detail"] == "Invalid credentials" + + def test_login_missing_fields(self, client): + """Test login with missing fields.""" + response = client.post( + "/auth/login", + json={"username": "testuser"} + ) + + assert response.status_code == 422 # Validation error + + +class TestParseEndpoint: + """Test incrementables parse endpoint.""" + + def create_mock_pdf(self, content: str) -> BytesIO: + """Create a simple mock PDF file for testing.""" + # This is a minimal PDF structure + pdf_content = f"""%PDF-1.4 +1 0 obj +<< /Type /Catalog /Pages 2 0 R >> +endobj +2 0 obj +<< /Type /Pages /Kids [3 0 R] /Count 1 >> +endobj +3 0 obj +<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents 4 0 R >> +endobj +4 0 obj +<< /Length {len(content)} >> +stream +{content} +endstream +endobj +xref +0 5 +0000000000 65535 f +0000000009 00000 n +0000000058 00000 n +0000000115 00000 n +0000000214 00000 n +trailer +<< /Size 5 /Root 1 0 R >> +startxref +{300 + len(content)} +%%EOF""" + return BytesIO(pdf_content.encode()) + + def test_parse_requires_auth(self, client): + """Test parse endpoint requires authentication.""" + files = {"file": ("test.pdf", BytesIO(b"fake pdf"), "application/pdf")} + response = client.post("/v1/incrementables/parse", files=files) + + assert response.status_code == 403 # Forbidden without auth + + def test_parse_invalid_file_type(self, client, auth_headers): + """Test parse endpoint rejects non-PDF files.""" + files = {"file": ("test.txt", BytesIO(b"not a pdf"), "text/plain")} + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=auth_headers + ) + + assert response.status_code == 400 + assert "PDF" in response.json()["detail"] + + def test_parse_invalid_file_extension(self, client, auth_headers): + """Test parse endpoint rejects files without .pdf extension.""" + files = {"file": ("test.txt", BytesIO(b"content"), "application/pdf")} + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=auth_headers + ) + + assert response.status_code == 400 + assert "PDF" in response.json()["detail"] + + def test_parse_with_correlation_id(self, client, auth_headers): + """Test parse endpoint accepts correlation ID.""" + pdf_content = self.create_mock_pdf("test content") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + headers = {**auth_headers, "X-Correlation-Id": "test-correlation-123"} + + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=headers + ) + + # Will fail at extraction/parsing stage, but should accept the correlation ID + # Either 422 (parsing failed) or 500 (extraction failed) + assert response.status_code in [422, 500] + + def test_parse_with_document_ref(self, client, auth_headers): + """Test parse endpoint accepts document reference.""" + pdf_content = self.create_mock_pdf("test content") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + data = {"document_ref": "REF-12345"} + + response = client.post( + "/v1/incrementables/parse", + files=files, + data=data, + headers=auth_headers + ) + + # Will fail at extraction/parsing stage + assert response.status_code in [422, 500] + + def test_parse_generates_correlation_id(self, client, auth_headers): + """Test parse endpoint generates correlation ID if not provided.""" + pdf_content = self.create_mock_pdf("test content") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=auth_headers + ) + + # Even on error, should have correlation_id in some responses + # This test verifies the endpoint is reachable + assert response.status_code in [422, 500] + + +class TestAuthIntegration: + """Test authentication integration.""" + + def test_expired_token_rejected(self, client, test_settings): + """Test that expired tokens are rejected.""" + from app.core.security import create_access_token + from datetime import timedelta + + # Create an already-expired token + expired_token = create_access_token( + data={"sub": "testuser"}, + expires_delta=timedelta(seconds=-1) + ) + + headers = {"Authorization": f"Bearer {expired_token}"} + pdf_content = BytesIO(b"fake pdf") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=headers + ) + + assert response.status_code == 401 + + def test_invalid_token_format(self, client): + """Test that invalid token format is rejected.""" + headers = {"Authorization": "Bearer invalid_token_xyz"} + pdf_content = BytesIO(b"fake pdf") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=headers + ) + + assert response.status_code == 401 + + def test_missing_bearer_prefix(self, client, auth_token): + """Test that token without Bearer prefix is rejected.""" + headers = {"Authorization": auth_token} + pdf_content = BytesIO(b"fake pdf") + files = {"file": ("test.pdf", pdf_content, "application/pdf")} + + response = client.post( + "/v1/incrementables/parse", + files=files, + headers=headers + ) + + assert response.status_code == 403 diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..4c8e009 --- /dev/null +++ b/tests/test_parser.py @@ -0,0 +1,201 @@ +"""Tests for the incrementables parser service.""" +import pytest +from app.services.parser import IncrementablesParser, ParsingError, parse_incrementables + + +# Sample text fixtures +SAMPLE_TEXT_COMPLETE = """ +CARTA INSTRUCTIVO DE EMBARQUE No. 228718 + +AJUSTE DE INCREMENTABLES EN: + +Fletes: $1,591.20 USD +Seguros: $250.50 USD +Almacenaje/Consolidación: $100.00 USD +Regalías: $75.00 USD + +Total: $2,016.70 USD +""" + +SAMPLE_TEXT_EMPTY_FIELDS = """ +AJUSTE DE INCREMENTABLES EN: + +Fletes: $1,591.20 USD +Seguros: USD +Almacenaje/Consolidación: $0.00 USD +Regalías: USD + +Observaciones: Los seguros y regalías están vacíos +""" + +SAMPLE_TEXT_NO_ANCHOR = """ +Este es un documento sin la sección de incrementables. +Solo tiene texto normal sin el ancla esperada. +""" + +SAMPLE_TEXT_VARIATIONS = """ +INCREMENTABLES EN: + +Fletes: $2,500.00 USD +Seguros: USD +Almacenaje Consolidacion: $150.75 USD +Regalias: $300.00 USD +""" + + +class TestIncrementablesParser: + """Test cases for IncrementablesParser class.""" + + def test_parse_complete_data(self): + """Test parsing with all fields present.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + result = parser.parse() + + assert result["currency"] == "USD" + assert result["fletes"] == 1591.20 + assert result["seguros"] == 250.50 + assert result["almacenaje_consolidacion"] == 100.00 + assert result["regalias"] == 75.00 + assert len(result["anchors_found"]) > 0 + + def test_parse_empty_fields(self): + """Test parsing with empty seguros and regalias.""" + parser = IncrementablesParser(SAMPLE_TEXT_EMPTY_FIELDS) + result = parser.parse() + + assert result["currency"] == "USD" + assert result["fletes"] == 1591.20 + assert result["seguros"] is None + assert result["almacenaje_consolidacion"] == 0.00 + assert result["regalias"] is None + + def test_parse_no_anchor(self): + """Test parsing fails when anchor not found.""" + parser = IncrementablesParser(SAMPLE_TEXT_NO_ANCHOR) + + with pytest.raises(ParsingError) as exc_info: + parser.parse() + + assert "not found" in str(exc_info.value).lower() + + def test_parse_variations(self): + """Test parsing with text variations.""" + parser = IncrementablesParser(SAMPLE_TEXT_VARIATIONS) + result = parser.parse() + + assert result["currency"] == "USD" + assert result["fletes"] == 2500.00 + assert result["seguros"] is None + assert result["almacenaje_consolidacion"] == 150.75 + assert result["regalias"] == 300.00 + + def test_parse_amount_with_commas(self): + """Test amount parsing with comma separators.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + amount = parser._parse_amount("1,591.20") + assert amount == 1591.20 + + def test_parse_amount_with_dollar(self): + """Test amount parsing with dollar sign.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + amount = parser._parse_amount("$1,591.20") + assert amount == 1591.20 + + def test_parse_amount_empty(self): + """Test amount parsing with empty string.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + amount = parser._parse_amount("") + assert amount is None + + def test_parse_amount_none(self): + """Test amount parsing with None.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + amount = parser._parse_amount(None) + assert amount is None + + def test_extract_currency_default(self): + """Test currency extraction defaults to USD.""" + parser = IncrementablesParser("Text without currency") + currency = parser._extract_currency("No currency here") + + assert currency == "USD" + assert len(parser.warnings) > 0 + assert "currency" in parser.warnings[0].lower() + + def test_find_incrementables_section(self): + """Test finding incrementables section.""" + parser = IncrementablesParser(SAMPLE_TEXT_COMPLETE) + section = parser._find_incrementables_section() + + assert section is not None + assert "Fletes" in section + assert len(parser.anchors_found) > 0 + + +def test_parse_incrementables_function(): + """Test the main parse_incrementables function.""" + result = parse_incrementables(SAMPLE_TEXT_COMPLETE) + + assert result["currency"] == "USD" + assert result["fletes"] == 1591.20 + assert "anchors_found" in result + assert "warnings" in result + + +def test_parse_incrementables_raises_error(): + """Test parse_incrementables raises error on invalid input.""" + with pytest.raises(ParsingError): + parse_incrementables(SAMPLE_TEXT_NO_ANCHOR) + + +class TestParserEdgeCases: + """Test edge cases and error conditions.""" + + def test_missing_fletes(self): + """Test that missing fletes raises error.""" + text = """ + AJUSTE DE INCREMENTABLES EN: + Seguros: $100.00 USD + Almacenaje: $50.00 USD + """ + parser = IncrementablesParser(text) + + with pytest.raises(ParsingError) as exc_info: + parser.parse() + + assert "fletes" in str(exc_info.value).lower() + + def test_missing_almacenaje(self): + """Test that missing almacenaje raises error.""" + text = """ + AJUSTE DE INCREMENTABLES EN: + Fletes: $1000.00 USD + Seguros: $100.00 USD + """ + parser = IncrementablesParser(text) + + with pytest.raises(ParsingError) as exc_info: + parser.parse() + + assert "almacenaje" in str(exc_info.value).lower() + + def test_case_insensitive_anchor(self): + """Test anchor detection is case insensitive.""" + text_lower = "ajuste de incrementables en:\nFletes: $100.00 USD\nAlmacenaje: $50.00 USD" + parser = IncrementablesParser(text_lower) + section = parser._find_incrementables_section() + + assert section is not None + + def test_accent_variations_regalias(self): + """Test that both 'regalías' and 'regalias' work.""" + text_with_accent = """ + AJUSTE DE INCREMENTABLES EN: + Fletes: $100.00 USD + Almacenaje: $50.00 USD + Regalías: $25.00 USD + """ + parser = IncrementablesParser(text_with_accent) + result = parser.parse() + + assert result["regalias"] == 25.00