rckrdmrd/erp-core

Fork 0

rckrdmrd 59f1e3badf Initial commit - erp-core

2026-01-04 06:12:07 -06:00

58 KiB

Raw Blame History

SPEC-TRAZABILIDAD-LOTES-SERIES: Trazabilidad Completa de Lotes y Números de Serie

Metadata

Código: SPEC-MGN-005
Módulo: Inventario / Trazabilidad
Gap Relacionado: GAP-MGN-005-002
Prioridad: P1
Esfuerzo Estimado: 13 SP
Versión: 1.0
Última Actualización: 2025-01-09
Referencias: Odoo stock.lot, product_expiry, mrp

1. Resumen Ejecutivo

1.1 Objetivo

Implementar un sistema completo de trazabilidad de lotes y números de serie que permita:

Seguimiento de lotes (batch tracking) para productos perecederos
Seguimiento de números de serie individuales para equipos/electrónica
Trazabilidad ascendente (origen) y descendente (destino)
Gestión de fechas de caducidad con estrategia FEFO
Recall de productos por lote/serie
Integración con manufactura (componentes → producto terminado)

1.2 Alcance

Modelo de datos para lotes y series
Tipos de seguimiento (none, lot, serial)
Validaciones de unicidad y consistencia
Algoritmos de trazabilidad bidireccional
Estrategias de salida (FIFO, LIFO, FEFO)
Gestión de caducidades y alertas
Integración con código de barras GS1
API REST para consultas de trazabilidad

1.3 Tipos de Seguimiento

Tipo	Descripción	Cantidad por Movimiento	Caso de Uso
`none`	Sin seguimiento	Cualquiera	Commodities, materiales a granel
`lot`	Por lote/batch	Múltiples unidades	Alimentos, farmacéuticos, químicos
`serial`	Por número de serie	Exactamente 1.0	Electrónica, equipos, vehículos

2. Modelo de Datos

2.1 Diagrama Entidad-Relación

┌─────────────────────────────────┐
│      products.products          │
│─────────────────────────────────│
│   id (PK)                       │
│   tracking                      │──┐
│   use_expiration_date           │  │
│   expiration_time               │  │
│   use_time                      │  │
│   removal_time                  │  │
│   alert_time                    │  │
│   lot_properties_definition     │  │
└────────────────┬────────────────┘  │
                 │                    │
                 │ 1:N                │
                 ▼                    │
┌─────────────────────────────────┐  │
│      inventory.lots             │  │
│─────────────────────────────────│  │
│   id (PK)                       │  │
│   name (UNIQUE per product)     │  │
│   ref                           │  │
│   product_id (FK) ◄─────────────┼──┘
│   company_id (FK)               │
│   expiration_date               │
│   use_date                      │
│   removal_date                  │
│   alert_date                    │
│   lot_properties (JSONB)        │
│   product_qty (computed)        │
└────────────────┬────────────────┘
                 │
        ┌────────┴────────┐
        │                 │
        ▼                 ▼
┌───────────────┐  ┌─────────────────────────┐
│ inventory     │  │ inventory.move_lines    │
│ .quants       │  │─────────────────────────│
│───────────────│  │   id (PK)               │
│ id (PK)       │  │   move_id (FK)          │
│ product_id    │  │   lot_id (FK)           │
│ lot_id (FK)   │  │   lot_name              │
│ location_id   │  │   quantity              │
│ quantity      │  │   tracking              │
│ reserved_qty  │  │   consume_line_ids (M2M)│
│ in_date       │  │   produce_line_ids (M2M)│
│ removal_date  │  └─────────────────────────┘
└───────────────┘              │
                               │
                               ▼
                 ┌─────────────────────────────┐
                 │ move_line_consume_rel       │
                 │─────────────────────────────│
                 │ consume_line_id (FK)        │
                 │ produce_line_id (FK)        │
                 │ (Tabla de relación M2M)     │
                 └─────────────────────────────┘

2.2 Definición de Tablas

2.2.1 Extensión de products.products

-- Agregar campos de tracking a productos
ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    tracking VARCHAR(16) NOT NULL DEFAULT 'none'
        CHECK (tracking IN ('none', 'lot', 'serial'));

-- Configuración de caducidad
ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    use_expiration_date BOOLEAN NOT NULL DEFAULT false;

ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    expiration_time INTEGER;  -- Días hasta caducidad desde recepción

ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    use_time INTEGER;  -- Días antes de caducidad para "consumir preferentemente"

ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    removal_time INTEGER;  -- Días antes de caducidad para remover de venta

ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    alert_time INTEGER;  -- Días antes de caducidad para alertar

-- Propiedades dinámicas por lote
ALTER TABLE products.products ADD COLUMN IF NOT EXISTS
    lot_properties_definition JSONB DEFAULT '[]';

-- Constraint de consistencia
ALTER TABLE products.products ADD CONSTRAINT chk_expiration_config CHECK (
    use_expiration_date = false OR (
        expiration_time IS NOT NULL AND
        expiration_time > 0
    )
);

-- Índice para productos con tracking
CREATE INDEX idx_products_tracking ON products.products(tracking)
    WHERE tracking != 'none';

2.2.2 inventory.lots

CREATE TABLE inventory.lots (
    -- Identificación
    id UUID PRIMARY KEY DEFAULT uuid_generate_v7(),
    name VARCHAR(128) NOT NULL,
    ref VARCHAR(256),  -- Referencia interna/externa

    -- Relaciones
    product_id UUID NOT NULL REFERENCES products.products(id),
    company_id UUID NOT NULL REFERENCES core.companies(id),

    -- Fechas de caducidad
    expiration_date TIMESTAMPTZ,
    use_date TIMESTAMPTZ,      -- Best-before
    removal_date TIMESTAMPTZ,  -- Fecha de retiro FEFO
    alert_date TIMESTAMPTZ,    -- Fecha de alerta

    -- Control de alertas
    expiry_alerted BOOLEAN NOT NULL DEFAULT false,

    -- Propiedades dinámicas (heredadas del producto)
    lot_properties JSONB DEFAULT '{}',

    -- Cantidad total (calculada desde quants)
    product_qty DECIMAL(20,6) GENERATED ALWAYS AS (
        COALESCE((
            SELECT SUM(quantity)
            FROM inventory.quants q
            JOIN inventory.locations l ON q.location_id = l.id
            WHERE q.lot_id = id
              AND l.usage IN ('internal', 'transit')
        ), 0)
    ) STORED,

    -- Ubicación (si solo hay una)
    location_id UUID REFERENCES inventory.locations(id),

    -- Auditoría
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    created_by UUID REFERENCES core.users(id),

    -- Constraints
    CONSTRAINT uk_lot_product_company UNIQUE (product_id, name, company_id)
);

