from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app from flask_login import login_required from sqlalchemy.orm import joinedload from sqlalchemy import asc from models import db, Employee, Menu, SubMenu, RoleAccess, Role from models import BrandingSetting,SystemSetting, ProductiveHoursRule, Department, Designation, Employee from sqlalchemy.exc import IntegrityError import os, time from werkzeug.utils import secure_filename import json 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) # ✅ ensure order is saved on edit 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: # 1) Delete role access rows referencing this submenu RoleAccess.query.filter_by(submenu_id=submenu_id).delete(synchronize_session=False) # 2) If you also have UserAccess tied to submenu_id, delete those too. # from models import UserAccess # UserAccess.query.filter_by(submenu_id=submenu_id).delete(synchronize_session=False) # 3) Now delete the submenu db.session.delete(sub) db.session.commit() flash("Submenu deleted.", "success") except IntegrityError as e: 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 (kept; small tidy) -------- @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) # security.py (or inside routes/auth.py if you prefer) from functools import wraps from flask import redirect, url_for, flash from flask_login import current_user from models import RoleAccess, UserAccess, SubMenu def has_menu_access(user, endpoint: str) -> bool: """Allow access if: - user is admin, OR - role has access to submenu, OR - user override grants access to submenu. """ if not user or not endpoint: return False role_name = (getattr(user.role, 'name', '') or '').strip().lower() if role_name == 'admin': return True sm = SubMenu.query.filter_by(endpoint=endpoint).first() if not sm: return False # Role-based if RoleAccess.query.filter_by(role=role_name, submenu_id=sm.id).first(): return True # User override if UserAccess.query.filter_by(user_id=user.id, 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 # routes/admin.py (additions) from flask import Blueprint, render_template, request, redirect, url_for, flash from flask_login import login_required from sqlalchemy.orm import joinedload from models import db, Employee, Menu, SubMenu, UserAccess, Role from routes.auth import roles_required # if you put it elsewhere, import accordingly @admin_bp.route('/user-access', methods=['GET', 'POST']) @login_required @roles_required('admin') def user_access(): employees = Employee.query.order_by(Employee.full_name).all() # Order menus and submenus; coalesce None -> 0 for order 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: 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: return False @admin_bp.route('/branding', methods=['GET', 'POST']) @login_required def branding_settings(): # Restrict to admin / HR admin only 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 # Optional logo upload 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) # --- helpers for settings ----------------------------------- def get_json_setting(key, default): setting = SystemSetting.query.filter_by(key=key).first() if setting and setting.json_value: try: return json.loads(setting.json_value) except Exception: return default return default def set_json_setting(key, data, user=None): setting = SystemSetting.query.filter_by(key=key).first() if not setting: setting = SystemSetting(key=key) db.session.add(setting) setting.json_value = json.dumps(data) setting.value = json.dumps(data) if user is not None: setting.updated_by = user db.session.commit() return setting @admin_bp.route("/brand-config", methods=["GET", "POST"]) @login_required def brand_config(): # Optional: restrict to HR/Admin if you want # if not _is_hr(current_user): abort(403) # Handle POST actions if request.method == "POST": form_type = request.form.get("form_type") # ---- Email settings form ----------------------------------- if form_type == "email": email_cfg = { "smtp_server": request.form.get("smtp_server", "").strip(), "smtp_port": int(request.form.get("smtp_port") or 587), "use_tls": bool(request.form.get("use_tls")), "use_ssl": bool(request.form.get("use_ssl")), "username": request.form.get("smtp_username", "").strip(), "password": request.form.get("smtp_password", "").strip(), "from_name": request.form.get("from_name", "").strip(), "from_email": request.form.get("from_email", "").strip(), } set_json_setting("email_settings", email_cfg, current_user) flash("Email settings updated.", "success") return redirect(url_for("admin.brand_config")) # ---- DeskTime settings form ------------------------------- elif form_type == "desktime": desktime_cfg = { "enabled": bool(request.form.get("desktime_enabled")), "api_key": request.form.get("desktime_api_key", "").strip(), } set_json_setting("desktime_settings", desktime_cfg, current_user) flash("DeskTime settings updated.", "success") return redirect(url_for("admin.brand_config")) # ---- Add productive hours rule ---------------------------- elif form_type == "prod_rule_add": scope_type = request.form.get("scope_type") # company/department/designation/employee required_hours_raw = request.form.get("required_hours") or "0" try: required_hours = float(required_hours_raw) except ValueError: required_hours = 0.0 dept_id = request.form.get("department_id") or None desig_id = request.form.get("designation_id") or None emp_id = request.form.get("employee_id") or None if dept_id: dept_id = int(dept_id) if desig_id: desig_id = int(desig_id) if emp_id: emp_id = int(emp_id) # Normalise based on scope if scope_type == "company": dept_id = None desig_id = None emp_id = None elif scope_type == "department": desig_id = None emp_id = None elif scope_type == "designation": emp_id = None if required_hours <= 0: flash("Please enter a valid required hours value.", "danger") else: rule = ProductiveHoursRule( scope_type=scope_type, department_id=dept_id, designation_id=desig_id, employee_id=emp_id, required_hours=required_hours, ) db.session.add(rule) db.session.commit() flash("Productive hours rule added.", "success") return redirect(url_for("admin.brand_config")) # ---- Delete rule ------------------------------------------ elif form_type == "prod_rule_delete": rule_id = request.form.get("rule_id") if rule_id: rule = ProductiveHoursRule.query.get(rule_id) if rule: db.session.delete(rule) db.session.commit() flash("Rule deleted.", "success") return redirect(url_for("admin.brand_config")) # ---- Biometric (API / DB) settings form ---------------------- elif form_type == "biometric": source_type = request.form.get("source_type", "api") # api / db biometric_cfg = { "enabled": bool(request.form.get("biometric_enabled")), "source_type": source_type, # "api" or "db" # ---- API related ---- "api": { "base_url": request.form.get("api_base_url", "").strip(), "http_method": request.form.get("api_http_method", "GET").upper(), # Example: query params template like "?date={date}&emp_code={employee_code}" "request_template": request.form.get("api_request_template", "").strip(), # Optional headers as JSON string, e.g. {"Authorization": "Bearer XXX"} "headers_json": request.form.get("api_headers_json", "").strip(), # Response parsing "response_list_key": request.form.get("api_response_list_key", "").strip(), "map_employee_code": request.form.get("api_map_employee_code", "").strip(), "map_punch_time": request.form.get("api_map_punch_time", "").strip(), "map_direction": request.form.get("api_map_direction", "").strip(), "time_format": request.form.get("api_time_format", "%Y-%m-%d %H:%M:%S").strip(), }, # ---- DB related ---- "db": { "db_type": request.form.get("db_type", "mysql").strip(), # mysql / mssql / postgres etc. "host": request.form.get("db_host", "").strip(), "port": request.form.get("db_port", "").strip(), "database": request.form.get("db_name", "").strip(), "username": request.form.get("db_username", "").strip(), "password": request.form.get("db_password", "").strip(), "table_name": request.form.get("db_table_name", "").strip(), "col_employee_code": request.form.get("db_col_employee_code", "").strip(), "col_punch_time": request.form.get("db_col_punch_time", "").strip(), "col_direction": request.form.get("db_col_direction", "").strip(), # optional WHERE template like "punch_date = {date}" "where_template": request.form.get("db_where_template", "").strip(), "time_format": request.form.get("db_time_format", "%Y-%m-%d %H:%M:%S").strip(), } } set_json_setting("biometric_settings", biometric_cfg, current_user) flash("Biometric integration settings updated.", "success") return redirect(url_for("admin.brand_config")) # -------- GET: load current settings --------------------------- email_settings = get_json_setting( "email_settings", { "smtp_server": "", "smtp_port": 587, "use_tls": True, "use_ssl": False, "username": "", "password": "", "from_name": "", "from_email": "", }, ) desktime_settings = get_json_setting( "desktime_settings", { "enabled": False, "api_key": "", }, ) rules = ProductiveHoursRule.query.order_by(ProductiveHoursRule.scope_type, ProductiveHoursRule.id).all() biometric_settings = get_json_setting( "biometric_settings", { "enabled": False, "source_type": "api", "api": { "base_url": "", "http_method": "GET", "request_template": "", "headers_json": "", "response_list_key": "", "map_employee_code": "", "map_punch_time": "", "map_direction": "", "time_format": "%Y-%m-%d %H:%M:%S", }, "db": { "db_type": "mysql", "host": "", "port": "", "database": "", "username": "", "password": "", "table_name": "", "col_employee_code": "", "col_punch_time": "", "col_direction": "", "where_template": "", "time_format": "%Y-%m-%d %H:%M:%S", }, }, ) departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() employees = Employee.query.order_by(Employee.full_name).all() return render_template( "admin/brand_config.html", email_settings=email_settings, desktime_settings=desktime_settings, biometric_settings=biometric_settings, rules=rules, departments=departments, designations=designations, employees=employees, ) # routes/admin.py from datetime import datetime from flask import Blueprint, request, redirect, url_for, flash from flask_login import login_required from routes.biometric import sync_biometric_for_date # ... existing admin_bp code ... @admin_bp.route("/biometric/test-sync", methods=["POST"]) @login_required def biometric_test_sync(): date_str = request.form.get("sync_date") or datetime.utcnow().strftime("%Y-%m-%d") try: target_date = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: flash("Invalid date format. Use YYYY-MM-DD.", "danger") return redirect(url_for("admin.brand_config")) count = sync_biometric_for_date(target_date) flash(f"Biometric sync done for {target_date} – {count} punches stored.", "success") return redirect(url_for("admin.brand_config"))