# ============================================================ # FASE 2: Training & Opleiding Dienst Transformatie # HSEQ Intelligence Dashboard # ============================================================ # Modules: # 1. Training Matrix Upgrade (AI-scan, competency, calendar) # 2. Toolbox Talk Generator # 3. Training Deliverable Lifecycle Integration # ============================================================ import os import sqlite3 import json import uuid from datetime import datetime, timedelta, date from flask import jsonify, request, render_template from search import get_db, DB_PATH # ─── DB Init ───────────────────────────────────────────────────────────── def init_training_fase2_db(): conn = sqlite3.connect(DB_PATH) c = conn.cursor() # Competency assessments c.execute('''CREATE TABLE IF NOT EXISTS trn_competency_assessments ( id INTEGER PRIMARY KEY AUTOINCREMENT, employee_id INTEGER, competency TEXT NOT NULL, level INTEGER DEFAULT 1, target_level INTEGER DEFAULT 3, assessor TEXT DEFAULT 'system', notes TEXT, assessed_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (employee_id) REFERENCES employees(id) )''') # Training calendar entries c.execute('''CREATE TABLE IF NOT EXISTS trn_calendar ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, training_program_id INTEGER, description TEXT, location TEXT, trainer TEXT, max_participants INTEGER, scheduled_date DATE, end_date DATE, recurrence TEXT, status TEXT DEFAULT 'planned', created_by TEXT DEFAULT 'system', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (training_program_id) REFERENCES training_programs(id) )''') # Toolbox talks c.execute('''CREATE TABLE IF NOT EXISTS trn_toolbox_talks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, theme TEXT NOT NULL, target_audience TEXT, risk_context TEXT, content_html TEXT, duration_minutes INTEGER DEFAULT 15, discussion_points TEXT, deliverable_id INTEGER, status TEXT DEFAULT 'draft', created_by TEXT DEFAULT 'system', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP )''') # Toolbox talk sessions (attendance tracking) c.execute('''CREATE TABLE IF NOT EXISTS trn_toolbox_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, talk_id INTEGER NOT NULL, session_date DATE NOT NULL, facilitator TEXT, attendees TEXT, notes TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (talk_id) REFERENCES trn_toolbox_talks(id) )''') # Quiz attempts with certificate linkage c.execute('''CREATE TABLE IF NOT EXISTS trn_certificates ( id INTEGER PRIMARY KEY AUTOINCREMENT, employee_id INTEGER, module_id INTEGER, assignment_id INTEGER, quiz_score INTEGER, passed INTEGER DEFAULT 0, certificate_ref TEXT UNIQUE, issued_at DATETIME DEFAULT CURRENT_TIMESTAMP, valid_until DATE, deliverable_id INTEGER, FOREIGN KEY (employee_id) REFERENCES employees(id) )''') # AI Training recommendations c.execute('''CREATE TABLE IF NOT EXISTS trn_recommendations ( id INTEGER PRIMARY KEY AUTOINCREMENT, employee_id INTEGER, training_program_id INTEGER, priority TEXT DEFAULT 'medium', reason TEXT, source TEXT DEFAULT 'ai_scan', status TEXT DEFAULT 'open', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (employee_id) REFERENCES employees(id), FOREIGN KEY (training_program_id) REFERENCES training_programs(id) )''') # Risk-based training matrix entries c.execute('''CREATE TABLE IF NOT EXISTS trn_risk_training_map ( id INTEGER PRIMARY KEY AUTOINCREMENT, risk_category TEXT NOT NULL, risk_level TEXT NOT NULL, role_id INTEGER, location TEXT, required_training_ids TEXT, refresher_months INTEGER DEFAULT 12, notes TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP )''') # Indexes for idx in [ 'idx_trn_comp_emp', 'idx_trn_cal_date', 'idx_trn_tb_status', 'idx_trn_cert_emp', 'idx_trn_rec_emp', 'idx_trn_rtm_risk' ]: try: pass # indexes handled below except: pass c.execute('CREATE INDEX IF NOT EXISTS idx_trn_comp_emp ON trn_competency_assessments(employee_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_trn_cal_date ON trn_calendar(scheduled_date)') c.execute('CREATE INDEX IF NOT EXISTS idx_trn_tb_status ON trn_toolbox_talks(status)') c.execute('CREATE INDEX IF NOT EXISTS idx_trn_cert_emp ON trn_certificates(employee_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_trn_rec_emp ON trn_recommendations(employee_id, status)') c.execute('CREATE INDEX IF NOT EXISTS idx_trn_rtm_risk ON trn_risk_training_map(risk_category, risk_level)') conn.commit() conn.close() print("[OK] FASE 2 training tables initialized") # ─── Helper: Link to Deliverable Lifecycle ─────────────────────────────── def _link_to_lifecycle(title, dtype, service, content=None, params=None): """Create a deliverable lifecycle entry and return the ID.""" try: from deliverable_lifecycle import DELIVERABLE_TYPES, DELIVERABLE_SERVICES if dtype not in DELIVERABLE_TYPES or service not in DELIVERABLE_SERVICES: return None except ImportError: return None db = get_db() try: cur = db.execute( '''INSERT INTO dlv_lifecycle (title, type, service, status, content, parameters, created_by) VALUES (?,?,?,?,?,'system','training_module')''', (title, dtype, service, 'draft', json.dumps(content) if content else None, json.dumps(params) if params else None) ) db.commit() lid = cur.lastrowid # Record creation approval db.execute('INSERT INTO dlv_approvals (lifecycle_id, action, reviewer, comment) VALUES (?,?,?,?)', (lid, 'created', 'system', 'Auto-generated by FASE 2 Training Module')) db.commit() return lid except Exception as e: print(f"[WARN] _link_to_lifecycle: {e}") return None finally: db.close() # ─── 1. AI Training Scan ──────────────────────────────────────────────── def _ai_training_scan(): """Analyze current training status and generate AI-powered recommendations.""" db = get_db() try: # Get all active employees with their roles and required training employees = db.execute(''' SELECT e.id, e.first_name, e.last_name, e.department, e.location, e.role_id, r.role_name FROM employees e LEFT JOIN roles r ON e.role_id = r.id WHERE e.status = 'active' ''').fetchall() # Get training matrix requirements matrix = db.execute(''' SELECT tm.role_id, tm.training_program_id, tp.program_name, tp.program_code, tp.is_mandatory, tp.valid_period_days FROM training_matrix tm JOIN training_programs tp ON tm.training_program_id = tp.id ''').fetchall() # Get existing certifications today = datetime.now().strftime('%Y-%m-%d') future90 = (datetime.now() + timedelta(days=90)).strftime('%Y-%m-%d') recommendations = [] risk_training = [] # Build role -> required training map role_training = {} for m in matrix: rid = m['role_id'] if rid not in role_training: role_training[rid] = [] role_training[rid].append(m) # Analyze each employee for emp in employees: eid = emp['id'] required = role_training.get(emp['role_id'], []) for req in required: # Check if cert exists and is valid cert = db.execute(''' SELECT expiry_date FROM certifications WHERE employee_id = ? AND training_program_id = ? ORDER BY expiry_date DESC LIMIT 1 ''', (eid, req['training_program_id'])).fetchone() priority = 'low' reason = '' if not cert: priority = 'high' reason = f"Ontbrekende verplichte training: {req['program_name']}" risk_training.append({ 'employee_id': eid, 'employee_name': f"{emp['first_name']} {emp['last_name']}", 'role': emp['role_name'], 'department': emp['department'], 'location': emp['location'], 'training': req['program_name'], 'training_id': req['training_program_id'], 'priority': priority, 'reason': reason, 'status': 'missing' }) elif cert and cert['expiry_date'] and cert['expiry_date'] < today: priority = 'critical' reason = f"Verlopen training: {req['program_name']} (vervallen op {cert['expiry_date']})" risk_training.append({ 'employee_id': eid, 'employee_name': f"{emp['first_name']} {emp['last_name']}", 'role': emp['role_name'], 'department': emp['department'], 'location': emp['location'], 'training': req['program_name'], 'training_id': req['training_program_id'], 'priority': priority, 'reason': reason, 'status': 'expired' }) elif cert and cert['expiry_date'] and cert['expiry_date'] < future90: priority = 'medium' reason = f"Binnenkort vervallend: {req['program_name']} (vervalt op {cert['expiry_date']})" risk_training.append({ 'employee_id': eid, 'employee_name': f"{emp['first_name']} {emp['last_name']}", 'role': emp['role_name'], 'department': emp['department'], 'location': emp['location'], 'training': req['program_name'], 'training_id': req['training_program_id'], 'priority': priority, 'reason': reason, 'status': 'expiring' }) if priority in ('high', 'critical'): recommendations.append({ 'employee_id': eid, 'training_program_id': req['training_program_id'], 'priority': priority, 'reason': reason, 'source': 'ai_scan' }) # Location-based risk analysis locations = db.execute('SELECT DISTINCT location FROM employees WHERE status="active" AND location IS NOT NULL').fetchall() location_risks = [] for loc in locations: loc_employees = [r for r in risk_training if r.get('location') == loc['location']] if loc_employees: location_risks.append({ 'location': loc['location'], 'total_gaps': len(loc_employees), 'critical': len([r for r in loc_employees if r['priority'] == 'critical']), 'high': len([r for r in loc_employees if r['priority'] == 'high']), 'medium': len([r for r in loc_employees if r['priority'] == 'medium']) }) # Summary stats summary = { 'total_employees_scanned': len(employees), 'total_gaps': len(risk_training), 'critical': len([r for r in risk_training if r['priority'] == 'critical']), 'high': len([r for r in risk_training if r['priority'] == 'high']), 'medium': len([r for r in risk_training if r['priority'] == 'medium']), 'locations_affected': len(location_risks) } return jsonify({ 'summary': summary, 'gaps': risk_training, 'location_analysis': location_risks, 'recommendations': recommendations, 'scanned_at': datetime.now().isoformat() }) finally: db.close() # ─── 2. Competency Assessment ──────────────────────────────────────────── def _competency_assessments_list(): db = get_db() try: eid = request.args.get('employee_id') if eid: rows = db.execute(''' SELECT ca.*, e.first_name, e.last_name FROM trn_competency_assessments ca JOIN employees e ON ca.employee_id = e.id WHERE ca.employee_id = ? ORDER BY ca.assessed_at DESC ''', (eid,)).fetchall() else: rows = db.execute(''' SELECT ca.*, e.first_name, e.last_name FROM trn_competency_assessments ca JOIN employees e ON ca.employee_id = e.id ORDER BY ca.assessed_at DESC LIMIT 200 ''').fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _competency_assessment_create(): data = request.get_json(force=True) if not data.get('employee_id') or not data.get('competency'): return jsonify({'error': 'employee_id and competency required'}), 400 db = get_db() try: db.execute('''INSERT INTO trn_competency_assessments (employee_id, competency, level, target_level, assessor, notes) VALUES (?,?,?,?,?,?)''', (data['employee_id'], data['competency'], data.get('level', 1), data.get('target_level', 3), data.get('assessor', 'system'), data.get('notes'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() def _competency_assessment_employee(eid): """Get competency profile for an employee with gap analysis.""" db = get_db() try: assessments = db.execute(''' SELECT competency, level, target_level, assessed_at FROM trn_competency_assessments WHERE employee_id = ? ORDER BY assessed_at DESC ''', (eid,)).fetchall() # Group by competency, take latest latest = {} for a in assessments: comp = a['competency'] if comp not in latest: latest[comp] = dict(a) latest[comp]['gap'] = a['target_level'] - a['level'] latest[comp]['status'] = 'on_track' if a['level'] >= a['target_level'] else 'gap' emp = db.execute('SELECT first_name, last_name, role_id, department FROM employees WHERE id=?', (eid,)).fetchone() return jsonify({ 'employee': dict(emp) if emp else None, 'competencies': list(latest.values()), 'total_gaps': len([v for v in latest.values() if v['status'] == 'gap']), 'total_on_track': len([v for v in latest.values() if v['status'] == 'on_track']) }) finally: db.close() # ─── 3. Training Calendar ──────────────────────────────────────────────── def _calendar_list(): db = get_db() try: from_date = request.args.get('from') to_date = request.args.get('to') status = request.args.get('status') q = '''SELECT tc.*, tp.program_name FROM trn_calendar tc LEFT JOIN training_programs tp ON tc.training_program_id = tp.id WHERE 1=1''' params = [] if from_date: q += ' AND tc.scheduled_date >= ?'; params.append(from_date) if to_date: q += ' AND tc.scheduled_date <= ?'; params.append(to_date) if status: q += ' AND tc.status = ?'; params.append(status) q += ' ORDER BY tc.scheduled_date ASC' rows = db.execute(q, params).fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _calendar_create(): data = request.get_json(force=True) if not data.get('title') or not data.get('scheduled_date'): return jsonify({'error': 'title and scheduled_date required'}), 400 db = get_db() try: db.execute('''INSERT INTO trn_calendar (title, training_program_id, description, location, trainer, max_participants, scheduled_date, end_date, recurrence, status, created_by) VALUES (?,?,?,?,?,?,?,?,?,?,?)''', (data['title'], data.get('training_program_id'), data.get('description'), data.get('location'), data.get('trainer'), data.get('max_participants'), data['scheduled_date'], data.get('end_date'), data.get('recurrence'), data.get('status', 'planned'), data.get('created_by', 'system'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() def _calendar_expiring(): """Get training calendar entries with upcoming expirations.""" db = get_db() try: today = datetime.now().strftime('%Y-%m-%d') future30 = (datetime.now() + timedelta(days=30)).strftime('%Y-%m-%d') future90 = (datetime.now() + timedelta(days=90)).strftime('%Y-%m-%d') expiring = db.execute(''' SELECT c.expiry_date, e.first_name || ' ' || e.last_name as employee_name, e.department, tp.program_name, tp.refresher_months FROM certifications c JOIN employees e ON c.employee_id = e.id JOIN training_programs tp ON c.training_program_id = tp.id WHERE c.expiry_date >= ? AND c.expiry_date <= ? AND e.status = 'active' ORDER BY c.expiry_date ASC ''', (today, future90)).fetchall() # Group by training program by_program = {} for exp in expiring: pname = exp['program_name'] if pname not in by_program: by_program[pname] = {'program': pname, 'valid_period_days': exp['valid_period_days'], 'expiring_count': 0, 'employees': []} by_program[pname]['expiring_count'] += 1 by_program[pname]['employees'].append({ 'name': exp['employee_name'], 'department': exp['department'], 'expiry_date': exp['expiry_date'] }) # Suggest calendar entries suggestions = [] for pname, info in by_program.items(): suggested_date = (datetime.now() + timedelta(days=7)).strftime('%Y-%m-%d') suggestions.append({ 'suggested_title': f"Herhaling: {pname}", 'suggested_date': suggested_date, 'participants_needed': info['expiring_count'], 'employees': info['employees'], 'valid_period_days': info['valid_period_days'] }) return jsonify({ 'expiring_90_days': [dict(r) for r in expiring], 'by_program': list(by_program.values()), 'calendar_suggestions': suggestions }) finally: db.close() # ─── 4. Toolbox Talk Generator ─────────────────────────────────────────── def _toolbox_talks_list(): db = get_db() try: status = request.args.get('status') theme = request.args.get('theme') q = 'SELECT * FROM trn_toolbox_talks WHERE 1=1' params = [] if status: q += ' AND status = ?'; params.append(status) if theme: q += ' AND theme LIKE ?'; params.append(f'%{theme}%') q += ' ORDER BY created_at DESC' rows = db.execute(q, params).fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _toolbox_talk_create(): """Generate a toolbox talk with AI content generation.""" data = request.get_json(force=True) if not data.get('title') or not data.get('theme'): return jsonify({'error': 'title and theme required'}), 400 db = get_db() try: # Generate structured content content_html = _generate_toolbox_content( data['title'], data['theme'], data.get('target_audience', 'Alle medewerkers'), data.get('risk_context', ''), data.get('duration_minutes', 15) ) discussion_points = json.dumps(data.get('discussion_points', [ f"Wat zijn de risico's in jouw werkgebied gerelateerd aan {data['theme']}?", "Welke veiligheidsmaatregelen zijn al getroffen?", "Wat kunnen we verbeteren?" ])) cur = db.execute('''INSERT INTO trn_toolbox_talks (title, theme, target_audience, risk_context, content_html, duration_minutes, discussion_points, created_by) VALUES (?,?,?,?,?,?,?,?)''', (data['title'], data['theme'], data.get('target_audience'), data.get('risk_context'), content_html, data.get('duration_minutes', 15), discussion_points, data.get('created_by', 'system'))) db.commit() talk_id = cur.lastrowid # Link to deliverable lifecycle dlv_id = _link_to_lifecycle( f"Toolbox Talk: {data['title']}", 'toolbox_talk', 'training', {'talk_id': talk_id, 'theme': data['theme']}, data ) if dlv_id: db.execute('UPDATE trn_toolbox_talks SET deliverable_id = ? WHERE id = ?', (dlv_id, talk_id)) db.commit() return jsonify({'status': 'created', 'id': talk_id, 'deliverable_id': dlv_id}), 201 finally: db.close() def _toolbox_talk_get(talk_id): db = get_db() try: talk = db.execute('SELECT * FROM trn_toolbox_talks WHERE id = ?', (talk_id,)).fetchone() if not talk: return jsonify({'error': 'Not found'}), 404 sessions = db.execute('SELECT * FROM trn_toolbox_sessions WHERE talk_id = ? ORDER BY session_date DESC', (talk_id,)).fetchall() return jsonify({'talk': dict(talk), 'sessions': [dict(s) for s in sessions]}) finally: db.close() def _toolbox_talk_publish(talk_id): """Publish a toolbox talk and link to deliverable lifecycle.""" db = get_db() try: talk = db.execute('SELECT * FROM trn_toolbox_talks WHERE id = ?', (talk_id,)).fetchone() if not talk: return jsonify({'error': 'Not found'}), 404 db.execute('UPDATE trn_toolbox_talks SET status = ?, updated_at = ? WHERE id = ?', ('published', datetime.now().isoformat(), talk_id)) # Update deliverable lifecycle if linked if talk['deliverable_id']: try: db.execute("UPDATE dlv_lifecycle SET status = 'published', published_at = ? WHERE id = ?", (datetime.now().isoformat(), talk['deliverable_id'])) db.execute('INSERT INTO dlv_approvals (lifecycle_id, action, reviewer, comment) VALUES (?,?,?,?)', (talk['deliverable_id'], 'published', 'system', 'Toolbox talk published')) except: pass db.commit() return jsonify({'status': 'published'}) finally: db.close() def _toolbox_session_create(talk_id): """Record a toolbox talk session with attendance.""" data = request.get_json(force=True) if not data.get('session_date'): return jsonify({'error': 'session_date required'}), 400 db = get_db() try: db.execute('''INSERT INTO trn_toolbox_sessions (talk_id, session_date, facilitator, attendees, notes) VALUES (?,?,?,?,?)''', (talk_id, data['session_date'], data.get('facilitator'), json.dumps(data.get('attendees', [])), data.get('notes'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() def _generate_toolbox_content(title, theme, audience, risk_context, duration): """Generate structured HTML content for a toolbox talk.""" return f'''

