📊 Dashboard
📋 Self-Assessment
🔍 Gap Analyse
📅 Audit Programma
📑 Management Review
♻️ Continu Verbetering
ISO Self-Assessment
Gap Analyse
Audit Programma Generator
Management Review
Continu Verbetering (PDCA)
'''
# ─── DB Init ──────────────────────────────────────────────────────────────
def init_kam_db():
"""Create KAM management tables."""
try:
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
# Self-Assessments
c.execute('''CREATE TABLE IF NOT EXISTS kam_assessments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
iso_standard TEXT NOT NULL,
assessor TEXT DEFAULT 'HSEQ Manager',
scope TEXT,
clauses TEXT NOT NULL,
total_score INTEGER DEFAULT 0,
status TEXT DEFAULT 'completed',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)''')
# Gap Analysis
c.execute('''CREATE TABLE IF NOT EXISTS kam_gaps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
iso_standard TEXT NOT NULL,
assessment_id INTEGER,
clause_ref TEXT NOT NULL,
clause_title TEXT,
description TEXT,
risk_level TEXT DEFAULT 'medium',
status TEXT DEFAULT 'open',
action_plan TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (assessment_id) REFERENCES kam_assessments(id)
)''')
# Audit Programme
c.execute('''CREATE TABLE IF NOT EXISTS kam_audit_plans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
iso_standard TEXT,
audit_type TEXT NOT NULL,
scope TEXT,
auditor TEXT,
scheduled_date DATE,
frequency TEXT DEFAULT 'jaarlijks',
status TEXT DEFAULT 'planned',
result TEXT,
notes TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)''')
# Management Reviews
c.execute('''CREATE TABLE IF NOT EXISTS kam_mgmt_reviews (
id INTEGER PRIMARY KEY AUTOINCREMENT,
review_date DATE,
review_type TEXT DEFAULT 'periodiek',
kpi_summary TEXT,
input_summary TEXT,
decisions TEXT,
followup_date DATE,
status TEXT DEFAULT 'planned',
notes TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)''')
# Continu Verbetering (PDCA)
c.execute('''CREATE TABLE IF NOT EXISTS kam_improvements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
iso_standard TEXT,
description TEXT,
pdca_stage TEXT DEFAULT 'plan',
priority TEXT DEFAULT 'medium',
metric_before TEXT,
metric_target TEXT,
metric_after TEXT,
owner TEXT,
status TEXT DEFAULT 'open',
notes TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)''')
# Indexes
c.execute('CREATE INDEX IF NOT EXISTS idx_kam_sa_iso ON kam_assessments(iso_standard)')
c.execute('CREATE INDEX IF NOT EXISTS idx_kam_gap_status ON kam_gaps(status)')
c.execute('CREATE INDEX IF NOT EXISTS idx_kam_gap_risk ON kam_gaps(risk_level)')
c.execute('CREATE INDEX IF NOT EXISTS idx_kam_aud_status ON kam_audit_plans(status)')
c.execute('CREATE INDEX IF NOT EXISTS idx_kam_imp_pdca ON kam_improvements(pdca_stage)')
conn.commit()
conn.close()
print("[OK] KAM management tables initialized")
except Exception as e:
print(f"[WARN] init_kam_db: {e}")
# ─── Route Registration ──────────────────────────────────────────────────
def register_kam_routes(app, page_fn, BASE_PATH="/hseq-dashboard"):
"""Register all KAM management routes."""
_page = page_fn
@app.route(BASE_PATH + '/kam')
def kam_dashboard_page():
return _page(KAM_DASHBOARD_TEMPLATE.replace('__BP__', BASE_PATH),
page_title='KAM Management', page_subtitle='FASE 5 — ISO 9001/14001/45001 Self-Assessment, Gap Analyse, Audit Programma, Management Review, PDCA', active='kam')
# ─── Dashboard Stats ───
@app.route(BASE_PATH + '/api/kam/dashboard')
def kam_api_dashboard():
db = get_db()
try:
ta = db.execute("SELECT COUNT(*) as c FROM kam_assessments").fetchone()['c']
avg = db.execute("SELECT AVG(total_score) as s FROM kam_assessments").fetchone()['s']
og = db.execute("SELECT COUNT(*) as c FROM kam_gaps WHERE status != 'closed'").fetchone()['c']
cg = db.execute("SELECT COUNT(*) as c FROM kam_gaps WHERE status != 'closed' AND risk_level IN ('high','critical')").fetchone()['c']
aud_t = db.execute("SELECT COUNT(*) as c FROM kam_audit_plans").fetchone()['c']
aud_c = db.execute("SELECT COUNT(*) as c FROM kam_audit_plans WHERE status='completed'").fetchone()['c']
imp_t = db.execute("SELECT COUNT(*) as c FROM kam_improvements").fetchone()['c']
imp_a = db.execute("SELECT COUNT(*) as c FROM kam_improvements WHERE status IN ('open','in_progress')").fetchone()['c']
rev_t = db.execute("SELECT COUNT(*) as c FROM kam_mgmt_reviews").fetchone()['c']
rev_c = db.execute("SELECT COUNT(*) as c FROM kam_mgmt_reviews WHERE status='completed'").fetchone()['c']
return jsonify({
'total_assessments': ta, 'avg_score': round(avg, 1) if avg else 0,
'open_gaps': og, 'critical_gaps': cg,
'total_audits': aud_t, 'audits_completed': aud_c,
'total_improvements': imp_t, 'improvements_active': imp_a,
'total_reviews': rev_t, 'reviews_completed': rev_c
})
finally:
db.close()
# ─── Clauses (static) ───
@app.route(BASE_PATH + '/api/kam/clauses')
def kam_api_clauses():
iso = request.args.get('iso_standard', '')
if iso not in ISO_CLAUSES:
return jsonify({'error': 'Unknown ISO standard'}), 400
clauses = ISO_CLAUSES[iso]['clauses']
return jsonify([{'clause_ref': ref, 'clause_title': title} for ref, title in clauses])
# ─── Self-Assessments CRUD ───
@app.route(BASE_PATH + '/api/kam/assessments')
def kam_api_assessments_list():
db = get_db()
try:
q = 'SELECT * FROM kam_assessments WHERE 1=1'
p = []
iso = request.args.get('iso_standard')
if iso:
q += ' AND iso_standard = ?'; p.append(iso)
q += ' ORDER BY created_at DESC'
limit = request.args.get('limit', type=int)
if limit:
q += ' LIMIT ?'; p.append(limit)
rows = db.execute(q, p).fetchall()
return jsonify([dict(r) for r in rows])
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/assessments/
')
def kam_api_assessment_get(aid):
db = get_db()
try:
d = db.execute('SELECT * FROM kam_assessments WHERE id=?', (aid,)).fetchone()
if not d:
return jsonify({'error': 'Not found'}), 404
return jsonify(dict(d))
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/assessments', methods=['POST'])
def kam_api_assessment_create():
data = request.get_json(force=True)
iso = data.get('iso_standard', '')
clauses = data.get('clauses', [])
if not iso or not clauses:
return jsonify({'error': 'iso_standard and clauses required'}), 400
total = sum(c.get('score', 0) for c in clauses)
max_score = len(clauses) * 5
pct = round((total / max_score * 100), 1) if max_score > 0 else 0
db = get_db()
try:
cur = db.execute(
'INSERT INTO kam_assessments (iso_standard, assessor, scope, clauses, total_score) VALUES (?,?,?,?,?)',
(iso, data.get('assessor', 'HSEQ Manager'), data.get('scope', ''),
json.dumps(clauses, ensure_ascii=False), pct)
)
aid = cur.lastrowid
# Auto-generate gaps for non-compliant clauses
for c in clauses:
gap = c.get('gap', 'none')
if gap != 'none':
risk = 'low' if gap == 'low' else ('medium' if gap == 'medium' else ('high' if gap == 'high' else 'critical'))
db.execute(
'INSERT INTO kam_gaps (iso_standard, assessment_id, clause_ref, clause_title, description, risk_level) VALUES (?,?,?,?,?,?)',
(iso, aid, c.get('clause_ref', ''), c.get('clause_title', ''),
c.get('evidence', ''), risk)
)
# Deliverable lifecycle integration
try:
dlv_type = 'audit_report' if 'audit' in iso.lower() else 'inspection_checklist'
db.execute(
'INSERT INTO dlv_lifecycle (title, type, service, status, description, created_by) VALUES (?,?,?,?,?,?)',
(f'ISO Self-Assessment — {iso}', dlv_type, 'kam', 'approved',
f'Assessment score: {pct}%, Assessor: {data.get("assessor","")}',
data.get('assessor', 'system'))
)
except Exception:
pass
db.commit()
return jsonify({'id': aid, 'total_score': pct}), 201
finally:
db.close()
# ─── Gaps CRUD ───
@app.route(BASE_PATH + '/api/kam/gaps')
def kam_api_gaps_list():
db = get_db()
try:
q = 'SELECT * FROM kam_gaps WHERE 1=1'
p = []
st = request.args.get('status')
if st:
q += ' AND status = ?'; p.append(st)
q += ' ORDER BY risk_level DESC, created_at DESC'
limit = request.args.get('limit', type=int)
if limit:
q += ' LIMIT ?'; p.append(limit)
rows = db.execute(q, p).fetchall()
return jsonify([dict(r) for r in rows])
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/gaps/')
def kam_api_gap_get(gid):
db = get_db()
try:
d = db.execute('SELECT * FROM kam_gaps WHERE id=?', (gid,)).fetchone()
if not d:
return jsonify({'error': 'Not found'}), 404
return jsonify(dict(d))
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/gaps/', methods=['PUT'])
def kam_api_gap_update(gid):
data = request.get_json(force=True)
db = get_db()
try:
db.execute('UPDATE kam_gaps SET description=?, risk_level=?, status=?, action_plan=?, updated_at=CURRENT_TIMESTAMP WHERE id=?',
(data.get('description', ''), data.get('risk_level', 'medium'),
data.get('status', 'open'), data.get('action_plan', ''), gid))
db.commit()
return jsonify({'ok': True})
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/gaps/generate', methods=['POST'])
def kam_api_gaps_generate():
"""Generate gaps from latest assessments."""
db = get_db()
try:
# Get latest assessment per ISO
standards = ['ISO 9001:2015', 'ISO 14001:2015', 'ISO 45001:2018']
generated = 0
updated = 0
for iso in standards:
latest = db.execute('SELECT id, clauses FROM kam_assessments WHERE iso_standard=? ORDER BY created_at DESC LIMIT 1', (iso,)).fetchone()
if not latest:
continue
clauses = json.loads(latest['clauses'])
for c in clauses:
gap = c.get('gap', 'none')
if gap == 'none':
continue
risk = {'low': 'low', 'medium': 'medium', 'high': 'high', 'critical': 'critical'}.get(gap, 'medium')
ref = c.get('clause_ref', '')
existing = db.execute('SELECT id FROM kam_gaps WHERE iso_standard=? AND clause_ref=? AND status != ?',
(iso, ref, 'closed')).fetchone()
if existing:
db.execute('UPDATE kam_gaps SET risk_level=?, description=?, assessment_id=?, updated_at=CURRENT_TIMESTAMP WHERE id=?',
(risk, c.get('evidence', ''), latest['id'], existing['id']))
updated += 1
else:
db.execute('INSERT INTO kam_gaps (iso_standard, assessment_id, clause_ref, clause_title, description, risk_level) VALUES (?,?,?,?,?,?)',
(iso, latest['id'], ref, c.get('clause_title', ''), c.get('evidence', ''), risk))
generated += 1
db.commit()
return jsonify({'generated': generated, 'updated': updated})
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/gaps/export')
def kam_api_gaps_export():
"""Export gaps as JSON report."""
db = get_db()
try:
rows = db.execute('SELECT * FROM kam_gaps ORDER BY iso_standard, risk_level DESC').fetchall()
gaps = [dict(r) for r in rows]
report = {
'title': 'KAM Gap Analyse Rapport',
'generated_at': datetime.now().isoformat(),
'total_gaps': len(gaps),
'by_status': {},
'by_risk': {},
'gaps': gaps
}
for g in gaps:
report['by_status'][g['status']] = report['by_status'].get(g['status'], 0) + 1
report['by_risk'][g['risk_level']] = report['by_risk'].get(g['risk_level'], 0) + 1
resp = jsonify(report)
resp.headers['Content-Disposition'] = 'attachment; filename=kam_gap_report.json'
return resp
finally:
db.close()
# ─── Audit Plans CRUD ───
@app.route(BASE_PATH + '/api/kam/audits')
def kam_api_audits_list():
db = get_db()
try:
q = 'SELECT * FROM kam_audit_plans WHERE 1=1 ORDER BY scheduled_date DESC'
limit = request.args.get('limit', type=int)
if limit:
q += ' LIMIT ?'
rows = db.execute(q, (limit,)).fetchall()
else:
rows = db.execute(q).fetchall()
return jsonify([dict(r) for r in rows])
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/audits/')
def kam_api_audit_get(aid):
db = get_db()
try:
d = db.execute('SELECT * FROM kam_audit_plans WHERE id=?', (aid,)).fetchone()
if not d:
return jsonify({'error': 'Not found'}), 404
return jsonify(dict(d))
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/audits', methods=['POST'])
def kam_api_audit_create():
data = request.get_json(force=True)
db = get_db()
try:
cur = db.execute(
'INSERT INTO kam_audit_plans (iso_standard, audit_type, scope, auditor, scheduled_date, frequency, notes) VALUES (?,?,?,?,?,?,?)',
(data.get('iso_standard'), data.get('audit_type', 'internal'), data.get('scope', ''),
data.get('auditor', ''), data.get('scheduled_date', ''), data.get('frequency', 'jaarlijks'),
data.get('notes', ''))
)
db.commit()
return jsonify({'id': cur.lastrowid}), 201
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/audits/', methods=['PUT'])
def kam_api_audit_update(aid):
data = request.get_json(force=True)
db = get_db()
try:
db.execute('UPDATE kam_audit_plans SET status=?, result=?, notes=?, updated_at=CURRENT_TIMESTAMP WHERE id=?',
(data.get('status', 'planned'), data.get('result', ''), data.get('notes', ''), aid))
db.commit()
return jsonify({'ok': True})
finally:
db.close()
# ─── Management Reviews CRUD ───
@app.route(BASE_PATH + '/api/kam/reviews')
def kam_api_reviews_list():
db = get_db()
try:
q = 'SELECT * FROM kam_mgmt_reviews WHERE 1=1 ORDER BY created_at DESC'
limit = request.args.get('limit', type=int)
if limit:
q += ' LIMIT ?'
rows = db.execute(q, (limit,)).fetchall()
else:
rows = db.execute(q).fetchall()
return jsonify([dict(r) for r in rows])
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/reviews/')
def kam_api_review_get(rid):
db = get_db()
try:
d = db.execute('SELECT * FROM kam_mgmt_reviews WHERE id=?', (rid,)).fetchone()
if not d:
return jsonify({'error': 'Not found'}), 404
return jsonify(dict(d))
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/reviews', methods=['POST'])
def kam_api_review_create():
data = request.get_json(force=True)
db = get_db()
try:
cur = db.execute(
'INSERT INTO kam_mgmt_reviews (review_date, review_type, kpi_summary, input_summary, decisions, followup_date) VALUES (?,?,?,?,?,?)',
(data.get('review_date', ''), data.get('review_type', 'periodiek'),
data.get('kpi_summary', ''), data.get('input_summary', ''),
data.get('decisions', ''), data.get('followup_date', ''))
)
db.commit()
return jsonify({'id': cur.lastrowid}), 201
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/reviews/', methods=['PUT'])
def kam_api_review_update(rid):
data = request.get_json(force=True)
db = get_db()
try:
db.execute('UPDATE kam_mgmt_reviews SET status=?, followup_date=?, notes=?, updated_at=CURRENT_TIMESTAMP WHERE id=?',
(data.get('status', 'planned'), data.get('followup_date', ''), data.get('notes', ''), rid))
db.commit()
return jsonify({'ok': True})
finally:
db.close()
# ─── Improvements CRUD ───
@app.route(BASE_PATH + '/api/kam/improvements')
def kam_api_improvements_list():
db = get_db()
try:
q = 'SELECT * FROM kam_improvements WHERE 1=1 ORDER BY created_at DESC'
limit = request.args.get('limit', type=int)
if limit:
q += ' LIMIT ?'
rows = db.execute(q, (limit,)).fetchall()
else:
rows = db.execute(q).fetchall()
return jsonify([dict(r) for r in rows])
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/improvements/')
def kam_api_improvement_get(iid):
db = get_db()
try:
d = db.execute('SELECT * FROM kam_improvements WHERE id=?', (iid,)).fetchone()
if not d:
return jsonify({'error': 'Not found'}), 404
return jsonify(dict(d))
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/improvements', methods=['POST'])
def kam_api_improvement_create():
data = request.get_json(force=True)
db = get_db()
try:
cur = db.execute(
'INSERT INTO kam_improvements (title, iso_standard, description, pdca_stage, priority, metric_before, metric_target, owner) VALUES (?,?,?,?,?,?,?,?)',
(data.get('title', ''), data.get('iso_standard', ''), data.get('description', ''),
data.get('pdca_stage', 'plan'), data.get('priority', 'medium'),
data.get('metric_before', ''), data.get('metric_target', ''), data.get('owner', ''))
)
db.commit()
return jsonify({'id': cur.lastrowid}), 201
finally:
db.close()
@app.route(BASE_PATH + '/api/kam/improvements/', methods=['PUT'])
def kam_api_improvement_update(iid):
data = request.get_json(force=True)
db = get_db()
try:
db.execute('UPDATE kam_improvements SET pdca_stage=?, status=?, metric_after=?, owner=?, notes=?, updated_at=CURRENT_TIMESTAMP WHERE id=?',
(data.get('pdca_stage', 'plan'), data.get('status', 'open'),
data.get('metric_after', ''), data.get('owner', ''),
data.get('notes', ''), iid))
db.commit()
return jsonify({'ok': True})
finally:
db.close()
print("[OK] KAM management routes registered")