""" 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 GClass 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(GClass).filter( and_( GClass.client_key == class_data.client_key, GClass.class_code == class_data.class_code ) ).first() if existing: raise HTTPException( status_code=400, detail=f"Class with client_key '{class_data.client_key}' and class_code '{class_data.class_code}' already exists" ) # Crear clase db_class = GClass( client_key=class_data.client_key, 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_key}-{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_key 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_key: int, class_code: str) -> Optional[ClassResponseDTO]: """ Obtiene una clase por clave compuesta Args: client_key: Clave del cliente class_code: Código de clase Returns: ClassResponseDTO o None si no existe """ class_obj = self.db.query(GClass).filter( and_( GClass.client_key == client_key, GClass.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(GClass) # Aplicar filtros si se proporcionan if search_params: if search_params.client_key: query = query.filter(GClass.client_key == search_params.client_key) if search_params.class_code: query = query.filter(GClass.class_code.ilike(f"%{search_params.class_code}%")) if search_params.description: description_pattern = f"%{search_params.description}%" query = query.filter( or_( GClass.description_spanish.ilike(description_pattern), GClass.description_english.ilike(description_pattern) ) ) if search_params.material_key: query = query.filter(GClass.material_key.ilike(f"%{search_params.material_key}%")) if search_params.fraction: query = query.filter(GClass.fraction.ilike(f"%{search_params.fraction}%")) if search_params.physical_review is not None: query = query.filter(GClass.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_key: int, class_code: str, class_data: ClassUpdateDTO) -> Optional[ClassResponseDTO]: """ Actualiza una clase Args: client_key: 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(GClass).filter( and_( GClass.client_key == client_key, GClass.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_key}-{class_code}") return ClassResponseDTO.model_validate(class_obj) except Exception as e: self.db.rollback() logger.error(f"Error updating class {client_key}-{class_code}: {str(e)}") raise HTTPException(status_code=500, detail="Error updating class") def delete_class(self, client_key: int, class_code: str) -> bool: """ Elimina una clase Args: client_key: Clave del cliente class_code: Código de clase Returns: True si se eliminó, False si no existe """ class_obj = self.db.query(GClass).filter( and_( GClass.client_key == client_key, GClass.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_key}-{class_code}") return True except Exception as e: self.db.rollback() logger.error(f"Error deleting class {client_key}-{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(GClass).filter(GClass.fraction.ilike(f"%{fraction}%")).all() return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes] def search_by_client(self, client_key: int, skip: int = 0, limit: int = 100) -> List[ClassBasicDTO]: """Obtiene todas las clases de un cliente específico""" classes = self.db.query(GClass).filter(GClass.client_key == client_key).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(GClass).filter(GClass.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(GClass).filter(GClass.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(GClass).count() # Contar por clientes clients_count = self.db.query(GClass.client_key).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(GClass).filter(GClass.physical_review == i).count() physical_review_stats[f"physical_review_{i}"] = count # Contar clases con fracciones with_fraction = self.db.query(GClass).filter(GClass.fraction.isnot(None)).count() with_us_fraction = self.db.query(GClass).filter(GClass.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(GClass).filter(GClass.unit_of_measure == unit_of_measure).all() return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]