#!/usr/bin/env python3 """ HSEQ Intelligence Hub V2.0 — AI Consultant Takes, Actieworkflow, Alerts Kas Backend Architecture — EA_02 Functies: - generate_consultant_take(): AI-powered HSEQ analyse via OpenRouter - intelligence_v2 tables: consultant_take_json, intelligence_actions, hseq_alerts - Dismiss/Action workflow endpoints - Dashboard alert integratie """ import json import sqlite3 import os import time from datetime import datetime, timedelta from flask import request, jsonify # ── Config ──────────────────────────────────────────────────────────────── OPENROUTER_URL = os.environ.get('OPENROUTER_URL', 'https://openrouter.ai/api/v1/chat/completions') OPENROUTER_KEY = os.environ.get('OPENROUTER_API_KEY', '') OPENROUTER_MODEL_CONSULTANT = 'google/gemini-2.0-flash-001' MAX_CONSULTANT_CALLS_PER_BATCH = 10 CONSULTANT_RATE_LIMIT_DELAY = 1.0 # seconds between calls def init_intelligence_v2_tables(db_path): """Create V2 tables if not exist, add consultant_take_json column to scraped_items.""" conn = sqlite3.connect(db_path) c = conn.cursor() # Add consultant_take_json to scraped_items try: c.execute("ALTER TABLE scraped_items ADD COLUMN consultant_take_json TEXT DEFAULT NULL") except: pass # Add dismissed column try: c.execute("ALTER TABLE scraped_items ADD COLUMN dismissed INTEGER DEFAULT 0") except: pass # Add risk_level column try: c.execute("ALTER TABLE scraped_items ADD COLUMN risk_level TEXT DEFAULT 'LAAG'") except: pass # Intelligence actions table c.execute("""CREATE TABLE IF NOT EXISTS intelligence_actions ( id INTEGER PRIMARY KEY AUTOINCREMENT, scraped_item_id INTEGER, action_type TEXT NOT NULL, action_data TEXT DEFAULT '{}', status TEXT DEFAULT 'open', created_at TEXT DEFAULT CURRENT_TIMESTAMP, closed_at TEXT DEFAULT NULL, title TEXT DEFAULT '', source TEXT DEFAULT '', url TEXT DEFAULT '' )""") # HSEQ Alerts table c.execute("""CREATE TABLE IF NOT EXISTS hseq_alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, message TEXT DEFAULT '', priority TEXT DEFAULT 'LAAG', source TEXT DEFAULT '', source_url TEXT DEFAULT '', is_read INTEGER DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP, scraped_item_id INTEGER DEFAULT NULL )""") conn.commit() conn.close() def generate_consultant_take(title, summary, source, category): """Generate AI consultant take using OpenRouter. Returns dict or None on failure.""" import urllib.request import urllib.error if not OPENROUTER_KEY: return None prompt = f"""Je bent een senior HSEQ consultant (Health, Safety, Environment & Quality) werkzaam in de Nederlandse olie-, gas- en petrochemische industrie. Analyseer het volgende HSEQ-nieuwsbericht en geef een praktisch advies: **Titel:** {title[:200]} **Bron:** {source} **Samenvatting:** {summary[:500]} **Categorie:** {category} Reageer ALLEEN in het volgende JSON-formaat (geen markdown, geen uitleg): {{"risk_level": "LAAG|MEDIUM|HOOG|KRITIEK", "consultant_take": "2-3 zinnen praktisch advies", "actions": ["actie 1", "actie 2", "actie 3"]}} Kies de risk_level op basis van: - KRITIEK: direct levensgevaar, boetes > €1M, onmiddellijke stillegging - HOOG: ernstig letsel risico, significante wetswijziging, milieuschade - MEDIUM: wijziging in regelgeving, best practices, industry alerts - LAAG: informatief, algemeen nieuws, lage impact""" try: payload = json.dumps({ "model": OPENROUTER_MODEL_CONSULTANT, "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "max_tokens": 500 }).encode('utf-8') req = urllib.request.Request(OPENROUTER_URL, data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {OPENROUTER_KEY}", "HTTP-Referer": "https://mescalinerabbit.shop", "X-Title": "HSEQ Intelligence Hub V2" }) resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read().decode('utf-8')) content = data['choices'][0]['message']['content'].strip() # Strip markdown code blocks if present if content.startswith('```'): content = content.split('\n', 1)[1] if '\n' in content else content[3:] content = content.rsplit('```', 1)[0] if '```' in content else content result = json.loads(content) # Validate risk_level rl = result.get('risk_level', 'LAAG').upper() if rl not in ('LAAG', 'MEDIUM', 'HOOG', 'KRITIEK'): rl = 'LAAG' result['risk_level'] = rl return result except Exception as e: print(f'[Intelligence V2] AI call failed: {e}') return None def generate_fallback_take(title, summary, source): """Rule-based fallback when AI is unavailable.""" text = (title + ' ' + summary).lower() if any(k in text for k in ['waarschuwing', 'alert', 'nood', 'ongeval', 'boete', 'dodelijk', 'explosie', 'brand', 'lekgift', 'blootstelling']): return {"risk_level": "HOOG", "consultant_take": "Dit item vereist directe aandacht. Neem contact op met de HSEQ manager voor evaluatie van de impact op de organisatie.", "actions": ["Evalueer impact op bedrijfsvoering", "Informeer betrokken afdelingen", "Update RI&E indien relevant"]} elif any(k in text for k in ['nieuwe regel', 'wet', 'verplicht', 'wijziging', 'besluit', 'richtlijn', 'wetgeving']): return {"risk_level": "MEDIUM", "consultant_take": "Wets- of regelwijziging gedetecteerd. Controleer of deze van toepassing is op de huidige activiteiten en vergunningen.", "actions": ["Controleer toepasselijkheid", "Evalueer compliance gap", "Plan implementatie"]} else: return {"risk_level": "LAAG", "consultant_take": "Informatief item. Bewaar voor referentie en monitor op verdere ontwikkelingen.", "actions": ["Archiveer als referentie", "Monitor op updates"]} def batch_generate_consultant_takes(db_path, limit=10): """Generate consultant takes for scraped items without one. Rate limited.""" conn = sqlite3.connect(db_path) rows = conn.execute(""" SELECT id, title, url, content, source_id, category FROM scraped_items WHERE consultant_take_json IS NULL AND dismissed = 0 ORDER BY id DESC LIMIT ? """, (limit,)).fetchall() if not rows: conn.close() return 0 generated = 0 for i, row in enumerate(rows): item_id, title, url, content, source, category = row take = generate_consultant_take( title or '', content or '', source or '', category or '' ) if take is None: take = generate_fallback_take(title or '', content or '', source or '') risk_level = take.get('risk_level', 'LAAG') conn.execute(""" UPDATE scraped_items SET consultant_take_json = ?, risk_level = ? WHERE id = ? """, (json.dumps(take, ensure_ascii=False), risk_level, item_id)) # Auto-create alert for HOOG/CRITICAL items if risk_level in ('HOOG', 'KRITIEK'): existing = conn.execute( "SELECT id FROM hseq_alerts WHERE scraped_item_id = ? AND priority = ?", (item_id, risk_level) ).fetchone() if not existing: conn.execute(""" INSERT INTO hseq_alerts (title, message, priority, source, source_url, scraped_item_id) VALUES (?, ?, ?, ?, ?, ?) """, (title[:200], take.get('consultant_take', ''), risk_level, source or '', url or '', item_id)) conn.commit() generated += 1 if i < len(rows) - 1: time.sleep(CONSULTANT_RATE_LIMIT_DELAY) conn.close() return generated def register_intelligence_v2_routes(app, page, BASE_PATH, DB_PATH): """Register all V2 API routes.""" init_intelligence_v2_tables(DB_PATH) # ── Generate Consultant Takes (batch) ───────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/generate', methods=['POST']) def api_intelligence_v2_generate(): """Generate AI consultant takes for items missing one.""" try: limit = min(request.json.get('limit', 10), MAX_CONSULTANT_CALLS_PER_BATCH) if request.is_json else MAX_CONSULTANT_CALLS_PER_BATCH count = batch_generate_consultant_takes(DB_PATH, limit=limit) return jsonify({'status': 'ok', 'generated': count}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 # ── Get items with consultant takes ────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/items') def api_intelligence_v2_items(): """Get scraped items with consultant takes, ordered by risk level.""" risk_filter = request.args.get('filter', '') try: conn = sqlite3.connect(DB_PATH) count_check = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE dismissed = 0 AND risk_level IN ('HOOG', 'KRITIEK')").fetchone()[0] print(f"[DEBUG] High-risk count in DB: {count_check}") if risk_filter == 'high-risk': rows = conn.execute(""" SELECT id, source_id, title, url, content, scraped_at, category, consultant_take_json, risk_level, dismissed FROM scraped_items WHERE dismissed = 0 AND risk_level IN ('HOOG', 'KRITIEK') ORDER BY id DESC LIMIT 50 """).fetchall() else: rows = conn.execute(""" SELECT id, source_id, title, url, content, scraped_at, category, consultant_take_json, risk_level, dismissed FROM scraped_items WHERE dismissed = 0 AND consultant_take_json IS NOT NULL ORDER BY CASE risk_level WHEN 'KRITIEK' THEN 1 WHEN 'HOOG' THEN 2 WHEN 'MEDIUM' THEN 3 ELSE 4 END, id DESC LIMIT 50 """).fetchall() items = [] for r in rows: take = {} if r[8]: try: take = json.loads(r[8]) except: take = {} items.append({ 'id': r[0], 'source': (r[1] or '').replace('_', ' ').title(), 'title': r[2][:200], 'url': r[3], 'summary': (r[4] or '')[:300], 'date': (r[5] or '')[:10], 'category': r[6] or '', 'consultant_take': take.get('consultant_take', ''), 'risk_level': r[9] or 'LAAG', 'actions': take.get('actions', []), }) conn.close() return jsonify({'items': items, 'count': len(items)}) except Exception as e: return jsonify({'items': [], 'count': 0, 'error': str(e)}) # ── Dismiss item ───────────────────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2//dismiss', methods=['POST']) def api_intelligence_v2_dismiss(item_id): try: conn = sqlite3.connect(DB_PATH) conn.execute("UPDATE scraped_items SET dismissed = 1 WHERE id = ?", (item_id,)) conn.commit() conn.close() return jsonify({'status': 'dismissed', 'id': item_id}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 # ── Create action from item ────────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2//action', methods=['POST']) def api_intelligence_v2_action(item_id): try: payload = request.get_json(force=True) action_type = payload.get('action_type', 'custom') action_data = payload.get('action_data', '{}') conn = sqlite3.connect(DB_PATH) item = conn.execute("SELECT title, source_id, url FROM scraped_items WHERE id = ?", (item_id,)).fetchone() conn.execute(""" INSERT INTO intelligence_actions (scraped_item_id, action_type, action_data, title, source, url) VALUES (?, ?, ?, ?, ?, ?) """, (item_id, action_type, action_data, item[0][:200] if item else '', (item[1] or '').replace('_', ' ').title() if item else '', item[2] or '' if item else '')) new_id = conn.execute('SELECT last_insert_rowid()').fetchone()[0] conn.commit() conn.close() return jsonify({'status': 'created', 'id': new_id}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 # ── Get open actions ───────────────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/actions') def api_intelligence_v2_actions(): try: conn = sqlite3.connect(DB_PATH) rows = conn.execute(""" SELECT ia.id, ia.scraped_item_id, ia.action_type, ia.action_data, ia.status, ia.created_at, ia.title, ia.source, ia.url FROM intelligence_actions ia WHERE ia.status = 'open' ORDER BY ia.created_at DESC LIMIT 50 """).fetchall() actions = [] for r in rows: actions.append({ 'id': r[0], 'scraped_item_id': r[1], 'action_type': r[2], 'action_data': r[3], 'status': r[4], 'created_at': r[5], 'title': r[6], 'source': r[7], 'url': r[8] }) conn.close() return jsonify({'actions': actions, 'count': len(actions)}) except Exception as e: return jsonify({'actions': [], 'count': 0, 'error': str(e)}) # ── Close action ───────────────────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/actions//close', methods=['POST']) def api_intelligence_v2_close_action(action_id): try: conn = sqlite3.connect(DB_PATH) conn.execute("UPDATE intelligence_actions SET status = 'closed', closed_at = ? WHERE id = ?", (datetime.now().isoformat(), action_id)) conn.commit() conn.close() return jsonify({'status': 'closed', 'id': action_id}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 # ── Alerts ─────────────────────────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/alerts') def api_intelligence_v2_alerts(): """Get unread alerts (HOOG/CRITICAL priority).""" try: conn = sqlite3.connect(DB_PATH) rows = conn.execute(""" SELECT id, title, message, priority, source, source_url, is_read, created_at, scraped_item_id FROM hseq_alerts WHERE is_read = 0 AND priority IN ('HOOG', 'KRITIEK') ORDER BY CASE priority WHEN 'KRITIEK' THEN 1 WHEN 'HOOG' THEN 2 END, created_at DESC LIMIT 20 """).fetchall() alerts = [] for r in rows: alerts.append({ 'id': r[0], 'title': r[1], 'message': r[2], 'priority': r[3], 'source': r[4], 'url': r[5], 'is_read': r[6], 'created_at': r[7], 'scraped_item_id': r[8] }) conn.close() return jsonify({'alerts': alerts, 'count': len(alerts)}) except Exception as e: return jsonify({'alerts': [], 'count': 0, 'error': str(e)}) @app.route(BASE_PATH + '/api/intelligence/v2/alerts/count') def api_intelligence_v2_alerts_count(): try: conn = sqlite3.connect(DB_PATH) count = conn.execute( "SELECT COUNT(*) FROM hseq_alerts WHERE is_read = 0 AND priority IN ('HOOG', 'KRITIEK')" ).fetchone()[0] conn.close() return jsonify({'count': count}) except: return jsonify({'count': 0}) @app.route(BASE_PATH + '/api/intelligence/v2/alerts//read', methods=['POST']) def api_intelligence_v2_alert_read(alert_id): try: conn = sqlite3.connect(DB_PATH) conn.execute("UPDATE hseq_alerts SET is_read = 1 WHERE id = ?", (alert_id,)) conn.commit() conn.close() return jsonify({'status': 'read'}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 @app.route(BASE_PATH + '/api/intelligence/v2/alerts/read-all', methods=['POST']) def api_intelligence_v2_alerts_read_all(): try: conn = sqlite3.connect(DB_PATH) conn.execute("UPDATE hseq_alerts SET is_read = 1") conn.commit() conn.close() return jsonify({'status': 'all_read'}) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 # ── Dashboard alert summary (for main page) ────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/dashboard-summary') def api_intelligence_v2_dashboard(): """Summary data for main dashboard integration.""" try: conn = sqlite3.connect(DB_PATH) open_alerts = conn.execute( "SELECT COUNT(*) FROM hseq_alerts WHERE is_read = 0 AND priority IN ('HOOG', 'KRITIEK')" ).fetchone()[0] total_items = conn.execute( "SELECT COUNT(*) FROM scraped_items WHERE consultant_take_json IS NOT NULL AND dismissed = 0" ).fetchone()[0] last_alert = conn.execute( "SELECT created_at FROM hseq_alerts ORDER BY created_at DESC LIMIT 1" ).fetchone() # Get top 5 high-risk items for dashboard widget rows = conn.execute(""" SELECT id, title, risk_level, source_id, scraped_at FROM scraped_items WHERE risk_level IN ('HOOG', 'KRITIEK') AND dismissed = 0 ORDER BY CASE risk_level WHEN 'KRITIEK' THEN 1 WHEN 'HOOG' THEN 2 END, id DESC LIMIT 5 """).fetchall() top_risks = [{'id': r[0], 'title': r[1][:100], 'risk': r[2], 'source': (r[3] or '').replace('_', ' ').title(), 'date': (r[4] or '')[:10]} for r in rows] conn.close() return jsonify({ 'open_alerts': open_alerts, 'total_analyzed': total_items, 'last_alert': last_alert[0] if last_alert else None, 'top_risks': top_risks }) except Exception as e: return jsonify({'open_alerts': 0, 'total_analyzed': 0, 'last_alert': None, 'top_risks': [], 'error': str(e)}) # ── Stats for intelligence page ────────────────────────────────────── @app.route(BASE_PATH + '/api/intelligence/v2/stats') def api_intelligence_v2_stats(): try: conn = sqlite3.connect(DB_PATH) total = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE dismissed = 0").fetchone()[0] analyzed = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE consultant_take_json IS NOT NULL AND dismissed = 0").fetchone()[0] dismissed = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE dismissed = 1").fetchone()[0] open_actions = conn.execute("SELECT COUNT(*) FROM intelligence_actions WHERE status = 'open'").fetchone()[0] risks = {} for level in ('KRITIEK', 'HOOG', 'MEDIUM', 'LAAG'): risks[level] = conn.execute( "SELECT COUNT(*) FROM scraped_items WHERE risk_level = ? AND dismissed = 0", (level,) ).fetchone()[0] conn.close() return jsonify({ 'total': total, 'analyzed': analyzed, 'dismissed': dismissed, 'open_actions': open_actions, 'risks': risks }) except Exception as e: return jsonify({'total': 0, 'analyzed': 0, 'dismissed': 0, 'open_actions': 0, 'risks': {}, 'error': str(e)}) def build_intelligence_v2_html(BASE_PATH, DB_PATH): """Build the V2 Intelligence Hub page HTML with consultant takes, action buttons, alerts.""" conn = sqlite3.connect(DB_PATH) init_intelligence_v2_tables(DB_PATH) # Stats total = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE dismissed = 0").fetchone()[0] analyzed = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE consultant_take_json IS NOT NULL AND dismissed = 0").fetchone()[0] dismissed = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE dismissed = 1").fetchone()[0] open_actions = conn.execute("SELECT COUNT(*) FROM intelligence_actions WHERE status = 'open'").fetchone()[0] open_alerts = conn.execute("SELECT COUNT(*) FROM hseq_alerts WHERE is_read = 0 AND priority IN ('HOOG','KRITIEK')").fetchone()[0] risks = {} for level in ('KRITIEK', 'HOOG', 'MEDIUM', 'LAAG'): risks[level] = conn.execute("SELECT COUNT(*) FROM scraped_items WHERE risk_level = ? AND dismissed = 0", (level,)).fetchone()[0] # Get items with consultant takes rows = conn.execute(""" SELECT id, source_id, title, url, content, scraped_at, category, consultant_take_json, risk_level FROM scraped_items WHERE dismissed = 0 AND consultant_take_json IS NOT NULL ORDER BY CASE risk_level WHEN 'KRITIEK' THEN 1 WHEN 'HOOG' THEN 2 WHEN 'MEDIUM' THEN 3 ELSE 4 END, id DESC LIMIT 50 """).fetchall() # Get open V2 actions v2_actions = conn.execute(""" SELECT id, scraped_item_id, action_type, action_data, title, source, created_at FROM intelligence_actions WHERE status = 'open' ORDER BY created_at DESC LIMIT 20 """).fetchall() # Get alerts alerts = conn.execute(""" SELECT id, title, message, priority, source, source_url, created_at FROM hseq_alerts WHERE is_read = 0 ORDER BY CASE priority WHEN 'KRITIEK' THEN 1 WHEN 'HOOG' THEN 2 WHEN 'MEDIUM' THEN 3 ELSE 4 END, created_at DESC LIMIT 10 """).fetchall() # Get pending items (no consultant take yet) pending = conn.execute(""" SELECT COUNT(*) FROM scraped_items WHERE consultant_take_json IS NULL AND dismissed = 0 """).fetchone()[0] conn.close() risk_colors = {'KRITIEK': '#DC2626', 'HOOG': '#EF4444', 'MEDIUM': '#F59E0B', 'LAAG': '#00A859'} risk_labels = {'KRITIEK': '🔴 Kritiek', 'HOOG': '🟠 Hoog', 'MEDIUM': '🟡 Medium', 'LAAG': '🟢 Laag'} body = '''
''' + str(pending) + '''
items wachten op AI analyse

