# ============================================================ # MODULE 1: PBZO Compliance & VBS Dashboard # Phoenix Metals HSEQ VBS — Separate module file # Imported by app.py before Flask app starts # ============================================================ import sqlite3 from datetime import datetime, timedelta from search import get_db from indexer import DB_PATH from flask import jsonify, request def init_db_pbzo(): """Create PBZO/VBS tables if not exist.""" try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS vbs_elements ( id INTEGER PRIMARY KEY, code TEXT NOT NULL UNIQUE, name_nl TEXT NOT NULL, name_en TEXT NOT NULL, description TEXT, sort_order INTEGER)''') c.execute('''CREATE TABLE IF NOT EXISTS legislation_register ( id INTEGER PRIMARY KEY AUTOINCREMENT, legislation_type TEXT NOT NULL, legislation_code TEXT NOT NULL, title TEXT NOT NULL, article TEXT, version_date TEXT, source_url TEXT, document_id INTEGER, active INTEGER DEFAULT 1, created_date TEXT DEFAULT CURRENT_TIMESTAMP, updated_date TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (document_id) REFERENCES documents(id) ON DELETE SET NULL)''') c.execute('''CREATE TABLE IF NOT EXISTS pbzo_requirements ( id INTEGER PRIMARY KEY AUTOINCREMENT, requirement_code TEXT NOT NULL UNIQUE, title TEXT NOT NULL, description TEXT NOT NULL, vbs_element_id INTEGER NOT NULL, legislation_id INTEGER, requirement_source TEXT, severity TEXT DEFAULT 'mandatory', applicability TEXT DEFAULT 'phoenix_metals', frequency TEXT, created_date TEXT DEFAULT CURRENT_TIMESTAMP, updated_date TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (vbs_element_id) REFERENCES vbs_elements(id), FOREIGN KEY (legislation_id) REFERENCES legislation_register(id) ON DELETE SET NULL)''') c.execute('''CREATE TABLE IF NOT EXISTS compliance_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, pbzo_requirement_id INTEGER NOT NULL, status TEXT DEFAULT 'not_assessed', compliance_score REAL DEFAULT 0.0, evidence_document_id INTEGER, evidence_action_id INTEGER, gap_description TEXT, remediation_plan TEXT, owner TEXT NOT NULL DEFAULT 'HSEQ Manager', assessor TEXT, assessment_date TEXT, next_review_date TEXT, priority TEXT DEFAULT 'medium', notes TEXT, created_date TEXT DEFAULT CURRENT_TIMESTAMP, updated_date TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (pbzo_requirement_id) REFERENCES pbzo_requirements(id) ON DELETE CASCADE, FOREIGN KEY (evidence_document_id) REFERENCES documents(id) ON DELETE SET NULL, FOREIGN KEY (evidence_action_id) REFERENCES action_items(id) ON DELETE SET NULL)''') c.execute('''CREATE TABLE IF NOT EXISTS inspection_schedule ( id INTEGER PRIMARY KEY AUTOINCREMENT, equipment_tag TEXT, inspection_type TEXT NOT NULL, legislation_id INTEGER, description TEXT NOT NULL, frequency_days INTEGER NOT NULL, last_done_date TEXT, next_due_date TEXT, status TEXT DEFAULT 'scheduled', certified_body TEXT, owner TEXT, related_vka_id INTEGER, notes TEXT, created_date TEXT DEFAULT CURRENT_TIMESTAMP, updated_date TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (legislation_id) REFERENCES legislation_register(id) ON DELETE SET NULL, FOREIGN KEY (related_vka_id) REFERENCES vka_items(id) ON DELETE SET NULL)''') c.execute('CREATE INDEX IF NOT EXISTS idx_cs_req ON compliance_status(pbzo_requirement_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_pbzo_vbs ON pbzo_requirements(vbs_element_id)') c.execute('CREATE INDEX IF NOT EXISTS idx_insp_due ON inspection_schedule(next_due_date, status)') count = c.execute("SELECT COUNT(*) FROM vbs_elements").fetchone()[0] if count == 0: c.executemany("INSERT OR IGNORE INTO vbs_elements VALUES (?,?,?,?,?,?)", [ (1,'VBS-01','Organisatie & Beleid','Organization & Policy','Veiligheidsbeleid, rollen, verantwoordelijkheden, M&A structuur',1), (2,'VBS-02','Risico-identificatie & Evaluatie','Hazard Identification & Risk Evaluation','RI&E, HAZOP, QRA, bronidentificatie',2), (3,'VBS-03','Operationele Beheersing','Operational Control','Procedures, werkplekinstructies, PTW, MOC',3), (4,'VBS-04','Noodplan & Respons','Emergency Response','Noodplannen, interne hulpverlening, alarmieren, oefeningen',4), (5,'VBS-05','Kwalificaties & Competenties','Competence & Training','Trainingmatrix, certificering, VCA, instructie',5), (6,'VBS-06','Monitoring & Meting','Monitoring & Measurement','KPIs, inspecties, emissiemetingen, audits',6), (7,'VBS-07','Continue Verbetering','Continuous Improvement','Incidenten, afwijkingen, correcties, management review',7), ]) conn.commit(); conn.close() except Exception as e: print(f"[WARN] init_db_pbzo: {e}") def register_pbzo_routes(app, page_fn, BASE_PATH="/hseq-dashboard"): """Register all PBZO/VBS routes on the Flask app.""" _page = page_fn @app.route(BASE_PATH + '/api/pbzo/dashboard') def api_pbzo_dashboard(): db = get_db() try: rows = db.execute(""" SELECT ve.id AS vbs_id, ve.code AS vbs_code, ve.name_nl AS vbs_name, COUNT(pr.id) AS total_requirements, SUM(CASE WHEN cs.status = 'compliant' THEN 1 ELSE 0 END) AS compliant_count, SUM(CASE WHEN cs.status = 'partially_compliant' THEN 1 ELSE 0 END) AS partial_count, SUM(CASE WHEN cs.status = 'non_compliant' THEN 1 ELSE 0 END) AS non_compliant_count, SUM(CASE WHEN cs.status IN ('not_assessed','non_compliant') OR cs.status IS NULL THEN 1 ELSE 0 END) AS gap_count, ROUND(AVG(COALESCE(cs.compliance_score, 0)), 1) AS avg_score FROM vbs_elements ve LEFT JOIN pbzo_requirements pr ON pr.vbs_element_id = ve.id LEFT JOIN compliance_status cs ON cs.pbzo_requirement_id = pr.id GROUP BY ve.id ORDER BY ve.sort_order """).fetchall() overall = db.execute("SELECT ROUND(AVG(COALESCE(compliance_score,0)),1) AS s FROM compliance_status").fetchone() overdue = db.execute("SELECT COUNT(*) AS c FROM inspection_schedule WHERE next_due_date < date('now') AND status='scheduled'").fetchone() return jsonify({ 'timestamp': datetime.utcnow().isoformat(), 'overall_score': overall['s'] if overall and overall['s'] else 0, 'overdue_inspections': overdue['c'] if overdue else 0, 'vbs_elements': [dict(r) for r in rows] }) finally: db.close() @app.route(BASE_PATH + '/api/pbzo/vbs/') def api_pbzo_vbs_detail(vbs_id): db = get_db() try: vbs = db.execute("SELECT * FROM vbs_elements WHERE id=?", (vbs_id,)).fetchone() if not vbs: return jsonify({'error':'VBS element not found'}), 404 reqs = db.execute(""" SELECT pr.id, pr.requirement_code, pr.title, pr.description, pr.requirement_source, pr.severity, pr.frequency, cs.status, cs.compliance_score, cs.gap_description, cs.remediation_plan, cs.owner, cs.assessor, cs.assessment_date, cs.next_review_date, cs.priority, d.title AS evidence_document_title, ai.title AS evidence_action_title, ai.status AS evidence_action_status, lr.legislation_code, lr.title AS legislation_title, lr.article AS legislation_article FROM pbzo_requirements pr LEFT JOIN compliance_status cs ON cs.pbzo_requirement_id = pr.id LEFT JOIN documents d ON cs.evidence_document_id = d.id LEFT JOIN action_items ai ON cs.evidence_action_id = ai.id LEFT JOIN legislation_register lr ON pr.legislation_id = lr.id WHERE pr.vbs_element_id = ? ORDER BY CASE cs.priority WHEN 'critical' THEN 1 WHEN 'high' THEN 2 WHEN 'medium' THEN 3 ELSE 4 END, pr.requirement_code """, (vbs_id,)).fetchall() return jsonify({'vbs_element': dict(vbs), 'requirements': [dict(r) for r in reqs], 'total': len(reqs)}) finally: db.close() @app.route(BASE_PATH + '/api/pbzo/status', methods=['POST']) def api_pbzo_status_update(): data = request.get_json() if not data or 'pbzo_requirement_id' not in data: return jsonify({'error': 'pbzo_requirement_id required'}), 400 db = get_db() try: now = datetime.utcnow().isoformat() nxt = (datetime.utcnow() + timedelta(days=365)).strftime('%Y-%m-%d') existing = db.execute("SELECT id FROM compliance_status WHERE pbzo_requirement_id=?", (data['pbzo_requirement_id'],)).fetchone() if existing: db.execute("""UPDATE compliance_status SET status=?, compliance_score=?, gap_description=?, remediation_plan=?, owner=?, assessor=?, evidence_document_id=?, evidence_action_id=?, priority=?, notes=?, assessment_date=?, next_review_date=?, updated_date=? WHERE pbzo_requirement_id=?""", (data.get('status','not_assessed'), data.get('compliance_score',0), data.get('gap_description'), data.get('remediation_plan'), data.get('owner','HSEQ Manager'), data.get('assessor'), data.get('evidence_document_id'), data.get('evidence_action_id'), data.get('priority','medium'), data.get('notes'), now, nxt, now, data['pbzo_requirement_id'])) else: db.execute("""INSERT INTO compliance_status (pbzo_requirement_id, status, compliance_score, gap_description, remediation_plan, owner, assessor, evidence_document_id, evidence_action_id, priority, notes, assessment_date, next_review_date) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", (data['pbzo_requirement_id'], data.get('status','not_assessed'), data.get('compliance_score',0), data.get('gap_description'), data.get('remediation_plan'), data.get('owner','HSEQ Manager'), data.get('assessor'), data.get('evidence_document_id'), data.get('evidence_action_id'), data.get('priority','medium'), data.get('notes'), now, nxt)) db.commit() return jsonify({'success': True, 'updated_date': now}) except Exception as e: db.rollback() return jsonify({'error': str(e)}), 500 finally: db.close() @app.route(BASE_PATH + '/api/pbzo/gaps') def api_pbzo_gaps(): db = get_db() try: gaps = db.execute(""" SELECT cs.id, pr.requirement_code, pr.title AS requirement_title, pr.requirement_source, ve.code AS vbs_code, ve.name_nl AS vbs_name, cs.status, cs.compliance_score, cs.gap_description, cs.remediation_plan, cs.owner, cs.priority, cs.assessment_date, cs.next_review_date, ai.id AS linked_action_id, ai.title AS linked_action_title FROM compliance_status cs JOIN pbzo_requirements pr ON cs.pbzo_requirement_id = pr.id JOIN vbs_elements ve ON pr.vbs_element_id = ve.id LEFT JOIN action_items ai ON cs.evidence_action_id = ai.id WHERE cs.status IN ('non_compliant','partially_compliant','not_assessed') ORDER BY CASE cs.priority WHEN 'critical' THEN 1 WHEN 'high' THEN 2 WHEN 'medium' THEN 3 ELSE 4 END, cs.compliance_score ASC """).fetchall() summary = db.execute(""" SELECT pr.requirement_source, COUNT(*) AS gap_count, ROUND(AVG(cs.compliance_score),1) AS avg_score FROM compliance_status cs JOIN pbzo_requirements pr ON cs.pbzo_requirement_id = pr.id WHERE cs.status IN ('non_compliant','partially_compliant','not_assessed') GROUP BY pr.requirement_source ORDER BY avg_score ASC """).fetchall() return jsonify({'gaps': [dict(r) for r in gaps], 'total_gaps': len(gaps), 'summary_by_source': [dict(r) for r in summary]}) finally: db.close() @app.route(BASE_PATH + '/pbzo') def pbzo_dashboard_page(): body = '''

📡 VBS Compliance Radar — Bijlage III Seveso III

🏗️ VBS-Elementen — Klik voor details

''' return _page(body, page_title='PBZO Compliance Tracker', page_subtitle='Phoenix Metals \u2014 Veiligheidsbeheersysteem (VBS) Module 1', active='pbzo')