f54e1bebc3
- Created warehouse_orders.py module with 8 order management functions - Added /warehouse/set-orders-on-boxes route and 7 API endpoints - Implemented 4-tab interface: assign, find, move, and view orders - Changed assign input from dropdown to text field with BOX validation - Fixed location join issue in warehouse.py (use boxes_crates.location_id) - Added debouncing flag to prevent multiple rapid form submissions - Added page refresh after successful order assignment - Disabled assign button during processing - Added page refresh with 2 second delay for UX feedback - Added CP code validation in inventory page - Improved modal styling with theme support - Fixed set_boxes_locations page to refresh box info after assignments
393 lines
11 KiB
Python
393 lines
11 KiB
Python
"""
|
|
Warehouse Orders Management - Helper Functions
|
|
Provides functions for managing orders on boxes
|
|
"""
|
|
import logging
|
|
import pymysql
|
|
from app.database import get_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_unassigned_orders(limit=100, offset=0):
|
|
"""Get all orders not yet assigned to any box
|
|
|
|
Returns:
|
|
List of unassigned orders
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
id,
|
|
CP_full_code,
|
|
operator_code,
|
|
quality_code,
|
|
date,
|
|
time,
|
|
approved_quantity,
|
|
rejected_quantity,
|
|
created_at
|
|
FROM scanfg_orders
|
|
WHERE box_id IS NULL
|
|
ORDER BY created_at DESC
|
|
LIMIT %s OFFSET %s
|
|
""", (limit, offset))
|
|
|
|
columns = [col[0] for col in cursor.description]
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
row_dict = {columns[i]: row[i] for i in range(len(columns))}
|
|
if row_dict.get('date'):
|
|
row_dict['date'] = str(row_dict['date'])
|
|
if row_dict.get('time'):
|
|
row_dict['time'] = str(row_dict['time'])
|
|
if row_dict.get('created_at'):
|
|
row_dict['created_at'] = str(row_dict['created_at'])
|
|
results.append(row_dict)
|
|
cursor.close()
|
|
|
|
return results if results else []
|
|
except Exception as e:
|
|
logger.error(f"Error getting unassigned orders: {e}")
|
|
return []
|
|
|
|
|
|
def get_orders_by_box(box_id):
|
|
"""Get all orders assigned to a specific box
|
|
|
|
Args:
|
|
box_id: ID of the box
|
|
|
|
Returns:
|
|
Tuple: (success: bool, data: dict, status_code: int)
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Get box info
|
|
cursor.execute("""
|
|
SELECT id, box_number, status, location_id
|
|
FROM boxes_crates
|
|
WHERE id = %s
|
|
""", (box_id,))
|
|
|
|
box = cursor.fetchone()
|
|
|
|
if not box:
|
|
cursor.close()
|
|
return False, {'error': 'Box not found'}, 404
|
|
|
|
# Get all orders in this box
|
|
cursor.execute("""
|
|
SELECT
|
|
id,
|
|
CP_full_code,
|
|
operator_code,
|
|
quality_code,
|
|
date,
|
|
time,
|
|
approved_quantity,
|
|
rejected_quantity,
|
|
created_at
|
|
FROM scanfg_orders
|
|
WHERE box_id = %s
|
|
ORDER BY created_at DESC
|
|
""", (box_id,))
|
|
|
|
orders = cursor.fetchall()
|
|
columns = [col[0] for col in cursor.description]
|
|
|
|
orders_list = []
|
|
for order in orders:
|
|
order_dict = {columns[i]: order[i] for i in range(len(columns))}
|
|
if order_dict.get('date'):
|
|
order_dict['date'] = str(order_dict['date'])
|
|
if order_dict.get('time'):
|
|
order_dict['time'] = str(order_dict['time'])
|
|
if order_dict.get('created_at'):
|
|
order_dict['created_at'] = str(order_dict['created_at'])
|
|
orders_list.append(order_dict)
|
|
|
|
# Get location info if box has a location
|
|
location_info = None
|
|
if box[3]: # location_id
|
|
cursor.execute("""
|
|
SELECT id, location_code, size, description
|
|
FROM warehouse_locations
|
|
WHERE id = %s
|
|
""", (box[3],))
|
|
loc = cursor.fetchone()
|
|
if loc:
|
|
location_info = {
|
|
'id': loc[0],
|
|
'location_code': loc[1],
|
|
'size': loc[2],
|
|
'description': loc[3]
|
|
}
|
|
|
|
box_data = {
|
|
'id': box[0],
|
|
'box_number': box[1],
|
|
'status': box[2],
|
|
'location_id': box[3],
|
|
'location': location_info,
|
|
'orders_count': len(orders_list)
|
|
}
|
|
|
|
cursor.close()
|
|
|
|
return True, {'box': box_data, 'orders': orders_list}, 200
|
|
except Exception as e:
|
|
logger.error(f"Error getting orders by box: {e}")
|
|
return False, {'error': str(e)}, 500
|
|
|
|
|
|
def search_orders_by_cp_code(cp_code):
|
|
"""Search for orders by CP code
|
|
|
|
Args:
|
|
cp_code: CP code to search for (can be partial)
|
|
|
|
Returns:
|
|
List of matching orders
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
search_term = cp_code.replace('-', '').strip().upper()
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
id,
|
|
CP_full_code,
|
|
operator_code,
|
|
quality_code,
|
|
date,
|
|
time,
|
|
approved_quantity,
|
|
rejected_quantity,
|
|
box_id,
|
|
created_at
|
|
FROM scanfg_orders
|
|
WHERE REPLACE(CP_full_code, '-', '') LIKE %s
|
|
ORDER BY created_at DESC
|
|
LIMIT 100
|
|
""", (f"{search_term}%",))
|
|
|
|
columns = [col[0] for col in cursor.description]
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
row_dict = {columns[i]: row[i] for i in range(len(columns))}
|
|
if row_dict.get('date'):
|
|
row_dict['date'] = str(row_dict['date'])
|
|
if row_dict.get('time'):
|
|
row_dict['time'] = str(row_dict['time'])
|
|
if row_dict.get('created_at'):
|
|
row_dict['created_at'] = str(row_dict['created_at'])
|
|
results.append(row_dict)
|
|
cursor.close()
|
|
|
|
return results if results else []
|
|
except Exception as e:
|
|
logger.error(f"Error searching orders: {e}")
|
|
return []
|
|
|
|
|
|
def assign_order_to_box(order_id, box_id):
|
|
"""Assign an order to a box
|
|
|
|
Args:
|
|
order_id: ID of the order (scanfg_orders.id)
|
|
box_id: ID of the box to assign to
|
|
|
|
Returns:
|
|
Tuple: (success: bool, message: str, status_code: int)
|
|
"""
|
|
try:
|
|
if not order_id or not box_id:
|
|
return False, 'Order ID and box ID are required', 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Verify order exists
|
|
cursor.execute("SELECT CP_full_code FROM scanfg_orders WHERE id = %s", (order_id,))
|
|
order = cursor.fetchone()
|
|
|
|
if not order:
|
|
cursor.close()
|
|
return False, 'Order not found', 404
|
|
|
|
# Verify box exists
|
|
cursor.execute("SELECT box_number FROM boxes_crates WHERE id = %s", (box_id,))
|
|
box = cursor.fetchone()
|
|
|
|
if not box:
|
|
cursor.close()
|
|
return False, 'Box not found', 404
|
|
|
|
# Assign order to box
|
|
cursor.execute("""
|
|
UPDATE scanfg_orders
|
|
SET box_id = %s
|
|
WHERE id = %s
|
|
""", (box_id, order_id))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, f'Order "{order[0]}" assigned to box "{box[0]}"', 200
|
|
except Exception as e:
|
|
logger.error(f"Error assigning order to box: {e}")
|
|
return False, f'Error: {str(e)}', 500
|
|
|
|
|
|
def move_order_to_box(order_id, new_box_id):
|
|
"""Move an order from one box to another
|
|
|
|
Args:
|
|
order_id: ID of the order
|
|
new_box_id: ID of the destination box
|
|
|
|
Returns:
|
|
Tuple: (success: bool, message: str, status_code: int)
|
|
"""
|
|
try:
|
|
if not order_id or not new_box_id:
|
|
return False, 'Order ID and destination box ID are required', 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Get order info
|
|
cursor.execute("""
|
|
SELECT CP_full_code, box_id FROM scanfg_orders WHERE id = %s
|
|
""", (order_id,))
|
|
|
|
order = cursor.fetchone()
|
|
|
|
if not order:
|
|
cursor.close()
|
|
return False, 'Order not found', 404
|
|
|
|
old_box_id = order[1]
|
|
|
|
# Verify new box exists
|
|
cursor.execute("SELECT box_number FROM boxes_crates WHERE id = %s", (new_box_id,))
|
|
new_box = cursor.fetchone()
|
|
|
|
if not new_box:
|
|
cursor.close()
|
|
return False, 'Destination box not found', 404
|
|
|
|
# Get old box info if it exists
|
|
old_box_number = None
|
|
if old_box_id:
|
|
cursor.execute("SELECT box_number FROM boxes_crates WHERE id = %s", (old_box_id,))
|
|
old_box = cursor.fetchone()
|
|
if old_box:
|
|
old_box_number = old_box[0]
|
|
|
|
# Move order to new box
|
|
cursor.execute("""
|
|
UPDATE scanfg_orders
|
|
SET box_id = %s
|
|
WHERE id = %s
|
|
""", (new_box_id, order_id))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
if old_box_number:
|
|
message = f'Order "{order[0]}" moved from "{old_box_number}" to "{new_box[0]}"'
|
|
else:
|
|
message = f'Order "{order[0]}" moved to "{new_box[0]}"'
|
|
|
|
return True, message, 200
|
|
except Exception as e:
|
|
logger.error(f"Error moving order to box: {e}")
|
|
return False, f'Error: {str(e)}', 500
|
|
|
|
|
|
def unassign_order_from_box(order_id):
|
|
"""Remove order from its assigned box
|
|
|
|
Args:
|
|
order_id: ID of the order
|
|
|
|
Returns:
|
|
Tuple: (success: bool, message: str, status_code: int)
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Get order info
|
|
cursor.execute("""
|
|
SELECT CP_full_code, box_id FROM scanfg_orders WHERE id = %s
|
|
""", (order_id,))
|
|
|
|
order = cursor.fetchone()
|
|
|
|
if not order:
|
|
cursor.close()
|
|
return False, 'Order not found', 404
|
|
|
|
# Remove from box
|
|
cursor.execute("""
|
|
UPDATE scanfg_orders
|
|
SET box_id = NULL
|
|
WHERE id = %s
|
|
""", (order_id,))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, f'Order "{order[0]}" removed from box', 200
|
|
except Exception as e:
|
|
logger.error(f"Error unassigning order: {e}")
|
|
return False, f'Error: {str(e)}', 500
|
|
|
|
|
|
def get_all_boxes_summary():
|
|
"""Get summary of all boxes with order counts
|
|
|
|
Returns:
|
|
List of boxes with order counts
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
bc.id,
|
|
bc.box_number,
|
|
bc.status,
|
|
wl.location_code,
|
|
wl.id as location_id,
|
|
COUNT(so.id) as order_count
|
|
FROM boxes_crates bc
|
|
LEFT JOIN warehouse_locations wl ON bc.location_id = wl.id
|
|
LEFT JOIN scanfg_orders so ON bc.id = so.box_id
|
|
GROUP BY bc.id, bc.box_number, bc.status, wl.location_code, wl.id
|
|
ORDER BY bc.box_number ASC
|
|
""")
|
|
|
|
columns = [col[0] for col in cursor.description]
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
row_dict = {columns[i]: row[i] for i in range(len(columns))}
|
|
results.append(row_dict)
|
|
cursor.close()
|
|
|
|
return results if results else []
|
|
except Exception as e:
|
|
logger.error(f"Error getting boxes summary: {e}")
|
|
return []
|