-- Índices
CREATE INDEX idx_lots_product ON inventory.lots(product_id);
CREATE INDEX idx_lots_expiration ON inventory.lots(expiration_date)
    WHERE expiration_date IS NOT NULL;
CREATE INDEX idx_lots_removal ON inventory.lots(removal_date)
    WHERE removal_date IS NOT NULL;
CREATE INDEX idx_lots_alert ON inventory.lots(alert_date)
    WHERE alert_date IS NOT NULL AND NOT expiry_alerted;
CREATE INDEX idx_lots_name_trgm ON inventory.lots USING GIN (name gin_trgm_ops);

2.2.3 inventory.quants (extensión)

-- Agregar campos de lote a quants existentes
ALTER TABLE inventory.quants ADD COLUMN IF NOT EXISTS
    lot_id UUID REFERENCES inventory.lots(id);

ALTER TABLE inventory.quants ADD COLUMN IF NOT EXISTS
    in_date TIMESTAMPTZ NOT NULL DEFAULT NOW();

-- Fecha de remoción para FEFO (heredada del lote)
ALTER TABLE inventory.quants ADD COLUMN IF NOT EXISTS
    removal_date TIMESTAMPTZ;

-- Indicador de duplicado (solo para seriales)
ALTER TABLE inventory.quants ADD COLUMN IF NOT EXISTS
    sn_duplicated BOOLEAN GENERATED ALWAYS AS (
        CASE
            WHEN lot_id IS NOT NULL AND
                 (SELECT tracking FROM products.products WHERE id = product_id) = 'serial' AND
                 (SELECT COUNT(*) FROM inventory.quants q2
                  WHERE q2.lot_id = lot_id
                    AND q2.quantity > 0) > 1
            THEN true
            ELSE false
        END
    ) STORED;

-- Constraint de unicidad para quants
ALTER TABLE inventory.quants DROP CONSTRAINT IF EXISTS uk_quant_composite;
ALTER TABLE inventory.quants ADD CONSTRAINT uk_quant_composite
    UNIQUE (product_id, location_id, lot_id, package_id, owner_id);

-- Índices optimizados
CREATE INDEX idx_quants_lot ON inventory.quants(lot_id)
    WHERE lot_id IS NOT NULL;

CREATE INDEX idx_quants_fefo ON inventory.quants(product_id, location_id, removal_date, in_date)
    WHERE quantity > 0;

CREATE INDEX idx_quants_fifo ON inventory.quants(product_id, location_id, in_date)
    WHERE quantity > 0;

2.2.4 inventory.move_lines (extensión)

-- Agregar campos de lote a líneas de movimiento
ALTER TABLE inventory.move_lines ADD COLUMN IF NOT EXISTS
    lot_id UUID REFERENCES inventory.lots(id);

ALTER TABLE inventory.move_lines ADD COLUMN IF NOT EXISTS
    lot_name VARCHAR(128);  -- Para creación on-the-fly

ALTER TABLE inventory.move_lines ADD COLUMN IF NOT EXISTS
    tracking VARCHAR(16);  -- Copia del producto

-- Tabla de relación para trazabilidad de manufactura
CREATE TABLE inventory.move_line_consume_rel (
    consume_line_id UUID NOT NULL REFERENCES inventory.move_lines(id) ON DELETE CASCADE,
    produce_line_id UUID NOT NULL REFERENCES inventory.move_lines(id) ON DELETE CASCADE,
    PRIMARY KEY (consume_line_id, produce_line_id)
);

-- Índices
CREATE INDEX idx_move_lines_lot ON inventory.move_lines(lot_id)
    WHERE lot_id IS NOT NULL;
CREATE INDEX idx_move_lines_lot_name ON inventory.move_lines(lot_name)
    WHERE lot_name IS NOT NULL;
CREATE INDEX idx_consume_rel_consume ON inventory.move_line_consume_rel(consume_line_id);
CREATE INDEX idx_consume_rel_produce ON inventory.move_line_consume_rel(produce_line_id);

2.2.5 inventory.removal_strategies

CREATE TABLE inventory.removal_strategies (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v7(),
    name VARCHAR(64) NOT NULL,
    code VARCHAR(16) NOT NULL UNIQUE
        CHECK (code IN ('fifo', 'lifo', 'fefo', 'closest')),
    description TEXT,
    is_active BOOLEAN NOT NULL DEFAULT true
);

-- Datos iniciales
INSERT INTO inventory.removal_strategies (name, code, description) VALUES
('First In, First Out', 'fifo', 'El stock más antiguo sale primero'),
('Last In, First Out', 'lifo', 'El stock más reciente sale primero'),
('First Expiry, First Out', 'fefo', 'El stock que caduca primero sale primero'),
('Closest Location', 'closest', 'El stock de ubicación más cercana sale primero');

-- Agregar estrategia a productos/categorías/ubicaciones
ALTER TABLE products.product_categories ADD COLUMN IF NOT EXISTS
    removal_strategy_id UUID REFERENCES inventory.removal_strategies(id);

ALTER TABLE inventory.locations ADD COLUMN IF NOT EXISTS
    removal_strategy_id UUID REFERENCES inventory.removal_strategies(id);

3. Algoritmos de Trazabilidad

3.1 Trazabilidad Ascendente (Upstream)

from typing import List, Set, Optional
from dataclasses import dataclass
from datetime import datetime
from uuid import UUID

@dataclass
class TraceabilityLine:
    """Línea de resultado de trazabilidad."""
    move_line_id: UUID
    lot_id: UUID
    lot_name: str
    product_id: UUID
    product_name: str
    quantity: float
    date: datetime
    location_from: str
    location_to: str
    reference: str
    reference_type: str  # 'picking', 'production', 'adjustment'
    level: int  # Nivel en el árbol de trazabilidad

