feat(pos/workshop): add lightweight workshop/taller module

- Add DB migration v4.4_workshop.sql (sale_id, service_catalog,
  reserved_quantity, SO_RESERVE/SO_RELEASE operation types).
- Extend service_order_engine with inventory reservation, release,
  convert-to-sale, mechanic assignment, and service catalog CRUD.
- Extend service_order_bp with /reserve, /convert-to-sale,
  /assign-mechanic, and /service-catalog endpoints.
- Create workshop Kanban UI: workshop.html, workshop.js, workshop.css.
- Add /pos/workshop route and sidebar navigation (sidebar.js + inline
  templates).
- Add 11 unit tests with mocked cursors.
- Update FASES_IMPLEMENTADAS.md with FASE 9 documentation.

Tests: 92 passing (61 console + 20 Facturapi + 11 workshop).
This commit is contained in:
2026-06-15 05:34:35 +00:00
parent d67887284d
commit ce66212223
15 changed files with 1842 additions and 14 deletions

View File

@@ -3,8 +3,11 @@
States: received -> diagnosis -> waiting_parts -> repair -> quality_check -> ready -> delivered
"""
import contextlib
from datetime import datetime
from services import inventory_engine
VALID_TRANSITIONS = {
'received': ['diagnosis', 'cancelled'],
'diagnosis': ['waiting_parts', 'repair', 'cancelled'],
@@ -30,10 +33,8 @@ def _generate_order_number(conn):
row = cur.fetchone()
last_num = 0
if row and row[0]:
try:
with contextlib.suppress(ValueError):
last_num = int(row[0].split('-')[-1])
except ValueError:
pass
new_num = last_num + 1
cur.close()
return f"{prefix}{new_num:04d}"
@@ -422,7 +423,7 @@ def get_kanban_summary(conn, branch_id=None):
GROUP BY status
""", params)
summary = {status: 0 for status in VALID_TRANSITIONS.keys() if status != 'cancelled'}
summary = {status: 0 for status in VALID_TRANSITIONS if status != 'cancelled'}
for r in cur.fetchall():
summary[r[0]] = r[1]
@@ -438,3 +439,413 @@ def get_kanban_summary(conn, branch_id=None):
cur.close()
summary['overdue'] = overdue
return summary
# ─── Workshop inventory integration ─────────────────────────────────────────
def reserve_item(conn, so_item_id, branch_id, employee_id=None):
"""Reserve inventory for a service order item.
Records a negative SO_RESERVE operation and updates reserved_quantity.
Raises ValueError if stock is insufficient.
"""
cur = conn.cursor()
cur.execute(
"""
SELECT soi.service_order_id, soi.inventory_id, soi.quantity, soi.status,
so.order_number
FROM service_order_items soi
JOIN service_orders so ON so.id = soi.service_order_id
WHERE soi.id = %s
""",
(so_item_id,),
)
row = cur.fetchone()
if not row:
cur.close()
raise ValueError("Service order item not found")
so_id, inventory_id, quantity, status, order_number = row
if status == "cancelled":
cur.close()
raise ValueError("Cannot reserve a cancelled item")
if not inventory_id:
cur.close()
raise ValueError("Item has no inventory linked")
qty = int(quantity)
available = inventory_engine.get_stock(conn, inventory_id, branch_id)
if available < qty:
cur.close()
raise ValueError(f"Insufficient stock. Available: {available}, requested: {qty}")
inventory_engine.record_operation(
conn,
inventory_id,
branch_id,
"SO_RESERVE",
-qty,
reference_id=so_id,
reference_type="service_order_item",
notes=f"Reserva orden {order_number}",
employee_id=employee_id,
)
cur.execute(
"UPDATE service_order_items SET reserved_quantity = %s WHERE id = %s",
(qty, so_item_id),
)
conn.commit()
cur.close()
return {"reserved": qty}
def release_item(conn, so_item_id, employee_id=None):
"""Release a previous reservation for a service order item.
Records a positive SO_RELEASE operation and resets reserved_quantity.
"""
cur = conn.cursor()
cur.execute(
"""
SELECT soi.service_order_id, soi.inventory_id, soi.reserved_quantity,
so.branch_id, so.order_number
FROM service_order_items soi
JOIN service_orders so ON so.id = soi.service_order_id
WHERE soi.id = %s
""",
(so_item_id,),
)
row = cur.fetchone()
if not row:
cur.close()
raise ValueError("Service order item not found")
so_id, inventory_id, reserved_qty, branch_id, order_number = row
if not inventory_id or not reserved_qty:
cur.close()
return {"released": 0}
qty = int(reserved_qty)
inventory_engine.record_operation(
conn,
inventory_id,
branch_id,
"SO_RELEASE",
qty,
reference_id=so_id,
reference_type="service_order_item",
notes=f"Liberacion reserva orden {order_number}",
employee_id=employee_id,
)
cur.execute(
"UPDATE service_order_items SET reserved_quantity = 0 WHERE id = %s",
(so_item_id,),
)
conn.commit()
cur.close()
return {"released": qty}
def _consume_item_inventory(conn, so_item, sale_id, order_number, branch_id, employee_id=None):
"""Release reservation and record final SALE for a service order item."""
inventory_id = so_item.get("inventory_id")
reserved_qty = so_item.get("reserved_quantity", 0)
qty = int(so_item.get("quantity", 0))
if not inventory_id or qty <= 0:
return
if reserved_qty:
inventory_engine.record_operation(
conn,
inventory_id,
branch_id,
"SO_RELEASE",
int(reserved_qty),
reference_id=so_item.get("service_order_id"),
reference_type="service_order",
notes=f"Liberacion para venta orden {order_number}",
employee_id=employee_id,
)
inventory_engine.record_operation(
conn,
inventory_id,
branch_id,
"SALE",
-qty,
reference_id=sale_id,
reference_type="sale",
notes=f"Venta desde orden {order_number}",
employee_id=employee_id,
)
def convert_to_sale(conn, so_id, sale_data, employee_id=None):
"""Convert a service order into a POS sale.
sale_data keys:
payment_method: 'efectivo' | 'transferencia' | 'tarjeta' | 'mixto'
sale_type: 'cash' | 'credit' | 'mixed'
register_id: int (optional)
amount_paid: float (optional)
payment_details: list (optional)
notes: str (optional)
Returns dict with sale_id, total, items_count.
"""
cur = conn.cursor()
so = get_service_order(conn, so_id)
if not so:
cur.close()
raise ValueError("Service order not found")
if so["status"] == "cancelled":
cur.close()
raise ValueError("Cannot convert a cancelled service order")
if so.get("sale_id"):
cur.close()
raise ValueError("Service order already converted to sale")
branch_id = so["branch_id"]
customer_id = so["customer_id"]
# Build sale items from SO parts and labor
sale_items = []
for item in so.get("items", []):
if item.get("status") == "cancelled":
continue
qty = int(item.get("quantity", 1))
unit_price = float(item.get("unit_price") or 0)
unit_cost = float(item.get("unit_cost") or 0)
sale_items.append(
{
"inventory_id": item.get("inventory_id"),
"part_number": item.get("part_number") or "PART",
"name": item.get("name") or "Refaccion",
"quantity": qty,
"unit_price": unit_price,
"unit_cost": unit_cost,
"tax_rate": 0.16,
}
)
for labor in so.get("labor", []):
if labor.get("status") == "cancelled":
continue
sale_items.append(
{
"inventory_id": None,
"part_number": "SERV",
"name": labor.get("description") or "Mano de obra",
"quantity": 1,
"unit_price": float(labor.get("total_cost") or 0),
"unit_cost": 0,
"tax_rate": 0.16,
}
)
if not sale_items:
cur.close()
raise ValueError("No items or labor to invoice")
# Calculate totals
subtotal = 0.0
tax_total = 0.0
for item in sale_items:
item_subtotal = item["quantity"] * item["unit_price"]
item_tax = item_subtotal * item["tax_rate"]
item["subtotal"] = item_subtotal
item["tax_amount"] = item_tax
subtotal += item_subtotal
tax_total += item_tax
total = subtotal + tax_total
payment_method = sale_data.get("payment_method", "efectivo")
sale_type = sale_data.get("sale_type", "cash")
register_id = sale_data.get("register_id")
amount_paid = float(sale_data.get("amount_paid", total if sale_type == "cash" else 0))
change_given = max(amount_paid - total, 0) if sale_type == "cash" and payment_method == "efectivo" else 0
notes = sale_data.get("notes") or f"Orden de servicio {so['order_number']}"
metodo_pago_sat = "PPD" if sale_type == "credit" else "PUE"
forma_pago_map = {"efectivo": "01", "transferencia": "03", "tarjeta": "04", "mixto": "99"}
forma_pago_sat = forma_pago_map.get(payment_method, "99")
cur.execute(
"""
INSERT INTO sales
(branch_id, customer_id, employee_id, register_id, sale_type,
payment_method, subtotal, discount_total, tax_total, total,
amount_paid, change_given, metodo_pago_sat, forma_pago_sat,
status, notes)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'completed', %s)
RETURNING id, created_at
""",
(
branch_id,
customer_id,
employee_id,
register_id,
sale_type,
payment_method,
subtotal,
0,
tax_total,
total,
amount_paid,
change_given,
metodo_pago_sat,
forma_pago_sat,
notes,
),
)
sale_id, _created_at = cur.fetchone()
# Insert sale_items
for item in sale_items:
cur.execute(
"""
INSERT INTO sale_items
(sale_id, inventory_id, part_number, name, quantity,
unit_price, unit_cost, tax_rate, tax_amount, subtotal)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
sale_id,
item["inventory_id"],
item["part_number"],
item["name"],
item["quantity"],
item["unit_price"],
item["unit_cost"],
item["tax_rate"],
item["tax_amount"],
item["subtotal"],
),
)
# Consume inventory for parts
for item in so.get("items", []):
if item.get("status") == "cancelled":
continue
_consume_item_inventory(
conn, item, sale_id, so["order_number"], branch_id, employee_id
)
# Link order to sale
cur.execute("UPDATE service_orders SET sale_id = %s WHERE id = %s", (sale_id, so_id))
conn.commit()
cur.close()
return {"sale_id": sale_id, "total": total, "items_count": len(sale_items)}
def assign_mechanic(conn, so_id, employee_id):
"""Assign a mechanic/technician to a service order."""
cur = conn.cursor()
cur.execute("SELECT id FROM service_orders WHERE id = %s", (so_id,))
if not cur.fetchone():
cur.close()
raise ValueError("Service order not found")
cur.execute(
"UPDATE service_orders SET employee_id = %s WHERE id = %s",
(employee_id, so_id),
)
conn.commit()
cur.close()
return {"employee_id": employee_id}
# ─── Service catalog (reusable labor concepts) ───────────────────────────────
def list_service_catalog(conn, active_only=True):
"""List reusable labor/service concepts."""
cur = conn.cursor()
where = "WHERE is_active = true" if active_only else ""
cur.execute(
f"""
SELECT id, tenant_id, name, description, suggested_hours, suggested_rate,
is_active, created_at, updated_at
FROM service_catalog
{where}
ORDER BY name
"""
)
items = []
for r in cur.fetchall():
items.append(
{
"id": r[0],
"tenant_id": r[1],
"name": r[2],
"description": r[3],
"suggested_hours": float(r[4]) if r[4] else 0,
"suggested_rate": float(r[5]) if r[5] else 0,
"is_active": r[6],
"created_at": str(r[7]) if r[7] else None,
"updated_at": str(r[8]) if r[8] else None,
}
)
cur.close()
return items
def create_service_catalog_item(conn, tenant_id, data):
"""Create a reusable labor concept."""
cur = conn.cursor()
cur.execute(
"""
INSERT INTO service_catalog
(tenant_id, name, description, suggested_hours, suggested_rate, is_active)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING id
""",
(
tenant_id,
data.get("name"),
data.get("description"),
data.get("suggested_hours", 0),
data.get("suggested_rate", 0),
data.get("is_active", True),
),
)
item_id = cur.fetchone()[0]
conn.commit()
cur.close()
return {"id": item_id}
def update_service_catalog_item(conn, item_id, data):
"""Update a reusable labor concept."""
cur = conn.cursor()
allowed = ["name", "description", "suggested_hours", "suggested_rate", "is_active"]
sets = []
vals = []
for field in allowed:
if field in data:
sets.append(f"{field} = %s")
vals.append(data[field])
if not sets:
cur.close()
return False
vals.append(item_id)
cur.execute(f"UPDATE service_catalog SET {', '.join(sets)} WHERE id = %s", vals)
conn.commit()
cur.close()
return True
def delete_service_catalog_item(conn, item_id):
"""Soft-delete a reusable labor concept by setting is_active = false."""
cur = conn.cursor()
cur.execute(
"UPDATE service_catalog SET is_active = false WHERE id = %s", (item_id,)
)
conn.commit()
cur.close()
return True