feat: Academy integration — training gap detection after document approval (F7)

- Migration 115: compliance_role_training_mapping table (org roles → training codes)
- TrainingLinkService: queries training_modules/matrix/assignments to find gaps
  per person and role. Gracefully degrades when Go training tables don't exist yet.
- document_review_routes: 2 new endpoints (training-requirements, training-gaps)
- _notify_approval() now checks training gaps and sends emails to persons
  with outstanding modules, linking to /sdk/training/learner

[migration-approved]

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-05-03 22:03:25 +02:00
parent 965af3a34c
commit 630fffc0cc
3 changed files with 249 additions and 0 deletions
@@ -252,6 +252,20 @@ def approve_review(
# Notify all OTHER roles mapped to this document type about the approval
_notify_approval(db, tenant_id, review)
# Check training gaps
training_info = {"training_gaps": 0, "academy_available": False}
try:
from compliance.services.training_link_service import TrainingLinkService
tls = TrainingLinkService(db)
gaps = tls.check_training_gaps(tenant_id, review["document_type"], review.get("project_id"))
training_info = {"training_gaps": gaps.get("total_gaps", 0), "academy_available": gaps.get("academy_available", False)}
# Send training notification emails for each gap
if gaps.get("gaps"):
_notify_training_gaps(gaps["gaps"], review)
except Exception as e:
logger.warning("Training gap check failed (non-blocking): %s", e)
review["training"] = training_info
return review
@@ -292,6 +306,32 @@ def _notify_approval(db: Session, tenant_id: str, review: dict):
logger.warning("Approval notification failed (non-blocking): %s", e)
def _notify_training_gaps(gaps: list[dict], review: dict):
"""Send training requirement emails to persons with outstanding modules."""
try:
from compliance.services.smtp_sender import send_email
for gap in gaps:
if not gap.get("person_email"):
continue
send_email(
recipient=gap["person_email"],
subject=f"[BreakPilot] Schulungsbedarf: {gap['module_title']}",
body_html=f"""
<h2>Schulungsbedarf nach Dokument-Freigabe</h2>
<p>Sehr geehrte/r <strong>{gap['person_name']}</strong>,</p>
<p>nach Freigabe des Dokuments <strong>{review['document_title']}</strong>
ist fuer Ihre Rolle (<strong>{gap['role']}</strong>) eine Schulung erforderlich:</p>
<p><strong>{gap['module_title']}</strong> ({gap['module_code']})</p>
<p>Status: {gap['status']}</p>
<p><a href="/sdk/training/learner" style="background:#7c3aed;color:white;padding:10px 20px;border-radius:6px;text-decoration:none;">Zur Academy</a></p>
<p style="color:#888;font-size:12px;">BreakPilot Compliance SDK</p>
""",
)
logger.info("Sent %d training gap notifications for %s", len(gaps), review["document_title"])
except Exception as e:
logger.warning("Training notification failed (non-blocking): %s", e)
@router.post("/{review_id}/reject")
def reject_review(
review_id: str,
@@ -310,3 +350,31 @@ def reject_review(
raise HTTPException(404, "Review not found")
db.commit()
return _row_to_dict(row)
# =============================================================================
# Training Integration
# =============================================================================
@router.get("/training-requirements")
def get_training_requirements(
document_type: str = Query(...),
db: Session = Depends(get_db),
tenant_id: str = Depends(_get_tenant_id),
):
from compliance.services.training_link_service import TrainingLinkService
service = TrainingLinkService(db)
return service.get_training_requirements(tenant_id, document_type)
@router.get("/training-gaps")
def get_training_gaps(
document_type: str = Query(...),
project_id: Optional[str] = Query(None),
db: Session = Depends(get_db),
tenant_id: str = Depends(_get_tenant_id),
):
from compliance.services.training_link_service import TrainingLinkService
service = TrainingLinkService(db)
return service.check_training_gaps(tenant_id, document_type, project_id)