class TraceabilityService:
    """
    Servicio de trazabilidad para lotes y números de serie.
    Implementa búsqueda bidireccional con soporte para manufactura.
    """

    def __init__(self, db_session):
        self.db = db_session

    def get_upstream_traceability(
        self,
        lot_id: UUID,
        max_depth: int = 10
    ) -> List[TraceabilityLine]:
        """
        Obtiene la trazabilidad ascendente (¿de dónde vino este lote?).

        Sigue la cadena:
        - Movimientos de origen (move_orig_ids) para MTO
        - Movimientos entrantes a la ubicación para MTS
        - Líneas consumidas en manufactura

        Args:
            lot_id: ID del lote a trazar
            max_depth: Profundidad máxima de búsqueda

        Returns:
            Lista de TraceabilityLine ordenadas por fecha descendente
        """
        lot = self._get_lot(lot_id)
        if not lot:
            return []

        # Obtener líneas de movimiento del lote
        initial_lines = self._get_lot_move_lines(lot_id)
        if not initial_lines:
            return []

        result = []
        lines_seen: Set[UUID] = set()
        lines_todo = list(initial_lines)
        current_level = 0

        while lines_todo and current_level < max_depth:
            current_level += 1
            next_level_lines = []

            for line_id in lines_todo:
                if line_id in lines_seen:
                    continue
                lines_seen.add(line_id)

                line = self._get_move_line_detail(line_id)
                if not line:
                    continue

                # Agregar a resultado
                result.append(TraceabilityLine(
                    move_line_id=line['id'],
                    lot_id=line['lot_id'],
                    lot_name=line['lot_name'],
                    product_id=line['product_id'],
                    product_name=line['product_name'],
                    quantity=line['quantity'],
                    date=line['date'],
                    location_from=line['location_src'],
                    location_to=line['location_dest'],
                    reference=line['reference'],
                    reference_type=line['reference_type'],
                    level=current_level
                ))

                # Buscar líneas de origen
                upstream_lines = self._find_upstream_lines(line, lot_id)
                for upstream_id in upstream_lines:
                    if upstream_id not in lines_seen:
                        next_level_lines.append(upstream_id)

            lines_todo = next_level_lines

        return sorted(result, key=lambda x: x.date, reverse=True)

    def _find_upstream_lines(
        self,
        line: dict,
        lot_id: UUID
    ) -> List[UUID]:
        """
        Encuentra líneas de movimiento anteriores.
        """
        upstream_ids = []

        # Caso 1: MTO - Seguir cadena de movimientos origen
        if line.get('move_orig_ids'):
            query = """
                SELECT ml.id
                FROM inventory.move_lines ml
                JOIN inventory.moves m ON ml.move_id = m.id
                WHERE m.id = ANY(:move_orig_ids)
                  AND ml.lot_id = :lot_id
                  AND ml.state = 'done'
            """
            result = self.db.execute(query, {
                'move_orig_ids': line['move_orig_ids'],
                'lot_id': lot_id
            })
            upstream_ids.extend([r['id'] for r in result])

        # Caso 2: MTS - Buscar movimientos entrantes a la ubicación origen
        elif line.get('location_src_usage') in ('internal', 'transit'):
            query = """
                SELECT ml.id
                FROM inventory.move_lines ml
                WHERE ml.product_id = :product_id
                  AND ml.lot_id = :lot_id
                  AND ml.location_dest_id = :location_id
                  AND ml.id != :current_id
                  AND ml.date <= :date
                  AND ml.state = 'done'
                ORDER BY ml.date DESC
            """
            result = self.db.execute(query, {
                'product_id': line['product_id'],
                'lot_id': lot_id,
                'location_id': line['location_src_id'],
                'current_id': line['id'],
                'date': line['date']
            })
            upstream_ids.extend([r['id'] for r in result])

        # Caso 3: Manufactura - Buscar líneas consumidas
        if line.get('consume_line_ids'):
            upstream_ids.extend(line['consume_line_ids'])

        return upstream_ids

    def get_downstream_traceability(
        self,
        lot_id: UUID,
        max_depth: int = 10
    ) -> List[TraceabilityLine]:
        """
        Obtiene la trazabilidad descendente (¿a dónde fue este lote?).

        Utiliza BFS iterativo para manejar grafos de manufactura.

        Args:
            lot_id: ID del lote a trazar
            max_depth: Profundidad máxima de búsqueda

        Returns:
            Lista de TraceabilityLine con deliveries y producciones
        """
        lot = self._get_lot(lot_id)
        if not lot:
            return []

        # Usar algoritmo BFS para grafos de manufactura
        all_lot_ids: Set[UUID] = {lot_id}
        barren_lines: dict = {}  # lot_id -> set of line_ids (hojas)
        parent_map: dict = {}    # child_lot_id -> set of parent_lot_ids

        queue = [lot_id]
        result = []
        level = 0

        while queue and level < max_depth:
            level += 1
            next_queue = []

            # Obtener todas las líneas salientes para lotes en cola
            query = """
                SELECT ml.*, l.usage as location_dest_usage
                FROM inventory.move_lines ml
                JOIN inventory.locations l ON ml.location_dest_id = l.id
                WHERE ml.lot_id = ANY(:lot_ids)
                  AND ml.state = 'done'
                  AND l.usage IN ('customer', 'internal', 'transit', 'production')
            """
            lines = self.db.execute(query, {'lot_ids': list(queue)}).fetchall()

            for line in lines:
                line_lot_id = line['lot_id']

                # Buscar si esta línea produjo otros lotes
                produce_query = """
                    SELECT DISTINCT ml2.lot_id
                    FROM inventory.move_line_consume_rel rel
                    JOIN inventory.move_lines ml2 ON rel.produce_line_id = ml2.id
                    WHERE rel.consume_line_id = :line_id
                      AND ml2.lot_id IS NOT NULL
                """
                produce_result = self.db.execute(
                    produce_query, {'line_id': line['id']}
                ).fetchall()

                produce_lot_ids = [r['lot_id'] for r in produce_result]

                if produce_lot_ids:
                    # Este lote fue usado para producir otros
                    for child_lot_id in produce_lot_ids:
                        if child_lot_id not in parent_map:
                            parent_map[child_lot_id] = set()
                        parent_map[child_lot_id].add(line_lot_id)

                        if child_lot_id not in all_lot_ids:
                            all_lot_ids.add(child_lot_id)
                            next_queue.append(child_lot_id)
                else:
                    # Línea hoja (delivery final o consumo)
                    if line_lot_id not in barren_lines:
                        barren_lines[line_lot_id] = set()
                    barren_lines[line_lot_id].add(line['id'])

                # Agregar a resultado
                result.append(TraceabilityLine(
                    move_line_id=line['id'],
                    lot_id=line['lot_id'],
                    lot_name=line['lot_name'],
                    product_id=line['product_id'],
                    product_name=line['product_name'],
                    quantity=line['quantity'],
                    date=line['date'],
                    location_from=line['location_src_name'],
                    location_to=line['location_dest_name'],
                    reference=line['reference'],
                    reference_type=self._get_reference_type(line),
                    level=level
                ))

            queue = next_queue

        return sorted(result, key=lambda x: x.date)

    def get_full_traceability_report(
        self,
        lot_id: UUID
    ) -> dict:
        """
        Genera un reporte completo de trazabilidad bidireccional.
        """
        lot = self._get_lot(lot_id)
        if not lot:
            return None

        upstream = self.get_upstream_traceability(lot_id)
        downstream = self.get_downstream_traceability(lot_id)

        # Obtener entregas (deliveries) finales
        deliveries = self._get_lot_deliveries(lot_id)

        return {
            'lot': {
                'id': lot_id,
                'name': lot['name'],
                'product_id': lot['product_id'],
                'product_name': lot['product_name'],
                'expiration_date': lot['expiration_date'],
                'current_qty': lot['product_qty']
            },
            'upstream': [self._line_to_dict(l) for l in upstream],
            'downstream': [self._line_to_dict(l) for l in downstream],
            'deliveries': deliveries,
            'summary': {
                'total_received': sum(l.quantity for l in upstream if l.level == 1),
                'total_shipped': sum(l.quantity for l in downstream
                                    if l.reference_type == 'delivery'),
                'total_consumed': sum(l.quantity for l in downstream
                                     if l.reference_type == 'production'),
                'upstream_levels': max((l.level for l in upstream), default=0),
                'downstream_levels': max((l.level for l in downstream), default=0)
            }
        }

