""" routes/dashboard.py =================== Dashboard and related API routes. Routes: /dashboard, /project//qr-codes, /dashboard/search, /api/dashboard/stats, /api/dashboard/realtime """ from flask import abort, Blueprint, render_template, request, redirect, flash, session, jsonify, url_for from datetime import datetime, timedelta, date, time from sqlalchemy import and_ from extensions import db, logger_handler from models.attendance import AttendanceData from models.project import Project from models.qrcode import QRCode from models.user import User from logger_handler import log_user_activity, log_database_operations from utils.helpers import login_required, load_project_manager_scope def _project_manager_qr_filter(): """ (is_pm, qr_filter) for the session user — the QR codes a Project Manager may see (§4). qr_filter is None for everyone else, and a never-true condition for a PM with no assignments. """ is_pm, allowed_project_ids, allowed_location_names = load_project_manager_scope() if not is_pm: return False, None # AND, not OR — the same rule the Attendance Report applies # (_attendance_filter_conditions in routes/attendance.py): the assigned # projects, narrowed to the assigned locations when there are any. # With OR, a PM assigned location "Bldg A" also saw every OTHER project's # QR codes that happened to use that same location name (Set 28). scope = [] if allowed_project_ids: scope.append(QRCode.project_id.in_(allowed_project_ids)) if allowed_location_names: scope.append(QRCode.location.in_(allowed_location_names)) if not scope: return True, QRCode.id.is_(None) # assigned nothing → sees nothing return True, and_(*scope) bp = Blueprint('dashboard', __name__) @bp.route('/dashboard', endpoint='dashboard') @login_required def dashboard(): """Enhanced project-centric dashboard with search filters""" try: user = db.session.get(User, session['user_id']) # Get search parameters from URL search_name = request.args.get('search_name', '').strip() search_status = request.args.get('search_status', '').strip() # Build QR codes query with filters qr_query = QRCode.query # Project Managers only see QR codes in their assigned projects/locations is_pm, pm_qr_filter = _project_manager_qr_filter() if pm_qr_filter is not None: qr_query = qr_query.filter(pm_qr_filter) # Apply name filter if provided if search_name: qr_query = qr_query.filter(QRCode.name.ilike(f'%{search_name}%')) # Apply status filter if provided if search_status == 'active': qr_query = qr_query.filter(QRCode.active_status == True) elif search_status == 'inactive': qr_query = qr_query.filter(QRCode.active_status == False) # Execute query qr_codes = qr_query.order_by(QRCode.created_date.desc()).all() project_query = Project.query if is_pm: # Only the projects the PM is assigned to, plus those behind their locations visible_project_ids = {qr.project_id for qr in qr_codes if qr.project_id} _, allowed_project_ids, _ = load_project_manager_scope() visible_project_ids.update(allowed_project_ids) project_query = (project_query.filter(Project.id.in_(sorted(visible_project_ids))) if visible_project_ids else project_query.filter(Project.id.is_(None))) projects = project_query.order_by(Project.name.asc()).all() # Log dashboard access with filter info filter_info = [] if search_name: filter_info.append(f"name contains '{search_name}'") if search_status: filter_info.append(f"status is {search_status}") log_message = f"User {session['username']} accessed dashboard: {len(qr_codes)} QR codes" if filter_info: log_message += f" (filtered: {', '.join(filter_info)})" logger_handler.logger.info(log_message) return render_template('dashboard.html', user=user, qr_codes=qr_codes, projects=projects, search_name=search_name, search_status=search_status) except Exception as e: db.session.rollback() logger_handler.log_database_error('dashboard_load', e) flash('Error loading dashboard. Please try again.', 'error') return redirect(url_for('auth.login')) @bp.route('/project//qr-codes', endpoint='project_qr_codes') @login_required def project_qr_codes(project_id): """ View all QR codes for a specific project with search filters Allows filtering by name and status within the project """ try: # Get the project project = db.session.get(Project, project_id) if project is None: abort(404) # Project Managers may only open their own projects (§4) is_pm, allowed_project_ids, _ = load_project_manager_scope() if is_pm and project_id not in allowed_project_ids: logger_handler.logger.warning( f"Project Manager {session.get('username')} tried to open project {project_id}" ) flash('You do not have permission to view that project.', 'error') return redirect(url_for('dashboard.dashboard')) # Get search parameters from URL search_name = request.args.get('search_name', '').strip() search_status = request.args.get('search_status', '').strip() # Build QR codes query with filters for this project only qr_query = QRCode.query.filter_by(project_id=project_id) # Apply name filter if provided if search_name: qr_query = qr_query.filter(QRCode.name.ilike(f'%{search_name}%')) # Apply status filter if provided if search_status == 'active': qr_query = qr_query.filter(QRCode.active_status == True) elif search_status == 'inactive': qr_query = qr_query.filter(QRCode.active_status == False) # Execute query qr_codes = qr_query.order_by(QRCode.created_date.desc()).all() # Log access with filter info filter_info = [] if search_name: filter_info.append(f"name contains '{search_name}'") if search_status: filter_info.append(f"status is {search_status}") log_message = f"User {session['username']} viewed project '{project.name}' QR codes: {len(qr_codes)} QR codes" if filter_info: log_message += f" (filtered: {', '.join(filter_info)})" logger_handler.logger.info(log_message) return render_template('project_qr_codes.html', project=project, qr_codes=qr_codes, search_name=search_name, search_status=search_status) except Exception as e: db.session.rollback() logger_handler.log_database_error('project_qr_codes_view', e) flash('Error loading project QR codes. Please try again.', 'error') return redirect(url_for('dashboard.dashboard')) @bp.route('/dashboard/search', methods=['GET'], endpoint='search_qr_codes') @login_required def search_qr_codes(): """Search QR codes - redirect to dashboard with filters""" search_name = request.args.get('search_name', '').strip() search_status = request.args.get('search_status', '').strip() # Log search activity logger_handler.logger.info( f"User {session['username']} searched QR codes: " f"name='{search_name}', status='{search_status}'" ) # Redirect to dashboard with search parameters return redirect(url_for('dashboard.dashboard', search_name=search_name, search_status=search_status)) @bp.route('/api/dashboard/stats', endpoint='dashboard_stats_api') @login_required def dashboard_stats_api(): """API endpoint for dashboard statistics""" try: # Project Manager scope (§4): every count below is restricted to the QR # codes / locations they are assigned to. is_pm, allowed_project_ids, allowed_location_names = load_project_manager_scope() _, pm_qr_filter = _project_manager_qr_filter() qr_base_query = QRCode.query.filter_by(active_status=True) attendance_base_query = AttendanceData.query location_base_query = db.session.query(AttendanceData.location_name) project_base_query = Project.query.filter_by(active_status=True) if is_pm: qr_base_query = qr_base_query.filter(pm_qr_filter) attendance_scope = [] if allowed_project_ids: attendance_scope.append(AttendanceData.qr_code_id.in_( db.session.query(QRCode.id).filter(QRCode.project_id.in_(allowed_project_ids)) )) project_base_query = project_base_query.filter(Project.id.in_(allowed_project_ids)) else: project_base_query = project_base_query.filter(Project.id.is_(None)) if allowed_location_names: attendance_scope.append(AttendanceData.location_name.in_(allowed_location_names)) attendance_condition = and_(*attendance_scope) if attendance_scope else AttendanceData.id.is_(None) attendance_base_query = attendance_base_query.filter(attendance_condition) location_base_query = location_base_query.filter(attendance_condition) # Get current stats total_qr_codes = qr_base_query.count() # Today's check-ins — local date, matching how check-ins are stored today = datetime.now().date() today_checkins = attendance_base_query.filter( AttendanceData.check_in_date == today ).count() # Active projects active_projects = project_base_query.count() # Unique locations unique_locations = location_base_query.distinct().count() # Calculate trends (compared to last month) last_month = datetime.now() - timedelta(days=30) # QR codes trend old_qr_count = qr_base_query.filter( QRCode.created_date <= last_month ).count() qr_change = ((total_qr_codes - old_qr_count) / max(old_qr_count, 1)) * 100 # Check-ins trend (yesterday) yesterday = today - timedelta(days=1) yesterday_checkins = attendance_base_query.filter( AttendanceData.check_in_date == yesterday ).count() checkin_change = ((today_checkins - yesterday_checkins) / max(yesterday_checkins, 1)) * 100 return jsonify({ 'success': True, 'total_qr_codes': total_qr_codes, 'today_checkins': today_checkins, 'active_projects': active_projects, 'unique_locations': unique_locations, 'qr_change': round(qr_change, 1), 'checkin_change': round(checkin_change, 1), 'project_change': 0, # You can calculate this based on your needs 'location_change': 0 # You can calculate this based on your needs }) except Exception as e: db.session.rollback() logger_handler.log_database_error('dashboard_stats_api', e) return jsonify({ 'success': False, 'error': 'Failed to fetch dashboard statistics' }), 500 @bp.route('/api/dashboard/realtime', endpoint='dashboard_realtime_api') @login_required def dashboard_realtime_api(): """API endpoint for real-time dashboard data""" try: # Get recent activity (last 10 check-ins) recent_query = db.session.query( AttendanceData.employee_id, AttendanceData.location_name, AttendanceData.check_in_time, AttendanceData.check_in_date ) # Project Managers only see activity at their own projects/locations (§4) is_pm, allowed_project_ids, allowed_location_names = load_project_manager_scope() if is_pm: recent_scope = [] if allowed_project_ids: recent_scope.append(AttendanceData.qr_code_id.in_( db.session.query(QRCode.id).filter(QRCode.project_id.in_(allowed_project_ids)) )) if allowed_location_names: recent_scope.append(AttendanceData.location_name.in_(allowed_location_names)) recent_query = recent_query.filter(and_(*recent_scope) if recent_scope else AttendanceData.id.is_(None)) recent_activity = recent_query.order_by( AttendanceData.check_in_date.desc(), AttendanceData.check_in_time.desc() ).limit(10).all() activity_data = [ { 'employee_id': activity.employee_id, 'location': activity.location_name, 'time': activity.check_in_time.strftime('%H:%M'), 'date': activity.check_in_date.strftime('%Y-%m-%d') } for activity in recent_activity ] return jsonify({ 'success': True, 'recent_activity': activity_data }) except Exception as e: db.session.rollback() logger_handler.log_database_error('dashboard_realtime_api', e) return jsonify({ 'success': False, 'error': 'Failed to fetch real-time data' }), 500 # USER MANAGEMENT ROUTES