diff --git a/backend/Dockerfile b/backend/Dockerfile index 0bccd57a..73e68e91 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -6,6 +6,10 @@ WORKDIR /app RUN apt-get update && apt-get install -y \ gcc \ libpq-dev \ + libpango1.0-dev \ + libcairo2-dev \ + libgdk-pixbuf2.0-dev \ + libffi-dev \ && rm -rf /var/lib/apt/lists/* # Install Python dependencies diff --git a/backend/app/api/endpoints/sessions.py b/backend/app/api/endpoints/sessions.py index e0e303a2..47a238a5 100644 --- a/backend/app/api/endpoints/sessions.py +++ b/backend/app/api/endpoints/sessions.py @@ -391,6 +391,22 @@ async def export_session( detail="You don't have access to this session" ) + # PDF export — separate path with binary response + if export_options.format == "pdf": + from app.services.export_service import generate_pdf_export + from fastapi.responses import Response + pdf_bytes = await generate_pdf_export(session, export_options, db) + + if session.completed_at: + session.exported = True + await db.commit() + + return Response( + content=pdf_bytes, + media_type="application/pdf", + headers={"Content-Disposition": f'attachment; filename="session-export-{session_id}.pdf"'}, + ) + # Generate export based on format if export_options.format == "markdown": content = generate_markdown_export(session, export_options) diff --git a/backend/app/services/export_service.py b/backend/app/services/export_service.py index 8654eb89..15e1426b 100644 --- a/backend/app/services/export_service.py +++ b/backend/app/services/export_service.py @@ -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 diff --git a/backend/app/templates/export_pdf.html b/backend/app/templates/export_pdf.html new file mode 100644 index 00000000..8a7b7c0a --- /dev/null +++ b/backend/app/templates/export_pdf.html @@ -0,0 +1,378 @@ + + + + + + + + + +
+
+
{{ report_type }}
+
{{ flow_title }}
+ {% if company_name %} +
{{ company_name }}
+ {% endif %} +
+ {% if logo_data %} + + {% endif %} +
+ + +
+
+ Engineer + {{ engineer_name or "N/A" }} +
+
+ Client + {{ client_name or "N/A" }} +
+
+ Ticket # + {{ ticket_number or "N/A" }} +
+
+ Date + {{ session_date }} +
+
+ Duration + {{ duration }} +
+
+ Outcome + {{ outcome_display }} +
+
+ + + {% if summary %} +
+
Summary
+
{{ summary }}
+
+ {% endif %} + + + {% if steps %} +
+
Troubleshooting Path
+ {% for step in steps %} +
+
+
{{ loop.index }}. {{ step.title }}
+ {% if step.decision %} +
{{ step.decision }}
+ {% endif %} + {% if step.notes %} +
{{ step.notes }}
+ {% endif %} + {% if step.duration %} +
{{ step.duration }}
+ {% endif %} +
+
+ {% endfor %} +
+ {% endif %} + + + {% if supporting_data %} +
+
Supporting Data
+ {% for item in supporting_data %} +
+
{{ item.label }}
+
+ {% if item.data_type == "screenshot" %} + {{ item.label }} + {% else %} +
{{ item.content }}
+ {% endif %} +
+
+ {% endfor %} +
+ {% endif %} + + + diff --git a/backend/requirements.txt b/backend/requirements.txt index 189c871e..b884f6e7 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -42,6 +42,10 @@ voyageai>=0.3.0 # Monitoring sentry-sdk[fastapi]>=2.54.0 +# PDF Export +weasyprint>=62.0 +jinja2>=3.1.0 + # Utilities python-dotenv==1.0.1 croniter>=2.0.0 diff --git a/backend/tests/test_pdf_export.py b/backend/tests/test_pdf_export.py new file mode 100644 index 00000000..da2666dc --- /dev/null +++ b/backend/tests/test_pdf_export.py @@ -0,0 +1,96 @@ +"""Tests for PDF export via WeasyPrint.""" + +import pytest +from httpx import AsyncClient + + +@pytest.mark.asyncio +class TestPDFExport: + """Test PDF export endpoint.""" + + async def test_export_pdf_returns_pdf_content( + self, client: AsyncClient, auth_headers: dict, test_tree: dict + ): + """Test that PDF export returns application/pdf content starting with %PDF.""" + # Create a session + create_response = await client.post( + "/api/v1/sessions", + json={"tree_id": test_tree["id"], "ticket_number": "PDF-001"}, + headers=auth_headers, + ) + assert create_response.status_code in (200, 201) + session_id = create_response.json()["id"] + + # Add a decision so there's content + await client.put( + f"/api/v1/sessions/{session_id}", + json={ + "decisions": [ + { + "node_id": "root", + "question": "Is this a test?", + "answer": "Yes", + "notes": "PDF export test", + "timestamp": "2026-03-17T10:00:00Z", + } + ] + }, + headers=auth_headers, + ) + + # Export as PDF + response = await client.post( + f"/api/v1/sessions/{session_id}/export", + json={"format": "pdf", "include_tree_info": True}, + headers=auth_headers, + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "application/pdf" + assert "session-export-" in response.headers.get("content-disposition", "") + # PDF files start with %PDF + assert response.content[:5] == b"%PDF-" + + async def test_export_pdf_with_no_supporting_data( + self, client: AsyncClient, auth_headers: dict, test_tree: dict + ): + """Test PDF export works when session has no supporting data.""" + create_response = await client.post( + "/api/v1/sessions", + json={"tree_id": test_tree["id"]}, + headers=auth_headers, + ) + assert create_response.status_code in (200, 201) + session_id = create_response.json()["id"] + + response = await client.post( + f"/api/v1/sessions/{session_id}/export", + json={"format": "pdf"}, + headers=auth_headers, + ) + + assert response.status_code == 200 + assert response.headers["content-type"] == "application/pdf" + assert response.content[:5] == b"%PDF-" + + async def test_existing_markdown_export_still_works( + self, client: AsyncClient, auth_headers: dict, test_tree: dict + ): + """Verify markdown export is unaffected by PDF addition.""" + create_response = await client.post( + "/api/v1/sessions", + json={"tree_id": test_tree["id"], "ticket_number": "MD-001"}, + headers=auth_headers, + ) + assert create_response.status_code in (200, 201) + session_id = create_response.json()["id"] + + response = await client.post( + f"/api/v1/sessions/{session_id}/export", + json={"format": "markdown", "include_tree_info": True}, + headers=auth_headers, + ) + + assert response.status_code == 200 + assert "text/markdown" in response.headers["content-type"] + assert "MD-001" in response.text