3.2 Estrategias de Salida (FIFO/LIFO/FEFO)

from enum import Enum
from typing import List, Optional
from decimal import Decimal

class RemovalStrategy(Enum):
    FIFO = 'fifo'
    LIFO = 'lifo'
    FEFO = 'fefo'
    CLOSEST = 'closest'

class QuantGatherService:
    """
    Servicio para seleccionar quants según estrategia de salida.
    """

    STRATEGY_ORDER = {
        RemovalStrategy.FIFO: 'in_date ASC, id ASC',
        RemovalStrategy.LIFO: 'in_date DESC, id DESC',
        RemovalStrategy.FEFO: 'removal_date ASC NULLS LAST, in_date ASC, id ASC',
        RemovalStrategy.CLOSEST: None  # Ordenamiento especial
    }

    def __init__(self, db_session):
        self.db = db_session

    def gather_quants(
        self,
        product_id: UUID,
        location_id: UUID,
        quantity_needed: Decimal,
        lot_id: Optional[UUID] = None,
        package_id: Optional[UUID] = None,
        owner_id: Optional[UUID] = None,
        strict: bool = False,
        exclude_expired: bool = True
    ) -> List[dict]:
        """
        Recolecta quants para satisfacer una cantidad requerida.

        Args:
            product_id: ID del producto
            location_id: ID de la ubicación origen
            quantity_needed: Cantidad a recolectar
            lot_id: Filtrar por lote específico (opcional)
            package_id: Filtrar por paquete (opcional)
            owner_id: Filtrar por propietario (opcional)
            strict: Si True, solo coincidencias exactas
            exclude_expired: Excluir lotes caducados

        Returns:
            Lista de quants con cantidades a tomar
        """
        # Determinar estrategia de salida
        strategy = self._get_removal_strategy(product_id, location_id)

        # Construir dominio de búsqueda
        domain_params = {
            'product_id': product_id,
            'location_id': location_id
        }

        where_clauses = [
            "q.product_id = :product_id",
            "q.location_id = :location_id",
            "q.quantity > q.reserved_quantity"
        ]

        if lot_id:
            where_clauses.append("q.lot_id = :lot_id")
            domain_params['lot_id'] = lot_id
        elif strict:
            where_clauses.append("q.lot_id IS NULL")

        if package_id:
            where_clauses.append("q.package_id = :package_id")
            domain_params['package_id'] = package_id
        elif strict:
            where_clauses.append("q.package_id IS NULL")

        if owner_id:
            where_clauses.append("q.owner_id = :owner_id")
            domain_params['owner_id'] = owner_id
        elif strict:
            where_clauses.append("q.owner_id IS NULL")

        # Excluir caducados para FEFO
        if exclude_expired and strategy == RemovalStrategy.FEFO:
            where_clauses.append("""
                (l.expiration_date IS NULL OR l.expiration_date > NOW())
            """)

        # Construir ORDER BY según estrategia
        order_clause = self.STRATEGY_ORDER.get(strategy)
        if strategy == RemovalStrategy.CLOSEST:
            order_clause = self._build_closest_order(location_id)

        # Query principal
        query = f"""
            SELECT
                q.id,
                q.product_id,
                q.lot_id,
                l.name as lot_name,
                l.expiration_date,
                l.removal_date,
                q.location_id,
                q.package_id,
                q.owner_id,
                q.quantity,
                q.reserved_quantity,
                (q.quantity - q.reserved_quantity) as available_quantity,
                q.in_date
            FROM inventory.quants q
            LEFT JOIN inventory.lots l ON q.lot_id = l.id
            WHERE {' AND '.join(where_clauses)}
            ORDER BY
                -- Preferir sin lote sobre con lote (para productos sin tracking)
                (CASE WHEN q.lot_id IS NULL THEN 0 ELSE 1 END),
                {order_clause}
        """

        quants = self.db.execute(query, domain_params).fetchall()

        # Seleccionar quants hasta cubrir cantidad necesaria
        result = []
        remaining = quantity_needed

        for quant in quants:
            if remaining <= 0:
                break

            available = Decimal(str(quant['available_quantity']))
            take_qty = min(available, remaining)

            result.append({
                'quant_id': quant['id'],
                'lot_id': quant['lot_id'],
                'lot_name': quant['lot_name'],
                'quantity_to_take': float(take_qty),
                'available_quantity': float(available),
                'in_date': quant['in_date'],
                'expiration_date': quant['expiration_date'],
                'removal_date': quant['removal_date']
            })

            remaining -= take_qty

        return result

    def _get_removal_strategy(
        self,
        product_id: UUID,
        location_id: UUID
    ) -> RemovalStrategy:
        """
        Determina la estrategia de salida según jerarquía:
        Producto > Categoría > Ubicación > Ubicación Padre
        """
        # Buscar en producto/categoría
        query = """
            SELECT
                COALESCE(
                    pc.removal_strategy_id,
                    (SELECT removal_strategy_id FROM inventory.locations WHERE id = :location_id)
                ) as strategy_id
            FROM products.products p
            LEFT JOIN products.product_categories pc ON p.category_id = pc.id
            WHERE p.id = :product_id
        """
        result = self.db.execute(query, {
            'product_id': product_id,
            'location_id': location_id
        }).fetchone()

        if result and result['strategy_id']:
            strategy_query = """
                SELECT code FROM inventory.removal_strategies
                WHERE id = :strategy_id
            """
            strategy = self.db.execute(
                strategy_query, {'strategy_id': result['strategy_id']}
            ).fetchone()
            if strategy:
                return RemovalStrategy(strategy['code'])

        # Default: FIFO
        return RemovalStrategy.FIFO

    def _build_closest_order(self, location_id: UUID) -> str:
        """
        Construye ORDER BY para estrategia 'closest'.
        Ordena por proximidad jerárquica de ubicación.
        """
        return """
            (SELECT complete_name FROM inventory.locations WHERE id = q.location_id) ASC,
            q.id DESC
        """

