Merge branch 'master' into feature-markdown-notes

This commit is contained in:
2025-07-07 21:26:44 +02:00
122 changed files with 23364 additions and 8651 deletions

189
routes/announcements.py Normal file
View File

@@ -0,0 +1,189 @@
"""
Announcement management routes
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g
from models import db, Announcement, Company, User, Role
from routes.auth import system_admin_required
from datetime import datetime
import json
import logging
logger = logging.getLogger(__name__)
announcements_bp = Blueprint('announcements', __name__, url_prefix='/system-admin/announcements')
@announcements_bp.route('')
@system_admin_required
def index():
"""System Admin: Manage announcements"""
page = request.args.get('page', 1, type=int)
per_page = 20
announcements = Announcement.query.order_by(Announcement.created_at.desc()).paginate(
page=page, per_page=per_page, error_out=False)
return render_template('system_admin_announcements.html',
title='System Admin - Announcements',
announcements=announcements)
@announcements_bp.route('/new', methods=['GET', 'POST'])
@system_admin_required
def create():
"""System Admin: Create new announcement"""
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
announcement_type = request.form.get('announcement_type', 'info')
is_urgent = request.form.get('is_urgent') == 'on'
is_active = request.form.get('is_active') == 'on'
# Handle date fields
start_date = request.form.get('start_date')
end_date = request.form.get('end_date')
start_datetime = None
end_datetime = None
if start_date:
try:
start_datetime = datetime.strptime(start_date, '%Y-%m-%dT%H:%M')
except ValueError:
pass
if end_date:
try:
end_datetime = datetime.strptime(end_date, '%Y-%m-%dT%H:%M')
except ValueError:
pass
# Handle targeting
target_all_users = request.form.get('target_all_users') == 'on'
target_roles = None
target_companies = None
if not target_all_users:
selected_roles = request.form.getlist('target_roles')
selected_companies = request.form.getlist('target_companies')
if selected_roles:
target_roles = json.dumps(selected_roles)
if selected_companies:
target_companies = json.dumps([int(c) for c in selected_companies])
announcement = Announcement(
title=title,
content=content,
announcement_type=announcement_type,
is_urgent=is_urgent,
is_active=is_active,
start_date=start_datetime,
end_date=end_datetime,
target_all_users=target_all_users,
target_roles=target_roles,
target_companies=target_companies,
created_by_id=g.user.id
)
db.session.add(announcement)
db.session.commit()
flash('Announcement created successfully.', 'success')
return redirect(url_for('announcements.index'))
# Get roles and companies for targeting options
roles = [role.value for role in Role]
companies = Company.query.order_by(Company.name).all()
return render_template('system_admin_announcement_form.html',
title='Create Announcement',
announcement=None,
roles=roles,
companies=companies)
@announcements_bp.route('/<int:id>/edit', methods=['GET', 'POST'])
@system_admin_required
def edit(id):
"""System Admin: Edit announcement"""
announcement = Announcement.query.get_or_404(id)
if request.method == 'POST':
announcement.title = request.form.get('title')
announcement.content = request.form.get('content')
announcement.announcement_type = request.form.get('announcement_type', 'info')
announcement.is_urgent = request.form.get('is_urgent') == 'on'
announcement.is_active = request.form.get('is_active') == 'on'
# Handle date fields
start_date = request.form.get('start_date')
end_date = request.form.get('end_date')
if start_date:
try:
announcement.start_date = datetime.strptime(start_date, '%Y-%m-%dT%H:%M')
except ValueError:
announcement.start_date = None
else:
announcement.start_date = None
if end_date:
try:
announcement.end_date = datetime.strptime(end_date, '%Y-%m-%dT%H:%M')
except ValueError:
announcement.end_date = None
else:
announcement.end_date = None
# Handle targeting
announcement.target_all_users = request.form.get('target_all_users') == 'on'
if not announcement.target_all_users:
selected_roles = request.form.getlist('target_roles')
selected_companies = request.form.getlist('target_companies')
if selected_roles:
announcement.target_roles = json.dumps(selected_roles)
else:
announcement.target_roles = None
if selected_companies:
announcement.target_companies = json.dumps([int(c) for c in selected_companies])
else:
announcement.target_companies = None
else:
announcement.target_roles = None
announcement.target_companies = None
announcement.updated_at = datetime.now()
db.session.commit()
flash('Announcement updated successfully.', 'success')
return redirect(url_for('announcements.index'))
# Get roles and companies for targeting options
roles = [role.value for role in Role]
companies = Company.query.order_by(Company.name).all()
return render_template('system_admin_announcement_form.html',
title='Edit Announcement',
announcement=announcement,
roles=roles,
companies=companies)
@announcements_bp.route('/<int:id>/delete', methods=['POST'])
@system_admin_required
def delete(id):
"""System Admin: Delete announcement"""
announcement = Announcement.query.get_or_404(id)
db.session.delete(announcement)
db.session.commit()
flash('Announcement deleted successfully.', 'success')
return redirect(url_for('announcements.index'))

View File

@@ -1,14 +1,13 @@
# Standard library imports
"""
Authentication decorators for route protection
"""
from functools import wraps
# Third-party imports
from flask import flash, g, redirect, request, url_for
# Local application imports
from models import Company, Role, User
from flask import g, redirect, url_for, flash, request
from models import Role, Company
def login_required(f):
"""Decorator to require login for routes"""
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user is None:
@@ -17,60 +16,85 @@ def login_required(f):
return decorated_function
def company_required(f):
"""
Decorator to ensure user has a valid company association and set company context.
"""
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user is None:
return redirect(url_for('login', next=request.url))
# System admins can access without company association
if g.user.role == Role.SYSTEM_ADMIN:
return f(*args, **kwargs)
if g.user.company_id is None:
flash('You must be associated with a company to access this page.', 'error')
return redirect(url_for('setup_company'))
# Set company context
g.company = Company.query.get(g.user.company_id)
if not g.company or not g.company.is_active:
flash('Your company account is inactive.', 'error')
return redirect(url_for('home'))
return f(*args, **kwargs)
return decorated_function
def role_required(*allowed_roles):
def role_required(min_role):
"""Decorator to require a minimum role for routes"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user.role not in allowed_roles:
flash('You do not have permission to access this page.', 'error')
return redirect(url_for('dashboard'))
if g.user is None:
return redirect(url_for('login', next=request.url))
# Admin and System Admin always have access
if g.user.role == Role.ADMIN or g.user.role == Role.SYSTEM_ADMIN:
return f(*args, **kwargs)
# Define role hierarchy
role_hierarchy = {
Role.TEAM_MEMBER: 1,
Role.TEAM_LEADER: 2,
Role.SUPERVISOR: 3,
Role.ADMIN: 4,
Role.SYSTEM_ADMIN: 5
}
user_role_value = role_hierarchy.get(g.user.role, 0)
min_role_value = role_hierarchy.get(min_role, 0)
if user_role_value < min_role_value:
flash('You do not have sufficient permissions to access this page.', 'error')
return redirect(url_for('home'))
return f(*args, **kwargs)
return decorated_function
return decorator
def admin_required(f):
def company_required(f):
"""Decorator to ensure user has a valid company association and set company context"""
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user is None:
return redirect(url_for('login', next=request.url))
# System admins can access without company association
if g.user.role == Role.SYSTEM_ADMIN:
return f(*args, **kwargs)
if g.user.company_id is None:
flash('You must be associated with a company to access this page.', 'error')
return redirect(url_for('setup_company'))
# Set company context
g.company = Company.query.get(g.user.company_id)
if not g.company or not g.company.is_active:
flash('Your company is not active. Please contact support.', 'error')
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
def admin_required(f):
"""Decorator to require admin role for routes"""
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user is None:
return redirect(url_for('login'))
if g.user.role not in [Role.ADMIN, Role.SYSTEM_ADMIN]:
flash('Admin access required.', 'error')
return redirect(url_for('dashboard'))
flash('You must be an administrator to access this page.', 'error')
return redirect(url_for('home'))
return f(*args, **kwargs)
return decorated_function
def system_admin_required(f):
"""Decorator to require system admin role for routes"""
@wraps(f)
def decorated_function(*args, **kwargs):
if g.user is None:
return redirect(url_for('login'))
if g.user.role != Role.SYSTEM_ADMIN:
flash('System admin access required.', 'error')
return redirect(url_for('dashboard'))
flash('You must be a system administrator to access this page.', 'error')
return redirect(url_for('home'))
return f(*args, **kwargs)
return decorated_function

360
routes/company.py Normal file
View File

