chat summary generator and pdf formatter

multi_llm_chatbot_backend/app/utils/chat_summary.py

@@ -0,0 +1,92 @@
+from typing import List
+from app.llm.llm_client import LLMClient
+import logging
+import re
+from typing import List, Dict
+
+logger = logging.getLogger(__name__)
+
+async def generate_summary_from_messages(messages: List[dict], llm: LLMClient, max_tokens: int = 800) -> str:
+    """
+    Summarize the conversation using the given LLM client.
+    """
+    try:
+        full_text = "\n\n".join([f"{m['role']}:\n{m['content']}" for m in messages])
+
+        system_prompt = (
+            "You are an academic assistant. Summarize the following PhD chat conversation "
+            "into concise bullet points (max 10) or short paragraphs. Focus on insights, questions, and advice."
+        )
+
+        context = [{"role": "user", "content": f"Chat Log:\n{full_text}"}]
+
+        summary = await llm.generate(
+            system_prompt=system_prompt,
+            context=context,
+            temperature=0.4,
+            max_tokens=max_tokens
+        )
+
+        return summary.strip()
+
+    except Exception as e:
+        logger.error(f"Error generating summary: {str(e)}")
+        return "Summary generation failed. Please try again later."
+
+def parse_summary_to_blocks(summary_text: str) -> List[Dict]:
+    #summary_text = re.sub(r'(?<!\n)([*•] )', r'\n\1', summary_text)
+    #summary_text = re.sub(r'(?<!\n)(\d+\.\s+)', r'\n\1', summary_text)
+    #summary_text = re.sub(r'(?<=[.!?])(?=\S)', ' ', summary_text)
+
+    lines = summary_text.strip().splitlines()
+    blocks = []
+    current_block = None
+
+    def flush_current_block():
+        if current_block:
+            blocks.append(current_block.copy())
+
+    for line in lines:
+        line = line.strip()
+        if not line:
+            continue
+
+        # Match section headings (e.g. **Title:**) as heading block
+        heading_match = re.match(r'^\*\*(.+?)\*\*:?$', line)
+        if heading_match:
+            flush_current_block()
+            current_block = {"type": "heading", "text": heading_match.group(1).strip()}
+            flush_current_block()
+            current_block = None
+            continue
+
+        # Match bullet list
+        if line.startswith("* "):
+            if current_block is None or current_block["type"] != "list" or current_block.get("style") != "bullet":
+                flush_current_block()
+                current_block = {"type": "list", "style": "bullet", "items": []}
+            current_block["items"].append(line[2:].strip())
+            continue
+
+        # Match numbered list
+        number_match = re.match(r'^\d+\.\s+(.*)', line)
+        if number_match:
+            if current_block is None or current_block["type"] != "list" or current_block.get("style") != "numbered":
+                flush_current_block()
+                current_block = {"type": "list", "style": "numbered", "items": []}
+            current_block["items"].append(number_match.group(1).strip())
+            continue
+
+        # Default: treat as paragraph
+        flush_current_block()
+        current_block = {"type": "paragraph", "text": line}
+        flush_current_block()
+        current_block = None
+
+    flush_current_block()
+
+    import pprint
+    print("[DEBUG] Summary Blocks:")
+    pprint.pprint(blocks)
+    return blocks
+

multi_llm_chatbot_backend/app/utils/file_export.py

@@ -1,9 +1,14 @@
 from io import BytesIO
-from typing import List, Tuple
+from typing import List, Tuple, Union
 from docx import Document
 from reportlab.lib.pagesizes import letter
 from reportlab.pdfgen import canvas