4. Gestión de Fechas de Caducidad

4.1 Cálculo Automático de Fechas

from datetime import datetime, timedelta

class LotExpirationService:
    """
    Servicio para gestionar fechas de caducidad de lotes.
    """

    def __init__(self, db_session):
        self.db = db_session

    def compute_expiration_dates(
        self,
        lot_id: UUID,
        reception_date: Optional[datetime] = None
    ) -> dict:
        """
        Calcula todas las fechas de caducidad basándose en el producto.

        Args:
            lot_id: ID del lote
            reception_date: Fecha de recepción (default: now)

        Returns:
            Dict con fechas calculadas
        """
        # Obtener configuración del producto
        query = """
            SELECT
                p.use_expiration_date,
                p.expiration_time,
                p.use_time,
                p.removal_time,
                p.alert_time
            FROM inventory.lots l
            JOIN products.products p ON l.product_id = p.id
            WHERE l.id = :lot_id
        """
        config = self.db.execute(query, {'lot_id': lot_id}).fetchone()

        if not config or not config['use_expiration_date']:
            return {}

        base_date = reception_date or datetime.now()

        # Calcular fechas
        expiration_date = base_date + timedelta(days=config['expiration_time'])

        dates = {
            'expiration_date': expiration_date
        }

        if config['use_time']:
            dates['use_date'] = expiration_date - timedelta(days=config['use_time'])

        if config['removal_time']:
            dates['removal_date'] = expiration_date - timedelta(days=config['removal_time'])

        if config['alert_time']:
            dates['alert_date'] = expiration_date - timedelta(days=config['alert_time'])

        return dates

    def update_lot_dates(self, lot_id: UUID, dates: dict):
        """Actualiza las fechas de un lote."""
        update_fields = []
        params = {'lot_id': lot_id}

        for field in ['expiration_date', 'use_date', 'removal_date', 'alert_date']:
            if field in dates:
                update_fields.append(f"{field} = :{field}")
                params[field] = dates[field]

        if update_fields:
            query = f"""
                UPDATE inventory.lots
                SET {', '.join(update_fields)},
                    updated_at = NOW()
                WHERE id = :lot_id
            """
            self.db.execute(query, params)

    def check_expiration_alerts(self) -> List[dict]:
        """
        Verifica lotes que han alcanzado su fecha de alerta.
        Ejecutar como job programado (cron).

        Returns:
            Lista de lotes que requieren alerta
        """
        query = """
            SELECT
                l.id,
                l.name as lot_name,
                l.product_id,
                p.name as product_name,
                l.expiration_date,
                l.alert_date,
                l.removal_date,
                SUM(q.quantity) as stock_qty
            FROM inventory.lots l
            JOIN products.products p ON l.product_id = p.id
            JOIN inventory.quants q ON q.lot_id = l.id
            JOIN inventory.locations loc ON q.location_id = loc.id
            WHERE l.alert_date <= NOW()
              AND l.expiry_alerted = false
              AND loc.usage = 'internal'
            GROUP BY l.id, p.id
            HAVING SUM(q.quantity) > 0
            ORDER BY l.removal_date ASC NULLS LAST
        """
        lots = self.db.execute(query).fetchall()

        # Marcar como alertados
        if lots:
            lot_ids = [lot['id'] for lot in lots]
            self.db.execute("""
                UPDATE inventory.lots
                SET expiry_alerted = true,
                    updated_at = NOW()
                WHERE id = ANY(:lot_ids)
            """, {'lot_ids': lot_ids})

        return [dict(lot) for lot in lots]

    def get_expiring_lots(
        self,
        days_ahead: int = 30,
        location_id: Optional[UUID] = None
    ) -> List[dict]:
        """
        Obtiene lotes próximos a caducar.

        Args:
            days_ahead: Días hacia adelante para buscar
            location_id: Filtrar por ubicación

        Returns:
            Lista de lotes con stock próximo a caducar
        """
        params = {
            'threshold_date': datetime.now() + timedelta(days=days_ahead)
        }

        location_filter = ""
        if location_id:
            location_filter = "AND q.location_id = :location_id"
            params['location_id'] = location_id

        query = f"""
            SELECT
                l.id,
                l.name as lot_name,
                l.product_id,
                p.name as product_name,
                p.default_code as sku,
                l.expiration_date,
                l.removal_date,
                EXTRACT(DAY FROM l.expiration_date - NOW()) as days_until_expiry,
                SUM(q.quantity) as stock_qty,
                array_agg(DISTINCT loc.name) as locations
            FROM inventory.lots l
            JOIN products.products p ON l.product_id = p.id
            JOIN inventory.quants q ON q.lot_id = l.id
            JOIN inventory.locations loc ON q.location_id = loc.id
            WHERE l.expiration_date <= :threshold_date
              AND l.expiration_date > NOW()
              AND loc.usage = 'internal'
              {location_filter}
            GROUP BY l.id, p.id
            HAVING SUM(q.quantity) > 0
            ORDER BY l.expiration_date ASC
        """

        return self.db.execute(query, params).fetchall()

5. Validaciones

5.1 Validaciones de Lote/Serie

