ske087/quality_app-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement approved/rejected quantity triggers and warehouse inventory

Browse Source 
Database Triggers Implementation:
- Added automatic quantity calculation triggers for scanfg_orders
- Added automatic quantity calculation triggers for scan1_orders (T1 phase)
- Triggers calculate based on CP_base_code grouping (8 digits)
- Quality code: 0 = approved, != 0 = rejected
- Quantities set at insertion time (BEFORE INSERT trigger)
- Added create_triggers() function to initialize_db.py

Warehouse Inventory Enhancement:
- Analyzed old app database quantity calculation logic
- Created comprehensive trigger implementation guide
- Added trigger verification and testing procedures
- Documented data migration strategy

Documentation Added:
- APPROVED_REJECTED_QUANTITIES_ANALYSIS.md - Old app logic analysis
- DATABASE_TRIGGERS_IMPLEMENTATION.md - v2 implementation guide
- WAREHOUSE_INVENTORY_IMPLEMENTATION.md - Inventory view feature

Files Modified:
- initialize_db.py: Added create_triggers() function and call in main()
- Documentation: 3 comprehensive guides for database and inventory management

Quality Metrics:
- Triggers maintain legacy compatibility
- Automatic calculation ensures data consistency
- Performance optimized at database level
- Comprehensive testing documented
This commit is contained in:
Quality App Developer
2026-01-30 12:30:56 +02:00
parent b15cc93b9d
commit 07f77603eb
7 changed files with 2246 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
app/modules/warehouse/routes.py
View File

