""" Service layer for Items business logic Handles CRUD operations for Item with complete one-to-one relationships: Item -> LineItem -> LineFinancial -> LineQuantity -> LineCustoms -> LineDescription -> LineReference -> FaLineItem (Fixed Assets - a24) """ import logging from typing import Optional, List, Tuple from fastapi import HTTPException from sqlalchemy import and_, or_ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, joinedload from core.exceptions import ErrorCollector from .imports.temporary.validators.create import validate_create from .imports.temporary.validators.update import validate_update from api.v1.modules.a76.items.line_items.schemas import LineItemCreate, LineItemUpdate from .schemas import ItemCreate, ItemUpdate from .line_items.models import LineItem from .line_financials.models import LineFinancial from .line_quantities.models import LineQuantity from .line_customs.models import LineCustom from .line_descriptions.models import LineDescription from .line_references.models import LineReference from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem from .models import Item from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.general_catalogs.company.models import Company logger = logging.getLogger(__name__) class ItemService: """ Service for managing Items and related entities with tenant/company isolation """ @staticmethod def get_by_id( db: Session, item_id: int, tenant_id: int, company_id: int ) -> Optional[Item]: """Get an item by ID with tenant/company validation""" return ( db.query(Item) .options( joinedload(Item.lines).joinedload(LineItem.financial), joinedload(Item.lines).joinedload(LineItem.quantity), joinedload(Item.lines).joinedload(LineItem.customs), joinedload(Item.lines).joinedload(LineItem.description), joinedload(Item.lines).joinedload(LineItem.reference), joinedload(Item.lines).joinedload(LineItem.class_info), joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info), joinedload(Item.lines).joinedload(LineItem.fa_data), ) .filter( Item.id == item_id, Item.tenant_id == tenant_id, Item.company_id == company_id, ) .first() ) @staticmethod def get_all( db: Session, tenant_id: int, company_id: int, skip: int = 0, limit: int = 100, filters: Optional[dict] = None, ) -> Tuple[List[Item], int]: """Get all items for a tenant/company with pagination and optional filters""" query = ( db.query(Item) .options( joinedload(Item.lines).joinedload(LineItem.financial), joinedload(Item.lines).joinedload(LineItem.quantity), joinedload(Item.lines).joinedload(LineItem.customs), joinedload(Item.lines).joinedload(LineItem.description), joinedload(Item.lines).joinedload(LineItem.reference), joinedload(Item.lines).joinedload(LineItem.class_info), joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info), joinedload(Item.lines).joinedload(LineItem.fa_data), ) .filter( Item.tenant_id == tenant_id, Item.company_id == company_id, ) ) # Apply filters if provided if filters: if filters.get("invoice_id"): query = query.filter(Item.invoice_id == filters["invoice_id"]) if filters.get("item_type"): query = query.filter(Item.item_type == filters["item_type"]) if filters.get("system_origin"): query = query.filter(Item.system_origin == filters["system_origin"]) if filters.get("search"): search_term = f"%{filters['search']}%" query = query.filter( or_( Item.invoice_number.ilike(search_term), Item.reference_number.ilike(search_term), Item.order.ilike(search_term), Item.guide_number.ilike(search_term), ) ) total = query.count() items = query.offset(skip).limit(limit).all() return items, total @staticmethod def get_by_invoice( db: Session, invoice_id: int, tenant_id: int, company_id: int, skip: int = 0, limit: int = 100, ) -> Tuple[List[Item], int]: """Get all items for a specific invoice""" query = ( db.query(Item) .options( joinedload(Item.lines).joinedload(LineItem.financial), joinedload(Item.lines).joinedload(LineItem.quantity), joinedload(Item.lines).joinedload(LineItem.customs), joinedload(Item.lines).joinedload(LineItem.description), joinedload(Item.lines).joinedload(LineItem.reference), joinedload(Item.lines).joinedload(LineItem.fa_data), ) .filter( Item.invoice_id == invoice_id, Item.tenant_id == tenant_id, Item.company_id == company_id, ) ) total = query.count() items = query.offset(skip).limit(limit).all() return items, total @staticmethod def create( db: Session, item_data: ItemCreate, tenant_id: int, company_id: int, ) -> Item: """Create a new item with all related nested data (multiple lines)""" # Validaciones con ErrorCollector errors = ErrorCollector() # Validar que la factura exista y no esté actualizada (si viene invoice_id) invoice = None if item_data.invoice_id: invoice = ( db.query(InvoiceHeader) .filter( InvoiceHeader.id == item_data.invoice_id, InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, ) .first() ) if not invoice: errors.add_error( field="invoice_id", message="La factura especificada no existe", code="NOT_FOUND", value=str(item_data.invoice_id), ) # Validar cada line item que se va a crear if item_data.lines: for idx, line_data in enumerate(item_data.lines): # Convertir a LineItemCreate para validar line_create = LineItemCreate(**line_data.model_dump()) validate_create(db, line_create, tenant_id, company_id, errors) # Validaciones adicionales específicas del negocio # Validar apóstrofes en número de parte if line_data.part_number and "'" in str(line_data.part_number): errors.add_error( field=f"lines[{idx}].part_number", message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos", code="WARNING_APOSTROPHE", ) # Validar tipo de partida if hasattr(line_data, "item_type"): tipo_partida = line_data.item_type if tipo_partida and tipo_partida not in ["N", "S"]: errors.add_error( field=f"lines[{idx}].item_type", message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'", code="INVALID_ITEM_TYPE", value=str(tipo_partida), ) # Si es subpartida (S), debe tener partida principal if tipo_partida == "S": if ( not hasattr(line_data, "main_line_id") or not line_data.main_line_id ): errors.add_error( field=f"lines[{idx}].main_line_id", message="Las subpartidas (tipo 'S') deben tener una partida principal", code="MISSING_MAIN_LINE", ) # Validar que el line_number sea consecutivo (si se especifica) if hasattr(line_data, "line_number") and line_data.line_number: expected_line = idx + 1 if line_data.line_number != expected_line: errors.add_error( field=f"lines[{idx}].line_number", message=f"Número de línea esperado: {expected_line}, recibido: {line_data.line_number}", code="INVALID_LINE_SEQUENCE", value=str(line_data.line_number), ) # Si hay errores, lanzar excepción ANTES de intentar crear errors.raise_if_errors("Error al crear el item") try: # Extract lines data lines_data = item_data.lines or [] item_dict = item_data.model_dump(exclude={"lines"}) # Add tenant and company item_dict["tenant_id"] = tenant_id item_dict["company_id"] = company_id # Create the item db_item = Item(**item_dict) db.add(db_item) db.flush() # Get the item ID # Create line items if provided for idx, line_data in enumerate(lines_data): # Extract nested data from line financial_data = line_data.financial quantity_data = line_data.quantity customs_data = line_data.customs description_data = line_data.description reference_data = line_data.reference fa_data = line_data.fa_data line_dict = line_data.model_dump( exclude={ "financial", "quantity", "customs", "description", "reference", "fa_data", } ) line_dict["item_id"] = db_item.id line_dict["tenant_id"] = tenant_id line_dict["company_id"] = company_id # Create line item db_line = LineItem(**line_dict) db.add(db_line) db.flush() # Get the line ID # Create financial data if provided if financial_data: financial_dict = financial_data.model_dump() financial_dict["item_line_id"] = db_line.id db_financial = LineFinancial(**financial_dict) db.add(db_financial) # Create quantity data if provided if quantity_data: quantity_dict = quantity_data.model_dump() quantity_dict["item_line_id"] = db_line.id db_quantity = LineQuantity(**quantity_dict) db.add(db_quantity) # Create customs data if provided if customs_data: customs_dict = customs_data.model_dump() customs_dict["item_line_id"] = db_line.id db_customs = LineCustom(**customs_dict) db.add(db_customs) # Create description data if provided if description_data: description_dict = description_data.model_dump() description_dict["item_line_id"] = db_line.id db_description = LineDescription(**description_dict) db.add(db_description) # Create reference data if provided if reference_data: reference_dict = reference_data.model_dump() reference_dict["item_line_id"] = db_line.id db_reference = LineReference(**reference_dict) db.add(db_reference) # Create FA data if provided if fa_data: fa_dict = fa_data.model_dump( exclude={"line_item_id"} ) # Exclude line_item_id from DTO fa_dict["id"] = db_line.id # FA table uses same ID as line item fa_dict["tenant_id"] = tenant_id fa_dict["company_id"] = company_id db_fa = FaLineItem(**fa_dict) db.add(db_fa) db.commit() db.refresh(db_item) return db_item except IntegrityError as e: db.rollback() logger.error(f"Error creating item: {e}") raise HTTPException( status_code=400, detail="Item creation failed - integrity constraint violated", ) except Exception as e: db.rollback() logger.error(f"Unexpected error creating item: {e}") raise HTTPException(status_code=500, detail="Error creating item") @staticmethod def update( db: Session, item_id: int, item_data: ItemUpdate, tenant_id: int, company_id: int, ) -> Item: """Update an item and optionally its nested data (multiple lines)""" # Get existing item db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id) if not db_item: raise HTTPException(status_code=404, detail="Item not found") # Validaciones con ErrorCollector errors = ErrorCollector() # Si se está actualizando el invoice_id, validar la factura invoice = None if item_data.invoice_id: invoice = ( db.query(InvoiceHeader) .filter( InvoiceHeader.id == item_data.invoice_id, InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, ) .first() ) if not invoice: errors.add_error( field="invoice_id", message="La factura especificada no existe", code="NOT_FOUND", value=str(item_data.invoice_id), ) else: # Si no se está actualizando invoice_id, obtener la factura actual por invoice_id from api.v1.modules.a76.invoices.models import InvoiceHeader invoice = ( db.query(InvoiceHeader) .filter(InvoiceHeader.id == db_item.invoice_id) .first() ) # Validar cada line item que se va a actualizar if item_data.lines: for idx, line_data in enumerate(item_data.lines): # Si el line tiene ID, es actualización; si no, es creación if hasattr(line_data, "id") and line_data.id: # Buscar el line item existente existing_line = next( (line for line in db_item.lines if line.id == line_data.id), None, ) if existing_line: # Convertir a LineItemUpdate para validar line_update = LineItemUpdate(**line_data.model_dump()) validate_update(db, line_update, tenant_id, company_id, errors) else: # Es un nuevo line item, validar como creación line_create = LineItemCreate(**line_data.model_dump()) validate_create(db, line_create, tenant_id, company_id, errors) # Validaciones adicionales específicas del negocio # (Aplican tanto para crear como actualizar) # Validar apóstrofes en número de parte if line_data.part_number and "'" in str(line_data.part_number): errors.add_error( field=f"lines[{idx}].part_number", message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos", solution=None, code="WARNING_APOSTROPHE", ) # Validar tipo de partida if hasattr(line_data, "item_type"): tipo_partida = line_data.item_type if tipo_partida and tipo_partida not in ["N", "S"]: errors.add_error( field=f"lines[{idx}].item_type", message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'", solution=None, code="INVALID_ITEM_TYPE", value=str(tipo_partida), ) # Si es subpartida (S), debe tener partida principal if tipo_partida == "S": if ( not hasattr(line_data, "main_line_id") or not line_data.main_line_id ): errors.add_error( field=f"lines[{idx}].main_line_id", message="Las subpartidas (tipo 'S') deben tener una partida principal", solution=None, code="MISSING_MAIN_LINE", ) # Validar que el line_number sea consecutivo (si se especifica) if hasattr(line_data, "line_number") and line_data.line_number: expected_line = idx + 1 if line_data.line_number != expected_line: errors.add_error( field=f"lines[{idx}].line_number", message=f"Número de línea esperado: {expected_line}, recibido: {line_data.line_number}", solution=None, code="INVALID_LINE_SEQUENCE", value=str(line_data.line_number), ) # Si hay errores, lanzar excepción ANTES de actualizar errors.raise_if_errors("Error al actualizar el item") try: # Extract lines data lines_data = item_data.lines item_dict = item_data.model_dump(exclude={"lines"}, exclude_unset=True) # Update item fields for key, value in item_dict.items(): setattr(db_item, key, value) # Update lines if provided (replace all lines) if lines_data is not None: # Delete existing lines (cascade will handle nested data) for existing_line in db_item.lines: db.delete(existing_line) db.flush() # Create new lines for line_data in lines_data: # Extract nested data from line financial_data = line_data.financial quantity_data = line_data.quantity customs_data = line_data.customs description_data = line_data.description reference_data = line_data.reference fa_data = line_data.fa_data line_dict = line_data.model_dump( exclude={ "financial", "quantity", "customs", "description", "reference", "fa_data", }, exclude_unset=True, ) line_dict["item_id"] = db_item.id line_dict["tenant_id"] = tenant_id line_dict["company_id"] = company_id db_line = LineItem(**line_dict) db.add(db_line) db.flush() # Create nested data if provided if financial_data is not None: financial_dict = financial_data.model_dump(exclude_unset=True) financial_dict["item_line_id"] = db_line.id db.add(LineFinancial(**financial_dict)) if quantity_data is not None: quantity_dict = quantity_data.model_dump(exclude_unset=True) quantity_dict["item_line_id"] = db_line.id db.add(LineQuantity(**quantity_dict)) if customs_data is not None: customs_dict = customs_data.model_dump(exclude_unset=True) customs_dict["item_line_id"] = db_line.id db.add(LineCustom(**customs_dict)) if description_data is not None: description_dict = description_data.model_dump( exclude_unset=True ) description_dict["item_line_id"] = db_line.id db.add(LineDescription(**description_dict)) if reference_data is not None: reference_dict = reference_data.model_dump(exclude_unset=True) reference_dict["item_line_id"] = db_line.id db.add(LineReference(**reference_dict)) # Create FA data if provided if fa_data is not None: fa_dict = fa_data.model_dump( exclude_unset=True, exclude={"line_item_id"} ) fa_dict["id"] = db_line.id # FA table uses same ID as line item fa_dict["tenant_id"] = tenant_id fa_dict["company_id"] = company_id db.add(FaLineItem(**fa_dict)) db.commit() db.refresh(db_item) return db_item except HTTPException: raise except Exception as e: db.rollback() logger.error(f"Unexpected error updating item: {e}") raise HTTPException(status_code=500, detail="Error updating item") @staticmethod def delete( db: Session, item_id: int, tenant_id: int, company_id: int, ) -> bool: """Delete an item and all its related data (cascade delete)""" try: db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id) if not db_item: return False db.delete(db_item) db.commit() return True except Exception as e: db.rollback() logger.error(f"Error deleting item: {e}") raise HTTPException(status_code=500, detail="Error deleting item")