feat: add PDF export generation via WeasyPrint with branded template
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
chihlasm
@@ -1,11 +1,13 @@
|
||||
"""
|
||||
Session export generators for ResolutionFlow.
|
||||
|
||||
Provides markdown, plain text, HTML, and PSA/ticket note export formatters
|
||||
Provides markdown, plain text, HTML, PDF, and PSA/ticket note export formatters
|
||||
for troubleshooting sessions.
|
||||
"""
|
||||
import html
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.models.session import Session
|
||||
@@ -904,3 +906,188 @@ def _generate_procedural_psa(session: Session, options: SessionExport) -> str:
|
||||
lines.append(f"Duration: {_format_duration(session.started_at, session.completed_at)}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
async def generate_pdf_export(session: Session, options: SessionExport, db) -> bytes:
|
||||
"""Generate PDF export using WeasyPrint and a Jinja2 HTML template.
|
||||
|
||||
Args:
|
||||
session: The session to export.
|
||||
options: Export options (redaction_mode, max_step_index, etc.).
|
||||
db: Async database session for loading supporting data and branding.
|
||||
|
||||
Returns:
|
||||
PDF file contents as bytes.
|
||||
"""
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
import weasyprint
|
||||
from sqlalchemy import select as sa_select
|
||||
|
||||
# Load Jinja2 template
|
||||
template_dir = Path(__file__).resolve().parent.parent / "templates"
|
||||
env = Environment(loader=FileSystemLoader(str(template_dir)), autoescape=True)
|
||||
template = env.get_template("export_pdf.html")
|
||||
|
||||
# Tree snapshot data
|
||||
tree_snapshot = session.tree_snapshot or {}
|
||||
flow_title = tree_snapshot.get("name", "Session Export")
|
||||
tree_type = tree_snapshot.get("tree_type", "troubleshooting")
|
||||
is_procedural = tree_type == "procedural"
|
||||
report_type = "Procedure Report" if is_procedural else "Troubleshooting Report"
|
||||
|
||||
# Branding — check team first, then user (solo pros)
|
||||
logo_data = None
|
||||
logo_content_type = None
|
||||
company_name = None
|
||||
|
||||
from app.models.user import User
|
||||
user_result = await db.execute(
|
||||
sa_select(User).where(User.id == session.user_id)
|
||||
)
|
||||
user = user_result.scalar_one_or_none()
|
||||
engineer_name = user.name if user else "Unknown"
|
||||
|
||||
if user and user.team_id:
|
||||
from app.models.team import Team
|
||||
team_result = await db.execute(
|
||||
sa_select(Team).where(Team.id == user.team_id)
|
||||
)
|
||||
team = team_result.scalar_one_or_none()
|
||||
if team:
|
||||
logo_data = team.logo_data
|
||||
logo_content_type = team.logo_content_type
|
||||
company_name = team.company_display_name or team.name
|
||||
elif user:
|
||||
logo_data = user.logo_data
|
||||
logo_content_type = user.logo_content_type
|
||||
company_name = user.company_display_name
|
||||
|
||||
has_custom_logo = bool(logo_data)
|
||||
|
||||
# Build steps list from decisions
|
||||
decisions = session.decisions or []
|
||||
if options.max_step_index is not None:
|
||||
decisions = decisions[:options.max_step_index]
|
||||
|
||||
steps = []
|
||||
for decision in decisions:
|
||||
title = decision.get("question") or decision.get("action_performed", "Step")
|
||||
answer = decision.get("answer", "")
|
||||
notes = decision.get("notes", "")
|
||||
duration_seconds = _get_step_duration_seconds(decision)
|
||||
duration_str = _format_step_duration(duration_seconds) if duration_seconds is not None else None
|
||||
|
||||
if is_procedural:
|
||||
completed = answer == "completed"
|
||||
decision_text = "Completed" if completed else ("Skipped" if answer else "")
|
||||
else:
|
||||
decision_text = answer
|
||||
|
||||
steps.append({
|
||||
"title": title,
|
||||
"decision": decision_text,
|
||||
"notes": notes,
|
||||
"duration": duration_str,
|
||||
})
|
||||
|
||||
# Query supporting data
|
||||
from app.models.supporting_data import SessionSupportingData
|
||||
sd_result = await db.execute(
|
||||
sa_select(SessionSupportingData)
|
||||
.where(SessionSupportingData.session_id == session.id)
|
||||
.order_by(SessionSupportingData.sort_order)
|
||||
)
|
||||
supporting_data_rows = sd_result.scalars().all()
|
||||
supporting_data = [
|
||||
{
|
||||
"label": sd.label,
|
||||
"data_type": sd.data_type,
|
||||
"content": sd.content,
|
||||
"content_type": sd.content_type,
|
||||
}
|
||||
for sd in supporting_data_rows
|
||||
]
|
||||
|
||||
# Calculate duration and format outcome
|
||||
duration = _format_duration(session.started_at, session.completed_at)
|
||||
session_date = session.started_at.strftime("%Y-%m-%d %H:%M")
|
||||
outcome_label = _get_outcome_label(session) or ("In Progress" if not session.completed_at else "Completed")
|
||||
outcome_raw = getattr(session, "outcome", None) or ""
|
||||
outcome_class = f"outcome-{outcome_raw}" if outcome_raw else ""
|
||||
|
||||
# Build summary text
|
||||
summary_text = ""
|
||||
if options.include_summary:
|
||||
summary_fields = _build_summary_fields(session)
|
||||
parts = []
|
||||
for label, value in summary_fields.items():
|
||||
if value:
|
||||
parts.append(f"{label.replace('_', ' ').title()}: {value}")
|
||||
summary_text = "\n".join(parts)
|
||||
|
||||
# Resolution / outcome notes as summary fallback
|
||||
if not summary_text:
|
||||
_raw_notes = getattr(session, "outcome_notes", None)
|
||||
if isinstance(_raw_notes, str) and _raw_notes.strip():
|
||||
summary_text = _raw_notes.strip()
|
||||
|
||||
generated_at = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
|
||||
|
||||
# Variable resolution
|
||||
session_vars = getattr(session, "session_variables", None) or {}
|
||||
if session_vars:
|
||||
from app.services.variable_service import resolve_variables
|
||||
flow_title = resolve_variables(flow_title, session_vars)
|
||||
summary_text = resolve_variables(summary_text, session_vars)
|
||||
for step in steps:
|
||||
step["title"] = resolve_variables(step["title"], session_vars)
|
||||
if step["decision"]:
|
||||
step["decision"] = resolve_variables(step["decision"], session_vars)
|
||||
if step["notes"]:
|
||||
step["notes"] = resolve_variables(step["notes"], session_vars)
|
||||
for sd in supporting_data:
|
||||
if sd["data_type"] == "text_snippet":
|
||||
sd["content"] = resolve_variables(sd["content"], session_vars)
|
||||
|
||||
# Apply redaction
|
||||
if options.redaction_mode == "mask":
|
||||
from app.services.redaction_service import apply_redaction_to_text
|
||||
try:
|
||||
flow_title, _ = apply_redaction_to_text(flow_title)
|
||||
summary_text, _ = apply_redaction_to_text(summary_text)
|
||||
for step in steps:
|
||||
step["title"], _ = apply_redaction_to_text(step["title"])
|
||||
if step["decision"]:
|
||||
step["decision"], _ = apply_redaction_to_text(step["decision"])
|
||||
if step["notes"]:
|
||||
step["notes"], _ = apply_redaction_to_text(step["notes"])
|
||||
for sd in supporting_data:
|
||||
if sd["data_type"] == "text_snippet":
|
||||
sd["content"], _ = apply_redaction_to_text(sd["content"])
|
||||
except Exception:
|
||||
pass # Redaction is best-effort for PDF
|
||||
|
||||
# Render HTML
|
||||
html_content = template.render(
|
||||
report_type=report_type,
|
||||
flow_title=flow_title,
|
||||
logo_data=logo_data,
|
||||
logo_content_type=logo_content_type or "image/png",
|
||||
has_custom_logo=has_custom_logo,
|
||||
company_name=company_name,
|
||||
engineer_name=engineer_name,
|
||||
client_name=session.client_name,
|
||||
ticket_number=session.ticket_number,
|
||||
session_date=session_date,
|
||||
duration=duration,
|
||||
outcome_class=outcome_class,
|
||||
outcome_display=outcome_label,
|
||||
summary=summary_text,
|
||||
steps=steps,
|
||||
supporting_data=supporting_data,
|
||||
generated_at=generated_at,
|
||||
)
|
||||
|
||||
# Convert to PDF
|
||||
pdf_bytes = weasyprint.HTML(string=html_content).write_pdf()
|
||||
return pdf_bytes
|
||||
|
||||
Reference in New Issue
Block a user