class LotValidationService:
    """
    Servicio de validación para lotes y números de serie.
    """

    def __init__(self, db_session):
        self.db = db_session

    def validate_lot_assignment(
        self,
        product_id: UUID,
        lot_id: Optional[UUID],
        lot_name: Optional[str],
        quantity: Decimal,
        move_line_ids_to_exclude: List[UUID] = None
    ) -> dict:
        """
        Valida la asignación de lote a un movimiento.

        Returns:
            Dict con 'valid', 'errors', 'warnings'
        """
        result = {'valid': True, 'errors': [], 'warnings': []}

        # Obtener tipo de tracking del producto
        product = self._get_product(product_id)
        if not product:
            result['valid'] = False
            result['errors'].append("Producto no encontrado")
            return result

        tracking = product['tracking']

        # Validación 1: Producto sin tracking no debe tener lote
        if tracking == 'none':
            if lot_id or lot_name:
                result['warnings'].append(
                    "Producto sin tracking, el lote será ignorado"
                )
            return result

        # Validación 2: Producto con tracking requiere lote
        if tracking in ('lot', 'serial') and not lot_id and not lot_name:
            result['valid'] = False
            result['errors'].append(
                f"Producto con tracking '{tracking}' requiere lote/serie"
            )
            return result

        # Validación 3: Serial requiere cantidad = 1
        if tracking == 'serial' and quantity != Decimal('1'):
            result['valid'] = False
            result['errors'].append(
                "Productos con tracking 'serial' deben tener cantidad = 1"
            )

        # Validación 4: Serial no puede estar duplicado en mismo picking
        if tracking == 'serial' and (lot_id or lot_name):
            duplicates = self._check_serial_duplicates(
                product_id=product_id,
                lot_id=lot_id,
                lot_name=lot_name,
                exclude_line_ids=move_line_ids_to_exclude
            )
            if duplicates:
                result['valid'] = False
                result['errors'].append(
                    f"Número de serie duplicado en el mismo movimiento"
                )

        # Validación 5: Lote debe pertenecer al producto correcto
        if lot_id:
            lot = self._get_lot(lot_id)
            if lot and lot['product_id'] != product_id:
                result['valid'] = False
                result['errors'].append(
                    f"Lote {lot['name']} no corresponde al producto"
                )

        # Validación 6: Verificar stock existente de serial
        if tracking == 'serial' and (lot_id or lot_name):
            existing_stock = self._check_serial_in_stock(
                product_id=product_id,
                lot_id=lot_id,
                lot_name=lot_name
            )
            if existing_stock:
                result['warnings'].append(
                    f"Número de serie ya existe en stock en: {existing_stock['locations']}"
                )

        return result

    def validate_lot_uniqueness(
        self,
        product_id: UUID,
        lot_name: str,
        company_id: UUID,
        lot_id_to_exclude: Optional[UUID] = None
    ) -> bool:
        """
        Valida que no exista otro lote con el mismo nombre para el producto.
        """
        query = """
            SELECT id FROM inventory.lots
            WHERE product_id = :product_id
              AND name = :lot_name
              AND company_id = :company_id
              AND (:exclude_id IS NULL OR id != :exclude_id)
        """
        result = self.db.execute(query, {
            'product_id': product_id,
            'lot_name': lot_name,
            'company_id': company_id,
            'exclude_id': lot_id_to_exclude
        }).fetchone()

        return result is None

    def _check_serial_duplicates(
        self,
        product_id: UUID,
        lot_id: Optional[UUID],
        lot_name: Optional[str],
        exclude_line_ids: List[UUID] = None
    ) -> bool:
        """Verifica si el serial está duplicado en líneas pendientes."""
        params = {'product_id': product_id}
        conditions = ["ml.product_id = :product_id", "ml.state NOT IN ('done', 'cancel')"]

        if lot_id:
            conditions.append("ml.lot_id = :lot_id")
            params['lot_id'] = lot_id
        elif lot_name:
            conditions.append("ml.lot_name = :lot_name")
            params['lot_name'] = lot_name

        if exclude_line_ids:
            conditions.append("ml.id != ALL(:exclude_ids)")
            params['exclude_ids'] = exclude_line_ids

        query = f"""
            SELECT COUNT(*) as count
            FROM inventory.move_lines ml
            WHERE {' AND '.join(conditions)}
        """
        result = self.db.execute(query, params).fetchone()
        return result['count'] > 0

    def _check_serial_in_stock(
        self,
        product_id: UUID,
        lot_id: Optional[UUID],
        lot_name: Optional[str]
    ) -> Optional[dict]:
        """Verifica si el serial tiene stock existente."""
        if lot_id:
            lot_filter = "l.id = :lot_id"
            params = {'lot_id': lot_id}
        else:
            lot_filter = "l.name = :lot_name AND l.product_id = :product_id"
            params = {'lot_name': lot_name, 'product_id': product_id}

        query = f"""
            SELECT
                l.id,
                l.name,
                SUM(q.quantity) as total_qty,
                array_agg(DISTINCT loc.name) as locations
            FROM inventory.lots l
            JOIN inventory.quants q ON q.lot_id = l.id
            JOIN inventory.locations loc ON q.location_id = loc.id
            WHERE {lot_filter}
              AND loc.usage IN ('internal', 'transit', 'customer')
              AND q.quantity > 0
            GROUP BY l.id
        """
        return self.db.execute(query, params).fetchone()

6. API REST

6.1 Endpoints

openapi: 3.0.3
info:
  title: Lot/Serial Traceability API
  version: 1.0.0

paths:
  /api/v1/inventory/lots:
    get:
      summary: Listar lotes
      parameters:
        - name: product_id
          in: query
          schema:
            type: string
            format: uuid
        - name: search
          in: query
          schema:
            type: string
          description: Búsqueda por nombre de lote
        - name: expiring_within_days
          in: query
          schema:
            type: integer
        - name: has_stock
          in: query
          schema:
            type: boolean
      responses:
        '200':
          description: Lista de lotes
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/Lot'

    post:
      summary: Crear lote
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/LotCreate'
      responses:
        '201':
          description: Lote creado
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Lot'

  /api/v1/inventory/lots/{id}:
    get:
      summary: Obtener lote
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: string
            format: uuid
      responses:
        '200':
          description: Detalle del lote
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/LotDetail'

    patch:
      summary: Actualizar lote
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/LotUpdate'
      responses:
        '200':
          description: Lote actualizado

  /api/v1/inventory/lots/{id}/traceability:
    get:
      summary: Obtener trazabilidad completa
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: string
            format: uuid
        - name: direction
          in: query
          schema:
            type: string
            enum: [upstream, downstream, both]
            default: both
        - name: max_depth
          in: query
          schema:
            type: integer
            default: 10
      responses:
        '200':
          description: Reporte de trazabilidad
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/TraceabilityReport'

  /api/v1/inventory/lots/{id}/deliveries:
    get:
      summary: Obtener entregas del lote
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: string
            format: uuid
      responses:
        '200':
          description: Lista de entregas
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/Delivery'

  /api/v1/inventory/lots/generate-names:
    post:
      summary: Generar nombres de lote secuenciales
      requestBody:
        content:
          application/json:
            schema:
              type: object
              properties:
                first_lot:
                  type: string
                  example: "LOT-2025-0001"
                count:
                  type: integer
                  example: 10
              required:
                - first_lot
                - count
      responses:
        '200':
          description: Nombres generados
          content:
            application/json:
              schema:
                type: array
                items:
                  type: string

  /api/v1/inventory/lots/expiring:
    get:
      summary: Obtener lotes próximos a caducar
      parameters:
        - name: days_ahead
          in: query
          schema:
            type: integer
            default: 30
        - name: location_id
          in: query
          schema:
            type: string
            format: uuid
      responses:
        '200':
          description: Lotes por caducar
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/ExpiringLot'

  /api/v1/inventory/lots/{id}/recall:
    post:
      summary: Iniciar recall de lote
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: string
            format: uuid
      requestBody:
        content:
          application/json:
            schema:
              type: object
              properties:
                reason:
                  type: string
                notify_customers:
                  type: boolean
                  default: true
              required:
                - reason
      responses:
        '200':
          description: Recall iniciado
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/RecallResult'

  /api/v1/inventory/removal-strategies:
    get:
      summary: Listar estrategias de salida
      responses:
        '200':
          description: Estrategias disponibles
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/RemovalStrategy'

