Executor / download_messages_service.py
Soumik Bose
go
bd8337b
Raw
History Blame Contribute Delete
29.3 kB
"""
Professional Chat Message Log PDF Generator - Soft Slate Blue Theme
Generates beautifully formatted PDF reports from chat conversations
Uses WeasyPrint for superior CSS support and rendering
Supports file attachments (CSV and Image files)
"""
import markdown
from weasyprint import HTML, CSS
import requests
import os
import tempfile
import re
from datetime import datetime
from typing import List, Dict, Optional
import json
import random
import string
import logging
# ---------------------------------------------------------
# LOGGING CONFIGURATION
# ---------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------
# SOFT SLATE BLUE THEME - WEASYPRINT OPTIMIZED
# ---------------------------------------------------------
CSS_STYLE = """
@page {
size: letter;
margin: 0.75in 0.75in 1in 0.75in;
@top-center {
content: element(header);
}
@bottom-center {
content: element(footer);
}
}
body {
font-family: 'Helvetica', Arial, sans-serif;
color: #2d3748;
line-height: 1.6;
font-size: 11pt;
}
.header {
position: running(header);
font-size: 9pt;
color: #718096;
border-bottom: 1px solid #e2e8f0;
padding-bottom: 4px;
margin-bottom: 20px;
}
.footer {
position: running(footer);
font-size: 9pt;
color: #718096;
border-top: 1px solid #e2e8f0;
padding-top: 4px;
text-align: center;
}
.cover-page {
text-align: center;
padding-top: 3in;
page-break-after: always;
}
.cover-title {
font-size: 32pt;
font-weight: bold;
color: #4a5568;
margin-bottom: 0.5in;
letter-spacing: 1px;
}
.cover-subtitle {
font-size: 18pt;
color: #5b7c99;
margin-bottom: 1in;
}
.cover-meta {
font-size: 14pt;
color: #718096;
margin-bottom: 0.2in;
}
.cover-date {
font-size: 11pt;
color: #a0aec0;
}
.message-container {
margin-bottom: 20px;
page-break-inside: avoid;
}
.message-user {
background-color: #ffffff;
border-left: 0.5px solid #5b7c99;
padding: 16px 20px;
margin-bottom: 16px;
box-shadow: 0 1px 3px rgba(0,0,0,0.05);
}
.message-assistant {
background-color: #ffffff;
border-left: 0.5px solid #4a5568;
padding: 16px 20px;
margin-bottom: 16px;
box-shadow: 0 1px 3px rgba(0,0,0,0.05);
}
.message-header {
font-weight: bold;
font-size: 10pt;
margin-bottom: 8px;
display: flex;
justify-content: space-between;
}
.role-user {
color: #4a5568;
font-size: 11pt;
}
.role-assistant {
color: #4a5568;
font-size: 11pt;
}
.timestamp {
font-size: 9pt;
color: #a0aec0;
font-weight: normal;
}
.message-content {
color: #4a5568;
font-size: 10pt;
line-height: 1.5;
word-wrap: break-word;
overflow-wrap: break-word;
}
a {
color: #5b7c99;
text-decoration: none;
word-wrap: break-word;
overflow-wrap: break-word;
}
a:hover {
text-decoration: underline;
}
.message-content a {
font-weight: 500;
}
.message-divider {
border-top: 1px solid #e2e8f0;
margin: 16px 0;
}
h1, h2, h3, h4 {
color: #4a5568;
margin-top: 12px;
margin-bottom: 8px;
}
h1 { font-size: 16pt; }
h2 { font-size: 14pt; color: #5b7c99; }
h3 { font-size: 12pt; }
h4 { font-size: 11pt; }
table {
width: 100%;
border-collapse: collapse;
margin: 12px 0;
font-size: 9pt;
table-layout: fixed;
}
/* For tables with many columns, reduce font size */
table.many-columns {
font-size: 7pt;
}
table.many-columns th,
table.many-columns td {
padding: 4px 6px;
}
th {
background-color: #5b7c99;
color: white;
padding: 8px;
text-align: left;
font-weight: bold;
border: 1px solid #4a6b85;
word-wrap: break-word;
overflow-wrap: break-word;
hyphens: auto;
}
td {
padding: 6px 8px;
border: 1px solid #e2e8f0;
word-wrap: break-word;
overflow-wrap: break-word;
max-width: 0;
overflow: hidden;
hyphens: auto;
font-size: inherit;
}
tr:nth-child(even) {
background-color: #f7fafc;
}
code {
background-color: #edf2f7;
padding: 2px 6px;
border-radius: 4px;
font-family: 'Courier New', monospace;
font-size: 9pt;
color: #2d3748;
}
pre {
background-color: #edf2f7;
border: 1px solid #cbd5e0;
border-left: none;
padding: 12px;
font-size: 8pt;
line-height: 1.4;
margin: 10px 0;
border-radius: 6px;
white-space: pre-wrap;
word-wrap: break-word;
overflow-wrap: break-word;
}
pre code {
background-color: transparent;
padding: 0;
color: #2d3748;
}
blockquote {
border-left: 3px solid #5b7c99;
margin-left: 0;
padding-left: 12px;
color: #4a5568;
font-style: italic;
background-color: #f7fafc;
padding: 8px 8px 8px 12px;
margin: 10px 0;
}
ul, ol {
margin: 8px 0;
padding-left: 20px;
list-style-type: none;
}
ul li:before {
content: "• ";
color: #5b7c99;
font-weight: bold;
display: inline-block;
width: 1em;
margin-left: -1em;
}
li {
margin: 4px 0;
}
.figure-container {
text-align: center;
margin: 16px 0;
page-break-inside: avoid;
}
.figure-container img {
max-width: 100%;
max-height: 400px;
height: auto;
border: 1px solid #e2e8f0;
padding: 4px;
background: white;
}
.figure-caption {
font-size: 9pt;
color: #718096;
margin-top: 6px;
font-style: italic;
}
.file-item {
margin-bottom: 12px;
padding: 12px;
background-color: #f7fafc;
border-left: 3px solid #5b7c99;
page-break-inside: avoid;
}
.file-label {
color: #4a5568;
font-weight: bold;
font-size: 9pt;
}
.file-value {
color: #4a5568;
font-size: 9pt;
word-wrap: break-word;
overflow-wrap: break-word;
}
.file-url {
color: #5b7c99;
font-size: 7pt;
font-family: 'Courier New', monospace;
display: block;
padding: 6px 8px;
background-color: #edf2f7;
border-radius: 3px;
margin-top: 4px;
word-wrap: break-word;
overflow-wrap: break-word;
line-height: 1.5;
}
.file-url a {
color: #5b7c99;
word-wrap: break-word;
overflow-wrap: break-word;
font-size: 7pt;
font-family: 'Courier New', monospace;
}
.stats-summary {
margin: 30px 0 20px 0;
}
.stats-summary p {
margin: 0;
padding: 0;
font-size: 10pt;
color: #718096;
}
.stats-summary strong {
color: #4a5568;
}
.section-divider {
border: none;
border-top: 2px solid #5b7c99;
margin: 20px 0 30px 0;
}
"""
# ---------------------------------------------------------
# CHAT LOG GENERATOR CLASS
# ---------------------------------------------------------
class ChatLogGenerator:
"""
Generates professional PDF reports from chat message logs.
Supports multiple JSON formats (createdAt, timestamp, chatId, chat_id, etc.)
"""
def __init__(self,
title: str = "Chat Conversation Log",
chat_id: str = None,
participants: List[str] = None,
date_range: str = None,
confidential: bool = False,
output_dir: str = "chat_pdfs"):
"""
Initialize the chat log generator.
Args:
title: Title of the chat log report
chat_id: Unique identifier for the chat session
participants: List of participant names
date_range: Date range of the conversation
confidential: Whether to mark as confidential
output_dir: Directory to save PDF files
"""
self.title = title
self.chat_id = chat_id
self.participants = participants or ["User", "Assistant"]
self.date_range = date_range
self.confidential = confidential
self.output_dir = output_dir
# Create output directory if it doesn't exist
os.makedirs(self.output_dir, exist_ok=True)
logger.info(f"ChatLogGenerator initialized - output_dir: {self.output_dir}")
def _get_timestamp_from_message(self, message: Dict) -> str:
"""
Extract timestamp from message, checking multiple possible field names.
Supports: timestamp, createdAt, created_at
"""
# Try different field names
for field in ['timestamp', 'createdAt', 'created_at']:
if field in message and message[field]:
return message[field]
# Return empty string if no timestamp found
return ''
def _generate_unique_filename(self, base_name: str = "chat_log") -> str:
"""
Generate a unique filename with random number.
Args:
base_name: Base name for the file
Returns:
Full path to the PDF file
"""
# Generate random 8-digit number
random_num = ''.join(random.choices(string.digits, k=8))
# Get current timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Create filename
filename = f"{base_name}_{timestamp}_{random_num}.pdf"
# Return full path
filepath = os.path.join(self.output_dir, filename)
logger.debug(f"Generated unique filename: {filepath}")
return filepath
def _format_timestamp(self, timestamp: str) -> str:
"""
Format ISO timestamp to readable format.
Handles both with and without timezone info.
"""
if not timestamp or timestamp.strip() == '':
return "No timestamp"
try:
# Try parsing with timezone first
if '+' in timestamp or timestamp.endswith('Z'):
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
else:
# Try parsing without timezone
dt = datetime.fromisoformat(timestamp)
return dt.strftime("%B %d, %Y at %I:%M:%S %p")
except Exception as e:
logger.warning(f"Failed to parse timestamp: {timestamp} - {e}")
return timestamp if timestamp else "No timestamp"
def _convert_markdown_with_inline_images(self, content: str) -> str:
"""
Convert markdown to HTML while preserving inline image positions.
Images will appear exactly where they're referenced.
Standalone image URLs are hidden from display.
Auto-converts URLs to "Click to View" links.
"""
# Pattern to match markdown images: ![alt](url)
image_pattern = r'!\[([^\]]*)\]\(([^\)]+)\)'
# Find all image URLs that will be converted to <img> tags
image_urls = set()
for match in re.finditer(image_pattern, content):
image_urls.add(match.group(2))
# Split content by images while keeping the images
parts = re.split(f'({image_pattern})', content)
html_output = ""
i = 0
while i < len(parts):
part = parts[i]
# Check if this part is an image (starts with ![)
if part.startswith('!['):
# Extract alt text and URL from the matched groups
match = re.match(image_pattern, part)
if match:
alt_text = match.group(1) or "Figure"
url = match.group(2)
# Add the image HTML inline
html_output += f"""
<div class="figure-container">
<img src="{url}" alt="{alt_text}">
<div class="figure-caption">{alt_text}</div>
</div>
"""
i += 1
else:
# Regular text content - process URLs and hide standalone image URLs
if part.strip():
# Remove standalone image URLs that appear on their own lines
processed_part = self._hide_standalone_image_urls(part, image_urls)
# Convert URLs to "Click to View" links BEFORE markdown conversion
processed_part = self._convert_urls_to_links(processed_part)
# Convert markdown to HTML
html_output += markdown.markdown(
processed_part,
extensions=['extra', 'sane_lists', 'tables', 'fenced_code', 'nl2br']
)
i += 1
# Post-process tables to add responsive class for wide tables
html_output = self._make_tables_responsive(html_output)
return html_output
def _make_tables_responsive(self, html: str) -> str:
"""
Add 'many-columns' class to tables with more than 10 columns.
This reduces font size and padding for better fit.
"""
# Pattern to find tables
table_pattern = r'<table>(.*?)</table>'
def process_table(match):
table_content = match.group(1)
# Count columns by finding <th> tags in the first row
th_count = table_content.count('<th>')
# If more than 10 columns, add the many-columns class
if th_count > 10:
return f'<table class="many-columns">{table_content}</table>'
else:
return match.group(0)
return re.sub(table_pattern, process_table, html, flags=re.DOTALL)
def _convert_urls_to_links(self, text: str) -> str:
"""
Convert standalone URLs (http:// or https://) to "Click to View" links.
"""
lines = text.split('\n')
processed_lines = []
for line in lines:
stripped = line.strip()
# Check if line starts with http:// or https://
if stripped.startswith('http://') or stripped.startswith('https://'):
# Check if it's already a markdown link [text](url)
if not line.strip().startswith('['):
# Convert to "Click to View" link
processed_lines.append(f'<a href="{stripped}" target="_blank" style="color: #5b7c99; text-decoration: underline; font-weight: bold;">Click to View</a>')
else:
processed_lines.append(line)
else:
# Check for URLs within the line
# Pattern to find URLs not already in markdown links
url_pattern = r'(?<!\])\(?(https?://[^\s\)]+)\)?(?!\])'
def replace_url(match):
url = match.group(1)
return f'<a href="{url}" target="_blank" style="color: #5b7c99; text-decoration: underline; font-weight: bold;">Click to View</a>'
processed_line = re.sub(url_pattern, replace_url, line)
processed_lines.append(processed_line)
return '\n'.join(processed_lines)
def _hide_standalone_image_urls(self, text: str, image_urls: set) -> str:
"""
Hide standalone image URLs that are already displayed as images.
Keep other HTTP links intact.
"""
lines = text.split('\n')
processed_lines = []
for line in lines:
stripped = line.strip()
# Check if this line is ONLY an image URL (already shown as image)
if stripped in image_urls:
# Skip this line entirely (don't add it)
continue
# Check if line is just a URL with optional markdown link wrapper
# Pattern: https://... or [text](https://...)
url_only_pattern = r'^<?https?://[^\s>]+>?$'
if re.match(url_only_pattern, stripped):
# Keep the line as is
processed_lines.append(line)
else:
processed_lines.append(line)
return '\n'.join(processed_lines)
def _build_files_section(self, files_data: Dict) -> str:
"""
Build HTML section for CSV and image files.
Shows "Click to View" as clickable link for URLs.
"""
if not files_data:
return ""
html = '<div style="margin: 30px 0; page-break-inside: avoid;">'
html += '<h2 style="color: #4a5568; border-bottom: 2px solid #5b7c99; padding-bottom: 8px; margin-bottom: 20px;">Attached Files</h2>'
# CSV Files section
csv_files = files_data.get('csv_files', [])
if csv_files and len(csv_files) > 0:
html += '<h3 style="color: #5b7c99; margin-top: 20px; margin-bottom: 12px;">CSV Files:</h3>'
for csv_file in csv_files:
file_name = csv_file.get('fileName', 'Unknown')
file_path = csv_file.get('filePath', 'Unknown')
# Check if it's a URL
is_url = file_path.startswith('http://') or file_path.startswith('https://')
html += f'''
<div class="file-item">
<div><span class="file-label">File Name:</span> <span class="file-value">{file_name}</span></div>
<div style="margin-top: 6px;"><span class="file-label">File URL:</span></div>
'''
if is_url:
# Show clickable "Click to View" link
html += f'''
<div style="margin-top: 2px;">
<a href="{file_path}" target="_blank" style="color: #5b7c99; text-decoration: underline; font-weight: bold; font-size: 9pt;">Click to View</a>
</div>
'''
else:
# Just show the text
html += f'''
<div class="file-value" style="margin-top: 2px;">{file_path}</div>
'''
html += f'''
<div style="margin-top: 6px;"><span class="file-label">File Type:</span> <span class="file-value">CSV</span></div>
</div>
'''
# Image Files section
image_files = files_data.get('image_files', [])
if image_files and len(image_files) > 0:
html += '<h3 style="color: #5b7c99; margin-top: 20px; margin-bottom: 12px;">Image Files:</h3>'
for img_file in image_files:
file_name = img_file.get('fileName', 'Unknown')
file_path = img_file.get('filePath', 'Unknown')
# Check if it's a URL
is_url = file_path.startswith('http://') or file_path.startswith('https://')
html += f'''
<div class="file-item">
<div><span class="file-label">File Name:</span> <span class="file-value">{file_name}</span></div>
<div style="margin-top: 6px;"><span class="file-label">File URL:</span></div>
'''
if is_url:
# Show clickable "Click to View" link
html += f'''
<div style="margin-top: 2px;">
<a href="{file_path}" target="_blank" style="color: #5b7c99; text-decoration: underline; font-weight: bold; font-size: 9pt;">Click to View</a>
</div>
'''
else:
# Just show the text
html += f'''
<div class="file-value" style="margin-top: 2px;">{file_path}</div>
'''
html += f'''
<div style="margin-top: 6px;"><span class="file-label">File Type:</span> <span class="file-value">Image</span></div>
</div>
'''
html += '</div>'
html += '<hr class="section-divider">'
return html
def _build_message_html(self, message: Dict) -> str:
"""
Build HTML for a single message with inline images.
"""
role = message.get('role', message.get('Role', 'user'))
content = message.get('content', message.get('Content', ''))
# Get timestamp from any available field
timestamp = self._get_timestamp_from_message(message)
# Get image if present
direct_image = message.get('image', message.get('Image', ''))
# Process markdown with images inline
html_content = self._convert_markdown_with_inline_images(content)
# Determine styling
container_class = "message-user" if role == "user" else "message-assistant"
role_class = "role-user" if role == "user" else "role-assistant"
role_display = "User" if role == "user" else "Assistant"
formatted_time = self._format_timestamp(timestamp)
# Build message HTML
html = f"""
<div class="message-container">
<div class="{container_class}">
<div class="message-header">
<span class="{role_class}">{role_display}</span>
<span class="timestamp">{formatted_time}</span>
</div>
<div class="message-content">
{html_content}
</div>
</div>
"""
# Add direct image attachment if present (at the end)
if direct_image:
html += f"""
<div class="figure-container">
<img src="{direct_image}" alt="Attached image">
<div class="figure-caption">Attached Image</div>
</div>
"""
html += "</div>"
return html
def _build_header(self) -> str:
"""Build the report header."""
header_text = self.title
if self.chat_id:
header_text += f" | {self.chat_id}"
return f"""
<div class="header">
{header_text}
</div>
"""
def _build_footer(self) -> str:
"""Build the report footer with page number."""
return """
<div class="footer">
Page <span class="page-number"></span>
</div>
"""
def _build_cover_page(self, message_count: int) -> str:
"""Build a professional cover page."""
current_date = datetime.now().strftime("%B %d, %Y")
cover_html = f"""
<div class="cover-page">
<div class="cover-title">{self.title}</div>
"""
if self.date_range:
cover_html += f'<div class="cover-subtitle">{self.date_range}</div>'
if self.chat_id:
cover_html += f'<div class="cover-meta">Chat ID: {self.chat_id}</div>'
cover_html += f'<div class="cover-meta">Total Messages: {message_count}</div>'
if self.participants and len(self.participants) > 0:
participants_str = ", ".join(self.participants)
cover_html += f'<div class="cover-meta">Participants: {participants_str}</div>'
cover_html += f'<div class="cover-date">Generated on {current_date}</div>'
cover_html += '</div>'
logger.debug("Built cover page")
return cover_html
def _build_stats_summary(self, messages: List[Dict]) -> str:
"""Build a compact statistics summary with horizontal line."""
total = len(messages)
user_count = sum(1 for m in messages if m.get('role', m.get('Role')) == 'user')
assistant_count = sum(1 for m in messages if m.get('role', m.get('Role')) == 'assistant')
# Get date range from messages - check all possible timestamp fields
timestamps = []
for m in messages:
ts = self._get_timestamp_from_message(m)
if ts:
timestamps.append(ts)
if timestamps:
try:
dates = []
for ts in timestamps:
if '+' in ts or ts.endswith('Z'):
dates.append(datetime.fromisoformat(ts.replace('Z', '+00:00')))
else:
dates.append(datetime.fromisoformat(ts))
first_date = min(dates).strftime("%B %d, %Y")
last_date = max(dates).strftime("%B %d, %Y")
date_info = f"{first_date} to {last_date}"
except Exception as e:
logger.warning(f"Failed to parse date range: {e}")
date_info = "Date unavailable"
else:
date_info = "Date unavailable"
logger.info(f"Stats - Total: {total}, User: {user_count}, Assistant: {assistant_count}")
return f"""
<div class="stats-summary">
<p>
<strong>Total Messages:</strong> {total} &nbsp;&nbsp;|&nbsp;&nbsp;
<strong>User:</strong> {user_count} &nbsp;&nbsp;|&nbsp;&nbsp;
<strong>Assistant:</strong> {assistant_count} &nbsp;&nbsp;|&nbsp;&nbsp;
<strong>Period:</strong> {date_info}
</p>
</div>
<hr class="section-divider">
"""
def generate_from_json(self,
json_data: str,
filename: str = None,
cover_page: bool = True,
include_stats: bool = True) -> bool:
"""
Generate PDF from JSON string.
"""
try:
messages = json.loads(json_data)
logger.info("Successfully parsed JSON data")
return self.generate(messages, filename, cover_page, include_stats)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON: {e}")
return False
def generate(self,
messages: List[Dict],
filename: str = None,
cover_page: bool = True,
include_stats: bool = True,
files: Dict = None) -> bool:
"""
Generate the PDF chat log using WeasyPrint.
"""
try:
if not messages or len(messages) == 0:
logger.error("No messages to process")
return False
logger.info(f"Starting PDF generation for {len(messages)} messages")
# Generate unique filename if not provided
if filename is None:
filename = self._generate_unique_filename()
else:
# If filename provided without path, put it in output_dir
if not os.path.dirname(filename):
filename = os.path.join(self.output_dir, filename)
# Build report body
report_body = ""
# Add cover page
if cover_page:
report_body += self._build_cover_page(len(messages))
# Add statistics summary with horizontal line
if include_stats:
report_body += self._build_stats_summary(messages)
# Add files section if provided
if files:
report_body += self._build_files_section(files)
# Add all messages
for i, message in enumerate(messages):
report_body += self._build_message_html(message)
# Add divider between messages (except last)
if i < len(messages) - 1:
report_body += '<div class="message-divider"></div>'
# Assemble full HTML
full_html = f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<style>
{CSS_STYLE}
.page-number:before {{ content: counter(page); }}
</style>
</head>
<body>
{self._build_header()}
{self._build_footer()}
{report_body}
</body>
</html>
"""
# Generate PDF with WeasyPrint
logger.info(f"Writing PDF to: {filename}")
# Create HTML object and render to PDF
html_obj = HTML(string=full_html)
html_obj.write_pdf(filename)
logger.info(f"Success! Chat log generated: {filename}")
logger.info(f"{len(messages)} messages processed")
return True
except Exception as e:
logger.exception(f"Error generating PDF: {e}")
return False