189 lines
7.1 KiB
Python
189 lines
7.1 KiB
Python
"""
|
|
Procesador de albaranes con OCR y vinculación automática
|
|
"""
|
|
from pathlib import Path
|
|
from typing import Dict, Optional, List
|
|
from django.utils import timezone
|
|
from datetime import datetime
|
|
from .ocr_service import OCRService
|
|
from ..models import (
|
|
Proveedor, Albaran, ReferenciaAlbaran,
|
|
ReferenciaPedidoProveedor, PedidoProveedor
|
|
)
|
|
|
|
|
|
class AlbaranProcessor:
|
|
"""Procesa albaranes y los vincula con pedidos pendientes"""
|
|
|
|
def __init__(self):
|
|
self.ocr_service = OCRService()
|
|
|
|
def _find_proveedor(self, datos: Dict) -> Optional[Proveedor]:
|
|
"""Busca el proveedor basándose en los datos del albarán"""
|
|
nombre = datos.get('proveedor', {}).get('nombre', '').strip()
|
|
numero = datos.get('proveedor', {}).get('numero', '').strip()
|
|
|
|
# Buscar por nombre
|
|
if nombre:
|
|
proveedor = Proveedor.objects.filter(
|
|
nombre__icontains=nombre
|
|
).first()
|
|
if proveedor:
|
|
return proveedor
|
|
|
|
# Buscar por número (si se implementa un campo número_proveedor)
|
|
# Por ahora, retornamos None si no se encuentra
|
|
return None
|
|
|
|
def _parse_fecha(self, fecha_str: str) -> Optional[datetime]:
|
|
"""Parsea una fecha desde string"""
|
|
if not fecha_str:
|
|
return None
|
|
|
|
formatos = [
|
|
'%Y-%m-%d',
|
|
'%d/%m/%Y',
|
|
'%d-%m-%Y',
|
|
'%Y/%m/%d',
|
|
]
|
|
|
|
for fmt in formatos:
|
|
try:
|
|
return datetime.strptime(fecha_str, fmt).date()
|
|
except ValueError:
|
|
continue
|
|
|
|
return None
|
|
|
|
def _match_referencias(
|
|
self,
|
|
referencias_albaran: List[Dict],
|
|
proveedor: Proveedor
|
|
) -> Dict[str, ReferenciaPedidoProveedor]:
|
|
"""
|
|
Busca referencias del albarán en pedidos pendientes del proveedor
|
|
|
|
Returns:
|
|
Dict mapping referencia -> ReferenciaPedidoProveedor
|
|
"""
|
|
matches = {}
|
|
|
|
# Obtener todas las referencias pendientes del proveedor
|
|
pedidos_pendientes = PedidoProveedor.objects.filter(
|
|
proveedor=proveedor,
|
|
estado__in=['pendiente_recepcion', 'parcial']
|
|
).prefetch_related('referencias')
|
|
|
|
for pedido in pedidos_pendientes:
|
|
for ref_pedido in pedido.referencias.filter(estado__in=['pendiente', 'parcial']):
|
|
# Buscar coincidencia en el albarán
|
|
for ref_albaran in referencias_albaran:
|
|
if ref_albaran['referencia'].strip().upper() == ref_pedido.referencia.strip().upper():
|
|
if ref_pedido.referencia not in matches:
|
|
matches[ref_pedido.referencia] = ref_pedido
|
|
break
|
|
|
|
return matches
|
|
|
|
def _match_and_update_referencias(self, albaran: Albaran):
|
|
"""Vincula y actualiza referencias del albarán con pedidos pendientes"""
|
|
if not albaran.proveedor:
|
|
return
|
|
|
|
referencias_albaran = albaran.referencias.all()
|
|
matches = self._match_referencias(
|
|
[{'referencia': ref.referencia} for ref in referencias_albaran],
|
|
albaran.proveedor
|
|
)
|
|
|
|
for ref_albaran in referencias_albaran:
|
|
ref_pedido_proveedor = matches.get(ref_albaran.referencia.strip().upper())
|
|
|
|
if ref_pedido_proveedor:
|
|
ref_albaran.referencia_pedido_proveedor = ref_pedido_proveedor
|
|
ref_albaran.save()
|
|
|
|
# Actualizar pedido proveedor
|
|
ref_pedido_proveedor.unidades_recibidas += ref_albaran.unidades
|
|
|
|
if ref_pedido_proveedor.unidades_recibidas >= ref_pedido_proveedor.unidades_pedidas:
|
|
ref_pedido_proveedor.estado = 'recibido'
|
|
elif ref_pedido_proveedor.unidades_recibidas > 0:
|
|
ref_pedido_proveedor.estado = 'parcial'
|
|
|
|
ref_pedido_proveedor.save()
|
|
|
|
# Actualizar referencia pedido cliente
|
|
if ref_pedido_proveedor.referencia_pedido_cliente:
|
|
ref_cliente = ref_pedido_proveedor.referencia_pedido_cliente
|
|
ref_cliente.unidades_en_stock += ref_albaran.unidades
|
|
ref_cliente.save()
|
|
|
|
def process_albaran_file(self, file_path: Path) -> Albaran:
|
|
"""
|
|
Procesa un archivo de albarán (imagen o PDF)
|
|
|
|
Returns:
|
|
Albaran creado o actualizado
|
|
"""
|
|
# Procesar con OCR
|
|
datos = self.ocr_service.process_albaran(file_path)
|
|
|
|
# Buscar proveedor
|
|
proveedor = self._find_proveedor(datos)
|
|
|
|
# Parsear fecha
|
|
fecha_albaran = self._parse_fecha(datos.get('fecha_albaran', ''))
|
|
|
|
# Crear albarán
|
|
albaran = Albaran.objects.create(
|
|
proveedor=proveedor,
|
|
numero_albaran=datos.get('numero_albaran', '').strip() or None,
|
|
fecha_albaran=fecha_albaran,
|
|
archivo_path=str(file_path),
|
|
estado_procesado='procesado' if proveedor else 'clasificacion',
|
|
fecha_procesado=timezone.now(),
|
|
datos_ocr=datos,
|
|
)
|
|
|
|
# Crear referencias del albarán
|
|
referencias_data = datos.get('referencias', [])
|
|
matches = {}
|
|
|
|
if proveedor:
|
|
matches = self._match_referencias(referencias_data, proveedor)
|
|
|
|
for ref_data in referencias_data:
|
|
ref_pedido_proveedor = matches.get(ref_data['referencia'].strip().upper())
|
|
|
|
referencia = ReferenciaAlbaran.objects.create(
|
|
albaran=albaran,
|
|
referencia=ref_data.get('referencia', '').strip(),
|
|
denominacion=ref_data.get('denominacion', '').strip(),
|
|
unidades=int(ref_data.get('unidades', 1)),
|
|
precio_unitario=float(ref_data.get('precio_unitario', 0)),
|
|
impuesto_tipo=ref_data.get('impuesto_tipo', '21'),
|
|
impuesto_valor=float(ref_data.get('impuesto_valor', 0)),
|
|
referencia_pedido_proveedor=ref_pedido_proveedor,
|
|
)
|
|
|
|
# Actualizar pedido proveedor si hay match
|
|
if ref_pedido_proveedor:
|
|
ref_pedido_proveedor.unidades_recibidas += referencia.unidades
|
|
|
|
# Actualizar estado
|
|
if ref_pedido_proveedor.unidades_recibidas >= ref_pedido_proveedor.unidades_pedidas:
|
|
ref_pedido_proveedor.estado = 'recibido'
|
|
elif ref_pedido_proveedor.unidades_recibidas > 0:
|
|
ref_pedido_proveedor.estado = 'parcial'
|
|
|
|
ref_pedido_proveedor.save()
|
|
|
|
# Actualizar referencia pedido cliente
|
|
if ref_pedido_proveedor.referencia_pedido_cliente:
|
|
ref_cliente = ref_pedido_proveedor.referencia_pedido_cliente
|
|
ref_cliente.unidades_en_stock += referencia.unidades
|
|
ref_cliente.save()
|
|
|
|
return albaran
|