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.exc import IntegrityError from sqlalchemy import asc, func from datetime import datetime from functools import wraps import os import json from werkzeug.utils import secure_filename from datetime import datetime, date, timedelta from models import ( db, Employee, Menu, SubMenu, RoleAccess, Role, UserAccess, # ✅ needed for has_menu_access + user_access BrandingSetting, SystemSetting, ProductiveHoursRule, Department, Designation, EarningHead, # ✅ NEW DeductionHead, # ✅ NEW DemoCompany, # ✅ for demo setup DemoAccount, # ✅ for demo setup DemoSignup # ✅ for demo setup ) from routes.biometric import sync_biometric_for_date # If you already have roles_required elsewhere, keep it. # Otherwise adjust import path accordingly. from routes.auth import roles_required 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 or '').lower())) if request.method == 'POST' and selected_role: 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") 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 or '').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) UserAccess.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) # ------------------ SECURITY HELPERS ------------------ 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 = SubMenu.query.filter_by(endpoint=endpoint).first() if not sm: return False if RoleAccess.query.filter_by(role=role_name, submenu_id=sm.id).first(): return True 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 # ------------------ USER ACCESS (USER OVERRIDE) ------------------ @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() 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)) # ------------------ BRANDING SETTINGS ------------------ @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) # ------------------ JSON SETTINGS HELPERS ------------------ 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 # ------------------ BRAND / SYSTEM CONFIG ------------------ @admin_bp.route("/brand-config", methods=["GET", "POST"]) @login_required def brand_config(): if request.method == "POST": form_type = request.form.get("form_type") # ---- Email settings ---------------------------------------- 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 ------------------------------------- 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")) # ---- Productive hours rule add ------------------------------ elif form_type == "prod_rule_add": scope_type = request.form.get("scope_type") 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) 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")) # ---- Productive hours rule delete --------------------------- 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 settings ------------------------------------ elif form_type == "biometric": source_type = request.form.get("source_type", "api") biometric_cfg = { "enabled": bool(request.form.get("biometric_enabled")), "source_type": source_type, "api": { "base_url": request.form.get("api_base_url", "").strip(), "http_method": request.form.get("api_http_method", "GET").upper(), "request_template": request.form.get("api_request_template", "").strip(), "headers_json": request.form.get("api_headers_json", "").strip(), "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": { "db_type": request.form.get("db_type", "mysql").strip(), "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(), "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")) # ---- Payroll Deduction proration (DB multi-select) ---------- elif form_type == "payroll_deduction": cfg = { "enable_proration": bool(request.form.get("enable_proration")), "prorate_mode": (request.form.get("prorate_mode") or "all_except").strip(), # ✅ multi-select values "exception_heads": [h.strip().lower() for h in request.form.getlist("exception_heads") if h.strip()], "only_heads": [h.strip().lower() for h in request.form.getlist("only_heads") if h.strip()], # keep hard default if you want "force_non_prorated_heads": ["pt"], } set_json_setting("payroll_deduction_settings", cfg, current_user) flash("Payroll deduction proration settings updated.", "success") return redirect(url_for("admin.brand_config")) # ---- Payroll Earning proration (DB multi-select) ------------ elif form_type == "payroll_earning": cfg = { "enable_proration": bool(request.form.get("enable_proration")), "prorate_mode": (request.form.get("prorate_mode") or "all_except").strip(), # ✅ multi-select values "exception_heads": [h.strip().lower() for h in request.form.getlist("exception_heads") if h.strip()], "only_heads": [h.strip().lower() for h in request.form.getlist("only_heads") if h.strip()], # optional extra safety list "force_non_prorated_heads": [h.strip().lower() for h in request.form.getlist("force_non_prorated_heads") if h.strip()], } set_json_setting("payroll_earning_settings", cfg, current_user) flash("Payroll earning proration 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": ""}) 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", }, }) payroll_deduction_settings = get_json_setting("payroll_deduction_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": ["pt"], "only_heads": [], "force_non_prorated_heads": ["pt"], }) payroll_earning_settings = get_json_setting("payroll_earning_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": [], "only_heads": [], "force_non_prorated_heads": [], }) # ✅ Heads from DB earning_heads = EarningHead.query.order_by(EarningHead.name.asc()).all() deduction_heads = DeductionHead.query.order_by(DeductionHead.name.asc()).all() # Existing rules = ProductiveHoursRule.query.order_by(ProductiveHoursRule.scope_type, ProductiveHoursRule.id).all() 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, payroll_deduction_settings=payroll_deduction_settings, payroll_earning_settings=payroll_earning_settings, # ✅ NEW earning_heads=earning_heads, # ✅ NEW deduction_heads=deduction_heads, # ✅ NEW rules=rules, departments=departments, designations=designations, employees=employees, ) # ------------------ BIOMETRIC TEST SYNC ------------------ @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")) # ------------------ ATTENDANCE SETTINGS JSON ------------------ @admin_bp.route("/attendance-settings", methods=["GET", "POST"]) @login_required def attendance_settings(): setting = SystemSetting.query.filter_by(key="attendance_source_priority").first() if not setting: setting = SystemSetting(key="attendance_source_priority") db.session.add(setting) db.session.flush() current_obj = {} if setting.json_value: try: current_obj = json.loads(setting.json_value) except Exception: current_obj = {} current_text = json.dumps(current_obj or { "default": { "order": ["manual", "biometric", "self", "desktime_hours", "desktime_productive"], "min_hours_present": 7.5, "min_hours_half_day": 4 } }, indent=2) if request.method == "POST": raw = (request.form.get("priority_json") or "").strip() if not raw: flash("Priority JSON cannot be empty.", "danger") return render_template("admin/attendance_settings.html", priority_json=current_text) try: obj = json.loads(raw) except Exception as e: flash(f"Invalid JSON: {e}", "danger") return render_template("admin/attendance_settings.html", priority_json=raw) setting.json_value = json.dumps(obj) db.session.commit() flash("✅ Attendance source priority updated.", "success") return render_template("admin/attendance_settings.html", priority_json=json.dumps(obj, indent=2)) return render_template("admin/attendance_settings.html", priority_json=current_text) # ------------------ 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, )