stuff/ScanLook
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.14.0 - Major Logic Overhaul & Real-Time Dashboard

Browse Source 
Logic: Implemented "One User, One Bin" locking to prevent duplicate counting.

    Integrity: Standardized is_deleted = 0 and tightened "Matched" criteria to require zero weight variance.

    Refresh: Added silent 30-second dashboard polling for all 6 status categories and active counter list.

    Tracking: Built user-specific activity tracking to identify who is counting where in real-time.

    Stability: Resolved persistent 500 errors by finalizing the active-counters-fragment structure.
This commit is contained in:
Javier
2026-01-31 19:17:36 -06:00
parent 288b390618
commit 2d333c16a3
12 changed files with 360 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
blueprints/__pycache__/admin_locations.cpython-313.pyc
View File

Binary file not shown.

BIN
blueprints/__pycache__/sessions.cpython-313.pyc
View File

Binary file not shown.

67
blueprints/admin_locations.py
View File

@@ -29,35 +29,6 @@ def reopen_location(location_count_id):
return jsonify({'success': True, 'message': 'Bin reopened for counting'})
@admin_locations_bp.route('/location/<int:location_count_id>/delete', methods=['POST'])
@login_required
def delete_location_count(location_count_id):
"""Delete all counts for a location (soft delete)"""
# Verify ownership
loc = query_db('SELECT * FROM LocationCounts WHERE location_count_id = ?', [location_count_id], one=True)
if not loc:
return jsonify({'success': False, 'message': 'Location not found'})
if loc['counted_by'] != session['user_id'] and session['role'] not in ['owner', 'admin']:
return jsonify({'success': False, 'message': 'Permission denied'})
# Soft delete all scan entries for this location
execute_db('''
UPDATE ScanEntries
SET is_deleted = 1
WHERE location_count_id = ?
''', [location_count_id])
# Delete the location count record
execute_db('''
DELETE FROM LocationCounts
WHERE location_count_id = ?
''', [location_count_id])
return jsonify({'success': True, 'message': 'Bin count deleted'})
@admin_locations_bp.route('/location/<int:location_count_id>/scans')
@login_required
def get_location_scans(location_count_id):
@@ -86,4 +57,40 @@ def get_location_scans(location_count_id):
return jsonify({'success': True, 'scans': scans_list})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
return jsonify({'success': False, 'message': str(e)})
@admin_locations_bp.route('/location/<int:location_count_id>/delete', methods=['POST'])
@login_required
def soft_delete_location(location_count_id):
"""Admin-only: Soft delete a bin count and its associated data"""
if session.get('role') not in ['owner', 'admin']:
return jsonify({'success': False, 'message': 'Admin role required'}), 403
# 1. Verify location exists
loc = query_db('SELECT session_id, location_name FROM LocationCounts WHERE location_count_id = ?',
[location_count_id], one=True)
if not loc:
return jsonify({'success': False, 'message': 'Location not found'})
# 2. Soft delete the bin count itself
execute_db('''
UPDATE LocationCounts
SET is_deleted = 1
WHERE location_count_id = ?
''', [location_count_id])
# 3. Soft delete all scans in that bin
execute_db('''
UPDATE ScanEntries
SET is_deleted = 1
WHERE location_count_id = ?
''', [location_count_id])
# 4. Remove any MissingLots records generated for this bin
execute_db('''
DELETE FROM MissingLots
WHERE session_id = ? AND master_expected_location = ?
''', [loc['session_id'], loc['location_name']])
return jsonify({'success': True, 'message': 'Bin count and associated data soft-deleted'})

127
blueprints/counting.py
View File

