""" 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_processed, ) 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 .exports.validators.create import validate_create as validate_create_export from .exports.validators.update import validate_update as validate_update_export 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 .series.models import Serie from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.parts.models import Part from api.v1.modules.a76.general_catalogs.identifiers.models import IdentifierDetail 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", "includes_subitems"} ) fa_dict.update( {"id": line.id, "tenant_id": tenant_id, "company_id": company_id} ) db.add(FaLineItem(**fa_dict)) # Serie data (list: multiple series per line) if hasattr(line_data, "series") and line_data.series: series_list = ( line_data.series if isinstance(line_data.series, list) else [line_data.series] ) for s in series_list: serie_dict = ( s.model_dump(exclude_unset=True) if hasattr(s, "model_dump") else (dict(s) if isinstance(s, dict) else {}) ) if not serie_dict: continue serie_dict["line_item_id"] = line.id serie_dict["tenant_id"] = tenant_id serie_dict["company_id"] = company_id if serie_dict.get("row") is None: serie_dict["row"] = 1 db.add(Serie(**serie_dict)) # Identifier Detail data if hasattr(line_data, "identifiers") and line_data.identifiers: id_list = ( line_data.identifiers if isinstance(line_data.identifiers, list) else [line_data.identifiers] ) for d in id_list: id_dict = ( d.model_dump(exclude_unset=True) if hasattr(d, "model_dump") else (dict(d) if isinstance(d, dict) else {}) ) if not id_dict: continue id_dict["item_line_id"] = line.id id_dict["tenant_id"] = tenant_id id_dict["company_id"] = company_id db.add(IdentifierDetail(**id_dict)) @staticmethod def _attach_series(db: Session, item: LineItem) -> None: """Query and attach all Serie rows for this item as a list.""" series = ( db.query(Serie) .filter(Serie.line_item_id == item.id) .order_by(Serie.row, Serie.id) .all() ) item.series = list(series) @staticmethod def _attach_identifiers(db: Session, item: LineItem) -> None: """Query and attach all IdentifierDetail rows for this item.""" identifiers = ( db.query(IdentifierDetail) .filter(IdentifierDetail.item_line_id == item.id) .all() ) item.identifiers = list(identifiers) @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""" result = ( 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() ) if result: ItemService._attach_series(db, result) ItemService._attach_identifiers(db, result) return result @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() for item in items: ItemService._attach_series(db, item) ItemService._attach_identifiers(db, item) 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() for item in items: ItemService._attach_series(db, item) ItemService._attach_identifiers(db, item) 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 - invoice_id es requerido") invoice = invoice_exists_by_id( db, item_data.invoice_id, tenant_id, company_id, errors ) if not invoice: errors.raise_if_errors("Error al encontra la factura para el item") if not invoice_processed(db, item_data.invoice_id, tenant_id, company_id, errors): errors.raise_if_errors("Error al crear el item - la factura ya fue actualizada, no se pueden agregar items") # 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 if invoice.operation_type == "exp": if invoice.invoice_type == "CR" and invoice.document_type == "AFIJO": errors.add_error("invoice_id", "No se pueden agregar items a una factura de tipo CR con documento AFIJO", code="INVALID_INVOICE_TYPE") validate_create_export( db, item_data, # Schema Pydantic completo tenant_id, company_id, errors, line_number, ) else: 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", "series", } ) # 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) ItemService._attach_series(db, db_item) ItemService._attach_identifiers(db, 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() invoice = invoice_exists_by_id( db, item_data.invoice_id, tenant_id, company_id, errors ) if not invoice: errors.raise_if_errors("Error al encontra la factura para el item") # 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 if invoice.operation_type == "exp": # Validar el item que se va a actualizar validate_update_export( db, item_data, # Schema de update db_item, # LineItem existente en DB tenant_id, company_id, errors, db_item.line_number, ) else: # 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", "series", }, 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.query(Serie).filter(Serie.line_item_id == db_item.id).delete() db.query(IdentifierDetail).filter(IdentifierDetail.item_line_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) ItemService._attach_series(db, db_item) ItemService._attach_identifiers(db, 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")