Here is your weekly intelligence briefing...
"
}"""
user_prompt = f"""Generate a Weekly Intelligence Report in HTML format for the week ending {datetime.now().strftime('%B %d, %Y')}.
USER'S DATA SUMMARY:
{data_context}
Create an executive-level weekly summary with trends, insights, and recommendations.
Return ONLY the JSON object with "subject" and "body" (HTML)."""
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
"https://api.groq.com/openai/v1/chat/completions",
headers={
"Authorization": f"Bearer {groq_api_key}",
"Content-Type": "application/json"
},
json={
"model": "llama-3.3-70b-versatile",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.7,
"max_tokens": 2000
}
)
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content'].strip()
try:
email_data = json.loads(content)
subject = email_data.get('subject', f"📈 DataVision Executive Briefing - Week of {datetime.now().strftime('%B %d, %Y')}")
body = email_data.get('body', content)
except Exception:
if content.lower().startswith("subject:"):
lines = content.split("\n", 1)
subject = lines[0].replace("Subject:", "").replace("subject:", "").strip()
body = lines[1].strip() if len(lines) > 1 else content
else:
subject = f"📈 DataVision Executive Briefing - Week of {datetime.now().strftime('%B %d, %Y')}"
body = content
else:
subject = f"📈 DataVision Executive Briefing - {datetime.now().strftime('%B %d, %Y')}"
body = generate_weekly_fallback(data_context)
except Exception as e:
logger.warning(f"AI weekly generation failed: {e}")
subject = f"📈 DataVision Weekly Report - {datetime.now().strftime('%B %d, %Y')}"
body = generate_weekly_fallback(data_context)
else:
subject = f"📈 DataVision Weekly Report - {datetime.now().strftime('%B %d, %Y')}"
body = generate_weekly_fallback(data_context)
# Send the email
await send_insight_email(
to_email=email_to_use,
title=subject,
body=body,
workspace_id=user_id
)
return {
"success": True,
"message": f"🤖 AI-generated Weekly Report sent to {email_to_use}",
"data_context": data_context
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to send weekly report: {e}")
raise HTTPException(500, f"Failed to send weekly report: {str(e)}")
def generate_weekly_fallback(data_context: str) -> str:
"""Fallback weekly email body"""
from datetime import datetime
return f"""
WEEKLY INTELLIGENCE BRIEFING - {datetime.now().strftime('%B %d, %Y')}
DataVision Autonomous Agents have compiled your weekly intelligence briefing regarding your connected datasets.
📊 Weekly Data Overview
{data_context.replace(chr(10), '
')}
🚀 Strategic Recommendations
- Review ML model predictions for emerging patterns across the week
- Consult the Dashboard for updated visualizations and anomalies
- Use the AI Chat to dive deeper into specific weekly metrics
"""
# ============================================================================
# AI EMAIL WRITING AGENT - Generates personalized emails with real data
# ============================================================================
class GenerateEmailRequest(BaseModel):
"""Request body for AI email generation"""
prompt: str # What the email should be about
email_address: str # Where to send
send_immediately: bool = False # If true, send after generation
class GeneratedEmail(BaseModel):
"""Generated email response"""
subject: str
body: str
data_summary: Optional[str] = None
async def gather_user_data_context(user_id: str) -> str:
"""
Gather REAL user data for context in email generation.
Reads actual uploaded CSV files, calculates real statistics,
and fetches ML training results.
"""
context_parts = []
try:
# Use the correct get_user_paths utility (same as rest of app)
from utils.paths import get_user_paths
import pandas as pd
paths = get_user_paths(user_id)
# 1. Get uploaded files and read REAL data
files_dir = paths.get('files')
if files_dir and files_dir.exists():
files = [f for f in files_dir.glob("*") if f.is_file()]
if files:
file_names = [f.name for f in files[:5]]
context_parts.append(f"📁 UPLOADED FILES ({len(files)} total): {', '.join(file_names)}")
# Read CSV/Excel files for real data summary
data_files = [f for f in files if f.suffix.lower() in ['.csv', '.xlsx', '.xls']]
total_rows = 0
total_columns = 0
all_column_stats = []
for data_file in data_files[:3]: # Process up to 3 files
try:
if data_file.suffix.lower() == '.csv':
df = pd.read_csv(data_file)
else:
df = pd.read_excel(data_file)
total_rows += len(df)
total_columns = max(total_columns, len(df.columns))
context_parts.append(f"\n📊 FILE: {data_file.name}")
context_parts.append(f" • Rows: {len(df):,} | Columns: {len(df.columns)}")
context_parts.append(f" • Columns: {', '.join(df.columns.tolist()[:8])}{'...' if len(df.columns) > 8 else ''}")
# Numeric column statistics
numeric_cols = df.select_dtypes(include=['int64', 'float64', 'int32', 'float32']).columns.tolist()
if numeric_cols:
context_parts.append(f" 📈 NUMERIC SUMMARY:")
for col in numeric_cols[:5]: # Top 5 numeric columns
col_mean = df[col].mean()
col_sum = df[col].sum()
col_min = df[col].min()
col_max = df[col].max()
context_parts.append(f" • {col}: Sum={col_sum:,.2f}, Mean={col_mean:,.2f}, Range=[{col_min:,.2f} to {col_max:,.2f}]")
all_column_stats.append({
'column': col,
'sum': col_sum,
'mean': col_mean,
'min': col_min,
'max': col_max
})
# Categorical column breakdown
cat_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()
if cat_cols:
context_parts.append(f" 📋 CATEGORY BREAKDOWN:")
for col in cat_cols[:3]: # Top 3 categories
value_counts = df[col].value_counts().head(5)
top_values = ', '.join([f"{v}: {c}" for v, c in value_counts.items()])
context_parts.append(f" • {col}: {top_values}")
except Exception as e:
logger.warning(f"Error reading {data_file.name}: {e}")
if total_rows > 0:
context_parts.insert(1, f"📈 TOTAL DATA: {total_rows:,} rows across {len(data_files)} file(s)")
# 2. Get ML training results from model storage (where models are actually saved)
from config.settings import Settings
ml_found = False
# Check the correct model storage path: storage/models/{user_id}/active_metadata.json
model_storage_dir = Settings.BASE_DIR / "storage" / "models" / user_id
active_metadata_path = model_storage_dir / "active_metadata.json"
if active_metadata_path.exists():
try:
with open(active_metadata_path, 'r') as f:
ml_data = json.load(f)
if ml_data:
ml_found = True
model_name = ml_data.get('model_name', 'AutoML Model')
task_type = ml_data.get('task_type', 'classification')
target = ml_data.get('target_column', 'unknown')
metrics = ml_data.get('metrics', {})
feature_columns = ml_data.get('feature_columns', [])
training_date = ml_data.get('training_date', '')
version = ml_data.get('version', 1)
context_parts.append(f"\n🤖 ML MODEL TRAINED:")
context_parts.append(f" • Model: {model_name}")
context_parts.append(f" • Task Type: {task_type.upper()}")
context_parts.append(f" • Target Column: {target}")
context_parts.append(f" • Features Used: {len(feature_columns)}")
context_parts.append(f" • Model Version: v{version}")
if training_date:
context_parts.append(f" • Trained On: {training_date}")
# Report metrics based on task type
if task_type == 'classification':
context_parts.append(f" 📊 PERFORMANCE METRICS:")
if 'accuracy' in metrics:
context_parts.append(f" • Accuracy: {metrics['accuracy']:.1%}")
if 'f1' in metrics or 'f1_score' in metrics:
f1 = metrics.get('f1', metrics.get('f1_score', 0))
context_parts.append(f" • F1 Score: {f1:.1%}")
if 'precision' in metrics:
context_parts.append(f" • Precision: {metrics['precision']:.1%}")
if 'recall' in metrics:
context_parts.append(f" • Recall: {metrics['recall']:.1%}")
else: # regression
context_parts.append(f" 📊 PERFORMANCE METRICS:")
if 'r2' in metrics or 'r2_score' in metrics:
r2 = metrics.get('r2', metrics.get('r2_score', 0))
context_parts.append(f" • R² Score: {r2:.3f}")
if 'rmse' in metrics:
context_parts.append(f" • RMSE: {metrics['rmse']:,.2f}")
if 'mae' in metrics:
context_parts.append(f" • MAE: {metrics['mae']:,.2f}")
# Show top features
if feature_columns:
context_parts.append(f" 🔑 TOP FEATURES:")
for idx, feat in enumerate(feature_columns[:5], 1):
context_parts.append(f" {idx}. {feat}")
if len(feature_columns) > 5:
context_parts.append(f" ... and {len(feature_columns) - 5} more features")
except Exception as e:
logger.warning(f"Error reading active model metadata: {e}")
# Fallback: Also check the old automl_results location for backward compatibility
if not ml_found:
automl_results_dir = Settings.BASE_DIR / "storage" / "automl_results" / user_id
if automl_results_dir.exists():
result_files = list(automl_results_dir.glob("*.json"))
if result_files:
# Sort by modification time to get latest
result_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
try:
with open(result_files[0], 'r') as f:
ml_data = json.load(f)
if ml_data:
ml_found = True
best_model = ml_data.get('best_model', {})
task_type = ml_data.get('task_type', 'classification')
target = ml_data.get('target_column', 'unknown')
metrics = best_model.get('metrics', {})
context_parts.append(f"\n🤖 ML MODEL TRAINED:")
context_parts.append(f" • Model: {best_model.get('name', 'AutoML Model')}")
context_parts.append(f" • Task Type: {task_type.upper()}")
context_parts.append(f" • Target Column: {target}")
# Report metrics based on task type
if task_type == 'classification':
acc = metrics.get('accuracy', 0)
f1 = metrics.get('f1_score', metrics.get('f1', 0))
precision = metrics.get('precision', 0)
recall = metrics.get('recall', 0)
context_parts.append(f" 📊 PERFORMANCE METRICS:")
context_parts.append(f" • Accuracy: {acc:.1%}")
if f1: context_parts.append(f" • F1 Score: {f1:.1%}")
if precision: context_parts.append(f" • Precision: {precision:.1%}")
if recall: context_parts.append(f" • Recall: {recall:.1%}")
else: # regression
r2 = metrics.get('r2', metrics.get('r2_score', 0))
rmse = metrics.get('rmse', 0)
mae = metrics.get('mae', 0)
context_parts.append(f" 📊 PERFORMANCE METRICS:")
context_parts.append(f" • R² Score: {r2:.3f}")
if rmse: context_parts.append(f" • RMSE: {rmse:,.2f}")
if mae: context_parts.append(f" • MAE: {mae:,.2f}")
# Feature importance
features = ml_data.get('feature_importance', [])[:5]
if features:
context_parts.append(f" 🔑 TOP FEATURES (most predictive):")
for idx, feat in enumerate(features, 1):
if isinstance(feat, dict):
fname = feat.get('feature', feat.get('name', str(feat)))
importance = feat.get('importance', 0)
context_parts.append(f" {idx}. {fname} (importance: {importance:.3f})")
else:
context_parts.append(f" {idx}. {feat}")
# Try to get sample predictions from stored model
try:
from ml.automl_engine import ProductionMLEngine
engine = ProductionMLEngine()
if engine.load(user_id):
# Read some sample data and make predictions
sample_preds = []
for data_file in data_files[:1]:
try:
sample_df = pd.read_csv(data_file) if data_file.suffix.lower() == '.csv' else pd.read_excel(data_file)
for row_idx, row in sample_df.head(3).iterrows():
try:
row_dict = row.to_dict()
X_sample = engine._preprocess_single(row_dict)
pred = engine.model.predict([X_sample])[0]
if hasattr(engine, 'target_encoder') and engine.target_encoder:
try:
pred_label = engine.target_encoder.inverse_transform([int(pred)])[0]
except:
pred_label = str(pred)
else:
pred_label = str(pred)
sample_preds.append(pred_label)
except:
continue
except:
continue
if sample_preds:
context_parts.append(f" 🔮 SAMPLE PREDICTIONS:")
for idx, pred in enumerate(sample_preds[:3], 1):
context_parts.append(f" {idx}. Predicted {target}: {pred}")
except Exception as pred_err:
logger.debug(f"Could not generate sample predictions: {pred_err}")
except Exception as e:
logger.warning(f"Error reading ML results: {e}")
# 3. Check for unified analytics cache
analytics_cache_dir = Settings.BASE_DIR / "storage" / "analytics_cache"
if analytics_cache_dir.exists():
cache_files = list(analytics_cache_dir.glob(f"{user_id}*.json"))
if cache_files:
try:
with open(cache_files[-1], 'r') as f:
analytics = json.load(f)
kpis = analytics.get('overviewLayout', {}).get('kpis', [])
if kpis:
context_parts.append(f"\n📈 KEY PERFORMANCE INDICATORS:")
for kpi in kpis[:4]:
title = kpi.get('title', 'Metric')
value = kpi.get('value', 0)
delta = kpi.get('delta')
delta_str = f" ({'+' if delta > 0 else ''}{delta:.1f}%)" if delta else ""
context_parts.append(f" • {title}: {value:,}{delta_str}")
except Exception as e:
logger.warning(f"Error reading analytics cache: {e}")
if not ml_found:
context_parts.append(f"\n💡 TIP: No ML model trained yet. Consider training a model for predictions!")
except Exception as e:
logger.error(f"Error gathering user data context: {e}")
import traceback
traceback.print_exc()
context_parts.append("(Limited data available due to error)")
return "\n".join(context_parts) if context_parts else "No data uploaded yet."
@router.post("/email-prefs/generate-email")
async def generate_ai_email(
request: GenerateEmailRequest,
x_user_id: str = Header(None, alias="X-User-ID")
):
"""
🤖 AI EMAIL WRITING AGENT
Generates personalized emails using LLM based on:
- User's uploaded data
- ML training results
- Dashboard insights
- Custom prompt
"""
import os
import httpx
from services.email_service import send_insight_email
user_id = x_user_id or "default_user"
if not request.email_address:
raise HTTPException(400, "Email address is required")
if not request.prompt or len(request.prompt.strip()) < 5:
raise HTTPException(400, "Please provide a description of what the email should contain")
try:
# 1. Gather real user data context
print(f"🔍 Gathering data context for user: {user_id}")
data_context = await gather_user_data_context(user_id)
print(f"📊 Data context:\n{data_context}")
# 2. Build LLM prompt
system_prompt = """You are DataVision AI, a professional business email writer.
Generate professional, data-driven emails based on user data and requirements.
Your emails should:
- Be clear, concise, and professional
- Include specific data insights when available
- Use a friendly but business-appropriate tone
- Format with proper paragraphs and bullet points where appropriate
- Always sign off as "DataVision AI Analytics"
Output format: Return ONLY valid JSON with exactly these fields:
{"subject": "Email subject line", "body": "Full email body text"}"""
user_prompt = f"""Generate a professional email based on this request:
USER REQUEST: {request.prompt}
AVAILABLE DATA CONTEXT:
{data_context}
Generate a personalized email that incorporates the available data insights.
If specific data is mentioned in the context, reference it naturally in the email.
Remember to return ONLY valid JSON with "subject" and "body" fields."""
# 3. Call Groq LLM
groq_api_key = os.getenv("GROQ_API_KEY")
if not groq_api_key:
raise HTTPException(500, "LLM service not configured. Add GROQ_API_KEY to enable AI email generation.")
print(f"🤖 Calling Groq LLM for email generation...")
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
"https://api.groq.com/openai/v1/chat/completions",
headers={
"Authorization": f"Bearer {groq_api_key}",
"Content-Type": "application/json"
},
json={
"model": "llama-3.3-70b-versatile",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.7,
"max_tokens": 1024
}
)
if response.status_code != 200:
logger.error(f"Groq API error: {response.text}")
raise HTTPException(500, f"LLM service error: {response.status_code}")
result = response.json()
content = result['choices'][0]['message']['content']
# Parse the JSON response
try:
# Clean up the response (remove markdown code blocks if present)
content = content.strip()
if content.startswith("```"):
content = content.split("```")[1]
if content.startswith("json"):
content = content[4:]
content = content.strip()
email_data = json.loads(content)
subject = email_data.get('subject', 'DataVision Insights')
body = email_data.get('body', content)
except json.JSONDecodeError:
# Fallback: use raw content
subject = f"DataVision: {request.prompt[:50]}..."
body = content
print(f"✅ Email generated successfully!")
# 4. Send immediately if requested
if request.send_immediately:
print(f"📧 Sending email to: {request.email_address}")
await send_insight_email(
to_email=request.email_address,
title=subject,
body=body,
workspace_id=user_id
)
return {
"success": True,
"sent": True,
"message": f"AI-generated email sent to {request.email_address}",
"email": {
"subject": subject,
"body": body
},
"data_context_used": data_context
}
# Return generated email without sending
return {
"success": True,
"sent": False,
"message": "Email generated successfully. Review and click send when ready.",
"email": {
"subject": subject,
"body": body
},
"data_context_used": data_context
}
except HTTPException:
raise
except Exception as e:
logger.error(f"AI email generation failed: {e}")
raise HTTPException(500, f"Failed to generate email: {str(e)}")
@router.post("/email-prefs/send-custom-email")
async def send_custom_email(
request: dict,
x_user_id: str = Header(None, alias="X-User-ID")
):
"""Send a custom email (after user reviews AI-generated content)"""
from services.email_service import send_insight_email
user_id = x_user_id or "default_user"
email_address = request.get('email_address')
subject = request.get('subject')
body = request.get('body')
if not email_address or not subject or not body:
raise HTTPException(400, "Missing email_address, subject, or body")
try:
await send_insight_email(
to_email=email_address,
title=subject,
body=body,
workspace_id=user_id
)
return {
"success": True,
"message": f"Email sent successfully to {email_address}"
}
except Exception as e:
raise HTTPException(500, f"Failed to send email: {str(e)}")
class PasswordResetRequest(BaseModel):
"""Password reset request model"""
email: str
@router.post("/auth/request-password-reset")
async def request_password_reset(request: PasswordResetRequest):
"""
🔐 Custom Password Reset Flow
Sends a branded password reset email using our email service.
Sends a branded password reset email using our email service.
Flow:
1. Generate a secure reset token
2. Build a branded reset link
3. Send via our Resend email service with DataVision branding
"""
import secrets
import hashlib
from datetime import datetime, timedelta
from services.email_service import send_password_reset_email
email = request.email.strip().lower()
if not email:
raise HTTPException(400, "Email address is required")
try:
# Generate a secure token
token = secrets.token_urlsafe(32)
token_hash = hashlib.sha256(token.encode()).hexdigest()
# Store the token with expiry (1 hour)
token_data = {
"email": email,
"token_hash": token_hash,
"created_at": datetime.utcnow().isoformat(),
"expires_at": (datetime.utcnow() + timedelta(hours=1)).isoformat(),
"used": False
}
# Save token to storage
tokens_dir = PREFS_DIR / "password_reset_tokens"
tokens_dir.mkdir(parents=True, exist_ok=True)
token_file = tokens_dir / f"{token_hash[:16]}.json"
with open(token_file, 'w') as f:
json.dump(token_data, f)
# Build reset link - points to our frontend update-password page
# The frontend will validate the token and call backend to update password
from urllib.parse import quote
APP_URL = os.getenv("APP_URL", "https://killerkumar-ai-business-analyst.hf.space")
# URL-encode the email to handle special characters like @ and +
encoded_email = quote(email, safe='')
reset_link = f"{APP_URL}/auth/update-password?token={token}&email={encoded_email}"
logger.info(f"📧 Generated reset link for {email}")
# Send branded email
await send_password_reset_email(
to_email=email,
reset_link=reset_link
)
logger.info(f"✅ Password reset email sent to {email}")
return {
"success": True,
"message": "If an account exists with this email, you will receive a password reset link shortly."
}
except Exception as e:
logger.error(f"Password reset request failed: {e}")
# Don't reveal if email exists or not for security
return {
"success": True,
"message": "If an account exists with this email, you will receive a password reset link shortly."
}
class PasswordUpdateRequest(BaseModel):
"""Password update request model"""
token: str
email: str
new_password: str
@router.post("/auth/update-password-with-token")
async def update_password_with_token(
request: PasswordUpdateRequest,
db=Depends(get_db)
):
"""
🔐 Update password using the reset token
Validates the token and updates the password via the auth system
"""
import hashlib
from datetime import datetime
token = request.token
email = request.email.strip().lower()
new_password = request.new_password
if not token or not email or not new_password:
raise HTTPException(400, "Token, email, and new_password are required")
if len(new_password) < 6:
raise HTTPException(400, "Password must be at least 6 characters")
try:
# Verify token
token_hash = hashlib.sha256(token.encode()).hexdigest()
tokens_dir = PREFS_DIR / "password_reset_tokens"
token_file = tokens_dir / f"{token_hash[:16]}.json"
if not token_file.exists():
raise HTTPException(400, "Invalid or expired reset token")
with open(token_file, 'r') as f:
token_data = json.load(f)
# Check if token matches email
if token_data.get("email") != email:
raise HTTPException(400, "Invalid reset token")
# Check if already used
if token_data.get("used"):
raise HTTPException(400, "This reset token has already been used")
# Check expiry
expires_at = datetime.fromisoformat(token_data["expires_at"])
if datetime.utcnow() > expires_at:
raise HTTPException(400, "Reset token has expired. Please request a new one.")
# Update password via SQLAlchemy
from sqlalchemy.future import select
from database.orm import UserProfile
from core.auth import get_password_hash
stmt = select(UserProfile).where(UserProfile.email == email)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(400, "No account found with this email")
user.hashed_password = get_password_hash(new_password)
await db.commit()
# Mark token as used
token_data["used"] = True
with open(token_file, 'w') as f:
json.dump(token_data, f)
logger.info(f"✅ Password updated successfully for {email}")
return {
"success": True,
"message": "Password updated successfully! You can now sign in with your new password."
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Password update failed: {e}")
raise HTTPException(500, f"Failed to update password: {str(e)}")