ghostersk
425 lines
19 KiB
Python
425 lines
19 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app
|
|
from flask_login import login_required, current_user
|
|
from auth.models import User, Company, UserCompany
|
|
from api.models import Log
|
|
from extensions import db
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy import or_
|
|
import pytz
|
|
import logging
|
|
|
|
# Initialize logger for this module
|
|
logger = logging.getLogger(__name__)
|
|
|
|
frontend = Blueprint('frontend', __name__)
|
|
|
|
frontend_bp = frontend
|
|
|
|
@frontend.route('/')
|
|
@frontend.route('/home')
|
|
def index():
|
|
try:
|
|
return render_template('frontend/home.html')
|
|
except Exception as e:
|
|
logger.exception("Home page error from IP %s: %s", request.remote_addr, str(e))
|
|
flash('An error occurred while loading the home page.', 'error')
|
|
return render_template('frontend/home.html') # Fallback to basic template
|
|
|
|
@frontend.route('/about')
|
|
def about():
|
|
try:
|
|
return render_template('frontend/about.html')
|
|
except Exception as e:
|
|
logger.exception("About page error from IP %s: %s", request.remote_addr, str(e))
|
|
flash('An error occurred while loading the about page.', 'error')
|
|
return render_template('frontend/about.html') # Fallback to basic template
|
|
|
|
@frontend.route('/profile')
|
|
@login_required
|
|
def profile():
|
|
try:
|
|
user = User.query.filter_by(id=current_user.id).first()
|
|
return render_template('frontend/profile.html', user=user)
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Profile page error for user %s (ID: %s) from IP %s: %s",
|
|
current_user.username if current_user.is_authenticated else 'anonymous',
|
|
current_user.id if current_user.is_authenticated else 'N/A',
|
|
request.remote_addr,
|
|
str(e)
|
|
)
|
|
flash('An error occurred while loading your profile. Please try again.', 'error')
|
|
return redirect(url_for('frontend.index'))
|
|
|
|
@frontend.route('/dashboard')
|
|
@login_required
|
|
def dashboard():
|
|
try:
|
|
# Get app timezone
|
|
app_tz = pytz.timezone(current_app.config['TIMEZONE'])
|
|
|
|
# Get date range from request or default to last 48 hours
|
|
date_range = request.args.get('daterange')
|
|
|
|
if date_range:
|
|
start_str, end_str = date_range.split(' - ')
|
|
start_date = datetime.strptime(start_str, '%Y-%m-%d %H:%M')
|
|
end_date = datetime.strptime(end_str, '%Y-%m-%d %H:%M')
|
|
|
|
# Make these timezone aware using the app's timezone
|
|
start_date = app_tz.localize(start_date)
|
|
end_date = app_tz.localize(end_date)
|
|
else:
|
|
end_date = datetime.now(app_tz) + timedelta(hours=1) # Include 1 hour in the future
|
|
start_date = end_date - timedelta(hours=49) # Look back 48 hours from future end time
|
|
|
|
# Get company filter if provided
|
|
company_id = request.args.get('company_id', type=int)
|
|
|
|
# Build base query with date range filter and eagerly load relationships
|
|
from auth.models import ApiKey, Company
|
|
|
|
# Convert timezone-aware dates to naive for comparison with database timestamps
|
|
# since the log timestamps are stored as naive datetime objects
|
|
start_date_naive = start_date.replace(tzinfo=None)
|
|
end_date_naive = end_date.replace(tzinfo=None)
|
|
|
|
# Use joinedload to eagerly load the API key and company relationships
|
|
query = Log.query.options(
|
|
db.joinedload(Log.api_key),
|
|
db.joinedload(Log.company)
|
|
).filter(Log.timestamp.between(start_date_naive, end_date_naive))
|
|
|
|
# Apply company-specific filtering based on user role
|
|
if current_user.is_global_admin():
|
|
# GlobalAdmin should be allowed to see all records, no matter what company/site
|
|
if company_id:
|
|
query = query.filter(Log.company_id == company_id)
|
|
# If no company_id specified, show all logs (no additional filtering)
|
|
else:
|
|
# CompanyAdmin and User should see only company log events and API Keys (sites)
|
|
# for companies they are member of
|
|
user_company_ids = [uc.company_id for uc in current_user.companies]
|
|
|
|
if not user_company_ids:
|
|
# If user has no company associations, show no logs
|
|
query = query.filter(Log.id == -1) # Impossible condition = no results
|
|
else:
|
|
if company_id and company_id in user_company_ids:
|
|
# Filter by the specific company if requested and user has access
|
|
query = query.filter(Log.company_id == company_id)
|
|
else:
|
|
# Show logs from all companies the user has access to
|
|
query = query.filter(Log.company_id.in_(user_company_ids))
|
|
|
|
# Get the logs ordered by timestamp (newest first)
|
|
logs = query.order_by(Log.timestamp.desc()).all()
|
|
|
|
# Get companies for the dropdown filter based on user role
|
|
if current_user.is_global_admin():
|
|
# GlobalAdmin can see all companies
|
|
companies = Company.query.all()
|
|
else:
|
|
# CompanyAdmin and User should see only companies they are member of
|
|
companies = current_user.get_companies()
|
|
|
|
return render_template('frontend/dashboard.html',
|
|
title='Dashboard',
|
|
logs=logs,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
companies=companies,
|
|
selected_company_id=company_id)
|
|
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Dashboard error for user %s (ID: %s, role: %s) from IP %s: %s",
|
|
current_user.username if current_user.is_authenticated else 'anonymous',
|
|
current_user.id if current_user.is_authenticated else 'N/A',
|
|
current_user.role if current_user.is_authenticated else 'N/A',
|
|
request.remote_addr,
|
|
str(e)
|
|
)
|
|
flash('An error occurred while loading the dashboard. Please try again.', 'error')
|
|
return redirect(url_for('frontend.index'))
|
|
|
|
@frontend.route('/time_spent_report')
|
|
@login_required
|
|
def time_spent_report():
|
|
try:
|
|
# Get app timezone
|
|
app_tz = pytz.timezone(current_app.config['TIMEZONE'])
|
|
|
|
# Get date range from request or default to last 7 days
|
|
date_range = request.args.get('daterange')
|
|
|
|
if date_range:
|
|
start_str, end_str = date_range.split(' - ')
|
|
start_date = datetime.strptime(start_str, '%Y-%m-%d %H:%M')
|
|
end_date = datetime.strptime(end_str, '%Y-%m-%d %H:%M')
|
|
|
|
# Make these timezone aware using the app's timezone
|
|
start_date = app_tz.localize(start_date)
|
|
end_date = app_tz.localize(end_date)
|
|
else:
|
|
end_date = datetime.now(app_tz) + timedelta(hours=1) # Include 1 hour in the future
|
|
start_date = end_date - timedelta(days=7, hours=1) # Look back 7 days from future end time
|
|
|
|
# Get filters
|
|
company_id = request.args.get('company_id', type=int)
|
|
api_key_id = request.args.get('api_key_id', type=int)
|
|
group_by = request.args.get('group_by', 'user') # Default to user grouping
|
|
continue_iterate = request.args.get('continue_iterate') == 'on' # Get checkbox value
|
|
|
|
# Build base query with date range filter
|
|
from auth.models import ApiKey, Company
|
|
|
|
# Start with all the login/logout events within the date range
|
|
# Convert timezone-aware dates to naive for comparison with database timestamps
|
|
start_date_naive = start_date.replace(tzinfo=None)
|
|
end_date_naive = end_date.replace(tzinfo=None)
|
|
logs_query = Log.query.filter(
|
|
Log.timestamp.between(start_date_naive, end_date_naive)
|
|
).order_by(Log.timestamp.asc())
|
|
|
|
# Apply company-specific filtering based on user role
|
|
if current_user.is_global_admin():
|
|
# GlobalAdmin should be allowed to see all records, no matter what company/site
|
|
if company_id:
|
|
logs_query = logs_query.filter(Log.company_id == company_id)
|
|
# If no company_id specified, show all logs (no additional filtering)
|
|
else:
|
|
# CompanyAdmin and User should see only company log events for companies they are member of
|
|
user_company_ids = [uc.company_id for uc in current_user.companies]
|
|
|
|
if not user_company_ids:
|
|
# If user has no company associations, show no logs
|
|
logs_query = logs_query.filter(Log.id == -1) # Impossible condition = no results
|
|
else:
|
|
if company_id and company_id in user_company_ids:
|
|
# Filter by the specific company if requested and user has access
|
|
logs_query = logs_query.filter(Log.company_id == company_id)
|
|
else:
|
|
# Show logs from all companies the user has access to
|
|
logs_query = logs_query.filter(Log.company_id.in_(user_company_ids))
|
|
|
|
# Apply API key filter if provided
|
|
if api_key_id:
|
|
# Ensure the API key belongs to a company the user has access to
|
|
api_key = ApiKey.query.get(api_key_id)
|
|
if api_key:
|
|
if current_user.is_global_admin():
|
|
# GlobalAdmin can use any API key
|
|
logs_query = logs_query.filter(Log.api_key_id == api_key_id)
|
|
else:
|
|
# Check if the API key belongs to a company the user has access to
|
|
user_company_ids = [uc.company_id for uc in current_user.companies]
|
|
if api_key.company_id in user_company_ids:
|
|
logs_query = logs_query.filter(Log.api_key_id == api_key_id)
|
|
# If API key doesn't belong to user's company, ignore the filter
|
|
|
|
# Get all the relevant logs
|
|
logs = logs_query.all()
|
|
|
|
# Process logs to calculate time spent
|
|
time_data = calculate_time_spent(logs, group_by, continue_iterate)
|
|
|
|
# Get all companies for the dropdown filter
|
|
if current_user.is_global_admin():
|
|
companies = Company.query.all()
|
|
else:
|
|
companies = current_user.get_companies()
|
|
|
|
# Get available API keys for filter
|
|
if current_user.is_global_admin():
|
|
api_keys = ApiKey.query.all()
|
|
else:
|
|
# Get API keys for companies user has access to
|
|
user_company_ids = [uc.company_id for uc in current_user.companies]
|
|
api_keys = ApiKey.query.filter(ApiKey.company_id.in_(user_company_ids)).all()
|
|
|
|
return render_template('frontend/time_spent_report.html',
|
|
title='Time Spent Report',
|
|
time_data=time_data,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
companies=companies,
|
|
api_keys=api_keys,
|
|
selected_company_id=company_id,
|
|
selected_api_key_id=api_key_id,
|
|
group_by=group_by,
|
|
continue_iterate=continue_iterate)
|
|
|
|
except Exception as e:
|
|
logger.exception(
|
|
"Time spent report error for user %s (ID: %s, role: %s) from IP %s: %s",
|
|
current_user.username if current_user.is_authenticated else 'anonymous',
|
|
current_user.id if current_user.is_authenticated else 'N/A',
|
|
current_user.role if current_user.is_authenticated else 'N/A',
|
|
request.remote_addr,
|
|
str(e)
|
|
)
|
|
flash('An error occurred while generating the time spent report. Please try again.', 'error')
|
|
return redirect(url_for('frontend.dashboard'))
|
|
|
|
def calculate_time_spent(logs, group_by='user', continue_iterate=False):
|
|
"""
|
|
Calculate time spent by users based on login/logout events.
|
|
|
|
Args:
|
|
logs: List of Log objects sorted by timestamp
|
|
group_by: Whether to group by 'user' or 'user_computer'
|
|
continue_iterate: Whether to continue iterating for additional login/logout pairs
|
|
|
|
Returns:
|
|
List of dictionaries with time spent information
|
|
"""
|
|
from auth.models import Company, ApiKey
|
|
|
|
# Dictionary to track user sessions
|
|
# Key: user_name or user_name+computer_name depending on group_by
|
|
# Value: dictionary with session tracking info
|
|
active_sessions = {}
|
|
|
|
# Dictionary to accumulate total time spent
|
|
# Key: date + user_name + (computer_name) + company_id
|
|
# Value: dictionary with accumulated time and session details
|
|
time_totals = {}
|
|
|
|
# Define login and logout event types - case-insensitive matching
|
|
login_events = ['login', 'unlock', 'logon'] # Events that start a session
|
|
logout_events = ['logout', 'lock', 'logoff'] # Events that end a session
|
|
|
|
# Create a set to track which users have appeared in logs
|
|
seen_users = set()
|
|
|
|
# First pass: populate the time_totals dictionary with user entries
|
|
# This ensures every user has an entry even if they don't have paired login/logout events
|
|
for log in logs:
|
|
# Determine the session key based on grouping option
|
|
session_key = log.user_name
|
|
if group_by == 'user_computer':
|
|
session_key = f"{log.user_name}:{log.computer_name}"
|
|
|
|
# Get date string in format YYYY-MM-DD
|
|
log_date = log.timestamp.strftime('%Y-%m-%d')
|
|
|
|
# Create a unique key for the time totals
|
|
total_key = f"{log_date}:{session_key}:{log.company_id}"
|
|
|
|
# Initialize the time total entry if it doesn't exist
|
|
if total_key not in time_totals:
|
|
# Get company info
|
|
company = Company.query.get(log.company_id) if log.company_id else None
|
|
company_name = company.name if company else "Unknown"
|
|
|
|
# Get API key info
|
|
api_key = ApiKey.query.get(log.api_key_id) if log.api_key_id else None
|
|
api_key_description = api_key.description if api_key else "Unknown"
|
|
|
|
# Initialize with zero time
|
|
time_totals[total_key] = {
|
|
'date': log_date,
|
|
'user_name': log.user_name,
|
|
'computer_name': log.computer_name if group_by == 'user_computer' else None,
|
|
'company_id': log.company_id,
|
|
'company_name': company_name,
|
|
'api_key_id': log.api_key_id,
|
|
'api_key_description': api_key_description,
|
|
'total_seconds': 0,
|
|
'first_login': None,
|
|
'last_logout': None,
|
|
'session_count': 0
|
|
}
|
|
|
|
# Track that we've seen this user
|
|
seen_users.add(log.user_name)
|
|
|
|
# Update login/logout timestamps even if we can't calculate duration
|
|
event_type = log.event_type.lower() if log.event_type else ''
|
|
|
|
# For all users, record their first login and last logout
|
|
if event_type in login_events:
|
|
if not time_totals[total_key]['first_login'] or log.timestamp < time_totals[total_key]['first_login']:
|
|
time_totals[total_key]['first_login'] = log.timestamp
|
|
|
|
if event_type in logout_events:
|
|
if not time_totals[total_key]['last_logout'] or log.timestamp > time_totals[total_key]['last_logout']:
|
|
time_totals[total_key]['last_logout'] = log.timestamp
|
|
|
|
# Second pass: calculate session durations
|
|
for log_index, log in enumerate(logs):
|
|
# Determine the session key based on grouping option
|
|
session_key = log.user_name
|
|
if group_by == 'user_computer':
|
|
session_key = f"{log.user_name}:{log.computer_name}"
|
|
|
|
# Get date string in format YYYY-MM-DD
|
|
log_date = log.timestamp.strftime('%Y-%m-%d')
|
|
|
|
# Create a unique key for the time totals
|
|
total_key = f"{log_date}:{session_key}:{log.company_id}"
|
|
|
|
# Convert event_type to lowercase for case-insensitive comparison
|
|
event_type = log.event_type.lower() if log.event_type else ''
|
|
|
|
# Process any login-like event
|
|
if event_type in login_events:
|
|
# Record login time for this session
|
|
active_sessions[session_key] = {
|
|
'login_time': log.timestamp,
|
|
'log_index': log_index,
|
|
'company_id': log.company_id,
|
|
'date': log_date
|
|
}
|
|
|
|
# Process any logout-like event
|
|
elif event_type in logout_events and session_key in active_sessions:
|
|
session = active_sessions[session_key]
|
|
|
|
# Only process if session is from the same day and company
|
|
if session['date'] == log_date and session['company_id'] == log.company_id:
|
|
# Calculate duration of this session
|
|
duration = (log.timestamp - session['login_time']).total_seconds()
|
|
|
|
# Only count if duration is positive and reasonable (< 24 hours)
|
|
if 0 < duration < 86400: # 24 hours = 86400 seconds
|
|
# Add to the total time for this user/day combination
|
|
time_totals[total_key]['total_seconds'] += duration
|
|
time_totals[total_key]['session_count'] += 1
|
|
|
|
# If we should continue iterating, leave the session active to match with future login events
|
|
# Otherwise, remove the active session after processing
|
|
if not continue_iterate:
|
|
del active_sessions[session_key]
|
|
else:
|
|
# If the session doesn't match day/company, remove it if we're not continuing to iterate
|
|
if not continue_iterate:
|
|
del active_sessions[session_key]
|
|
|
|
# Special case: If we only have one event of each type per user per day,
|
|
# calculate duration between first login and last logout
|
|
for total_key, entry in time_totals.items():
|
|
if entry['session_count'] == 0 and entry['first_login'] and entry['last_logout']:
|
|
# Calculate duration between first login and last logout
|
|
duration = (entry['last_logout'] - entry['first_login']).total_seconds()
|
|
|
|
# Only use if duration is positive and reasonable
|
|
if 0 < duration < 86400: # 24 hours = 86400 seconds
|
|
entry['total_seconds'] += duration
|
|
entry['session_count'] = 1
|
|
|
|
# Convert the time_totals dictionary to a list of dictionaries
|
|
result = []
|
|
for entry in time_totals.values():
|
|
# Format the total time as hours:minutes:seconds
|
|
hours, remainder = divmod(entry['total_seconds'], 3600)
|
|
minutes, seconds = divmod(remainder, 60)
|
|
entry['formatted_time'] = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}"
|
|
|
|
result.append(entry)
|
|
|
|
# Sort by date (newest first) and then by user_name
|
|
result.sort(key=lambda x: (x['date'], x['user_name']), reverse=True)
|
|
|
|
return result |