|
|
from typing import Dict, List, Any, Optional, Union
|
|
|
import pandas as pd
|
|
|
import smtplib
|
|
|
from email.mime.multipart import MIMEMultipart
|
|
|
from email.mime.text import MIMEText
|
|
|
from email.mime.image import MIMEImage
|
|
|
import base64
|
|
|
import io
|
|
|
from pathlib import Path
|
|
|
import json
|
|
|
import datetime
|
|
|
from llama_index.tools import FunctionTool
|
|
|
|
|
|
class ExportTools:
|
|
|
"""Tools for exporting data, generating reports, and sending emails."""
|
|
|
|
|
|
def __init__(self, output_directory: str = "./exports"):
|
|
|
"""Initialize with directory for saved exports."""
|
|
|
self.output_directory = Path(output_directory)
|
|
|
self.output_directory.mkdir(exist_ok=True, parents=True)
|
|
|
self.tools = self._create_tools()
|
|
|
|
|
|
def _create_tools(self) -> List[FunctionTool]:
|
|
|
"""Create LlamaIndex function tools for export operations."""
|
|
|
tools = [
|
|
|
FunctionTool.from_defaults(
|
|
|
name="generate_report",
|
|
|
description="Generate a report from conversation and results",
|
|
|
fn=self.generate_report
|
|
|
),
|
|
|
FunctionTool.from_defaults(
|
|
|
name="save_results_to_csv",
|
|
|
description="Save query results to a CSV file",
|
|
|
fn=self.save_results_to_csv
|
|
|
),
|
|
|
FunctionTool.from_defaults(
|
|
|
name="send_email",
|
|
|
description="Send results via email",
|
|
|
fn=self.send_email
|
|
|
)
|
|
|
]
|
|
|
return tools
|
|
|
|
|
|
def get_tools(self) -> List[FunctionTool]:
|
|
|
"""Get all available export tools."""
|
|
|
return self.tools
|
|
|
|
|
|
def generate_report(self, title: str, content: str,
|
|
|
images: List[str] = None) -> Dict[str, Any]:
|
|
|
"""Generate HTML report from content and images."""
|
|
|
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
filename = f"{title.replace(' ', '_')}_{timestamp}.html"
|
|
|
file_path = self.output_directory / filename
|
|
|
|
|
|
|
|
|
html = f"""
|
|
|
<!DOCTYPE html>
|
|
|
<html>
|
|
|
<head>
|
|
|
<title>{title}</title>
|
|
|
<style>
|
|
|
body {{ font-family: Arial, sans-serif; margin: 40px; line-height: 1.6; }}
|
|
|
h1 {{ color: #333366; }}
|
|
|
.report-container {{ max-width: 800px; margin: 0 auto; }}
|
|
|
.timestamp {{ color: #666; font-size: 0.8em; }}
|
|
|
img {{ max-width: 100%; height: auto; margin: 20px 0; }}
|
|
|
</style>
|
|
|
</head>
|
|
|
<body>
|
|
|
<div class="report-container">
|
|
|
<h1>{title}</h1>
|
|
|
<div class="timestamp">Generated on: {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</div>
|
|
|
<div class="content">
|
|
|
{content.replace('\n', '<br>')}
|
|
|
</div>
|
|
|
"""
|
|
|
|
|
|
|
|
|
if images and len(images) > 0:
|
|
|
html += "<div class='images'>"
|
|
|
for i, img_base64 in enumerate(images):
|
|
|
html += f"<img src='data:image/png;base64,{img_base64}' alt='Figure {i+1}'>"
|
|
|
html += "</div>"
|
|
|
|
|
|
html += """
|
|
|
</div>
|
|
|
</body>
|
|
|
</html>
|
|
|
"""
|
|
|
|
|
|
|
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
|
f.write(html)
|
|
|
|
|
|
return {
|
|
|
"success": True,
|
|
|
"report_path": str(file_path),
|
|
|
"title": title,
|
|
|
"timestamp": timestamp
|
|
|
}
|
|
|
|
|
|
def save_results_to_csv(self, data: List[Dict[str, Any]],
|
|
|
filename: str = None) -> Dict[str, Any]:
|
|
|
"""Save query results to a CSV file."""
|
|
|
if not data or len(data) == 0:
|
|
|
return {"success": False, "error": "No data provided"}
|
|
|
|
|
|
|
|
|
df = pd.DataFrame(data)
|
|
|
|
|
|
|
|
|
if not filename:
|
|
|
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
filename = f"query_results_{timestamp}.csv"
|
|
|
|
|
|
|
|
|
if not filename.lower().endswith('.csv'):
|
|
|
filename += '.csv'
|
|
|
|
|
|
file_path = self.output_directory / filename
|
|
|
|
|
|
|
|
|
df.to_csv(file_path, index=False)
|
|
|
|
|
|
return {
|
|
|
"success": True,
|
|
|
"file_path": str(file_path),
|
|
|
"row_count": len(df),
|
|
|
"column_count": len(df.columns)
|
|
|
}
|
|
|
|
|
|
def send_email(self, to_email: str, subject: str, body: str,
|
|
|
from_email: str = None, smtp_server: str = None,
|
|
|
smtp_port: int = 587, username: str = None,
|
|
|
password: str = None, images: List[str] = None) -> Dict[str, Any]:
|
|
|
"""
|
|
|
Send email with results.
|
|
|
Note: In production, credentials should be securely managed.
|
|
|
For demo purposes, this will log the email content instead.
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
email_content = {
|
|
|
"to": to_email,
|
|
|
"subject": subject,
|
|
|
"body": body[:100] + "..." if len(body) > 100 else body,
|
|
|
"images": f"{len(images) if images else 0} images would be attached",
|
|
|
"note": "Email sending is simulated for demo purposes"
|
|
|
}
|
|
|
|
|
|
|
|
|
print(f"SIMULATED EMAIL: {json.dumps(email_content, indent=2)}")
|
|
|
|
|
|
return {
|
|
|
"success": True,
|
|
|
"to": to_email,
|
|
|
"subject": subject,
|
|
|
"simulated": True,
|
|
|
"timestamp": datetime.datetime.now().isoformat()
|
|
|
}
|
|
|
|
