#!/usr/bin/env python3 """ LMS Academy Module v1.0 JvG Consultancy — HSEQ Intelligence Dashboard Flask module: User portal, Admin portal, Quiz engine, Compliance matrix All routes prefixed /lms/ — register via app.register_blueprint(lms_bp) or direct import. """ import os import json import uuid import sqlite3 from datetime import datetime, date, timedelta from flask import ( Blueprint, request, render_template_string, redirect, url_for, jsonify, session ) # ── Config ─────────────────────────────────────────────────────────────────── BASE_PATH = os.environ.get('BASE_PATH', '/hseq-dashboard') DB_PATH = os.environ.get('LMS_DB_PATH', '/root/projects/jg/HSEQ/HSEQ-Intelligence-Monitor/app/hseq_kennisbank.db') lms_bp = Blueprint('lms', __name__, url_prefix='/lms') # Note: nginx strips BASE_PATH, so Flask sees /lms/ not /hseq-dashboard/lms/ @lms_bp.before_request def _lms_admin_guard(): """Protect all /admin/ routes — redirect non-admin users to user dashboard.""" from flask import request as req, redirect if req.path.startswith(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/lms/admin'): uid = session.get('user_id') if not uid: return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/lms/') db = get_db() try: user = db.execute('SELECT role FROM users WHERE id = ?', (uid,)).fetchone() if not user or user['role'] not in ('admin', 'hseq_manager'): return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/lms/') finally: db.close() # Try to use existing get_db from search module; fallback to direct connection try: import sys, importlib _app_dir = os.path.join(os.path.dirname(os.path.abspath(__file__))) if _app_dir not in sys.path: sys.path.insert(0, _app_dir) from search import get_db as _get_db def get_db(): return _get_db() except Exception: def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn # ── Helpers ────────────────────────────────────────────────────────────────── def _current_employee_id(): """Stub: returns Jorick's employee ID. Replace with real session auth.""" return 7 def _current_employee_name(): """Stub: returns Jorick's full name. Replace with real session auth.""" conn = get_db() try: r = conn.execute("SELECT first_name, last_name FROM employees WHERE id=7").fetchone() return f"{r['first_name']} {r['last_name']}" if r else 'Medewerker' finally: conn.close() def _base_css(): return """ :root{--primary:#003366;--primary-light:#1a5276;--success:#00A859;--warning:#F59E0B;--danger:#EF4444;--bg:#F3F4F6;--card:#fff;--text:#1F2937;--text-muted:#6B7280;--border:#E5E7EB} *{box-sizing:border-box;margin:0;padding:0} body{font-family:'Segoe UI',system-ui,-apple-system,sans-serif;background:var(--bg);color:var(--text);line-height:1.6} .container{max-width:1200px;margin:0 auto;padding:20px} h1{font-size:1.5rem;font-weight:700;color:var(--primary);margin-bottom:16px} h2{font-size:1.2rem;font-weight:600;color:var(--primary);margin:20px 0 12px} .cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(200px,1fr));gap:16px;margin-bottom:24px} .card{background:var(--card);border-radius:10px;padding:20px;box-shadow:0 1px 3px rgba(0,0,0,.08)} .card .label{font-size:.75rem;color:var(--text-muted);text-transform:uppercase;letter-spacing:.5px} .card .value{font-size:1.8rem;font-weight:700;color:var(--primary)} .card.success .value{color:var(--success)} .card.warning .value{color:var(--warning)} .card.danger .value{color:var(--danger)} table{width:100%;border-collapse:collapse;background:var(--card);border-radius:8px;overflow:hidden;box-shadow:0 1px 3px rgba(0,0,0,.08)} th,td{padding:10px 14px;text-align:left;font-size:.85rem;border-bottom:1px solid var(--border)} th{background:var(--primary);color:#fff;font-weight:600;font-size:.75rem;text-transform:uppercase;letter-spacing:.5px} tr:hover{background:#f0f4ff} .badge{display:inline-block;padding:2px 10px;border-radius:12px;font-size:.7rem;font-weight:600} .badge-success{background:#d1fae5;color:#065f46} .badge-warning{background:#fef3c7;color:#92400e} .badge-danger{background:#fee2e2;color:#991b1b} .badge-info{background:#dbeafe;color:#1e40af} .badge-muted{background:#e5e7eb;color:#4b5563} .progress-bar{background:#e5e7eb;border-radius:6px;height:8px;overflow:hidden} .progress-fill{height:100%;border-radius:6px;background:var(--success);transition:width .3s} .btn{display:inline-block;padding:8px 18px;border-radius:6px;font-size:.85rem;font-weight:600;text-decoration:none;border:none;cursor:pointer;transition:all .15s} .btn-primary{background:var(--primary);color:#fff} .btn-primary:hover{background:var(--primary-light)} .btn-success{background:var(--success);color:#fff} .btn-danger{background:var(--danger);color:#fff} .btn-warning{background:var(--warning);color:#fff} .btn-sm{padding:5px 12px;font-size:.75rem} .tabs{display:flex;gap:4px;margin-bottom:20px;border-bottom:2px solid var(--border);padding-bottom:0} .tabs a{padding:10px 20px;text-decoration:none;color:var(--text-muted);font-size:.85rem;font-weight:600;border-bottom:3px solid transparent;margin-bottom:-2px} .tabs a.active{color:var(--primary);border-bottom-color:var(--primary)} .tabs a:hover{color:var(--primary)} .heatmap{border-collapse:collapse;font-size:.75rem} .heatmap th,.heatmap td{padding:6px 10px;text-align:center;border:1px solid var(--border)} .heatmap .green{background:#d1fae5}.heatmap .amber{background:#fef3c7}.heatmap .red{background:#fee2e2} .iframe-wrap{position:relative;width:100%;padding-bottom:65%;height:0;overflow:hidden;border-radius:8px;box-shadow:0 2px 8px rgba(0,0,0,.1)} .iframe-wrap iframe{position:absolute;top:0;left:0;width:100%;height:100%;border:none} @media(max-width:768px){.cards{grid-template-columns:1fr 1fr}.container{padding:12px}table{font-size:.75rem}th,td{padding:6px 8px}} """ def _is_admin(): """Check if current user has admin role. Plant Managers and HSEQ Officers are admin.""" conn = get_db() try: emp = conn.execute("SELECT role_id FROM employees WHERE id=?", (_current_employee_id(),)).fetchone() if not emp: return False return emp['role_id'] in (6, 10) # HSEQ Officer=6, Plant Manager=10 finally: conn.close() def _deadline_html(due_date_str): """Return deadline with color-coded badge.""" if not due_date_str or due_date_str in ('None', '', 'null'): return '—' try: due = datetime.strptime(due_date_str, '%Y-%m-%d').date() except: return due_date_str today = date.today() days_left = (due - today).days if days_left < 0: return f'Verlopen ({abs(days_left)}d)' elif days_left <= 7: return f'⚠️ {days_left}d' elif days_left <= 14: return f'{days_left}d' elif days_left <= 30: return f'{days_left}d' else: return f'✅ {due_date_str}' uid = session.get('user_id') if not uid: return False db = get_db() try: user = db.execute('SELECT role FROM users WHERE id = ?', (uid,)).fetchone() return user and user['role'] in ('admin', 'hseq_manager') finally: db.close() _page_fn = None def _page(title, body_html, active_tab='dashboard'): global _page_fn if _page_fn is not None: is_admin_view = active_tab.startswith('admin') tabs_user = f'''
📊 Dashboard 📚 Mijn Trainingen 🏆 Certificaten
''' tabs_admin = f'''
📖 Cursussen 👥 Inschrijvingen 🛡️ Compliance 📈 Analytics
''' tabs = tabs_admin if is_admin_view else tabs_user # Build action buttons for the main header area user_btn_class = 'btn-active-portal' if not is_admin_view else 'btn btn-sm btn-outline' admin_btn_html = '' if _is_admin(): admin_btn_class = 'btn-active-portal' if is_admin_view else 'btn btn-sm btn-outline' admin_btn_html = f'⚙️ Admin' action_btns = '
🏠 User Portal' + admin_btn_html + '
' full_body = action_btns + tabs + '
' + body_html + '
' return _page_fn(full_body, active='lms', page_title=title) # ═══════════════════════════════════════════════════════════════════════════════ # USER PORTAL # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/') def user_dashboard(): emp_id = _current_employee_id() conn = get_db() try: enrollments = conn.execute(""" SELECT e.*, m.title, m.duration_minutes, m.category FROM lms_enrollments e JOIN elearning_modules m ON m.id = e.elearning_module_id WHERE e.employee_id = ? ORDER BY e.due_date ASC """, (emp_id,)).fetchall() completed = sum(1 for e in enrollments if e['status'] == 'completed') in_progress = sum(1 for e in enrollments if e['status'] == 'in_progress') not_started = sum(1 for e in enrollments if e['status'] == 'not_started') expiring = sum(1 for e in enrollments if e['due_date'] and e['due_date'] < str(date.today() + timedelta(days=30)) and e['status'] != 'completed') certs = conn.execute(""" SELECT c.*, m.title FROM trn_certificates c JOIN elearning_modules m ON m.id = c.module_id WHERE c.employee_id = ? ORDER BY c.id DESC """, (emp_id,)).fetchall() finally: conn.close() open_rows = "" completed_rows = "" for e in enrollments: badge = {'not_started':'badge-muted','in_progress':'badge-info','completed':'badge-success','failed':'badge-danger','expired':'badge-warning'}.get(e['status'],'badge-muted') status_label = e['status'].replace('_',' ').title() pct = e['progress_percent'] or 0 deadline = _deadline_html(e['due_date'] if 'due_date' in e.keys() else None) row = f""" {e['title']}
{e['category'] or ''}
{pct}% {status_label} {deadline} Openen """ if e['status'] in ('completed', 'failed', 'expired', 'withdrawn'): completed_rows += row else: open_rows += row body = f"""
📚
Totaal Trainingen
{len(enrollments)}
✅
Voltooid
{completed}
⏳
In Uitvoering
{in_progress}
⚠️
Nakomend / Verlopen
{expiring}

📋 Openstaande Trainingen

{open_rows or ''}
CursusVoortgangStatusDeadline
Geen openstaande trainingen 🎉

✅ Afgeronde Trainingen

{completed_rows or ''}
CursusScoreStatusDeadline
Geen afgeronde trainingen

Recente Certificaten

{"".join(f'' for c in certs) or ''}
TrainingScoreGeldigRef
{c["title"]}{c["quiz_score"]}%{c["valid_until"] or "—"}{c["certificate_ref"]}📄
Nog geen certificaten
""" return _page('Dashboard', body, 'dashboard') @lms_bp.route('/my-courses') def my_courses(): emp_id = _current_employee_id() conn = get_db() try: enrollments = conn.execute(""" SELECT e.*, m.title, m.duration_minutes, m.category, m.status as module_status FROM lms_enrollments e JOIN elearning_modules m ON m.id = e.elearning_module_id WHERE e.employee_id = ? ORDER BY e.status, e.due_date """, (emp_id,)).fetchall() finally: conn.close() rows = "" for e in enrollments: badge = {'not_started':'badge-muted','in_progress':'badge-info','completed':'badge-success','failed':'badge-danger','expired':'badge-warning'}.get(e['status'],'badge-muted') pct = e['progress_percent'] or 0 rows += f""" {e['title']} {e['category'] or '—'}
{pct}% {e['status'].replace('_',' ').title()} {_deadline_html(e['due_date'] if 'due_date' in e.keys() else None)} Openen """ body = f"""

Alle Toegewezen Trainingen

{rows or ''}
CursusCategorieVoortgangStatusDeadlineActie
Geen trainingen
""" return _page('Mijn Trainingen', body, 'courses') @lms_bp.route('/my-certificates') def my_certificates(): emp_id = _current_employee_id() conn = get_db() try: certs = conn.execute(""" SELECT c.*, m.title FROM trn_certificates c JOIN elearning_modules m ON m.id = c.module_id WHERE c.employee_id = ? ORDER BY c.id DESC """, (emp_id,)).fetchall() finally: conn.close() rows = "" for c in certs: valid_badge = '' if c['valid_until'] and c['valid_until'] not in ('None', '', 'null'): try: vdate = datetime.strptime(c['valid_until'], '%Y-%m-%d').date() if vdate < date.today(): valid_badge = 'Verlopen' elif vdate < date.today() + timedelta(days=30): valid_badge = f'Geldig t/m {c["valid_until"]}' else: valid_badge = f'Geldig t/m {c["valid_until"]}' except: pass rows += f""" {c['title']}
{valid_badge} {c['quiz_score']}% {'✅ Behaald' if c['passed'] else '❌ Niet behaald'} {c['certificate_ref'] or '—'} 📄 Bekijk """ body = f"""

🏆 Mijn Certificaten

{rows or ''}
TrainingScoreStatusReferentie
Nog geen certificaten behaald. Voltooi een training om een certificaat te ontvangen.
""" return _page('Certificaten', body, 'certs') @lms_bp.route('/certificate/') def certificate_view(cert_id): emp_id = _current_employee_id() conn = get_db() try: c = conn.execute(""" SELECT c.*, m.title, m.category, e.first_name, e.last_name FROM trn_certificates c JOIN elearning_modules m ON m.id = c.module_id JOIN employees e ON e.id = c.employee_id WHERE c.id = ? AND c.employee_id = ? """, (cert_id, emp_id)).fetchone() finally: conn.close() if not c: return _page('Certificaat', '

Certificaat niet gevonden.

') body = f"""
JvG HSEQ Academy
Certificaat van Voltooiing

🏆 {c['title']}

Dit certificaat wordt uitgereikt aan

{c['first_name']} {c['last_name']}

Score
{c['quiz_score']}%
Status
Behaald ✅
Geldig tot
{c['valid_until'] or 'Onbeperkt'}
Referentie: {c['certificate_ref']} | Uitgegeven: {c['issued_at']}
← Terug
""" return _page('Certificaat', body, 'certs') # ═══════════════════════════════════════════════════════════════════════════════ # COURSE PLAYER # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/course/') def course_player(module_id): emp_id = _current_employee_id() conn = get_db() try: module = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (module_id,)).fetchone() if not module: return _page('Fout', '

Cursus niet gevonden.

', 'courses') content = conn.execute("SELECT * FROM lms_course_content WHERE elearning_module_id=? ORDER BY display_order", (module_id,)).fetchall() enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() # Auto-create enrollment if missing if not enrollment: conn.execute("INSERT OR IGNORE INTO lms_enrollments (employee_id, elearning_module_id, status) VALUES (?,?, 'not_started')", (emp_id, module_id)) conn.commit() enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() # Mark as in_progress if enrollment and enrollment['status'] == 'not_started': conn.execute("UPDATE lms_enrollments SET status='in_progress', started_at=datetime('now'), last_accessed_at=datetime('now') WHERE id=?", (enrollment['id'],)) conn.commit() # Update last accessed if enrollment: conn.execute("UPDATE lms_enrollments SET last_accessed_at=datetime('now') WHERE id=?", (enrollment['id'],)) conn.commit() finally: conn.close() # Build content tabs content_tabs = "" iframe_src = "" for i, c in enumerate(content): active = 'btn-primary' if i == 0 else 'btn-sm' fname = c['file_name'] or os.path.basename(c['file_path']) content_tabs += f'{fname}\n' if i == 0: iframe_src = f"{BASE_PATH}/lms/serve-content/{c['id']}" # If no linked content, check if module has content_html if not content and module['content_html']: iframe_src = f"{BASE_PATH}/lms/serve-content-inline/{module_id}" content_tabs = 'Inline content' pct = enrollment['progress_percent'] or 0 if enrollment else 0 quiz_url = f"{BASE_PATH}/lms/course/{module_id}/quiz" complete_url = f"{BASE_PATH}/lms/course/{module_id}/complete" body = f"""

{module['title']}

{module['duration_minutes'] or 0} min
{content_tabs}
📝 Quiz Starten
""" return _page(module['title'], body, 'courses') @lms_bp.route('/serve-content/') def serve_content(content_id): """Serve linked HTML content file.""" conn = get_db() try: c = conn.execute("SELECT * FROM lms_course_content WHERE id=?", (content_id,)).fetchone() finally: conn.close() if not c: return "

Content niet gevonden.

", 404 fpath = c['file_path'] if os.path.exists(fpath): with open(fpath, 'r', encoding='utf-8') as f: return f.read() return f"

Bestand niet gevonden op server: {fpath}

", 404 @lms_bp.route('/serve-content-inline/') def serve_content_inline(module_id): """Serve inline content_html from elearning_modules.""" conn = get_db() try: m = conn.execute("SELECT content_html FROM elearning_modules WHERE id=?", (module_id,)).fetchone() finally: conn.close() if m and m['content_html']: return m['content_html'] return "

Geen content beschikbaar.

", 404 @lms_bp.route('/course//start', methods=['POST']) def course_start(module_id): emp_id = _current_employee_id() conn = get_db() try: conn.execute("""UPDATE lms_enrollments SET status='in_progress', started_at=datetime('now'), last_accessed_at=datetime('now') WHERE employee_id=? AND elearning_module_id=?""", (emp_id, module_id)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/course/{module_id}") @lms_bp.route('/course//progress', methods=['POST']) def course_progress(module_id): emp_id = _current_employee_id() data = request.get_json(silent=True) or {} pct = data.get('progress_percent', 0) ts = data.get('time_spent_seconds', 0) conn = get_db() try: conn.execute("""UPDATE lms_enrollments SET progress_percent=?, time_spent_seconds=time_spent_seconds+?, last_accessed_at=datetime('now') WHERE employee_id=? AND elearning_module_id=?""", (pct, ts, emp_id, module_id)) conn.commit() finally: conn.close() return jsonify(success=True) @lms_bp.route('/course//complete', methods=['POST']) def course_complete(module_id): emp_id = _current_employee_id() conn = get_db() try: # Check if there are quiz questions for this module module = conn.execute("SELECT quiz_questions, passing_score FROM elearning_modules WHERE id=?", (module_id,)).fetchone() has_quiz = False if module and module['quiz_questions']: try: questions = json.loads(module['quiz_questions']) has_quiz = len(questions) > 0 except: pass # Also check if there's a linked quiz in lms_course_content if not has_quiz: quiz_content = conn.execute("SELECT id FROM lms_course_content WHERE elearning_module_id=? AND (file_name LIKE '%quiz%' OR content_type='quiz')", (module_id,)).fetchone() has_quiz = bool(quiz_content) if has_quiz: # Has quiz — user clicked complete, mark as completed (quiz was done via iframe) conn.execute("""UPDATE lms_enrollments SET progress_percent=100, final_score=100, final_passed=1, status='completed', completed_at=datetime('now'), last_accessed_at=datetime('now') WHERE employee_id=? AND elearning_module_id=? AND status IN ('in_progress','not_started')""", (emp_id, module_id)) # Generate certificate enrollment = conn.execute("SELECT id FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() if enrollment: existing_cert = conn.execute("SELECT id FROM trn_certificates WHERE employee_id=? AND module_id=?", (emp_id, module_id)).fetchone() if not existing_cert: cert_ref = f"JvG-LMS-{module_id:03d}-{emp_id:04d}-{uuid.uuid4().hex[:6].upper()}" conn.execute("""INSERT INTO trn_certificates (employee_id, module_id, quiz_score, passed, certificate_ref) VALUES (?,?,100,1,?)""", (emp_id, module_id, cert_ref)) else: # No quiz — mark as completed with full score conn.execute("""UPDATE lms_enrollments SET progress_percent=100, final_score=100, final_passed=1, status='completed', completed_at=datetime('now'), last_accessed_at=datetime('now') WHERE employee_id=? AND elearning_module_id=? AND status IN ('in_progress','not_started')""", (emp_id, module_id)) # Generate certificate enrollment = conn.execute("SELECT id FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() if enrollment: existing_cert = conn.execute("SELECT id FROM trn_certificates WHERE employee_id=? AND module_id=?", (emp_id, module_id)).fetchone() if not existing_cert: cert_ref = f"JvG-LMS-{module_id:03d}-{emp_id:04d}-{uuid.uuid4().hex[:6].upper()}" conn.execute("""INSERT INTO trn_certificates (employee_id, module_id, quiz_score, passed, certificate_ref) VALUES (?,?,100,1,?)""", (emp_id, module_id, cert_ref)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/course/{module_id}") # ═══════════════════════════════════════════════════════════════════════════════ # QUIZ ENGINE # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/course//quiz') def quiz_view(module_id): emp_id = _current_employee_id() conn = get_db() try: module = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (module_id,)).fetchone() enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() finally: conn.close() if not module: return _page('Fout', '

Module niet gevonden.

', 'courses') # Try loading quiz questions from module questions = [] if module['quiz_questions']: try: questions = json.loads(module['quiz_questions']) except (json.JSONDecodeError, TypeError): pass # Check if there's a linked quiz HTML file conn = get_db() try: quiz_content = conn.execute("SELECT * FROM lms_course_content WHERE elearning_module_id=? AND file_name LIKE '%quiz%'", (module_id,)).fetchone() finally: conn.close() if quiz_content and os.path.exists(quiz_content['file_path']): # Serve external quiz HTML in iframe quiz_src = f"{BASE_PATH}/lms/serve-content/{quiz_content['id']}" body = f"""

Quiz: {module['title']}

De quiz wordt automatisch nagekeken. Resultaat wordt opgeslagen na afronding.

""" return _page(f'Quiz: {module["title"]}', body, 'courses') # Built-in quiz rendering from quiz_questions JSON if not questions: body = f"""

Quiz: {module['title']}

Geen quiz beschikbaar voor deze module.

Terug naar cursus
""" return _page(f'Quiz: {module["title"]}', body, 'courses') # Render questions q_html = "" for i, q in enumerate(questions): opts = "" for j, opt in enumerate(q.get('options', [])): letter = chr(65 + j) opts += f'' q_html += f'

{i+1}. {q.get("question","")}

{opts}
' body = f"""

Quiz: {module['title']}

{q_html}
""" return _page(f'Quiz: {module["title"]}', body, 'courses') @lms_bp.route('/course//quiz/submit', methods=['POST']) def quiz_submit(module_id): """Internal quiz submission — scores against quiz_questions JSON.""" emp_id = _current_employee_id() data = request.get_json(silent=True) or {} answers = data.get('answers', {}) conn = get_db() try: module = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (module_id,)).fetchone() if not module: return jsonify(success=False, error="Module niet gevonden"), 404 questions = [] if module['quiz_questions']: try: questions = json.loads(module['quiz_questions']) except (json.JSONDecodeError, TypeError): pass correct = 0 total = len(questions) for i, q in enumerate(questions): user_ans = answers.get(str(i + 1), '') if user_ans.upper() == q.get('correct', '').upper(): correct += 1 score = int((correct / total) * 100) if total > 0 else 0 passing_score = module['passing_score'] or 70 passed = 1 if score >= passing_score else 0 # Get or create enrollment enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() if not enrollment: conn.execute("INSERT OR IGNORE INTO lms_enrollments (employee_id, elearning_module_id, status) VALUES (?,?, 'in_progress')", (emp_id, module_id)) conn.commit() enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() # Determine attempt number attempt = conn.execute("SELECT MAX(attempt_number) as m FROM lms_quiz_results WHERE enrollment_id=?", (enrollment['id'],)).fetchone() attempt_num = (attempt['m'] or 0) + 1 # Insert quiz result conn.execute("""INSERT INTO lms_quiz_results (enrollment_id, attempt_number, answers_json, total_questions, correct_answers, score, passed, submitted_at) VALUES (?,?,?,?,?,?,?,datetime('now'))""", (enrollment['id'], attempt_num, json.dumps(answers), total, correct, score, passed)) # Update enrollment new_status = 'completed' if passed else 'failed' conn.execute("""UPDATE lms_enrollments SET progress_percent=100, final_score=?, final_passed=?, completed_at=datetime('now'), status=?, last_accessed_at=datetime('now') WHERE id=?""", (score, passed, new_status, enrollment['id'])) # Generate certificate if passed cert_ref = None if passed: cert_ref = f"JvG-LMS-{module_id:03d}-{emp_id:04d}-{uuid.uuid4().hex[:6].upper()}" conn.execute("""INSERT INTO trn_certificates (employee_id, module_id, quiz_score, passed, certificate_ref) VALUES (?,?,?,?,?)""", (emp_id, module_id, score, 1, cert_ref)) # Update certifications table if training_program linked compliance_rule = conn.execute("SELECT training_program_id FROM lms_compliance_rules WHERE elearning_module_id=?", (module_id,)).fetchone() if compliance_rule and compliance_rule['training_program_id']: tp_id = compliance_rule['training_program_id'] # Get valid_period from training_programs tp = conn.execute("SELECT valid_period_days FROM training_programs WHERE id=?", (tp_id,)).fetchone() vdays = tp['valid_period_days'] if tp else 365 expiry = str(date.today() + timedelta(days=vdays)) existing = conn.execute("SELECT id FROM certifications WHERE employee_id=? AND training_program_id=?", (emp_id, tp_id)).fetchone() if existing: conn.execute("UPDATE certifications SET expiry_date=?, status='active', score=?, updated_at=datetime('now') WHERE id=?", (expiry, score, existing['id'])) else: conn.execute("INSERT INTO certifications (employee_id, training_program_id, expiry_date, status, score) VALUES (?,?,?,?,?)", (emp_id, tp_id, expiry, 'active', score)) conn.commit() finally: conn.close() return jsonify(success=True, score=score, total_questions=total, correct_answers=correct, passed=bool(passed), passing_score=passing_score, certificate_ref=cert_ref) # ═══════════════════════════════════════════════════════════════════════════════ # EXTERNAL SCORE API (for existing HTML quiz files) # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/api/score', methods=['POST']) def api_score(): """External score endpoint — called by existing HTML quiz files via fetch/XHR.""" data = request.get_json(silent=True) or {} emp_id = data.get('employee_id', _current_employee_id()) module_id = data.get('module_id') score = data.get('score', 0) total_q = data.get('total_questions', 0) correct = data.get('correct_answers', 0) answers = data.get('answers', {}) time_spent = data.get('time_spent', 0) if not module_id: return jsonify(success=False, error="module_id verplicht"), 400 conn = get_db() try: module = conn.execute("SELECT id, passing_score FROM elearning_modules WHERE id=?", (module_id,)).fetchone() if not module: return jsonify(success=False, error="Module niet gevonden"), 404 passing_score = module['passing_score'] or 70 passed = 1 if score >= passing_score else 0 # Get or create enrollment conn.execute("INSERT OR IGNORE INTO lms_enrollments (employee_id, elearning_module_id, status) VALUES (?,?, 'in_progress')", (emp_id, module_id)) conn.commit() enrollment = conn.execute("SELECT * FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=?", (emp_id, module_id)).fetchone() # Attempt number attempt = conn.execute("SELECT MAX(attempt_number) as m FROM lms_quiz_results WHERE enrollment_id=?", (enrollment['id'],)).fetchone() attempt_num = (attempt['m'] or 0) + 1 # Insert quiz result conn.execute("""INSERT INTO lms_quiz_results (enrollment_id, attempt_number, answers_json, total_questions, correct_answers, score, passed, time_spent_seconds, submitted_at) VALUES (?,?,?,?,?,?,?,?,datetime('now'))""", (enrollment['id'], attempt_num, json.dumps(answers), total_q, correct, score, passed, time_spent)) # Update enrollment new_status = 'completed' if passed else 'failed' conn.execute("""UPDATE lms_enrollments SET progress_percent=100, final_score=?, final_passed=?, completed_at=datetime('now'), status=?, time_spent_seconds=time_spent_seconds+?, last_accessed_at=datetime('now') WHERE id=?""", (score, passed, new_status, time_spent, enrollment['id'])) # Certificate cert_ref = None if passed: cert_ref = f"JvG-LMS-{module_id:03d}-{emp_id:04d}-{uuid.uuid4().hex[:6].upper()}" conn.execute("""INSERT INTO trn_certificates (employee_id, module_id, quiz_score, passed, certificate_ref) VALUES (?,?,?,?,?)""", (emp_id, module_id, score, 1, cert_ref)) # Compliance link cr = conn.execute("SELECT training_program_id FROM lms_compliance_rules WHERE elearning_module_id=?", (module_id,)).fetchone() if cr and cr['training_program_id']: tp_id = cr['training_program_id'] tp = conn.execute("SELECT valid_period_days FROM training_programs WHERE id=?", (tp_id,)).fetchone() vdays = tp['valid_period_days'] if tp else 365 expiry = str(date.today() + timedelta(days=vdays)) existing = conn.execute("SELECT id FROM certifications WHERE employee_id=? AND training_program_id=?", (emp_id, tp_id)).fetchone() if existing: conn.execute("UPDATE certifications SET expiry_date=?, status='active', score=? WHERE id=?", (expiry, score, existing['id'])) else: conn.execute("INSERT INTO certifications (employee_id, training_program_id, expiry_date, status, score) VALUES (?,?,?,?,?)", (emp_id, tp_id, expiry, 'active', score)) conn.commit() finally: conn.close() return jsonify(success=True, score=score, passed=bool(passed), certificate_ref=cert_ref) # ═══════════════════════════════════════════════════════════════════════════════ # JSON API ENDPOINTS # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/api/modules') def api_modules(): conn = get_db() try: rows = conn.execute("SELECT * FROM elearning_modules WHERE status='published' ORDER BY id").fetchall() finally: conn.close() return jsonify([dict(r) for r in rows]) @lms_bp.route('/api/modules/') def api_module_detail(module_id): conn = get_db() try: m = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (module_id,)).fetchone() if not m: return jsonify(error="Not found"), 404 content = conn.execute("SELECT * FROM lms_course_content WHERE elearning_module_id=?", (module_id,)).fetchall() finally: conn.close() result = dict(m) result['content_files'] = [dict(c) for c in content] return jsonify(result) @lms_bp.route('/api/enrollments/') def api_enrollments(employee_id): conn = get_db() try: rows = conn.execute(""" SELECT e.*, m.title, m.category FROM lms_enrollments e JOIN elearning_modules m ON m.id = e.elearning_module_id WHERE e.employee_id=? ORDER BY e.status """, (employee_id,)).fetchall() finally: conn.close() return jsonify([dict(r) for r in rows]) @lms_bp.route('/api/compliance/summary') def api_compliance_summary(): conn = get_db() try: employees = conn.execute("SELECT id, first_name, last_name, role_id FROM employees WHERE status='active'").fetchall() total = len(employees) compliant = 0 for emp in employees: rules = conn.execute(""" SELECT cr.elearning_module_id, cr.valid_period_days, cr.is_mandatory FROM lms_compliance_rules cr WHERE cr.is_mandatory=1 """).fetchall() if not rules: compliant += 1 continue all_ok = True for rule in rules: enr = conn.execute("""SELECT final_passed, completed_at FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=? ORDER BY completed_at DESC LIMIT 1""", (emp['id'], rule['elearning_module_id'])).fetchone() if not enr or not enr['final_passed']: all_ok = False break expiry = datetime.strptime(enr['completed_at'], '%Y-%m-%d %H:%M:%S') + timedelta(days=rule['valid_period_days']) if expiry.date() < date.today(): all_ok = False break if all_ok: compliant += 1 finally: conn.close() pct = int((compliant / total) * 100) if total > 0 else 0 return jsonify(total_employees=total, compliant=compliant, non_compliant=total - compliant, compliance_pct=pct) @lms_bp.route('/api/certificates/') def api_certificates(employee_id): conn = get_db() try: rows = conn.execute(""" SELECT c.*, m.title FROM trn_certificates c JOIN elearning_modules m ON m.id = c.module_id WHERE c.employee_id=? """, (employee_id,)).fetchall() finally: conn.close() return jsonify([dict(r) for r in rows]) # ═══════════════════════════════════════════════════════════════════════════════ # ADMIN PORTAL — COURSES # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/admin/courses') def admin_courses(): # Scan projects for deliverables import os projects_dir = '/root/projects/jg' projects = [] if os.path.isdir(projects_dir): for d in sorted(os.listdir(projects_dir)): full = os.path.join(projects_dir, d) if not os.path.isdir(full) or d.startswith('.'): continue deliv_dir = os.path.join(full, 'deliverables') if os.path.isdir(deliv_dir): files = [] for ext, label in [('.html','HTML'),('.pptx','PPTX'),('.zip','SCORM')]: for root, dirs, fnames in os.walk(deliv_dir): if '/archive/' in root: continue for f in fnames: if f.endswith(ext): rel = os.path.relpath(os.path.join(root, f), deliv_dir) files.append({'file': rel, 'type': label, 'name': f}) if files: projects.append({'dir': d, 'files': files}) proj_opts = ''.join(f'' for p in projects) proj_json = json.dumps(projects) conn = get_db() try: modules = conn.execute(""" SELECT m.*, (SELECT COUNT(*) FROM lms_enrollments WHERE elearning_module_id=m.id) as enroll_count, (SELECT ROUND(AVG(final_score)) FROM lms_enrollments WHERE elearning_module_id=m.id AND final_score IS NOT NULL) as avg_score FROM elearning_modules m ORDER BY m.id """).fetchall() finally: conn.close() rows = "" for m in modules: badge = {'published':'badge-success','draft':'badge-muted','archived':'badge-warning'}.get(m['status'], 'badge-muted') rows += f""" {m['id']} {m['title']} {m['category'] or '—'} {m['status']} {m['enroll_count']} {m['avg_score'] or '—'}% {m['passing_score'] or 70}% 📎 Content
""" body = f"""

Cursusbeheer

➕ Nieuwe Cursus Toevoegen
📂 Stap 1: Kies een bestaand project (optioneel)
{rows or ''}
IDTitelCategorieStatusInschrijvingenGem. ScoreSlagings%Acties
Geen cursussen
""" return _page('Cursusbeheer', body, 'admin-courses') @lms_bp.route('/admin/courses/new', methods=['POST']) def admin_course_create(): data = request.form title = data.get('title', '').strip() category = data.get('category', '').strip() duration = data.get('duration', 30, type=int) passing_score = data.get('passing_score', 70, type=int) content_path = data.get('content_path', '').strip() if not title or not category: return redirect(f"{BASE_PATH}/lms/admin/courses") conn = get_db() try: cur = conn.execute("""INSERT INTO elearning_modules (title, category, duration_minutes, passing_score, status, created_by) VALUES (?, ?, ?, ?, 'draft', 'admin')""", (title, category, duration, passing_score)) mod_id = cur.lastrowid conn.commit() # If content_path is an HTML file, read and store import os if content_path and content_path.endswith('.html') and os.path.isfile(content_path): with open(content_path, 'r', encoding='utf-8') as f: html_content = f.read() conn.execute("UPDATE elearning_modules SET content_html=? WHERE id=?", (html_content, mod_id)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/courses") @lms_bp.route('/admin/courses//content') def admin_course_content(course_id): # Scan projects for deliverables (same logic as course creation) projects_dir = '/root/projects/jg' projects = [] if os.path.isdir(projects_dir): for d in sorted(os.listdir(projects_dir)): full = os.path.join(projects_dir, d) if not os.path.isdir(full) or d.startswith('.'): continue deliv_dir = os.path.join(full, 'deliverables') if os.path.isdir(deliv_dir): files = [] for ext, label in [('.html','HTML'),('.pptx','PPTX'),('.zip','SCORM')]: for root, dirs, fnames in os.walk(deliv_dir): if '/archive/' in root: continue for f in fnames: if f.endswith(ext): rel = os.path.relpath(os.path.join(root, f), deliv_dir) files.append({'file': rel, 'type': label, 'name': f}) if files: projects.append({'dir': d, 'files': files}) proj_opts = ''.join(f'' for p in projects) proj_json = json.dumps(projects) conn = get_db() try: module = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (course_id,)).fetchone() content = conn.execute("SELECT * FROM lms_course_content WHERE elearning_module_id=? ORDER BY display_order", (course_id,)).fetchall() finally: conn.close() rows = "" for c in content: ctype_badge = {'html':'badge-success','pdf':'badge-danger','video':'badge-info','scorm':'badge-warning'}.get(c['content_type'],'badge-muted') rows += f""" {c['id']} {c['content_type'].upper()} {c['file_name'] or '—'} {c['file_path'][:60]}{'...' if len(c['file_path'] or '')>60 else ''} {'✅' if c['is_primary'] else '—'}
""" body = f"""

📎 Content: {module['title'] if module else 'Onbekend'}

📂 Kies een deliverable uit een bestaand project
{rows or ''}
IDTypeBestandPadPrimair
Geen content gekoppeld — selecteer hierboven een bestand
""" return _page(f'Content: {module["title"] if module else "?"}', body, 'admin-courses') @lms_bp.route('/admin/courses//content', methods=['POST']) def admin_course_content_add(course_id): data = request.form fpath = data.get('file_path', '') fname = data.get('file_name', '') or os.path.basename(fpath) ctype = data.get('content_type', 'html') conn = get_db() try: conn.execute("""INSERT INTO lms_course_content (elearning_module_id, content_type, file_path, file_name, is_primary) VALUES (?,?,?,?, (SELECT CASE WHEN (SELECT COUNT(*) FROM lms_course_content WHERE elearning_module_id=? AND is_primary=1)=0 THEN 1 ELSE 0 END))""", (course_id, ctype, fpath, fname, course_id)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/courses/{course_id}/content") @lms_bp.route('/admin/courses//content/', methods=['POST']) def admin_course_content_delete(course_id, content_id): conn = get_db() try: conn.execute("DELETE FROM lms_course_content WHERE id=? AND elearning_module_id=?", (content_id, course_id)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/courses/{course_id}/content") @lms_bp.route('/admin/courses//publish', methods=['POST']) def admin_course_publish(course_id): conn = get_db() try: conn.execute("UPDATE elearning_modules SET status='published' WHERE id=?", (course_id,)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/courses") @lms_bp.route('/admin/courses//delete', methods=['POST']) def admin_course_delete(course_id): conn = get_db() try: conn.execute("DELETE FROM lms_quiz_results WHERE enrollment_id IN (SELECT id FROM lms_enrollments WHERE elearning_module_id=?)", (course_id,)) conn.execute("DELETE FROM lms_enrollments WHERE elearning_module_id=?", (course_id,)) conn.execute("DELETE FROM lms_course_content WHERE elearning_module_id=?", (course_id,)) conn.execute("DELETE FROM lms_compliance_rules WHERE elearning_module_id=?", (course_id,)) conn.execute("DELETE FROM elearning_modules WHERE id=?", (course_id,)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/courses") # ═══════════════════════════════════════════════════════════════════════════════ # ADMIN PORTAL — ENROLLMENTS # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/admin/enrollments') def admin_enrollments(): conn = get_db() try: enrollments = conn.execute(""" SELECT e.*, em.first_name, em.last_name, em.employee_number, m.title as module_title FROM lms_enrollments e JOIN employees em ON em.id = e.employee_id JOIN elearning_modules m ON m.id = e.elearning_module_id ORDER BY e.status, e.due_date """).fetchall() modules = conn.execute("SELECT id, title FROM elearning_modules ORDER BY id").fetchall() employees = conn.execute("SELECT id, first_name, last_name FROM employees WHERE status='active' ORDER BY id").fetchall() roles = conn.execute("SELECT id, role_name FROM roles ORDER BY id").fetchall() finally: conn.close() rows = "" for e in enrollments: badge = {'not_started':'badge-muted','in_progress':'badge-info','completed':'badge-success','failed':'badge-danger','expired':'badge-warning','withdrawn':'badge-muted'}.get(e['status'],'badge-muted') rows += f""" {e['employee_number'] or e['employee_id']} {e['first_name']} {e['last_name']} {e['module_title']} {e['status'].replace('_',' ').title()} {e['final_score'] or '—'} {_deadline_html(e['due_date'] if 'due_date' in e.keys() else None)}
""" emp_opts = "".join(f'' for e in employees) mod_opts = "".join(f'' for m in modules) role_opts = "".join(f'' for r in roles) body = f"""

Inschrijvingen

Training Toewijzen

Ctrl+klik voor meerdere
{rows or ''}
Medew.NaamCursusStatusScoreDeadlineActie
Geen inschrijvingen
""" return _page('Inschrijvingen', body, 'admin-enroll') @lms_bp.route('/admin/enrollments', methods=['POST']) def admin_enrollment_create(): data = request.form module_id = data.get('module_id', type=int) emp_id = data.getlist('employee_id') due_date = data.get('due_date') or (date.today() + timedelta(days=30)).isoformat() if not module_id or not emp_id: return redirect(f"{BASE_PATH}/lms/admin/enrollments") conn = get_db() try: for eid in emp_id: conn.execute("INSERT OR IGNORE INTO lms_enrollments (employee_id, elearning_module_id, status, due_date) VALUES (?,?, 'not_started', ?)", (int(eid), module_id, due_date)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/enrollments") @lms_bp.route('/admin/enrollments/bulk', methods=['POST']) def admin_enrollment_bulk(): data = request.form module_id = data.get('module_id', type=int) role_id = data.get('role_id', type=int) due_date = data.get('due_date') or (date.today() + timedelta(days=30)).isoformat() if not module_id or not role_id: return redirect(f"{BASE_PATH}/lms/admin/enrollments") conn = get_db() try: emps = conn.execute("SELECT id FROM employees WHERE role_id=? AND status='active'", (role_id,)).fetchall() for e in emps: conn.execute("INSERT OR IGNORE INTO lms_enrollments (employee_id, elearning_module_id, status, due_date) VALUES (?,?, 'not_started', ?)", (e['id'], module_id, due_date)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/enrollments") @lms_bp.route('/admin/enrollments/', methods=['POST']) def admin_enrollment_delete(enrollment_id): conn = get_db() try: conn.execute("DELETE FROM lms_quiz_results WHERE enrollment_id=?", (enrollment_id,)) conn.execute("DELETE FROM lms_enrollments WHERE id=?", (enrollment_id,)) conn.commit() finally: conn.close() return redirect(f"{BASE_PATH}/lms/admin/enrollments") # ═══════════════════════════════════════════════════════════════════════════════ # ADMIN PORTAL — COMPLIANCE # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/admin/compliance') def admin_compliance(): conn = get_db() try: employees = conn.execute("SELECT e.id, e.first_name, e.last_name, e.role_id, r.role_name FROM employees e LEFT JOIN roles r ON r.id=e.role_id WHERE e.status='active' ORDER BY e.id").fetchall() modules = conn.execute("SELECT id, title FROM elearning_modules WHERE status='published' ORDER BY id").fetchall() rules = conn.execute("SELECT * FROM lms_compliance_rules").fetchall() rules_map = {r['elearning_module_id']: r for r in rules} finally: conn.close() # Build heatmap matrix: role × module role_modules = {} for m in modules: for e in employees: rid = e['role_id'] or 0 if rid not in role_modules: role_modules[rid] = {'role_name': e['role_name'] or f'Rol {rid}', 'cells': {}} # Calculate compliance per employee conn = get_db() try: emp_status = {} for e in employees: emp_status[e['id']] = {} for m in modules: rule = rules_map.get(m['id']) is_mandatory = rule and rule['is_mandatory'] enr = conn.execute("SELECT final_passed, completed_at FROM lms_enrollments WHERE employee_id=? AND elearning_module_id=? ORDER BY completed_at DESC LIMIT 1", (e['id'], m['id'])).fetchone() if not is_mandatory: status = 'na' elif not enr or not enr['final_passed']: status = 'red' elif enr['completed_at']: vdays = rule['valid_period_days'] if rule else 365 try: comp_date = datetime.strptime(enr['completed_at'], '%Y-%m-%d %H:%M:%S') except ValueError: comp_date = datetime.strptime(enr['completed_at'][:10], '%Y-%m-%d') expiry = comp_date + timedelta(days=vdays) if expiry.date() < date.today(): status = 'red' elif expiry.date() < date.today() + timedelta(days=30): status = 'amber' else: status = 'green' else: status = 'red' emp_status[e['id']][m['id']] = status finally: conn.close() # Compliance summary cards total_mandatory = 0 compliant_count = 0 for e in employees: for m in modules: rule = rules_map.get(m['id']) if rule and rule['is_mandatory']: total_mandatory += 1 if emp_status[e['id']][m['id']] == 'green': compliant_count += 1 pct = int((compliant_count / total_mandatory) * 100) if total_mandatory > 0 else 100 # Build employee detail table emp_rows = "" for e in employees: cells = "" for m in modules: s = emp_status[e['id']].get(m['id'], 'na') color = {'green':'green','amber':'amber','red':'red','na':''}.get(s, '') label = {'green':'✅','amber':'⚠️','red':'❌','na':'—'}.get(s, '—') cells += f'{label}' emp_rows += f'{e["first_name"]} {e["last_name"]}{e["role_name"] or "—"}{cells}' mod_headers = "".join(f'{m["title"][:20]}' for m in modules) body = f"""
🛡️
Compliance
{pct}%
📋
Verplichte Trainingen
{compliant_count}/{total_mandatory}

Compliance Matrix (Medewerker × Training)

{mod_headers} {emp_rows}
MedewerkerFunctie

✅ Actief   ⚠️ Vervalt binnen 30 dagen   ❌ Verlopen/Ontbreekt   — Niet verplicht

""" return _page('Compliance', body, 'admin-compliance') @lms_bp.route('/admin/compliance/matrix') def admin_compliance_matrix(): return redirect(f"{BASE_PATH}/lms/admin/compliance") @lms_bp.route('/admin/compliance/expired') def admin_compliance_expired(): conn = get_db() try: expired = conn.execute(""" SELECT e.first_name, e.last_name, e.employee_number, m.title, enr.completed_at, cr.valid_period_days, date(enr.completed_at, '+' || cr.valid_period_days || ' days') as expiry_date FROM lms_enrollments enr JOIN employees e ON e.id=enr.employee_id JOIN elearning_modules m ON m.id=enr.elearning_module_id LEFT JOIN lms_compliance_rules cr ON cr.elearning_module_id=enr.elearning_module_id WHERE enr.final_passed=1 AND cr.is_mandatory=1 AND date(enr.completed_at, '+' || cr.valid_period_days || ' days') < date('now') ORDER BY expiry_date """).fetchall() finally: conn.close() rows = "".join(f""" {e['employee_number'] or ''} {e['first_name']} {e['last_name']} {e['title']} Verlopen {e['expiry_date']} """ for e in expired) body = f"""

Verlopen Trainingen

{rows or ''}
NrNaamTrainingStatusVervaldatum
Geen verlopen trainingen
""" return _page('Verlopen Trainingen', body, 'admin-compliance') @lms_bp.route('/admin/compliance/alerts') def admin_compliance_alerts(): conn = get_db() try: alerts = conn.execute(""" SELECT e.first_name, e.last_name, e.employee_number, m.title, enr.completed_at, cr.valid_period_days, date(enr.completed_at, '+' || cr.valid_period_days || ' days') as expiry_date FROM lms_enrollments enr JOIN employees e ON e.id=enr.employee_id JOIN elearning_modules m ON m.id=enr.elearning_module_id LEFT JOIN lms_compliance_rules cr ON cr.elearning_module_id=enr.elearning_module_id WHERE enr.final_passed=1 AND cr.is_mandatory=1 AND date(enr.completed_at, '+' || cr.valid_period_days || ' days') BETWEEN date('now') AND date('now','+30 days') ORDER BY expiry_date """).fetchall() finally: conn.close() rows = "".join(f""" {a['employee_number'] or ''} {a['first_name']} {a['last_name']} {a['title']} Vervalt binnen 30 dagen {a['expiry_date']} """ for a in alerts) body = f"""

Compliance Alerts — Nakomend

{rows or ''}
NrNaamTrainingStatusVervaldatum
Geen nakomende vervallen
""" return _page('Compliance Alerts', body, 'admin-compliance') @lms_bp.route('/admin/compliance/rules', methods=['POST']) def admin_compliance_rules(): data = request.get_json(silent=True) or request.form module_id = data.get('elearning_module_id', type=int) or int(data.get('elearning_module_id', 0)) tp_id = data.get('training_program_id', type=int) or int(data.get('training_program_id', 0)) or None is_mandatory = data.get('is_mandatory', 0) auto_assign = data.get('auto_assign', 0) valid_days = data.get('valid_period_days', 365) if not module_id: return jsonify(success=False, error="module_id verplicht"), 400 conn = get_db() try: conn.execute("""INSERT INTO lms_compliance_rules (elearning_module_id, training_program_id, is_mandatory, auto_assign, valid_period_days, updated_at) VALUES (?,?,?,?,?,datetime('now'))""", (module_id, tp_id, is_mandatory, auto_assign, valid_days)) conn.commit() finally: conn.close() return jsonify(success=True) # ═══════════════════════════════════════════════════════════════════════════════ # ADMIN PORTAL — ANALYTICS # ═══════════════════════════════════════════════════════════════════════════════ @lms_bp.route('/admin/analytics') def admin_analytics(): conn = get_db() try: total_enroll = conn.execute("SELECT COUNT(*) as c FROM lms_enrollments").fetchone()['c'] completed = conn.execute("SELECT COUNT(*) as c FROM lms_enrollments WHERE status='completed'").fetchone()['c'] failed = conn.execute("SELECT COUNT(*) as c FROM lms_enrollments WHERE status='failed'").fetchone()['c'] in_progress = conn.execute("SELECT COUNT(*) as c FROM lms_enrollments WHERE status='in_progress'").fetchone()['c'] avg_score = conn.execute("SELECT ROUND(AVG(final_score),1) as s FROM lms_enrollments WHERE final_score IS NOT NULL").fetchone()['s'] or 0 pass_rate = int((completed / total_enroll) * 100) if total_enroll > 0 else 0 # Compliance summary total_emp = conn.execute("SELECT COUNT(*) as c FROM employees WHERE status='active'").fetchone()['c'] comp_pct = 0 # placeholder until compliance rules are configured # Per-course stats course_stats = conn.execute(""" SELECT m.id, m.title, COUNT(e.id) as enrollments, SUM(CASE WHEN e.status='completed' THEN 1 ELSE 0 END) as completed, SUM(CASE WHEN e.final_passed=1 THEN 1 ELSE 0 END) as passed, ROUND(AVG(CASE WHEN e.final_score IS NOT NULL THEN e.final_score END),1) as avg_score FROM elearning_modules m LEFT JOIN lms_enrollments e ON e.elearning_module_id=m.id GROUP BY m.id ORDER BY m.id """).fetchall() finally: conn.close() rows = "" for cs in course_stats: total = cs['enrollments'] or 0 comp_count = cs['completed'] or 0 rate = int((comp_count / total) * 100) if total > 0 else 0 rows += f""" {cs['title']} {total} {comp_count} {rate}% {cs['avg_score'] or '—'}% Details """ body = f"""
👥
Totaal Inschrijvingen
{total_enroll}
✅
Voltooid
{completed}
⏳
In Uitvoering
{in_progress}
❌
Gezakten
{failed}
📊
Gemiddelde Score
{avg_score}%
🎯
Slagingspercentage
{pass_rate}%

Per Cursus

{rows or ''}
CursusInschrijvingenVoltooidCompletionGem. ScoreDetails
Geen data
""" return _page('Analytics', body, 'admin-analytics') @lms_bp.route('/admin/analytics/course/') def admin_analytics_course(course_id): conn = get_db() try: module = conn.execute("SELECT * FROM elearning_modules WHERE id=?", (course_id,)).fetchone() enrollments = conn.execute(""" SELECT e.*, em.first_name, em.last_name, em.employee_number FROM lms_enrollments e JOIN employees em ON em.id = e.employee_id WHERE e.elearning_module_id=? ORDER BY e.status, em.last_name """, (course_id,)).fetchall() quiz_results = conn.execute(""" SELECT qr.*, em.first_name, em.last_name FROM lms_quiz_results qr JOIN lms_enrollments enr ON enr.id = qr.enrollment_id JOIN employees em ON em.id = enr.employee_id WHERE enr.elearning_module_id=? ORDER BY qr.submitted_at DESC LIMIT 50 """, (course_id,)).fetchall() finally: conn.close() rows = "" for e in enrollments: badge = {'not_started':'badge-muted','in_progress':'badge-info','completed':'badge-success','failed':'badge-danger'}.get(e['status'],'badge-muted') rows += f""" {e['employee_number'] or ''} {e['first_name']} {e['last_name']} {e['status'].replace('_',' ').title()} {e['final_score'] or '—'} {e['progress_percent'] or 0}% """ qr_rows = "" for qr in quiz_results: qr_rows += f""" {qr['first_name']} {qr['last_name']} Poging {qr['attempt_number']} {qr['score']}% {qr['correct_answers']}/{qr['total_questions']} {'✅' if qr['passed'] else '❌'} {qr['submitted_at'] or '—'} """ title = module['title'] if module else 'Onbekend' body = f"""

{title} — Cursusstatistieken

Inschrijvingen

{rows or ''}
NrNaamStatusScoreVoortgang
Geen inschrijvingen

Recente Quiz-resultaten

{qr_rows or ''}
MedewerkerPogingScoreCorrectResultaatDatum
Geen quiz-resultaten
← Terug naar Analytics""" return _page(f'Analytics: {title}', body, 'admin-analytics') @lms_bp.route('/admin/analytics/employee/') def admin_analytics_employee(employee_id): conn = get_db() try: emp = conn.execute("SELECT * FROM employees WHERE id=?", (employee_id,)).fetchone() enrollments = conn.execute(""" SELECT e.*, m.title FROM lms_enrollments e JOIN elearning_modules m ON m.id = e.elearning_module_id WHERE e.employee_id=? ORDER BY e.status """, (employee_id,)).fetchall() certs = conn.execute(""" SELECT c.*, m.title FROM trn_certificates c JOIN elearning_modules m ON m.id = c.module_id WHERE c.employee_id=? ORDER BY c.id DESC """, (employee_id,)).fetchall() finally: conn.close() name = f"{emp['first_name']} {emp['last_name']}" if emp else 'Onbekend' rows = "".join(f""" {e['title']} {e['status'].replace('_',' ').title()} {e['progress_percent'] or 0}% {e['final_score'] or '—'} """ for e in enrollments) cert_rows = "".join(f""" {c['title']}{c['quiz_score']}%{c['certificate_ref'] or '—'} """ for c in certs) body = f"""

{name} — Training Historie

Inschrijvingen

{rows or ''}
TrainingStatusVoortgangScore
Geen trainingen

Certificaten

{cert_rows or ''}
TrainingScoreReferentie
Geen certificaten
""" return _page(f'{name}', body, 'admin-analytics') # ═══════════════════════════════════════════════════════════════════════════════ # REGISTRATION HELPER # ═══════════════════════════════════════════════════════════════════════════════ def register_lms_routes(app, page_fn=None, BASE_PATH="/hseq-dashboard"): """Register LMS blueprint on Flask app. Call from app.py.""" global _page_fn if page_fn: _page_fn = page_fn bp = os.environ.get('BASE_PATH', '/hseq-dashboard') app.register_blueprint(lms_bp, url_prefix=BASE_PATH + '/lms') # Also register the API score endpoint (external quiz HTML posts here) @app.route(BASE_PATH + '/api/lms/score', methods=['POST']) def api_lms_score_proxy(): return api_score()