# /home/Autopartes/pos/services/inventory_engine.py """Inventory operations engine. All stock mutations go through here. Stock is NEVER stored as a field — it is always computed as: SUM(inventory_operations.quantity) WHERE inventory_id = X AND branch_id = Y Operations are append-only. No UPDATE, no DELETE on inventory_operations. """ from flask import g from services.audit import log_action from services.redis_stock_cache import ( get_cached_stock, set_cached_stock, invalidate_stock ) def _safe_g(attr, default=None): """Safely read flask.g attribute outside of app context.""" try: return getattr(g, attr, default) except RuntimeError: return default def get_stock(conn, inventory_id, branch_id=None): """Get current stock for an inventory item. Optionally filter by branch. Uses Redis cache first, then inventory_stock_summary, falls back to PostgreSQL SUM query. """ # Try Redis first cached = get_cached_stock(inventory_id, branch_id) if cached is not None: return cached # Use inventory_stock_summary (O(1) lookup) cur = conn.cursor() if branch_id: cur.execute( "SELECT stock FROM inventory_stock_summary WHERE inventory_id = %s AND branch_id = %s", (inventory_id, branch_id) ) else: cur.execute( "SELECT stock FROM inventory_stock_summary WHERE inventory_id = %s", (inventory_id,) ) row = cur.fetchone() if row is not None: cur.close() set_cached_stock(inventory_id, row[0], branch_id) return row[0] # Fallback to PostgreSQL SUM (legacy, should not reach here if trigger works) if branch_id: cur.execute( "SELECT COALESCE(SUM(quantity), 0) FROM inventory_operations WHERE inventory_id = %s AND branch_id = %s", (inventory_id, branch_id) ) else: cur.execute( "SELECT COALESCE(SUM(quantity), 0) FROM inventory_operations WHERE inventory_id = %s", (inventory_id,) ) stock = cur.fetchone()[0] cur.close() # Cache the result set_cached_stock(inventory_id, stock, branch_id) return stock def get_stock_bulk(conn, branch_id=None): """Get stock for all items. Returns dict {inventory_id: stock_quantity}. Uses inventory_stock_summary for O(1) bulk lookup. """ cur = conn.cursor() if branch_id: cur.execute(""" SELECT inventory_id, stock FROM inventory_stock_summary WHERE branch_id = %s """, (branch_id,)) else: cur.execute(""" SELECT inventory_id, stock FROM inventory_stock_summary """) stock_map = {r[0]: r[1] for r in cur.fetchall()} cur.close() # Populate Redis cache with results for inv_id, qty in stock_map.items(): set_cached_stock(inv_id, qty, branch_id) return stock_map def record_operation(conn, inventory_id, branch_id, operation_type, quantity, reference_id=None, reference_type=None, cost_at_time=None, notes=None, employee_id=None): """Record a single inventory operation. Does NOT commit — caller controls transaction. Args: quantity: positive for entries (PURCHASE, RETURN, INITIAL), negative for exits (SALE) operation_type: SALE, PURCHASE, RETURN, ADJUST, TRANSFER, INITIAL, QUOTE_RESERVE, QUOTE_RELEASE """ cur = conn.cursor() cur.execute(""" INSERT INTO inventory_operations (inventory_id, branch_id, operation_type, quantity, reference_id, reference_type, cost_at_time, employee_id, device_id, notes) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( inventory_id, branch_id, operation_type, quantity, reference_id, reference_type, cost_at_time, employee_id if employee_id is not None else _safe_g('employee_id'), _safe_g('device_id'), notes )) op_id = cur.fetchone()[0] cur.close() return op_id def record_purchase(conn, inventory_id, branch_id, quantity, unit_cost, supplier_invoice=None, notes=None): """Record a purchase entry. Updates weighted average cost on the inventory item. IMPORTANT: Cost is stored globally on the inventory item (not per-branch), so we must use TOTAL stock across ALL branches when computing the weighted average. Using branch-scoped stock would produce incorrect averages when the same item exists in multiple branches. Uses SELECT ... FOR UPDATE to prevent race conditions on concurrent purchases of the same item. """ from decimal import Decimal, ROUND_HALF_UP TWO = Decimal('0.01') cur = conn.cursor() cur.execute("SELECT cost FROM inventory WHERE id = %s FOR UPDATE", (inventory_id,)) row = cur.fetchone() current_cost = Decimal(str(row[0] or 0)) if row else Decimal('0') # Use GLOBAL stock (all branches) because cost is a global field on the inventory item current_stock = Decimal(str(get_stock(conn, inventory_id, branch_id=None) or 0)) qty_dec = Decimal(str(quantity)) unit_cost_dec = Decimal(str(unit_cost)) # Weighted average cost (Decimal arithmetic) stock_plus_qty = current_stock + qty_dec if stock_plus_qty > 0: numerator = (current_cost * current_stock) + (unit_cost_dec * qty_dec) new_cost = (numerator / stock_plus_qty).quantize(TWO, rounding=ROUND_HALF_UP) else: new_cost = unit_cost_dec # Update cost on inventory item cur.execute("UPDATE inventory SET cost = %s WHERE id = %s", (float(new_cost), inventory_id)) cur.close() ref_note = f"Compra: {quantity} uds @ ${float(unit_cost_dec):.2f}" if supplier_invoice: ref_note += f" | Factura: {supplier_invoice}" if notes: ref_note += f" | {notes}" result = record_operation( conn, inventory_id, branch_id, 'PURCHASE', quantity, cost_at_time=float(unit_cost_dec), notes=ref_note ) invalidate_stock(inventory_id, branch_id) invalidate_stock(inventory_id, None) return result def record_sale(conn, inventory_id, branch_id, quantity, sale_id=None, cost_at_time=None, remaining_stock=None): """Record a sale (negative quantity). NOT exposed via HTTP endpoint — called directly by the POS blueprint which imports inventory_engine as part of the full sale transaction. Args: remaining_stock: optional pre-calculated stock to avoid redundant SUM query. If None, stock will be calculated internally. """ op_id = record_operation( conn, inventory_id, branch_id, 'SALE', -abs(quantity), reference_id=sale_id, reference_type='sale', cost_at_time=cost_at_time ) # Invalidate cache immediately invalidate_stock(inventory_id, branch_id) invalidate_stock(inventory_id, None) # Check if stock hit zero — push to owner (best-effort) try: remaining = remaining_stock if remaining_stock is not None else get_stock(conn, inventory_id, branch_id) if remaining <= 0: cur = conn.cursor() cur.execute("SELECT part_number, name FROM inventory WHERE id = %s", (inventory_id,)) inv_row = cur.fetchone() cur.close() if inv_row: from services.push_service import notify_owner notify_owner( conn, 'Stock en Cero', f'{inv_row[1] or inv_row[0]} se quedo sin existencias', '/pos' ) except Exception: pass # Push failures never block sales return op_id def record_return(conn, inventory_id, branch_id, quantity, sale_id=None, notes=None): """Record a customer return (positive quantity).""" result = record_operation( conn, inventory_id, branch_id, 'RETURN', abs(quantity), reference_id=sale_id, reference_type='return', notes=notes ) invalidate_stock(inventory_id, branch_id) invalidate_stock(inventory_id, None) return result def record_adjustment(conn, inventory_id, branch_id, quantity, reason): """Record a manual stock adjustment. Reason is mandatory.""" if not reason or len(reason.strip()) < 3: raise ValueError("Adjustment reason is mandatory (min 3 characters)") log_action(conn, 'STOCK_ADJUST', 'inventory', inventory_id, old_value={'stock': get_stock(conn, inventory_id, branch_id)}, new_value={'adjustment': quantity, 'reason': reason}) result = record_operation( conn, inventory_id, branch_id, 'ADJUST', quantity, notes=f"Ajuste: {reason}" ) invalidate_stock(inventory_id, branch_id) invalidate_stock(inventory_id, None) return result def record_transfer(conn, inventory_id, from_branch_id, to_branch_id, quantity, notes=None): """Transfer stock between branches. Creates two operations (out + in).""" out_id = record_operation( conn, inventory_id, from_branch_id, 'TRANSFER', -abs(quantity), notes=f"Transferencia a sucursal {to_branch_id}" + (f" | {notes}" if notes else "") ) in_id = record_operation( conn, inventory_id, to_branch_id, 'TRANSFER', abs(quantity), notes=f"Transferencia desde sucursal {from_branch_id}" + (f" | {notes}" if notes else "") ) invalidate_stock(inventory_id, from_branch_id) invalidate_stock(inventory_id, to_branch_id) invalidate_stock(inventory_id, None) return out_id, in_id def record_initial(conn, inventory_id, branch_id, quantity, cost=None): """Record initial stock load.""" result = record_operation( conn, inventory_id, branch_id, 'INITIAL', quantity, cost_at_time=cost, notes="Carga inicial de inventario" ) invalidate_stock(inventory_id, branch_id) invalidate_stock(inventory_id, None) return result def get_alerts(conn, branch_id=None): """Get stock alerts: zero stock, below minimum, above maximum.""" stock_map = get_stock_bulk(conn, branch_id) cur = conn.cursor() where = "WHERE i.is_active = true" params = [] if branch_id: where += " AND i.branch_id = %s" params.append(branch_id) cur.execute(f""" SELECT i.id, i.part_number, i.name, i.min_stock, i.max_stock, i.branch_id FROM inventory i {where} """, params) alerts = [] for row in cur.fetchall(): inv_id, part_num, name, min_s, max_s, br_id = row stock = stock_map.get(inv_id, 0) if stock <= 0: alerts.append({'type': 'zero', 'severity': 'critical', 'inventory_id': inv_id, 'part_number': part_num, 'name': name, 'stock': stock, 'branch_id': br_id}) elif min_s and stock < min_s: alerts.append({'type': 'low', 'severity': 'warning', 'inventory_id': inv_id, 'part_number': part_num, 'name': name, 'stock': stock, 'min_stock': min_s, 'branch_id': br_id}) elif max_s and stock > max_s: alerts.append({'type': 'over', 'severity': 'info', 'inventory_id': inv_id, 'part_number': part_num, 'name': name, 'stock': stock, 'max_stock': max_s, 'branch_id': br_id}) cur.close() return alerts def get_movement_history(conn, inventory_id, limit=50): """Get operation history for a specific item.""" cur = conn.cursor() cur.execute(""" SELECT io.id, io.operation_type, io.quantity, io.cost_at_time, io.notes, io.created_at, e.name as employee_name, io.branch_id FROM inventory_operations io LEFT JOIN employees e ON io.employee_id = e.id WHERE io.inventory_id = %s ORDER BY io.created_at DESC LIMIT %s """, (inventory_id, limit)) history = [] for r in cur.fetchall(): history.append({ 'id': r[0], 'type': r[1], 'quantity': r[2], 'cost': float(r[3]) if r[3] else None, 'notes': r[4], 'date': str(r[5]), 'employee': r[6], 'branch_id': r[7] }) cur.close() return history