244 lines
11 KiB
Python
244 lines
11 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, session
|
|
from db import query_db, execute_db
|
|
from utils import role_required
|
|
|
|
sessions_bp = Blueprint('sessions', __name__)
|
|
|
|
@sessions_bp.route('/session/create', methods=['GET', 'POST'])
|
|
@role_required('owner', 'admin')
|
|
def create_session():
|
|
"""Create new count session"""
|
|
if request.method == 'POST':
|
|
session_name = request.form.get('session_name', '').strip()
|
|
session_type = request.form.get('session_type')
|
|
|
|
if not session_name:
|
|
flash('Session name is required', 'danger')
|
|
return redirect(url_for('sessions.create_session'))
|
|
|
|
session_id = execute_db('''
|
|
INSERT INTO CountSessions (session_name, session_type, created_by, branch)
|
|
VALUES (?, ?, ?, ?)
|
|
''', [session_name, session_type, session['user_id'], 'Main'])
|
|
|
|
flash(f'Session "{session_name}" created successfully!', 'success')
|
|
return redirect(url_for('sessions.session_detail', session_id=session_id))
|
|
|
|
return render_template('create_session.html')
|
|
|
|
|
|
@sessions_bp.route('/session/<int:session_id>')
|
|
@role_required('owner', 'admin')
|
|
def session_detail(session_id):
|
|
"""Session detail and monitoring page"""
|
|
sess = query_db('SELECT * FROM CountSessions WHERE session_id = ?', [session_id], one=True)
|
|
|
|
if not sess:
|
|
flash('Session not found', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
|
|
# Get statistics
|
|
stats = query_db('''
|
|
SELECT
|
|
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.is_deleted = 0) as total_scans,
|
|
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.master_status = 'match' AND se.duplicate_status = '00' AND se.is_deleted = 0 AND ABS(se.actual_weight - se.master_expected_weight) < 0.01) as matched,
|
|
COUNT(DISTINCT se.lot_number) FILTER (WHERE se.duplicate_status IN ('01', '03', '04') AND se.is_deleted = 0) as duplicates,
|
|
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.master_status = 'match' AND se.duplicate_status = '00' AND se.is_deleted = 0 AND ABS(se.actual_weight - se.master_expected_weight) >= 0.01) as weight_discrepancy,
|
|
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.master_status = 'wrong_location' AND se.is_deleted = 0) as wrong_location,
|
|
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.master_status = 'ghost_lot' AND se.is_deleted = 0) as ghost_lots,
|
|
COUNT(DISTINCT ml.missing_id) as missing_lots
|
|
FROM CountSessions cs
|
|
LEFT JOIN ScanEntries se ON cs.session_id = se.session_id
|
|
LEFT JOIN MissingLots ml ON cs.session_id = ml.session_id
|
|
WHERE cs.session_id = ?
|
|
''', [session_id], one=True)
|
|
|
|
# Get location progress
|
|
locations = query_db('''
|
|
SELECT lc.*, u.full_name as counter_name
|
|
FROM LocationCounts lc
|
|
LEFT JOIN Users u ON lc.counted_by = u.user_id
|
|
WHERE lc.session_id = ?
|
|
ORDER BY lc.status DESC, lc.location_name
|
|
''', [session_id])
|
|
|
|
# Get active counters
|
|
active_counters = query_db('''
|
|
SELECT DISTINCT u.full_name, lc.location_name, lc.start_timestamp
|
|
FROM LocationCounts lc
|
|
JOIN Users u ON lc.counted_by = u.user_id
|
|
WHERE lc.session_id = ? AND lc.status = 'in_progress'
|
|
ORDER BY lc.start_timestamp DESC
|
|
''', [session_id])
|
|
|
|
return render_template('session_detail.html',
|
|
count_session=sess,
|
|
stats=stats,
|
|
locations=locations,
|
|
active_counters=active_counters)
|
|
|
|
|
|
@sessions_bp.route('/session/<int:session_id>/status-details/<status>')
|
|
@role_required('owner', 'admin')
|
|
def get_status_details(session_id, status):
|
|
"""Get detailed breakdown for a specific status"""
|
|
|
|
try:
|
|
if status == 'match':
|
|
# Matched lots (not duplicates) - JOIN with CURRENT for live data
|
|
items = query_db('''
|
|
SELECT
|
|
se.*,
|
|
u.full_name as scanned_by_name,
|
|
bic.system_bin as current_system_location,
|
|
bic.system_quantity as current_system_weight
|
|
FROM ScanEntries se
|
|
JOIN Users u ON se.scanned_by = u.user_id
|
|
LEFT JOIN BaselineInventory_Current bic ON se.lot_number = bic.lot_number
|
|
WHERE se.session_id = ?
|
|
AND se.master_status = 'match'
|
|
AND se.duplicate_status = '00'
|
|
AND se.is_deleted = 0
|
|
ORDER BY se.scan_timestamp DESC
|
|
''', [session_id])
|
|
|
|
elif status == 'duplicates':
|
|
# Duplicate lots (grouped by lot number) - JOIN with CURRENT
|
|
items = query_db('''
|
|
SELECT
|
|
se.lot_number,
|
|
se.item,
|
|
se.description,
|
|
GROUP_CONCAT(DISTINCT se.scanned_location) as scanned_location,
|
|
SUM(se.actual_weight) as actual_weight,
|
|
se.master_expected_location,
|
|
se.master_expected_weight,
|
|
GROUP_CONCAT(DISTINCT u.full_name) as scanned_by_name,
|
|
MIN(se.scan_timestamp) as scan_timestamp,
|
|
bic.system_bin as current_system_location,
|
|
bic.system_quantity as current_system_weight
|
|
FROM ScanEntries se
|
|
JOIN Users u ON se.scanned_by = u.user_id
|
|
LEFT JOIN BaselineInventory_Current bic ON se.lot_number = bic.lot_number
|
|
WHERE se.session_id = ?
|
|
AND se.duplicate_status IN ('01', '03', '04')
|
|
AND se.is_deleted = 0
|
|
GROUP BY se.lot_number
|
|
ORDER BY se.lot_number
|
|
''', [session_id])
|
|
|
|
elif status == 'wrong_location':
|
|
# Wrong location lots - JOIN with CURRENT
|
|
items = query_db('''
|
|
SELECT
|
|
se.*,
|
|
u.full_name as scanned_by_name,
|
|
bic.system_bin as current_system_location,
|
|
bic.system_quantity as current_system_weight
|
|
FROM ScanEntries se
|
|
JOIN Users u ON se.scanned_by = u.user_id
|
|
LEFT JOIN BaselineInventory_Current bic ON se.lot_number = bic.lot_number
|
|
WHERE se.session_id = ?
|
|
AND se.master_status = 'wrong_location'
|
|
AND se.is_deleted = 0
|
|
ORDER BY se.scan_timestamp DESC
|
|
''', [session_id])
|
|
|
|
elif status == 'weight_discrepancy':
|
|
# Weight discrepancies (right location, wrong weight) - JOIN with CURRENT
|
|
items = query_db('''
|
|
SELECT
|
|
se.*,
|
|
u.full_name as scanned_by_name,
|
|
bic.system_bin as current_system_location,
|
|
bic.system_quantity as current_system_weight
|
|
FROM ScanEntries se
|
|
JOIN Users u ON se.scanned_by = u.user_id
|
|
LEFT JOIN BaselineInventory_Current bic ON se.lot_number = bic.lot_number
|
|
WHERE se.session_id = ?
|
|
AND se.master_status = 'match'
|
|
AND se.duplicate_status = '00'
|
|
AND ABS(se.actual_weight - se.master_expected_weight) >= 0.01
|
|
AND se.is_deleted = 0
|
|
ORDER BY ABS(se.actual_weight - se.master_expected_weight) DESC
|
|
''', [session_id])
|
|
|
|
elif status == 'ghost_lot':
|
|
# Ghost lots (not in master baseline) - JOIN with CURRENT
|
|
items = query_db('''
|
|
SELECT
|
|
se.*,
|
|
u.full_name as scanned_by_name,
|
|
bic.system_bin as current_system_location,
|
|
bic.system_quantity as current_system_weight
|
|
FROM ScanEntries se
|
|
JOIN Users u ON se.scanned_by = u.user_id
|
|
LEFT JOIN BaselineInventory_Current bic ON se.lot_number = bic.lot_number
|
|
WHERE se.session_id = ?
|
|
AND se.master_status = 'ghost_lot'
|
|
AND se.is_deleted = 0
|
|
ORDER BY se.scan_timestamp DESC
|
|
''', [session_id])
|
|
|
|
elif status == 'missing':
|
|
# Missing lots (in master but not scanned)
|
|
items = query_db('''
|
|
SELECT
|
|
bim.lot_number,
|
|
bim.item,
|
|
bim.description,
|
|
bim.system_bin,
|
|
bim.system_quantity
|
|
FROM BaselineInventory_Master bim
|
|
WHERE bim.session_id = ?
|
|
AND bim.lot_number NOT IN (
|
|
SELECT lot_number
|
|
FROM ScanEntries
|
|
WHERE session_id = ? AND is_deleted = 0
|
|
)
|
|
ORDER BY bim.system_bin, bim.lot_number
|
|
''', [session_id, session_id])
|
|
else:
|
|
return jsonify({'success': False, 'message': 'Invalid status'})
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'items': [dict(item) for item in items] if items else []
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"Error in get_status_details: {str(e)}")
|
|
return jsonify({'success': False, 'message': f'Error: {str(e)}'})
|
|
|
|
@sessions_bp.route('/session/<int:session_id>/archive', methods=['POST'])
|
|
@role_required('owner', 'admin')
|
|
def archive_session(session_id):
|
|
"""Archive a count session"""
|
|
sess = query_db('SELECT * FROM CountSessions WHERE session_id = ?', [session_id], one=True)
|
|
|
|
if not sess:
|
|
return jsonify({'success': False, 'message': 'Session not found'})
|
|
|
|
if sess['status'] == 'archived':
|
|
return jsonify({'success': False, 'message': 'Session is already archived'})
|
|
|
|
execute_db('UPDATE CountSessions SET status = ? WHERE session_id = ?', ['archived', session_id])
|
|
|
|
return jsonify({'success': True, 'message': 'Session archived successfully'})
|
|
|
|
|
|
@sessions_bp.route('/session/<int:session_id>/activate', methods=['POST'])
|
|
@role_required('owner', 'admin')
|
|
def activate_session(session_id):
|
|
"""Reactivate an archived session"""
|
|
sess = query_db('SELECT * FROM CountSessions WHERE session_id = ?', [session_id], one=True)
|
|
|
|
if not sess:
|
|
return jsonify({'success': False, 'message': 'Session not found'})
|
|
|
|
if sess['status'] != 'archived':
|
|
return jsonify({'success': False, 'message': 'Session is not archived'})
|
|
|
|
execute_db('UPDATE CountSessions SET status = ? WHERE session_id = ?', ['active', session_id])
|
|
|
|
return jsonify({'success': True, 'message': 'Session activated successfully'}) |