""" Capa de servicio para lógica de negocio de clases SCAII y SCAF """ from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from sqlalchemy import or_, and_, func from fastapi import HTTPException from typing import List, Optional import logging from .models import Class from .dto import ( ClassCreateDTO, ClassUpdateDTO, ClassResponseDTO, ClassBasicDTO, ClassListDTO, ClassSearchDTO, ) logger = logging.getLogger(__name__) class ClassService: """Servicio para gestión de clases SCAII y SCAF""" def __init__(self, db: Session): self.db = db def create_class(self, class_data: ClassCreateDTO) -> ClassResponseDTO: """ Crea una nueva clase en el sistema Args: class_data: Datos de la clase a crear Returns: ClassResponseDTO con información de la clase creada Raises: HTTPException: Si la clase ya existe o error en la creación """ try: # Verificar que no exista la clase existing = ( self.db.query(Class) .filter( and_( Class.client_id == class_data.client_id, Class.class_code == class_data.class_code, ) ) .first() ) if existing: raise HTTPException( status_code=400, detail=f"Class with client_id '{class_data.client_id}' and class_code '{class_data.class_code}' already exists", ) # Crear clase db_class = Class( client_id=class_data.client_id, class_code=class_data.class_code, description_spanish=class_data.description_spanish, description_english=class_data.description_english, material_key=class_data.material_key, unit_of_measure=class_data.unit_of_measure, fraction=class_data.fraction, us_fraction=class_data.us_fraction, sub_key=class_data.sub_key, physical_review=class_data.physical_review, iva_exempt_fraction=class_data.iva_exempt_fraction, ) self.db.add(db_class) self.db.commit() self.db.refresh(db_class) logger.info(f"Class created: {db_class.client_id}-{db_class.class_code}") return ClassResponseDTO.model_validate(db_class) except IntegrityError as e: self.db.rollback() logger.error(f"IntegrityError creating class: {str(e)}") raise HTTPException( status_code=400, detail="Class with this client_id and class_code already exists", ) except HTTPException: raise except Exception as e: self.db.rollback() logger.error(f"Error creating class: {str(e)}") raise HTTPException(status_code=500, detail="Error creating class") def get_class(self, client_id: int, class_code: str) -> Optional[ClassResponseDTO]: """ Obtiene una clase por clave compuesta Args: client_id: Clave del cliente class_code: Código de clase Returns: ClassResponseDTO o None si no existe """ class_obj = ( self.db.query(Class) .filter(and_(Class.client_id == client_id, Class.class_code == class_code)) .first() ) if not class_obj: return None return ClassResponseDTO.model_validate(class_obj) def list_classes( self, skip: int = 0, limit: int = 100, search_params: Optional[ClassSearchDTO] = None, ) -> ClassListDTO: """ Lista clases con filtros Args: skip: Número de registros a omitir limit: Número máximo de registros a retornar search_params: Parámetros de búsqueda Returns: ClassListDTO con la lista paginada """ query = self.db.query(Class) # Aplicar filtros si se proporcionan if search_params: if search_params.client_id: query = query.filter(Class.client_id == search_params.client_id) if search_params.class_code: query = query.filter( Class.class_code.ilike(f"%{search_params.class_code}%") ) if search_params.description: description_pattern = f"%{search_params.description}%" query = query.filter( or_( Class.description_spanish.ilike(description_pattern), Class.description_english.ilike(description_pattern), ) ) if search_params.material_key: query = query.filter( Class.material_key.ilike(f"%{search_params.material_key}%") ) if search_params.fraction: query = query.filter( Class.fraction.ilike(f"%{search_params.fraction}%") ) if search_params.physical_review is not None: query = query.filter( Class.physical_review == search_params.physical_review ) # Contar total total = query.count() # Aplicar paginación classes = query.offset(skip).limit(limit).all() # Convertir a DTOs básicos class_dtos = [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] return ClassListDTO( classes=class_dtos, total=total, page=(skip // limit) + 1 if limit > 0 else 1, size=len(class_dtos), ) def update_class( self, client_id: int, class_code: str, class_data: ClassUpdateDTO ) -> Optional[ClassResponseDTO]: """ Actualiza una clase Args: client_id: Clave del cliente class_code: Código de clase class_data: Datos a actualizar Returns: ClassResponseDTO actualizado o None si no existe """ class_obj = ( self.db.query(Class) .filter(and_(Class.client_id == client_id, Class.class_code == class_code)) .first() ) if not class_obj: return None try: # Actualizar solo campos proporcionados update_data = class_data.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(class_obj, field, value) self.db.commit() self.db.refresh(class_obj) logger.info(f"Class updated: {client_id}-{class_code}") return ClassResponseDTO.model_validate(class_obj) except Exception as e: self.db.rollback() logger.error(f"Error updating class {client_id}-{class_code}: {str(e)}") raise HTTPException(status_code=500, detail="Error updating class") def delete_class(self, client_id: int, class_code: str) -> bool: """ Elimina una clase Args: client_id: Clave del cliente class_code: Código de clase Returns: True si se eliminó, False si no existe """ class_obj = ( self.db.query(Class) .filter(and_(Class.client_id == client_id, Class.class_code == class_code)) .first() ) if not class_obj: return False try: self.db.delete(class_obj) self.db.commit() logger.info(f"Class deleted: {client_id}-{class_code}") return True except Exception as e: self.db.rollback() logger.error(f"Error deleting class {client_id}-{class_code}: {str(e)}") raise HTTPException(status_code=500, detail="Error deleting class") def search_by_fraction(self, fraction: str) -> List[ClassBasicDTO]: """Busca clases por fracción arancelaria""" classes = ( self.db.query(Class).filter(Class.fraction.ilike(f"%{fraction}%")).all() ) return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] def search_by_client( self, client_id: int, skip: int = 0, limit: int = 100 ) -> List[ClassBasicDTO]: """Obtiene todas las clases de un cliente específico""" classes = ( self.db.query(Class) .filter(Class.client_id == client_id) .offset(skip) .limit(limit) .all() ) return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] def search_by_material(self, material_key: str) -> List[ClassBasicDTO]: """Busca clases por clave de material""" classes = ( self.db.query(Class) .filter(Class.material_key.ilike(f"%{material_key}%")) .all() ) return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] def get_classes_by_physical_review( self, physical_review: int ) -> List[ClassBasicDTO]: """Obtiene clases por indicador de revisión física""" classes = ( self.db.query(Class).filter(Class.physical_review == physical_review).all() ) return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] def get_classes_statistics(self) -> dict: """Obtiene estadísticas básicas de clases""" total_classes = self.db.query(Class).count() # Contar por clientes clients_count = self.db.query(Class.client_id).distinct().count() # Contar por revisión física physical_review_stats = {} for i in range(3): # Asumiendo valores 0, 1, 2 count = self.db.query(Class).filter(Class.physical_review == i).count() physical_review_stats[f"physical_review_{i}"] = count # Contar clases con fracciones with_fraction = self.db.query(Class).filter(Class.fraction.isnot(None)).count() with_us_fraction = ( self.db.query(Class).filter(Class.us_fraction.isnot(None)).count() ) return { "total_classes": total_classes, "clients_with_classes": clients_count, "classes_with_fraction": with_fraction, "classes_with_us_fraction": with_us_fraction, **physical_review_stats, } def get_classes_by_unit_measure(self, unit_of_measure: str) -> List[ClassBasicDTO]: """Obtiene clases por unidad de medida""" classes = ( self.db.query(Class).filter(Class.unit_of_measure == unit_of_measure).all() ) return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]