@@ -0,0 +1,360 @@
"""
Company management routes
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, session
from models import db, Company, User, Role, Team, Project, SystemSettings, CompanyWorkConfig, WorkRegion
from routes.auth import admin_required, company_required, login_required
import logging
import re
logger = logging.getLogger(__name__)
companies_bp = Blueprint('companies', __name__, url_prefix='/admin/company')
@companies_bp.route('', methods=['GET', 'POST'])
@admin_required
@company_required
def admin_company():
"""View and manage company settings"""
company = g.company
# Handle form submissions
if request.method == 'POST':
action = request.form.get('action')
if action == 'update_company_details':
# Handle company details update
name = request.form.get('name')
description = request.form.get('description', '')
max_users = request.form.get('max_users')
is_active = 'is_active' in request.form
# Validate input
error = None
if not name:
error = 'Company name is required'
elif name != company.name and Company.query.filter_by(name=name).first():
error = 'Company name already exists'
if max_users:
try:
max_users = int(max_users)
if max_users < 1:
error = 'Maximum users must be at least 1'
except ValueError:
error = 'Maximum users must be a valid number'
else:
max_users = None
if error is None:
company.name = name
company.description = description
company.max_users = max_users
company.is_active = is_active
db.session.commit()
flash('Company details updated successfully!', 'success')
else:
flash(error, 'error')
return redirect(url_for('companies.admin_company'))
elif action == 'update_system_settings':
# Update registration setting
registration_enabled = 'registration_enabled' in request.form
reg_setting = SystemSettings.query.filter_by(key='registration_enabled').first()
if reg_setting:
reg_setting.value = 'true' if registration_enabled else 'false'
# Update email verification setting
email_verification_required = 'email_verification_required' in request.form
email_setting = SystemSettings.query.filter_by(key='email_verification_required').first()
if email_setting:
email_setting.value = 'true' if email_verification_required else 'false'
db.session.commit()
flash('System settings updated successfully!', 'success')
return redirect(url_for('companies.admin_company'))
elif action == 'update_work_policies':
# Get or create company work config
work_config = CompanyWorkConfig.query.filter_by(company_id=g.user.company_id).first()
if not work_config:
# Create default config for the company
preset = CompanyWorkConfig.get_regional_preset(WorkRegion.GERMANY)
work_config = CompanyWorkConfig(
company_id=g.user.company_id,
standard_hours_per_day=preset['standard_hours_per_day'],
standard_hours_per_week=preset['standard_hours_per_week'],
work_region=WorkRegion.GERMANY,
overtime_enabled=preset['overtime_enabled'],
overtime_rate=preset['overtime_rate'],
double_time_enabled=preset['double_time_enabled'],
double_time_threshold=preset['double_time_threshold'],
double_time_rate=preset['double_time_rate'],
require_breaks=preset['require_breaks'],
break_duration_minutes=preset['break_duration_minutes'],
break_after_hours=preset['break_after_hours'],
weekly_overtime_threshold=preset['weekly_overtime_threshold'],
weekly_overtime_rate=preset['weekly_overtime_rate']
)
db.session.add(work_config)
db.session.flush()
try:
# Handle regional preset selection
if request.form.get('apply_preset'):
region_code = request.form.get('region_preset')
if region_code:
region = WorkRegion(region_code)
preset = CompanyWorkConfig.get_regional_preset(region)
work_config.standard_hours_per_day = preset['standard_hours_per_day']
work_config.standard_hours_per_week = preset['standard_hours_per_week']
work_config.work_region = region
work_config.overtime_enabled = preset['overtime_enabled']
work_config.overtime_rate = preset['overtime_rate']
work_config.double_time_enabled = preset['double_time_enabled']
work_config.double_time_threshold = preset['double_time_threshold']
work_config.double_time_rate = preset['double_time_rate']
work_config.require_breaks = preset['require_breaks']
work_config.break_duration_minutes = preset['break_duration_minutes']
work_config.break_after_hours = preset['break_after_hours']
work_config.weekly_overtime_threshold = preset['weekly_overtime_threshold']
work_config.weekly_overtime_rate = preset['weekly_overtime_rate']
db.session.commit()
flash(f'Applied {preset["region_name"]} work policy preset', 'success')
else:
# Handle manual configuration update
work_config.standard_hours_per_day = float(request.form.get('standard_hours_per_day', 8.0))
work_config.standard_hours_per_week = float(request.form.get('standard_hours_per_week', 40.0))
work_config.overtime_enabled = request.form.get('overtime_enabled') == 'on'
work_config.overtime_rate = float(request.form.get('overtime_rate', 1.5))
work_config.double_time_enabled = request.form.get('double_time_enabled') == 'on'
work_config.double_time_threshold = float(request.form.get('double_time_threshold', 12.0))
work_config.double_time_rate = float(request.form.get('double_time_rate', 2.0))
work_config.require_breaks = request.form.get('require_breaks') == 'on'
work_config.break_duration_minutes = int(request.form.get('break_duration_minutes', 30))
work_config.break_after_hours = float(request.form.get('break_after_hours', 6.0))
work_config.weekly_overtime_threshold = float(request.form.get('weekly_overtime_threshold', 40.0))
work_config.weekly_overtime_rate = float(request.form.get('weekly_overtime_rate', 1.5))
work_config.work_region = WorkRegion.OTHER
# region_name removed - using work_region enum value instead
db.session.commit()
flash('Work policies updated successfully!', 'success')
except ValueError:
flash('Please enter valid numbers for all fields', 'error')
return redirect(url_for('companies.admin_company'))
# Get company statistics
stats = {
'total_users': User.query.filter_by(company_id=company.id).count(),
'total_teams': Team.query.filter_by(company_id=company.id).count(),
'total_projects': Project.query.filter_by(company_id=company.id).count(),
'active_projects': Project.query.filter_by(company_id=company.id, is_active=True).count(),
}
# Get current system settings
settings = {}
for setting in SystemSettings.query.all():
if setting.key == 'registration_enabled':
settings['registration_enabled'] = setting.value == 'true'
elif setting.key == 'email_verification_required':
settings['email_verification_required'] = setting.value == 'true'
# Get or create company work config
work_config = CompanyWorkConfig.query.filter_by(company_id=g.user.company_id).first()
if not work_config:
# Create default config for the company
preset = CompanyWorkConfig.get_regional_preset(WorkRegion.GERMANY)
work_config = CompanyWorkConfig(
company_id=g.user.company_id,
standard_hours_per_day=preset['standard_hours_per_day'],
standard_hours_per_week=preset['standard_hours_per_week'],
work_region=WorkRegion.GERMANY,
overtime_enabled=preset['overtime_enabled'],
overtime_rate=preset['overtime_rate'],
double_time_enabled=preset['double_time_enabled'],
double_time_threshold=preset['double_time_threshold'],
double_time_rate=preset['double_time_rate'],
require_breaks=preset['require_breaks'],
break_duration_minutes=preset['break_duration_minutes'],
break_after_hours=preset['break_after_hours'],
weekly_overtime_threshold=preset['weekly_overtime_threshold'],
weekly_overtime_rate=preset['weekly_overtime_rate']
)
db.session.add(work_config)
db.session.commit()
# Get available regional presets
regional_presets = []
for region in WorkRegion:
preset = CompanyWorkConfig.get_regional_preset(region)
regional_presets.append({
'code': region.value,
'name': preset['region_name'],
'description': f"{preset['standard_hours_per_day']}h/day, {preset['break_duration_minutes']}min break after {preset['break_after_hours']}h"
})
return render_template('admin_company.html',
title='Company Management',
company=company,
stats=stats,
settings=settings,
work_config=work_config,
regional_presets=regional_presets,
WorkRegion=WorkRegion)
@companies_bp.route('/users')
@admin_required
@company_required
def company_users():
"""List all users in the company with detailed information"""
users = User.query.filter_by(company_id=g.company.id).order_by(User.created_at.desc()).all()
# Calculate user statistics
user_stats = {
'total': len(users),
'verified': len([u for u in users if u.is_verified]),
'unverified': len([u for u in users if not u.is_verified]),
'blocked': len([u for u in users if u.is_blocked]),
'active': len([u for u in users if not u.is_blocked and u.is_verified]),
'admins': len([u for u in users if u.role == Role.ADMIN]),
'supervisors': len([u for u in users if u.role == Role.SUPERVISOR]),
'team_leaders': len([u for u in users if u.role == Role.TEAM_LEADER]),
'team_members': len([u for u in users if u.role == Role.TEAM_MEMBER]),
}
return render_template('company_users.html',
title='Company Users',
company=g.company,
users=users,
stats=user_stats)
# Setup company route (separate from company blueprint due to different URL)
def setup_company():
"""Company setup route for creating new companies with admin users"""
existing_companies = Company.query.count()
# Determine access level
is_initial_setup = existing_companies == 0
is_system_admin = g.user and g.user.role == Role.SYSTEM_ADMIN
is_authorized = is_initial_setup or is_system_admin
# Check authorization for non-initial setups
if not is_initial_setup and not is_system_admin:
flash('You do not have permission to create new companies.', 'error')
return redirect(url_for('home') if g.user else url_for('login'))
if request.method == 'POST':
company_name = request.form.get('company_name')
company_description = request.form.get('company_description', '')
admin_username = request.form.get('admin_username')
admin_email = request.form.get('admin_email')
admin_password = request.form.get('admin_password')
confirm_password = request.form.get('confirm_password')
# Validate input
error = None
if not company_name:
error = 'Company name is required'
elif not admin_username:
error = 'Admin username is required'
elif not admin_email:
error = 'Admin email is required'
elif not admin_password:
error = 'Admin password is required'
elif admin_password != confirm_password:
error = 'Passwords do not match'
elif len(admin_password) < 6:
error = 'Password must be at least 6 characters long'
if error is None:
try:
# Generate company slug
slug = re.sub(r'[^\w\s-]', '', company_name.lower())
slug = re.sub(r'[-\s]+', '-', slug).strip('-')
# Ensure slug uniqueness
base_slug = slug
counter = 1
while Company.query.filter_by(slug=slug).first():
slug = f"{base_slug}-{counter}"
counter += 1
# Create company
company = Company(
name=company_name,
slug=slug,
description=company_description,
is_active=True
)
db.session.add(company)
db.session.flush() # Get company.id without committing
# Check if username/email already exists in this company context
existing_user_by_username = User.query.filter_by(
username=admin_username,
company_id=company.id
).first()
existing_user_by_email = User.query.filter_by(
email=admin_email,
company_id=company.id
).first()
if existing_user_by_username:
error = 'Username already exists in this company'
elif existing_user_by_email:
error = 'Email already registered in this company'
if error is None:
# Create admin user
admin_user = User(
username=admin_username,
email=admin_email,
company_id=company.id,
role=Role.ADMIN,
is_verified=True # Auto-verify company admin
)
admin_user.set_password(admin_password)
db.session.add(admin_user)
db.session.commit()
if is_initial_setup:
# Auto-login the admin user for initial setup
session['user_id'] = admin_user.id
session['username'] = admin_user.username
session['role'] = admin_user.role.value
flash(f'Company "{company_name}" created successfully! You are now logged in as the administrator.', 'success')
return redirect(url_for('home'))
else:
# For super admin creating additional companies, don't auto-login
flash(f'Company "{company_name}" created successfully! Admin user "{admin_username}" has been created with the company code "{slug}".', 'success')
return redirect(url_for('companies.admin_company') if g.user else url_for('login'))
else:
db.session.rollback()
except Exception as e:
db.session.rollback()
logger.error(f"Error during company setup: {str(e)}")
error = f"An error occurred during setup: {str(e)}"
if error:
flash(error, 'error')
return render_template('setup_company.html',
title='Company Setup',
existing_companies=existing_companies,
is_initial_setup=is_initial_setup,
is_super_admin=is_system_admin)

102
routes/company_api.py Normal file
View File

@@ -0,0 +1,102 @@
"""
Company API endpoints
"""
from flask import Blueprint, jsonify, g
from models import db, Company, User, Role, Team, Project, TimeEntry
from routes.auth import system_admin_required
from datetime import datetime, timedelta
company_api_bp = Blueprint('company_api', __name__, url_prefix='/api')
@company_api_bp.route('/system-admin/companies/<int:company_id>/users')
@system_admin_required
def api_company_users(company_id):
"""API: Get users for a specific company (System Admin only)"""
company = Company.query.get_or_404(company_id)
users = User.query.filter_by(company_id=company.id).order_by(User.username).all()
return jsonify({
'company': {
'id': company.id,
'name': company.name,
'is_personal': company.is_personal
},
'users': [{
'id': user.id,
'username': user.username,
'email': user.email,
'role': user.role.value,
'is_blocked': user.is_blocked,
'is_verified': user.is_verified,
'created_at': user.created_at.isoformat(),
'team_id': user.team_id
} for user in users]
})
@company_api_bp.route('/system-admin/companies/<int:company_id>/stats')
@system_admin_required
def api_company_stats(company_id):
"""API: Get detailed statistics for a specific company"""
company = Company.query.get_or_404(company_id)
# User counts by role
role_counts = {}
for role in Role:
count = User.query.filter_by(company_id=company.id, role=role).count()
if count > 0:
role_counts[role.value] = count
# Team and project counts
team_count = Team.query.filter_by(company_id=company.id).count()
project_count = Project.query.filter_by(company_id=company.id).count()
active_projects = Project.query.filter_by(company_id=company.id, is_active=True).count()
# Time entries statistics
week_ago = datetime.now() - timedelta(days=7)
month_ago = datetime.now() - timedelta(days=30)
weekly_entries = TimeEntry.query.join(User).filter(
User.company_id == company.id,
TimeEntry.arrival_time >= week_ago
).count()
monthly_entries = TimeEntry.query.join(User).filter(
User.company_id == company.id,
TimeEntry.arrival_time >= month_ago
).count()
# Active sessions
active_sessions = TimeEntry.query.join(User).filter(
User.company_id == company.id,
TimeEntry.departure_time == None,
TimeEntry.is_paused == False
).count()
return jsonify({
'company': {
'id': company.id,
'name': company.name,
'is_personal': company.is_personal,
'is_active': company.is_active
},
'users': {
'total': User.query.filter_by(company_id=company.id).count(),
'verified': User.query.filter_by(company_id=company.id, is_verified=True).count(),
'blocked': User.query.filter_by(company_id=company.id, is_blocked=True).count(),
'roles': role_counts
},
'teams': team_count,
'projects': {
'total': project_count,
'active': active_projects,
'inactive': project_count - active_projects
},
'time_entries': {
'weekly': weekly_entries,
'monthly': monthly_entries,
'active_sessions': active_sessions
}
})

94
routes/export.py Normal file
View File

@@ -0,0 +1,94 @@
"""
Export routes for TimeTrack application.
Handles data export functionality for time entries and analytics.
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g
from datetime import datetime, time, timedelta
from models import db, TimeEntry, Role, Project
from data_formatting import prepare_export_data
from data_export import export_to_csv, export_to_excel
from routes.auth import login_required, company_required
# Create blueprint
export_bp = Blueprint('export', __name__, url_prefix='/export')
@export_bp.route('/')
@login_required
@company_required
def export_page():
"""Display the export page."""
return render_template('export.html', title='Export Data')
def get_date_range(period, start_date_str=None, end_date_str=None):
"""Get start and end date based on period or custom date range."""
today = datetime.now().date()
if period:
if period == 'today':
return today, today
elif period == 'week':
start_date = today - timedelta(days=today.weekday())
return start_date, today
elif period == 'month':
start_date = today.replace(day=1)
return start_date, today
elif period == 'all':
earliest_entry = TimeEntry.query.order_by(TimeEntry.arrival_time).first()
start_date = earliest_entry.arrival_time.date() if earliest_entry else today
return start_date, today
else:
# Custom date range
try:
start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date()
end_date = datetime.strptime(end_date_str, '%Y-%m-%d').date()
return start_date, end_date
except (ValueError, TypeError):
raise ValueError('Invalid date format')
@export_bp.route('/download')
@login_required
@company_required
def download_export():
"""Handle export download requests."""
export_format = request.args.get('format', 'csv')
period = request.args.get('period')
try:
start_date, end_date = get_date_range(
period,
request.args.get('start_date'),
request.args.get('end_date')
)
except ValueError:
flash('Invalid date format. Please use YYYY-MM-DD format.')
return redirect(url_for('export.export_page'))
# Query entries within the date range
start_datetime = datetime.combine(start_date, time.min)
end_datetime = datetime.combine(end_date, time.max)
entries = TimeEntry.query.filter(
TimeEntry.arrival_time >= start_datetime,
TimeEntry.arrival_time <= end_datetime
).order_by(TimeEntry.arrival_time).all()
if not entries:
flash('No entries found for the selected date range.')
return redirect(url_for('export.export_page'))
# Prepare data and filename
data = prepare_export_data(entries)
filename = f"timetrack_export_{start_date.strftime('%Y%m%d')}_to_{end_date.strftime('%Y%m%d')}"
# Export based on format
if export_format == 'csv':
return export_to_csv(data, filename)
elif export_format == 'excel':
return export_to_excel(data, filename)
else:
flash('Invalid export format.')
return redirect(url_for('export.export_page'))

96
routes/export_api.py Normal file
View File

@@ -0,0 +1,96 @@
"""
Export API routes for TimeTrack application.
Handles API endpoints for data export functionality.
"""
from flask import Blueprint, request, redirect, url_for, flash, g
from datetime import datetime
from models import Role
from routes.auth import login_required, role_required, company_required
from data_export import export_analytics_csv, export_analytics_excel
import logging
logger = logging.getLogger(__name__)
# Create blueprint
export_api_bp = Blueprint('export_api', __name__, url_prefix='/api')
def get_filtered_analytics_data(user, mode, start_date=None, end_date=None, project_filter=None):
"""Get filtered time entry data for analytics"""
from models import TimeEntry, User
from sqlalchemy import func
# Base query
query = TimeEntry.query
# Apply user/team filter
if mode == 'personal':
query = query.filter(TimeEntry.user_id == user.id)
elif mode == 'team' and user.team_id:
team_user_ids = [u.id for u in User.query.filter_by(team_id=user.team_id).all()]
query = query.filter(TimeEntry.user_id.in_(team_user_ids))
# Apply date filters
if start_date:
query = query.filter(func.date(TimeEntry.arrival_time) >= start_date)
if end_date:
query = query.filter(func.date(TimeEntry.arrival_time) <= end_date)
# Apply project filter
if project_filter:
if project_filter == 'none':
query = query.filter(TimeEntry.project_id.is_(None))
else:
try:
project_id = int(project_filter)
query = query.filter(TimeEntry.project_id == project_id)
except ValueError:
pass
return query.order_by(TimeEntry.arrival_time.desc()).all()
@export_api_bp.route('/analytics/export')
@login_required
@company_required
def analytics_export():
"""Export analytics data in various formats"""
export_format = request.args.get('format', 'csv')
view_type = request.args.get('view', 'table')
mode = request.args.get('mode', 'personal')
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
project_filter = request.args.get('project_id')
# Validate permissions
if mode == 'team':
if not g.user.team_id:
flash('No team assigned', 'error')
return redirect(url_for('analytics'))
if g.user.role not in [Role.TEAM_LEADER, Role.SUPERVISOR, Role.ADMIN]:
flash('Insufficient permissions', 'error')
return redirect(url_for('analytics'))
try:
# Parse dates
if start_date:
start_date = datetime.strptime(start_date, '%Y-%m-%d').date()
if end_date:
end_date = datetime.strptime(end_date, '%Y-%m-%d').date()
# Get data
data = get_filtered_analytics_data(g.user, mode, start_date, end_date, project_filter)
if export_format == 'csv':
return export_analytics_csv(data, view_type, mode)
elif export_format == 'excel':
return export_analytics_excel(data, view_type, mode)
else:
flash('Invalid export format', 'error')
return redirect(url_for('analytics'))
except Exception as e:
logger.error(f"Error in analytics export: {str(e)}")
flash('Error generating export', 'error')
return redirect(url_for('analytics'))

217
routes/invitations.py Normal file
View File

