plantillas-proyectos/backend/api/v1/modules/a76/items/service.py

"""
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 datetime
import logging
from decimal import Decimal
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, case, func, or_, select
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
from api.v1.modules.a24.balance_movements.models import BalanceMovement, NEGATIVE_MOVEMENTS
from api.v1.modules.a24.discharges.models import DischargeDetail, DischargeHeader, DischargeStatus

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, None
        )

        if not invoice:
            errors.add_error("invoice_id", "La factura no existe", code="NOT_FOUND", value=str(item_data.invoice_id))
            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, None
        )
        if not invoice:
            errors.add_error("invoice_id", "La factura no existe", code="NOT_FOUND", value=str(item_data.invoice_id))
            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")

    @staticmethod
    def get_lines_with_balance(
        db: Session,
        invoice_id: int,
        tenant_id: int,
        company_id: int,
        as_of_date: Optional[datetime.date] = None,
    ) -> List[dict]:
        """
        Returns every line of an import invoice together with its current
        available balance calculated from the a24.balance_movement ledger.

        Lines with balance <= 0 are included but marked as unavailable so
        the frontend can grey them out / disable them.

        Parameters
        ----------
        as_of_date : optional cut-off date.  Only negative movements
                     (consumptions, etc.) on or before this date are counted,
                     mirroring the CALCULA_SALDO_FECHA_EXPO Clarion logic.
                     If None, all movements are counted (no date restriction).
        """
        lines: List[LineItem] = (
            db.query(LineItem)
            .filter(
                LineItem.invoice_id == invoice_id,
                LineItem.tenant_id == tenant_id,
                LineItem.company_id == company_id,
            )
            .options(
                joinedload(LineItem.quantity),
                joinedload(LineItem.description),
                joinedload(LineItem.part_info),
                joinedload(LineItem.class_info),
                joinedload(LineItem.unit_of_measure_info),
                joinedload(LineItem.fa_data),
                joinedload(LineItem.invoice),
            )
            .order_by(LineItem.line_number)
            .all()
        )

        result = []
        used_map = ItemService._used_quantities_by_procedure(
            db=db,
            import_line_ids=[line.id for line in lines],
            as_of_date=as_of_date,
        )
        for line in lines:
            available_balance = ItemService._compute_balance(db, line.id, as_of_date)
            qty = line.quantity
            desc = line.description
            fa = line.fa_data
            inv = line.invoice
            qty_used_temp = used_map.get((line.id, "TEM"), Decimal(0))
            qty_used_def = used_map.get((line.id, "DEF"), Decimal(0))

            # Count subitems (lines that reference this line as parent via subitem_number)
            subitem_count = 0
            if fa and fa.contains_subitems:
                from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem as FaModel
                subitem_count = (
                    db.query(func.count(LineItem.id))
                    .join(FaModel, FaModel.id == LineItem.id)
                    .filter(
                        LineItem.invoice_id == invoice_id,
                        LineItem.tenant_id == tenant_id,
                        FaModel.is_subitem == True,
                        FaModel.subitem_number == line.line_number,
                    )
                    .scalar() or 0
                )

            result.append({
                "id": line.id,
                "line_number": line.line_number,
                # Invoice info
                "invoice_number": inv.invoice_number if inv else None,
                "invoice_date": inv.invoice_date.isoformat() if inv and inv.invoice_date else None,
                "invoice_status": inv.status if inv and inv.status else None,
                # Part / class
                "part_number": line.part_info.part_number if line.part_info else None,
                "class_code": line.class_info.class_code if line.class_info else None,
                "description_spanish": desc.description_spanish if desc else None,
                "unit_of_measure_code": line.unit_of_measure_info.code if line.unit_of_measure_info else None,
                # Quantities
                "quantity": float(qty.quantity) if qty and qty.quantity is not None else None,
                "quantity_used_temp": float(qty_used_temp),
                "quantity_used_def": float(qty_used_def),
                # Balance
                "available_balance": float(available_balance),
                "has_balance": available_balance > Decimal(0),
                # FA / subitem info
                "is_subitem": fa.is_subitem if fa else None,
                "contains_subitems": fa.contains_subitems if fa else None,
                "subitem_count": subitem_count,
            })
        return result

    @staticmethod
    def _compute_balance(
        db: Session,
        item_line_id: int,
        as_of_date: Optional[datetime.date],
    ) -> Decimal:
        """Net available balance for one import line from the ledger."""
        sign_expr = case(
            (BalanceMovement.movement_type.in_(NEGATIVE_MOVEMENTS), Decimal(-1)),
            else_=Decimal(1),
        )
        if as_of_date is not None:
            date_filter = case(
                (
                    BalanceMovement.movement_type.in_(NEGATIVE_MOVEMENTS),
                    BalanceMovement.operation_date <= as_of_date,
                ),
                else_=True,
            )
        else:
            date_filter = True  # type: ignore[assignment]

        result = db.execute(
            select(func.sum(sign_expr * BalanceMovement.quantity)).where(
                BalanceMovement.import_item_line_id == item_line_id,
                date_filter,
            )
        ).scalar()
        return Decimal(str(result or 0))

    @staticmethod
    def _used_quantities_by_procedure(
        db: Session,
        import_line_ids: List[int],
        as_of_date: Optional[datetime.date],
    ) -> dict[tuple[int, str], Decimal]:
        """
        Returns net used quantity by import line and procedence (TEM/DEF),
        based on active discharge records only (new ledger logic).
        """
        if not import_line_ids:
            return {}

        query = (
            select(
                DischargeDetail.import_item_line_id,
                DischargeDetail.procedence,
                func.sum(DischargeDetail.quantity_discharged),
            )
            .join(DischargeHeader, DischargeHeader.id == DischargeDetail.discharge_header_id)
            .where(
                DischargeDetail.import_item_line_id.in_(import_line_ids),
                DischargeHeader.status == DischargeStatus.APPLIED,
                DischargeDetail.procedence.in_(["TEM", "DEF"]),
            )
            .group_by(DischargeDetail.import_item_line_id, DischargeDetail.procedence)
        )
        if as_of_date is not None:
            query = query.where(DischargeHeader.discharge_date <= as_of_date)

        rows = db.execute(query).all()
        out: dict[tuple[int, str], Decimal] = {}
        for line_id, procedence, qty in rows:
            out[(int(line_id), str(procedence))] = Decimal(str(qty or 0))
        return out