""" Service layer for Items business logic Handles CRUD operations for LineItem with complete one-to-one relationships: LineItem -> LineFinancial -> LineQuantity -> LineCustoms -> LineDescription -> LineReference -> FaLineItem (Fixed Assets - a24) After refactoring: LineItem is the main entity, representing a single line item in an invoice. There is no intermediate Item entity anymore. Each LineItem belongs directly to an InvoiceHeader. """ 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 api.v1.modules.a76.invoices.common.common_validators import ( invoice_exists_by_id, invoice_updated, ) from core.exceptions import ErrorCollector from .imports.validators.create import validate_create as validate_create_import from .imports.validators.update import validate_update as validate_update_import from .schemas import LineItemCreate, LineItemUpdate 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 LineItem from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.parts.models import Part logger = logging.getLogger(__name__) class ItemService: """ Service for managing Items and related entities with tenant/company isolation """ @staticmethod def _resolve_part_number( db: Session, part_number: Optional[str], tenant_id: int, company_id: int, ) -> Optional[int]: """Try to resolve a part number string to its database ID.""" if not part_number: return None # If it's already an integer (or a string representing an integer), it might be the ID try: return int(part_number) except (ValueError, TypeError): # It's a string part number (e.g., "MAQ-001"), look it up part = ( db.query(Part) .filter( Part.part_number == part_number, Part.tenant_id == tenant_id, Part.company_id == company_id, ) .first() ) return part.id if part else None @staticmethod def _get_next_line_number(db: Session, invoice_id: int) -> int: """Calculate the next line_number for a given invoice based on database.""" from sqlalchemy import func max_line = ( db.query(func.max(LineItem.line_number)) .filter(LineItem.invoice_id == invoice_id) .scalar() ) return 1 if max_line is None else max_line + 1 @staticmethod def _renumber_all_invoice_lines(db: Session, invoice_id: int) -> None: """Renumber all line_items for a given invoice to be consecutive (1, 2, 3, ...).""" items = ( db.query(LineItem) .filter(LineItem.invoice_id == invoice_id) .order_by(LineItem.line_number) .all() ) for idx, item in enumerate(items, start=1): item.line_number = idx @staticmethod def _lock_invoice( db: Session, invoice_id: int, tenant_id: int, company_id: int, errors: ErrorCollector, ) -> Optional[InvoiceHeader]: """Lock invoice to prevent concurrent modifications. Returns locked invoice or adds error.""" try: invoice = ( db.query(InvoiceHeader) .filter( InvoiceHeader.id == invoice_id, InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, ) .with_for_update() .first() ) if not invoice: errors.add_error( field="invoice_id", message="La factura no existe o no se pudo bloquear", code="LOCK_FAILED", value=str(invoice_id), ) return invoice except Exception as e: logger.error(f"Error locking invoice {invoice_id}: {e}") errors.add_error( field="invoice_id", message="Error al intentar bloquear la factura", code="LOCK_ERROR", ) return None @staticmethod def _create_line_nested_data( db: Session, line: LineItem, line_data, tenant_id: int, company_id: int ) -> None: """Create all nested data for a line item.""" nested_models = [ (line_data.financial, LineFinancial), (line_data.quantity, LineQuantity), (line_data.customs, LineCustom), (line_data.description, LineDescription), (line_data.reference, LineReference), ] for data, model_class in nested_models: if data: nested_dict = ( data.model_dump(exclude_unset=True) if hasattr(data, "model_dump") else data.model_dump() ) nested_dict["item_line_id"] = line.id db.add(model_class(**nested_dict)) # FA data uses line.id as primary key if line_data.fa_data: fa_dict = line_data.fa_data.model_dump( exclude_unset=True, exclude={"line_item_id"} ) fa_dict.update( {"id": line.id, "tenant_id": tenant_id, "company_id": company_id} ) db.add(FaLineItem(**fa_dict)) @staticmethod def get_by_id( db: Session, item_id: int, tenant_id: int, company_id: int ) -> Optional[LineItem]: """Get an item by ID with tenant/company validation""" return ( db.query(LineItem) .options( joinedload(LineItem.financial), joinedload(LineItem.quantity), joinedload(LineItem.customs), joinedload(LineItem.description), joinedload(LineItem.reference), joinedload(LineItem.class_info), joinedload(LineItem.unit_of_measure_info), joinedload(LineItem.fa_data), ) .filter( LineItem.id == item_id, LineItem.tenant_id == tenant_id, LineItem.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[LineItem], int]: """Get all items for a tenant/company with pagination and optional filters""" query = ( db.query(LineItem) .options( joinedload(LineItem.financial), joinedload(LineItem.quantity), joinedload(LineItem.customs), joinedload(LineItem.description), joinedload(LineItem.reference), joinedload(LineItem.class_info), joinedload(LineItem.unit_of_measure_info), joinedload(LineItem.fa_data), ) .filter( LineItem.tenant_id == tenant_id, LineItem.company_id == company_id, ) ) # Apply filters if provided if filters: if filters.get("invoice_id"): query = query.filter(LineItem.invoice_id == filters["invoice_id"]) if filters.get("item_type"): query = query.filter(LineItem.item_type == filters["item_type"]) if filters.get("system_origin"): query = query.filter(LineItem.system_origin == filters["system_origin"]) if filters.get("search"): search_term = f"%{filters['search']}%" query = query.filter( or_( LineItem.invoice_id.ilike(search_term), LineItem.reference_number.ilike(search_term), LineItem.order.ilike(search_term), LineItem.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[LineItem], int]: """Get all items for a specific invoice""" query = ( db.query(LineItem) .options( joinedload(LineItem.financial), joinedload(LineItem.quantity), joinedload(LineItem.customs), joinedload(LineItem.description), joinedload(LineItem.reference), joinedload(LineItem.fa_data), ) .filter( LineItem.invoice_id == invoice_id, LineItem.tenant_id == tenant_id, LineItem.company_id == company_id, ) ) total = query.count() items = query.offset(skip).limit(limit).all() return items, total @staticmethod def create( db: Session, item_data: LineItemCreate, tenant_id: int, company_id: int, ) -> LineItem: """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) if not item_data.invoice_id: errors.add_required_error(field="invoice_id") errors.raise_if_errors("Error al crear el item") if not invoice_exists_by_id( db, item_data.invoice_id, tenant_id, company_id, errors ): errors.raise_if_errors("Error al crear el item") if not invoice_updated(db, item_data.invoice_id, tenant_id, company_id, errors): errors.raise_if_errors("Error al crear el item") # Lock invoice and calculate line number if not ItemService._lock_invoice( db, item_data.invoice_id, tenant_id, company_id, errors ): errors.raise_if_errors("Error al crear el item") # Calculate the next line number for this single item line_number = ItemService._get_next_line_number(db, item_data.invoice_id) # Resolve part ID if a string is provided in part_number (alias for part_number_id) if item_data.part_number_id and not isinstance(item_data.part_number_id, int): resolved_id = ItemService._resolve_part_number( db, str(item_data.part_number_id), tenant_id, company_id ) if resolved_id: item_data.part_number_id = resolved_id # Resolve component part ID if item_data.component_part_number_id and not isinstance( item_data.component_part_number_id, int ): resolved_id = ItemService._resolve_part_number( db, str(item_data.component_part_number_id), tenant_id, company_id ) if resolved_id: item_data.component_part_number_id = resolved_id # Validar el item validate_create_import( db, item_data, # Schema Pydantic completo tenant_id, company_id, errors, line_number, ) # Validaciones adicionales específicas del negocio if item_data.fa_data and item_data.fa_data.is_subitem is None: errors.add_required_error(field=f"fa_data.is_subitem") if item_data.fa_data and item_data.fa_data.subitem_number is None: errors.add_required_error(field=f"fa_data.subitem_number") # Si hay errores, lanzar excepción ANTES de intentar crear errors.raise_if_errors("Error al crear el item") try: # Prepare item data item_dict = item_data.model_dump( exclude={ "financial", "quantity", "customs", "description", "reference", "fa_data", } ) # Add tenant, company and line number item_dict.update( { "tenant_id": tenant_id, "company_id": company_id, "line_number": line_number, } ) # Create the item db_item = LineItem(**item_dict) db.add(db_item) db.flush() # Get the item ID # Create all nested data ItemService._create_line_nested_data( db, db_item, item_data, tenant_id, company_id ) 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="LineItem creation failed - integrity constraint violated", ) except Exception as e: db.rollback() logger.error(f"Unexpected error creating LineItem: {e}") raise HTTPException(status_code=500, detail="Error creating LineItem") @staticmethod def update( db: Session, item_id: int, item_data: LineItemUpdate, tenant_id: int, company_id: int, ) -> LineItem: """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="LineItem not found") # Validaciones con ErrorCollector errors = ErrorCollector() # Lock invoice invoice_id_to_lock = ( item_data.invoice_id if item_data.invoice_id else db_item.invoice_id ) if not ItemService._lock_invoice( db, invoice_id_to_lock, tenant_id, company_id, errors ): errors.raise_if_errors("Error al actualizar el item") # Resolve part ID if a string is provided in part_number (alias for part_number_id) if hasattr(item_data, 'part_number_id') and item_data.part_number_id and not isinstance(item_data.part_number_id, int): resolved_id = ItemService._resolve_part_number( db, str(item_data.part_number_id), tenant_id, company_id ) if resolved_id: item_data.part_number_id = resolved_id # Resolve component part ID if hasattr(item_data, 'component_part_number_id') and item_data.component_part_number_id and not isinstance( item_data.component_part_number_id, int ): resolved_id = ItemService._resolve_part_number( db, str(item_data.component_part_number_id), tenant_id, company_id ) if resolved_id: item_data.component_part_number_id = resolved_id # Validar el item que se va a actualizar validate_update_import( db, item_data, # Schema de update db_item, # LineItem existente en DB tenant_id, company_id, errors, db_item.line_number, ) # Validar tipo de partida if hasattr(item_data, "item_type") and item_data.item_type: tipo_partida = item_data.item_type if tipo_partida and tipo_partida not in ["N", "S"]: errors.add_error( field=f"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(item_data, "main_line_id") or not item_data.main_line_id: errors.add_error( field=f"main_line_id", message="Las subpartidas (tipo 'S') deben tener una partida principal", solution=None, code="MISSING_MAIN_LINE", ) # Si hay errores, lanzar excepción ANTES de actualizar errors.raise_if_errors("Error al actualizar el item") try: # Get item data excluding nested objects item_dict = item_data.model_dump( exclude={ "financial", "quantity", "customs", "description", "reference", "fa_data", }, exclude_unset=True, ) # Update item fields for key, value in item_dict.items(): setattr(db_item, key, value) # Delete existing nested data db.query(LineFinancial).filter( LineFinancial.item_line_id == db_item.id ).delete() db.query(LineQuantity).filter( LineQuantity.item_line_id == db_item.id ).delete() db.query(LineCustom).filter(LineCustom.item_line_id == db_item.id).delete() db.query(LineDescription).filter( LineDescription.item_line_id == db_item.id ).delete() db.query(LineReference).filter( LineReference.item_line_id == db_item.id ).delete() db.query(FaLineItem).filter(FaLineItem.id == db_item.id).delete() db.flush() # Create new nested data ItemService._create_line_nested_data( db, db_item, item_data, tenant_id, company_id ) # Renumber all lines for this invoice to ensure consecutive numbering ItemService._renumber_all_invoice_lines(db, db_item.invoice_id) 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 invoice_id = db_item.invoice_id errors = ErrorCollector() # Lock the invoice if not ItemService._lock_invoice( db, invoice_id, tenant_id, company_id, errors ): raise HTTPException( status_code=404, detail="Invoice not found or could not be locked" ) db.delete(db_item) db.flush() ItemService._renumber_all_invoice_lines(db, invoice_id) 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")