# ============================================================ # MODULE: Scraper Engine - FASE 2 # HSEQ Intelligence Dashboard # Geautomatiseerde scraping van HSEQ-bronnen → FTS5 index # ============================================================ import sqlite3 import hashlib import json import os import re import threading import time from datetime import datetime, timedelta from flask import jsonify, request, render_template_string import requests from bs4 import BeautifulSoup from search import get_db, DB_PATH BASE_PATH = os.environ.get('BASE_PATH', '/hseq-dashboard') # ─── Tier Configuratie ───────────────────────────────────────────────── SOURCES = { 'arboportaal': { 'name': 'Arboportaal', 'url': 'https://www.arboportaal.nl/actueel', 'tier': 1, 'category': 'wetgeving', }, 'inspectie_szw': { 'name': 'Inspectie SZW', 'url': 'https://www.rijksoverheid.nl/onderwerpen/arbeidsomstandigheden/nieuws', 'tier': 1, 'category': 'wetgeving', }, 'rivm': { 'name': 'RIVM', 'url': 'https://www.rivm.nl/nieuws', 'tier': 1, 'category': 'scraped', 'link_pattern': '/nieuws/', # Only follow links containing this pattern }, 'brzoplus': { 'name': 'BRZOplus', 'url': 'https://brzoplus.nl/actueel/', 'tier': 1, 'category': 'wetgeving', }, 'infomil': { 'name': 'InfoMil', 'url': 'https://www.infomil.nl/actueel', 'tier': 1, 'category': 'scraped', }, 'wetten': { 'name': 'Wetten.nl', 'url': 'https://wetten.overheid.nl/zoeken/', 'tier': 1, 'category': 'wetgeving', }, 'arbeidshygiene': { 'name': 'Arbeidshygiëne', 'url': 'https://arbeidshygiene.nl/nieuws/', # Redirected URL (old URL redirects here) 'tier': 2, 'category': 'scraped', }, 'nen': { 'name': 'NEN', 'url': 'https://www.nen.nl/nieuws', 'tier': 2, 'category': 'scraped', }, 'sodm': { 'name': 'SODM', 'url': 'https://www.sodm.nl/actueel/nieuws', 'tier': 2, 'category': 'scraped', }, 'prgs': { 'name': 'Publicatiereeks Gevaarlijke Stoffen', 'url': 'https://publicatiereeksgevaarlijkestoffen.nl/', 'tier': 2, 'category': 'wetgeving', }, 'veiligheidskunde': { 'name': 'Veiligheidskunde', 'url': 'https://www.veiligheidskunde.nl/nieuws', 'tier': 3, 'category': 'scraped', }, 'vca': { 'name': 'VCA', 'url': 'https://www.vca.nl/actueel', 'tier': 3, 'category': 'scraped', }, 'ser': { 'name': 'SER', 'url': 'https://www.ser.nl/nl/actueel', 'tier': 3, 'category': 'wetgeving', }, 'crow': { 'name': 'CROW', 'url': 'https://www.crow.nl/over-crow/nieuws', 'tier': 3, 'category': 'scraped', }, 'dcmr': { 'name': 'DCMR', 'url': 'https://www.dcmr.nl/nieuws', 'tier': 3, 'category': 'scraped', }, 'gezondheidsraad': { 'name': 'Gezondheidsraad', 'url': 'https://www.gezondheidsraad.nl/documenten', 'tier': 3, 'category': 'scraped', }, 'deltalinqs': { 'name': 'Deltalinqs', 'url': 'https://www.deltalinqs.nl/nieuws', 'tier': 3, 'category': 'scraped', }, 'onderzoeksraad': { 'name': 'Onderzoeksraad', 'url': 'https://www.onderzoeksraad.nl/nl/nieuws', 'tier': 3, 'category': 'scraped', }, 'nipv': { 'name': 'NIPV', 'url': 'https://www.nipv.nl/nieuws', 'tier': 3, 'category': 'scraped', }, 'volandis': { 'name': 'Volandis', 'url': 'https://www.volandis.nl/nieuws/', 'tier': 3, 'category': 'scraped', }, 'iplo': { 'name': 'IPLO', 'url': 'https://iplo.nl/actueel/', 'tier': 3, 'category': 'scraped', }, 'osha': { 'name': 'EU-OSHA', 'url': 'https://osha.europa.eu/nl/oshanews', 'tier': 3, 'category': 'scraped', }, 'iso': { 'name': 'ISO', 'url': 'https://www.iso.org/news.html', 'tier': 3, 'category': 'scraped', }, 'hse_uk': { 'name': 'HSE UK', 'url': 'https://www.hse.gov.uk/news.htm', 'tier': 3, 'category': 'scraped', }, 'ilocations': { 'name': 'ILocations (Olie & Gas)', 'url': 'https://www.ilocations.nl/nieuws', 'tier': 2, 'category': 'scraped', }, 'maintenance_nl': { 'name': 'MaintenanceNL', 'url': 'https://www.maintenancenl.com/nieuws', 'tier': 2, 'category': 'scraped', }, 'procesveiligheid': { 'name': 'ProcesVeiligheid.nl', 'url': 'https://www.procesveiligheid.nl/nieuws', 'tier': 1, 'category': 'wetgeving', }, 'petrochemicals': { 'name': 'European Petrochemical Association', 'url': 'https://www.e-pc.org/news', 'tier': 3, 'category': 'scraped', }, 'echa_disabled': { 'name': 'ECHA (disabled - 403)', 'url': 'https://echa.europa.eu/nl/news-and-events/news', 'tier': 3, 'category': 'scraped', }, } # ─── Scheduler defaults ───────────────────────────────────────────────── SCHEDULE_DEFAULTS = { 1: {'times': ['06:00', '12:00', '18:00'], 'days': 'daily'}, 2: {'times': ['06:00'], 'days': 'weekly_monday'}, 3: {'times': ['06:00'], 'days': 'monthly_1st'}, } HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'nl-NL,nl;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'DNT': '1', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', } # Retry settings MAX_RETRIES = 3 RETRY_BACKOFF_BASE = 2 # seconds (exponential: 2, 4, 8) ARTICLE_DELAY = 2.0 # seconds between article fetches MAX_ARTICLES_PER_SOURCE = 10 # ─── Scraper Base Class ────────────────────────────────────────────────── class Scraper: """Base scraper klasse voor HSEQ bronnen.""" def __init__(self, source_id, config): self.source_id = source_id self.name = config['name'] self.url = config['url'] self.tier = config['tier'] self.category = config['category'] self.session = requests.Session() self.session.headers.update(HEADERS) self.session.timeout = 30 self.config_extra = config # Full config for extra options like link_pattern def _fetch_with_retry(self, url, timeout=30): """Fetch URL met retry-logica bij 403/429 errors (exponential backoff).""" for attempt in range(MAX_RETRIES): try: resp = self.session.get(url, timeout=timeout) if resp.status_code in (403, 429): wait = RETRY_BACKOFF_BASE * (2 ** attempt) print(f"[SCRAPER] {self.name}: HTTP {resp.status_code} bij {url}, retry {attempt+1}/{MAX_RETRIES} over {wait}s") time.sleep(wait) continue resp.raise_for_status() return resp except requests.exceptions.RequestException as e: if attempt < MAX_RETRIES - 1: wait = RETRY_BACKOFF_BASE * (2 ** attempt) print(f"[SCRAPER] {self.name}: fetch error {e}, retry {attempt+1}/{MAX_RETRIES} over {wait}s") time.sleep(wait) else: raise return None # All retries exhausted def fetch(self): """Haal HTML op van bron-URL.""" resp = self._fetch_with_retry(self.url) if resp is None: raise requests.exceptions.HTTPError(f"All retries exhausted for {self.url}") return resp.text def parse(self, html): """Parse HTML → lijst van items: [{title, url, content, date}].""" soup = BeautifulSoup(html, 'html.parser') items = [] seen_urls = set() from urllib.parse import urljoin # Optionele link_pattern filter (source-specifiek) link_pattern = self.config_extra.get('link_pattern', None) if hasattr(self, 'config_extra') else None # Algemene link-artikel extractie for link in soup.find_all('a', href=True): text = link.get_text(strip=True) href = link['href'] if not text or len(text) < 15: continue if any(skip in text.lower() for skip in ['cookie', 'privacy', 'contact', 'sitemap', 'login', 'zoek', 'toegankelijkheid', 'rss', 'abonneren', 'deel op', 'delen op']): continue # Maak absolute URL if href.startswith('/'): href = urljoin(self.url, href) # Skip non-http and same-page links if not href.startswith('http') or href == self.url or '#' in href.split('?')[0]: continue # Apply link_pattern filter if configured if link_pattern and link_pattern not in href: continue # Deduplicate URLs if href in seen_urls: continue seen_urls.add(href) items.append({ 'title': text[:300], 'url': href, 'content': '', # Will be filled when fetching article 'date': datetime.now().strftime('%Y-%m-%d'), }) if len(items) >= MAX_ARTICLES_PER_SOURCE: break # Fetch article content met delay en retry for item in items: article_content = item['title'] # Default fallback try: time.sleep(ARTICLE_DELAY) # Rate limiting art_resp = self._fetch_with_retry(item['url'], timeout=15) if art_resp and art_resp.text: art_soup = BeautifulSoup(art_resp.text, 'html.parser') for tag in art_soup.find_all(['nav', 'footer', 'script', 'style', 'header', 'aside', 'iframe']): tag.decompose() main = art_soup.find('article') or art_soup.find('main') or art_soup.find(class_=re.compile(r'content|article|body|text|entry', re.I)) if main: article_content = main.get_text(separator=' ', strip=True) else: body = art_soup.find('body') if body: article_content = body.get_text(separator=' ', strip=True) article_content = re.sub(r'\s+', ' ', article_content).strip() if len(article_content) > 5000: article_content = article_content[:5000] except Exception: pass item['content'] = article_content return items def deduplicate(self, items): """Verwijder duplicaten op basis van content hash.""" conn = get_db() cursor = conn.cursor() new_items = [] for item in items: content_hash = hashlib.sha256( f"{item['url']}:{item['title']}".encode() ).hexdigest() existing = cursor.execute( 'SELECT id FROM scraped_items WHERE content_hash = ?', (content_hash,) ).fetchone() if not existing: item['content_hash'] = content_hash new_items.append(item) conn.close() return new_items def categorize(self, item): """Bepaal categorie voor item.""" return self.category def index(self, item): """Indexeer item in FTS5 via documents tabel.""" conn = get_db() cursor = conn.cursor() try: has_content = len(item.get('content', '')) > 100 cursor.execute(''' INSERT INTO documents (filename, filepath, title, author, pages, file_size, created_date, indexed_date, category, tags, full_text) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( f"scrape_{self.source_id}", item['url'], item['title'], self.name, 1 if has_content else 0, len(item.get('content', '')), item.get('date', ''), datetime.now().strftime('%Y-%m-%d %H:%M'), self.categorize(item), self.source_id, item.get('content', item['title']), )) conn.commit() except Exception: pass finally: conn.close() def alert(self, items): """Genereer alert als er nieuwe items zijn (uitbreidbaar).""" if items: print(f"[SCRAPER] {self.name}: {len(items)} nieuwe items gevonden") return items def run(self): """Volledige scrape pipeline: fetch → parse → deduplicate → index.""" items_found = 0 items_new = 0 try: html = self.fetch() items = self.parse(html) items_found = len(items) new_items = self.deduplicate(items) items_new = len(new_items) for item in new_items: item['category'] = self.categorize(item) # Sla op in scraped_items (INSERT OR IGNORE voor dedup safety) self._save_item(item) # Indexeer in FTS5 self.index(item) self.alert(new_items) self._log_run(items_found, items_new, 'success') return {'source': self.source_id, 'found': items_found, 'new': items_new, 'status': 'success'} except Exception as e: self._log_run(items_found, items_new, f'error: {str(e)[:200]}') return {'source': self.source_id, 'found': items_found, 'new': items_new, 'status': 'error', 'error': str(e)[:200]} def _save_item(self, item): conn = get_db() try: conn.execute(''' INSERT OR IGNORE INTO scraped_items (source_id, url, title, content, content_hash, scraped_at, category) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( self.source_id, item['url'], item['title'], item.get('content', ''), item['content_hash'], datetime.now().strftime('%Y-%m-%d %H:%M:%S'), item.get('category', self.category), )) conn.commit() finally: conn.close() def _log_run(self, items_found, items_new, status): conn = get_db() try: conn.execute(''' INSERT INTO scraper_runs (source_id, source_name, run_at, status, items_found, items_new) VALUES (?, ?, ?, ?, ?, ?) ''', (self.source_id, self.name, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), status, items_found, items_new)) conn.commit() finally: conn.close() # ─── DB Init ───────────────────────────────────────────────────────────── def init_scraper_db(): conn = get_db() conn.executescript(''' CREATE TABLE IF NOT EXISTS scraper_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id TEXT NOT NULL, source_name TEXT, run_at TEXT NOT NULL, status TEXT NOT NULL, items_found INTEGER DEFAULT 0, items_new INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS scraped_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id TEXT NOT NULL, url TEXT NOT NULL, title TEXT, content TEXT, content_hash TEXT UNIQUE, scraped_at TEXT NOT NULL, category TEXT DEFAULT 'scraped' ); CREATE TABLE IF NOT EXISTS scraper_schedule ( source_id TEXT PRIMARY KEY, tier INTEGER DEFAULT 1, times TEXT DEFAULT '["06:00"]', days TEXT DEFAULT 'daily', enabled INTEGER DEFAULT 1 ); ''') # Insert default schedules for sid, cfg in SOURCES.items(): sched = SCHEDULE_DEFAULTS.get(cfg['tier'], SCHEDULE_DEFAULTS[3]) conn.execute(''' INSERT OR IGNORE INTO scraper_schedule (source_id, tier, times, days, enabled) VALUES (?, ?, ?, ?, 1) ''', (sid, cfg['tier'], json.dumps(sched['times']), sched['days'])) conn.commit() conn.close() # ─── Scheduler (background thread) ────────────────────────────────────── _scheduler_running = False _scheduler_thread = None def _should_run_now(schedule): """Check of een bron nu moet draaien op basis van schedule.""" now = datetime.now() times = json.loads(schedule['times']) if isinstance(schedule['times'], str) else schedule['times'] days = schedule['days'] # Check tijd (binnen 5 minuten window) current_time = now.strftime('%H:%M') time_match = False for t in times: h, m = map(int, t.split(':')) now_minutes = now.hour * 60 + now.minute target_minutes = h * 60 + m if abs(now_minutes - target_minutes) <= 5: time_match = True break if not time_match: return False # Check dag if days == 'daily': return True elif days == 'weekly_monday': return now.weekday() == 0 elif days == 'monthly_1st': return now.day == 1 return False def _run_scheduled_scrapes(): """Background scheduler loop.""" global _scheduler_running print("[SCRAPER] Scheduler gestart") while _scheduler_running: try: conn = get_db() schedules = conn.execute( 'SELECT ss.*, sr.run_at as last_run FROM scraper_schedule ss ' 'LEFT JOIN scraper_runs sr ON ss.source_id = sr.source_id ' 'WHERE ss.enabled = 1 ' 'ORDER BY ss.tier' ).fetchall() conn.close() ran_any = False for sched in schedules: if not _scheduler_running: break if _should_run_now(dict(sched)): # Smart trigger: check of recent al gedraaid if sched['last_run']: last = datetime.strptime(sched['last_run'], '%Y-%m-%d %H:%M:%S') if (datetime.now() - last).total_seconds() < 3600: # < 1 uur geleden continue cfg = SOURCES.get(sched['source_id']) if cfg: scraper = Scraper(sched['source_id'], cfg) result = scraper.run() print(f"[SCRAPER] Scheduled: {cfg['name']} → {result}") if result.get('new', 0) > 0: ran_any = True # Intelligence pipeline: analyseer nieuwe items + update JSON if ran_any: try: # Auto-analyse: genereer consultant_takes voor nieuwe items from module_intelligence_v2 import batch_generate_consultant_takes count = batch_generate_consultant_takes(DB_PATH, limit=20) if count > 0: print(f"[SCRAPER] Auto-analyse: {count} nieuwe items geanalyseerd") except Exception as e: print(f"[SCRAPER] Auto-analyse error: {e}") try: from module_intelligence_pipeline import save_intelligence_json save_intelligence_json() print("[SCRAPER] Intelligence pipeline bijgewerkt na sweep") except Exception as e: print(f"[SCRAPER] Intelligence pipeline error: {e}") except Exception as e: print(f"[SCRAPER] Scheduler error: {e}") # Check elke 5 minuten for _ in range(30): if not _scheduler_running: break time.sleep(10) print("[SCRAPER] Scheduler gestopt") def start_scheduler(): global _scheduler_running, _scheduler_thread if _scheduler_running: return _scheduler_running = True _scheduler_thread = threading.Thread(target=_run_scheduled_scrapes, daemon=True) _scheduler_thread.start() def stop_scheduler(): global _scheduler_running _scheduler_running = False # ─── Smart Trigger: hash-check ────────────────────────────────────────── def smart_scrape(source_id): """Scrape met hash-check: alleen als content gewijzigd.""" cfg = SOURCES.get(source_id) if not cfg: return {'error': f'Onbekende bron: {source_id}'} scraper = Scraper(source_id, cfg) try: html = scraper.fetch() current_hash = hashlib.md5(html.encode()).hexdigest() conn = get_db() last = conn.execute( 'SELECT content_hash FROM scraper_runs WHERE source_id = ? AND status = ? ORDER BY run_at DESC LIMIT 1', (source_id, 'success') ).fetchone() conn.close() if last and last['content_hash'] == current_hash: return {'source': source_id, 'status': 'skipped', 'reason': 'no_changes'} # Content gewijzigd - full scrape return scraper.run() except Exception as e: return {'source': source_id, 'status': 'error', 'error': str(e)[:200]} # ─── Route Registratie ────────────────────────────────────────────────── SCRAPER_PAGE_HTML = '''

