""" 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__) def _validate_stock_uom_conversion( db: Session, tenant_id: int, company_id: int, system: str, stock_unit_of_measure: Optional[str], unit_of_measure: Optional[str], ) -> None: """SCAII: si UM Existencia ≠ UM Comercial, debe existir conversión en catálogo.""" if system != "inventory": return stock_um = (stock_unit_of_measure or "").strip().upper() commercial_um = (unit_of_measure or "").strip().upper() if not stock_um or not commercial_um or stock_um == commercial_um: return from api.v1.modules.a76.general_catalogs.unit_conversions.models import ( UnitConversion, ) conversion_exists = ( db.query(UnitConversion) .filter( UnitConversion.tenant_id == tenant_id, UnitConversion.company_id == company_id, or_( (UnitConversion.from_unit_code == stock_um) & (UnitConversion.to_unit_code == commercial_um), (UnitConversion.from_unit_code == commercial_um) & (UnitConversion.to_unit_code == stock_um), ), ) .first() ) if not conversion_exists: raise HTTPException( status_code=422, detail=( f"No existe un factor de conversión para '{stock_um}' ↔ '{commercial_um}'. " "Configúrelo primero en el catálogo de Conversiones de UM." ), ) class ClassService: """Servicio para gestión de clases SCAII y SCAF""" @staticmethod def get_all( db: Session, tenant_id: int, company_id: Optional[int], skip: int = 0, limit: int = 100, filters: Optional[Dict[str, Any]] = None, sort_by: Optional[str] = None, sort_order: Optional[str] = "asc", ) -> tuple[List[Class], int]: """ Get all classes for a tenant with pagination and filters """ query = db.query(Class).filter(Class.tenant_id == tenant_id) if company_id is not None: query = query.filter(Class.company_id == company_id) if filters: # Búsqueda libre: OR en clave y descripciones (selectores / listados) search_raw = filters.get("search") or filters.get("q") if search_raw and str(search_raw).strip(): pattern = f"%{str(search_raw).strip()}%" query = query.filter( or_( Class.class_code.ilike(pattern), Class.description_es.ilike(pattern), Class.description_en.ilike(pattern), ) ) else: 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"] ) if filters.get("system"): query = query.filter(Class.system == filters["system"]) # Apply sorting if sort_by: column = getattr(Class, sort_by, None) if column: if sort_order == "desc": query = query.order_by(column.desc()) else: query = query.order_by(column.asc()) else: # Default sorting query = query.order_by(Class.class_code.asc()) total = query.count() items = query.offset(skip).limit(limit).all() return items, total @staticmethod def get_all_with_fa_data( db: Session, tenant_id: int, company_id: Optional[int], skip: int = 0, limit: int = 1000, filters: Optional[Dict[str, Any]] = None, sort_by: Optional[str] = None, sort_order: Optional[str] = "asc", ) -> tuple[List[Dict[str, Any]], int]: """ Get all classes with their FA data in a single query using LEFT JOIN. This eliminates the N+1 query problem. Returns a list of dicts with combined base class + FA data. """ from api.v1.modules.a24.fa.fa_classes.models import QClasses # Build query with LEFT JOIN query = ( db.query(Class, QClasses) .outerjoin(QClasses, and_( Class.id == QClasses.class_id, QClasses.tenant_id == tenant_id )) .filter(Class.tenant_id == tenant_id) ) if company_id is not None: query = query.filter(Class.company_id == company_id) # Apply filters if provided if filters: if filters.get("q"): search = f"%{filters['q']}%" query = query.filter( or_( Class.class_code.ilike(search), Class.description_es.ilike(search), Class.description_en.ilike(search) ) ) 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("system"): query = query.filter(Class.system == filters["system"]) # Apply sorting if sort_by: # Check if sort_by belongs to Class or QClasses column = getattr(Class, sort_by, None) if column: if sort_order == "desc": query = query.order_by(column.desc()) else: query = query.order_by(column.asc()) else: # Handle FA extension fields if sort_by is one of them # (Simple approach for now, assuming base class fields are prioritized) pass else: # Default sorting query = query.order_by(Class.class_code.asc()) # Count total before pagination total = query.count() # Apply pagination results = query.offset(skip).limit(limit).all() # Combine base class + FA data into dicts combined = [] for base_class, fa_class in results: class_dict = { # Base class fields "id": base_class.id, "tenant_id": base_class.tenant_id, "company_id": base_class.company_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, "stock_unit_of_measure": base_class.stock_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, "system": base_class.system, "created_at": base_class.created_at, "updated_at": base_class.updated_at, # FA extension fields (None if no FA record exists) "fa_class_id": fa_class.id if fa_class else None, "import_tariff_code": fa_class.import_tariff_code if fa_class else None, "import_tariff_type": fa_class.import_tariff_type if fa_class else None, "export_tariff_code": fa_class.export_tariff_code if fa_class else None, "export_tariff_type": fa_class.export_tariff_type if fa_class else None, "depreciation_rate": fa_class.depreciation_rate if fa_class else None, "fda_code": fa_class.fda_code if fa_class else None, "eccn_code": fa_class.eccn_code if fa_class else None, "class_enabled": fa_class.class_enabled if fa_class else None, } combined.append(class_dict) return combined, 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.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" ) _validate_stock_uom_conversion( db, tenant_id, company_id, data_dict.get("system", "fixed_asset"), data_dict.get("stock_unit_of_measure"), data_dict.get("unit_of_measure"), ) 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""" 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) effective_system = update_data.get("system", class_obj.system) effective_stock_um = update_data.get( "stock_unit_of_measure", class_obj.stock_unit_of_measure ) effective_commercial_um = update_data.get( "unit_of_measure", class_obj.unit_of_measure ) _validate_stock_uom_conversion( db, tenant_id, company_id, effective_system, effective_stock_um, effective_commercial_um, ) # 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, class_code) existing_class = db.query(Class).filter( Class.class_code == new_code, Class.tenant_id == tenant_id, Class.company_id == company_id, Class.id != class_id # Exclude current class ).first() 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: db.commit() db.refresh(class_obj) 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 from api.v1.modules.a76.items.models import LineItem from api.v1.modules.a76.parts.models import Part class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id) if not class_obj: return False used_in_parts = ( db.query(Part.id) .filter( Part.part_class == class_obj.class_code, Part.tenant_id == tenant_id, Part.company_id == company_id, ) .first() is not None ) used_in_invoices = ( db.query(LineItem.id) .filter( LineItem.class_id == class_id, LineItem.tenant_id == tenant_id, LineItem.company_id == company_id, ) .first() is not None ) if used_in_parts or used_in_invoices: reasons: list[str] = [] if used_in_parts: reasons.append("partes") if used_in_invoices: reasons.append("facturas") usage = " y ".join(reasons) raise HTTPException( status_code=409, detail=f"No se puede eliminar la clase porque ya fue utilizada en {usage}.", ) # 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 """ from api.v1.modules.a24.fa.fa_classes.models import QClasses # Extract base class fields base_fields = { "class_code", "description_es", "description_en", "material_key", "unit_of_measure", "stock_unit_of_measure", "fraction", "us_fraction", "sub_key", "physical_review", "iva_exempt_fraction", "system" } 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, "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, "stock_unit_of_measure": base_class.stock_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, "system": base_class.system, "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}" ) 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.class_code == class_data.class_code, ) ) .first() ) if existing: raise HTTPException( status_code=400, detail=f"Class with class_code '{class_data.class_code}' already exists", ) # Crear clase db_class = Class( 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 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, class_code: str) -> Optional[ClassResponseDTO]: """ Obtiene una clase por clave compuesta Args: class_code: Código de clase Returns: ClassResponseDTO o None si no existe """ class_obj = ( self.db.query(Class) .filter(and_(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.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, class_code: str, class_data: ClassUpdateDTO ) -> Optional[ClassResponseDTO]: """ Actualiza una clase Args: 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.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 {class_code}: {str(e)}") raise HTTPException(status_code=500, detail="Error updating class") def delete_class(self, class_code: str) -> bool: """ Elimina una clase Args: class_code: Código de clase Returns: True si se eliminó, False si no existe """ class_obj = ( self.db.query(Class) .filter(and_(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 {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_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 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, "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]