diff --git a/app.py b/app.py index be47068..0e51199 100644 --- a/app.py +++ b/app.py @@ -1,5 +1,5 @@ """ -ScanLook V1.0 - Inventory Management System +ScanLook - Inventory Management System Flask Application Production-Ready Release """ @@ -17,6 +17,7 @@ from blueprints.data_imports import data_imports_bp from blueprints.users import users_bp from blueprints.sessions import sessions_bp from blueprints.admin_locations import admin_locations_bp +from blueprints.counting import counting_bp # Add this import from utils import login_required # Register Blueprints @@ -24,6 +25,7 @@ app.register_blueprint(data_imports_bp) app.register_blueprint(users_bp) app.register_blueprint(sessions_bp) app.register_blueprint(admin_locations_bp) +app.register_blueprint(counting_bp) # Add this registration # V1.0: Use environment variable for production, fallback to demo key for development app.secret_key = os.environ.get('SCANLOOK_SECRET_KEY', 'scanlook-demo-key-replace-for-production') @@ -33,6 +35,14 @@ app.config['DATABASE'] = os.path.join(os.path.dirname(__file__), 'database', 'sc app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=1) +# 1. Define the version +APP_VERSION = '0.8.6' + +# 2. Inject it into all templates automatically +@app.context_processor +def inject_version(): + return dict(version=APP_VERSION) + # ==================== ROUTES: AUTHENTICATION ==================== @app.route('/') @@ -126,635 +136,6 @@ def staff_mode(): return render_template('staff_dashboard.html', sessions=active_sessions, is_admin_mode=True) -# ==================== ROUTES: COUNTING (STAFF) ==================== - -@app.route('/count/') -@login_required -def count_session(session_id): - """Select session and begin counting""" - sess = query_db('SELECT * FROM CountSessions WHERE session_id = ? AND status = "active"', - [session_id], one=True) - - if not sess: - flash('Session not found or not active', 'danger') - return redirect(url_for('dashboard')) - - # Redirect to my_counts page (staff can manage multiple bins) - return redirect(url_for('my_counts', session_id=session_id)) - - -@app.route('/session//my-counts') -@login_required -def my_counts(session_id): - """Staff view of their active and completed bins""" - 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 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']]) - - # Get this user's completed bins - completed_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 = 'completed' - GROUP BY lc.location_count_id - ORDER BY lc.end_timestamp DESC - ''', [session_id, session['user_id']]) - - return render_template('my_counts.html', - count_session=sess, - active_bins=active_bins, - completed_bins=completed_bins) - - -@app.route('/session//start-bin', methods=['POST']) -@login_required -def start_bin_count(session_id): - """Start counting a new bin""" - location_name = request.form.get('location_name', '').strip().upper() - - if not location_name: - flash('Bin number is required', 'danger') - return redirect(url_for('my_counts', session_id=session_id)) - - # Count expected lots from MASTER baseline for this location - expected_lots = query_db(''' - SELECT COUNT(DISTINCT lot_number) as count - FROM BaselineInventory_Master - WHERE session_id = ? AND system_bin = ? - ''', [session_id, location_name], one=True) - - expected_count = expected_lots['count'] if expected_lots else 0 - - # Create new location count - conn = get_db() - cursor = conn.cursor() - - cursor.execute(''' - INSERT INTO LocationCounts (session_id, location_name, counted_by, status, start_timestamp, expected_lots_master) - VALUES (?, ?, ?, 'in_progress', CURRENT_TIMESTAMP, ?) - ''', [session_id, location_name, session['user_id'], expected_count]) - - location_count_id = cursor.lastrowid - conn.commit() - conn.close() - - flash(f'Started counting bin: {location_name}', 'success') - return redirect(url_for('count_location', session_id=session_id, location_count_id=location_count_id)) - - -@app.route('/location//complete', methods=['POST']) -@login_required -def complete_location(location_count_id): - """Mark a location count as complete""" - # 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'}) - - # Mark as completed - execute_db(''' - UPDATE LocationCounts - SET status = 'completed', end_timestamp = CURRENT_TIMESTAMP - WHERE location_count_id = ? - ''', [location_count_id]) - - return jsonify({'success': True, 'message': 'Bin marked as complete'}) - - -@app.route('/count//location/') -@login_required -def count_location(session_id, location_count_id): - """Count lots in a specific location""" - # Get session info to determine type (Cycle Count vs Physical) - sess = query_db('SELECT * FROM CountSessions WHERE session_id = ?', [session_id], one=True) - - location = query_db(''' - SELECT * FROM LocationCounts - WHERE location_count_id = ? AND session_id = ? - ''', [location_count_id, session_id], one=True) - - if not location: - flash('Location not found', 'danger') - return redirect(url_for('count_session', session_id=session_id)) - - # Check if location is completed and user is staff (not admin/owner) - if location['status'] == 'completed' and session['role'] == 'staff': - flash(f'Location {location["location_name"]} has been finalized and cannot accept new scans', 'warning') - return redirect(url_for('my_counts', session_id=session_id)) - - # Get scans for this location (Scanned Lots) - scans = query_db(''' - SELECT * FROM ScanEntries - WHERE location_count_id = ? AND is_deleted = 0 - ORDER BY scan_timestamp DESC - ''', [location_count_id]) - - # NEW LOGIC: Get Expected Lots for Cycle Counts (Grouped & Summed) - expected_lots = [] - if sess and sess['session_type'] == 'cycle_count': - expected_lots = query_db(''' - SELECT - lot_number, - MAX(item) as item, -- Pick one item code if they differ (rare) - SUM(system_quantity) as total_weight - FROM BaselineInventory_Master - WHERE session_id = ? - AND system_bin = ? - AND lot_number NOT IN ( - SELECT lot_number - FROM ScanEntries - WHERE location_count_id = ? - AND is_deleted = 0 - ) - GROUP BY lot_number - ORDER BY lot_number - ''', [session_id, location['location_name'], location_count_id]) - - return render_template('count_location.html', - session_id=session_id, - location=location, - scans=scans, - expected_lots=expected_lots, - session_type=sess['session_type'] if sess else '') - -@app.route('/count//location//scan', methods=['POST']) -@login_required -def scan_lot(session_id, location_count_id): - """Process a lot scan with duplicate detection""" - data = request.get_json() - lot_number = data.get('lot_number', '').strip() - weight = data.get('weight') - confirm_duplicate = data.get('confirm_duplicate', False) - check_only = data.get('check_only', False) # Just checking for duplicates, not saving - - if not lot_number: - return jsonify({'success': False, 'message': 'Lot number required'}) - - if not check_only and not weight: - return jsonify({'success': False, 'message': 'Weight required'}) - - if not check_only: - try: - weight = float(weight) - except ValueError: - return jsonify({'success': False, 'message': 'Invalid weight value'}) - - # Get location info - location = query_db('SELECT * FROM LocationCounts WHERE location_count_id = ?', - [location_count_id], one=True) - - # Check for duplicates in this session - existing_scans = query_db(''' - SELECT se.*, lc.location_name, u.full_name - FROM ScanEntries se - JOIN LocationCounts lc ON se.location_count_id = lc.location_count_id - JOIN Users u ON se.scanned_by = u.user_id - WHERE se.session_id = ? AND se.lot_number = ? AND se.is_deleted = 0 - ''', [session_id, lot_number]) - - duplicate_status = '00' # Default: no duplicate - duplicate_info = None - needs_confirmation = False - - if existing_scans: - # Check for same location duplicates (by this user) - same_location = [s for s in existing_scans - if s['location_name'] == location['location_name'] - and s['scanned_by'] == session['user_id']] - - # Check for different location duplicates (by anyone) - diff_location = [s for s in existing_scans - if s['location_name'] != location['location_name']] - - if same_location and diff_location: - # Status 04: Duplicate in both same and different locations - duplicate_status = '04' - other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) - duplicate_info = f"Also found in {other_locs}. Duplicate Lot" - needs_confirmation = True - - elif same_location: - # Status 01: Duplicate in same location only - duplicate_status = '01' - duplicate_info = "Duplicate" - needs_confirmation = True - - elif diff_location: - # Status 03: Duplicate in different location only - duplicate_status = '03' - other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) - duplicate_info = f"Also found in {other_locs}" - - # If just checking, return early with baseline info - if check_only: - # Get baseline info to show what they're scanning - master_info = query_db(''' - SELECT item, description FROM BaselineInventory_Master - WHERE session_id = ? AND lot_number = ? - LIMIT 1 - ''', [session_id, lot_number], one=True) - - if needs_confirmation: - return jsonify({ - 'success': False, - 'needs_confirmation': True, - 'message': 'Lot already scanned, Are you sure?', - 'duplicate_status': duplicate_status, - 'item': master_info['item'] if master_info else None, - 'description': master_info['description'] if master_info else None - }) - else: - return jsonify({ - 'success': True, - 'needs_confirmation': False, - 'item': master_info['item'] if master_info else None, - 'description': master_info['description'] if master_info else None - }) - - # If needs confirmation and not yet confirmed, ask user - if needs_confirmation and not confirm_duplicate: - return jsonify({ - 'success': False, - 'needs_confirmation': True, - 'message': 'Lot already scanned, Are you sure?', - 'duplicate_status': duplicate_status - }) - - # Check against MASTER baseline - master = query_db(''' - SELECT * FROM BaselineInventory_Master - WHERE session_id = ? AND lot_number = ? AND system_bin = ? - ''', [session_id, lot_number, location['location_name']], one=True) - - # Determine master_status (only if not a duplicate issue) - if duplicate_status == '00': - if master: - # Lot exists in correct location - master_status = 'match' - if master['system_quantity'] is not None: - variance_lbs = weight - master['system_quantity'] - variance_pct = (variance_lbs / master['system_quantity'] * 100) if master['system_quantity'] > 0 else 0 - else: - variance_lbs = None - variance_pct = None - else: - # Check if lot exists in different location - master_other = query_db(''' - SELECT * FROM BaselineInventory_Master - WHERE session_id = ? AND lot_number = ? - ''', [session_id, lot_number], one=True) - - if master_other: - master_status = 'wrong_location' - master = master_other - variance_lbs = None - variance_pct = None - else: - # Ghost lot - master_status = 'ghost_lot' - variance_lbs = None - variance_pct = None - else: - # For duplicates, still check baseline for item info - if not master: - master = query_db(''' - SELECT * FROM BaselineInventory_Master - WHERE session_id = ? AND lot_number = ? - ''', [session_id, lot_number], one=True) - master_status = 'match' # Don't override with wrong_location for duplicates - variance_lbs = None - variance_pct = None - - # Insert scan - entry_id = execute_db(''' - INSERT INTO ScanEntries - (session_id, location_count_id, lot_number, item, description, - scanned_location, actual_weight, scanned_by, - master_status, master_expected_location, master_expected_weight, - master_variance_lbs, master_variance_pct, - duplicate_status, duplicate_info, comment) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ''', [ - session_id, location_count_id, lot_number, - master['item'] if master else None, - master['description'] if master else None, - location['location_name'], weight, session['user_id'], - master_status, - master['system_bin'] if master else None, - master['system_quantity'] if master else None, - variance_lbs, variance_pct, - duplicate_status, duplicate_info, duplicate_info - ]) - - # If this is a confirmed duplicate (01 or 04), update previous scans in same location - updated_entry_ids = [] - if duplicate_status in ['01', '04'] and confirm_duplicate: - same_location_ids = [s['entry_id'] for s in existing_scans - if s['location_name'] == location['location_name'] - and s['scanned_by'] == session['user_id']] - - for scan_id in same_location_ids: - execute_db(''' - UPDATE ScanEntries - SET duplicate_status = ?, - duplicate_info = ?, - comment = ?, - modified_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [duplicate_status, duplicate_info, duplicate_info, scan_id]) - updated_entry_ids.append(scan_id) - - # Update location count - execute_db(''' - UPDATE LocationCounts - SET lots_found = lots_found + 1 - WHERE location_count_id = ? - ''', [location_count_id]) - - return jsonify({ - 'success': True, - 'entry_id': entry_id, - 'master_status': master_status, - 'duplicate_status': duplicate_status, - 'duplicate_info': duplicate_info, - 'master_expected_location': master['system_bin'] if master else None, - 'master_expected_weight': master['system_quantity'] if master else None, - 'actual_weight': weight, - 'variance_lbs': variance_lbs, - 'item': master['item'] if master else 'Unknown Item', - 'description': master['description'] if master else 'Not in system', - 'updated_entry_ids': updated_entry_ids # IDs of scans that were updated to duplicate - }) - - -@app.route('/scan//delete', methods=['POST']) -@login_required -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) - - if not scan: - return jsonify({'success': False, 'message': 'Scan not found'}) - - # Only allow user to delete their own scans - if scan['scanned_by'] != session['user_id'] and session['role'] not in ['owner', 'admin']: - return jsonify({'success': False, 'message': 'Permission denied'}) - - # Soft delete the scan - execute_db(''' - UPDATE ScanEntries - SET is_deleted = 1, - deleted_by = ?, - deleted_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [session['user_id'], entry_id]) - - # Recalculate duplicate statuses for this lot number in this session - updated_entries = recalculate_duplicate_status(scan['session_id'], scan['lot_number'], scan['scanned_location']) - - # Update location count - execute_db(''' - UPDATE LocationCounts - SET lots_found = lots_found - 1 - WHERE location_count_id = ? - ''', [scan['location_count_id']]) - - return jsonify({ - 'success': True, - 'message': 'Scan deleted', - 'updated_entries': updated_entries # Return which scans were updated - }) - - -@app.route('/scan//details', methods=['GET']) -@login_required -def get_scan_details(entry_id): - """Get detailed information about a scan""" - 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'}) - - return jsonify({ - 'success': True, - 'scan': dict(scan) - }) - - -@app.route('/scan//update', methods=['POST']) -@login_required -def update_scan(entry_id): - """Update scan item, weight and comment""" - data = request.get_json() - item = data.get('item', '').strip() - weight = data.get('weight') - comment = data.get('comment', '') - - # Get the scan - scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ?', [entry_id], one=True) - - if not scan: - return jsonify({'success': False, 'message': 'Scan not found'}) - - # Only allow user to update their own scans - if scan['scanned_by'] != session['user_id'] and session['role'] not in ['owner', 'admin']: - return jsonify({'success': False, 'message': 'Permission denied'}) - - try: - weight = float(weight) - except ValueError: - return jsonify({'success': False, 'message': 'Invalid weight value'}) - - # Update the scan - execute_db(''' - UPDATE ScanEntries - SET item = ?, - actual_weight = ?, - comment = ?, - modified_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [item, weight, comment, entry_id]) - - return jsonify({'success': True, 'message': 'Scan updated'}) - - -def recalculate_duplicate_status(session_id, lot_number, current_location): - """Recalculate duplicate statuses for a lot after deletion""" - # Track which entries were updated - updated_entries = [] - - # Get all active scans for this lot in this session - scans = query_db(''' - SELECT se.*, lc.location_name, u.full_name, u.user_id as scan_user_id - FROM ScanEntries se - JOIN LocationCounts lc ON se.location_count_id = lc.location_count_id - JOIN Users u ON se.scanned_by = u.user_id - WHERE se.session_id = ? AND se.lot_number = ? AND se.is_deleted = 0 - ORDER BY se.scan_timestamp - ''', [session_id, lot_number]) - - if not scans: - return updated_entries - - # Reset all to status 00 - for scan in scans: - execute_db(''' - UPDATE ScanEntries - SET duplicate_status = '00', - duplicate_info = NULL, - comment = NULL, - modified_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [scan['entry_id']]) - updated_entries.append({ - 'entry_id': scan['entry_id'], - 'duplicate_status': '00', - 'duplicate_info': None - }) - - # Recalculate statuses - for i, scan in enumerate(scans): - # Get previous scans (before this one chronologically) - prev_scans = scans[:i] - - if not prev_scans: - continue # First scan, stays 00 - - same_location = [s for s in prev_scans if s['location_name'] == scan['location_name'] and s['scan_user_id'] == scan['scan_user_id']] - diff_location = [s for s in prev_scans if s['location_name'] != scan['location_name']] - - duplicate_status = '00' - duplicate_info = None - - if same_location and diff_location: - # Status 04 - duplicate_status = '04' - other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) - duplicate_info = f"Also found in {other_locs}. Duplicate Lot" - elif same_location: - # Status 01 - duplicate_status = '01' - duplicate_info = "Duplicate" - elif diff_location: - # Status 03 - duplicate_status = '03' - other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) - duplicate_info = f"Also found in {other_locs}" - - # Update this scan if it changed from 00 - if duplicate_status != '00': - execute_db(''' - UPDATE ScanEntries - SET duplicate_status = ?, - duplicate_info = ?, - comment = ?, - modified_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [duplicate_status, duplicate_info, duplicate_info, scan['entry_id']]) - - # Update our tracking list - for entry in updated_entries: - if entry['entry_id'] == scan['entry_id']: - entry['duplicate_status'] = duplicate_status - entry['duplicate_info'] = duplicate_info - break - - # If status 01 or 04, also update previous scans in same location - if duplicate_status in ['01', '04'] and same_location: - for prev_scan in same_location: - execute_db(''' - UPDATE ScanEntries - SET duplicate_status = ?, - duplicate_info = ?, - comment = ?, - modified_timestamp = CURRENT_TIMESTAMP - WHERE entry_id = ? - ''', [duplicate_status, duplicate_info, duplicate_info, prev_scan['entry_id']]) - - # Update tracking for previous scans - for entry in updated_entries: - if entry['entry_id'] == prev_scan['entry_id']: - entry['duplicate_status'] = duplicate_status - entry['duplicate_info'] = duplicate_info - break - - return updated_entries - - -@app.route('/count//location//finish', methods=['POST']) -@login_required -def finish_location(session_id, location_count_id): - """Finish counting a location""" - # Get location info - location = query_db('SELECT * FROM LocationCounts WHERE location_count_id = ?', - [location_count_id], one=True) - - if not location: - return jsonify({'success': False, 'message': 'Location not found'}) - - # Mark location as completed - execute_db(''' - UPDATE LocationCounts - SET status = 'completed', end_timestamp = CURRENT_TIMESTAMP - WHERE location_count_id = ? - ''', [location_count_id]) - - # V1.0: Mark missing lots from MASTER baseline that weren't scanned - # Get all expected lots for this location 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, location['location_name']]) - - # Get all scanned lots for this location - scanned_lots = query_db(''' - SELECT DISTINCT lot_number - FROM ScanEntries - WHERE location_count_id = ? AND is_deleted = 0 - ''', [location_count_id]) - - scanned_lot_numbers = {s['lot_number'] for s in scanned_lots} - - # Insert missing 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'], location['location_name'], - expected['item'], expected['system_quantity'], session['user_id']]) - - flash('Location count completed!', 'success') - return jsonify({ - 'success': True, - 'redirect': url_for('count_session', session_id=session_id) - }) - # ==================== PWA SUPPORT ROUTES ==================== diff --git a/blueprints/__pycache__/counting.cpython-313.pyc b/blueprints/__pycache__/counting.cpython-313.pyc new file mode 100644 index 0000000..6fa32d0 Binary files /dev/null and b/blueprints/__pycache__/counting.cpython-313.pyc differ diff --git a/blueprints/__pycache__/data_imports.cpython-313.pyc b/blueprints/__pycache__/data_imports.cpython-313.pyc index 944775e..acb87fb 100644 Binary files a/blueprints/__pycache__/data_imports.cpython-313.pyc and b/blueprints/__pycache__/data_imports.cpython-313.pyc differ diff --git a/blueprints/counting.py b/blueprints/counting.py new file mode 100644 index 0000000..58e4b1c --- /dev/null +++ b/blueprints/counting.py @@ -0,0 +1,633 @@ +from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, session +from db import query_db, execute_db, get_db +from utils import login_required + +counting_bp = Blueprint('counting', __name__) + +@counting_bp.route('/count/') +@login_required +def count_session(session_id): + """Select session and begin counting""" + sess = query_db('SELECT * FROM CountSessions WHERE session_id = ? AND status = "active"', + [session_id], one=True) + + if not sess: + flash('Session not found or not active', 'danger') + return redirect(url_for('dashboard')) + + # Redirect to my_counts page (staff can manage multiple bins) + return redirect(url_for('counting.my_counts', session_id=session_id)) + + +@counting_bp.route('/session//my-counts') +@login_required +def my_counts(session_id): + """Staff view of their active and completed bins""" + 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 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']]) + + # Get this user's completed bins + completed_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 = 'completed' + GROUP BY lc.location_count_id + ORDER BY lc.end_timestamp DESC + ''', [session_id, session['user_id']]) + + return render_template('my_counts.html', + count_session=sess, + active_bins=active_bins, + completed_bins=completed_bins) + + +@counting_bp.route('/session//start-bin', methods=['POST']) +@login_required +def start_bin_count(session_id): + """Start counting a new bin""" + location_name = request.form.get('location_name', '').strip().upper() + + if not location_name: + flash('Bin number is required', 'danger') + return redirect(url_for('counting.my_counts', session_id=session_id)) + + # Count expected lots from MASTER baseline for this location + expected_lots = query_db(''' + SELECT COUNT(DISTINCT lot_number) as count + FROM BaselineInventory_Master + WHERE session_id = ? AND system_bin = ? + ''', [session_id, location_name], one=True) + + expected_count = expected_lots['count'] if expected_lots else 0 + + # Create new location count + conn = get_db() + cursor = conn.cursor() + + cursor.execute(''' + INSERT INTO LocationCounts (session_id, location_name, counted_by, status, start_timestamp, expected_lots_master) + VALUES (?, ?, ?, 'in_progress', CURRENT_TIMESTAMP, ?) + ''', [session_id, location_name, session['user_id'], expected_count]) + + location_count_id = cursor.lastrowid + conn.commit() + conn.close() + + 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//complete', methods=['POST']) +@login_required +def complete_location(location_count_id): + """Mark a location count as complete (Simple toggle)""" + # 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'}) + + # Mark as completed + execute_db(''' + UPDATE LocationCounts + SET status = 'completed', end_timestamp = CURRENT_TIMESTAMP + WHERE location_count_id = ? + ''', [location_count_id]) + + return jsonify({'success': True, 'message': 'Bin marked as complete'}) + + +@counting_bp.route('/count//location/') +@login_required +def count_location(session_id, location_count_id): + """Count lots in a specific location""" + # Get session info to determine type (Cycle Count vs Physical) + sess = query_db('SELECT * FROM CountSessions WHERE session_id = ?', [session_id], one=True) + + location = query_db(''' + SELECT * FROM LocationCounts + WHERE location_count_id = ? AND session_id = ? + ''', [location_count_id, session_id], one=True) + + if not location: + flash('Location not found', 'danger') + return redirect(url_for('counting.count_session', session_id=session_id)) + + # Check if location is completed and user is staff (not admin/owner) + if location['status'] == 'completed' and session['role'] == 'staff': + flash(f'Location {location["location_name"]} has been finalized and cannot accept new scans', 'warning') + return redirect(url_for('counting.my_counts', session_id=session_id)) + + # Get scans for this location (Scanned Lots) + scans = query_db(''' + SELECT * FROM ScanEntries + WHERE location_count_id = ? AND is_deleted = 0 + ORDER BY scan_timestamp DESC + ''', [location_count_id]) + + # NEW LOGIC: Get Expected Lots for Cycle Counts (Grouped & Summed) + expected_lots = [] + if sess and sess['session_type'] == 'cycle_count': + expected_lots = query_db(''' + SELECT + lot_number, + MAX(item) as item, -- Pick one item code if they differ (rare) + SUM(system_quantity) as total_weight + FROM BaselineInventory_Master + WHERE session_id = ? + AND system_bin = ? + AND lot_number NOT IN ( + SELECT lot_number + FROM ScanEntries + WHERE location_count_id = ? + AND is_deleted = 0 + ) + GROUP BY lot_number + ORDER BY lot_number + ''', [session_id, location['location_name'], location_count_id]) + + return render_template('count_location.html', + session_id=session_id, + location=location, + scans=scans, + expected_lots=expected_lots, + session_type=sess['session_type'] if sess else '') + + +@counting_bp.route('/count//location//scan', methods=['POST']) +@login_required +def scan_lot(session_id, location_count_id): + """Process a lot scan with duplicate detection""" + data = request.get_json() + lot_number = data.get('lot_number', '').strip() + weight = data.get('weight') + confirm_duplicate = data.get('confirm_duplicate', False) + check_only = data.get('check_only', False) # Just checking for duplicates, not saving + + if not lot_number: + return jsonify({'success': False, 'message': 'Lot number required'}) + + if not check_only and not weight: + return jsonify({'success': False, 'message': 'Weight required'}) + + if not check_only: + try: + weight = float(weight) + except ValueError: + return jsonify({'success': False, 'message': 'Invalid weight value'}) + + # Get location info + location = query_db('SELECT * FROM LocationCounts WHERE location_count_id = ?', + [location_count_id], one=True) + + # Check for duplicates in this session + existing_scans = query_db(''' + SELECT se.*, lc.location_name, u.full_name + FROM ScanEntries se + JOIN LocationCounts lc ON se.location_count_id = lc.location_count_id + JOIN Users u ON se.scanned_by = u.user_id + WHERE se.session_id = ? AND se.lot_number = ? AND se.is_deleted = 0 + ''', [session_id, lot_number]) + + duplicate_status = '00' # Default: no duplicate + duplicate_info = None + needs_confirmation = False + + if existing_scans: + # Check for same location duplicates (by this user) + same_location = [s for s in existing_scans + if s['location_name'] == location['location_name'] + and s['scanned_by'] == session['user_id']] + + # Check for different location duplicates (by anyone) + diff_location = [s for s in existing_scans + if s['location_name'] != location['location_name']] + + if same_location and diff_location: + # Status 04: Duplicate in both same and different locations + duplicate_status = '04' + other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) + duplicate_info = f"Also found in {other_locs}. Duplicate Lot" + needs_confirmation = True + + elif same_location: + # Status 01: Duplicate in same location only + duplicate_status = '01' + duplicate_info = "Duplicate" + needs_confirmation = True + + elif diff_location: + # Status 03: Duplicate in different location only + duplicate_status = '03' + other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) + duplicate_info = f"Also found in {other_locs}" + + # If just checking, return early with baseline info + if check_only: + # Get baseline info to show what they're scanning + master_info = query_db(''' + SELECT item, description FROM BaselineInventory_Master + WHERE session_id = ? AND lot_number = ? + LIMIT 1 + ''', [session_id, lot_number], one=True) + + if needs_confirmation: + return jsonify({ + 'success': False, + 'needs_confirmation': True, + 'message': 'Lot already scanned, Are you sure?', + 'duplicate_status': duplicate_status, + 'item': master_info['item'] if master_info else None, + 'description': master_info['description'] if master_info else None + }) + else: + return jsonify({ + 'success': True, + 'needs_confirmation': False, + 'item': master_info['item'] if master_info else None, + 'description': master_info['description'] if master_info else None + }) + + # If needs confirmation and not yet confirmed, ask user + if needs_confirmation and not confirm_duplicate: + return jsonify({ + 'success': False, + 'needs_confirmation': True, + 'message': 'Lot already scanned, Are you sure?', + 'duplicate_status': duplicate_status + }) + + # Check against MASTER baseline + master = query_db(''' + SELECT * FROM BaselineInventory_Master + WHERE session_id = ? AND lot_number = ? AND system_bin = ? + ''', [session_id, lot_number, location['location_name']], one=True) + + # Determine master_status (only if not a duplicate issue) + if duplicate_status == '00': + if master: + # Lot exists in correct location + master_status = 'match' + if master['system_quantity'] is not None: + variance_lbs = weight - master['system_quantity'] + variance_pct = (variance_lbs / master['system_quantity'] * 100) if master['system_quantity'] > 0 else 0 + else: + variance_lbs = None + variance_pct = None + else: + # Check if lot exists in different location + master_other = query_db(''' + SELECT * FROM BaselineInventory_Master + WHERE session_id = ? AND lot_number = ? + ''', [session_id, lot_number], one=True) + + if master_other: + master_status = 'wrong_location' + master = master_other + variance_lbs = None + variance_pct = None + else: + # Ghost lot + master_status = 'ghost_lot' + variance_lbs = None + variance_pct = None + else: + # For duplicates, still check baseline for item info + if not master: + master = query_db(''' + SELECT * FROM BaselineInventory_Master + WHERE session_id = ? AND lot_number = ? + ''', [session_id, lot_number], one=True) + master_status = 'match' # Don't override with wrong_location for duplicates + variance_lbs = None + variance_pct = None + + # Insert scan + entry_id = execute_db(''' + INSERT INTO ScanEntries + (session_id, location_count_id, lot_number, item, description, + scanned_location, actual_weight, scanned_by, + master_status, master_expected_location, master_expected_weight, + master_variance_lbs, master_variance_pct, + duplicate_status, duplicate_info, comment) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ''', [ + session_id, location_count_id, lot_number, + master['item'] if master else None, + master['description'] if master else None, + location['location_name'], weight, session['user_id'], + master_status, + master['system_bin'] if master else None, + master['system_quantity'] if master else None, + variance_lbs, variance_pct, + duplicate_status, duplicate_info, duplicate_info + ]) + + # If this is a confirmed duplicate (01 or 04), update previous scans in same location + updated_entry_ids = [] + if duplicate_status in ['01', '04'] and confirm_duplicate: + same_location_ids = [s['entry_id'] for s in existing_scans + if s['location_name'] == location['location_name'] + and s['scanned_by'] == session['user_id']] + + for scan_id in same_location_ids: + execute_db(''' + UPDATE ScanEntries + SET duplicate_status = ?, + duplicate_info = ?, + comment = ?, + modified_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [duplicate_status, duplicate_info, duplicate_info, scan_id]) + updated_entry_ids.append(scan_id) + + # Update location count + execute_db(''' + UPDATE LocationCounts + SET lots_found = lots_found + 1 + WHERE location_count_id = ? + ''', [location_count_id]) + + return jsonify({ + 'success': True, + 'entry_id': entry_id, + 'master_status': master_status, + 'duplicate_status': duplicate_status, + 'duplicate_info': duplicate_info, + 'master_expected_location': master['system_bin'] if master else None, + 'master_expected_weight': master['system_quantity'] if master else None, + 'actual_weight': weight, + 'variance_lbs': variance_lbs, + 'item': master['item'] if master else 'Unknown Item', + 'description': master['description'] if master else 'Not in system', + 'updated_entry_ids': updated_entry_ids # IDs of scans that were updated to duplicate + }) + + +@counting_bp.route('/scan//delete', methods=['POST']) +@login_required +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) + + if not scan: + return jsonify({'success': False, 'message': 'Scan not found'}) + + # Only allow user to delete their own scans + if scan['scanned_by'] != session['user_id'] and session['role'] not in ['owner', 'admin']: + return jsonify({'success': False, 'message': 'Permission denied'}) + + # Soft delete the scan + execute_db(''' + UPDATE ScanEntries + SET is_deleted = 1, + deleted_by = ?, + deleted_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [session['user_id'], entry_id]) + + # Recalculate duplicate statuses for this lot number in this session + updated_entries = recalculate_duplicate_status(scan['session_id'], scan['lot_number'], scan['scanned_location']) + + # Update location count + execute_db(''' + UPDATE LocationCounts + SET lots_found = lots_found - 1 + WHERE location_count_id = ? + ''', [scan['location_count_id']]) + + return jsonify({ + 'success': True, + 'message': 'Scan deleted', + 'updated_entries': updated_entries # Return which scans were updated + }) + + +@counting_bp.route('/scan//details', methods=['GET']) +@login_required +def get_scan_details(entry_id): + """Get detailed information about a scan""" + 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'}) + + return jsonify({ + 'success': True, + 'scan': dict(scan) + }) + + +@counting_bp.route('/scan//update', methods=['POST']) +@login_required +def update_scan(entry_id): + """Update scan item, weight and comment""" + data = request.get_json() + item = data.get('item', '').strip() + weight = data.get('weight') + comment = data.get('comment', '') + + # Get the scan + scan = query_db('SELECT * FROM ScanEntries WHERE entry_id = ?', [entry_id], one=True) + + if not scan: + return jsonify({'success': False, 'message': 'Scan not found'}) + + # Only allow user to update their own scans + if scan['scanned_by'] != session['user_id'] and session['role'] not in ['owner', 'admin']: + return jsonify({'success': False, 'message': 'Permission denied'}) + + try: + weight = float(weight) + except ValueError: + return jsonify({'success': False, 'message': 'Invalid weight value'}) + + # Update the scan + execute_db(''' + UPDATE ScanEntries + SET item = ?, + actual_weight = ?, + comment = ?, + modified_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [item, weight, comment, entry_id]) + + return jsonify({'success': True, 'message': 'Scan updated'}) + + +def recalculate_duplicate_status(session_id, lot_number, current_location): + """Recalculate duplicate statuses for a lot after deletion""" + # Track which entries were updated + updated_entries = [] + + # Get all active scans for this lot in this session + scans = query_db(''' + SELECT se.*, lc.location_name, u.full_name, u.user_id as scan_user_id + FROM ScanEntries se + JOIN LocationCounts lc ON se.location_count_id = lc.location_count_id + JOIN Users u ON se.scanned_by = u.user_id + WHERE se.session_id = ? AND se.lot_number = ? AND se.is_deleted = 0 + ORDER BY se.scan_timestamp + ''', [session_id, lot_number]) + + if not scans: + return updated_entries + + # Reset all to status 00 + for scan in scans: + execute_db(''' + UPDATE ScanEntries + SET duplicate_status = '00', + duplicate_info = NULL, + comment = NULL, + modified_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [scan['entry_id']]) + updated_entries.append({ + 'entry_id': scan['entry_id'], + 'duplicate_status': '00', + 'duplicate_info': None + }) + + # Recalculate statuses + for i, scan in enumerate(scans): + # Get previous scans (before this one chronologically) + prev_scans = scans[:i] + + if not prev_scans: + continue # First scan, stays 00 + + same_location = [s for s in prev_scans if s['location_name'] == scan['location_name'] and s['scan_user_id'] == scan['scan_user_id']] + diff_location = [s for s in prev_scans if s['location_name'] != scan['location_name']] + + duplicate_status = '00' + duplicate_info = None + + if same_location and diff_location: + # Status 04 + duplicate_status = '04' + other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) + duplicate_info = f"Also found in {other_locs}. Duplicate Lot" + elif same_location: + # Status 01 + duplicate_status = '01' + duplicate_info = "Duplicate" + elif diff_location: + # Status 03 + duplicate_status = '03' + other_locs = ', '.join(set(f"{s['location_name']} by {s['full_name']}" for s in diff_location)) + duplicate_info = f"Also found in {other_locs}" + + # Update this scan if it changed from 00 + if duplicate_status != '00': + execute_db(''' + UPDATE ScanEntries + SET duplicate_status = ?, + duplicate_info = ?, + comment = ?, + modified_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [duplicate_status, duplicate_info, duplicate_info, scan['entry_id']]) + + # Update our tracking list + for entry in updated_entries: + if entry['entry_id'] == scan['entry_id']: + entry['duplicate_status'] = duplicate_status + entry['duplicate_info'] = duplicate_info + break + + # If status 01 or 04, also update previous scans in same location + if duplicate_status in ['01', '04'] and same_location: + for prev_scan in same_location: + execute_db(''' + UPDATE ScanEntries + SET duplicate_status = ?, + duplicate_info = ?, + comment = ?, + modified_timestamp = CURRENT_TIMESTAMP + WHERE entry_id = ? + ''', [duplicate_status, duplicate_info, duplicate_info, prev_scan['entry_id']]) + + # Update tracking for previous scans + for entry in updated_entries: + if entry['entry_id'] == prev_scan['entry_id']: + entry['duplicate_status'] = duplicate_status + entry['duplicate_info'] = duplicate_info + break + + return updated_entries + + +@counting_bp.route('/count//location//finish', methods=['POST']) +@login_required +def finish_location(session_id, location_count_id): + """Finish counting a location""" + # Get location info + location = query_db('SELECT * FROM LocationCounts WHERE location_count_id = ?', + [location_count_id], one=True) + + if not location: + return jsonify({'success': False, 'message': 'Location not found'}) + + # Mark location as completed + execute_db(''' + UPDATE LocationCounts + SET status = 'completed', end_timestamp = CURRENT_TIMESTAMP + WHERE location_count_id = ? + ''', [location_count_id]) + + # V1.0: Mark missing lots from MASTER baseline that weren't scanned + # Get all expected lots for this location 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, location['location_name']]) + + # Get all scanned lots for this location + scanned_lots = query_db(''' + SELECT DISTINCT lot_number + FROM ScanEntries + WHERE location_count_id = ? AND is_deleted = 0 + ''', [location_count_id]) + + scanned_lot_numbers = {s['lot_number'] for s in scanned_lots} + + # Insert missing 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'], location['location_name'], + expected['item'], expected['system_quantity'], session['user_id']]) + + flash('Location count completed!', 'success') + return jsonify({ + 'success': True, + 'redirect': url_for('counting.count_session', session_id=session_id) + }) \ No newline at end of file diff --git a/blueprints/data_imports.py b/blueprints/data_imports.py index 89cca65..fcb802f 100644 --- a/blueprints/data_imports.py +++ b/blueprints/data_imports.py @@ -17,12 +17,12 @@ def upload_current(session_id): if 'csv_file' not in request.files: flash('No file part', 'danger') - return redirect(url_for('session_detail', session_id=session_id)) + return redirect(url_for('sessions.session_detail', session_id=session_id)) file = request.files['csv_file'] if file.filename == '': flash('No selected file', 'danger') - return redirect(url_for('session_detail', session_id=session_id)) + return redirect(url_for('sessions.session_detail', session_id=session_id)) if file: conn = get_db() @@ -76,7 +76,7 @@ def upload_current(session_id): finally: conn.close() - return redirect(url_for('session_detail', session_id=session_id)) + return redirect(url_for('sessions.session_detail', session_id=session_id)) # --- ROUTE 2: Upload MASTER Baseline (Session Specific) --- @data_imports_bp.route('/session//upload_master', methods=['POST']) @@ -85,12 +85,12 @@ def upload_master(session_id): if 'csv_file' not in request.files: flash('No file uploaded', 'danger') - return redirect(url_for('session_detail', session_id=session_id)) + return redirect(url_for('sessions.session_detail', session_id=session_id)) file = request.files['csv_file'] if file.filename == '': flash('No file selected', 'danger') - return redirect(url_for('session_detail', session_id=session_id)) + return redirect(url_for('sessions.session_detail', session_id=session_id)) conn = get_db() cursor = conn.cursor() @@ -152,4 +152,4 @@ def upload_master(session_id): finally: conn.close() - return redirect(url_for('session_detail', session_id=session_id)) \ No newline at end of file + return redirect(url_for('sessions. session_detail', session_id=session_id)) \ No newline at end of file diff --git a/templates/base.html b/templates/base.html index 975de22..d0e80d5 100644 --- a/templates/base.html +++ b/templates/base.html @@ -62,6 +62,8 @@

