281 lines
7.3 KiB
Python
281 lines
7.3 KiB
Python
import os
|
|
import uuid
|
|
from datetime import timedelta
|
|
from flask import render_template
|
|
from weasyprint import HTML
|
|
|
|
import db
|
|
|
|
|
|
CERT_OUTPUT_DIR = "/app/files/certificates"
|
|
|
|
|
|
def get_module_courses(course_rows, module_code):
|
|
return [
|
|
c for c in course_rows
|
|
if (c["code"] or "").upper().startswith(module_code)
|
|
]
|
|
|
|
|
|
def get_user_module_completion(user_id, module_code):
|
|
conn = db.get_connection()
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT u.id, u.name, u.mandant_id, m.name
|
|
FROM app_user u
|
|
JOIN mandant m ON m.id = u.mandant_id
|
|
WHERE u.id = %s
|
|
""", (user_id,))
|
|
user_row = cur.fetchone()
|
|
if not user_row:
|
|
cur.close()
|
|
conn.close()
|
|
return None
|
|
|
|
_, user_name, mandant_id, mandant_name = user_row
|
|
|
|
cur.execute("""
|
|
SELECT id, code, title
|
|
FROM course
|
|
WHERE is_active = TRUE
|
|
ORDER BY code
|
|
""")
|
|
all_courses = db.fetchall_dict(cur)
|
|
module_courses = get_module_courses(all_courses, module_code)
|
|
|
|
if not module_courses:
|
|
cur.close()
|
|
conn.close()
|
|
return None
|
|
|
|
course_ids = [c["id"] for c in module_courses]
|
|
|
|
cur.execute("""
|
|
SELECT course_id, MIN(created_at) AS first_passed_at, MAX(created_at) AS last_passed_at
|
|
FROM user_assessment
|
|
WHERE user_id = %s
|
|
AND passed = TRUE
|
|
AND course_id = ANY(%s)
|
|
GROUP BY course_id
|
|
""", (user_id, course_ids))
|
|
passed_rows = db.fetchall_dict(cur)
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
if len(passed_rows) != len(module_courses):
|
|
return None
|
|
|
|
first_pass_dates = [r["first_passed_at"] for r in passed_rows]
|
|
last_pass_dates = [r["last_passed_at"] for r in passed_rows]
|
|
|
|
valid_from = max(last_pass_dates)
|
|
valid_until = min(first_pass_dates) + timedelta(days=365)
|
|
|
|
return {
|
|
"user_id": user_id,
|
|
"user_name": user_name,
|
|
"mandant_id": mandant_id,
|
|
"mandant_name": mandant_name,
|
|
"module_code": module_code,
|
|
"valid_from": valid_from,
|
|
"valid_until": valid_until,
|
|
}
|
|
|
|
|
|
def generate_certificate_pdf_for_user(user_id, module_code):
|
|
data = get_user_module_completion(user_id, module_code)
|
|
if not data:
|
|
return None
|
|
|
|
conn = db.get_connection()
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT guid
|
|
FROM certificate
|
|
WHERE user_id = %s
|
|
AND module_code = %s
|
|
""", (user_id, module_code))
|
|
existing = cur.fetchone()
|
|
|
|
if existing:
|
|
guid_value = str(existing[0])
|
|
cur.execute("""
|
|
UPDATE certificate
|
|
SET user_name = %s,
|
|
mandant_id = %s,
|
|
mandant_name = %s,
|
|
valid_from = %s,
|
|
valid_until = %s
|
|
WHERE guid = %s
|
|
""", (
|
|
data["user_name"],
|
|
data["mandant_id"],
|
|
data["mandant_name"],
|
|
data["valid_from"],
|
|
data["valid_until"],
|
|
guid_value
|
|
))
|
|
else:
|
|
guid_value = str(uuid.uuid4())
|
|
cur.execute("""
|
|
INSERT INTO certificate (
|
|
guid, user_id, mandant_id, user_name, mandant_name,
|
|
module_code, valid_from, valid_until
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
""", (
|
|
guid_value,
|
|
data["user_id"],
|
|
data["mandant_id"],
|
|
data["user_name"],
|
|
data["mandant_name"],
|
|
data["module_code"],
|
|
data["valid_from"],
|
|
data["valid_until"],
|
|
))
|
|
|
|
conn.commit()
|
|
cur.close()
|
|
conn.close()
|
|
|
|
os.makedirs(CERT_OUTPUT_DIR, exist_ok=True)
|
|
|
|
reference_url = f"https://cert.compliance-verification.info/verification/{guid_value}"
|
|
pdf_path = os.path.join(CERT_OUTPUT_DIR, f"{guid_value}.pdf")
|
|
|
|
html = render_template(
|
|
"certificate_template.html",
|
|
user_name=data["user_name"],
|
|
mandant_name=data["mandant_name"],
|
|
module_code=data["module_code"],
|
|
valid_from=data["valid_from"].strftime("%d.%m.%Y"),
|
|
valid_until=data["valid_until"].strftime("%d.%m.%Y"),
|
|
reference_url=reference_url,
|
|
reference_id=guid_value,
|
|
logo_path="file:///app/certificates/assets/logo.png",
|
|
signature_path="file:///app/certificates/assets/signature.png",
|
|
)
|
|
|
|
HTML(string=html, base_url="/").write_pdf(pdf_path)
|
|
|
|
return {
|
|
"guid": guid_value,
|
|
"pdf_path": pdf_path,
|
|
"reference_url": reference_url,
|
|
}
|
|
|
|
#########
|
|
def ensure_certificate_for_user_module(user_id, module_code):
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
|
|
# User + Mandant laden
|
|
cur.execute("""
|
|
SELECT u.id, u.name, u.mandant_id, m.name
|
|
FROM app_user u
|
|
JOIN mandant m ON m.id = u.mandant_id
|
|
WHERE u.id = %s
|
|
""", (user_id,))
|
|
row = cur.fetchone()
|
|
|
|
if not row:
|
|
cur.close()
|
|
conn.close()
|
|
return None
|
|
|
|
_, user_name, mandant_id, mandant_name = row
|
|
|
|
# Alle Kurse dieses Moduls
|
|
cur.execute("""
|
|
SELECT id, code, title
|
|
FROM course
|
|
WHERE is_active = TRUE
|
|
ORDER BY code
|
|
""")
|
|
all_courses = fetchall_dict(cur)
|
|
|
|
module_courses = get_module_courses(all_courses, module_code)
|
|
|
|
if not module_courses:
|
|
cur.close()
|
|
conn.close()
|
|
return None
|
|
|
|
course_ids = [c["id"] for c in module_courses]
|
|
|
|
# Bestandene Assessments des Users für diese Kurse
|
|
cur.execute("""
|
|
SELECT course_id, MIN(created_at) AS first_passed_at, MAX(created_at) AS last_passed_at
|
|
FROM user_assessment
|
|
WHERE user_id = %s
|
|
AND passed = TRUE
|
|
AND course_id = ANY(%s)
|
|
GROUP BY course_id
|
|
""", (user_id, course_ids))
|
|
passed_rows = fetchall_dict(cur)
|
|
|
|
if len(passed_rows) != len(module_courses):
|
|
cur.close()
|
|
conn.close()
|
|
return None
|
|
|
|
first_pass_dates = [r["first_passed_at"] for r in passed_rows]
|
|
last_pass_dates = [r["last_passed_at"] for r in passed_rows]
|
|
|
|
valid_from = max(last_pass_dates)
|
|
valid_until = min(first_pass_dates) + timedelta(days=365)
|
|
|
|
# Gibt es schon ein Zertifikat?
|
|
cur.execute("""
|
|
SELECT guid
|
|
FROM certificate
|
|
WHERE user_id = %s
|
|
AND module_code = %s
|
|
""", (user_id, module_code))
|
|
existing = cur.fetchone()
|
|
|
|
if existing:
|
|
guid_value = existing[0]
|
|
cur.execute("""
|
|
UPDATE certificate
|
|
SET user_name = %s,
|
|
mandant_id = %s,
|
|
mandant_name = %s,
|
|
valid_from = %s,
|
|
valid_until = %s
|
|
WHERE guid = %s
|
|
""", (
|
|
user_name,
|
|
mandant_id,
|
|
mandant_name,
|
|
valid_from,
|
|
valid_until,
|
|
guid_value
|
|
))
|
|
else:
|
|
guid_value = str(uuid.uuid4())
|
|
cur.execute("""
|
|
INSERT INTO certificate (
|
|
guid, user_id, mandant_id, user_name, mandant_name,
|
|
module_code, valid_from, valid_until
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
""", (
|
|
guid_value,
|
|
user_id,
|
|
mandant_id,
|
|
user_name,
|
|
mandant_name,
|
|
module_code,
|
|
valid_from,
|
|
valid_until
|
|
))
|
|
|
|
conn.commit()
|
|
cur.close()
|
|
conn.close()
|
|
|
|
return guid_value |