@@ -111,16 +111,23 @@ def my_counts(session_id):
# Get this user's active bins
active_bins = query_db('''
SELECT lc.*,
COUNT(se.entry_id) as scan_count
FROM LocationCounts lc
LEFT JOIN ScanEntries se ON lc.location_count_id = se.location_count_id AND se.is_deleted = 0
WHERE lc.session_id = ?
AND lc.counted_by = ?
AND lc.status = 'in_progress'
GROUP BY lc.location_count_id
ORDER BY lc.start_timestamp DESC
''', [session_id, session['user_id']])
SELECT lc.*,
COUNT(se.entry_id) as scan_count
FROM LocationCounts lc
LEFT JOIN ScanEntries se ON lc.location_count_id = se.location_count_id AND se.is_deleted = 0
WHERE lc.session_id = ?
AND lc.status = 'in_progress'
AND lc.is_deleted = 0
AND (
lc.counted_by = ?
OR lc.location_count_id IN (
SELECT location_count_id FROM ScanEntries
WHERE scanned_by = ? AND is_deleted = 0
)
)
GROUP BY lc.location_count_id
ORDER BY lc.start_timestamp DESC
''', [session_id, session['user_id'], session['user_id']])
# Get this user's completed bins
completed_bins = query_db('''
@@ -129,11 +136,17 @@ def my_counts(session_id):
FROM LocationCounts lc
LEFT JOIN ScanEntries se ON lc.location_count_id = se.location_count_id AND se.is_deleted = 0
WHERE lc.session_id = ?
AND lc.counted_by = ?
AND lc.status = 'completed'
AND (
lc.counted_by = ?
OR lc.location_count_id IN (
SELECT location_count_id FROM ScanEntries
WHERE scanned_by = ? AND is_deleted = 0
)
)
GROUP BY lc.location_count_id
ORDER BY lc.end_timestamp DESC
''', [session_id, session['user_id']])
ORDER BY lc.start_timestamp DESC
''', [session_id, session['user_id'], session['user_id']])
return render_template('counts/my_counts.html',
count_session=sess,
@@ -144,7 +157,7 @@ def my_counts(session_id):
@counting_bp.route('/session/<int:session_id>/start-bin', methods=['POST'])
@login_required
def start_bin_count(session_id):
"""Start counting a new bin"""
"""Start counting a new bin or resume an existing in-progress one"""
sess = get_active_session(session_id)
if not sess:
flash('Session not found or archived', 'warning')
@@ -158,7 +171,21 @@ def start_bin_count(session_id):
if not location_name:
flash('Bin number is required', 'danger')
return redirect(url_for('counting.my_counts', session_id=session_id))
# --- NEW LOGIC: Check for existing in-progress bin ---
existing_bin = query_db('''
SELECT location_count_id
FROM LocationCounts
WHERE session_id = ? AND location_name = ? AND status = 'in_progress'
''', [session_id, location_name], one=True)
if existing_bin:
flash(f'Resuming bin: {location_name}', 'info')
return redirect(url_for('counting.count_location',
session_id=session_id,
location_count_id=existing_bin['location_count_id']))
# --- END NEW LOGIC ---
# Count expected lots from MASTER baseline for this location
expected_lots = query_db('''
SELECT COUNT(DISTINCT lot_number) as count
@@ -168,7 +195,7 @@ def start_bin_count(session_id):
expected_count = expected_lots['count'] if expected_lots else 0
# Create new location count
# Create new location count if none existed
conn = get_db()
cursor = conn.cursor()
@@ -184,7 +211,6 @@ def start_bin_count(session_id):
flash(f'Started counting bin: {location_name}', 'success')
return redirect(url_for('counting.count_location', session_id=session_id, location_count_id=location_count_id))
@counting_bp.route('/location/<int:location_count_id>/complete', methods=['POST'])
@login_required
def complete_location(location_count_id):
@@ -512,7 +538,7 @@ def scan_lot(session_id, location_count_id):
def delete_scan(entry_id):
"""Soft delete a scan and recalculate duplicate statuses"""
# Get the scan being deleted
scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ?', [entry_id], one=True)
scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ? AND is_deleted = 0', [entry_id], one=True)
if not scan:
return jsonify({'success': False, 'message': 'Scan not found'})
@@ -572,7 +598,7 @@ def update_scan(entry_id):
comment = data.get('comment', '')
# Get the scan
scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ?', [entry_id], one=True)
scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ? AND is_deleted = 0', [entry_id], one=True)
if not scan:
return jsonify({'success': False, 'message': 'Scan not found'})
@@ -593,7 +619,7 @@ def update_scan(entry_id):
actual_weight = ?,
comment = ?,
modified_timestamp = CURRENT_TIMESTAMP
WHERE entry_id = ?
WHERE entry_id = ? and is_deleted = 0
''', [item, weight, comment, entry_id])
return jsonify({'success': True, 'message': 'Scan updated'})
@@ -625,7 +651,7 @@ def recalculate_duplicate_status(session_id, lot_number, current_location):
duplicate_info = NULL,
comment = NULL,
modified_timestamp = CURRENT_TIMESTAMP
WHERE entry_id = ?
WHERE entry_id = ? and is_deleted = 0
''', [scan['entry_id']])
updated_entries.append({
'entry_id': scan['entry_id'],
@@ -670,7 +696,7 @@ def recalculate_duplicate_status(session_id, lot_number, current_location):
duplicate_info = ?,
comment = ?,
modified_timestamp = CURRENT_TIMESTAMP
WHERE entry_id = ?
WHERE entry_id = ? and is_deleted = 0
''', [duplicate_status, duplicate_info, duplicate_info, scan['entry_id']])
# Update our tracking list
@@ -689,7 +715,7 @@ def recalculate_duplicate_status(session_id, lot_number, current_location):
duplicate_info = ?,
comment = ?,
modified_timestamp = CURRENT_TIMESTAMP
WHERE entry_id = ?
WHERE entry_id = ? and is_deleted = 0
''', [duplicate_status, duplicate_info, duplicate_info, prev_scan['entry_id']])
# Update tracking for previous scans
@@ -754,4 +780,57 @@ def finish_location(session_id, location_count_id):
return jsonify({
'success': True,
'redirect': url_for('counting.count_session', session_id=session_id)
})
})
@counting_bp.route('/session/<int:session_id>/finalize-all', methods=['POST'])
@login_required
def finalize_all_locations(session_id):
"""Finalize all 'in_progress' locations in a session"""
if session.get('role') not in ['owner', 'admin']:
return jsonify({'success': False, 'message': 'Permission denied'}), 403
# 1. Get all in_progress locations for this session
locations = query_db('''
SELECT location_count_id, location_name
FROM LocationCounts
WHERE session_id = ?
AND status = 'in_progress'
AND is_deleted = 0
''', [session_id])
if not locations:
return jsonify({'success': True, 'message': 'No open bins to finalize.'})
# 2. Loop through and run the finalize logic for each
for loc in locations:
# We reuse the logic from your existing finish_location route
execute_db('''
UPDATE LocationCounts
SET status = 'completed', end_timestamp = CURRENT_TIMESTAMP
WHERE location_count_id = ?
''', [loc['location_count_id']])
# Identify missing lots from MASTER baseline
expected_lots = query_db('''
SELECT lot_number, item, description, system_quantity
FROM BaselineInventory_Master
WHERE session_id = ? AND system_bin = ?
''', [session_id, loc['location_name']])
scanned_lots = query_db('''
SELECT DISTINCT lot_number
FROM ScanEntries
WHERE location_count_id = ? AND is_deleted = 0
''', [loc['location_count_id']])
scanned_lot_numbers = {s['lot_number'] for s in scanned_lots}
for expected in expected_lots:
if expected['lot_number'] not in scanned_lot_numbers:
execute_db('''
INSERT INTO MissingLots (session_id, lot_number, master_expected_location, item, master_expected_quantity, marked_by)
VALUES (?, ?, ?, ?, ?, ?)
''', [session_id, expected['lot_number'], loc['location_name'],
expected['item'], expected['system_quantity'], session['user_id']])
return jsonify({'success': True, 'message': f'Successfully finalized {len(locations)} bins.'})

92
blueprints/sessions.py
View File

@@ -24,7 +24,7 @@ def create_session():
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')
return render_template('/counts/create_session.html')
@sessions_bp.route('/session/<int:session_id>')
@@ -54,24 +54,38 @@ def session_detail(session_id):
''', [session_id], one=True)
# Get location progress
# We add a subquery to count the actual missing lots for each bin
locations = query_db('''
SELECT lc.*, u.full_name as counter_name
SELECT
lc.*,
u.full_name as counter_name,
(SELECT COUNT(*) FROM MissingLots ml
WHERE ml.session_id = lc.session_id
AND ml.master_expected_location = lc.location_name) as lots_missing_calc
FROM LocationCounts lc
LEFT JOIN Users u ON lc.counted_by = u.user_id
WHERE lc.session_id = ?
AND lc.is_deleted = 0
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
SELECT
u.full_name,
u.user_id,
MAX(lc.start_timestamp) AS start_timestamp, -- Add the alias here!
lc.location_name
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
WHERE lc.session_id = ?
AND lc.status = 'in_progress'
AND lc.is_deleted = 0
GROUP BY u.user_id
ORDER BY start_timestamp DESC
''', [session_id])
return render_template('session_detail.html',
return render_template('/counts/session_detail.html',
count_session=sess,
stats=stats,
locations=locations,
@@ -98,6 +112,7 @@ def get_status_details(session_id, status):
WHERE se.session_id = ?
AND se.master_status = 'match'
AND se.duplicate_status = '00'
AND se.master_variance_lbs = 0
AND se.is_deleted = 0
ORDER BY se.scan_timestamp DESC
''', [session_id])
@@ -184,20 +199,21 @@ def get_status_details(session_id, status):
# Missing lots (in master but not scanned)
items = query_db('''
SELECT
bim.lot_number,
bim.item,
ml.lot_number,
ml.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])
ml.master_expected_location as system_bin,
ml.master_expected_quantity as system_quantity
FROM MissingLots ml
LEFT JOIN BaselineInventory_Master bim ON
ml.lot_number = bim.lot_number AND
ml.item = bim.item AND
ml.master_expected_location = bim.system_bin AND
ml.session_id = bim.session_id
WHERE ml.session_id = ?
GROUP BY ml.lot_number, ml.item, ml.master_expected_location
ORDER BY ml.master_expected_location, ml.lot_number
''', [session_id])
else:
return jsonify({'success': False, 'message': 'Invalid status'})
@@ -241,4 +257,40 @@ def activate_session(session_id):
execute_db('UPDATE CountSessions SET status = ? WHERE session_id = ?', ['active', session_id])
return jsonify({'success': True, 'message': 'Session activated successfully'})
return jsonify({'success': True, 'message': 'Session activated successfully'})
@sessions_bp.route('/session/<int:session_id>/get_stats')
@role_required('owner', 'admin')
def get_session_stats(session_id):
stats = query_db('''
SELECT
COUNT(DISTINCT se.entry_id) FILTER (WHERE se.master_status = 'match' AND se.duplicate_status = '00' AND se.master_variance_lbs = 0 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 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
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)
return jsonify(success=True, stats=dict(stats))
@sessions_bp.route('/session/<int:session_id>/active-counters-fragment')
@role_required('owner', 'admin')
def active_counters_fragment(session_id):
# Use that unique-user query we just built together
active_counters = query_db('''
SELECT
u.full_name,
MAX(lc.start_timestamp) AS start_timestamp,
lc.location_name
FROM LocationCounts lc
JOIN Users u ON lc.counted_by = u.user_id
WHERE lc.session_id = ? AND lc.status = 'in_progress' AND lc.is_deleted = 0
GROUP BY u.user_id
''', [session_id])
# This renders JUST the list part, not the whole page
return render_template('counts/partials/_active_counters.html', active_counters=active_counters)