© 2026 Javier Torres. All Rights Reserved.

+

v{{ version }}

+
diff --git a/templates/count_location.html b/templates/count_location.html index dadc302..33c2a10 100644 --- a/templates/count_location.html +++ b/templates/count_location.html @@ -97,9 +97,9 @@ {% set row_class = 'weight_discrepancy' %} {% endif %} -
+
{{ scan.lot_number }}
{{ scan.item or 'N/A' }}
{{ scan.actual_weight }} lbs
@@ -160,7 +160,7 @@
- + ← Back to My Counts
- + Resume Counting
- -
@@ -105,7 +105,7 @@
-
+
diff --git a/templates/session_detail.html b/templates/session_detail.html index 9cf43b4..446cf82 100644 --- a/templates/session_detail.html +++ b/templates/session_detail.html @@ -8,8 +8,9 @@
← Back to Dashboard

{{ count_session.session_name }}

+ - {{ 'Full Physical' if session.session_type == 'full_physical' else 'Cycle Count' }} + {{ 'Full Physical' if count_session.session_type == 'full_physical' else 'Cycle Count' }}
@@ -30,6 +31,7 @@ {% endif %}
{% if not count_session.master_baseline_timestamp %} + @@ -41,8 +43,8 @@
CURRENT Baseline (Optional)
{% if count_session.current_baseline_timestamp %} - Last Updated: -
{{ count_session.current_baseline_timestamp[:16] if count_session.current_baseline_timestamp else 'Never' }}
+ ✓ Uploaded + {{ count_session.current_baseline_timestamp[:16] }} {% else %} Not Uploaded {% endif %} @@ -65,27 +67,27 @@

