AlexeerCT
73462fcec3
- Introduced a new movement type 'ENTRY_VOID' to handle invoice reversals, ensuring net balance remains zero. - Updated the FaLineItem DTOs to replace 'download' with 'discharge' for consistency. - Implemented balance entry creation in the invoice processing and reverting workflows, enhancing inventory management. - Adjusted progress tracking messages to reflect the new balance entry generation steps.
650 lines
23 KiB
Python
650 lines
23 KiB
Python
"""
|
|
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
|
|
|
|
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))
|
|
|
|
@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 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)
|
|
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)
|
|
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)
|
|
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)
|
|
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.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)
|
|
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")
|