-
+from fastapi.responses import StreamingResponse
+from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, ListFlowable, ListItem
+from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
+from reportlab.lib.enums import TA_CENTER, TA_LEFT
+from io import BytesIO
+import re
 
 def format_messages_for_export(messages: List[dict]) -> str:
     """
@@ -33,42 +38,123 @@ def generate_docx_file(text: str) -> BytesIO:
 
 def generate_pdf_file(text: str) -> BytesIO:
     buffer = BytesIO()
-    c = canvas.Canvas(buffer, pagesize=letter)
+    doc = SimpleDocTemplate(buffer, pagesize=letter)
 
-    width, height = letter
-    y = height - 40
-    line_height = 14
+    styles = getSampleStyleSheet()
+    story = []
 
-    for line in text.splitlines():
-        if y < 40:
-            c.showPage()
-            y = height - 40
-        c.drawString(40, y, line)
-        y -= line_height
+    for block in text.split("\n\n"):
+        story.append(Paragraph(block.strip(), styles["Normal"]))
+        story.append(Spacer(1, 12))
 
-    c.save()
+    doc.build(story)
     buffer.seek(0)
     return buffer
 
-
-def export_chat_as_file(messages: List[dict], format: str) -> Tuple[BytesIO, str, str]:
+def export_chat_as_file(content: Union[str, List[dict]], format: str) -> Tuple[BytesIO, str, str]:
     """
-    Export chat messages to the requested file format.
-    Returns: (file_stream, filename, media_type)
+    Export either a list of chat messages or a summary string to the specified format.
     """
-    text = format_messages_for_export(messages)
+    if isinstance(content, list):
+        text = format_messages_for_export(content)
+    elif isinstance(content, str):
+        text = content.strip()
+    else:
+        raise ValueError("Unsupported content type")
 
     if format == "txt":
-        stream = generate_txt_file(text)
-        return stream, "chat_export.txt", "text/plain"
+        return generate_txt_file(text), "chat_export.txt", "text/plain"
 
     elif format == "docx":
-        stream = generate_docx_file(text)
-        return stream, "chat_export.docx", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+        return generate_docx_file(text), "chat_export.docx", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
 
     elif format == "pdf":
-        stream = generate_pdf_file(text)
-        return stream, "chat_export.pdf", "application/pdf"
+        return generate_pdf_file(text), "chat_export.pdf", "application/pdf"
 
     else:
         raise ValueError(f"Unsupported export format: {format}")
+    
+def prepare_export_response(
+    content: Union[str, List[dict]],
+    format: str,
+    filename_prefix: str = "chat_export"
+) -> StreamingResponse:
+    """
+    Prepare a StreamingResponse for export, using the given filename prefix.
+    """
+    stream, filename, media_type = export_chat_as_file(content, format)
+
+    # Replace "chat_export" with custom prefix if needed
+    final_filename = filename.replace("chat_export", filename_prefix)
+
+    return StreamingResponse(
+        stream,
+        media_type=media_type,
+        headers={"Content-Disposition": f"attachment; filename={final_filename}"}
+    )
+
+
+def _render_rich_text(text: str) -> str:
+    """
+    Convert markdown-style **bold** to reportlab <b>bold</b> tags.
+    """
+    return re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
+
+
+def generate_pdf_file_from_blocks(blocks: List[dict]) -> BytesIO:
+    buffer = BytesIO()
+    doc = SimpleDocTemplate(buffer, pagesize=letter)
+
+    styles = getSampleStyleSheet()
+
+    heading_style = ParagraphStyle(
+        name="Heading",
+        parent=styles["Heading1"],
+        alignment=TA_CENTER,
+        spaceAfter=20
+    )
+
+    paragraph_style = ParagraphStyle(
+        name="Paragraph",
+        parent=styles["Normal"],
+        alignment=TA_LEFT,
+        leading=14,
+        spaceAfter=8
+    )
+
+    list_style = ParagraphStyle(
+        name="List",
+        parent=styles["Normal"],
+        leftIndent=20,
+        leading=14
+    )
+
+    story = []
+
+    for block in blocks:
+        if block["type"] == "heading":
+            story.append(Paragraph(_render_rich_text(block["text"]), heading_style))
+            story.append(Spacer(1, 12))
+
+        elif block["type"] == "paragraph":
+            story.append(Paragraph(_render_rich_text(block["text"]), paragraph_style))
+
+        elif block["type"] == "list":
+            list_items = [
+                ListItem(Paragraph(_render_rich_text(item), list_style))
+                for item in block["items"]
+            ]
+            story.append(
+                ListFlowable(
+                    list_items,
+                    bulletType="1" if block.get("style") == "numbered" else "bullet",
+                    start="1" if block.get("style") == "numbered" else None,
+                    leftIndent=20
+                )
+            )
+            story.append(Spacer(1, 8))
+
+    doc.build(story)
+    buffer.seek(0)
+    return buffer
+