import os
import io
import time
import json
import random
import tempfile
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
import gradio as gr
try:
import openai
OPENAI_AVAILABLE = True
except Exception:
OPENAI_AVAILABLE = False
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
if OPENAI_AVAILABLE and OPENAI_API_KEY:
openai.api_key = OPENAI_API_KEY
def generate_sample_csv(n=50):
base_date = datetime.now() + timedelta(days=7)
rows = []
interests = ["Data Engineering", "Machine Learning", "Web Dev", "Cloud", "Product"]
channels = ["Organic", "Ads", "Referral", "Partner", "Email Campaign"]
for i in range(1, n + 1):
name = f"User{i:03d}"
email = f"user{i:03d}@example.com"
phone = f"+91100000{i:04d}"[-13:]
interest = random.choice(interests)
channel = random.choice(channels)
registered_at = (datetime.now() - timedelta(days=random.randint(0, 14))).strftime('%Y-%m-%d')
rows.append({
"id": i,
"name": name,
"email": email,
"phone": phone,
"interest": interest,
"channel": channel,
"registered_at": registered_at
})
df = pd.DataFrame(rows)
tmpfile = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
df.to_csv(tmpfile.name, index=False)
tmpfile.close()
return tmpfile.name
def load_leads_from_file(fileobj):
if fileobj is None:
return pd.DataFrame()
try:
df = pd.read_csv(fileobj)
except Exception:
fileobj.seek(0)
df = pd.read_csv(io.StringIO(fileobj.read().decode('utf-8')))
for col in ["name", "email", "phone", "interest", "registered_at"]:
if col not in df.columns:
df[col] = ""
return df
def estimate_attendance_prob(df, event_date=None):
df = df.copy()
if event_date is None:
event_date = datetime.now() + timedelta(days=7)
else:
if isinstance(event_date, str):
event_date = datetime.fromisoformat(event_date)
def channel_score(ch):
ch = str(ch).lower()
if 'ref' in ch: return 0.12
if 'organic' in ch: return 0.08
if 'partner' in ch: return 0.10
if 'email' in ch: return 0.06
return 0.03
interest_map = {
'Data Engineering': 0.12,
'Machine Learning': 0.11,
'Web Dev': 0.07,
'Cloud': 0.09,
'Product': 0.06
}
probs = []
for _, r in df.iterrows():
try:
reg = datetime.fromisoformat(str(r.get('registered_at')))
except Exception:
reg = datetime.now()
days_until = (event_date - reg).days
if days_until <= 0:
base = 0.25
elif days_until <= 2:
base = 0.22
elif days_until <= 7:
base = 0.18
else:
base = 0.12
ch_boost = channel_score(r.get('channel', ''))
interest_boost = interest_map.get(r.get('interest'), 0.05)
noise = random.uniform(-0.03, 0.03)
p = base + ch_boost + interest_boost + noise
p = max(0.01, min(0.95, p))
probs.append(round(p, 3))
df['predicted_prob'] = probs
return df
def generate_personalized_message(row, event_name="Scaler Live: Roadmap to Data Engineering", event_date=None, use_openai=False):
if event_date is None:
event_date = (datetime.now() + timedelta(days=7)).strftime('%b %d, %Y')
name = row.get('name', 'there')
interest = row.get('interest', '')
prob = row.get('predicted_prob', None)
if use_openai and OPENAI_AVAILABLE and OPENAI_API_KEY:
prompt = f"Write a short personalized reminder message (1-2 sentences) for {name} who is interested in {interest} to attend the online event '{event_name}' on {event_date}. Make it friendly and include a single call-to-action 'Join here: '."
try:
resp = openai.Completion.create(
engine='text-davinci-003',
prompt=prompt,
max_tokens=80,
temperature=0.7,
n=1
)
return resp.choices[0].text.strip()
except Exception:
pass
urgency = "Don't miss out!" if (prob is not None and prob < 0.25) else "Can't wait to see you there!"
return f"Hi {name},\nWe have a short live session '{event_name}' on {event_date} that covers {interest} topics you care about. {urgency} Join here: "
def batch_generate_messages(df, event_name, event_date, use_openai=False):
df = df.copy()
messages = []
for _, r in df.iterrows():
msg = generate_personalized_message(r, event_name=event_name, event_date=event_date, use_openai=use_openai)
messages.append(msg)
df['message'] = messages
return df
def simulate_send_campaign(df, channel='email'):
sent_rows = []
opens = 0
clicks = 0
for _, r in df.iterrows():
p = float(r.get('predicted_prob', 0.1))
open_prob = min(0.9, 0.2 + p * 0.6)
click_prob = min(0.8, 0.05 + p * 0.5)
opened = random.random() < open_prob
clicked = opened and (random.random() < click_prob)
sent_rows.append({
'id': r.get('id'),
'name': r.get('name'),
'email': r.get('email'),
'phone': r.get('phone'),
'predicted_prob': r.get('predicted_prob'),
'opened': opened,
'clicked': clicked,
'message': r.get('message')
})
opens += int(opened)
clicks += int(clicked)
sent_df = pd.DataFrame(sent_rows)
stats = {
'total_sent': len(sent_df),
'opens': int(opens),
'clicks': int(clicks),
'open_rate': round(opens / max(1, len(sent_df)), 3),
'click_rate': round(clicks / max(1, len(sent_df)), 3)
}
out_path = '/mnt/data/simulated_sent_log.csv'
try:
sent_df.to_csv(out_path, index=False)
except Exception:
pass
return stats, sent_df
def ui_generate_sample_csv(n=50):
return generate_sample_csv(n)
def ui_load_and_preview(file):
if file is None:
return pd.DataFrame()
df = load_leads_from_file(file)
df = estimate_attendance_prob(df)
return df
def ui_estimate_and_generate(df, event_name, event_date, use_openai=False):
if isinstance(df, str):
df = pd.read_csv(io.StringIO(df))
if df is None or len(df) == 0:
return pd.DataFrame(), "No leads provided"
df = estimate_attendance_prob(df, event_date=event_date)
df = batch_generate_messages(df, event_name, event_date, use_openai=use_openai)
return df, f"Generated messages and probabilities for {len(df)} leads"
def ui_simulate_send(df):
if isinstance(df, str):
df = pd.read_csv(io.StringIO(df))
stats, sent_df = simulate_send_campaign(df)
return stats, sent_df
def ui_export_csv(df):
if isinstance(df, str):
df = pd.read_csv(io.StringIO(df))
tmpfile = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
df.to_csv(tmpfile.name, index=False)
tmpfile.close()
return tmpfile.name
with gr.Blocks(title="AI Conversion Automator - Scaler APM MVP") as demo:
gr.Markdown("# AI Conversion Automator — Increase joining % for free live class")
with gr.Tab("1. Sample CSV"):
with gr.Row():
n_samples = gr.Slider(minimum=10, maximum=500, value=50, label="Number of sample leads")
gen_btn = gr.Button("Generate sample CSV")
sample_download = gr.File()
gen_btn.click(fn=ui_generate_sample_csv, inputs=[n_samples], outputs=[sample_download])
with gr.Tab("2. Upload & Preview Leads"):
uploader = gr.File(label="Upload leads CSV (columns: id,name,email,phone,interest,channel,registered_at)")
preview = gr.Dataframe(headers=None, row_count=10)
load_btn = gr.Button("Load & Preview")
load_btn.click(fn=ui_load_and_preview, inputs=[uploader], outputs=[preview])
with gr.Tab("3. Generate Messages & Predictions"):
event_name = gr.Textbox(label="Event name", value="Roadmap to Data Engineering — Live Class")
event_date = gr.Textbox(label="Event date (YYYY-MM-DD)", value=(datetime.now() + timedelta(days=7)).strftime('%Y-%m-%d'))
use_openai = gr.Checkbox(label="Use OpenAI for message generation (requires OPENAI_API_KEY)", value=False)
generate_btn = gr.Button("Generate Messages & Probabilities")
generated_table = gr.Dataframe(headers=None, row_count=20)
status_txt = gr.Textbox(label="Status")
generate_btn.click(fn=ui_estimate_and_generate, inputs=[preview, event_name, event_date, use_openai], outputs=[generated_table, status_txt])
with gr.Tab("4. Simulate Campaign"):
simulate_btn = gr.Button("Simulate Send")
sim_stats = gr.JSON()
sim_table = gr.Dataframe(headers=None, row_count=20)
simulate_btn.click(fn=ui_simulate_send, inputs=[generated_table], outputs=[sim_stats, sim_table])
with gr.Tab("5. Export"):
export_btn = gr.Button("Export CSV")
download_file = gr.File()
export_btn.click(fn=ui_export_csv, inputs=[generated_table], outputs=[download_file])
if __name__ == '__main__':
demo.launch()