🛡️ Toolbox Talk

{title}

Thema: {theme} | Doelgroep: {audience} | Duur: {duration} min

📋 Doel

Deze toolbox talk richt zich op bewustwording en gedragsverandering rondom {theme}. Het doel is dat elke deelnemer na afloop de risico's herkent en de juiste veiligheidsmaatregelen kent.

⚠️ Risicobeschrijving

{risk_context if risk_context else "Zie specifieke risicoinventarisatie voor de actuele risico's gerelateerd aan " + theme + "."}

🛡️ Veiligheidsmaatregelen

  1. Volg altijd de geldende werkprocedures en instructies
  2. Gebruik de voorgeschreven persoonlijke beschermingsmiddelen (PBM)
  3. Meld onveilige situaties direct aan je leidinggevende
  4. Stop het werk bij twijfel — veiligheid gaat altijd voor

💬 Discussiepunten

📝 Aanwezigheidslijst

NaamFunctieHandtekening
   
   
   
   

Gegenereerd door HSEQ Intelligence Dashboard — Training & Opleiding Dienst
{datetime.now().strftime('%d-%m-%Y %H:%M')}

''' # ─── 5. Certificates & Quiz Results ────────────────────────────────────── def _certificates_list(): db = get_db() try: eid = request.args.get('employee_id') if eid: rows = db.execute(''' SELECT c.*, e.first_name, e.last_name FROM trn_certificates c JOIN employees e ON c.employee_id = e.id WHERE c.employee_id = ? ORDER BY c.issued_at DESC ''', (eid,)).fetchall() else: rows = db.execute(''' SELECT c.*, e.first_name, e.last_name FROM trn_certificates c JOIN employees e ON c.employee_id = e.id ORDER BY c.issued_at DESC LIMIT 200 ''').fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _certificate_issue(): """Issue a certificate after passing a quiz.""" data = request.get_json(force=True) if not data.get('employee_id') or not data.get('module_id'): return jsonify({'error': 'employee_id and module_id required'}), 400 if not data.get('passed'): return jsonify({'error': 'Certificate only issued for passed quizzes'}), 400 cert_ref = f"CERT-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:8].upper()}" # Calculate valid_until based on refresher months valid_months = data.get('valid_months', 12) valid_until = (datetime.now() + timedelta(days=valid_months * 30)).strftime('%Y-%m-%d') db = get_db() try: cur = db.execute('''INSERT INTO trn_certificates (employee_id, module_id, assignment_id, quiz_score, passed, certificate_ref, valid_until) VALUES (?,?,?,?,?,?,?)''', (data['employee_id'], data['module_id'], data.get('assignment_id'), data.get('quiz_score', 0), 1, cert_ref, valid_until)) db.commit() cert_id = cur.lastrowid # Link to deliverable lifecycle dlv_id = _link_to_lifecycle( f"Certificaat: {cert_ref}", 'training_module', 'training', {'certificate_id': cert_id, 'ref': cert_ref, 'employee_id': data['employee_id']}, data ) if dlv_id: db.execute('UPDATE trn_certificates SET deliverable_id = ? WHERE id = ?', (dlv_id, cert_id)) db.commit() return jsonify({'status': 'issued', 'id': cert_id, 'certificate_ref': cert_ref, 'valid_until': valid_until}), 201 finally: db.close() def _certificate_get(cert_id): db = get_db() try: cert = db.execute(''' SELECT c.*, e.first_name, e.last_name, e.department FROM trn_certificates c JOIN employees e ON c.employee_id = e.id WHERE c.id = ? ''', (cert_id,)).fetchone() if not cert: return jsonify({'error': 'Not found'}), 404 return jsonify(dict(cert)) finally: db.close() # ─── 6. Risk-Training Mapping ──────────────────────────────────────────── def _risk_training_map_list(): db = get_db() try: rows = db.execute('SELECT * FROM trn_risk_training_map ORDER BY risk_category, risk_level').fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _risk_training_map_create(): data = request.get_json(force=True) if not data.get('risk_category') or not data.get('risk_level'): return jsonify({'error': 'risk_category and risk_level required'}), 400 db = get_db() try: db.execute('''INSERT INTO trn_risk_training_map (risk_category, risk_level, role_id, location, required_training_ids, refresher_months, notes) VALUES (?,?,?,?,?,?,?)''', (data['risk_category'], data['risk_level'], data.get('role_id'), data.get('location'), json.dumps(data.get('required_training_ids', [])), data.get('refresher_months', 12), data.get('notes'))) db.commit() return jsonify({'status': 'created'}), 201 finally: db.close() # ─── 7. Recommendations CRUD ───────────────────────────────────────────── def _recommendations_list(): db = get_db() try: status = request.args.get('status', 'open') rows = db.execute(''' SELECT r.*, e.first_name, e.last_name, tp.program_name FROM trn_recommendations r LEFT JOIN employees e ON r.employee_id = e.id LEFT JOIN training_programs tp ON r.training_program_id = tp.id WHERE r.status = ? ORDER BY CASE r.priority WHEN 'critical' THEN 1 WHEN 'high' THEN 2 WHEN 'medium' THEN 3 ELSE 4 END, r.created_at DESC ''', (status,)).fetchall() return jsonify([dict(r) for r in rows]) finally: db.close() def _recommendations_save(): """Save AI scan recommendations to the database.""" data = request.get_json(force=True) recommendations = data.get('recommendations', []) if not recommendations: return jsonify({'error': 'No recommendations provided'}), 400 db = get_db() try: # Clear previous open recommendations from ai_scan db.execute("DELETE FROM trn_recommendations WHERE source = 'ai_scan' AND status = 'open'") saved = 0 for rec in recommendations: db.execute('''INSERT INTO trn_recommendations (employee_id, training_program_id, priority, reason, source) VALUES (?,?,?,?,?)''', (rec.get('employee_id'), rec.get('training_program_id'), rec.get('priority', 'medium'), rec.get('reason', ''), rec.get('source', 'ai_scan'))) saved += 1 db.commit() return jsonify({'status': 'saved', 'count': saved}) finally: db.close() def _recommendations_update(rec_id): data = request.get_json(force=True) db = get_db() try: db.execute("UPDATE trn_recommendations SET status = ? WHERE id = ?", (data.get('status', 'closed'), rec_id)) db.commit() return jsonify({'status': 'updated'}) finally: db.close() # ─── Route Registration ────────────────────────────────────────────────── TRAINING_FASE2_TEMPLATE = ''' {% extends "base.html" %} {% block content %}
← Training Dashboard