@@ -0,0 +1,217 @@
"""
Company invitation routes
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, jsonify
from models import db, CompanyInvitation, User, Role
from routes.auth import login_required, admin_required
from flask_mail import Message
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
invitations_bp = Blueprint('invitations', __name__, url_prefix='/invitations')
@invitations_bp.route('/')
@login_required
@admin_required
def list_invitations():
"""List all invitations for the company"""
invitations = CompanyInvitation.query.filter_by(
company_id=g.user.company_id
).order_by(CompanyInvitation.created_at.desc()).all()
# Separate into pending and accepted
pending_invitations = [inv for inv in invitations if inv.is_valid()]
accepted_invitations = [inv for inv in invitations if inv.accepted]
expired_invitations = [inv for inv in invitations if not inv.accepted and inv.is_expired()]
return render_template('invitations/list.html',
pending_invitations=pending_invitations,
accepted_invitations=accepted_invitations,
expired_invitations=expired_invitations,
title='Manage Invitations')
@invitations_bp.route('/send', methods=['GET', 'POST'])
@login_required
@admin_required
def send_invitation():
"""Send a new invitation"""
if request.method == 'POST':
email = request.form.get('email', '').strip()
role = request.form.get('role', 'Team Member')
custom_message = request.form.get('custom_message', '').strip()
if not email:
flash('Email address is required', 'error')
return redirect(url_for('invitations.send_invitation'))
# Check if user already exists in the company
existing_user = User.query.filter_by(
email=email,
company_id=g.user.company_id
).first()
if existing_user:
flash(f'A user with email {email} already exists in your company', 'error')
return redirect(url_for('invitations.send_invitation'))
# Check for pending invitations
pending_invitation = CompanyInvitation.query.filter_by(
email=email,
company_id=g.user.company_id,
accepted=False
).filter(CompanyInvitation.expires_at > datetime.now()).first()
if pending_invitation:
flash(f'An invitation has already been sent to {email} and is still pending', 'warning')
return redirect(url_for('invitations.list_invitations'))
# Create new invitation
invitation = CompanyInvitation(
company_id=g.user.company_id,
email=email,
role=role,
invited_by_id=g.user.id
)
db.session.add(invitation)
db.session.commit()
# Send invitation email
try:
from app import mail
# Build invitation URL
invitation_url = url_for('register_with_invitation',
token=invitation.token,
_external=True)
msg = Message(
f'Invitation to join {g.user.company.name} on {g.branding.app_name}',
recipients=[email]
)
msg.html = render_template('emails/invitation.html',
invitation=invitation,
invitation_url=invitation_url,
custom_message=custom_message,
sender=g.user)
msg.body = f"""
Hello,
{g.user.username} has invited you to join {g.user.company.name} on {g.branding.app_name}.
{custom_message if custom_message else ''}
Click the link below to accept the invitation and create your account:
{invitation_url}
This invitation will expire in 7 days.
Best regards,
The {g.branding.app_name} Team
"""
mail.send(msg)
logger.info(f"Invitation sent to {email} by {g.user.username}")
flash(f'Invitation sent successfully to {email}', 'success')
except Exception as e:
logger.error(f"Error sending invitation email: {str(e)}")
flash('Invitation created but failed to send email. The user can still use the invitation link.', 'warning')
return redirect(url_for('invitations.list_invitations'))
# GET request - show the form
roles = ['Team Member', 'Team Leader', 'Administrator']
return render_template('invitations/send.html', roles=roles, title='Send Invitation')
@invitations_bp.route('/revoke/<int:invitation_id>', methods=['POST'])
@login_required
@admin_required
def revoke_invitation(invitation_id):
"""Revoke a pending invitation"""
invitation = CompanyInvitation.query.filter_by(
id=invitation_id,
company_id=g.user.company_id,
accepted=False
).first()
if not invitation:
flash('Invitation not found or already accepted', 'error')
return redirect(url_for('invitations.list_invitations'))
# Instead of deleting, we'll expire it immediately
invitation.expires_at = datetime.now()
db.session.commit()
flash(f'Invitation to {invitation.email} has been revoked', 'success')
return redirect(url_for('invitations.list_invitations'))
@invitations_bp.route('/resend/<int:invitation_id>', methods=['POST'])
@login_required
@admin_required
def resend_invitation(invitation_id):
"""Resend an invitation email"""
invitation = CompanyInvitation.query.filter_by(
id=invitation_id,
company_id=g.user.company_id,
accepted=False
).first()
if not invitation:
flash('Invitation not found or already accepted', 'error')
return redirect(url_for('invitations.list_invitations'))
# Extend expiration if needed
if invitation.is_expired():
invitation.expires_at = datetime.now() + timedelta(days=7)
db.session.commit()
# Resend email
try:
from app import mail
invitation_url = url_for('register_with_invitation',
token=invitation.token,
_external=True)
msg = Message(
f'Reminder: Invitation to join {g.user.company.name}',
recipients=[invitation.email]
)
msg.html = render_template('emails/invitation_reminder.html',
invitation=invitation,
invitation_url=invitation_url,
sender=g.user)
msg.body = f"""
Hello,
This is a reminder that you have been invited to join {g.user.company.name} on {g.branding.app_name}.
Click the link below to accept the invitation and create your account:
{invitation_url}
This invitation will expire on {invitation.expires_at.strftime('%B %d, %Y')}.
Best regards,
The {g.branding.app_name} Team
"""
mail.send(msg)
flash(f'Invitation resent to {invitation.email}', 'success')
except Exception as e:
logger.error(f"Error resending invitation email: {str(e)}")
flash('Failed to resend invitation email', 'error')
return redirect(url_for('invitations.list_invitations'))

View File

@@ -195,6 +195,7 @@ def create_note():
tags = request.form.get('tags', '').strip()
project_id = request.form.get('project_id')
task_id = request.form.get('task_id')
is_pinned = request.form.get('is_pinned') == '1'
# Validate
if not title:
@@ -240,9 +241,13 @@ def create_note():
company_id=g.user.company_id,
created_by_id=g.user.id,
project_id=project.id if project else None,
task_id=task.id if task else None
task_id=task.id if task else None,
is_pinned=is_pinned
)
# Generate slug before saving
note.generate_slug()
db.session.add(note)
db.session.commit()
@@ -258,7 +263,7 @@ def create_note():
# Get projects for dropdown
projects = Project.query.filter_by(
company_id=g.user.company_id,
is_archived=False
is_active=True
).order_by(Project.name).all()
# Get task if specified in URL
@@ -359,6 +364,7 @@ def edit_note(slug):
tags = request.form.get('tags', '').strip()
project_id = request.form.get('project_id')
task_id = request.form.get('task_id')
is_pinned = request.form.get('is_pinned') == '1'
# Validate
if not title:
@@ -402,6 +408,7 @@ def edit_note(slug):
note.tags = tags if tags else None
note.project_id = project.id if project else None
note.task_id = task.id if task else None
note.is_pinned = is_pinned
note.updated_at = datetime.now(timezone.utc)
# Update slug if title changed
@@ -421,7 +428,7 @@ def edit_note(slug):
# Get projects for dropdown
projects = Project.query.filter_by(
company_id=g.user.company_id,
is_archived=False
is_active=True
).order_by(Project.name).all()
return render_template('note_editor.html',

View File

@@ -245,6 +245,36 @@ def update_note_folder(slug):
})
@notes_api_bp.route('/<int:note_id>/move', methods=['POST'])
@login_required
@company_required
def move_note_to_folder(note_id):
"""Move a note to a different folder (used by drag and drop)"""
note = Note.query.filter_by(id=note_id, company_id=g.user.company_id).first()
if not note:
return jsonify({'success': False, 'error': 'Note not found'}), 404
# Check permissions
if not note.can_user_edit(g.user):
return jsonify({'success': False, 'error': 'Permission denied'}), 403
data = request.get_json()
new_folder = data.get('folder', '').strip()
# Update note folder
note.folder = new_folder if new_folder else None
note.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
'success': True,
'message': 'Note moved successfully',
'folder': note.folder or ''
})
@notes_api_bp.route('/<int:note_id>/tags', methods=['POST'])
@login_required
@company_required

238
routes/projects.py Normal file
View File

@@ -0,0 +1,238 @@
"""
Project Management Routes
Handles all project-related views and operations
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, abort
from datetime import datetime
from models import db, Project, Team, ProjectCategory, TimeEntry, Role, Task, User
from routes.auth import role_required, company_required, admin_required
from utils.validation import FormValidator
from utils.repository import ProjectRepository
projects_bp = Blueprint('projects', __name__, url_prefix='/admin/projects')
@projects_bp.route('')
@role_required(Role.SUPERVISOR) # Supervisors and Admins can manage projects
@company_required
def admin_projects():
project_repo = ProjectRepository()
projects = project_repo.get_by_company_ordered(g.user.company_id, Project.created_at.desc())
categories = ProjectCategory.query.filter_by(company_id=g.user.company_id).order_by(ProjectCategory.name).all()
return render_template('admin_projects.html', title='Project Management', projects=projects, categories=categories)
@projects_bp.route('/create', methods=['GET', 'POST'])
@role_required(Role.SUPERVISOR)
@company_required
def create_project():
if request.method == 'POST':
validator = FormValidator()
project_repo = ProjectRepository()
name = request.form.get('name')
description = request.form.get('description')
code = request.form.get('code')
team_id = request.form.get('team_id') or None
category_id = request.form.get('category_id') or None
start_date_str = request.form.get('start_date')
end_date_str = request.form.get('end_date')
# Validate required fields
validator.validate_required(name, 'Project name')
validator.validate_required(code, 'Project code')
# Validate code uniqueness
if validator.is_valid():
validator.validate_unique(Project, 'code', code, company_id=g.user.company_id)
# Parse dates
start_date = validator.parse_date(start_date_str, 'Start date')
end_date = validator.parse_date(end_date_str, 'End date')
# Validate date range
if start_date and end_date:
validator.validate_date_range(start_date, end_date)
if validator.is_valid():
project = project_repo.create(
name=name,
description=description,
code=code.upper(),
company_id=g.user.company_id,
team_id=int(team_id) if team_id else None,
category_id=int(category_id) if category_id else None,
start_date=start_date,
end_date=end_date,
created_by_id=g.user.id
)
project_repo.save()
flash(f'Project "{name}" created successfully!', 'success')
return redirect(url_for('projects.admin_projects'))
else:
validator.flash_errors()
# Get available teams and categories for the form (company-scoped)
teams = Team.query.filter_by(company_id=g.user.company_id).order_by(Team.name).all()
categories = ProjectCategory.query.filter_by(company_id=g.user.company_id).order_by(ProjectCategory.name).all()
return render_template('create_project.html', title='Create Project', teams=teams, categories=categories)
@projects_bp.route('/edit/<int:project_id>', methods=['GET', 'POST'])
@role_required(Role.SUPERVISOR)
@company_required
def edit_project(project_id):
project_repo = ProjectRepository()
project = project_repo.get_by_id_and_company(project_id, g.user.company_id)
if not project:
abort(404)
if request.method == 'POST':
validator = FormValidator()
name = request.form.get('name')
description = request.form.get('description')
code = request.form.get('code')
team_id = request.form.get('team_id') or None
category_id = request.form.get('category_id') or None
is_active = request.form.get('is_active') == 'on'
start_date_str = request.form.get('start_date')
end_date_str = request.form.get('end_date')
# Validate required fields
validator.validate_required(name, 'Project name')
validator.validate_required(code, 'Project code')
# Validate code uniqueness (exclude current project)
if validator.is_valid() and code != project.code:
validator.validate_unique(Project, 'code', code, company_id=g.user.company_id)
# Parse dates
start_date = validator.parse_date(start_date_str, 'Start date')
end_date = validator.parse_date(end_date_str, 'End date')
# Validate date range
if start_date and end_date:
validator.validate_date_range(start_date, end_date)
if validator.is_valid():
project_repo.update(project,
name=name,
description=description,
code=code.upper(),
team_id=int(team_id) if team_id else None,
category_id=int(category_id) if category_id else None,
is_active=is_active,
start_date=start_date,
end_date=end_date
)
project_repo.save()
flash(f'Project "{name}" updated successfully!', 'success')
return redirect(url_for('projects.admin_projects'))
else:
validator.flash_errors()
# Get available teams and categories for the form (company-scoped)
teams = Team.query.filter_by(company_id=g.user.company_id).order_by(Team.name).all()
categories = ProjectCategory.query.filter_by(company_id=g.user.company_id).order_by(ProjectCategory.name).all()
return render_template('edit_project.html', title='Edit Project', project=project, teams=teams, categories=categories)
@projects_bp.route('/delete/<int:project_id>', methods=['POST'])
@company_required
def delete_project(project_id):
# Check if user is admin or system admin
if g.user.role not in [Role.ADMIN, Role.SYSTEM_ADMIN]:
flash('You do not have permission to delete projects.', 'error')
return redirect(url_for('projects.admin_projects'))
project_repo = ProjectRepository()
project = project_repo.get_by_id_and_company(project_id, g.user.company_id)
if not project:
abort(404)
project_name = project.name
try:
# Import models needed for cascading deletions
from models import Sprint, SubTask, TaskDependency, Comment
# Delete all related data in the correct order
# Delete comments on tasks in this project
Comment.query.filter(Comment.task_id.in_(
db.session.query(Task.id).filter(Task.project_id == project_id)
)).delete(synchronize_session=False)
# Delete subtasks
SubTask.query.filter(SubTask.task_id.in_(
db.session.query(Task.id).filter(Task.project_id == project_id)
)).delete(synchronize_session=False)
# Delete task dependencies
TaskDependency.query.filter(
TaskDependency.blocked_task_id.in_(
db.session.query(Task.id).filter(Task.project_id == project_id)
) | TaskDependency.blocking_task_id.in_(
db.session.query(Task.id).filter(Task.project_id == project_id)
)
).delete(synchronize_session=False)
# Delete tasks
Task.query.filter_by(project_id=project_id).delete()
# Delete sprints
Sprint.query.filter_by(project_id=project_id).delete()
# Delete time entries
TimeEntry.query.filter_by(project_id=project_id).delete()
# Finally, delete the project
project_repo.delete(project)
db.session.commit()
flash(f'Project "{project_name}" and all related data have been permanently deleted.', 'success')
except Exception as e:
db.session.rollback()
flash(f'Error deleting project: {str(e)}', 'error')
return redirect(url_for('projects.edit_project', project_id=project_id))
return redirect(url_for('projects.admin_projects'))
@projects_bp.route('/<int:project_id>/tasks')
@role_required(Role.TEAM_MEMBER) # All authenticated users can view tasks
@company_required
def manage_project_tasks(project_id):
project_repo = ProjectRepository()
project = project_repo.get_by_id_and_company(project_id, g.user.company_id)
if not project:
abort(404)
# Check if user has access to this project
if not project.is_user_allowed(g.user):
flash('You do not have access to this project.', 'error')
return redirect(url_for('projects.admin_projects'))
# Get all tasks for this project
tasks = Task.query.filter_by(project_id=project_id).order_by(Task.created_at.desc()).all()
# Get team members for assignment dropdown
if project.team_id:
# If project is assigned to a specific team, only show team members
team_members = User.query.filter_by(team_id=project.team_id, company_id=g.user.company_id).all()
else:
# If project is available to all teams, show all company users
team_members = User.query.filter_by(company_id=g.user.company_id).all()
return render_template('manage_project_tasks.html',
title=f'Tasks - {project.name}',
project=project,
tasks=tasks,
team_members=team_members)

170
routes/projects_api.py Normal file
View File

