from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app from flask_login import login_required, current_user from sqlalchemy.orm import joinedload from sqlalchemy import asc, func from models import db, Employee, Menu, SubMenu, RoleAccess, Role, DemoCompany, DemoAccount, DemoSignup from models import BrandingSetting from sqlalchemy.exc import IntegrityError import os, time from werkzeug.utils import secure_filename from datetime import datetime, date, timedelta admin_bp = Blueprint('admin', __name__) # ------------------ ACCESS CONTROL ------------------ @admin_bp.route('/access-control', methods=['GET', 'POST']) @login_required def access_control(): roles = Role.query.order_by(Role.name.asc()).all() # unified role picker selected_role = (request.args.get('role') or request.form.get('role') or (roles[0].name if roles else None)) # ordered menus + ordered submenus all_menus = (Menu.query .options(joinedload(Menu.submenus)) .order_by(Menu.order.asc(), Menu.name.asc()) .all()) for m in all_menus: m.submenus.sort(key=lambda s: (s.order or 0, s.name.lower())) if request.method == 'POST' and selected_role: # Save access atomically selected_ids = [int(x) for x in request.form.getlist('submenu_ids')] RoleAccess.query.filter_by(role=selected_role).delete() db.session.bulk_save_objects([RoleAccess(role=selected_role, submenu_id=sid) for sid in selected_ids]) db.session.commit() flash(f"✅ Access updated for role: {selected_role}", "success") # keep the role on reload return redirect(url_for('admin.access_control', role=selected_role)) selected_ids = [] if selected_role: selected_ids = [ra.submenu_id for ra in RoleAccess.query.filter_by(role=selected_role)] return render_template('admin/access_control.html', roles=roles, all_menus=all_menus, selected_role=selected_role, selected_ids=set(selected_ids)) # ------------------ MENU MANAGEMENT ------------------ @admin_bp.route('/menu-management', methods=['GET', 'POST']) @login_required def menu_management(): if request.method == 'POST': # simple add (left for backwards compatibility) name = (request.form.get('name') or '').strip() icon = (request.form.get('icon') or '').strip() or None order = int(request.form.get('menu_order') or 0) if not name: flash('Menu name is required.', 'danger') return redirect(url_for('admin.menu_management')) db.session.add(Menu(name=name, icon=icon, order=order)) db.session.commit() flash('Menu added successfully', 'success') return redirect(url_for('admin.menu_management')) menus = (Menu.query .options(joinedload(Menu.submenus)) .order_by(Menu.order.asc(), Menu.name.asc()) .all()) for m in menus: m.submenus.sort(key=lambda s: (s.order or 0, s.name.lower())) return render_template('admin/menu_management.html', menus=menus) @admin_bp.route('/menu-management/add-menu', methods=['POST']) @login_required def add_menu(): name = (request.form.get('menu_name') or '').strip() icon = (request.form.get('menu_icon') or '').strip() or None order = int(request.form.get('menu_order') or 0) if not name: flash('Menu name is required.', 'danger') return redirect(url_for('admin.menu_management')) db.session.add(Menu(name=name, icon=icon, order=order)) db.session.commit() flash(f'Menu "{name}" added successfully.', 'success') return redirect(url_for('admin.menu_management')) @admin_bp.route('/menu-management/edit-menu/', methods=['POST']) @login_required def edit_menu(menu_id): menu = Menu.query.get_or_404(menu_id) menu.name = (request.form.get('menu_name') or '').strip() menu.icon = (request.form.get('menu_icon') or '').strip() or None menu.order = int(request.form.get('menu_order') or 0) db.session.commit() flash(f'Menu "{menu.name}" updated.', 'success') return redirect(url_for('admin.menu_management')) @admin_bp.route('/menu-management/delete-menu/', methods=['POST']) @login_required def delete_menu(menu_id): menu = Menu.query.get_or_404(menu_id) db.session.delete(menu) db.session.commit() flash(f'Menu "{menu.name}" deleted.', 'info') return redirect(url_for('admin.menu_management')) @admin_bp.route('/menu-management/add-submenu', methods=['POST']) @login_required def add_submenu(): menu_id = request.form.get('menu_id') name = (request.form.get('submenu_name') or '').strip() url = (request.form.get('submenu_url') or '').strip() endpoint = (request.form.get('submenu_endpoint') or '').strip() or None order = int(request.form.get('submenu_order') or 0) if not (menu_id and name and url): flash('SubMenu name, URL, and Menu are required.', 'danger') return redirect(url_for('admin.menu_management')) submenu = SubMenu( name=name, url=url, endpoint=endpoint, order=order, menu_id=int(menu_id) ) db.session.add(submenu) db.session.commit() flash(f'Submenu "{name}" added.', 'success') return redirect(url_for('admin.menu_management')) @admin_bp.route('/menu-management/edit-submenu/', methods=['POST']) @login_required def edit_submenu(submenu_id): submenu = SubMenu.query.get_or_404(submenu_id) submenu.name = (request.form.get('submenu_name') or '').strip() submenu.url = (request.form.get('submenu_url') or '').strip() submenu.endpoint = (request.form.get('submenu_endpoint') or '').strip() or None submenu.order = int(request.form.get('submenu_order') or 0) db.session.commit() flash(f'Submenu "{submenu.name}" updated.', 'success') return redirect(url_for('admin.menu_management')) @admin_bp.post("/menu-management/delete-submenu/") @login_required def delete_submenu(submenu_id): sub = SubMenu.query.get_or_404(submenu_id) try: RoleAccess.query.filter_by(submenu_id=submenu_id).delete(synchronize_session=False) db.session.delete(sub) db.session.commit() flash("Submenu deleted.", "success") except IntegrityError: db.session.rollback() flash("Cannot delete submenu because it is still referenced. Remove role/user access first.", "danger") except Exception as e: db.session.rollback() flash(f"Delete failed: {e}", "danger") return redirect(url_for("admin.menu_management")) # -------- Route Discovery -------- @admin_bp.route('/route-discovery') @login_required def route_discovery(): search = (request.args.get('search') or '').strip().lower() page = request.args.get('page', 1, type=int) routes = [] for rule in current_app.url_map.iter_rules(): if rule.endpoint.startswith('static'): continue methods = list(rule.methods - {'HEAD', 'OPTIONS'}) routes.append({'endpoint': rule.endpoint, 'url': rule.rule, 'methods': methods}) if search: routes = [r for r in routes if search in r['endpoint'].lower() or search in r['url'].lower()] existing = {(sm.endpoint, sm.url) for sm in SubMenu.query.with_entities(SubMenu.endpoint, SubMenu.url)} for r in routes: r['assigned'] = (r['endpoint'], r['url']) in existing PER_PAGE = 20 total = len(routes) start = (page - 1) * PER_PAGE paginated = routes[start:start + PER_PAGE] class Pagination: def __init__(self, items, page, per_page, total): self.items = items; self.page = page; self.per_page = per_page self.total = total; self.pages = (total + per_page - 1) // per_page self.has_prev = page > 1; self.has_next = page < self.pages self.prev_num = page - 1; self.next_num = page + 1 return render_template('admin/route_discovery.html', routes=Pagination(paginated, page, PER_PAGE, total), search=search) # ------------------ SIMPLE PERMISSION HELPERS ------------------ from functools import wraps from models import RoleAccess as RA, SubMenu as SM # avoid name clash def has_menu_access(user, endpoint: str) -> bool: if not user or not endpoint: return False role_name = (getattr(user.role, 'name', '') or '').strip().lower() if role_name == 'admin': return True sm = SM.query.filter_by(endpoint=endpoint).first() if not sm: return False if RA.query.filter_by(role=role_name, submenu_id=sm.id).first(): return True return False def permission_required(endpoint_name: str): def wrapper(f): @wraps(f) def decorated(*args, **kwargs): if not current_user.is_authenticated or not current_user.role: flash("Please log in to continue.", "warning") return redirect(url_for('auth.login')) if has_menu_access(current_user, endpoint_name): return f(*args, **kwargs) flash("You do not have permission to access this feature.", "danger") return redirect(url_for('auth.forbidden')) return decorated return wrapper # ------------------ USER ACCESS (per-user overrides) ------------------ from models import Role as RoleModel # alias to avoid confusion @admin_bp.route('/user-access', methods=['GET', 'POST']) @login_required def user_access(): from models import UserAccess employees = Employee.query.order_by(Employee.full_name).all() all_menus = (Menu.query .options(joinedload(Menu.submenus)) .order_by(Menu.order.asc(), Menu.name.asc()) .all()) for m in all_menus: m.submenus.sort(key=lambda s: ((s.order or 0), (s.name or '').lower())) selected_user_id = request.args.get('user_id') or request.form.get('user_id') selected_ids = [] if request.method == 'POST': if not selected_user_id: flash("Please choose a user first.", "warning") return redirect(url_for('admin.user_access')) submenu_ids = [int(x) for x in request.form.getlist('submenu_ids')] UserAccess.query.filter_by(user_id=int(selected_user_id)).delete() db.session.bulk_save_objects([UserAccess(user_id=int(selected_user_id), submenu_id=sid) for sid in submenu_ids]) db.session.commit() flash("✅ User-specific permissions updated.", "success") return redirect(url_for('admin.user_access', user_id=selected_user_id)) if selected_user_id: from models import UserAccess selected_ids = [ua.submenu_id for ua in UserAccess.query.filter_by(user_id=int(selected_user_id))] return render_template('admin/user_access.html', employees=employees, all_menus=all_menus, selected_user_id=int(selected_user_id) if selected_user_id else None, selected_ids=set(selected_ids)) def _is_admin(): try: return current_user and current_user.role and current_user.role.name.lower() in ('admin', 'hr admin', 'hr') except Exception: return False # ------------------ BRANDING ------------------ @admin_bp.route('/branding', methods=['GET', 'POST']) @login_required def branding_settings(): if not (current_user.role and current_user.role.name.lower() in ['admin', 'hr admin']): flash("Access denied. Only Admin/HR Admin can modify branding.", "danger") return redirect(url_for('dashboard.index')) bs = BrandingSetting.query.get(1) or BrandingSetting(id=1) if request.method == 'POST': bs.sidebar_bg = request.form.get('sidebar_bg') or bs.sidebar_bg bs.topbar_bg = request.form.get('topbar_bg') or bs.topbar_bg bs.accent = request.form.get('accent') or bs.accent file = request.files.get('logo_file') if file and file.filename: filename = secure_filename(file.filename) ext = os.path.splitext(filename)[1].lower() if ext not in ('.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp'): flash("Unsupported logo format.", "warning") else: static_dir = os.path.join(current_app.root_path, 'static', 'images') os.makedirs(static_dir, exist_ok=True) save_name = f"custom_logo{ext}" path = os.path.join(static_dir, save_name) file.save(path) bs.logo_url = url_for('static', filename=f'images/{save_name}', _external=False) db.session.add(bs) db.session.commit() flash("Branding updated successfully!", "success") return redirect(url_for('admin.branding_settings')) return render_template('admin/branding_settings.html', bs=bs) # ------------------ DEMO SETUP ------------------ def create_demo_admin_for_company(company: DemoCompany) -> DemoAccount: """ Creates: - Role 'admin' if not present - Employee with that role, unique demo employee_code - DemoAccount pointing to this employee Returns the DemoAccount. """ admin_role = Role.query.filter(func.lower(Role.name) == 'admin').first() if not admin_role: admin_role = Role(name='admin', description='Administrator (auto-created)') db.session.add(admin_role) db.session.flush() # Generate employee code like DEMO0001, DEMO0002... base_code = f"DEMO{company.id:04d}" emp_code = base_code i = 1 while Employee.query.filter_by(employee_code=emp_code).first(): emp_code = f"{base_code}_{i}" i += 1 today = date.today() demo_email = f"demo_admin_{company.id}@demo.local" demo_emp = Employee( employee_code=emp_code, full_name=f"{company.name} Admin", email=demo_email, date_of_joining=today, role=admin_role, status='active', ) plain_password = "admin@123" demo_emp.set_password(plain_password) db.session.add(demo_emp) db.session.flush() login_url = url_for('employee.employee_login', _external=True) demo_account = DemoAccount( company_id=company.id, display_name="Admin", role_id=admin_role.id, employee_id=demo_emp.id, login_url=login_url, login_username=demo_emp.employee_code, login_password=plain_password, expires_on=company.expires_on, is_active=True, notes="Auto-created demo admin account.", ) db.session.add(demo_account) return demo_account @admin_bp.route('/demo-setup', methods=['GET', 'POST']) @login_required def demo_setup(): """ Admin page to: - Create demo companies (auto-creates Admin demo account) - Toggle Active / Inactive status """ if request.method == 'POST': action = request.form.get('action') try: # 1) Add new company (auto admin) if action == 'add_company': company_id = request.form.get('company_id') name = request.form.get('company_name', '').strip() website = request.form.get('company_website', '').strip() contact_person = request.form.get('company_contact_person', '').strip() contact_email = request.form.get('company_contact_email', '').strip() contact_phone = request.form.get('company_contact_phone', '').strip() notes = request.form.get('company_notes', '').strip() expires_on_str = request.form.get('company_expires_on', '').strip() if not name: flash('Company name is required.', 'danger') return redirect(url_for('admin.demo_setup')) expires_on = None if expires_on_str: try: expires_on = datetime.strptime(expires_on_str, '%Y-%m-%d').date() except ValueError: flash('Invalid company expiry date format.', 'danger') return redirect(url_for('admin.demo_setup')) if company_id: flash('Editing demo companies is not implemented yet.', 'warning') return redirect(url_for('admin.demo_setup')) company = DemoCompany( name=name, website=website, contact_person=contact_person, contact_email=contact_email, contact_phone=contact_phone, notes=notes, expires_on=expires_on, is_active=True, ) db.session.add(company) db.session.commit() # get company.id create_demo_admin_for_company(company) db.session.commit() flash('Demo company and default admin account created.', 'success') # 2) Toggle Active / Inactive elif action == 'toggle_company_status': company_id = request.form.get('company_id') try: company_id = int(company_id) except (TypeError, ValueError): flash('Invalid company id.', 'danger') return redirect(url_for('admin.demo_setup')) company = DemoCompany.query.get(company_id) if not company: flash('Demo company not found.', 'danger') return redirect(url_for('admin.demo_setup')) new_status = request.form.get('new_status') # 'activate' or 'deactivate' make_active = (new_status == 'activate') company.is_active = make_active for acc in company.demo_accounts: acc.is_active = make_active db.session.commit() flash( f"Demo company '{company.name}' marked as " f"{'Active' if make_active else 'Inactive'}.", 'success' ) except Exception: current_app.logger.exception('Error in demo_setup') flash('Something went wrong while saving. Please try again.', 'danger') return redirect(url_for('admin.demo_setup')) # GET companies = (DemoCompany.query.options(joinedload(DemoCompany.demo_accounts)) .order_by(DemoCompany.created_at.desc()) .all()) roles = Role.query.order_by(Role.name).all() # kept if needed later return render_template( 'admin/demo_setup.html', companies=companies, roles=roles, ) @admin_bp.route('/demo-requests', methods=['GET', 'POST']) @login_required def demo_requests(): from flask_login import current_user if request.method == 'POST': action = request.form.get('action') signup_id = request.form.get('signup_id', type=int) signup = DemoSignup.query.get_or_404(signup_id) if action == 'approve': expiry_str = (request.form.get('expiry_date') or '').strip() if expiry_str: try: expires_on = datetime.strptime(expiry_str, '%Y-%m-%d').date() except ValueError: flash("Invalid expiry date, using default 14 days.", "warning") expires_on = date.today() + timedelta(days=14) else: expires_on = date.today() + timedelta(days=14) # Create DemoCompany company = DemoCompany( name=signup.company_name, website=signup.website, contact_person=signup.contact_name, contact_email=signup.work_email, contact_phone=signup.phone, notes=signup.message, expires_on=expires_on, is_active=True, ) db.session.add(company) db.session.flush() # get company.id # Auto admin + account create_demo_admin_for_company(company) # Mark signup approved signup.status = 'approved' signup.approved_at = datetime.utcnow() signup.approved_by_id = current_user.id signup.demo_company_id = company.id db.session.commit() flash( f"Demo created for {signup.company_name}. " f"Admin login is visible in Demo Setup.", "success" ) elif action == 'reject': reason = (request.form.get('reject_reason') or '').strip() signup.status = 'rejected' signup.approved_at = datetime.utcnow() signup.approved_by_id = current_user.id if reason: # append reason to message existing = signup.message or '' signup.message = existing + f"\n\n[Rejected reason] {reason}" db.session.commit() flash(f"Request from {signup.company_name} marked as rejected.", "info") return redirect(url_for('admin.demo_requests')) # GET: lists pending = (DemoSignup.query .filter_by(status='pending') .order_by(DemoSignup.created_at.desc()) .all()) approved = (DemoSignup.query .filter_by(status='approved') .order_by(DemoSignup.created_at.desc()) .limit(20) .all()) rejected = (DemoSignup.query .filter_by(status='rejected') .order_by(DemoSignup.created_at.desc()) .limit(20) .all()) return render_template( 'admin/demo_requests.html', pending=pending, approved=approved, rejected=rejected, )