chihlasm/resolutionflow
1
0
Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Implement session outcomes, step timing, and live timer fixes

Browse Source
This commit is contained in:
Michael Chihlas
2026-02-11 17:52:12 -05:00
parent 2a1ed4d250
commit ca4ce7cad6
15 changed files with 574 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
backend/app/services/export_service.py
View File

@@ -5,12 +5,21 @@ Provides markdown, plain text, HTML, and PSA/ticket note export formatters
for troubleshooting sessions.
"""
import html
from datetime import datetime, timezone
from datetime import datetime
from typing import Any
from app.models.session import Session
from app.schemas.session import SessionExport
OUTCOME_LABELS = {
"resolved": "Resolved",
"escalated": "Escalated",
"workaround": "Workaround Applied",
"unresolved": "Unresolved",
}
def _format_duration(started_at: datetime, completed_at: datetime | None) -> str:
"""Format duration between two datetimes as human-readable string."""
if not completed_at:
@@ -26,9 +35,64 @@ def _format_duration(started_at: datetime, completed_at: datetime | None) -> str
return f"{minutes} minutes"
def _format_step_duration(duration_seconds: int) -> str:
"""Format step duration seconds as compact human-readable text."""
if duration_seconds < 0:
return "0s"
if duration_seconds < 60:
return f"{duration_seconds}s"
hours, remainder = divmod(duration_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
if hours > 0:
if seconds > 0:
return f"{hours}h {minutes}m {seconds}s"
return f"{hours}h {minutes}m"
if seconds > 0:
return f"{minutes}m {seconds}s"
return f"{minutes}m"
def _parse_iso_datetime(value: Any) -> datetime | None:
"""Parse ISO datetime strings from JSONB with support for trailing Z."""
if isinstance(value, datetime):
return value
if not isinstance(value, str) or not value:
return None
try:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
def _get_step_duration_seconds(decision: dict[str, Any]) -> int | None:
"""Get step duration from explicit field or entered/exited timestamps."""
explicit_duration = decision.get("duration_seconds")
if isinstance(explicit_duration, (int, float)):
duration = int(explicit_duration)
if duration >= 0:
return duration
entered_at = _parse_iso_datetime(decision.get("entered_at"))
exited_at = _parse_iso_datetime(decision.get("exited_at"))
if entered_at and exited_at:
total_seconds = int((exited_at - entered_at).total_seconds())
if total_seconds >= 0:
return total_seconds
return None
def _get_outcome_label(session: Session) -> str | None:
"""Map stored outcome enum to human-friendly label."""
outcome = getattr(session, "outcome", None)
if not outcome:
return None
return OUTCOME_LABELS.get(outcome, str(outcome).replace("_", " ").title())
def generate_markdown_export(session: Session, options: SessionExport) -> str:
"""Generate markdown export."""
lines = []
outcome_label = _get_outcome_label(session)
if options.include_tree_info:
tree_name = session.tree_snapshot.get("name", "Troubleshooting Session")
@@ -42,6 +106,9 @@ def generate_markdown_export(session: Session, options: SessionExport) -> str:
lines.append(f"**Started:** {session.started_at.strftime('%Y-%m-%d %H:%M')}")
if session.completed_at:
lines.append(f"**Completed:** {session.completed_at.strftime('%Y-%m-%d %H:%M')}")
lines.append(f"**Duration:** {_format_duration(session.started_at, session.completed_at)}")
if outcome_label:
lines.append(f"**Outcome:** {outcome_label}")
lines.append("")
lines.append("---")
lines.append("")
@@ -63,12 +130,15 @@ def generate_markdown_export(session: Session, options: SessionExport) -> str:
question = decision.get("question") or decision.get("action_performed", "Step")
answer = decision.get("answer", "")
notes = decision.get("notes", "")
duration_seconds = _get_step_duration_seconds(decision)
lines.append(f"### Step {i}: {question}")
if answer:
lines.append(f"**Answer:** {answer}")
if notes:
lines.append(f"**Notes:** {notes}")
if duration_seconds is not None:
lines.append(f"**Duration:** {_format_step_duration(duration_seconds)}")
if options.include_timestamps and decision.get("timestamp"):
lines.append(f"*{decision['timestamp']}*")
lines.append("")
@@ -79,6 +149,7 @@ def generate_markdown_export(session: Session, options: SessionExport) -> str:
def generate_text_export(session: Session, options: SessionExport) -> str:
"""Generate plain text export."""
lines = []
outcome_label = _get_outcome_label(session)
if options.include_tree_info:
tree_name = session.tree_snapshot.get("name", "Troubleshooting Session")
@@ -92,6 +163,9 @@ def generate_text_export(session: Session, options: SessionExport) -> str:
lines.append(f"Started: {session.started_at.strftime('%Y-%m-%d %H:%M')}")
if session.completed_at:
lines.append(f"Completed: {session.completed_at.strftime('%Y-%m-%d %H:%M')}")
lines.append(f"Duration: {_format_duration(session.started_at, session.completed_at)}")
if outcome_label:
lines.append(f"Outcome: {outcome_label}")
lines.append("")
# Scratchpad / Evidence section
@@ -109,12 +183,15 @@ def generate_text_export(session: Session, options: SessionExport) -> str:
question = decision.get("question") or decision.get("action_performed", "Step")
answer = decision.get("answer", "")
notes = decision.get("notes", "")
duration_seconds = _get_step_duration_seconds(decision)
lines.append(f"\n{i}. {question}")
if answer:
lines.append(f" Answer: {answer}")
if notes:
lines.append(f" Notes: {notes}")
if duration_seconds is not None:
lines.append(f" Duration: {_format_step_duration(duration_seconds)}")
return "\n".join(lines)
@@ -122,6 +199,7 @@ def generate_text_export(session: Session, options: SessionExport) -> str:
def generate_html_export(session: Session, options: SessionExport) -> str:
"""Generate HTML export."""
tree_name = html.escape(session.tree_snapshot.get("name", "Troubleshooting Session"))
outcome_label = _get_outcome_label(session)
html_parts = ['<!DOCTYPE html>', '<html>', '<head>',
'<meta charset="UTF-8">',
@@ -134,6 +212,7 @@ def generate_html_export(session: Session, options: SessionExport) -> str:
'.step h3 { margin: 0 0 10px 0; color: #444; }',
'.answer { font-weight: bold; }',
'.notes { font-style: italic; color: #555; }',
'.duration { color: #444; margin-top: 6px; }',
'.timestamp { font-size: 0.85em; color: #888; }',
'</style>',
'</head>', '<body>']
@@ -149,6 +228,9 @@ def generate_html_export(session: Session, options: SessionExport) -> str:
html_parts.append(f'<p><strong>Started:</strong> {session.started_at.strftime("%Y-%m-%d %H:%M")}</p>')
if session.completed_at:
html_parts.append(f'<p><strong>Completed:</strong> {session.completed_at.strftime("%Y-%m-%d %H:%M")}</p>')
html_parts.append(f'<p><strong>Duration:</strong> {_format_duration(session.started_at, session.completed_at)}</p>')
if outcome_label:
html_parts.append(f'<p><strong>Outcome:</strong> {html.escape(outcome_label)}</p>')
html_parts.append('</div>')
# Scratchpad / Evidence section
@@ -163,6 +245,7 @@ def generate_html_export(session: Session, options: SessionExport) -> str:
question = html.escape(decision.get("question") or decision.get("action_performed", "Step"))
answer = html.escape(decision.get("answer", ""))
notes = html.escape(decision.get("notes", ""))
duration_seconds = _get_step_duration_seconds(decision)
html_parts.append('<div class="step">')
html_parts.append(f'<h3>Step {i}: {question}</h3>')
@@ -170,6 +253,8 @@ def generate_html_export(session: Session, options: SessionExport) -> str:
html_parts.append(f'<p class="answer">Answer: {answer}</p>')
if notes:
html_parts.append(f'<p class="notes">Notes: {notes}</p>')
if duration_seconds is not None:
html_parts.append(f'<p class="duration"><strong>Duration:</strong> {_format_step_duration(duration_seconds)}</p>')
if options.include_timestamps and decision.get("timestamp"):
html_parts.append(f'<p class="timestamp">{html.escape(str(decision["timestamp"]))}</p>')
html_parts.append('</div>')
@@ -181,6 +266,7 @@ def generate_html_export(session: Session, options: SessionExport) -> str:
def generate_psa_export(session: Session, options: SessionExport) -> str:
"""Generate PSA/ticket note export optimized for ConnectWise and similar PSA tools."""
lines = []
outcome_label = _get_outcome_label(session)
tree_name = session.tree_snapshot.get("name", "Troubleshooting Session")
tree_description = session.tree_snapshot.get("description", "")
@@ -191,6 +277,9 @@ def generate_psa_export(session: Session, options: SessionExport) -> str:
lines.append("=== TROUBLESHOOTING NOTES ===")
lines.append(f"Ticket: {ticket_number} | Client: {client_name}")
lines.append(f"Tree: {tree_name} | Date: {date_str}")
lines.append(f"Duration: {_format_duration(session.started_at, session.completed_at)}")
if outcome_label:
lines.append(f"Outcome: {outcome_label}")
lines.append("")
# Problem section
@@ -205,10 +294,13 @@ def generate_psa_export(session: Session, options: SessionExport) -> str:
question = decision.get("question") or decision.get("action_performed", "Step")
answer = decision.get("answer", "")
notes = decision.get("notes", "")
duration_seconds = _get_step_duration_seconds(decision)
line = f"{i}. {question}"
if answer:
line += f" -> {answer}"
if duration_seconds is not None:
line += f" ({_format_step_duration(duration_seconds)})"
lines.append(line)
if notes:
lines.append(f" Notes: {notes}")
@@ -224,6 +316,8 @@ def generate_psa_export(session: Session, options: SessionExport) -> str:
lines.append(resolution)
else:
lines.append("No resolution recorded.")
if outcome_label:
lines.append(f"Outcome: {outcome_label}")
lines.append("")
# Time spent