@@ -0,0 +1,170 @@
"""
Project Management API Routes
Handles all project-related API endpoints including categories
"""
from flask import Blueprint, jsonify, request, g
from sqlalchemy import or_ as sql_or
from models import db, Project, ProjectCategory, Role
from routes.auth import role_required, company_required, admin_required
import logging
logger = logging.getLogger(__name__)
projects_api_bp = Blueprint('projects_api', __name__, url_prefix='/api')
# Category Management API Routes
@projects_api_bp.route('/admin/categories', methods=['POST'])
@role_required(Role.ADMIN)
@company_required
def create_category():
try:
data = request.get_json()
name = data.get('name')
description = data.get('description', '')
color = data.get('color', '#007bff')
icon = data.get('icon', '')
if not name:
return jsonify({'success': False, 'message': 'Category name is required'})
# Check if category already exists
existing = ProjectCategory.query.filter_by(
name=name,
company_id=g.user.company_id
).first()
if existing:
return jsonify({'success': False, 'message': 'Category name already exists'})
category = ProjectCategory(
name=name,
description=description,
color=color,
icon=icon,
company_id=g.user.company_id,
created_by_id=g.user.id
)
db.session.add(category)
db.session.commit()
return jsonify({'success': True, 'message': 'Category created successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@projects_api_bp.route('/admin/categories/<int:category_id>', methods=['PUT'])
@role_required(Role.ADMIN)
@company_required
def update_category(category_id):
try:
category = ProjectCategory.query.filter_by(
id=category_id,
company_id=g.user.company_id
).first()
if not category:
return jsonify({'success': False, 'message': 'Category not found'})
data = request.get_json()
name = data.get('name')
if not name:
return jsonify({'success': False, 'message': 'Category name is required'})
# Check if name conflicts with another category
existing = ProjectCategory.query.filter(
ProjectCategory.name == name,
ProjectCategory.company_id == g.user.company_id,
ProjectCategory.id != category_id
).first()
if existing:
return jsonify({'success': False, 'message': 'Category name already exists'})
category.name = name
category.description = data.get('description', '')
category.color = data.get('color', category.color)
category.icon = data.get('icon', '')
db.session.commit()
return jsonify({'success': True, 'message': 'Category updated successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@projects_api_bp.route('/admin/categories/<int:category_id>', methods=['DELETE'])
@role_required(Role.ADMIN)
@company_required
def delete_category(category_id):
try:
category = ProjectCategory.query.filter_by(
id=category_id,
company_id=g.user.company_id
).first()
if not category:
return jsonify({'success': False, 'message': 'Category not found'})
# Unassign projects from this category
projects = Project.query.filter_by(category_id=category_id).all()
for project in projects:
project.category_id = None
db.session.delete(category)
db.session.commit()
return jsonify({'success': True, 'message': 'Category deleted successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@projects_api_bp.route('/search/projects')
@role_required(Role.TEAM_MEMBER)
@company_required
def search_projects():
"""Search for projects for smart search auto-completion"""
try:
query = request.args.get('q', '').strip()
if not query:
return jsonify({'success': True, 'projects': []})
# Search projects the user has access to
projects = Project.query.filter(
Project.company_id == g.user.company_id,
sql_or(
Project.code.ilike(f'%{query}%'),
Project.name.ilike(f'%{query}%')
)
).limit(10).all()
# Filter projects user has access to
accessible_projects = [
project for project in projects
if project.is_user_allowed(g.user)
]
project_list = [
{
'id': project.id,
'code': project.code,
'name': project.name
}
for project in accessible_projects
]
return jsonify({'success': True, 'projects': project_list})
except Exception as e:
logger.error(f"Error in search_projects: {str(e)}")
return jsonify({'success': False, 'message': str(e)})

47
routes/sprints.py Normal file
View File

@@ -0,0 +1,47 @@
"""
Sprint Management Routes
Handles all sprint-related views
"""
from flask import Blueprint, render_template, g, redirect, url_for, flash
from sqlalchemy import or_
from models import db, Role, Project, Sprint
from routes.auth import login_required, role_required, company_required
sprints_bp = Blueprint('sprints', __name__, url_prefix='/sprints')
@sprints_bp.route('')
@role_required(Role.TEAM_MEMBER)
@company_required
def sprint_management():
"""Sprint management interface"""
# Get all projects the user has access to (for sprint assignment)
if g.user.role in [Role.ADMIN, Role.SUPERVISOR]:
# Admins and Supervisors can see all company projects
available_projects = Project.query.filter_by(
company_id=g.user.company_id,
is_active=True
).order_by(Project.name).all()
elif g.user.team_id:
# Team members see team projects + unassigned projects
available_projects = Project.query.filter(
Project.company_id == g.user.company_id,
Project.is_active == True,
or_(Project.team_id == g.user.team_id, Project.team_id == None)
).order_by(Project.name).all()
# Filter by actual access permissions
available_projects = [p for p in available_projects if p.is_user_allowed(g.user)]
else:
# Unassigned users see only unassigned projects
available_projects = Project.query.filter_by(
company_id=g.user.company_id,
team_id=None,
is_active=True
).order_by(Project.name).all()
available_projects = [p for p in available_projects if p.is_user_allowed(g.user)]
return render_template('sprint_management.html',
title='Sprint Management',
available_projects=available_projects)

224
routes/sprints_api.py Normal file
View File

@@ -0,0 +1,224 @@
"""
Sprint Management API Routes
Handles all sprint-related API endpoints
"""
from flask import Blueprint, jsonify, request, g
from datetime import datetime
from models import db, Role, Project, Sprint, SprintStatus, Task
from routes.auth import login_required, role_required, company_required
import logging
logger = logging.getLogger(__name__)
sprints_api_bp = Blueprint('sprints_api', __name__, url_prefix='/api')
@sprints_api_bp.route('/sprints')
@role_required(Role.TEAM_MEMBER)
@company_required
def get_sprints():
"""Get all sprints for the user's company"""
try:
# Base query for sprints in user's company
query = Sprint.query.filter(Sprint.company_id == g.user.company_id)
# Apply access restrictions based on user role and team
if g.user.role not in [Role.ADMIN, Role.SUPERVISOR]:
# Regular users can only see sprints they have access to
accessible_sprint_ids = []
sprints = query.all()
for sprint in sprints:
if sprint.can_user_access(g.user):
accessible_sprint_ids.append(sprint.id)
if accessible_sprint_ids:
query = query.filter(Sprint.id.in_(accessible_sprint_ids))
else:
# No accessible sprints, return empty list
return jsonify({'success': True, 'sprints': []})
sprints = query.order_by(Sprint.created_at.desc()).all()
sprint_list = []
for sprint in sprints:
task_summary = sprint.get_task_summary()
sprint_data = {
'id': sprint.id,
'name': sprint.name,
'description': sprint.description,
'status': sprint.status.name,
'company_id': sprint.company_id,
'project_id': sprint.project_id,
'project_name': sprint.project.name if sprint.project else None,
'project_code': sprint.project.code if sprint.project else None,
'start_date': sprint.start_date.isoformat(),
'end_date': sprint.end_date.isoformat(),
'goal': sprint.goal,
'capacity_hours': sprint.capacity_hours,
'created_by_id': sprint.created_by_id,
'created_by_name': sprint.created_by.username if sprint.created_by else None,
'created_at': sprint.created_at.isoformat(),
'is_current': sprint.is_current,
'duration_days': sprint.duration_days,
'days_remaining': sprint.days_remaining,
'progress_percentage': sprint.progress_percentage,
'task_summary': task_summary
}
sprint_list.append(sprint_data)
return jsonify({'success': True, 'sprints': sprint_list})
except Exception as e:
logger.error(f"Error in get_sprints: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@sprints_api_bp.route('/sprints', methods=['POST'])
@role_required(Role.TEAM_LEADER) # Team leaders and above can create sprints
@company_required
def create_sprint():
"""Create a new sprint"""
try:
data = request.get_json()
# Validate required fields
name = data.get('name')
start_date = data.get('start_date')
end_date = data.get('end_date')
if not name:
return jsonify({'success': False, 'message': 'Sprint name is required'})
if not start_date:
return jsonify({'success': False, 'message': 'Start date is required'})
if not end_date:
return jsonify({'success': False, 'message': 'End date is required'})
# Parse dates
try:
start_date = datetime.strptime(start_date, '%Y-%m-%d').date()
end_date = datetime.strptime(end_date, '%Y-%m-%d').date()
except ValueError:
return jsonify({'success': False, 'message': 'Invalid date format'})
if start_date >= end_date:
return jsonify({'success': False, 'message': 'End date must be after start date'})
# Verify project access if project is specified
project_id = data.get('project_id')
if project_id:
project = Project.query.filter_by(id=project_id, company_id=g.user.company_id).first()
if not project or not project.is_user_allowed(g.user):
return jsonify({'success': False, 'message': 'Project not found or access denied'})
# Create sprint
sprint = Sprint(
name=name,
description=data.get('description', ''),
status=SprintStatus[data.get('status', 'PLANNING')],
company_id=g.user.company_id,
project_id=int(project_id) if project_id else None,
start_date=start_date,
end_date=end_date,
goal=data.get('goal'),
capacity_hours=int(data.get('capacity_hours')) if data.get('capacity_hours') else None,
created_by_id=g.user.id
)
db.session.add(sprint)
db.session.commit()
return jsonify({'success': True, 'message': 'Sprint created successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error creating sprint: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@sprints_api_bp.route('/sprints/<int:sprint_id>', methods=['PUT'])
@role_required(Role.TEAM_LEADER)
@company_required
def update_sprint(sprint_id):
"""Update an existing sprint"""
try:
sprint = Sprint.query.filter_by(id=sprint_id, company_id=g.user.company_id).first()
if not sprint or not sprint.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Sprint not found or access denied'})
data = request.get_json()
# Update sprint fields
if 'name' in data:
sprint.name = data['name']
if 'description' in data:
sprint.description = data['description']
if 'status' in data:
sprint.status = SprintStatus[data['status']]
if 'goal' in data:
sprint.goal = data['goal']
if 'capacity_hours' in data:
sprint.capacity_hours = int(data['capacity_hours']) if data['capacity_hours'] else None
if 'project_id' in data:
project_id = data['project_id']
if project_id:
project = Project.query.filter_by(id=project_id, company_id=g.user.company_id).first()
if not project or not project.is_user_allowed(g.user):
return jsonify({'success': False, 'message': 'Project not found or access denied'})
sprint.project_id = int(project_id)
else:
sprint.project_id = None
# Update dates if provided
if 'start_date' in data:
try:
sprint.start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date()
except ValueError:
return jsonify({'success': False, 'message': 'Invalid start date format'})
if 'end_date' in data:
try:
sprint.end_date = datetime.strptime(data['end_date'], '%Y-%m-%d').date()
except ValueError:
return jsonify({'success': False, 'message': 'Invalid end date format'})
# Validate date order
if sprint.start_date >= sprint.end_date:
return jsonify({'success': False, 'message': 'End date must be after start date'})
db.session.commit()
return jsonify({'success': True, 'message': 'Sprint updated successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error updating sprint: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@sprints_api_bp.route('/sprints/<int:sprint_id>', methods=['DELETE'])
@role_required(Role.TEAM_LEADER)
@company_required
def delete_sprint(sprint_id):
"""Delete a sprint and remove it from all associated tasks"""
try:
sprint = Sprint.query.filter_by(id=sprint_id, company_id=g.user.company_id).first()
if not sprint or not sprint.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Sprint not found or access denied'})
# Remove sprint assignment from all tasks
Task.query.filter_by(sprint_id=sprint_id).update({'sprint_id': None})
# Delete the sprint
db.session.delete(sprint)
db.session.commit()
return jsonify({'success': True, 'message': 'Sprint deleted successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error deleting sprint: {str(e)}")
return jsonify({'success': False, 'message': str(e)})

511
routes/system_admin.py Normal file
View File

@@ -0,0 +1,511 @@
"""
System Administrator routes
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, current_app
from models import (db, Company, User, Role, Team, Project, TimeEntry, SystemSettings,
SystemEvent, BrandingSettings, Task, SubTask, TaskDependency, Sprint,
Comment, UserPreferences, UserDashboard, WorkConfig, CompanySettings,
CompanyWorkConfig, ProjectCategory)
from routes.auth import system_admin_required
from flask import session
from sqlalchemy import func
from datetime import datetime, timedelta
import logging
import os
logger = logging.getLogger(__name__)
system_admin_bp = Blueprint('system_admin', __name__, url_prefix='/system-admin')
@system_admin_bp.route('/dashboard')
@system_admin_required
def system_admin_dashboard():
"""System Administrator Dashboard - view all data across companies"""
# Global statistics
total_companies = Company.query.count()
total_users = User.query.count()
total_teams = Team.query.count()
total_projects = Project.query.count()
total_time_entries = TimeEntry.query.count()
# System admin count
system_admins = User.query.filter_by(role=Role.SYSTEM_ADMIN).count()
regular_admins = User.query.filter_by(role=Role.ADMIN).count()
# Recent activity (last 7 days)
week_ago = datetime.now() - timedelta(days=7)
recent_users = User.query.filter(User.created_at >= week_ago).count()
recent_companies = Company.query.filter(Company.created_at >= week_ago).count()
recent_time_entries = TimeEntry.query.filter(TimeEntry.arrival_time >= week_ago).count()
# Top companies by user count
top_companies = db.session.query(
Company.name,
Company.id,
db.func.count(User.id).label('user_count')
).join(User).group_by(Company.id).order_by(db.func.count(User.id).desc()).limit(5).all()
# Recent companies
recent_companies_list = Company.query.order_by(Company.created_at.desc()).limit(5).all()
# System health checks
orphaned_users = User.query.filter_by(company_id=None).count()
orphaned_time_entries = TimeEntry.query.filter_by(user_id=None).count()
blocked_users = User.query.filter_by(is_blocked=True).count()
return render_template('system_admin_dashboard.html',
title='System Administrator Dashboard',
total_companies=total_companies,
total_users=total_users,
total_teams=total_teams,
total_projects=total_projects,
total_time_entries=total_time_entries,
system_admins=system_admins,
regular_admins=regular_admins,
recent_users=recent_users,
recent_companies=recent_companies,
recent_time_entries=recent_time_entries,
top_companies=top_companies,
recent_companies_list=recent_companies_list,
orphaned_users=orphaned_users,
orphaned_time_entries=orphaned_time_entries,
blocked_users=blocked_users)
@system_admin_bp.route('/companies')
@system_admin_required
def system_admin_companies():
"""System admin view of all companies"""
# Get filter parameters
search_query = request.args.get('search', '')
# Base query
query = Company.query
# Apply search filter
if search_query:
query = query.filter(
db.or_(
Company.name.ilike(f'%{search_query}%'),
Company.slug.ilike(f'%{search_query}%')
)
)
# Get all companies
companies = query.order_by(Company.created_at.desc()).all()
# Create a paginated response object
from flask_sqlalchemy import Pagination
page = request.args.get('page', 1, type=int)
per_page = 20
# Paginate companies
companies_paginated = query.paginate(page=page, per_page=per_page, error_out=False)
# Calculate statistics for each company
company_stats = {}
for company in companies_paginated.items:
company_stats[company.id] = {
'user_count': User.query.filter_by(company_id=company.id).count(),
'admin_count': User.query.filter_by(company_id=company.id, role=Role.ADMIN).count(),
'team_count': Team.query.filter_by(company_id=company.id).count(),
'project_count': Project.query.filter_by(company_id=company.id).count(),
'active_projects': Project.query.filter_by(company_id=company.id, is_active=True).count(),
}
return render_template('system_admin_companies.html',
title='System Admin - Companies',
companies=companies_paginated,
company_stats=company_stats,
search_query=search_query)
@system_admin_bp.route('/companies/<int:company_id>')
@system_admin_required
def system_admin_company_detail(company_id):
"""System admin detailed view of a specific company"""
company = Company.query.get_or_404(company_id)
# Get recent time entries count
week_ago = datetime.now() - timedelta(days=7)
recent_time_entries = TimeEntry.query.join(User).filter(
User.company_id == company.id,
TimeEntry.arrival_time >= week_ago
).count()
# Get role distribution
role_counts = {}
for role in Role:
count = User.query.filter_by(company_id=company.id, role=role).count()
if count > 0:
role_counts[role.value] = count
# Get users list
users = User.query.filter_by(company_id=company.id).order_by(User.created_at.desc()).all()
# Get teams list
teams = Team.query.filter_by(company_id=company.id).all()
# Get projects list
projects = Project.query.filter_by(company_id=company.id).order_by(Project.created_at.desc()).all()
return render_template('system_admin_company_detail.html',
title=f'Company Details - {company.name}',
company=company,
users=users,
teams=teams,
projects=projects,
recent_time_entries=recent_time_entries,
role_counts=role_counts)
@system_admin_bp.route('/companies/<int:company_id>/delete', methods=['POST'])
@system_admin_required
def delete_company(company_id):
"""System Admin: Delete a company and all its data"""
company = Company.query.get_or_404(company_id)
company_name = company.name
try:
# Delete all related data in the correct order to avoid foreign key constraints
# Delete comments (must be before tasks)
Comment.query.filter(Comment.task_id.in_(
db.session.query(Task.id).join(Project).filter(Project.company_id == company_id)
)).delete(synchronize_session=False)
# Delete subtasks
SubTask.query.filter(SubTask.task_id.in_(
db.session.query(Task.id).join(Project).filter(Project.company_id == company_id)
)).delete(synchronize_session=False)
# Delete task dependencies
TaskDependency.query.filter(
TaskDependency.blocked_task_id.in_(
db.session.query(Task.id).join(Project).filter(Project.company_id == company_id)
) | TaskDependency.blocking_task_id.in_(
db.session.query(Task.id).join(Project).filter(Project.company_id == company_id)
)
).delete(synchronize_session=False)
# Delete tasks
Task.query.filter(Task.project_id.in_(
db.session.query(Project.id).filter(Project.company_id == company_id)
)).delete(synchronize_session=False)
# Delete sprints
Sprint.query.filter(Sprint.project_id.in_(
db.session.query(Project.id).filter(Project.company_id == company_id)
)).delete(synchronize_session=False)
# Delete time entries
TimeEntry.query.filter(TimeEntry.user_id.in_(
db.session.query(User.id).filter(User.company_id == company_id)
)).delete(synchronize_session=False)
# Delete projects
Project.query.filter_by(company_id=company_id).delete()
# Delete teams
Team.query.filter_by(company_id=company_id).delete()
# Delete user preferences, dashboards, and work configs
UserPreferences.query.filter(UserPreferences.user_id.in_(
db.session.query(User.id).filter(User.company_id == company_id)
)).delete(synchronize_session=False)
UserDashboard.query.filter(UserDashboard.user_id.in_(
db.session.query(User.id).filter(User.company_id == company_id)
)).delete(synchronize_session=False)
WorkConfig.query.filter(WorkConfig.user_id.in_(
db.session.query(User.id).filter(User.company_id == company_id)
)).delete(synchronize_session=False)
# Delete users
User.query.filter_by(company_id=company_id).delete()
# Delete company settings and work config
CompanySettings.query.filter_by(company_id=company_id).delete()
CompanyWorkConfig.query.filter_by(company_id=company_id).delete()
# Delete project categories
ProjectCategory.query.filter_by(company_id=company_id).delete()
# Finally, delete the company
db.session.delete(company)
db.session.commit()
flash(f'Company "{company_name}" and all its data have been permanently deleted.', 'success')
logger.info(f"System admin {g.user.username} deleted company {company_name} (ID: {company_id})")
except Exception as e:
db.session.rollback()
logger.error(f"Error deleting company {company_id}: {str(e)}")
flash(f'Error deleting company: {str(e)}', 'error')
return redirect(url_for('system_admin.system_admin_company_detail', company_id=company_id))
return redirect(url_for('system_admin.system_admin_companies'))
@system_admin_bp.route('/time-entries')
@system_admin_required
def system_admin_time_entries():
"""System Admin: View time entries across all companies"""
page = request.args.get('page', 1, type=int)
company_filter = request.args.get('company', '')
per_page = 50
# Build query properly with explicit joins
query = db.session.query(
TimeEntry,
User.username,
Company.name.label('company_name'),
Project.name.label('project_name')
).join(
User, TimeEntry.user_id == User.id
).join(
Company, User.company_id == Company.id
).outerjoin(
Project, TimeEntry.project_id == Project.id
)
# Apply company filter
if company_filter:
query = query.filter(Company.id == company_filter)
# Order by arrival time (newest first)
query = query.order_by(TimeEntry.arrival_time.desc())
# Paginate
entries = query.paginate(page=page, per_page=per_page, error_out=False)
# Get companies for filter dropdown
companies = Company.query.order_by(Company.name).all()
# Get today's date for the template
today = datetime.now().date()
return render_template('system_admin_time_entries.html',
title='System Admin - Time Entries',
entries=entries,
companies=companies,
current_company=company_filter,
today=today)
@system_admin_bp.route('/settings', methods=['GET', 'POST'])
@system_admin_required
def system_admin_settings():
"""System Admin: Global system settings"""
if request.method == 'POST':
# Update system settings
registration_enabled = request.form.get('registration_enabled') == 'on'
email_verification = request.form.get('email_verification_required') == 'on'
tracking_script_enabled = request.form.get('tracking_script_enabled') == 'on'
tracking_script_code = request.form.get('tracking_script_code', '')
# Update or create settings
reg_setting = SystemSettings.query.filter_by(key='registration_enabled').first()
if reg_setting:
reg_setting.value = 'true' if registration_enabled else 'false'
else:
reg_setting = SystemSettings(
key='registration_enabled',
value='true' if registration_enabled else 'false',
description='Controls whether new user registration is allowed'
)
db.session.add(reg_setting)
email_setting = SystemSettings.query.filter_by(key='email_verification_required').first()
if email_setting:
email_setting.value = 'true' if email_verification else 'false'
else:
email_setting = SystemSettings(
key='email_verification_required',
value='true' if email_verification else 'false',
description='Controls whether email verification is required for new accounts'
)
db.session.add(email_setting)
tracking_enabled_setting = SystemSettings.query.filter_by(key='tracking_script_enabled').first()
if tracking_enabled_setting:
tracking_enabled_setting.value = 'true' if tracking_script_enabled else 'false'
else:
tracking_enabled_setting = SystemSettings(
key='tracking_script_enabled',
value='true' if tracking_script_enabled else 'false',
description='Controls whether custom tracking script is enabled'
)
db.session.add(tracking_enabled_setting)
tracking_code_setting = SystemSettings.query.filter_by(key='tracking_script_code').first()
if tracking_code_setting:
tracking_code_setting.value = tracking_script_code
else:
tracking_code_setting = SystemSettings(
key='tracking_script_code',
value=tracking_script_code,
description='Custom tracking script code (HTML/JavaScript)'
)
db.session.add(tracking_code_setting)
db.session.commit()
flash('System settings updated successfully.', 'success')
return redirect(url_for('system_admin.system_admin_settings'))
# Get current settings
settings = {}
all_settings = SystemSettings.query.all()
for setting in all_settings:
if setting.key == 'registration_enabled':
settings['registration_enabled'] = setting.value == 'true'
elif setting.key == 'email_verification_required':
settings['email_verification_required'] = setting.value == 'true'
elif setting.key == 'tracking_script_enabled':
settings['tracking_script_enabled'] = setting.value == 'true'
elif setting.key == 'tracking_script_code':
settings['tracking_script_code'] = setting.value
# System statistics
total_companies = Company.query.count()
total_users = User.query.count()
total_system_admins = User.query.filter_by(role=Role.SYSTEM_ADMIN).count()
return render_template('system_admin_settings.html',
title='System Administrator Settings',
settings=settings,
total_companies=total_companies,
total_users=total_users,
total_system_admins=total_system_admins)
@system_admin_bp.route('/health')
@system_admin_required
def system_admin_health():
"""System Admin: System health check and event log"""
# Get system health summary
health_summary = SystemEvent.get_system_health_summary()
# Get recent events (last 7 days)
recent_events = SystemEvent.get_recent_events(days=7, limit=100)
# Get events by severity for quick stats
errors = SystemEvent.get_events_by_severity('error', days=7, limit=20)
warnings = SystemEvent.get_events_by_severity('warning', days=7, limit=20)
# System metrics
now = datetime.now()
# Database connection test
db_healthy = True
db_error = None
try:
db.session.execute('SELECT 1')
except Exception as e:
db_healthy = False
db_error = str(e)
SystemEvent.log_event(
'database_check_failed',
f'Database health check failed: {str(e)}',
'system',
'error'
)
# Application uptime (approximate based on first event)
first_event = SystemEvent.query.order_by(SystemEvent.timestamp.asc()).first()
uptime_start = first_event.timestamp if first_event else now
uptime_duration = now - uptime_start
# Recent activity stats
today = now.date()
today_events = SystemEvent.query.filter(
func.date(SystemEvent.timestamp) == today
).count()
# Log the health check
SystemEvent.log_event(
'system_health_check',
f'System health check performed by {session.get("username", "unknown")}',
'system',
'info',
user_id=session.get('user_id'),
ip_address=request.remote_addr,
user_agent=request.headers.get('User-Agent')
)
return render_template('system_admin_health.html',
title='System Health Check',
health_summary=health_summary,
recent_events=recent_events,
errors=errors,
warnings=warnings,
db_healthy=db_healthy,
db_error=db_error,
uptime_duration=uptime_duration,
today_events=today_events)
@system_admin_bp.route('/branding', methods=['GET', 'POST'])
@system_admin_required
def branding():
"""System Admin: Branding settings"""
if request.method == 'POST':
branding = BrandingSettings.get_current()
# Handle form data
branding.app_name = request.form.get('app_name', g.branding.app_name).strip()
branding.logo_alt_text = request.form.get('logo_alt_text', '').strip()
branding.primary_color = request.form.get('primary_color', '#007bff').strip()
# Handle imprint settings
branding.imprint_enabled = 'imprint_enabled' in request.form
branding.imprint_title = request.form.get('imprint_title', 'Imprint').strip()
branding.imprint_content = request.form.get('imprint_content', '').strip()
branding.updated_by_id = g.user.id
# Handle logo upload
if 'logo_file' in request.files:
logo_file = request.files['logo_file']
if logo_file and logo_file.filename:
# Create uploads directory if it doesn't exist
upload_dir = os.path.join(current_app.static_folder, 'uploads', 'branding')
os.makedirs(upload_dir, exist_ok=True)
# Save the file with a timestamp to avoid conflicts
import time
filename = f"logo_{int(time.time())}_{logo_file.filename}"
logo_path = os.path.join(upload_dir, filename)
logo_file.save(logo_path)
branding.logo_filename = filename
# Handle favicon upload
if 'favicon_file' in request.files:
favicon_file = request.files['favicon_file']
if favicon_file and favicon_file.filename:
# Create uploads directory if it doesn't exist
upload_dir = os.path.join(current_app.static_folder, 'uploads', 'branding')
os.makedirs(upload_dir, exist_ok=True)
# Save the file with a timestamp to avoid conflicts
import time
filename = f"favicon_{int(time.time())}_{favicon_file.filename}"
favicon_path = os.path.join(upload_dir, filename)
favicon_file.save(favicon_path)
branding.favicon_filename = filename
db.session.commit()
flash('Branding settings updated successfully.', 'success')
return redirect(url_for('system_admin.branding'))
# Get current branding settings
branding = BrandingSettings.get_current()
return render_template('system_admin_branding.html',
title='System Administrator - Branding Settings',
branding=branding)

128
routes/tasks.py Normal file
View File

@@ -0,0 +1,128 @@
"""
Task Management Routes
Handles all task-related views and operations
"""
from flask import Blueprint, render_template, g, redirect, url_for, flash
from sqlalchemy import or_
from models import db, Role, Project, Task, User
from routes.auth import login_required, role_required, company_required
tasks_bp = Blueprint('tasks', __name__, url_prefix='/tasks')
def get_filtered_tasks_for_burndown(user, mode, start_date=None, end_date=None, project_filter=None):
"""Get filtered tasks for burndown chart"""
from datetime import datetime, time
# Base query - get tasks from user's company
query = Task.query.join(Project).filter(Project.company_id == user.company_id)
# Apply user/team filter
if mode == 'personal':
# For personal mode, get tasks assigned to the user or created by them
query = query.filter(
(Task.assigned_to_id == user.id) |
(Task.created_by_id == user.id)
)
elif mode == 'team' and user.team_id:
# For team mode, get tasks from projects assigned to the team
query = query.filter(Project.team_id == user.team_id)
# Apply project filter
if project_filter:
if project_filter == 'none':
# No project filter for tasks - they must belong to a project
return []
else:
try:
project_id = int(project_filter)
query = query.filter(Task.project_id == project_id)
except ValueError:
pass
# Apply date filters - use task creation date and completion date
if start_date:
query = query.filter(
(Task.created_at >= datetime.combine(start_date, time.min)) |
(Task.completed_date >= start_date)
)
if end_date:
query = query.filter(
Task.created_at <= datetime.combine(end_date, time.max)
)
return query.order_by(Task.created_at.desc()).all()
@tasks_bp.route('')
@role_required(Role.TEAM_MEMBER)
@company_required
def unified_task_management():
"""Unified task management interface"""
# Get all projects the user has access to (for filtering and task creation)
if g.user.role in [Role.ADMIN, Role.SUPERVISOR]:
# Admins and Supervisors can see all company projects
available_projects = Project.query.filter_by(
company_id=g.user.company_id,
is_active=True
).order_by(Project.name).all()
elif g.user.team_id:
# Team members see team projects + unassigned projects
available_projects = Project.query.filter(
Project.company_id == g.user.company_id,
Project.is_active == True,
or_(Project.team_id == g.user.team_id, Project.team_id == None)
).order_by(Project.name).all()
# Filter by actual access permissions
available_projects = [p for p in available_projects if p.is_user_allowed(g.user)]
else:
# Unassigned users see only unassigned projects
available_projects = Project.query.filter_by(
company_id=g.user.company_id,
team_id=None,
is_active=True
).order_by(Project.name).all()
available_projects = [p for p in available_projects if p.is_user_allowed(g.user)]
# Get team members for task assignment (company-scoped)
if g.user.role in [Role.ADMIN, Role.SUPERVISOR]:
# Admins can assign to anyone in the company
team_members = User.query.filter_by(
company_id=g.user.company_id,
is_blocked=False
).order_by(User.username).all()
elif g.user.team_id:
# Team members can assign to team members + supervisors/admins
team_members = User.query.filter(
User.company_id == g.user.company_id,
User.is_blocked == False,
or_(
User.team_id == g.user.team_id,
User.role.in_([Role.ADMIN, Role.SUPERVISOR])
)
).order_by(User.username).all()
else:
# Unassigned users can assign to supervisors/admins only
team_members = User.query.filter(
User.company_id == g.user.company_id,
User.is_blocked == False,
User.role.in_([Role.ADMIN, Role.SUPERVISOR])
).order_by(User.username).all()
# Convert team members to JSON-serializable format
team_members_data = [{
'id': member.id,
'username': member.username,
'email': member.email,
'role': member.role.value if member.role else 'Team Member',
'avatar_url': member.get_avatar_url(32)
} for member in team_members]
return render_template('unified_task_management.html',
title='Task Management',
available_projects=available_projects,
team_members=team_members_data)

985
routes/tasks_api.py Normal file
View File

@@ -0,0 +1,985 @@
"""
Task Management API Routes
Handles all task-related API endpoints including subtasks and dependencies
"""
from flask import Blueprint, jsonify, request, g
from datetime import datetime
from models import (db, Role, Project, Task, User, TaskStatus, TaskPriority, SubTask,
TaskDependency, Sprint, CompanySettings, Comment, CommentVisibility)
from routes.auth import login_required, role_required, company_required
import logging
logger = logging.getLogger(__name__)
tasks_api_bp = Blueprint('tasks_api', __name__, url_prefix='/api')
@tasks_api_bp.route('/tasks', methods=['POST'])
@role_required(Role.TEAM_MEMBER)
@company_required
def create_task():
try:
data = request.get_json()
project_id = data.get('project_id')
# Verify project access
project = Project.query.filter_by(id=project_id, company_id=g.user.company_id).first()
if not project or not project.is_user_allowed(g.user):
return jsonify({'success': False, 'message': 'Project not found or access denied'})
# Validate required fields
name = data.get('name')
if not name:
return jsonify({'success': False, 'message': 'Task name is required'})
# Parse dates
start_date = None
due_date = None
if data.get('start_date'):
start_date = datetime.strptime(data.get('start_date'), '%Y-%m-%d').date()
if data.get('due_date'):
due_date = datetime.strptime(data.get('due_date'), '%Y-%m-%d').date()
# Generate task number
task_number = Task.generate_task_number(g.user.company_id)
# Create task
task = Task(
task_number=task_number,
name=name,
description=data.get('description', ''),
status=TaskStatus[data.get('status', 'TODO')],
priority=TaskPriority[data.get('priority', 'MEDIUM')],
estimated_hours=float(data.get('estimated_hours')) if data.get('estimated_hours') else None,
project_id=project_id,
assigned_to_id=int(data.get('assigned_to_id')) if data.get('assigned_to_id') else None,
sprint_id=int(data.get('sprint_id')) if data.get('sprint_id') else None,
start_date=start_date,
due_date=due_date,
created_by_id=g.user.id
)
db.session.add(task)
db.session.commit()
return jsonify({
'success': True,
'message': 'Task created successfully',
'task': {
'id': task.id,
'task_number': task.task_number
}
})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>', methods=['GET'])
@role_required(Role.TEAM_MEMBER)
@company_required
def get_task(task_id):
try:
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
task_data = {
'id': task.id,
'task_number': getattr(task, 'task_number', f'TSK-{task.id:03d}'),
'name': task.name,
'description': task.description,
'status': task.status.name,
'priority': task.priority.name,
'estimated_hours': task.estimated_hours,
'assigned_to_id': task.assigned_to_id,
'assigned_to_name': task.assigned_to.username if task.assigned_to else None,
'project_id': task.project_id,
'project_name': task.project.name if task.project else None,
'project_code': task.project.code if task.project else None,
'start_date': task.start_date.isoformat() if task.start_date else None,
'due_date': task.due_date.isoformat() if task.due_date else None,
'completed_date': task.completed_date.isoformat() if task.completed_date else None,
'archived_date': task.archived_date.isoformat() if task.archived_date else None,
'sprint_id': task.sprint_id,
'subtasks': [{
'id': subtask.id,
'name': subtask.name,
'status': subtask.status.name,
'priority': subtask.priority.name,
'assigned_to_id': subtask.assigned_to_id,
'assigned_to_name': subtask.assigned_to.username if subtask.assigned_to else None
} for subtask in task.subtasks] if task.subtasks else []
}
return jsonify(task_data)
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>', methods=['PUT'])
@role_required(Role.TEAM_MEMBER)
@company_required
def update_task(task_id):
try:
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
data = request.get_json()
# Update task fields
if 'name' in data:
task.name = data['name']
if 'description' in data:
task.description = data['description']
if 'status' in data:
task.status = TaskStatus[data['status']]
if data['status'] == 'COMPLETED':
task.completed_date = datetime.now().date()
else:
task.completed_date = None
if 'priority' in data:
task.priority = TaskPriority[data['priority']]
if 'estimated_hours' in data:
task.estimated_hours = float(data['estimated_hours']) if data['estimated_hours'] else None
if 'assigned_to_id' in data:
task.assigned_to_id = int(data['assigned_to_id']) if data['assigned_to_id'] else None
if 'sprint_id' in data:
task.sprint_id = int(data['sprint_id']) if data['sprint_id'] else None
if 'start_date' in data:
task.start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date() if data['start_date'] else None
if 'due_date' in data:
task.due_date = datetime.strptime(data['due_date'], '%Y-%m-%d').date() if data['due_date'] else None
db.session.commit()
return jsonify({'success': True, 'message': 'Task updated successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>', methods=['DELETE'])
@role_required(Role.TEAM_LEADER) # Only team leaders and above can delete tasks
@company_required
def delete_task(task_id):
try:
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
db.session.delete(task)
db.session.commit()
return jsonify({'success': True, 'message': 'Task deleted successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/unified')
@role_required(Role.TEAM_MEMBER)
@company_required
def get_unified_tasks():
"""Get all tasks for unified task view"""
try:
# Base query for tasks in user's company
query = Task.query.join(Project).filter(Project.company_id == g.user.company_id)
# Apply access restrictions based on user role and team
if g.user.role not in [Role.ADMIN, Role.SUPERVISOR]:
# Regular users can only see tasks from projects they have access to
accessible_project_ids = []
projects = Project.query.filter_by(company_id=g.user.company_id).all()
for project in projects:
if project.is_user_allowed(g.user):
accessible_project_ids.append(project.id)
if accessible_project_ids:
query = query.filter(Task.project_id.in_(accessible_project_ids))
else:
# No accessible projects, return empty list
return jsonify({'success': True, 'tasks': []})
tasks = query.order_by(Task.created_at.desc()).all()
task_list = []
for task in tasks:
# Determine if this is a team task
is_team_task = (
g.user.team_id and
task.project and
task.project.team_id == g.user.team_id
)
task_data = {
'id': task.id,
'task_number': getattr(task, 'task_number', f'TSK-{task.id:03d}'), # Fallback for existing tasks
'name': task.name,
'description': task.description,
'status': task.status.name,
'priority': task.priority.name,
'estimated_hours': task.estimated_hours,
'project_id': task.project_id,
'project_name': task.project.name if task.project else None,
'project_code': task.project.code if task.project else None,
'assigned_to_id': task.assigned_to_id,
'assigned_to_name': task.assigned_to.username if task.assigned_to else None,
'created_by_id': task.created_by_id,
'created_by_name': task.created_by.username if task.created_by else None,
'start_date': task.start_date.isoformat() if task.start_date else None,
'due_date': task.due_date.isoformat() if task.due_date else None,
'completed_date': task.completed_date.isoformat() if task.completed_date else None,
'created_at': task.created_at.isoformat(),
'is_team_task': is_team_task,
'subtask_count': len(task.subtasks) if task.subtasks else 0,
'subtasks': [{
'id': subtask.id,
'name': subtask.name,
'status': subtask.status.name,
'priority': subtask.priority.name,
'assigned_to_id': subtask.assigned_to_id,
'assigned_to_name': subtask.assigned_to.username if subtask.assigned_to else None
} for subtask in task.subtasks] if task.subtasks else [],
'sprint_id': task.sprint_id,
'sprint_name': task.sprint.name if task.sprint else None,
'is_current_sprint': task.sprint.is_current if task.sprint else False
}
task_list.append(task_data)
return jsonify({'success': True, 'tasks': task_list})
except Exception as e:
logger.error(f"Error in get_unified_tasks: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>/status', methods=['PUT'])
@role_required(Role.TEAM_MEMBER)
@company_required
def update_task_status(task_id):
"""Update task status"""
try:
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
data = request.get_json()
new_status = data.get('status')
if not new_status:
return jsonify({'success': False, 'message': 'Status is required'})
# Validate status value - convert from enum name to enum object
try:
task_status = TaskStatus[new_status]
except KeyError:
return jsonify({'success': False, 'message': 'Invalid status value'})
# Update task status
old_status = task.status
task.status = task_status
# Set completion date if status is DONE
if task_status == TaskStatus.DONE:
task.completed_date = datetime.now().date()
elif old_status == TaskStatus.DONE:
# Clear completion date if moving away from done
task.completed_date = None
# Set archived date if status is ARCHIVED
if task_status == TaskStatus.ARCHIVED:
task.archived_date = datetime.now().date()
elif old_status == TaskStatus.ARCHIVED:
# Clear archived date if moving away from archived
task.archived_date = None
db.session.commit()
return jsonify({
'success': True,
'message': 'Task status updated successfully',
'old_status': old_status.name,
'new_status': task_status.name
})
except Exception as e:
db.session.rollback()
logger.error(f"Error updating task status: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
# Task Dependencies APIs
@tasks_api_bp.route('/tasks/<int:task_id>/dependencies')
@role_required(Role.TEAM_MEMBER)
@company_required
def get_task_dependencies(task_id):
"""Get dependencies for a specific task"""
try:
# Get the task and verify ownership through project
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Get blocked by dependencies (tasks that block this one)
blocked_by_query = db.session.query(Task).join(
TaskDependency, Task.id == TaskDependency.blocking_task_id
).filter(TaskDependency.blocked_task_id == task_id)
# Get blocks dependencies (tasks that this one blocks)
blocks_query = db.session.query(Task).join(
TaskDependency, Task.id == TaskDependency.blocked_task_id
).filter(TaskDependency.blocking_task_id == task_id)
blocked_by_tasks = blocked_by_query.all()
blocks_tasks = blocks_query.all()
def task_to_dict(t):
return {
'id': t.id,
'name': t.name,
'task_number': t.task_number
}
return jsonify({
'success': True,
'dependencies': {
'blocked_by': [task_to_dict(t) for t in blocked_by_tasks],
'blocks': [task_to_dict(t) for t in blocks_tasks]
}
})
except Exception as e:
logger.error(f"Error getting task dependencies: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>/dependencies', methods=['POST'])
@role_required(Role.TEAM_MEMBER)
@company_required
def add_task_dependency(task_id):
"""Add a dependency for a task"""
try:
data = request.get_json()
task_number = data.get('task_number')
dependency_type = data.get('type') # 'blocked_by' or 'blocks'
if not task_number or not dependency_type:
return jsonify({'success': False, 'message': 'Task number and type are required'})
# Get the main task
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Find the dependency task by task number
dependency_task = Task.query.join(Project).filter(
Task.task_number == task_number,
Project.company_id == g.user.company_id
).first()
if not dependency_task:
return jsonify({'success': False, 'message': f'Task {task_number} not found'})
# Prevent self-dependency
if dependency_task.id == task_id:
return jsonify({'success': False, 'message': 'A task cannot depend on itself'})
# Create the dependency based on type
if dependency_type == 'blocked_by':
# Current task is blocked by the dependency task
blocked_task_id = task_id
blocking_task_id = dependency_task.id
elif dependency_type == 'blocks':
# Current task blocks the dependency task
blocked_task_id = dependency_task.id
blocking_task_id = task_id
else:
return jsonify({'success': False, 'message': 'Invalid dependency type'})
# Check if dependency already exists
existing_dep = TaskDependency.query.filter_by(
blocked_task_id=blocked_task_id,
blocking_task_id=blocking_task_id
).first()
if existing_dep:
return jsonify({'success': False, 'message': 'This dependency already exists'})
# Check for circular dependencies
def would_create_cycle(blocked_id, blocking_id):
# Use a simple DFS to check if adding this dependency would create a cycle
visited = set()
def dfs(current_blocked_id):
if current_blocked_id in visited:
return False
visited.add(current_blocked_id)
# If we reach the original blocking task, we have a cycle
if current_blocked_id == blocking_id:
return True
# Check all tasks that block the current task
dependencies = TaskDependency.query.filter_by(blocked_task_id=current_blocked_id).all()
for dep in dependencies:
if dfs(dep.blocking_task_id):
return True
return False
return dfs(blocked_id)
if would_create_cycle(blocked_task_id, blocking_task_id):
return jsonify({'success': False, 'message': 'This dependency would create a circular dependency'})
# Create the new dependency
new_dependency = TaskDependency(
blocked_task_id=blocked_task_id,
blocking_task_id=blocking_task_id
)
db.session.add(new_dependency)
db.session.commit()
return jsonify({'success': True, 'message': 'Dependency added successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error adding task dependency: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>/dependencies/<int:dependency_task_id>', methods=['DELETE'])
@role_required(Role.TEAM_MEMBER)
@company_required
def remove_task_dependency(task_id, dependency_task_id):
"""Remove a dependency for a task"""
try:
data = request.get_json()
dependency_type = data.get('type') # 'blocked_by' or 'blocks'
if not dependency_type:
return jsonify({'success': False, 'message': 'Dependency type is required'})
# Get the main task
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Determine which dependency to remove based on type
if dependency_type == 'blocked_by':
# Remove dependency where current task is blocked by dependency_task_id
dependency = TaskDependency.query.filter_by(
blocked_task_id=task_id,
blocking_task_id=dependency_task_id
).first()
elif dependency_type == 'blocks':
# Remove dependency where current task blocks dependency_task_id
dependency = TaskDependency.query.filter_by(
blocked_task_id=dependency_task_id,
blocking_task_id=task_id
).first()
else:
return jsonify({'success': False, 'message': 'Invalid dependency type'})
if not dependency:
return jsonify({'success': False, 'message': 'Dependency not found'})
db.session.delete(dependency)
db.session.commit()
return jsonify({'success': True, 'message': 'Dependency removed successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error removing task dependency: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
# Task Archive/Restore APIs
@tasks_api_bp.route('/tasks/<int:task_id>/archive', methods=['POST'])
@role_required(Role.TEAM_MEMBER)
@company_required
def archive_task(task_id):
"""Archive a completed task"""
try:
# Get the task and verify ownership through project
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Only allow archiving completed tasks
if task.status != TaskStatus.COMPLETED:
return jsonify({'success': False, 'message': 'Only completed tasks can be archived'})
# Archive the task
task.status = TaskStatus.ARCHIVED
task.archived_date = datetime.now().date()
db.session.commit()
return jsonify({
'success': True,
'message': 'Task archived successfully',
'archived_date': task.archived_date.isoformat()
})
except Exception as e:
db.session.rollback()
logger.error(f"Error archiving task: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/tasks/<int:task_id>/restore', methods=['POST'])
@role_required(Role.TEAM_MEMBER)
@company_required
def restore_task(task_id):
"""Restore an archived task to completed status"""
try:
# Get the task and verify ownership through project
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Only allow restoring archived tasks
if task.status != TaskStatus.ARCHIVED:
return jsonify({'success': False, 'message': 'Only archived tasks can be restored'})
# Restore the task to completed status
task.status = TaskStatus.COMPLETED
task.archived_date = None
db.session.commit()
return jsonify({
'success': True,
'message': 'Task restored successfully'
})
except Exception as e:
db.session.rollback()
logger.error(f"Error restoring task: {str(e)}")
return jsonify({'success': False, 'message': str(e)})
# Subtask API Routes
@tasks_api_bp.route('/subtasks', methods=['POST'])
@role_required(Role.TEAM_MEMBER)
@company_required
def create_subtask():
try:
data = request.get_json()
task_id = data.get('task_id')
# Verify task access
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
# Validate required fields
name = data.get('name')
if not name:
return jsonify({'success': False, 'message': 'Subtask name is required'})
# Parse dates
start_date = None
due_date = None
if data.get('start_date'):
start_date = datetime.strptime(data.get('start_date'), '%Y-%m-%d').date()
if data.get('due_date'):
due_date = datetime.strptime(data.get('due_date'), '%Y-%m-%d').date()
# Create subtask
subtask = SubTask(
name=name,
description=data.get('description', ''),
status=TaskStatus[data.get('status', 'TODO')],
priority=TaskPriority[data.get('priority', 'MEDIUM')],
estimated_hours=float(data.get('estimated_hours')) if data.get('estimated_hours') else None,
task_id=task_id,
assigned_to_id=int(data.get('assigned_to_id')) if data.get('assigned_to_id') else None,
start_date=start_date,
due_date=due_date,
created_by_id=g.user.id
)
db.session.add(subtask)
db.session.commit()
return jsonify({'success': True, 'message': 'Subtask created successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/subtasks/<int:subtask_id>', methods=['GET'])
@role_required(Role.TEAM_MEMBER)
@company_required
def get_subtask(subtask_id):
try:
subtask = SubTask.query.join(Task).join(Project).filter(
SubTask.id == subtask_id,
Project.company_id == g.user.company_id
).first()
if not subtask or not subtask.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Subtask not found or access denied'})
subtask_data = {
'id': subtask.id,
'name': subtask.name,
'description': subtask.description,
'status': subtask.status.name,
'priority': subtask.priority.name,
'estimated_hours': subtask.estimated_hours,
'assigned_to_id': subtask.assigned_to_id,
'start_date': subtask.start_date.isoformat() if subtask.start_date else None,
'due_date': subtask.due_date.isoformat() if subtask.due_date else None
}
return jsonify({'success': True, 'subtask': subtask_data})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/subtasks/<int:subtask_id>', methods=['PUT'])
@role_required(Role.TEAM_MEMBER)
@company_required
def update_subtask(subtask_id):
try:
subtask = SubTask.query.join(Task).join(Project).filter(
SubTask.id == subtask_id,
Project.company_id == g.user.company_id
).first()
if not subtask or not subtask.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Subtask not found or access denied'})
data = request.get_json()
# Update subtask fields
if 'name' in data:
subtask.name = data['name']
if 'description' in data:
subtask.description = data['description']
if 'status' in data:
subtask.status = TaskStatus[data['status']]
if data['status'] == 'COMPLETED':
subtask.completed_date = datetime.now().date()
else:
subtask.completed_date = None
if 'priority' in data:
subtask.priority = TaskPriority[data['priority']]
if 'estimated_hours' in data:
subtask.estimated_hours = float(data['estimated_hours']) if data['estimated_hours'] else None
if 'assigned_to_id' in data:
subtask.assigned_to_id = int(data['assigned_to_id']) if data['assigned_to_id'] else None
if 'start_date' in data:
subtask.start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date() if data['start_date'] else None
if 'due_date' in data:
subtask.due_date = datetime.strptime(data['due_date'], '%Y-%m-%d').date() if data['due_date'] else None
db.session.commit()
return jsonify({'success': True, 'message': 'Subtask updated successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
@tasks_api_bp.route('/subtasks/<int:subtask_id>', methods=['DELETE'])
@role_required(Role.TEAM_LEADER) # Only team leaders and above can delete subtasks
@company_required
def delete_subtask(subtask_id):
try:
subtask = SubTask.query.join(Task).join(Project).filter(
SubTask.id == subtask_id,
Project.company_id == g.user.company_id
).first()
if not subtask or not subtask.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Subtask not found or access denied'})
db.session.delete(subtask)
db.session.commit()
return jsonify({'success': True, 'message': 'Subtask deleted successfully'})
except Exception as e:
db.session.rollback()
return jsonify({'success': False, 'message': str(e)})
# Comment API Routes
@tasks_api_bp.route('/tasks/<int:task_id>/comments', methods=['GET', 'POST'])
@login_required
@company_required
def handle_task_comments(task_id):
"""Handle GET and POST requests for task comments"""
if request.method == 'GET':
return get_task_comments(task_id)
else: # POST
return create_task_comment(task_id)
@tasks_api_bp.route('/comments/<int:comment_id>', methods=['PUT', 'DELETE'])
@login_required
@company_required
def handle_comment(comment_id):
"""Handle PUT and DELETE requests for a specific comment"""
if request.method == 'DELETE':
return delete_comment(comment_id)
else: # PUT
return update_comment(comment_id)
def delete_comment(comment_id):
"""Delete a comment"""
try:
comment = Comment.query.join(Task).join(Project).filter(
Comment.id == comment_id,
Project.company_id == g.user.company_id
).first()
if not comment:
return jsonify({'success': False, 'message': 'Comment not found'}), 404
# Check if user can delete this comment
if not comment.can_user_delete(g.user):
return jsonify({'success': False, 'message': 'You do not have permission to delete this comment'}), 403
# Delete the comment (replies will be cascade deleted)
db.session.delete(comment)
db.session.commit()
return jsonify({'success': True, 'message': 'Comment deleted successfully'})
except Exception as e:
db.session.rollback()
logger.error(f"Error deleting comment {comment_id}: {e}")
return jsonify({'success': False, 'message': 'Failed to delete comment'}), 500
def update_comment(comment_id):
"""Update a comment"""
try:
comment = Comment.query.join(Task).join(Project).filter(
Comment.id == comment_id,
Project.company_id == g.user.company_id
).first()
if not comment:
return jsonify({'success': False, 'message': 'Comment not found'}), 404
# Check if user can edit this comment
if not comment.can_user_edit(g.user):
return jsonify({'success': False, 'message': 'You do not have permission to edit this comment'}), 403
data = request.json
new_content = data.get('content', '').strip()
if not new_content:
return jsonify({'success': False, 'message': 'Comment content is required'}), 400
# Update the comment
comment.content = new_content
comment.is_edited = True
comment.edited_at = datetime.now()
db.session.commit()
return jsonify({
'success': True,
'message': 'Comment updated successfully',
'comment': {
'id': comment.id,
'content': comment.content,
'is_edited': comment.is_edited,
'edited_at': comment.edited_at.isoformat()
}
})
except Exception as e:
db.session.rollback()
logger.error(f"Error updating comment {comment_id}: {e}")
return jsonify({'success': False, 'message': 'Failed to update comment'}), 500
def get_task_comments(task_id):
"""Get all comments for a task that the user can view"""
try:
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task or not task.can_user_access(g.user):
return jsonify({'success': False, 'message': 'Task not found or access denied'})
# Get all comments for the task
comments = []
for comment in task.comments.order_by(Comment.created_at.desc()):
if comment.can_user_view(g.user):
comment_data = {
'id': comment.id,
'content': comment.content,
'visibility': comment.visibility.value,
'is_edited': comment.is_edited,
'edited_at': comment.edited_at.isoformat() if comment.edited_at else None,
'created_at': comment.created_at.isoformat(),
'author': {
'id': comment.created_by.id,
'username': comment.created_by.username,
'avatar_url': comment.created_by.get_avatar_url(40)
},
'can_edit': comment.can_user_edit(g.user),
'can_delete': comment.can_user_delete(g.user),
'replies': []
}
# Add replies if any
for reply in comment.replies:
if reply.can_user_view(g.user):
reply_data = {
'id': reply.id,
'content': reply.content,
'is_edited': reply.is_edited,
'edited_at': reply.edited_at.isoformat() if reply.edited_at else None,
'created_at': reply.created_at.isoformat(),
'author': {
'id': reply.created_by.id,
'username': reply.created_by.username,
'avatar_url': reply.created_by.get_avatar_url(40)
},
'can_edit': reply.can_user_edit(g.user),
'can_delete': reply.can_user_delete(g.user)
}
comment_data['replies'].append(reply_data)
comments.append(comment_data)
# Check if user can use team visibility
company_settings = CompanySettings.query.filter_by(company_id=g.user.company_id).first()
allow_team_visibility = company_settings.allow_team_visibility_comments if company_settings else True
return jsonify({
'success': True,
'comments': comments,
'allow_team_visibility': allow_team_visibility
})
except Exception as e:
return jsonify({'success': False, 'message': str(e)})
def create_task_comment(task_id):
"""Create a new comment on a task"""
try:
# Get the task and verify access through project
task = Task.query.join(Project).filter(
Task.id == task_id,
Project.company_id == g.user.company_id
).first()
if not task:
return jsonify({'success': False, 'message': 'Task not found'})
# Check if user has access to this task's project
if not task.project.is_user_allowed(g.user):
return jsonify({'success': False, 'message': 'Access denied'})
data = request.json
content = data.get('content', '').strip()
visibility = data.get('visibility', CommentVisibility.COMPANY.value)
if not content:
return jsonify({'success': False, 'message': 'Comment content is required'})
# Validate visibility - handle case conversion
try:
# Convert from frontend format (TEAM/COMPANY) to enum format (Team/Company)
if visibility == 'TEAM':
visibility_enum = CommentVisibility.TEAM
elif visibility == 'COMPANY':
visibility_enum = CommentVisibility.COMPANY
else:
# Try to use the value directly in case it's already in the right format
visibility_enum = CommentVisibility(visibility)
except ValueError:
return jsonify({'success': False, 'message': f'Invalid visibility option: {visibility}'})
# Create the comment
comment = Comment(
content=content,
task_id=task_id,
created_by_id=g.user.id,
visibility=visibility_enum
)
db.session.add(comment)
db.session.commit()
return jsonify({
'success': True,
'message': 'Comment added successfully',
'comment': {
'id': comment.id,
'content': comment.content,
'visibility': comment.visibility.value,
'created_at': comment.created_at.isoformat(),
'user': {
'id': comment.created_by.id,
'username': comment.created_by.username
}
}
})
except Exception as e:
db.session.rollback()
logger.error(f"Error creating task comment: {str(e)}")
return jsonify({'success': False, 'message': str(e)})

191
routes/teams.py Normal file
View File

@@ -0,0 +1,191 @@
"""
Team Management Routes
Handles all team-related views and operations
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, abort
from models import db, Team, User
from routes.auth import admin_required, company_required
from utils.validation import FormValidator
from utils.repository import TeamRepository
teams_bp = Blueprint('teams', __name__, url_prefix='/admin/teams')
@teams_bp.route('')
@admin_required
@company_required
def admin_teams():
team_repo = TeamRepository()
teams = team_repo.get_with_member_count(g.user.company_id)
return render_template('admin_teams.html', title='Team Management', teams=teams)
@teams_bp.route('/create', methods=['GET', 'POST'])
@admin_required
@company_required
def create_team():
if request.method == 'POST':
validator = FormValidator()
team_repo = TeamRepository()
name = request.form.get('name')
description = request.form.get('description')
# Validate input
validator.validate_required(name, 'Team name')
if validator.is_valid() and team_repo.exists_by_name_in_company(name, g.user.company_id):
validator.errors.add('Team name already exists in your company')
if validator.is_valid():
new_team = team_repo.create(
name=name,
description=description,
company_id=g.user.company_id
)
team_repo.save()
flash(f'Team "{name}" created successfully!', 'success')
return redirect(url_for('teams.admin_teams'))
validator.flash_errors()
return render_template('team_form.html', title='Create Team', team=None)
@teams_bp.route('/edit/<int:team_id>', methods=['GET', 'POST'])
@admin_required
@company_required
def edit_team(team_id):
team_repo = TeamRepository()
team = team_repo.get_by_id_and_company(team_id, g.user.company_id)
if not team:
abort(404)
if request.method == 'POST':
validator = FormValidator()
name = request.form.get('name')
description = request.form.get('description')
# Validate input
validator.validate_required(name, 'Team name')
if validator.is_valid() and name != team.name:
if team_repo.exists_by_name_in_company(name, g.user.company_id):
validator.errors.add('Team name already exists in your company')
if validator.is_valid():
team_repo.update(team, name=name, description=description)
team_repo.save()
flash(f'Team "{name}" updated successfully!', 'success')
return redirect(url_for('teams.admin_teams'))
validator.flash_errors()
return render_template('edit_team.html', title='Edit Team', team=team)
@teams_bp.route('/delete/<int:team_id>', methods=['POST'])
@admin_required
@company_required
def delete_team(team_id):
team_repo = TeamRepository()
team = team_repo.get_by_id_and_company(team_id, g.user.company_id)
if not team:
abort(404)
# Check if team has members
if team.users:
flash('Cannot delete team with members. Remove all members first.', 'error')
return redirect(url_for('teams.admin_teams'))
team_name = team.name
team_repo.delete(team)
team_repo.save()
flash(f'Team "{team_name}" deleted successfully!', 'success')
return redirect(url_for('teams.admin_teams'))
@teams_bp.route('/<int:team_id>', methods=['GET', 'POST'])
@admin_required
@company_required
def manage_team(team_id):
team_repo = TeamRepository()
team = team_repo.get_by_id_and_company(team_id, g.user.company_id)
if not team:
abort(404)
if request.method == 'POST':
action = request.form.get('action')
if action == 'update_team':
# Update team details
validator = FormValidator()
name = request.form.get('name')
description = request.form.get('description')
# Validate input
validator.validate_required(name, 'Team name')
if validator.is_valid() and name != team.name:
if team_repo.exists_by_name_in_company(name, g.user.company_id):
validator.errors.add('Team name already exists in your company')
if validator.is_valid():
team_repo.update(team, name=name, description=description)
team_repo.save()
flash(f'Team "{name}" updated successfully!', 'success')
else:
validator.flash_errors()
elif action == 'add_member':
# Add user to team
user_id = request.form.get('user_id')
if user_id:
user = User.query.get(user_id)
if user:
user.team_id = team.id
db.session.commit()
flash(f'User {user.username} added to team!', 'success')
else:
flash('User not found', 'error')
else:
flash('No user selected', 'error')
elif action == 'remove_member':
# Remove user from team
user_id = request.form.get('user_id')
if user_id:
user = User.query.get(user_id)
if user and user.team_id == team.id:
user.team_id = None
db.session.commit()
flash(f'User {user.username} removed from team!', 'success')
else:
flash('User not found or not in this team', 'error')
else:
flash('No user selected', 'error')
# Get team members
team_members = User.query.filter_by(team_id=team.id).all()
# Get users not in this team for the add member form (company-scoped)
available_users = User.query.filter(
User.company_id == g.user.company_id,
(User.team_id != team.id) | (User.team_id == None)
).all()
return render_template(
'team_form.html',
title=f'Manage Team: {team.name}',
team=team,
team_members=team_members,
available_users=available_users
)

124
routes/teams_api.py Normal file
View File

@@ -0,0 +1,124 @@
"""
Team Management API Routes
Handles all team-related API endpoints
"""
from flask import Blueprint, jsonify, request, g
from datetime import datetime, time, timedelta
from models import Team, User, TimeEntry, Role
from routes.auth import login_required, role_required, company_required, system_admin_required
teams_api_bp = Blueprint('teams_api', __name__, url_prefix='/api')
@teams_api_bp.route('/team/hours_data', methods=['GET'])
@login_required
@role_required(Role.TEAM_LEADER) # Only team leaders and above can access
@company_required
def team_hours_data():
# Get the current user's team
team = Team.query.get(g.user.team_id)
if not team:
return jsonify({
'success': False,
'message': 'You are not assigned to any team.'
}), 400
# Get date range from query parameters or use current week as default
today = datetime.now().date()
start_of_week = today - timedelta(days=today.weekday())
end_of_week = start_of_week + timedelta(days=6)
start_date_str = request.args.get('start_date', start_of_week.strftime('%Y-%m-%d'))
end_date_str = request.args.get('end_date', end_of_week.strftime('%Y-%m-%d'))
include_self = request.args.get('include_self', 'false') == 'true'
try:
start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date()
end_date = datetime.strptime(end_date_str, '%Y-%m-%d').date()
except ValueError:
return jsonify({
'success': False,
'message': 'Invalid date format.'
}), 400
# Get all team members
team_members = User.query.filter_by(team_id=team.id).all()
# Prepare data structure for team members' hours
team_data = []
for member in team_members:
# Skip if the member is the current user (team leader) and include_self is False
if member.id == g.user.id and not include_self:
continue
# Get time entries for this member in the date range
entries = TimeEntry.query.filter(
TimeEntry.user_id == member.id,
TimeEntry.arrival_time >= datetime.combine(start_date, time.min),
TimeEntry.arrival_time <= datetime.combine(end_date, time.max)
).order_by(TimeEntry.arrival_time).all()
# Calculate daily and total hours
daily_hours = {}
total_seconds = 0
for entry in entries:
if entry.duration: # Only count completed entries
entry_date = entry.arrival_time.date()
date_str = entry_date.strftime('%Y-%m-%d')
if date_str not in daily_hours:
daily_hours[date_str] = 0
daily_hours[date_str] += entry.duration
total_seconds += entry.duration
# Convert seconds to hours for display
for date_str in daily_hours:
daily_hours[date_str] = round(daily_hours[date_str] / 3600, 2) # Convert to hours
total_hours = round(total_seconds / 3600, 2) # Convert to hours
# Format entries for JSON response
formatted_entries = []
for entry in entries:
formatted_entries.append({
'id': entry.id,
'arrival_time': entry.arrival_time.isoformat(),
'departure_time': entry.departure_time.isoformat() if entry.departure_time else None,
'duration': entry.duration,
'total_break_duration': entry.total_break_duration
})
# Add member data to team data
team_data.append({
'member_id': member.id,
'member_name': member.username,
'daily_hours': daily_hours,
'total_hours': total_hours,
'entries': formatted_entries
})
return jsonify({
'success': True,
'team_name': team.name,
'team_id': team.id,
'start_date': start_date_str,
'end_date': end_date_str,
'team_data': team_data
})
@teams_api_bp.route('/companies/<int:company_id>/teams')
@system_admin_required
def api_company_teams(company_id):
"""API: Get teams for a specific company (System Admin only)"""
teams = Team.query.filter_by(company_id=company_id).order_by(Team.name).all()
return jsonify([{
'id': team.id,
'name': team.name,
'description': team.description
} for team in teams])

532
routes/users.py Normal file
View File

@@ -0,0 +1,532 @@
"""
User management routes
"""
from flask import Blueprint, render_template, request, redirect, url_for, flash, g, session, abort
from models import db, User, Role, Team, TimeEntry, WorkConfig, UserPreferences, Project, Task, SubTask, ProjectCategory, UserDashboard, Comment, Company
from routes.auth import admin_required, company_required, login_required, system_admin_required
from flask_mail import Message
from flask import current_app
from utils.validation import FormValidator
from utils.repository import UserRepository
import logging
logger = logging.getLogger(__name__)
users_bp = Blueprint('users', __name__, url_prefix='/admin/users')
def get_available_roles():
"""Get roles available for user creation/editing based on current user's role"""
current_user_role = g.user.role
if current_user_role == Role.SYSTEM_ADMIN:
# System admin can assign any role
return [Role.TEAM_MEMBER, Role.TEAM_LEADER, Role.SUPERVISOR, Role.ADMIN, Role.SYSTEM_ADMIN]
elif current_user_role == Role.ADMIN:
# Admin can assign any role except system admin
return [Role.TEAM_MEMBER, Role.TEAM_LEADER, Role.SUPERVISOR, Role.ADMIN]
elif current_user_role == Role.SUPERVISOR:
# Supervisor can only assign team member and team leader roles
return [Role.TEAM_MEMBER, Role.TEAM_LEADER]
else:
# Others cannot assign roles
return []
@users_bp.route('')
@admin_required
@company_required
def admin_users():
user_repo = UserRepository()
users = user_repo.get_by_company(g.user.company_id)
return render_template('admin_users.html', title='User Management', users=users)
@users_bp.route('/create', methods=['GET', 'POST'])
@admin_required
@company_required
def create_user():
if request.method == 'POST':
validator = FormValidator()
user_repo = UserRepository()
username = request.form.get('username')
email = request.form.get('email')
password = request.form.get('password')
auto_verify = 'auto_verify' in request.form
# Get role and team
role_name = request.form.get('role')
team_id = request.form.get('team_id')
# Validate required fields
validator.validate_required(username, 'Username')
validator.validate_required(email, 'Email')
validator.validate_required(password, 'Password')
# Validate uniqueness
if validator.is_valid():
validator.validate_unique(User, 'username', username, company_id=g.user.company_id)
validator.validate_unique(User, 'email', email, company_id=g.user.company_id)
if validator.is_valid():
# Convert role string to enum
try:
role = Role[role_name] if role_name else Role.TEAM_MEMBER
except KeyError:
role = Role.TEAM_MEMBER
# Create new user with role and team
new_user = user_repo.create(
username=username,
email=email,
company_id=g.user.company_id,
is_verified=auto_verify,
role=role,
team_id=team_id if team_id else None
)
new_user.set_password(password)
if not auto_verify:
# Generate verification token and send email
token = new_user.generate_verification_token()
verification_url = url_for('verify_email', token=token, _external=True)
try:
from flask_mail import Mail
mail = Mail(current_app)
# Get branding for email
from models import BrandingSettings
branding = BrandingSettings.get_settings()
msg = Message(
f'Welcome to {branding.app_name} - Verify Your Email',
sender=(branding.app_name, current_app.config['MAIL_USERNAME']),
recipients=[email]
)
msg.body = f'''Welcome to {branding.app_name}!
Your administrator has created an account for you. Please verify your email address to activate your account.
Username: {username}
Company: {g.company.name}
Click the link below to verify your email:
{verification_url}
This link will expire in 24 hours.
If you did not expect this email, please ignore it.
Best regards,
The {branding.app_name} Team
'''
mail.send(msg)
logger.info(f"Verification email sent to {email}")
except Exception as e:
logger.error(f"Failed to send verification email: {str(e)}")
flash('User created but verification email could not be sent. Please contact the user directly.', 'warning')
user_repo.save()
flash(f'User {username} created successfully!', 'success')
return redirect(url_for('users.admin_users'))
validator.flash_errors()
# Get all teams for the form (company-scoped)
teams = Team.query.filter_by(company_id=g.user.company_id).all()
roles = get_available_roles()
return render_template('create_user.html', title='Create User', teams=teams, roles=roles)
@users_bp.route('/edit/<int:user_id>', methods=['GET', 'POST'])
@admin_required
@company_required
def edit_user(user_id):
user_repo = UserRepository()
user = user_repo.get_by_id_and_company(user_id, g.user.company_id)
if not user:
abort(404)
if request.method == 'POST':
validator = FormValidator()
username = request.form.get('username')
email = request.form.get('email')
password = request.form.get('password')
# Get role and team
role_name = request.form.get('role')
team_id = request.form.get('team_id')
# Validate required fields
validator.validate_required(username, 'Username')
validator.validate_required(email, 'Email')
# Validate uniqueness (exclude current user)
if validator.is_valid():
if username != user.username:
validator.validate_unique(User, 'username', username, company_id=g.user.company_id)
if email != user.email:
validator.validate_unique(User, 'email', email, company_id=g.user.company_id)
if validator.is_valid():
# Convert role string to enum
try:
role = Role[role_name] if role_name else Role.TEAM_MEMBER
except KeyError:
role = Role.TEAM_MEMBER
user_repo.update(user,
username=username,
email=email,
role=role,
team_id=team_id if team_id else None
)
if password:
user.set_password(password)
user_repo.save()
flash(f'User {username} updated successfully!', 'success')
return redirect(url_for('users.admin_users'))
validator.flash_errors()
# Get all teams for the form (company-scoped)
teams = Team.query.filter_by(company_id=g.user.company_id).all()
roles = get_available_roles()
return render_template('edit_user.html', title='Edit User', user=user, teams=teams, roles=roles)
@users_bp.route('/delete/<int:user_id>', methods=['POST'])
@admin_required
@company_required
def delete_user(user_id):
user = User.query.filter_by(id=user_id, company_id=g.user.company_id).first_or_404()
# Prevent deleting yourself
if user.id == session.get('user_id'):
flash('You cannot delete your own account', 'error')
return redirect(url_for('users.admin_users'))
username = user.username
try:
# Check if user owns any critical resources
owns_projects = Project.query.filter_by(created_by_id=user_id).count() > 0
owns_tasks = Task.query.filter_by(created_by_id=user_id).count() > 0
owns_subtasks = SubTask.query.filter_by(created_by_id=user_id).count() > 0
needs_ownership_transfer = owns_projects or owns_tasks or owns_subtasks
if needs_ownership_transfer:
# Find an alternative admin/supervisor to transfer ownership to
alternative_admin = User.query.filter(
User.company_id == g.user.company_id,
User.role.in_([Role.ADMIN, Role.SUPERVISOR]),
User.id != user_id
).first()
if alternative_admin:
# Transfer ownership of projects to alternative admin
if owns_projects:
Project.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
# Transfer ownership of tasks to alternative admin
if owns_tasks:
Task.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
# Transfer ownership of subtasks to alternative admin
if owns_subtasks:
SubTask.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
else:
# No alternative admin found but user owns resources
flash('Cannot delete this user. They own resources but no other administrator or supervisor found to transfer ownership to.', 'error')
return redirect(url_for('users.admin_users'))
# Delete user-specific records that can be safely removed
TimeEntry.query.filter_by(user_id=user_id).delete()
WorkConfig.query.filter_by(user_id=user_id).delete()
UserPreferences.query.filter_by(user_id=user_id).delete()
# Delete user dashboards (cascades to widgets)
UserDashboard.query.filter_by(user_id=user_id).delete()
# Clear task and subtask assignments
Task.query.filter_by(assigned_to_id=user_id).update({'assigned_to_id': None})
SubTask.query.filter_by(assigned_to_id=user_id).update({'assigned_to_id': None})
# Now safe to delete the user
db.session.delete(user)
db.session.commit()
if needs_ownership_transfer and alternative_admin:
flash(f'User {username} deleted successfully. Projects and tasks transferred to {alternative_admin.username}', 'success')
else:
flash(f'User {username} deleted successfully.', 'success')
except Exception as e:
db.session.rollback()
logger.error(f"Error deleting user {user_id}: {str(e)}")
flash(f'Error deleting user: {str(e)}', 'error')
return redirect(url_for('users.admin_users'))
@users_bp.route('/toggle-status/<int:user_id>', methods=['POST'])
@admin_required
@company_required
def toggle_user_status(user_id):
"""Toggle user active/blocked status"""
user = User.query.filter_by(id=user_id, company_id=g.user.company_id).first_or_404()
# Prevent blocking yourself
if user.id == g.user.id:
flash('You cannot block your own account', 'error')
return redirect(url_for('users.admin_users'))
# Toggle the blocked status
user.is_blocked = not user.is_blocked
db.session.commit()
status = 'blocked' if user.is_blocked else 'unblocked'
flash(f'User {user.username} has been {status}', 'success')
return redirect(url_for('users.admin_users'))
# System Admin User Routes
@users_bp.route('/system-admin')
@system_admin_required
def system_admin_users():
"""System admin view of all users across all companies"""
# Get filter parameters
company_id = request.args.get('company_id', type=int)
search_query = request.args.get('search', '')
filter_type = request.args.get('filter', '')
page = request.args.get('page', 1, type=int)
per_page = 50
# Build query that returns tuples of (User, company_name)
query = db.session.query(User, Company.name).join(Company)
# Apply company filter
if company_id:
query = query.filter(User.company_id == company_id)
# Apply search filter
if search_query:
query = query.filter(
db.or_(
User.username.ilike(f'%{search_query}%'),
User.email.ilike(f'%{search_query}%')
)
)
# Apply type filter
if filter_type == 'system_admins':
query = query.filter(User.role == Role.SYSTEM_ADMIN)
elif filter_type == 'admins':
query = query.filter(User.role == Role.ADMIN)
elif filter_type == 'blocked':
query = query.filter(User.is_blocked == True)
elif filter_type == 'unverified':
query = query.filter(User.is_verified == False)
elif filter_type == 'freelancers':
query = query.filter(Company.is_personal == True)
# Order by company name and username
query = query.order_by(Company.name, User.username)
# Paginate the results
try:
users = query.paginate(page=page, per_page=per_page, error_out=False)
# Debug log
if users.items:
logger.info(f"First item type: {type(users.items[0])}")
logger.info(f"First item: {users.items[0]}")
except Exception as e:
logger.error(f"Error paginating users: {str(e)}")
# Fallback to empty pagination
from flask_sqlalchemy import Pagination
users = Pagination(query=None, page=1, per_page=per_page, total=0, items=[])
# Get all companies for filter dropdown
companies = Company.query.order_by(Company.name).all()
# Calculate statistics
all_users = User.query.all()
stats = {
'total_users': len(all_users),
'verified_users': len([u for u in all_users if u.is_verified]),
'blocked_users': len([u for u in all_users if u.is_blocked]),
'system_admins': len([u for u in all_users if u.role == Role.SYSTEM_ADMIN]),
'company_admins': len([u for u in all_users if u.role == Role.ADMIN]),
}
return render_template('system_admin_users.html',
title='System User Management',
users=users,
companies=companies,
stats=stats,
selected_company_id=company_id,
search_query=search_query,
current_filter=filter_type)
@users_bp.route('/system-admin/<int:user_id>/edit', methods=['GET', 'POST'])
@system_admin_required
def system_admin_edit_user(user_id):
"""System admin edit any user"""
user = User.query.get_or_404(user_id)
if request.method == 'POST':
username = request.form.get('username')
email = request.form.get('email')
password = request.form.get('password')
role_name = request.form.get('role')
company_id = request.form.get('company_id', type=int)
team_id = request.form.get('team_id', type=int)
is_verified = 'is_verified' in request.form
is_blocked = 'is_blocked' in request.form
# Validate input
validator = FormValidator()
validator.validate_required(username, 'Username')
validator.validate_required(email, 'Email')
# Validate uniqueness (exclude current user)
if validator.is_valid():
if username != user.username:
validator.validate_unique(User, 'username', username, company_id=user.company_id)
if email != user.email:
validator.validate_unique(User, 'email', email, company_id=user.company_id)
# Prevent removing the last system admin
if validator.is_valid() and user.role == Role.SYSTEM_ADMIN and role_name != 'SYSTEM_ADMIN':
system_admin_count = User.query.filter_by(role=Role.SYSTEM_ADMIN, is_blocked=False).count()
if system_admin_count <= 1:
validator.errors.add('Cannot remove the last system administrator')
if validator.is_valid():
user.username = username
user.email = email
user.is_verified = is_verified
user.is_blocked = is_blocked
# Update company and team
if company_id:
user.company_id = company_id
if team_id:
user.team_id = team_id
else:
user.team_id = None # Clear team if not selected
# Update role
try:
user.role = Role[role_name] if role_name else Role.TEAM_MEMBER
except KeyError:
user.role = Role.TEAM_MEMBER
if password:
user.set_password(password)
db.session.commit()
flash(f'User {username} updated successfully!', 'success')
return redirect(url_for('users.system_admin_users'))
validator.flash_errors()
# Get all companies and teams for the form
companies = Company.query.order_by(Company.name).all()
teams = Team.query.filter_by(company_id=user.company_id).order_by(Team.name).all()
return render_template('system_admin_edit_user.html',
title='Edit User (System Admin)',
user=user,
companies=companies,
teams=teams,
roles=list(Role),
Role=Role)
@users_bp.route('/system-admin/<int:user_id>/delete', methods=['POST'])
@system_admin_required
def system_admin_delete_user(user_id):
"""System admin delete any user"""
user = User.query.get_or_404(user_id)
# Prevent deleting yourself
if user.id == g.user.id:
flash('You cannot delete your own account', 'error')
return redirect(url_for('users.system_admin_users'))
# Prevent deleting the last system admin
if user.role == Role.SYSTEM_ADMIN:
system_admin_count = User.query.filter_by(role=Role.SYSTEM_ADMIN).count()
if system_admin_count <= 1:
flash('Cannot delete the last system administrator', 'error')
return redirect(url_for('users.system_admin_users'))
username = user.username
company_name = user.company.name
try:
# Check if this is the last admin/supervisor in the company
admin_count = User.query.filter(
User.company_id == user.company_id,
User.role.in_([Role.ADMIN, Role.SUPERVISOR]),
User.id != user_id
).count()
if admin_count == 0:
# This is the last admin - need to handle company data
flash(f'User {username} is the last administrator in {company_name}. Company data will need to be handled.', 'warning')
# For now, just prevent deletion
return redirect(url_for('users.system_admin_users'))
# Otherwise proceed with normal deletion
# Delete user-specific records
TimeEntry.query.filter_by(user_id=user_id).delete()
WorkConfig.query.filter_by(user_id=user_id).delete()
UserPreferences.query.filter_by(user_id=user_id).delete()
UserDashboard.query.filter_by(user_id=user_id).delete()
# Clear assignments
Task.query.filter_by(assigned_to_id=user_id).update({'assigned_to_id': None})
SubTask.query.filter_by(assigned_to_id=user_id).update({'assigned_to_id': None})
# Transfer ownership of created items
alternative_admin = User.query.filter(
User.company_id == user.company_id,
User.role.in_([Role.ADMIN, Role.SUPERVISOR]),
User.id != user_id
).first()
if alternative_admin:
Project.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
Task.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
SubTask.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
ProjectCategory.query.filter_by(created_by_id=user_id).update({'created_by_id': alternative_admin.id})
# Delete the user
db.session.delete(user)
db.session.commit()
flash(f'User {username} from {company_name} deleted successfully!', 'success')
except Exception as e:
db.session.rollback()
logger.error(f"Error deleting user {user_id}: {str(e)}")
flash(f'Error deleting user: {str(e)}', 'error')
return redirect(url_for('users.system_admin_users'))

75
routes/users_api.py Normal file
View File

@@ -0,0 +1,75 @@
"""
User API endpoints
"""
from flask import Blueprint, jsonify, request, g
from models import db, User, Role
from routes.auth import system_admin_required, role_required
from sqlalchemy import or_
users_api_bp = Blueprint('users_api', __name__, url_prefix='/api')
@users_api_bp.route('/system-admin/users/<int:user_id>/toggle-block', methods=['POST'])
@system_admin_required
def api_toggle_user_block(user_id):
"""API: Toggle user blocked status (System Admin only)"""
user = User.query.get_or_404(user_id)
# Safety check: prevent blocking yourself
if user.id == g.user.id:
return jsonify({'error': 'Cannot block your own account'}), 400
# Safety check: prevent blocking the last system admin
if user.role == Role.SYSTEM_ADMIN and not user.is_blocked:
system_admin_count = User.query.filter_by(role=Role.SYSTEM_ADMIN, is_blocked=False).count()
if system_admin_count <= 1:
return jsonify({'error': 'Cannot block the last system administrator'}), 400
user.is_blocked = not user.is_blocked
db.session.commit()
return jsonify({
'id': user.id,
'username': user.username,
'is_blocked': user.is_blocked,
'message': f'User {"blocked" if user.is_blocked else "unblocked"} successfully'
})
@users_api_bp.route('/search/users')
@role_required(Role.TEAM_MEMBER)
def search_users():
"""Search for users within the company"""
query = request.args.get('q', '').strip()
exclude_id = request.args.get('exclude', type=int)
if not query or len(query) < 2:
return jsonify({'users': []})
# Search users in the same company
users_query = User.query.filter(
User.company_id == g.user.company_id,
or_(
User.username.ilike(f'%{query}%'),
User.email.ilike(f'%{query}%')
),
User.is_blocked == False,
User.is_verified == True
)
if exclude_id:
users_query = users_query.filter(User.id != exclude_id)
users = users_query.limit(10).all()
return jsonify({
'users': [{
'id': user.id,
'username': user.username,
'email': user.email,
'avatar_url': user.get_avatar_url(32),
'role': user.role.value,
'team': user.team.name if user.team else None
} for user in users]
})