# 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 ```sql -- 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 ```sql 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) ```sql -- 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) ```sql -- 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 ```sql 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) ```python 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) ```python 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 ```python 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 ```python 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 ```yaml 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 ```python 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 ```sql -- 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 ```python # 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