b15cc93b9d
- Fixed 3 JavaScript syntax errors in fg_scan.html (lines 951, 840-950, 1175-1215) - Restored form field validation with proper null safety checks - Re-enabled auto-advance between form fields - Re-enabled CP code auto-complete with hyphen detection - Updated validation error messages with clear format specifications and examples - Added autocomplete='off' to all input fields - Removed auto-prefix correction feature - Updated warehouse routes and modules for box assignment workflow - Added/improved database initialization scripts - Updated requirements.txt dependencies Format specifications implemented: - Operator Code: OP + 2 digits (example: OP01, OP99) - CP Code: CP + 8 digits + hyphen + 4 digits (example: CP00000000-0001) - OC1/OC2 Codes: OC + 2 digits (example: OC01, OC99) - Defect Code: 3 digits only
438 lines
13 KiB
Python
438 lines
13 KiB
Python
"""
|
|
Warehouse Module - Helper Functions
|
|
Provides functions for warehouse operations
|
|
"""
|
|
import logging
|
|
from app.database import get_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def ensure_warehouse_locations_table():
|
|
"""Ensure warehouse_locations table exists"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS warehouse_locations (
|
|
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
|
location_code VARCHAR(12) UNIQUE NOT NULL,
|
|
size INT,
|
|
description VARCHAR(250),
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
INDEX idx_location_code (location_code)
|
|
)
|
|
""")
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
logger.info("warehouse_locations table ensured")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error ensuring warehouse_locations table: {e}")
|
|
return False
|
|
|
|
|
|
def add_location(location_code, size, description):
|
|
"""Add a new warehouse location"""
|
|
try:
|
|
ensure_warehouse_locations_table()
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
INSERT INTO warehouse_locations (location_code, size, description)
|
|
VALUES (%s, %s, %s)
|
|
""", (location_code, size if size else None, description))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, "Location added successfully."
|
|
except Exception as e:
|
|
if "Duplicate entry" in str(e):
|
|
return False, f"Failed: Location code '{location_code}' already exists."
|
|
logger.error(f"Error adding location: {e}")
|
|
return False, f"Error adding location: {str(e)}"
|
|
|
|
|
|
def get_all_locations():
|
|
"""Get all warehouse locations"""
|
|
try:
|
|
ensure_warehouse_locations_table()
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT id, location_code, size, description, created_at, updated_at
|
|
FROM warehouse_locations
|
|
ORDER BY id DESC
|
|
""")
|
|
|
|
locations = cursor.fetchall()
|
|
cursor.close()
|
|
|
|
result = []
|
|
for loc in locations:
|
|
result.append({
|
|
'id': loc[0],
|
|
'location_code': loc[1],
|
|
'size': loc[2],
|
|
'description': loc[3],
|
|
'created_at': loc[4],
|
|
'updated_at': loc[5]
|
|
})
|
|
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Error getting locations: {e}")
|
|
return []
|
|
|
|
|
|
def get_location_by_id(location_id):
|
|
"""Get a specific location by ID"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT id, location_code, size, description, created_at, updated_at
|
|
FROM warehouse_locations
|
|
WHERE id = %s
|
|
""", (location_id,))
|
|
|
|
loc = cursor.fetchone()
|
|
cursor.close()
|
|
|
|
if loc:
|
|
return {
|
|
'id': loc[0],
|
|
'location_code': loc[1],
|
|
'size': loc[2],
|
|
'description': loc[3],
|
|
'created_at': loc[4],
|
|
'updated_at': loc[5]
|
|
}
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting location: {e}")
|
|
return None
|
|
|
|
|
|
def update_location(location_id, location_code=None, size=None, description=None):
|
|
"""Update a warehouse location
|
|
|
|
Args:
|
|
location_id: ID of location to update
|
|
location_code: New location code (optional - cannot be changed in form)
|
|
size: New size (optional)
|
|
description: New description (optional)
|
|
"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Build update query dynamically
|
|
updates = []
|
|
params = []
|
|
|
|
if location_code:
|
|
updates.append("location_code = %s")
|
|
params.append(location_code)
|
|
if size is not None:
|
|
updates.append("size = %s")
|
|
params.append(size)
|
|
if description is not None:
|
|
updates.append("description = %s")
|
|
params.append(description)
|
|
|
|
if not updates:
|
|
return False, "No fields to update"
|
|
|
|
params.append(location_id)
|
|
|
|
query = f"UPDATE warehouse_locations SET {', '.join(updates)} WHERE id = %s"
|
|
cursor.execute(query, params)
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, "Location updated successfully."
|
|
except Exception as e:
|
|
if "Duplicate entry" in str(e):
|
|
return False, f"Failed: Location code already exists."
|
|
logger.error(f"Error updating location: {e}")
|
|
return False, f"Error updating location: {str(e)}"
|
|
|
|
|
|
def delete_location(location_id):
|
|
"""Delete a warehouse location"""
|
|
try:
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("DELETE FROM warehouse_locations WHERE id = %s", (location_id,))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, "Location deleted successfully."
|
|
except Exception as e:
|
|
logger.error(f"Error deleting location: {e}")
|
|
return False, f"Error deleting location: {str(e)}"
|
|
|
|
|
|
def delete_multiple_locations(location_ids):
|
|
"""Delete multiple warehouse locations"""
|
|
try:
|
|
if not location_ids:
|
|
return False, "No locations to delete."
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
deleted_count = 0
|
|
for loc_id in location_ids:
|
|
try:
|
|
cursor.execute("DELETE FROM warehouse_locations WHERE id = %s", (int(loc_id),))
|
|
if cursor.rowcount > 0:
|
|
deleted_count += 1
|
|
except:
|
|
pass
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, f"Deleted {deleted_count} location(s)."
|
|
except Exception as e:
|
|
logger.error(f"Error deleting multiple locations: {e}")
|
|
return False, f"Error deleting locations: {str(e)}"
|
|
|
|
|
|
# ============================================================================
|
|
# Set Boxes Locations - Functions for assigning boxes to locations
|
|
# ============================================================================
|
|
|
|
def search_box_by_number(box_number):
|
|
"""Search for a box by its number
|
|
|
|
Returns:
|
|
tuple: (success: bool, box_data: dict or None, status_code: int)
|
|
"""
|
|
try:
|
|
if not box_number or not str(box_number).strip():
|
|
return False, None, 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
b.id,
|
|
b.box_number,
|
|
b.status,
|
|
b.location_id,
|
|
COALESCE(l.location_code, 'Not assigned') as location_code,
|
|
b.created_at
|
|
FROM boxes_crates b
|
|
LEFT JOIN warehouse_locations l ON b.location_id = l.id
|
|
WHERE b.box_number = %s
|
|
""", (str(box_number).strip(),))
|
|
|
|
result = cursor.fetchone()
|
|
cursor.close()
|
|
|
|
if not result:
|
|
return False, None, 404
|
|
|
|
box_data = {
|
|
'id': result[0],
|
|
'box_number': result[1],
|
|
'status': result[2],
|
|
'location_id': result[3],
|
|
'location_code': result[4],
|
|
'created_at': str(result[5])
|
|
}
|
|
|
|
return True, box_data, 200
|
|
except Exception as e:
|
|
logger.error(f"Error searching box: {e}")
|
|
return False, None, 500
|
|
|
|
|
|
def search_location_with_boxes(location_code):
|
|
"""Search for a location and get all boxes assigned to it
|
|
|
|
Returns:
|
|
tuple: (success: bool, data: dict, status_code: int)
|
|
"""
|
|
try:
|
|
if not location_code or not str(location_code).strip():
|
|
return False, {}, 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Get location info
|
|
cursor.execute("""
|
|
SELECT id, location_code, size, description
|
|
FROM warehouse_locations
|
|
WHERE location_code = %s
|
|
""", (str(location_code).strip(),))
|
|
|
|
location = cursor.fetchone()
|
|
|
|
if not location:
|
|
cursor.close()
|
|
return False, {'error': f'Location "{location_code}" not found'}, 404
|
|
|
|
location_id = location[0]
|
|
|
|
# Get all boxes in this location
|
|
cursor.execute("""
|
|
SELECT
|
|
id,
|
|
box_number,
|
|
status,
|
|
created_at
|
|
FROM boxes_crates
|
|
WHERE location_id = %s
|
|
ORDER BY id DESC
|
|
""", (location_id,))
|
|
|
|
boxes = cursor.fetchall()
|
|
cursor.close()
|
|
|
|
location_data = {
|
|
'id': location[0],
|
|
'location_code': location[1],
|
|
'size': location[2],
|
|
'description': location[3]
|
|
}
|
|
|
|
boxes_list = []
|
|
for box in boxes:
|
|
boxes_list.append({
|
|
'id': box[0],
|
|
'box_number': box[1],
|
|
'status': box[2],
|
|
'created_at': str(box[3])
|
|
})
|
|
|
|
return True, {'location': location_data, 'boxes': boxes_list}, 200
|
|
except Exception as e:
|
|
logger.error(f"Error searching location: {e}")
|
|
return False, {'error': str(e)}, 500
|
|
|
|
|
|
def assign_box_to_location(box_id, location_code):
|
|
"""Assign a box to a warehouse location
|
|
|
|
Returns:
|
|
tuple: (success: bool, message: str, status_code: int)
|
|
"""
|
|
try:
|
|
if not box_id or not location_code:
|
|
return False, 'Box ID and location code are required', 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Check if location exists
|
|
cursor.execute("""
|
|
SELECT id FROM warehouse_locations
|
|
WHERE location_code = %s
|
|
""", (location_code,))
|
|
|
|
location = cursor.fetchone()
|
|
|
|
if not location:
|
|
cursor.close()
|
|
return False, f'Location "{location_code}" not found', 404
|
|
|
|
location_id = location[0]
|
|
|
|
# Get box info
|
|
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
|
|
|
|
# Update box location
|
|
cursor.execute("""
|
|
UPDATE boxes_crates
|
|
SET location_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
""", (location_id, box_id))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, f'Box "{box[0]}" assigned to location "{location_code}"', 200
|
|
except Exception as e:
|
|
logger.error(f"Error assigning box to location: {e}")
|
|
return False, f'Error: {str(e)}', 500
|
|
|
|
|
|
def move_box_to_new_location(box_id, new_location_code):
|
|
"""Move a box from current location to a new location
|
|
|
|
Returns:
|
|
tuple: (success: bool, message: str, status_code: int)
|
|
"""
|
|
try:
|
|
if not box_id or not new_location_code:
|
|
return False, 'Box ID and new location code are required', 400
|
|
|
|
conn = get_db()
|
|
cursor = conn.cursor()
|
|
|
|
# Check if new location exists
|
|
cursor.execute("""
|
|
SELECT id FROM warehouse_locations
|
|
WHERE location_code = %s
|
|
""", (new_location_code,))
|
|
|
|
location = cursor.fetchone()
|
|
|
|
if not location:
|
|
cursor.close()
|
|
return False, f'Location "{new_location_code}" not found', 404
|
|
|
|
new_location_id = location[0]
|
|
|
|
# Get box info
|
|
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
|
|
|
|
# Update box location
|
|
cursor.execute("""
|
|
UPDATE boxes_crates
|
|
SET location_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
""", (new_location_id, box_id))
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
|
|
return True, f'Box "{box[0]}" moved to location "{new_location_code}"', 200
|
|
except Exception as e:
|
|
logger.error(f"Error moving box: {e}")
|
|
return False, f'Error: {str(e)}', 500
|