288 lines
11 KiB
Python
288 lines
11 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
CommCoach Export Service.
|
|
Generates Markdown and PDF exports for dossiers and sessions.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def buildDossierMarkdown(context: Dict[str, Any], sessions: List[Dict[str, Any]],
|
|
tasks: List[Dict[str, Any]], scores: List[Dict[str, Any]]) -> str:
|
|
"""Build a Markdown export of a full coaching dossier (context)."""
|
|
title = context.get("title", "Coaching Dossier")
|
|
description = context.get("description", "")
|
|
category = context.get("category", "custom")
|
|
createdAt = _formatDate(context.get("createdAt"))
|
|
|
|
lines = [
|
|
f"# {title}",
|
|
"",
|
|
f"**Kategorie:** {category} ",
|
|
f"**Erstellt:** {createdAt} ",
|
|
]
|
|
if description:
|
|
lines.append(f"**Beschreibung:** {description} ")
|
|
|
|
goalsRaw = context.get("goals")
|
|
goals = _parseJson(goalsRaw, [])
|
|
if goals:
|
|
lines += ["", "## Ziele", ""]
|
|
for g in goals:
|
|
text = g.get("text", g) if isinstance(g, dict) else str(g)
|
|
status = g.get("status", "open") if isinstance(g, dict) else "open"
|
|
marker = "[x]" if status in ("done", "completed") else "[ ]"
|
|
lines.append(f"- {marker} {text}")
|
|
|
|
insightsRaw = context.get("insights")
|
|
insights = _parseJson(insightsRaw, [])
|
|
if insights:
|
|
lines += ["", "## Erkenntnisse", ""]
|
|
for ins in insights:
|
|
text = ins.get("text", ins) if isinstance(ins, dict) else str(ins)
|
|
lines.append(f"- {text}")
|
|
|
|
completedSessions = [s for s in sessions if s.get("status") == "completed"]
|
|
completedSessions.sort(key=lambda s: s.get("startedAt") or s.get("createdAt") or "")
|
|
if completedSessions:
|
|
lines += ["", "## Sessions", ""]
|
|
for i, s in enumerate(completedSessions, 1):
|
|
dateStr = _formatDate(s.get("startedAt") or s.get("createdAt"))
|
|
duration = s.get("durationSeconds", 0)
|
|
durationMin = duration // 60 if duration else 0
|
|
score = s.get("competenceScore")
|
|
persona = s.get("personaId") or "Coach"
|
|
lines.append(f"### Session {i} -- {dateStr}")
|
|
lines.append("")
|
|
lines.append(f"**Dauer:** {durationMin} Min. | **Score:** {score or '--'} | **Persona:** {persona} ")
|
|
summary = s.get("summary")
|
|
if summary:
|
|
lines.append(f"\n{summary}")
|
|
lines.append("")
|
|
|
|
if tasks:
|
|
openTasks = [t for t in tasks if t.get("status") in ("open", "inProgress")]
|
|
doneTasks = [t for t in tasks if t.get("status") == "done"]
|
|
lines += ["", "## Aufgaben", ""]
|
|
if openTasks:
|
|
lines.append("**Offen:**")
|
|
for t in openTasks:
|
|
lines.append(f"- [ ] {t.get('title')} ({t.get('priority', 'medium')})")
|
|
lines.append("")
|
|
if doneTasks:
|
|
lines.append("**Erledigt:**")
|
|
for t in doneTasks:
|
|
lines.append(f"- [x] {t.get('title')}")
|
|
lines.append("")
|
|
|
|
if scores:
|
|
lines += ["", "## Kompetenz-Scores", ""]
|
|
dimScores = _groupScoresByDimension(scores)
|
|
for dim, entries in dimScores.items():
|
|
latest = entries[-1]
|
|
lines.append(f"- **{dim}**: {latest.get('score', '--')} ({latest.get('trend', 'stable')})")
|
|
|
|
lines += ["", "---", f"*Exportiert am {_formatDate(None)}*", ""]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def buildSessionMarkdown(session: Dict[str, Any], messages: List[Dict[str, Any]],
|
|
tasks: List[Dict[str, Any]], scores: List[Dict[str, Any]]) -> str:
|
|
"""Build a Markdown export of a single session."""
|
|
dateStr = _formatDate(session.get("startedAt") or session.get("createdAt"))
|
|
duration = session.get("durationSeconds", 0)
|
|
durationMin = duration // 60 if duration else 0
|
|
score = session.get("competenceScore")
|
|
persona = session.get("personaId") or "Coach"
|
|
|
|
lines = [
|
|
f"# Coaching Session -- {dateStr}",
|
|
"",
|
|
f"**Dauer:** {durationMin} Min. | **Score:** {score or '--'} | **Persona:** {persona} ",
|
|
]
|
|
|
|
summary = session.get("summary")
|
|
if summary:
|
|
lines += ["", "## Zusammenfassung", "", summary]
|
|
|
|
if messages:
|
|
lines += ["", "## Gespraechsverlauf", ""]
|
|
for msg in messages:
|
|
role = "Du" if msg.get("role") == "user" else "Coach"
|
|
content = msg.get("content", "")
|
|
lines.append(f"**{role}:** {content}")
|
|
lines.append("")
|
|
|
|
sessionTasks = [t for t in tasks if t.get("sessionId") == session.get("id")]
|
|
if sessionTasks:
|
|
lines += ["## Aufgaben", ""]
|
|
for t in sessionTasks:
|
|
marker = "[x]" if t.get("status") == "done" else "[ ]"
|
|
lines.append(f"- {marker} {t.get('title')}")
|
|
lines.append("")
|
|
|
|
sessionScores = [s for s in scores if s.get("sessionId") == session.get("id")]
|
|
if sessionScores:
|
|
lines += ["## Scores", ""]
|
|
for s in sessionScores:
|
|
lines.append(f"- **{s.get('dimension')}**: {s.get('score')} ({s.get('trend', 'stable')})")
|
|
if s.get("evidence"):
|
|
lines.append(f" _{s.get('evidence')}_")
|
|
lines.append("")
|
|
|
|
lines += ["---", f"*Exportiert am {_formatDate(None)}*", ""]
|
|
return "\n".join(lines)
|
|
|
|
|
|
async def renderDossierPdf(context: Dict[str, Any], sessions: List[Dict[str, Any]],
|
|
tasks: List[Dict[str, Any]], scores: List[Dict[str, Any]],
|
|
aiService=None) -> Optional[bytes]:
|
|
"""Render a dossier as PDF using the existing RendererPdf."""
|
|
try:
|
|
from modules.services.serviceGeneration.renderers.rendererPdf import RendererPdf
|
|
extractedContent = _buildPdfContent(context, sessions, tasks, scores, isDossier=True)
|
|
renderer = RendererPdf()
|
|
docs = await renderer.render(extractedContent=extractedContent, title=context.get("title", "Dossier"), aiService=aiService)
|
|
if docs and len(docs) > 0:
|
|
return docs[0].documentData
|
|
except ImportError:
|
|
logger.warning("RendererPdf not available, falling back to markdown-based PDF")
|
|
except Exception as e:
|
|
logger.warning(f"PDF rendering failed: {e}")
|
|
return None
|
|
|
|
|
|
async def renderSessionPdf(session: Dict[str, Any], messages: List[Dict[str, Any]],
|
|
tasks: List[Dict[str, Any]], scores: List[Dict[str, Any]],
|
|
aiService=None) -> Optional[bytes]:
|
|
"""Render a session as PDF."""
|
|
try:
|
|
from modules.services.serviceGeneration.renderers.rendererPdf import RendererPdf
|
|
title = f"Session {_formatDate(session.get('startedAt'))}"
|
|
extractedContent = _buildPdfContent({"title": title}, [session], tasks, scores, isDossier=False, messages=messages)
|
|
renderer = RendererPdf()
|
|
docs = await renderer.render(extractedContent=extractedContent, title=title, aiService=aiService)
|
|
if docs and len(docs) > 0:
|
|
return docs[0].documentData
|
|
except ImportError:
|
|
logger.warning("RendererPdf not available")
|
|
except Exception as e:
|
|
logger.warning(f"Session PDF rendering failed: {e}")
|
|
return None
|
|
|
|
|
|
def _buildPdfContent(context, sessions, tasks, scores, isDossier=True, messages=None) -> Dict[str, Any]:
|
|
"""Convert dossier/session data into the extractedContent format expected by RendererPdf."""
|
|
title = context.get("title", "Export")
|
|
sections = []
|
|
|
|
sections.append({
|
|
"id": "header",
|
|
"content_type": "heading",
|
|
"elements": [{"text": title, "level": 1}],
|
|
})
|
|
|
|
if isDossier and context.get("description"):
|
|
sections.append({
|
|
"id": "desc",
|
|
"content_type": "paragraph",
|
|
"elements": [{"text": context.get("description")}],
|
|
})
|
|
|
|
completedSessions = [s for s in sessions if s.get("status") == "completed"] if isDossier else sessions
|
|
if completedSessions:
|
|
sessionRows = []
|
|
for s in completedSessions:
|
|
sessionRows.append({
|
|
"cells": [
|
|
_formatDate(s.get("startedAt") or s.get("createdAt")),
|
|
str(s.get("competenceScore") or "--"),
|
|
s.get("summary", "")[:200] if s.get("summary") else "",
|
|
]
|
|
})
|
|
sections.append({
|
|
"id": "sessions",
|
|
"content_type": "heading",
|
|
"elements": [{"text": "Sessions", "level": 2}],
|
|
})
|
|
sections.append({
|
|
"id": "sessions_table",
|
|
"content_type": "table",
|
|
"elements": [{
|
|
"headers": ["Datum", "Score", "Zusammenfassung"],
|
|
"rows": sessionRows,
|
|
}],
|
|
})
|
|
|
|
if messages:
|
|
chatElements = []
|
|
for msg in messages:
|
|
role = "Du" if msg.get("role") == "user" else "Coach"
|
|
chatElements.append({"text": f"{role}: {msg.get('content', '')}"})
|
|
sections.append({
|
|
"id": "chat",
|
|
"content_type": "heading",
|
|
"elements": [{"text": "Gespraechsverlauf", "level": 2}],
|
|
})
|
|
sections.append({
|
|
"id": "chat_content",
|
|
"content_type": "paragraph",
|
|
"elements": chatElements,
|
|
})
|
|
|
|
if tasks:
|
|
taskItems = [{"text": f"{'[x]' if t.get('status') == 'done' else '[ ]'} {t.get('title')}"} for t in tasks]
|
|
sections.append({
|
|
"id": "tasks",
|
|
"content_type": "heading",
|
|
"elements": [{"text": "Aufgaben", "level": 2}],
|
|
})
|
|
sections.append({
|
|
"id": "task_list",
|
|
"content_type": "bullet_list",
|
|
"elements": taskItems,
|
|
})
|
|
|
|
return {
|
|
"metadata": {"title": title},
|
|
"documents": [{"id": "main", "title": title, "sections": sections}],
|
|
}
|
|
|
|
|
|
def _formatDate(isoStr: Optional[str]) -> str:
|
|
if not isoStr:
|
|
return datetime.now().strftime("%d.%m.%Y")
|
|
try:
|
|
dt = datetime.fromisoformat(str(isoStr).replace("Z", "+00:00"))
|
|
return dt.strftime("%d.%m.%Y")
|
|
except Exception:
|
|
return isoStr
|
|
|
|
|
|
def _parseJson(value, fallback):
|
|
if not value:
|
|
return fallback
|
|
if isinstance(value, (list, dict)):
|
|
return value
|
|
try:
|
|
return json.loads(value)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return fallback
|
|
|
|
|
|
def _groupScoresByDimension(scores: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
|
|
groups: Dict[str, List[Dict[str, Any]]] = {}
|
|
for s in scores:
|
|
dim = s.get("dimension", "unknown")
|
|
if dim not in groups:
|
|
groups[dim] = []
|
|
groups[dim].append(s)
|
|
for dim in groups:
|
|
groups[dim].sort(key=lambda x: x.get("createdAt") or "")
|
|
return groups
|