""" Capa de servicio para lógica de negocio de clases SCAII y SCAF """ import logging from typing import Any, Dict, List, Optional from fastapi import HTTPException from sqlalchemy import and_, or_ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from .dto import ( ClassBasicDTO, ClassCreateDTO, ClassCreateDTOFA, ClassListDTO, ClassResponseDTO, ClassResponseDTOFA, ClassSearchDTO, ClassUpdateDTO, ) from .models import Class logger = logging.getLogger(__name__) class ClassService: """Servicio para gestión de clases SCAII y SCAF""" @staticmethod def get_all( db: Session, tenant_id: int, company_id: int, skip: int = 0, limit: int = 100, filters: Optional[Dict[str, Any]] = None, ) -> tuple[List[Class], int]: """ Get all classes for a tenant with pagination and filters """ logger.info(f"get_all called with tenant_id={tenant_id}, company_id={company_id}, skip={skip}, limit={limit}") query = db.query(Class).filter( Class.tenant_id == tenant_id, Class.company_id == company_id ) if filters: if filters.get("client_id"): query = query.filter(Class.client_id == filters["client_id"]) if filters.get("class_code"): query = query.filter( Class.class_code.ilike(f"%{filters['class_code']}%") ) if filters.get("description"): description_pattern = f"%{filters['description']}%" query = query.filter( or_( Class.description_es.ilike(description_pattern), Class.description_en.ilike(description_pattern), ) ) if filters.get("material_key"): query = query.filter( Class.material_key.ilike(f"%{filters['material_key']}%") ) if filters.get("fraction"): query = query.filter(Class.fraction.ilike(f"%{filters['fraction']}%")) if filters.get("physical_review") is not None: query = query.filter( Class.physical_review == filters["physical_review"] ) total = query.count() items = query.offset(skip).limit(limit).all() logger.info(f"get_all returning {len(items)} items out of {total} total") return items, total @staticmethod def get_by_id( db: Session, class_id: int, tenant_id: int, company_id: int ) -> Optional[Class]: """Get a class by ID""" return ( db.query(Class) .filter( Class.id == class_id, Class.tenant_id == tenant_id, Class.company_id == company_id, ) .first() ) @staticmethod def create( db: Session, class_data: ClassCreateDTO, tenant_id: int, company_id: int ) -> Class: """Create a new class""" from fastapi import HTTPException from sqlalchemy.exc import IntegrityError data_dict = class_data.model_dump() # Check if class_code already exists for this tenant and company existing = db.query(Class).filter( Class.tenant_id == tenant_id, Class.company_id == company_id, Class.client_id == data_dict["client_id"], Class.class_code == data_dict["class_code"] ).first() if existing: raise HTTPException( status_code=400, detail=f" El código de clase '{data_dict['class_code']}' ya existe. Por favor use un código diferente." ) # Validate material_key exists (now required) from api.v1.modules.public.reference_data.material_types.models import MaterialType material_exists = db.query(MaterialType).filter( MaterialType.key == data_dict["material_key"] ).first() if not material_exists: raise HTTPException( status_code=400, detail=f"Material type '{data_dict['material_key']}' does not exist" ) class_obj = Class(**data_dict) class_obj.tenant_id = tenant_id class_obj.company_id = company_id try: db.add(class_obj) db.commit() db.refresh(class_obj) return class_obj except IntegrityError as e: db.rollback() raise HTTPException( status_code=400, detail=f"Failed to create class: {str(e.orig)}" ) @staticmethod def update( db: Session, class_id: int, tenant_id: int, class_data: ClassUpdateDTO, company_id: int, ) -> Optional[Class]: """Update a class""" logger.info(f"Update called for class_id={class_id}, tenant_id={tenant_id}, company_id={company_id}") logger.info(f"Update data received: {class_data.model_dump(exclude_unset=True)}") class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id) if not class_obj: logger.warning(f"Class {class_id} not found for tenant {tenant_id}, company {company_id}") return None update_data = class_data.model_dump(exclude_unset=True) logger.info(f"Update data after model_dump: {update_data}") # Validate material_key exists if provided if "material_key" in update_data and update_data["material_key"]: from api.v1.modules.public.reference_data.material_types.models import MaterialType material_exists = db.query(MaterialType).filter( MaterialType.key == update_data["material_key"] ).first() if not material_exists: # Set to None if material_key doesn't exist update_data["material_key"] = None # Validate class_code is unique if being changed if "class_code" in update_data and update_data["class_code"]: new_code = update_data["class_code"] # Check if another class with this code exists (excluding current class) # The unique constraint is on (tenant_id, company_id, client_id, class_code) existing_class = db.query(Class).filter( Class.class_code == new_code, Class.tenant_id == tenant_id, Class.company_id == company_id, Class.client_id == class_obj.client_id, # Same client Class.id != class_id # Exclude current class ).first() logger.info(f"Checking for duplicate class_code '{new_code}' for client {class_obj.client_id}") if existing_class: logger.warning(f"Duplicate class_code found: {existing_class.id}") raise HTTPException( status_code=400, detail=f"El código '{new_code}' ya está en uso para este cliente. Por favor ingrese un código diferente." ) for field, value in update_data.items(): setattr(class_obj, field, value) try: logger.info(f"Attempting to commit changes for class {class_id}") db.commit() db.refresh(class_obj) logger.info(f"Successfully updated class {class_id}") return class_obj except IntegrityError as e: db.rollback() error_msg = str(e.orig) logger.error(f"IntegrityError updating class {class_id}: {error_msg}") # Check if it's a duplicate class_code error if "already exists" in error_msg.lower() or "duplicate" in error_msg.lower(): # Extract the code from update_data if it was changed code = update_data.get("class_code", class_obj.class_code) raise HTTPException( status_code=400, detail=f"El código '{code}' ya está en uso. Por favor ingrese un código diferente." ) raise HTTPException( status_code=400, detail=f"Error al actualizar la clase: {error_msg}" ) except Exception as e: db.rollback() logger.error(f"Unexpected error updating class {class_id}: {type(e).__name__}: {str(e)}") raise @staticmethod def delete(db: Session, class_id: int, tenant_id: int, company_id: int) -> bool: """Delete a class (and its FA extension if exists)""" from api.v1.modules.a24.fa.fa_classes.models import QClasses class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id) if not class_obj: return False # Delete FA extension first (if exists) to avoid FK constraint violation fa_extension = db.query(QClasses).filter( QClasses.class_id == class_id, QClasses.tenant_id == tenant_id ).first() if fa_extension: db.delete(fa_extension) # Now delete the base class db.delete(class_obj) db.commit() return True @staticmethod def create_fa_class( db: Session, class_data: ClassCreateDTOFA, tenant_id: int, company_id: int ) -> Dict[str, Any]: """ Create a fixed asset class (both a76.classes and a24.fa_classes) Returns a dict with both records combined """ import logging logger = logging.getLogger(__name__) logger.info(f"create_fa_class called with data: {class_data.model_dump()}") from api.v1.modules.a24.fa.fa_classes.models import QClasses # Extract base class fields base_fields = { "client_id", "class_code", "description_es", "description_en", "material_key", "unit_of_measure", "fraction", "us_fraction", "sub_key", "physical_review", "iva_exempt_fraction" } base_data = {k: v for k, v in class_data.model_dump().items() if k in base_fields} # Extract FA-specific fields fa_fields = { "import_tariff_code", "import_tariff_type", "export_tariff_code", "export_tariff_type", "depreciation_rate", "fda_code", "eccn_code", "class_enabled" } fa_data = {k: v for k, v in class_data.model_dump().items() if k in fa_fields} try: # 1. Create base class base_dto = ClassCreateDTO(**base_data) base_class = ClassService.create(db, base_dto, tenant_id, company_id) # 2. Create FA extension fa_obj = QClasses(**fa_data) fa_obj.class_id = base_class.id fa_obj.tenant_id = tenant_id fa_obj.company_id = company_id db.add(fa_obj) db.commit() db.refresh(fa_obj) # 3. Combine response - build dict manually to avoid SQLAlchemy internals combined_response = { # Base class fields "id": base_class.id, "tenant_id": base_class.tenant_id, "company_id": base_class.company_id, "client_id": base_class.client_id, "class_code": base_class.class_code, "description_es": base_class.description_es, "description_en": base_class.description_en, "material_key": base_class.material_key, "unit_of_measure": base_class.unit_of_measure, "fraction": base_class.fraction, "us_fraction": base_class.us_fraction, "sub_key": base_class.sub_key, "physical_review": base_class.physical_review, "iva_exempt_fraction": base_class.iva_exempt_fraction, "created_at": base_class.created_at, "updated_at": base_class.updated_at, # FA extension fields "fa_id": fa_obj.id, "import_tariff_code": fa_obj.import_tariff_code, "import_tariff_type": fa_obj.import_tariff_type, "export_tariff_code": fa_obj.export_tariff_code, "export_tariff_type": fa_obj.export_tariff_type, "depreciation_rate": fa_obj.depreciation_rate, "fda_code": fa_obj.fda_code, "eccn_code": fa_obj.eccn_code, "class_enabled": fa_obj.class_enabled, } return combined_response except Exception as e: db.rollback() # If FA creation fails, rollback base class too if 'base_class' in locals(): try: db.delete(base_class) db.commit() except: pass # Extract and improve error message error_msg = str(e) if "already exists" in error_msg.lower() or "duplicad" in error_msg.lower(): # Extract code from error if possible code = class_data.class_code raise HTTPException( status_code=400, detail=f"El código '{code}' ya está en uso. Por favor ingrese un código diferente." ) raise HTTPException( status_code=400, detail=f"Error al crear clase de activo fijo: {error_msg}" ) @staticmethod def seed_initial_data( db: Session, tenant_id: int, company_id: int, client_id: int ) -> int: """ Seed initial fixed asset class data Returns: number of records created """ from .seed import seed created_count = 0 for record in seed: ( class_code, description_es, description_en, material_key, unit_of_measure, fraction, us_fraction, bom, ) = record # Check if already exists existing = ( db.query(Class) .filter( Class.tenant_id == tenant_id, Class.company_id == company_id, Class.client_id == client_id, Class.class_code == class_code, ) .first() ) if not existing: class_obj = Class( tenant_id=tenant_id, company_id=company_id, client_id=client_id, class_code=class_code, description_es=description_es, description_en=description_en, material_key=material_key if material_key else None, unit_of_measure=unit_of_measure if unit_of_measure else None, fraction=fraction if fraction else None, us_fraction=us_fraction if us_fraction else None, ) db.add(class_obj) created_count += 1 if created_count > 0: db.commit() return created_count 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) 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) 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() 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]