@@ -6,7 +6,8 @@ from app.modules.warehouse.warehouse import (
get_all_locations, add_location, update_location, delete_location,
delete_multiple_locations, get_location_by_id,
search_box_by_number, search_location_with_boxes,
assign_box_to_location, move_box_to_new_location
assign_box_to_location, move_box_to_new_location,
get_cp_inventory_list, search_cp_code, search_by_box_number, get_cp_details
)
import logging
@@ -215,3 +216,120 @@ def api_get_locations():
locations = get_all_locations()
return jsonify({'success': True, 'locations': locations}), 200
# ============================================================================
# API Routes for CP Inventory View
# ============================================================================
@warehouse_bp.route('/api/cp-inventory', methods=['GET'], endpoint='api_cp_inventory')
def api_cp_inventory():
"""Get CP inventory list - all CP articles with box and location info"""
if 'user_id' not in session:
return jsonify({'success': False, 'error': 'Unauthorized'}), 401
try:
limit = request.args.get('limit', 100, type=int)
offset = request.args.get('offset', 0, type=int)
# Validate pagination parameters
if limit > 1000:
limit = 1000
if limit < 1:
limit = 50
if offset < 0:
offset = 0
inventory = get_cp_inventory_list(limit=limit, offset=offset)
return jsonify({
'success': True,
'inventory': inventory,
'count': len(inventory),
'limit': limit,
'offset': offset
}), 200
except Exception as e:
logger.error(f"Error getting CP inventory: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@warehouse_bp.route('/api/search-cp', methods=['POST'], endpoint='api_search_cp')
def api_search_cp():
"""Search for CP code in warehouse inventory"""
if 'user_id' not in session:
return jsonify({'success': False, 'error': 'Unauthorized'}), 401
try:
data = request.get_json()
cp_code = data.get('cp_code', '').strip()
if not cp_code or len(cp_code) < 2:
return jsonify({'success': False, 'error': 'CP code must be at least 2 characters'}), 400
results = search_cp_code(cp_code)
return jsonify({
'success': True,
'results': results,
'count': len(results),
'search_term': cp_code
}), 200
except Exception as e:
logger.error(f"Error searching CP code: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@warehouse_bp.route('/api/search-cp-box', methods=['POST'], endpoint='api_search_cp_box')
def api_search_cp_box():
"""Search for box number and get all CP codes in it"""
if 'user_id' not in session:
return jsonify({'success': False, 'error': 'Unauthorized'}), 401
try:
data = request.get_json()
box_number = data.get('box_number', '').strip()
if not box_number or len(box_number) < 1:
return jsonify({'success': False, 'error': 'Box number is required'}), 400
results = search_by_box_number(box_number)
return jsonify({
'success': True,
'results': results,
'count': len(results),
'search_term': box_number
}), 200
except Exception as e:
logger.error(f"Error searching by box number: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@warehouse_bp.route('/api/cp-details/<cp_code>', methods=['GET'], endpoint='api_cp_details')
def api_cp_details(cp_code):
"""Get detailed information for a CP code and all its variations"""
if 'user_id' not in session:
return jsonify({'success': False, 'error': 'Unauthorized'}), 401
try:
cp_code = cp_code.strip().upper()
# Ensure CP code is properly formatted (at least "CP00000001")
if not cp_code.startswith('CP') or len(cp_code) < 10:
return jsonify({'success': False, 'error': 'Invalid CP code format'}), 400
# Extract base CP code (first 10 characters)
cp_base = cp_code[:10]
details = get_cp_details(cp_base)
return jsonify({
'success': True,
'cp_code': cp_base,
'details': details,
'count': len(details)
}), 200
except Exception as e:
logger.error(f"Error getting CP details: {e}")
return jsonify({'success': False, 'error': str(e)}), 500

194
app/modules/warehouse/warehouse.py
View File

@@ -3,6 +3,7 @@ Warehouse Module - Helper Functions
Provides functions for warehouse operations
"""
import logging
import pymysql
from app.database import get_db
logger = logging.getLogger(__name__)
@@ -435,3 +436,196 @@ def move_box_to_new_location(box_id, new_location_code):
except Exception as e:
logger.error(f"Error moving box: {e}")
return False, f'Error: {str(e)}', 500
def get_cp_inventory_list(limit=100, offset=0):
"""
Get CP articles from scanfg_orders with box and location info
Groups by CP_full_code (8 digits) to show all entries with that base CP
Returns latest entries first
Returns:
List of CP inventory records with box and location info
"""
try:
conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Get all CP codes with their box and location info (latest entries first)
cursor.execute("""
SELECT
s.id,
s.CP_full_code,
SUBSTRING(s.CP_full_code, 1, 10) as cp_base,
COUNT(*) as total_entries,
s.box_id,
bc.box_number,
wl.location_code,
wl.id as location_id,
MAX(s.date) as latest_date,
MAX(s.time) as latest_time,
SUM(s.approved_quantity) as total_approved,
SUM(s.rejected_quantity) as total_rejected
FROM scanfg_orders s
LEFT JOIN boxes_crates bc ON s.box_id = bc.id
LEFT JOIN warehouse_locations wl ON s.location_id = wl.id
GROUP BY SUBSTRING(s.CP_full_code, 1, 10), s.box_id
ORDER BY MAX(s.created_at) DESC
LIMIT %s OFFSET %s
""", (limit, offset))
results = cursor.fetchall()
cursor.close()
return results if results else []
except Exception as e:
logger.error(f"Error getting CP inventory: {e}")
return []
def search_cp_code(cp_code_search):
"""
Search for CP codes - can search by full CP code or CP base (8 digits)
Args:
cp_code_search: Search string (can be "CP00000001" or "CP00000001-0001")
Returns:
List of matching CP inventory records
"""
try:
conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Remove hyphen and get base CP code if full CP provided
search_term = cp_code_search.replace('-', '').strip().upper()
# Search for matching CP codes
cursor.execute("""
SELECT
s.id,
s.CP_full_code,
SUBSTRING(s.CP_full_code, 1, 10) as cp_base,
COUNT(*) as total_entries,
s.box_id,
bc.box_number,
wl.location_code,
wl.id as location_id,
MAX(s.date) as latest_date,
MAX(s.time) as latest_time,
SUM(s.approved_quantity) as total_approved,
SUM(s.rejected_quantity) as total_rejected
FROM scanfg_orders s
LEFT JOIN boxes_crates bc ON s.box_id = bc.id
LEFT JOIN warehouse_locations wl ON s.location_id = wl.id
WHERE REPLACE(s.CP_full_code, '-', '') LIKE %s
GROUP BY SUBSTRING(s.CP_full_code, 1, 10), s.box_id
ORDER BY MAX(s.created_at) DESC
""", (f"{search_term}%",))
results = cursor.fetchall()
cursor.close()
return results if results else []
except Exception as e:
logger.error(f"Error searching CP code: {e}")
return []
def search_by_box_number(box_number_search):
"""
Search for box number and get all CP codes in that box
Args:
box_number_search: Box number to search for
Returns:
List of CP entries in the box
"""
try:
conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor)
box_search = box_number_search.strip().upper()
cursor.execute("""
SELECT
s.id,
s.CP_full_code,
s.operator_code,
s.quality_code,
s.date,
s.time,
s.approved_quantity,
s.rejected_quantity,
bc.box_number,
bc.id as box_id,
wl.location_code,
wl.id as location_id,
s.created_at
FROM scanfg_orders s
LEFT JOIN boxes_crates bc ON s.box_id = bc.id
LEFT JOIN warehouse_locations wl ON s.location_id = wl.id
WHERE bc.box_number LIKE %s
ORDER BY s.created_at DESC
LIMIT 500
""", (f"%{box_search}%",))
results = cursor.fetchall()
cursor.close()
return results if results else []
except Exception as e:
logger.error(f"Error searching by box number: {e}")
return []
def get_cp_details(cp_code):
"""
Get detailed information for a specific CP code (8 digits)
Shows all variations (with different 4-digit suffixes) and their locations
Args:
cp_code: CP base code (8 digits, e.g., "CP00000001")
Returns:
List of all CP variations with their box and location info
"""
try:
conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Search for all entries with this CP base
cursor.execute("""
SELECT
s.id,
s.CP_full_code,
s.operator_code,
s.OC1_code,
s.OC2_code,
s.quality_code,
s.date,
s.time,
s.approved_quantity,
s.rejected_quantity,
bc.box_number,
bc.id as box_id,
wl.location_code,
wl.id as location_id,
wl.description,
s.created_at
FROM scanfg_orders s
LEFT JOIN boxes_crates bc ON s.box_id = bc.id
LEFT JOIN warehouse_locations wl ON s.location_id = wl.id
WHERE SUBSTRING(s.CP_full_code, 1, 10) = %s
ORDER BY s.created_at DESC
LIMIT 1000
""", (cp_code.upper(),))
results = cursor.fetchall()
cursor.close()
return results if results else []
except Exception as e:
logger.error(f"Error getting CP details: {e}")
return []