# ============================================================ # MODULE: Rule-Based Intelligence Pipeline — V1.0 # HSEQ Intelligence Dashboard # Genereert intelligence insights uit scraped_items (geen LLM nodig) # ============================================================ import sqlite3 import json import re from datetime import datetime, timedelta from pathlib import Path DB_PATH = Path(__file__).parent / 'hseq_kennisbank.db' OUTPUT_PATH = Path(__file__).parent / 'static' / 'hseq_latest.json' # ─── Keyword Scoring ──────────────────────────────────────────────────── CRITICAL_KEYWORDS = [ 'lood', 'diisocyanaten', 'asbest', 'explosie', 'incident', 'dodelijk', 'doden', 'ernstig gewond', 'boete', 'handhaving', 'stillegging', 'acute blootstelling', 'chemische ongeval', 'brand', 'lek', 'gaswolk', 'seveso', 'brzo', 'major accident', 'verbod', 'intrekking', 'in werking', 'per direct', 'uiterlijk', 'verplicht', ] HIGH_KEYWORDS = [ 'wetgeving', 'wijziging', 'nieuwe regel', 'amendement', 'richtlijn', 'pfus', 'pfas', 'gevaarlijke stof', 'veiligheidsbeleid', 'toezicht', 'inspectie', 'arbowet', 'arbobesluit', 'owerheid', 'eu-richtlijn', 'iso 45001', 'iso 14001', 'certificering', 'compliance', 'ria', 'risicoanalyse', 'plan van aanpak', 'pva', 'machineveiligheid', 'veilig werken', 'lockout tagout', 'loto', 'height safety', 'valbeveiliging', 'confined space', 'besloten ruimte', 'straling', 'geluid', 'trilling', 'ergonomie', 'psychosociale', 'werkdruk', 'agressie', 'pestgedrag', ] STANDARD_KEYWORDS = [ 'nieuws', 'update', 'publicatie', 'rapport', 'onderzoek', 'advies', 'toolkit', 'checklist', 'training', 'cursus', 'kennisbank', 'publicatie', 'richtlijn', 'norm', ] # ─── Category Mapping ─────────────────────────────────────────────────── CATEGORY_MAP = { 'wetgeving': 'WETGEVING', 'scraped': 'NIEUWS', 'default': 'NIEUWS', } # ─── Context-Based Consultant Take Engine ──────────────────────────────── HSEQ_CONTEXT_RULES = { 'arbowetgeving': { 'keywords': ['arbowet', 'arbobesluit', 'arbeidsomstandigheden', 'arbovergo', 'preventiemedewerker', 'arbocatalogus'], 'takes': [ 'Toets de impact op uw RI&E: identificeer alle werkplekken waar deze arbowijziging van toepassing is en update de risicoinventarisatie binnen de wettelijke termijn.', 'Update uw arbobeleid en Plan van Aanpak (PVA): verifieer of de preventiemaatregelen nog voldoen aan de gewijzigde wettelijke kaders.', 'Plan een overleg met uw preventiemedewerker en bedrijfsarts om de operationele impact van deze arbowijziging te Inventariseren.', ], 'actions': [ ('DOCUMENT_UPDATE', 'Controleer of de RI&E van uw BRZO-locaties volledig is bijgewerkt conform de nieuwe arbowetgeving.'), ('COMPLIANCE_CHECK', 'Toets of uw PVA de nieuwe arboverplichtingen dekt; plan een update met de preventiemedewerker.'), ('TRAINING_NEEDED', 'Plan een toolbox-meeting over de gewijzigde arboverplichtingen voor operationeel leidinggevenden.'), ], 'impact': 'HOOG', }, 'brzo_seveso': { 'keywords': ['brzo', 'seveso', 'veiligheidsrapport', 'noodplan', 'qra', 'mimd', 'veiligheidsbeheerssysteem', 'vbs'], 'takes': [ 'BRZO/Seveso-impact: toets of uw veiligheidsrapport en noodprocedures moeten worden geüpdatet. Verifieer de MIMD-scenario\'s tegen de nieuwe inzichten.', 'Controleer uw Veiligheidsbeheerssysteem (VBS): deze ontwikkeling kan impact hebben op uw veiligheidsrapportageplicht aan DCMR/Inspectie SZW.', 'Evalueer of de QRA (Quantitative Risk Assessment) herziening nodig is op basis van nieuwe incidentdata of gewijzigde vergunningsvoorwaarden.', ], 'actions': [ ('AUDIT_PREP', 'Bereid een gap-analyse voor tussen huidige veiligheidsrapportage en de nieuwe BRZO-eisen.'), ('DOCUMENT_UPDATE', 'Update het interne noodplan en oefen de noodprocedures met alle BHV-ers binnen 30 dagen.'), ('RISK_ASSESSMENT', 'Heroverweeg de MIMD-scenario\'s in het kader van de huidige procesveiligheidsstatus.'), ], 'impact': 'HOOG', }, 'asbest': { 'keywords': ['asbest', 'asbestinventarisatie', 'asbestverwijdering', 'asbesthoudend', 'asbesthoudende'], 'takes': [ 'Asbest-specifiek: verifieer dat alle asbestinventarisaties op uw locaties actueel zijn (max. 3 jaar oud). Controleer of de saneringsplannen conform de nieuwe richtlijn zijn opgesteld.', 'Controleer de blootstellingsregistratie en de geldigheid van het asbestattest. Werkgevers zijn aansprakelijk voor onvoldoende documentatie bij Inspectie SZW.', ], 'actions': [ ('COMPLIANCE_CHECK', 'Controleer of de RI&E van uw klantenlocaties asbest-inventarisatie bevat conform de actuele richtlijn.'), ('DOCUMENT_UPDATE', 'Update de asbestbeheerplannen en verifieer dat alle saneringsdossiers compleet zijn.'), ('RISK_ASSESSMENT', 'Inventariseer alle asbestverdachte materialen in installaties ouder dan 1994.'), ], 'impact': 'HOOG', }, 'lood': { 'keywords': ['lood', 'loodhoudend', 'loodvergiftiging', 'loodblootstelling', 'blood lead', 'loodverbindingen'], 'takes': [ 'Loodblootstelling: verifieer de blootstellingsregistratie en de resultaten van medische keuringen. De grenswaarden voor lood zijn recent aangescherpt.', 'Controleer of de RI&E de loodrisico\'s adequaat identificeert en of de PBM-protocollen up-to-date zijn.', ], 'actions': [ ('COMPLIANCE_CHECK', 'Controleer de blootstellingsregistratie lood bij onderhouds- en schilderwerkzaamheden.'), ('RISK_ASSESSMENT', 'Heroverweeg de loodblootstellingsrisico\'s in het kader van de huidige grenswaarden.'), ('TRAINING_NEEDED', 'Plan bijscholing over loodveilig werken voor betrokken medewerkers.'), ], 'impact': 'HOOG', }, 'pfas': { 'keywords': ['pfas', 'pfus', 'polyfluor', 'perfluor', 'forever chemicals'], 'takes': [ 'PFAS-impact: map de bodem- en waterkwaliteitsdata naar uw locatiespecifieke risico\'s. PFAS-verplichtingen kunnen impact hebben op milieuvergunningen en saneringsplichten.', 'Evalueer of PFAS-houdende materialen (schuimblusmiddelen, coatings, packing) op uw locaties geïdentificeerd zijn in de inventarisatie.', ], 'actions': [ ('RISK_ASSESSMENT', 'Inventariseer PFAS-bronnen op locatie: blusschuim, coatings, packing, en afwateringssystemen.'), ('COMPLIANCE_CHECK', 'Controleer of uw milieuvergunning PFAS-emissielimieten bevat en of u hieraan voldoet.'), ('DOCUMENT_UPDATE', 'Update de milieuzorgsysteemdocumentatie met PFAS-specifieke procedures.'), ], 'impact': 'MIDDEN', }, 'diisocyanaten': { 'keywords': ['diisocyanaten', 'isocyanaat', 'mdi', 'tdi', 'polyurethaan', 'pur', 'hardschuim', 'kitschuim'], 'takes': [ 'Diisocyanaten: de EU-restrictie (2023) vereist training voor beroepsmatige gebruikers. Verifieer dat alle medewerkers die met PU-producten werken getraind zijn.', 'Controleer de blootstellingsregistratie voor diisocyanaten en toets of de ventilatie- en PBM-maatregelen voldoen aan de nieuwe eisen.', ], 'actions': [ ('TRAINING_NEEDED', 'Verifieer dat alle medewerkers die met diisocyanaten werken een geldig trainingcertificaat hebben.'), ('COMPLIANCE_CHECK', 'Controleer de blootstellingsregistratie diisocyanaten op alle BRZO-locaties.'), ('RISK_ASSESSMENT', 'Heroverweeg de risicobeoordeling voor werkzaamheden met isocyanaten (isolatie, coating, kitten).'), ], 'impact': 'HOOG', }, 'machineveiligheid': { 'keywords': ['machineveiligheid', 'ce-markering', 'machinerichtlijn', 'lockout tagout', 'loto', 'veilige afstand', 'kbm', 'afdichting', 'machinebeveiliging'], 'takes': [ 'Machineveiligheid: verifieer CE-markeringen en machineveiligheidsrapporten. Controleer LOTO-procedures bij onderhoudswerkzaamheden op gevaarlijke installaties.', 'Toets of de risicobeoordeling per machine conform de Machinerichtlijn 2006/42/EG is uitgevoerd en documenteer eventuele afwijkingen.', ], 'actions': [ ('COMPLIANCE_CHECK', 'Controleer of alle machines een geldige CE-markering hebben en de risicobeoordeling up-to-date is.'), ('AUDIT_PREP', 'Bereid de machineveiligheidsdocumentatie voor voor de komende VCA/ISO-audit.'), ('TRAINING_NEEDED', 'Plan LOTO-training voor onderhoudspersoneel en contractors op BRZO-locaties.'), ], 'impact': 'MIDDEN', }, 'ergonomie': { 'keywords': ['ergonomie', 'rsi', 'werkdruk', 'fysieke belasting', 'tillen', 'repetitief', 'werken op hoogte', 'valbeveiliging', 'valongeval'], 'takes': [ 'Ergonomie/fysieke belasting: voer een werkplekanalyse uit op locaties waar tillen, repetitieve bewegingen of langdurig staan voorkomt.', 'RSI-preventie: evalueer de werkplekinrichting en plan interventies voor medewerkers met klachten aan bewegingsapparaat.', ], 'actions': [ ('RISK_ASSESSMENT', 'Plan een ergonomische werkplekanalyse voor de hoog-risicofuncties (onderhoud, logistiek, produktie).'), ('DOCUMENT_UPDATE', 'Update de RI&E met specifieke ergonomische risico\'s per werkplek.'), ('TRAINING_NEEDED', 'Organiseer een training over tillen en fysieke belasting voor operationeel personeel.'), ], 'impact': 'MIDDEN', }, 'incidenten': { 'keywords': ['incident', 'ongeval', 'dodelijk', 'doden', 'ernstig gewond', 'explo', 'brand', 'lek', 'gaswolk', 'near miss', 'forval', 'bodemvalling'], 'takes': [ 'Incident-les: voer een lesson-learned analyse uit. Vergelijk de oorzaken met uw eigen installaties en identificeer potentiële zwakke plekken.', 'Toets of vergelijkbare risico\'s op uw locaties aanwezig zijn. Update het inspectieprogramma op basis van de bevindingen uit dit incident.', 'Communiceer de lessons learned naar alle operationele teams en neem op in het komende veiligheidsoverleg.', ], 'actions': [ ('RISK_ASSESSMENT', 'Voer een lesson-learned analyse uit en identificeer vergelijkbare risico\'s op uw eigen locaties.'), ('COMPLIANCE_CHECK', 'Toets of de inspectie SZW na dit incident verhoogde aandacht heeft voor uw sector; bereid u voor.'), ('STAKEHOLDER_COMM', 'Communiceer de lessons learned naar operationele teams en neem op in het veiligheidsoverleg.'), ], 'impact': 'HOOG', }, 'certificering': { 'keywords': ['iso 45001', 'iso 14001', 'vca', 'certificering', 'certificaat', 'audit', 'managementsysteem', 'normering'], 'takes': [ 'Certificering: bereid de documentatie voor voor de komende audit. Identificeer gaps tussen de huidige praktijk en de normeisen.', 'Controleer of het managementsysteem up-to-date is en of alle interne audits zijn afgerond. Plan eventuele correctieve acties.', ], 'actions': [ ('AUDIT_PREP', 'Start de audit-voorbereiding: inventariseer documentatie, identificeer gaps, en plan interne audits.'), ('DOCUMENT_UPDATE', 'Update het managementsysteemdocumentatie met de nieuwste normeisen en wijzigingen.'), ('COMPLIANCE_CHECK', 'Voer een gap-analyse uit tussen de huidige VCA/ISO-status en de herziene normeisen.'), ], 'impact': 'MIDDEN', }, 'milieu': { 'keywords': ['emissie', 'bodem', 'water', 'lucht', 'milieuvergunning', 'mja', 'co2', 'stikstof', 'noxe', 'afval', 'lozing', 'vergunning'], 'takes': [ 'Milieubeheer: toets of uw milieuvergunning en emissielimieten nog actueel zijn. MJA-afspraken kunnen impact hebben op operationele processen.', 'Controleer de rapportageverplichtingen (E-PRTR, DPR, luchtemissies) en verifieer dat de monitoringssystemen correct kalibreren.', ], 'actions': [ ('COMPLIANCE_CHECK', 'Controleer of de milieuvergunning up-to-date is en of de MJA-doelstellingen gehaald worden.'), ('DOCUMENT_UPDATE', 'Update de emissieregistratie en verifieer de rapportageverplichtingen (E-PRTR, DPR).'), ('RISK_ASSESSMENT', 'Heroverweeg de milieurisico\'s in het kader van gewijzigde emissienormen of vergunningseisen.'), ], 'impact': 'MIDDEN', }, 'gevaarlijke_stoffen': { 'keywords': ['gevaarlijke stof', 'ghs', 'clp', 'sds', 'veiligheidsinformatieblad', 'msds', 'giffige', 'brandbaar', 'ontplofbaar', 'corrosief', 'carcinogeen', 'mutageen', 'reproductietoxisch'], 'takes': [ 'Gevaarlijke stoffen: verifieer dat alle Veiligheidsinformatiebladen (VIB/SDS) up-to-date zijn en dat de blootstellingsregistratie compleet is.', 'Controleer de CLP/GHS-classificatie van opgeslagen stoffen en update de etikettering conform de laatste wijzigingen.', ], 'actions': [ ('COMPLIANCE_CHECK', 'Inventariseer alle gevaarlijke stoffen op locatie en verifieer de geldigheid van de VIB\'s.'), ('DOCUMENT_UPDATE', 'Update de stoffeninventarisatie en de blootstellingsregistratie.'), ('TRAINING_NEEDED', 'Plan training over omgang met gevaarlijke stoffen voor nieuw personeel en contractors.'), ], 'impact': 'HOOG', }, 'handhaving': { 'keywords': ['boete', 'handhaving', 'stillegging', 'dwangsom', 'intrekking', 'inspectie', 'toezicht', 'overtreding'], 'takes': [ 'Handhaving: analyseer de overtreding en toets of vergelijkbare risico\'s bestaan op uw locaties. Pas de preventiemaatregelen aan waar nodig.', 'Deze handhaving actie kan een trend signaleren. Verhoog de interne inspectiefrequentie op de betreffende aandachtsgebieden.', ], 'actions': [ ('RISK_ASSESSMENT', 'Analyseer de handhaving en toets of vergelijkbare risico\'s bestaan op uw BRZO-locaties.'), ('COMPLIANCE_CHECK', 'Verhoog de interne inspectiefrequentie op de betreffende aandachtsgebieden.'), ('STAKEHOLDER_COMM', 'Informeer management over de handhaving trend en de genomen voorzorgsmaatregelen.'), ], 'impact': 'HOOG', }, } # Fallback templates voor items zonder specifieke context — EERLIJK: nooit doen alsof dit consultant-advies is FALLBACK_TEMPLATES = { 'CRITICAL': 'Automatische signalering (trefwoordanalyse, geen AI-consultant-analyse): dit item bevat mogelijke kritieke signalen. Voer een handmatige review uit om de werkelijke impact te bepalen.', 'HIGH': 'Automatische signalering (trefwoordanalyse, geen AI-consultant-analyse): mogelijk relevante ontwikkeling. Toets handmatig de relevantie voor uw organisatie, vergunningen en RI&E.', 'STANDARD': 'Automatische signalering (trefwoordanalyse, geen AI-consultant-analyse): informatief item — handmatige beoordeling aanbevolen bij twijfel.', } FALLBACK_ACTIONS = { 'WETGEVING': ('DOCUMENT_UPDATE', 'Controleer of uw RI&E en Arbobeleid aansluiten bij de nieuwe regelgeving en plan een update indien nodig.'), 'NIEUWS': ('COMPLIANCE_CHECK', 'Toets de relevantie van dit nieuws voor actieve projecten en klantencontracten in de petrochemische sector.'), 'INCIDENT': ('RISK_ASSESSMENT', 'Voer een lesson-learned analyse uit en toets of vergelijkbare risico\'s op uw locaties aanwezig zijn.'), } # ─── Core Functions ───────────────────────────────────────────────────── def get_db(): conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row return conn def classify_priority(title, content): """Classificeer item prioriteit op basis van keywords.""" text = f'{title} {content}'.lower() critical_hits = sum(1 for kw in CRITICAL_KEYWORDS if kw in text) high_hits = sum(1 for kw in HIGH_KEYWORDS if kw in text) if critical_hits >= 2: return 'CRITICAL', 10 elif critical_hits >= 1: return 'CRITICAL', 9 elif high_hits >= 3: return 'HIGH', 8 elif high_hits >= 1: return 'HIGH', 7 else: return 'STANDARD', 5 def _detect_context(title, content): """Detecteer HSEQ-context op basis van keywords. Retourneert (context_key, rule) of None.""" text = f'{title} {content}'.lower() best_match = None best_score = 0 for ctx_key, rule in HSEQ_CONTEXT_RULES.items(): score = sum(1 for kw in rule['keywords'] if kw in text) if score > best_score: best_score = score best_match = (ctx_key, rule) return best_match def _extract_deadline(title, content): """Extraheer een regulatory deadline uit de content.""" import re text = f'{title} {content}' # Patronen: "uiterlijk 1 juli", "voor 1 maart 2026", "per 1 januari" patterns = [ r'uiterlijk\s+(\d{1,2}\s+\w+\s+\d{4})', r'voor\s+(\d{1,2}\s+\w+\s+\d{4})', r'per\s+(\d{1,2}\s+\w+\s+\d{4})', r'\b(\d{1,2}\s+(?:januari|februari|maart|april|mei|juni|juli|augustus|september|oktober|november|december)\s+\d{4})\b', r'uiterlijk\s+(\d{1,2}\s+\w+)', r'voor\s+(\d{1,2}\s+\w+)', ] for pat in patterns: m = re.search(pat, text, re.IGNORECASE) if m: return m.group(1) return None def generate_consultant_take(priority, title, content, category): """Genereer een context-gebaseerde consultant analyse.""" import hashlib context = _detect_context(title, content) text = f'{title} {content}'.lower() if context: ctx_key, rule = context # Deterministische selectie op basis van content hash idx = int(hashlib.md5(text.encode()).hexdigest(), 16) % len(rule['takes']) return rule['takes'][idx] # Fallback return FALLBACK_TEMPLATES.get(priority, FALLBACK_TEMPLATES['STANDARD']) def generate_action(title, content, category): """Genereer een context-gebaseerde voorgestelde actie met action_type.""" import hashlib context = _detect_context(title, content) text = f'{title} {content}'.lower() if context: ctx_key, rule = context idx = int(hashlib.md5(text.encode()).hexdigest(), 16) % len(rule['actions']) return rule['actions'][idx] # (action_type, action_text) # Fallback fallback = FALLBACK_ACTIONS.get(category, FALLBACK_ACTIONS['NIEUWS']) return fallback # (action_type, action_text) def generate_impact(title, content): """Bepaal impact assessment (HOOG/MIDDEN/LAAG) voor BRZO-petrochemisch bedrijf.""" context = _detect_context(title, content) if context: return context[1]['impact'] # Fallback: check critical keywords text = f'{title} {content}'.lower() critical = any(kw in text for kw in ['brzo', 'seveso', 'explosie', 'dodelijk', 'stillegging', 'boete', 'handhaving']) if critical: return 'HOOG' return 'LAAG' def build_intelligence_json(max_items=100): """Bouw hseq_latest.json op basis van scraped_items (rule-based).""" conn = get_db() # Haal laatste scraped items op (met content) rows = conn.execute(''' SELECT si.source_id, si.url, si.title, si.content, si.scraped_at, si.category FROM scraped_items si WHERE LENGTH(si.content) > 50 ORDER BY si.scraped_at DESC LIMIT ? ''', (max_items,)).fetchall() conn.close() insights = [] trending = {} watchlist = [] for row in rows: r = dict(row) priority, relevance_score = classify_priority(r['title'], r['content']) category = CATEGORY_MAP.get(r['category'], 'NIEUWS') action_type, action_text = generate_action(r['title'], r['content'], category) deadline = _extract_deadline(r['title'], r['content']) impact = generate_impact(r['title'], r['content']) insight = { 'title': r['title'][:200], 'source': r['source_id'], 'url': r['url'], 'date_found': r['scraped_at'][:10] if r['scraped_at'] else '', 'status': 'NEW', 'intelligence_level': priority, 'summary': r['content'][:300] + ('...' if len(r['content']) > 300 else ''), 'category': category, 'priority': 'HOOG' if priority == 'CRITICAL' else 'MIDDEN' if priority == 'HIGH' else 'STANDAARD', 'relevance_score': relevance_score, 'consultant_take': generate_consultant_take(priority, r['title'], r['content'], category), 'suggested_action': action_text, 'action_type': action_type, 'impact_assessment': impact, 'regulatory_deadline': deadline, } insights.append(insight) # Trending topics text = f'{r["title"]} {r["content"]}'.lower() for kw in CRITICAL_KEYWORDS + HIGH_KEYWORDS: if kw in text: trending[kw] = trending.get(kw, 0) + 1 # Sorteer: CRITICAL eerst, dan HIGH, dan STANDARD insights.sort(key=lambda x: {'CRITICAL': 0, 'HIGH': 1, 'STANDARD': 2}.get(x['intelligence_level'], 3)) # Top trending trending_sorted = sorted(trending.items(), key=lambda x: -x[1])[:10] # Watchlist for kw, count in trending_sorted[:5]: if count >= 2: watchlist.append({ 'topic': kw.title(), 'reason': f'{count} relevante items gevonden', 'priority': 'HOOG' if count >= 3 else 'MIDDEN', }) # Bouw JSON output = { 'version': '3.0', 'date': datetime.now().strftime('%Y-%m-%d'), 'generated_at': datetime.now().isoformat() + 'Z', 'generator': 'rule-based-v2.0', 'total_sources_scraped': len(set(r['source_id'] for r in [dict(row) for row in rows])), 'total_items_found': len(insights), 'items_visible_in_ui': min(len(insights), 50), 'ui_threshold': 50, 'insights': insights[:50], 'trending_topics': [{'topic': kw, 'count': count} for kw, count in trending_sorted], 'watchlist': watchlist, } return output def save_intelligence_json(): """Genereer en sla intelligence JSON op.""" data = build_intelligence_json() OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) with open(OUTPUT_PATH, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"[INTELLIGENCE] Bijgewerkt: {data['total_items_found']} insights, {len(data['trending_topics'])} trending topics") return data if __name__ == '__main__': save_intelligence_json()