components:
  schemas:
    Lot:
      type: object
      properties:
        id:
          type: string
          format: uuid
        name:
          type: string
        ref:
          type: string
        product_id:
          type: string
          format: uuid
        product_name:
          type: string
        expiration_date:
          type: string
          format: date-time
        use_date:
          type: string
          format: date-time
        removal_date:
          type: string
          format: date-time
        alert_date:
          type: string
          format: date-time
        product_qty:
          type: number
        lot_properties:
          type: object

    LotDetail:
      allOf:
        - $ref: '#/components/schemas/Lot'
        - type: object
          properties:
            quants:
              type: array
              items:
                type: object
                properties:
                  location_id:
                    type: string
                    format: uuid
                  location_name:
                    type: string
                  quantity:
                    type: number
                  reserved_quantity:
                    type: number

    LotCreate:
      type: object
      properties:
        name:
          type: string
        ref:
          type: string
        product_id:
          type: string
          format: uuid
        expiration_date:
          type: string
          format: date-time
        lot_properties:
          type: object
      required:
        - name
        - product_id

    LotUpdate:
      type: object
      properties:
        ref:
          type: string
        expiration_date:
          type: string
          format: date-time
        lot_properties:
          type: object

    TraceabilityReport:
      type: object
      properties:
        lot:
          $ref: '#/components/schemas/Lot'
        upstream:
          type: array
          items:
            $ref: '#/components/schemas/TraceabilityLine'
        downstream:
          type: array
          items:
            $ref: '#/components/schemas/TraceabilityLine'
        deliveries:
          type: array
          items:
            $ref: '#/components/schemas/Delivery'
        summary:
          type: object
          properties:
            total_received:
              type: number
            total_shipped:
              type: number
            total_consumed:
              type: number
            upstream_levels:
              type: integer
            downstream_levels:
              type: integer

    TraceabilityLine:
      type: object
      properties:
        move_line_id:
          type: string
          format: uuid
        lot_id:
          type: string
          format: uuid
        lot_name:
          type: string
        product_id:
          type: string
          format: uuid
        product_name:
          type: string
        quantity:
          type: number
        date:
          type: string
          format: date-time
        location_from:
          type: string
        location_to:
          type: string
        reference:
          type: string
        reference_type:
          type: string
          enum: [picking, production, adjustment, scrap]
        level:
          type: integer

    ExpiringLot:
      type: object
      properties:
        id:
          type: string
          format: uuid
        lot_name:
          type: string
        product_id:
          type: string
          format: uuid
        product_name:
          type: string
        sku:
          type: string
        expiration_date:
          type: string
          format: date-time
        days_until_expiry:
          type: integer
        stock_qty:
          type: number
        locations:
          type: array
          items:
            type: string

    RecallResult:
      type: object
      properties:
        success:
          type: boolean
        recall_id:
          type: string
          format: uuid
        affected_deliveries:
          type: integer
        return_pickings_created:
          type: integer
        customers_notified:
          type: integer

    RemovalStrategy:
      type: object
      properties:
        id:
          type: string
          format: uuid
        name:
          type: string
        code:
          type: string
          enum: [fifo, lifo, fefo, closest]
        description:
          type: string

    Delivery:
      type: object
      properties:
        id:
          type: string
          format: uuid
        name:
          type: string
        partner_id:
          type: string
          format: uuid
        partner_name:
          type: string
        date:
          type: string
          format: date-time
        quantity:
          type: number
        state:
          type: string

7. Integración con Código de Barras GS1

7.1 Formato GS1-128