Risico Verdeling

''' total_risk = max(sum(risks.values()), 1) for level, color in risk_colors.items(): pct = round((risks.get(level, 0) / total_risk) * 100) if risks.get(level, 0) > 0: body += f'
' body += '''
🔴 Kritiek''' + str(risks.get('KRITIEK', 0)) + '''
🟠 Hoog''' + str(risks.get('HOOG', 0)) + '''
🟡 Medium''' + str(risks.get('MEDIUM', 0)) + '''
🟢 Laag''' + str(risks.get('LAAG', 0)) + '''

Statistieken

📊 Totaal items''' + str(total) + '''
🧠 Geanalyseerd''' + str(analyzed) + '''
✅ Afgehandeld''' + str(dismissed) + '''
📋 Open acties''' + str(open_actions) + '''
🚨 Unread alerts''' + str(open_alerts) + '''

Snelle Acties

🧠 Intelligence Feed
📋 Acties ''' + str(open_actions) + '''
🚨 Alerts ''' + str(open_alerts) + '''
''' if rows: for r in rows: item_id, source, title, url, content, scraped_at, category, take_json, risk_level = r take = {} if take_json: try: take = json.loads(take_json) except: pass risk_cls = risk_level or 'LAAG' rc = risk_colors.get(risk_cls, '#868E96') take_text = take.get('consultant_take', '') actions_list = take.get('actions', []) src_name = (source or '').replace('_', ' ').title() body += f'''
{title[:200]}
{risk_labels.get(risk_cls, risk_cls)}
{src_name} · {category or 'Algemeen'} · {(scraped_at or '')[:10]}
{(content or '')[:250]}
''' if take_text: body += f'''
💡 Consultant Take: {take_text}
''' if actions_list: body += '
Voorgestelde acties:
    ' for a in actions_list[:3]: body += f'
  • {a}
    • ' body += '
' body += f'''
🎓 Maak Training (LMS)
📰 Genereer Nieuwsbericht
📋 Update RI&E
📢 Stuur Team Alert
📝 Eigen Actie
{f'🔗 Bron' if url else ''}
''' else: body += '
🧠
Geen geanalyseerde items. Klik op "Genereer Consultant Takes" om te starten.
' body += '''
''' if v2_actions: action_type_labels = {'training': '🎓 Training', 'news': '📰 Nieuws', 'rie': '📋 RI&E', 'alert': '📢 Alert', 'custom': '📝 Actie'} action_type_colors = {'training': '#003366', 'news': '#7950F2', 'rie': '#00A859', 'alert': '#EF4444', 'custom': '#F59E0B'} for a in v2_actions: aid, sitem_id, atype, adata, atitle, asource, acreated = a atc = action_type_colors.get(atype, '#868E96') atl = action_type_labels.get(atype, atype) body += f'''
{atl}
{(atitle or 'Onbekend')[:100]}
{asource} · {acreated[:16] if acreated else ''}
''' else: body += '
📋
Geen openstaande acties
' body += '''
''' if alerts: for al in alerts: alid, altitle, almsg, alpri, alsrc, alurl, alcreated = al alc = '#DC2626' if alpri == 'KRITIEK' else '#EF4444' body += f'''
{(altitle or '')[:150]}
{alpri}
{(almsg or '')[:200]}
{alsrc} · {alcreated[:16] if alcreated else ''}
{f'🔗 Bron' if alurl else ''}
''' body += '
' else: body += '
🟢
Geen openstaande alerts
' body += '''
''' return body def build_dashboard_alert_widget(BASE_PATH): """Build the HSEQ Alerts widget HTML for the main dashboard page.""" return '''

🚨 HSEQ Alerts -

Laden...
'''