578 lines
21 KiB
Python
578 lines
21 KiB
Python
"""
|
|
utils/helpers.py
|
|
================
|
|
Shared utility functions, decorators, QR-code generation helpers,
|
|
and role/permission helpers.
|
|
|
|
Extracted verbatim from app.py (lines 234-329, 910-969, 1274-1467).
|
|
No logic changes — only import paths updated.
|
|
"""
|
|
|
|
import io
|
|
import re
|
|
import os
|
|
import base64
|
|
from datetime import datetime, date, time, timedelta
|
|
from functools import wraps
|
|
|
|
import qrcode
|
|
from flask import session, redirect, flash, request, url_for, jsonify
|
|
from user_agents import parse
|
|
|
|
from extensions import logger_handler
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Role constants
|
|
# ---------------------------------------------------------------------------
|
|
VALID_ROLES = ['admin', 'staff', 'payroll', 'project_manager', 'accounting']
|
|
STAFF_LEVEL_ROLES = ['staff', 'payroll', 'project_manager', 'accounting']
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Employee ID filter helpers — SP / PW / PT aware
|
|
# ---------------------------------------------------------------------------
|
|
# Extra-work records are stored with a work-type code attached to the employee
|
|
# ID ("1234SP", "1234 PW", "PT-1234", ...) in both attendance_data and
|
|
# time_attendance. Filtering with a plain equality test on the numeric ID drops
|
|
# every one of those records. These helpers expand a selected ID into all of its
|
|
# spellings, and normalize IDs so separators/spacing differences still match.
|
|
#
|
|
# Used by: routes/attendance.py, routes/attendance_export.py,
|
|
# routes/time_attendance.py
|
|
|
|
WORK_TYPE_CODES = ('SP', 'PW', 'PT', 'C')
|
|
|
|
def get_base_employee_id(raw_employee_id):
|
|
"""Return the numeric base ID for a possibly work-type-suffixed employee ID."""
|
|
from working_hours_calculator import parse_employee_id_for_work_type
|
|
|
|
base_id, _ = parse_employee_id_for_work_type(str(raw_employee_id or '').strip())
|
|
return base_id
|
|
|
|
|
|
def _work_type_codes_for(raw_employee_id):
|
|
"""
|
|
Return (base_id, codes) for one selected ID.
|
|
|
|
A plain ID ("1234") matches regular AND every work type. An ID that already
|
|
carries a work type ("1234SP") matches only that work type — the user picked
|
|
it deliberately, so don't broaden the result set.
|
|
"""
|
|
from working_hours_calculator import parse_employee_id_for_work_type
|
|
|
|
base_id, work_type = parse_employee_id_for_work_type(str(raw_employee_id or '').strip())
|
|
codes = WORK_TYPE_CODES if work_type == 'regular' else (work_type,)
|
|
return base_id, work_type, codes
|
|
|
|
|
|
def build_employee_id_variants(raw_employee_id):
|
|
"""
|
|
Expand one selected employee ID into the exact ID spellings stored in the
|
|
database — suffix and prefix forms, with and without a space.
|
|
|
|
"1234" -> 1234, 1234SP, "1234 SP", SP1234, "SP 1234", ... (PW / PT / C too)
|
|
"1234SP" -> the four SP spellings only
|
|
"""
|
|
raw = str(raw_employee_id or '').strip()
|
|
if not raw:
|
|
return []
|
|
|
|
base_id, work_type, codes = _work_type_codes_for(raw)
|
|
|
|
variants = [raw]
|
|
if work_type == 'regular':
|
|
variants.append(base_id)
|
|
|
|
for code in codes:
|
|
variants.extend([
|
|
f"{base_id}{code}",
|
|
f"{base_id} {code}",
|
|
f"{code}{base_id}",
|
|
f"{code} {base_id}",
|
|
])
|
|
|
|
return _dedupe(variants)
|
|
|
|
|
|
def build_employee_id_regex(raw_employee_id):
|
|
"""
|
|
Build a MySQL REGEXP pattern matching this employee's ID in ANY spelling
|
|
stored in the database, whatever separator the source file used:
|
|
|
|
1759, "1759 SP", 1759SP, 1759.PW, 1759-PT, SP1759, "SP 1759", "01759 SP"
|
|
|
|
This is what a fixed variant list cannot do — imported IDs come straight from
|
|
customer Excel files, so the separator is unpredictable.
|
|
|
|
Precision is preserved: the numeric part is anchored, so 17590, 11759 and
|
|
1759.0 do NOT match a search for 1759.
|
|
|
|
Returns None when the base ID is not purely numeric. The pattern is built by
|
|
string interpolation, and the ID is user-supplied text, so anything that could
|
|
carry regex metacharacters is refused here — callers fall back to exact matching.
|
|
"""
|
|
raw = str(raw_employee_id or '').strip()
|
|
if not raw:
|
|
return None
|
|
|
|
base_id, work_type, codes = _work_type_codes_for(raw)
|
|
if not base_id.isdigit():
|
|
return None
|
|
|
|
codes_alt = '|'.join(codes)
|
|
# 0* tolerates zero-padded IDs. The separator class excludes letters and digits,
|
|
# so it matches any run of " ", ".", "-", "_" — or nothing at all. That covers
|
|
# everything people actually type: "1759 PW", "1759.PW", "1759. PW", "1759 . PW".
|
|
# It never runs over a word, so "1759 SPX" stays a different ID.
|
|
sep = '[^0-9A-Z]*'
|
|
number = f'0*{base_id}'
|
|
|
|
# Leading and trailing sep runs absorb stray spaces or a trailing dot
|
|
# ("1759 PW ", "1759 PW.", " SP 1759").
|
|
if work_type == 'regular':
|
|
# Regular records AND every work type — code optional on either side
|
|
return f'^{sep}({codes_alt})?{sep}{number}{sep}({codes_alt})?{sep}$'
|
|
|
|
# An explicitly picked work type ("1759SP") must NOT pull in regular records,
|
|
# so the code is required — on one side or the other.
|
|
return f'^{sep}(({codes_alt}){sep}{number}|{number}{sep}({codes_alt})){sep}$'
|
|
|
|
|
|
def expand_employee_id_filter(employee_ids):
|
|
"""
|
|
Expand a list of selected employee IDs into (exact_variants, regex_patterns),
|
|
both de-duplicated and order-preserving.
|
|
|
|
Callers OR the two together: the exact list is index-friendly and covers the
|
|
common spellings, the regex list catches every other separator style.
|
|
"""
|
|
exact, patterns = [], []
|
|
for raw in employee_ids or []:
|
|
exact.extend(build_employee_id_variants(raw))
|
|
pattern = build_employee_id_regex(raw)
|
|
if pattern:
|
|
patterns.append(pattern)
|
|
return _dedupe(exact), _dedupe(patterns)
|
|
|
|
|
|
def employee_id_regex_condition(column, patterns):
|
|
"""
|
|
SQLAlchemy condition: column matches any of the REGEXP patterns, upper-cased so
|
|
the match does not depend on the column's collation. Patterns are bound as
|
|
query parameters, never inlined into the SQL string.
|
|
"""
|
|
from sqlalchemy import func, or_
|
|
|
|
return or_(*[func.upper(column).op('REGEXP')(pattern) for pattern in patterns])
|
|
|
|
|
|
def _dedupe(values):
|
|
"""Order-preserving de-duplication, dropping empties."""
|
|
seen, unique_values = set(), []
|
|
for value in values:
|
|
if value and value not in seen:
|
|
seen.add(value)
|
|
unique_values.append(value)
|
|
return unique_values
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Role helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def is_valid_role(role):
|
|
"""Check if role is valid"""
|
|
return role in VALID_ROLES
|
|
|
|
|
|
def has_admin_privileges(role):
|
|
"""Check if role has admin privileges"""
|
|
return role == 'admin'
|
|
|
|
|
|
def has_staff_level_access(role):
|
|
"""Check if role has staff-level access (includes new roles)"""
|
|
return role in STAFF_LEVEL_ROLES
|
|
|
|
|
|
def get_role_permissions(role):
|
|
"""Get permissions description for a role"""
|
|
permissions = {
|
|
'admin': {
|
|
'title': 'Administrator Permissions',
|
|
'permissions': [
|
|
'Full QR code management (create, edit, delete)',
|
|
'Complete user management capabilities',
|
|
'System configuration access',
|
|
'View all system analytics',
|
|
'Bulk operations and data export',
|
|
'Access to all admin features'
|
|
],
|
|
'restrictions': ['With great power comes great responsibility!']
|
|
},
|
|
'staff': {
|
|
'title': 'Staff User Permissions',
|
|
'permissions': [
|
|
'Create and edit QR codes',
|
|
'View all QR codes in the system',
|
|
'Download QR code images',
|
|
'Update personal profile information',
|
|
],
|
|
'restrictions': [
|
|
'Cannot delete QR codes',
|
|
'Cannot manage other users',
|
|
'Cannot access admin settings'
|
|
]
|
|
},
|
|
'payroll': {
|
|
'title': 'Payroll Specialist Permissions',
|
|
'permissions': [
|
|
'Create and edit QR codes',
|
|
'View all QR codes in the system',
|
|
'Download QR code images',
|
|
'Update personal profile information',
|
|
'Access dashboard and reports',
|
|
'Same permissions as Staff (additional features coming soon)'
|
|
],
|
|
'restrictions': [
|
|
'Cannot delete QR codes',
|
|
'Cannot manage other users',
|
|
'Cannot access admin settings'
|
|
]
|
|
},
|
|
'project_manager': {
|
|
'title': 'Project Manager Permissions',
|
|
'permissions': [
|
|
'Create and edit QR codes',
|
|
'View all QR codes in the system',
|
|
'Download QR code images',
|
|
'Update personal profile information',
|
|
'Access dashboard and reports',
|
|
'Same permissions as Staff (additional features coming soon)'
|
|
],
|
|
'restrictions': [
|
|
'Cannot delete QR codes',
|
|
'Cannot manage other users',
|
|
'Cannot access admin settings'
|
|
]
|
|
},
|
|
'accounting': {
|
|
'title': 'Accounting Specialist Permissions',
|
|
'permissions': [
|
|
'View and modify employee records',
|
|
'Access attendance reports and analytics',
|
|
'View and manage time attendance data',
|
|
'Export payroll and attendance data',
|
|
'Access financial reports and statistics',
|
|
'Update personal profile information',
|
|
'Delete attendance records (same as payroll)'
|
|
],
|
|
'restrictions': [
|
|
'Cannot create or delete QR codes',
|
|
'Cannot manage other users',
|
|
'Cannot access admin settings',
|
|
'Cannot manage projects'
|
|
]
|
|
}
|
|
}
|
|
return permissions.get(role, {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth decorators
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def login_required(f):
|
|
"""Decorator to ensure user is logged in"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if 'user_id' not in session:
|
|
flash('Please log in to access this page.', 'error')
|
|
return redirect(url_for('auth.login'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
|
|
def admin_required(f):
|
|
"""Decorator to ensure user has admin privileges"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if 'username' not in session:
|
|
flash('Please log in to access this page.', 'error')
|
|
return redirect(url_for('auth.login'))
|
|
user_role = session.get('role')
|
|
if not has_admin_privileges(user_role):
|
|
flash('Administrator privileges required for this action.', 'error')
|
|
return redirect(url_for('dashboard.dashboard'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
|
|
def staff_or_admin_required(f):
|
|
"""Decorator to ensure user has staff-level or admin privileges"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if 'username' not in session:
|
|
flash('Please log in to access this page.', 'error')
|
|
return redirect(url_for('auth.login'))
|
|
user_role = session.get('role')
|
|
if not (has_admin_privileges(user_role) or has_staff_level_access(user_role)):
|
|
flash('Insufficient privileges to access this page.', 'error')
|
|
return redirect(url_for('dashboard.dashboard'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
|
|
# Role sets for areas the sidebar only offers to some roles. Enforced on the
|
|
# server so a user cannot reach them by typing the URL (base_authenticated.html
|
|
# is the reference for who sees what).
|
|
PAYROLL_AREA_ROLES = ('admin', 'payroll', 'accounting') # Time Attendance, Legacy, Employees, Statistics
|
|
QR_MANAGEMENT_ROLES = ('admin', 'staff', 'payroll', 'accounting') # create / edit / bulk-import QR codes
|
|
|
|
|
|
def _wants_json_response():
|
|
"""API and fetch() callers get JSON errors instead of a redirect to an HTML page."""
|
|
return (request.path.startswith('/api/')
|
|
or request.headers.get('X-Requested-With') == 'XMLHttpRequest'
|
|
or request.accept_mimetypes.best == 'application/json')
|
|
|
|
|
|
def _role_denied_response(allowed_roles):
|
|
"""None when the logged-in user's role is allowed; otherwise the response to return."""
|
|
if 'user_id' not in session:
|
|
if _wants_json_response():
|
|
return jsonify({'success': False, 'error': 'Authentication required'}), 401
|
|
flash('Please log in to access this page.', 'error')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
role = session.get('role')
|
|
if role in allowed_roles:
|
|
return None
|
|
|
|
logger_handler.logger.warning(
|
|
f"Access denied: user {session.get('username')} (role {role}) -> {request.endpoint}"
|
|
)
|
|
if _wants_json_response():
|
|
return jsonify({'success': False, 'error': 'Access denied'}), 403
|
|
flash('You do not have permission to access that page.', 'error')
|
|
return redirect(url_for('dashboard.dashboard'))
|
|
|
|
|
|
def roles_required(*allowed_roles):
|
|
"""Decorator: the user must be logged in with one of allowed_roles."""
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
denied = _role_denied_response(allowed_roles)
|
|
if denied is not None:
|
|
return denied
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
return decorator
|
|
|
|
|
|
def restrict_blueprint_to_roles(blueprint, allowed_roles):
|
|
"""Apply the role check to every route of a blueprint (one line per module)."""
|
|
@blueprint.before_request
|
|
def _enforce_blueprint_roles():
|
|
return _role_denied_response(allowed_roles)
|
|
return blueprint
|
|
|
|
|
|
def is_admin_user(user_id):
|
|
"""Helper function to safely check if user is admin"""
|
|
from extensions import db
|
|
from models import set_db
|
|
try:
|
|
# User model is available through the app context
|
|
from flask import current_app
|
|
with current_app.app_context():
|
|
# Access via db session to avoid circular import
|
|
from sqlalchemy import text
|
|
result = db.session.execute(
|
|
text("SELECT role, active_status FROM users WHERE id = :uid"),
|
|
{'uid': user_id}
|
|
).fetchone()
|
|
return result and result.active_status and result.role == 'admin'
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Request helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def detect_device_info(user_agent_string):
|
|
"""Extract device information from user agent"""
|
|
try:
|
|
user_agent = parse(user_agent_string)
|
|
device_info = f"{user_agent.device.family}"
|
|
if user_agent.os.family:
|
|
device_info += f" - {user_agent.os.family}"
|
|
if user_agent.os.version_string:
|
|
device_info += f" {user_agent.os.version_string}"
|
|
if user_agent.browser.family:
|
|
device_info += f" ({user_agent.browser.family})"
|
|
return device_info[:200]
|
|
except Exception:
|
|
return "Unknown Device"
|
|
|
|
|
|
def get_client_ip():
|
|
"""Get client IP address"""
|
|
if request.environ.get('HTTP_X_FORWARDED_FOR') is None:
|
|
return request.environ['REMOTE_ADDR']
|
|
else:
|
|
return request.environ['HTTP_X_FORWARDED_FOR']
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# QR code generation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def generate_qr_url(name, qr_id):
|
|
"""Generate a unique URL for QR code destination"""
|
|
clean_name = re.sub(r'[^a-zA-Z0-9\s-]', '', name)
|
|
clean_name = re.sub(r'\s+', '-', clean_name.strip())
|
|
clean_name = clean_name.lower()
|
|
url_slug = f"qr-{qr_id}-{clean_name}"
|
|
return url_slug[:200]
|
|
|
|
|
|
def generate_qr_code(data, fill_color="black", back_color="white", box_size=10, border=4, error_correction='L'):
|
|
"""Generate a QR code image and return as base64 string"""
|
|
error_correction_map = {
|
|
'L': qrcode.constants.ERROR_CORRECT_L,
|
|
'M': qrcode.constants.ERROR_CORRECT_M,
|
|
'Q': qrcode.constants.ERROR_CORRECT_Q,
|
|
'H': qrcode.constants.ERROR_CORRECT_H
|
|
}
|
|
|
|
try:
|
|
qr = qrcode.QRCode(
|
|
version=1,
|
|
error_correction=error_correction_map.get(error_correction, qrcode.constants.ERROR_CORRECT_L),
|
|
box_size=int(box_size),
|
|
border=int(border),
|
|
)
|
|
qr.add_data(data)
|
|
qr.make(fit=True)
|
|
|
|
img = qr.make_image(fill_color=fill_color, back_color=back_color)
|
|
|
|
buffer = io.BytesIO()
|
|
img.save(buffer, format='PNG')
|
|
img_str = base64.b64encode(buffer.getvalue()).decode()
|
|
|
|
try:
|
|
logger_handler.log_qr_code_generated(
|
|
data_length=len(data),
|
|
fill_color=fill_color,
|
|
back_color=back_color,
|
|
box_size=box_size,
|
|
border=border,
|
|
error_correction=error_correction
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return img_str
|
|
|
|
except Exception as e:
|
|
logger_handler.log_database_error('qr_code_generation', e)
|
|
return generate_default_qr_code(data)
|
|
|
|
|
|
def generate_default_qr_code(data):
|
|
"""Fallback function for basic QR code generation"""
|
|
qr = qrcode.QRCode(
|
|
version=1,
|
|
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
|
box_size=10,
|
|
border=4,
|
|
)
|
|
qr.add_data(data)
|
|
qr.make(fit=True)
|
|
|
|
img = qr.make_image(fill_color="black", back_color="white")
|
|
buffer = io.BytesIO()
|
|
img.save(buffer, format='PNG')
|
|
img_str = base64.b64encode(buffer.getvalue()).decode()
|
|
return img_str
|
|
|
|
|
|
def get_qr_styling(qr_code):
|
|
"""Extract QR code styling parameters from database record"""
|
|
return {
|
|
'fill_color': getattr(qr_code, 'fill_color', '#000000') or '#000000',
|
|
'back_color': getattr(qr_code, 'back_color', '#FFFFFF') or '#FFFFFF',
|
|
'box_size': getattr(qr_code, 'box_size', 10) or 10,
|
|
'border': getattr(qr_code, 'border', 4) or 4,
|
|
'error_correction': getattr(qr_code, 'error_correction', 'L') or 'L'
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Check-in history helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_employee_checkin_history(employee_id, qr_code_id, date_filter=None):
|
|
"""Get check-in history for an employee at a specific location"""
|
|
from extensions import db
|
|
try:
|
|
if date_filter is None:
|
|
date_filter = date.today()
|
|
# AttendanceData imported at call site to avoid circular import
|
|
from flask import current_app
|
|
from models.attendance import AttendanceData
|
|
if AttendanceData:
|
|
checkins = AttendanceData.query.filter_by(
|
|
employee_id=employee_id.upper(),
|
|
qr_code_id=qr_code_id,
|
|
check_in_date=date_filter
|
|
).order_by(AttendanceData.check_in_time.asc()).all()
|
|
return checkins
|
|
return []
|
|
except Exception as e:
|
|
print(f"❌ Error retrieving checkin history: {e}")
|
|
return []
|
|
|
|
|
|
def format_checkin_intervals(checkins):
|
|
"""Format time intervals between check-ins for display"""
|
|
if len(checkins) < 2:
|
|
return []
|
|
|
|
intervals = []
|
|
for i in range(1, len(checkins)):
|
|
previous_time = datetime.combine(checkins[i - 1].check_in_date, checkins[i - 1].check_in_time)
|
|
current_time = datetime.combine(checkins[i].check_in_date, checkins[i].check_in_time)
|
|
interval = current_time - previous_time
|
|
interval_minutes = int(interval.total_seconds() / 60)
|
|
intervals.append({
|
|
'from_time': checkins[i - 1].check_in_time.strftime('%H:%M'),
|
|
'to_time': checkins[i].check_in_time.strftime('%H:%M'),
|
|
'interval_minutes': interval_minutes,
|
|
'interval_text': format_time_interval(interval_minutes)
|
|
})
|
|
return intervals
|
|
|
|
|
|
def format_time_interval(minutes):
|
|
"""Format minutes into human-readable time interval"""
|
|
if minutes < 60:
|
|
return f"{minutes} minutes"
|
|
elif minutes < 1440:
|
|
hours = minutes // 60
|
|
remaining_minutes = minutes % 60
|
|
if remaining_minutes == 0:
|
|
return f"{hours} hour{'s' if hours != 1 else ''}"
|
|
else:
|
|
return f"{hours}h {remaining_minutes}m"
|
|
else:
|
|
days = minutes // 1440
|
|
remaining_hours = (minutes % 1440) // 60
|
|
if remaining_hours == 0:
|
|
return f"{days} day{'s' if days != 1 else ''}"
|
|
else:
|
|
return f"{days}d {remaining_hours}h" |