🌐 Scraper Engine

Geautomatiseerde HSEQ-bron monitoring & indexering

Bronnen
-
Vandaag
-
Deze Week
-
Totaal Items
-
Bron Tier Status Laatste Scrape 24u 7d Actie
Laden...

Scrape Geschiedenis

Tijdstip Bron Status Gevonden Nieuw
Laden...
→ Bekijk alle items in Market Intelligence
''' def register_scraper_routes(app, page, BASE_PATH="/hseq-dashboard"): """Registreer alle scraper routes.""" @app.route(BASE_PATH + '/scraper') def scraper_dashboard(): return page( SCRAPER_PAGE_HTML.replace('{{BASE_PATH}}', BASE_PATH), page_title='Scraper Engine', active='scraper' ) @app.route(BASE_PATH + '/api/scraper/status') def api_scraper_status(): conn = get_db() sources = [] for sid, cfg in SOURCES.items(): last_run = conn.execute( 'SELECT run_at, status FROM scraper_runs WHERE source_id = ? ORDER BY run_at DESC LIMIT 1', (sid,) ).fetchone() items_24h = conn.execute( 'SELECT COUNT(*) as c FROM scraped_items WHERE source_id = ? AND scraped_at >= ?', (sid, (datetime.now() - timedelta(hours=24)).strftime('%Y-%m-%d %H:%M:%S')) ).fetchone()['c'] items_7d = conn.execute( 'SELECT COUNT(*) as c FROM scraped_items WHERE source_id = ? AND scraped_at >= ?', (sid, (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d %H:%M:%S')) ).fetchone()['c'] total_items = conn.execute( 'SELECT COUNT(*) as c FROM scraped_items WHERE source_id = ?', (sid,) ).fetchone()['c'] sources.append({ 'id': sid, 'name': cfg['name'], 'url': cfg['url'], 'tier': cfg['tier'], 'status': dict(last_run)['status'].split(':')[0] if last_run else 'pending', 'last_run': dict(last_run)['run_at'] if last_run else None, 'items_24h': items_24h, 'items_7d': items_7d, 'total_items': total_items, }) conn.close() return jsonify({'sources': sources}) @app.route(BASE_PATH + '/api/scraper/run', methods=['POST']) def api_scraper_run_all(): results = [] total_new = 0 for sid, cfg in SOURCES.items(): scraper = Scraper(sid, cfg) result = scraper.run() results.append(result) total_new += result.get('new', 0) # Update intelligence pipeline try: from module_intelligence_pipeline import save_intelligence_json intel = save_intelligence_json() print(f"[SCRAPER] Intelligence pipeline: {intel['total_items_found']} insights") except Exception as e: print(f"[SCRAPER] Intelligence pipeline error: {e}") return jsonify({'message': f'{len(results)} bronnen gescrapet, {total_new} nieuw', 'results': results}) @app.route(BASE_PATH + '/api/scraper/run/', methods=['POST']) def api_scraper_run_one(source_id): cfg = SOURCES.get(source_id) if not cfg: return jsonify({'error': f'Onbekende bron: {source_id}'}), 404 scraper = Scraper(source_id, cfg) result = scraper.run() return jsonify({'message': f'{cfg["name"]} gescrapet', 'result': result}) @app.route(BASE_PATH + '/api/scraper/history') def api_scraper_history(): limit = request.args.get('limit', 50, type=int) conn = get_db() rows = conn.execute( 'SELECT * FROM scraper_runs ORDER BY run_at DESC LIMIT ?', (limit,) ).fetchall() conn.close() return jsonify({'items': [dict(r) for r in rows]}) @app.route(BASE_PATH + '/api/scraper/stats') def api_scraper_stats(): conn = get_db() today = datetime.now().strftime('%Y-%m-%d') week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d') total = conn.execute('SELECT COUNT(*) as c FROM scraped_items').fetchone()['c'] today_count = conn.execute( 'SELECT COUNT(*) as c FROM scraped_items WHERE scraped_at >= ?', (today,) ).fetchone()['c'] week_count = conn.execute( 'SELECT COUNT(*) as c FROM scraped_items WHERE scraped_at >= ?', (week_ago,) ).fetchone()['c'] runs = conn.execute('SELECT COUNT(*) as c FROM scraper_runs').fetchone()['c'] errors = conn.execute( "SELECT COUNT(*) as c FROM scraper_runs WHERE status LIKE 'error%'" ).fetchone()['c'] conn.close() return jsonify({ 'total_items': total, 'items_today': today_count, 'items_week': week_count, 'total_runs': runs, 'error_runs': errors, 'sources': len(SOURCES), }) @app.route(BASE_PATH + '/api/scraper/scheduler/toggle', methods=['POST']) def api_scraper_scheduler_toggle(): global _scheduler_running if _scheduler_running: stop_scheduler() return jsonify({'running': False}) else: start_scheduler() return jsonify({'running': True}) # Start scheduler automatisch start_scheduler()