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()