Real-Time Statistics

-
+
{{ stats.matched or 0 }}
✓ Matched
-
+
{{ stats.duplicates or 0 }}
🔵 Duplicates
-
+
{{ stats.weight_discrepancy or 0 }}
⚖️ Weight Discrepancy
-
+
{{ stats.wrong_location or 0 }}
⚠ Wrong Location
-
+
{{ stats.ghost_lots or 0 }}
🟣 Ghost Lots
-
+
{{ stats.missing_lots or 0 }}
🔴 Missing
@@ -129,7 +131,12 @@ {% for loc in locations %} - + + {{ loc.location_name }} @@ -246,7 +253,10 @@
-{% endblock %} +{% endblock %} \ No newline at end of file diff --git a/templates/staff_dashboard.html b/templates/staff_dashboard.html index 655c1fe..c401697 100644 --- a/templates/staff_dashboard.html +++ b/templates/staff_dashboard.html @@ -7,11 +7,11 @@ {% if session.role in ['owner', 'admin'] %}
- + + Admin Console +
{% endif %} @@ -22,11 +22,11 @@ {% if sessions %}
- {% for session in sessions %} - + {% for s in sessions %} +
-

{{ session.session_name }}

- {{ 'Full Physical' if session.session_type == 'full_physical' else 'Cycle Count' }} +

{{ s.session_name }}

+ {{ 'Full Physical' if s.session_type == 'full_physical' else 'Cycle Count' }}
→