🎓 Training & Opleiding — FASE 2

AI-Powered

AI Training Compliance Scan

Scan alle medewerkers op training compliance gaps. Analyseert per rol, locatie en risico-expositie.

{% endblock %} ''' def register_training_fase2_routes(app, page, BASE_PATH="/hseq-dashboard"): init_training_fase2_db() @app.route(BASE_PATH + '/training-fase2') def training_fase2_page(): bp = os.environ.get('BASE_PATH', '') body = render_template('training_fase2.html', BASE_PATH=bp) return page(body, active='training-fase2', page_title='Training & Opleiding FASE 2') # ── AI Training Scan ── app.add_url_rule(BASE_PATH + '/api/training/fase2/ai-scan', view_func=_ai_training_scan, methods=['GET']) # ── Competency Assessments ── app.add_url_rule(BASE_PATH + '/api/training/fase2/competencies', view_func=_competency_assessments_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/competencies', view_func=_competency_assessment_create, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/competencies//profile', view_func=_competency_assessment_employee, methods=['GET']) # ── Training Calendar ── app.add_url_rule(BASE_PATH + '/api/training/fase2/calendar', view_func=_calendar_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/calendar', view_func=_calendar_create, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/calendar/expiring', view_func=_calendar_expiring, methods=['GET']) # ── Toolbox Talks ── app.add_url_rule(BASE_PATH + '/api/training/fase2/toolbox-talks', view_func=_toolbox_talks_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/toolbox-talks', view_func=_toolbox_talk_create, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/toolbox-talks/', view_func=_toolbox_talk_get, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/toolbox-talks//publish', view_func=_toolbox_talk_publish, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/toolbox-talks//sessions', view_func=_toolbox_session_create, methods=['POST']) # ── Certificates ── app.add_url_rule(BASE_PATH + '/api/training/fase2/certificates', view_func=_certificates_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/certificates', view_func=_certificate_issue, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/certificates/', view_func=_certificate_get, methods=['GET']) # ── Risk-Training Mapping ── app.add_url_rule(BASE_PATH + '/api/training/fase2/risk-map', view_func=_risk_training_map_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/risk-map', view_func=_risk_training_map_create, methods=['POST']) # ── Recommendations ── app.add_url_rule(BASE_PATH + '/api/training/fase2/recommendations', view_func=_recommendations_list, methods=['GET']) app.add_url_rule(BASE_PATH + '/api/training/fase2/recommendations/save', view_func=_recommendations_save, methods=['POST']) app.add_url_rule(BASE_PATH + '/api/training/fase2/recommendations/', view_func=_recommendations_update, methods=['PUT'])