""" Capa de servicio para lógica de negocio de partes/componentes """ import logging from typing import List, Optional from fastapi import HTTPException from sqlalchemy import and_, func, or_ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from .dto import PartCreateDTO, PartUpdateDTO from .models import Part logger = logging.getLogger(__name__) class PartService: """ Servicio para gestión de partes/componentes """ @staticmethod def create_part(db: Session, part_data: PartCreateDTO) -> Part: """ Crear una nueva parte """ try: db_part = Part(**part_data.model_dump()) db.add(db_part) db.commit() db.refresh(db_part) return db_part except IntegrityError as e: db.rollback() logger.error(f"Error creating part: {e}") raise HTTPException( status_code=400, detail="Part with this client_id and part_number already exists", ) except Exception as e: db.rollback() logger.error(f"Unexpected error creating part: {e}") raise HTTPException(status_code=500, detail="Error creating part") @staticmethod def get_part(db: Session, client_id: int, part_number: str) -> Optional[Part]: """ Obtener una parte por clave de cliente y número de parte """ try: return ( db.query(Part) .filter( and_(Part.client_id == client_id, Part.part_number == part_number) ) .first() ) except Exception as e: logger.error(f"Error getting part: {e}") raise HTTPException(status_code=500, detail="Error retrieving part") @staticmethod def get_parts_paginated( db: Session, skip: int = 0, limit: int = 100, search: Optional[str] = None, client_id: Optional[int] = None, fraction: Optional[str] = None, country_of_origin: Optional[str] = None, ) -> tuple[List[Part], int]: """ Obtener partes con paginación y filtros """ try: query = db.query(Part) # Aplicar filtros if search: query = query.filter( or_( Part.description_spanish.ilike(f"%{search}%"), Part.description_english.ilike(f"%{search}%"), Part.part_number.ilike(f"%{search}%"), ) ) if client_id is not None: query = query.filter(Part.client_id == client_id) if fraction: query = query.filter(Part.fraction == fraction) if country_of_origin: query = query.filter(Part.country_of_origin == country_of_origin) # Contar total total = query.count() # Aplicar paginación parts = query.offset(skip).limit(limit).all() return parts, total except Exception as e: logger.error(f"Error getting paginated parts: {e}") raise HTTPException(status_code=500, detail="Error retrieving parts") @staticmethod def get_parts_by_client(db: Session, client_id: int) -> List[Part]: """ Obtener todas las partes de un cliente específico """ try: return db.query(Part).filter(Part.client_id == client_id).all() except Exception as e: logger.error(f"Error getting parts by client: {e}") raise HTTPException(status_code=500, detail="Error retrieving client parts") @staticmethod def search_parts_by_fraction(db: Session, fraction: str) -> List[Part]: """ Buscar partes por fracción arancelaria """ try: return ( db.query(Part) .filter( or_( Part.fraction.ilike(f"%{fraction}%"), Part.us_fraction.ilike(f"%{fraction}%"), ) ) .all() ) except Exception as e: logger.error(f"Error searching parts by fraction: {e}") raise HTTPException( status_code=500, detail="Error searching parts by fraction" ) @staticmethod def search_parts_by_supplier(db: Session, supplier: str) -> List[Part]: """ Buscar partes por proveedor """ try: return db.query(Part).filter(Part.supplier.ilike(f"%{supplier}%")).all() except Exception as e: logger.error(f"Error searching parts by supplier: {e}") raise HTTPException( status_code=500, detail="Error searching parts by supplier" ) @staticmethod def search_parts_by_country(db: Session, country_code: str) -> List[Part]: """ Buscar partes por país de origen """ try: return db.query(Part).filter(Part.country_of_origin == country_code).all() except Exception as e: logger.error(f"Error searching parts by country: {e}") raise HTTPException( status_code=500, detail="Error searching parts by country" ) @staticmethod def update_part( db: Session, client_id: int, part_number: str, part_data: PartUpdateDTO ) -> Optional[Part]: """ Actualizar una parte existente """ try: db_part = PartService.get_part(db, client_id, part_number) if not db_part: return None # Actualizar campos for field, value in part_data.model_dump(exclude_unset=True).items(): setattr(db_part, field, value) db.commit() db.refresh(db_part) return db_part except Exception as e: db.rollback() logger.error(f"Error updating part: {e}") raise HTTPException(status_code=500, detail="Error updating part") @staticmethod def delete_part(db: Session, client_id: int, part_number: str) -> bool: """ Eliminar una parte """ try: db_part = PartService.get_part(db, client_id, part_number) if not db_part: return False db.delete(db_part) db.commit() return True except Exception as e: db.rollback() logger.error(f"Error deleting part: {e}") raise HTTPException(status_code=500, detail="Error deleting part") @staticmethod def toggle_part_status( db: Session, client_id: int, part_number: str ) -> Optional[Part]: """ Cambiar el estado habilitado/deshabilitado de una parte """ try: db_part = PartService.get_part(db, client_id, part_number) if not db_part: return None # Toggle status (assuming 1 = enabled, 0 = disabled) db_part.is_active = 1 if db_part.is_active == 0 else 0 db.commit() db.refresh(db_part) return db_part except Exception as e: db.rollback() logger.error(f"Error toggling part status: {e}") raise HTTPException(status_code=500, detail="Error toggling part status") @staticmethod def get_parts_statistics(db: Session) -> dict: """ Obtener estadísticas de partes """ try: total_parts = db.query(Part).count() # Partes por cliente parts_by_client = ( db.query(Part.client_id, func.count(Part.part_number).label("count")) .group_by(Part.client_id) .all() ) # Partes por país de origen parts_by_country = ( db.query( Part.country_of_origin, func.count(Part.part_number).label("count") ) .filter(Part.country_of_origin.isnot(None)) .group_by(Part.country_of_origin) .all() ) # Partes habilitadas vs deshabilitadas enabled_parts = db.query(Part).filter(Part.is_active == 1).count() disabled_parts = db.query(Part).filter(Part.is_active == 0).count() return { "total_parts": total_parts, "enabled_parts": enabled_parts, "disabled_parts": disabled_parts, "parts_by_client": [ {"client_id": item[0], "count": item[1]} for item in parts_by_client ], "parts_by_country": [ {"country": item[0], "count": item[1]} for item in parts_by_country ], } except Exception as e: logger.error(f"Error getting parts statistics: {e}") raise HTTPException( status_code=500, detail="Error retrieving parts statistics" ) @staticmethod def get_part_regulatory_info( db: Session, client_id: int, part_number: str ) -> Optional[dict]: """ Obtener información regulatoria específica de una parte """ try: db_part = PartService.get_part(db, client_id, part_number) if not db_part: return None return { "client_id": db_part.client_id, "part_number": db_part.part_number, "fraction": db_part.fraction, "us_fraction": db_part.us_fraction, "fda_key": db_part.fda_key, "fcc_key": db_part.fcc_key, "license_code": db_part.license_code, "eccn": db_part.eccn, "export_code": db_part.export_code, "exclusion_symbol": db_part.exclusion_symbol, "country_of_origin": db_part.country_of_origin, } except Exception as e: logger.error(f"Error getting part regulatory info: {e}") raise HTTPException( status_code=500, detail="Error retrieving part regulatory information" )