class GS1BarcodeService:
    """
    Servicio para generación y parseo de códigos de barras GS1.
    """

    # Application Identifiers (AI) comunes
    AI_CODES = {
        '01': 'gtin',           # Global Trade Item Number (14 dígitos)
        '02': 'content',        # Contenido del contenedor
        '10': 'lot',            # Número de lote
        '11': 'prod_date',      # Fecha de producción (YYMMDD)
        '13': 'pack_date',      # Fecha de empaque (YYMMDD)
        '15': 'best_before',    # Consumir preferentemente (YYMMDD)
        '17': 'expiry_date',    # Fecha de caducidad (YYMMDD)
        '21': 'serial',         # Número de serie
        '30': 'var_count',      # Cantidad variable
        '310': 'net_weight_kg', # Peso neto en kg (6 decimales)
        '37': 'count',          # Cantidad de unidades
    }

    def generate_gs1_barcode(
        self,
        lot: dict,
        quant: Optional[dict] = None,
        include_dates: bool = True
    ) -> str:
        """
        Genera código de barras GS1-128 para un lote.

        Args:
            lot: Información del lote
            quant: Información del quant (para cantidad)
            include_dates: Incluir fechas de caducidad

        Returns:
            String del código de barras GS1-128
        """
        parts = []

        # AI 10: Número de lote
        parts.append(f"10{lot['name']}")

        # AI 17: Fecha de caducidad
        if include_dates and lot.get('expiration_date'):
            exp_date = lot['expiration_date'].strftime('%y%m%d')
            parts.append(f"17{exp_date}")

        # AI 15: Best before date
        if include_dates and lot.get('use_date'):
            use_date = lot['use_date'].strftime('%y%m%d')
            parts.append(f"15{use_date}")

        # AI 21: Serial (si es tracking serial)
        if lot.get('tracking') == 'serial':
            parts.append(f"21{lot['name']}")

        # AI 37: Cantidad
        if quant and quant.get('quantity'):
            parts.append(f"37{int(quant['quantity'])}")

        return ''.join(parts)

    def parse_gs1_barcode(self, barcode: str) -> dict:
        """
        Parsea un código de barras GS1-128.

        Returns:
            Dict con los campos extraídos
        """
        result = {}
        i = 0

        while i < len(barcode):
            # Buscar AI de 2 dígitos
            ai2 = barcode[i:i+2]
            if ai2 in self.AI_CODES:
                field = self.AI_CODES[ai2]
                i += 2

                if ai2 in ('11', '13', '15', '17'):
                    # Fechas: 6 dígitos YYMMDD
                    value = barcode[i:i+6]
                    result[field] = self._parse_date(value)
                    i += 6
                elif ai2 == '10':
                    # Lote: variable hasta FNC1 o fin
                    end = self._find_end(barcode, i)
                    result[field] = barcode[i:end]
                    i = end
                elif ai2 == '21':
                    # Serial: variable hasta FNC1 o fin
                    end = self._find_end(barcode, i)
                    result[field] = barcode[i:end]
                    i = end
                elif ai2 == '30':
                    # Cantidad variable: hasta 8 dígitos
                    end = self._find_end(barcode, i, max_len=8)
                    result[field] = int(barcode[i:end])
                    i = end
                elif ai2 == '37':
                    # Cantidad: hasta 8 dígitos
                    end = self._find_end(barcode, i, max_len=8)
                    result[field] = int(barcode[i:end])
                    i = end
                continue

            # Buscar AI de 3 dígitos
            ai3 = barcode[i:i+3]
            if ai3 in ('310', '311', '312', '313', '314', '315'):
                # Peso con decimales
                i += 4  # AI + dígito de decimales
                result['weight'] = float(barcode[i:i+6]) / 1000000
                i += 6
                continue

            # AI de 4 dígitos (GTIN)
            ai4 = barcode[i:i+4]
            if ai4.startswith('01'):
                i += 2
                result['gtin'] = barcode[i:i+14]
                i += 14
                continue

            i += 1  # Avanzar si no se reconoce

        return result

    def _parse_date(self, yymmdd: str) -> datetime:
        """Parsea fecha YYMMDD a datetime."""
        year = int(yymmdd[0:2])
        month = int(yymmdd[2:4])
        day = int(yymmdd[4:6])

        # Asumir siglo 21 para años < 50
        full_year = 2000 + year if year < 50 else 1900 + year

        return datetime(full_year, month, day)

    def _find_end(self, s: str, start: int, max_len: int = 20) -> int:
        """Encuentra el fin de un campo variable."""
        # GS (Group Separator) = chr(29) o FNC1
        for i in range(start, min(start + max_len, len(s))):
            if s[i] == chr(29):
                return i
        return min(start + max_len, len(s))

8. Reglas de Negocio

8.1 Validaciones

Regla	Descripción	Acción
RN-001	Serial debe ser único por producto	Error
RN-002	Cantidad de serial debe ser 1.0	Error
RN-003	Lote debe pertenecer al producto	Error
RN-004	Producto con tracking requiere lote	Error en confirmación
RN-005	Lote caducado no puede enviarse	Warning/Block configurable
RN-006	FEFO: priorizar removal_date	Automático

8.2 Comportamientos Automáticos

Trigger	Acción
Crear lote	Calcular fechas de caducidad
Recibir con lote nuevo	Crear lote automáticamente
Alcanzar alert_date	Crear actividad de alerta
Validar picking	Resolver lot_name → lot_id
Manufactura completada	Crear relaciones consume/produce

9. Consideraciones de Rendimiento

9.1 Índices Optimizados

-- Búsqueda de lotes por nombre (trigram)
CREATE INDEX idx_lots_name_trgm ON inventory.lots USING GIN (name gin_trgm_ops);

-- Quants para FEFO
CREATE INDEX idx_quants_fefo_lookup
    ON inventory.quants(product_id, location_id, removal_date NULLS LAST, in_date)
    WHERE quantity > reserved_quantity;

-- Trazabilidad
CREATE INDEX idx_move_lines_traceability
    ON inventory.move_lines(lot_id, product_id, date)
    WHERE state = 'done';

-- Alertas de caducidad
CREATE INDEX idx_lots_expiry_alert
    ON inventory.lots(alert_date)
    WHERE alert_date IS NOT NULL
      AND expiry_alerted = false;

9.2 Caching

# Cache de estrategias de salida por producto/ubicación
REMOVAL_STRATEGY_CACHE = TTLCache(maxsize=1000, ttl=300)

# Cache de quants para operaciones masivas
QUANTS_CACHE_KEY = (product_id, location_id, lot_id, package_id, owner_id)

10. Referencias

Odoo stock.lot model
Odoo product_expiry module
GS1 General Specifications (https://www.gs1.org/)
ISO 22005:2007 - Traceability in feed and food chain

Documento generado para ERP-SUITE Versión: 1.0 Fecha: 2025-01-09

58 KiB Raw Blame History

SPEC-TRAZABILIDAD-LOTES-SERIES: Trazabilidad Completa de Lotes y Números de Serie

Metadata

1. Resumen Ejecutivo

1.1 Objetivo

1.2 Alcance

1.3 Tipos de Seguimiento

2. Modelo de Datos

2.1 Diagrama Entidad-Relación

2.2 Definición de Tablas

2.2.1 Extensión de products.products

2.2.2 inventory.lots

2.2.3 inventory.quants (extensión)

2.2.4 inventory.move_lines (extensión)

2.2.5 inventory.removal_strategies

3. Algoritmos de Trazabilidad

3.1 Trazabilidad Ascendente (Upstream)

3.2 Estrategias de Salida (FIFO/LIFO/FEFO)

4. Gestión de Fechas de Caducidad

4.1 Cálculo Automático de Fechas

5. Validaciones

5.1 Validaciones de Lote/Serie

6. API REST

6.1 Endpoints

7. Integración con Código de Barras GS1

7.1 Formato GS1-128

8. Reglas de Negocio

8.1 Validaciones

8.2 Comportamientos Automáticos

9. Consideraciones de Rendimiento

9.1 Índices Optimizados

9.2 Caching

10. Referencias

58 KiB

Raw Blame History