# -*- coding: utf-8 -*- import os import re import urllib.parse import time import random import gradio as gr import requests from dotenv import load_dotenv from openai import OpenAI from quiz_data import QUESTIONS, PROFILES from utils import ( start_state, compute_winning_profile, build_profile_summary_prompt, ) # ------------------------------------------------------ # Config # ------------------------------------------------------ load_dotenv() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) WELCOME_TITLE = "CX Management Maturity Navigator" WELCOME_INTRO = ( "Where Does Your CX Stand? Find Out in 5 Minutes.\n" "Discover your CX Management Maturity level – and your next move." ) PARDOT_FORM_HANDLER_URL = "http://go.demo.pardot.com/l/718473/2026-02-27/3mwk" # Fallback responses (used only if LLM call fails) BOT_ACKS_FALLBACK = [ "That's really helpful context, thank you.", "Interesting — that tells me a lot about where you're at.", "Got it, that makes sense given what you're working with.", "Thanks for being so candid about that.", "That's a common challenge at this stage, actually.", "Good to know — it shapes the picture quite a bit.", ] # ------------------------------------------------------ # Helpers # ------------------------------------------------------ def welcome_messages(): return [ {"role": "assistant", "content": f"Hi there! 👋\n\n{WELCOME_INTRO}"}, {"role": "assistant", "content": QUESTIONS[0]["text"]}, ] def get_current_options(state): idx = state["current_question_index"] if idx >= len(QUESTIONS): return [] return [opt[0] for opt in QUESTIONS[idx]["options"]] def generate_gpt_summary(profile_key, answers): prompt = build_profile_summary_prompt(profile_key, answers) try: completion = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "You are a concise, professional Merkle CX expert."}, {"role": "user", "content": prompt}, ], temperature=0.6, ) return completion.choices[0].message.content.strip() except Exception as e: print("Error calling OpenAI:", e) return "Your answers show a clear vision for your CX journey. Let's look at the details below." def generate_ack(question_text: str, user_answer: str) -> str: """ Uses the LLM to generate a short, natural, contextual acknowledgment based on the question and the user's answer. Falls back to a static response if the API call fails. """ try: completion = client.chat.completions.create( model="gpt-4o-mini", messages=[ { "role": "system", "content": ( "You are a warm, sharp CX consultant at Merkle having a real conversation. " "The user just answered a maturity assessment question. " "React briefly and naturally to their specific answer — 1 to 2 sentences max. " "Be conversational, empathetic and human. Show you actually understood what they said. " "Vary your tone and openings. Never start with 'Great!' or 'Absolutely!'. " "No bullet points. Minimal emojis. Don't parrot the answer back verbatim." ) }, { "role": "user", "content": f"Question: {question_text}\nTheir answer: {user_answer}" } ], temperature=0.85, max_tokens=60, ) return completion.choices[0].message.content.strip() except Exception as e: print("Error generating ack:", e) return random.choice(BOT_ACKS_FALLBACK) def format_answers_for_pardot(answers): parts = [f"{a['id']}: {a['answer']}" for a in answers] return " | ".join(parts) def send_to_pardot_backend(state): if not state.get("final_profile_key"): return profile = PROFILES[state["final_profile_key"]] payload = { "email": state["email"], "firstname": state["first_name"], "Lastname": state["last_name"], "Company": state["company"], "Jobtitle": state["job_title"], "agentic_ai_personality": profile["label"], "agentic_ai_profile_key": profile["key"], "agentic_ai_quiz_answers": format_answers_for_pardot(state["answers"]), } try: resp = requests.post(PARDOT_FORM_HANDLER_URL, data=payload, timeout=5) print("Pardot response:", resp.status_code) except Exception as e: print("Error sending to Pardot:", e) def is_professional_email(email: str) -> bool: email = email.strip().lower() if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email): return False personal_domains = { "gmail.com", "yahoo.com", "yahoo.fr", "hotmail.com", "outlook.com", "live.com", "icloud.com", "me.com", "wanadoo.fr", "orange.fr" } domain = email.split("@", 1)[1] return domain not in personal_domains # ------------------------------------------------------ # Logic: Quiz avec effet Typewriter # ------------------------------------------------------ def chatbot_step(selected_option, state, history): if state is None: state = start_state() history = welcome_messages() yield history, state, gr.update(choices=get_current_options(state), value=None, visible=True), gr.update(visible=False) return if not selected_option: yield history, state, gr.update(), gr.update() return idx = state["current_question_index"] q = QUESTIONS[idx] profile_key = None for opt_text, prof in q["options"]: if opt_text == selected_option: profile_key = prof break state["answers"].append({ "id": q["id"], "question": q["text"], "answer": selected_option, "profile": profile_key, }) if profile_key != "none": state["profile_counts"][profile_key] += 1 # 1. Affiche la réponse de l'utilisateur history.append({"role": "user", "content": selected_option}) yield history, state, gr.update(value=None, interactive=False), gr.update(visible=False) # 2. BULLE 1 — Acquiescement LLM (typewriter) ack_msg = generate_ack(q["text"], selected_option) history.append({"role": "assistant", "content": ""}) for i in range(0, len(ack_msg), 2): history[-1]["content"] += ack_msg[i:i+2] time.sleep(0.01) yield history, state, gr.update(value=None, interactive=False), gr.update(visible=False) # Yield explicite pour que Gradio commite la bulle avant d'en ouvrir une nouvelle yield history, state, gr.update(value=None, interactive=False), gr.update(visible=False) time.sleep(0.4) state["current_question_index"] += 1 # 3. BULLE 2 — Question suivante, texte brut sans numérotation if state["current_question_index"] < len(QUESTIONS): next_q = QUESTIONS[state["current_question_index"]] history.append({"role": "assistant", "content": ""}) # nouvelle bulle for i in range(0, len(next_q["text"]), 2): history[-1]["content"] += next_q["text"][i:i+2] time.sleep(0.01) yield history, state, gr.update(value=None, interactive=False), gr.update(visible=False) yield history, state, gr.update(choices=[opt[0] for opt in next_q["options"]], value=None, interactive=True), gr.update(visible=False) else: state["phase"] = "lead_gen" transition_msg = "That's everything I need! Based on your answers, I've built your CX Maturity Profile. Fill in your details below and I'll reveal your results." history.append({"role": "assistant", "content": ""}) # nouvelle bulle for i in range(0, len(transition_msg), 2): history[-1]["content"] += transition_msg[i:i+2] time.sleep(0.01) yield history, state, gr.update(visible=False), gr.update(visible=False) yield history, state, gr.update(visible=False), gr.update(visible=False) time.sleep(0.3) yield history, state, gr.update(visible=False), gr.update(visible=True) # ------------------------------------------------------ # Logic: Lead Gen Form Submission (Typewriter) # ------------------------------------------------------ def handle_form_submit(fname, lname, email, company, job, consent, state, history): if not fname or not lname or not email or not company: history.append({"role": "assistant", "content": "⚠️ Please fill in all required fields (First Name, Last Name, Email, Company)."}) yield history, state, gr.update(visible=True) return if not consent: history.append({"role": "assistant", "content": "⚠️ Please agree to the data usage policy to see your results."}) yield history, state, gr.update(visible=True) return if not is_professional_email(email): history.append({"role": "assistant", "content": "⚠️ Please enter a valid professional email address (no Gmail, Outlook, etc.)."}) yield history, state, gr.update(visible=True) return state["first_name"] = fname state["last_name"] = lname state["email"] = email state["company"] = company state["job_title"] = job history.append({"role": "user", "content": "My details are submitted!"}) yield history, state, gr.update(visible=False) winning_key = compute_winning_profile(state["profile_counts"]) state["final_profile_key"] = winning_key profile = PROFILES[winning_key] summary = generate_gpt_summary(winning_key, state["answers"]) send_to_pardot_backend(state) subject = f"I just took the CX Maturity Quiz – I'm a {profile['label']}" body = f"Hi,\n\n{profile['email_body_template']}\n\n{state['first_name']} {state['last_name']} | {state['company']} | {state['job_title']}" mailto_link = f"mailto:enquiries-dach@merkle.com?subject={urllib.parse.quote(subject)}&body={urllib.parse.quote(body)}" cta_html = f'
✉️ {profile["cta_text"]}
' combined = ( f"**Your Profile: {profile['label']}**\n\n" f"{summary}\n\n" f"**{profile['result_title']}**\n\n" f"{profile['result_body']}\n\n" f"Feel free to get in contact to schedule an appointment to discuss your specific challenges and provide further inspiration.\n\n" f"{cta_html}\n\n" "*(I've also logged your result for the Merkle team so they can prepare tailored ideas for you!)*" ) history.append({"role": "assistant", "content": ""}) for i in range(0, len(combined), 5): history[-1]["content"] += combined[i:i+5] time.sleep(0.01) yield history, state, gr.update(visible=False) state["phase"] = "done" yield history, state, gr.update(visible=False) # ------------------------------------------------------ # UI # ------------------------------------------------------ def build_interface(): with gr.Blocks(fill_height=True) as demo: gr.Markdown(f"### {WELCOME_TITLE}") state = gr.State(start_state()) chatbot = gr.Chatbot( value=welcome_messages(), height=500, label="CX Guide", show_label=False ) options = gr.Radio( choices=[opt[0] for opt in QUESTIONS[0]["options"]], label="Choose your answer:", interactive=True, ) with gr.Group(visible=False) as lead_form: gr.Markdown("#### Where should we send your full maturity report?") with gr.Row(): fname = gr.Textbox(label="First Name *") lname = gr.Textbox(label="Last Name *") with gr.Row(): email = gr.Textbox(label="Business Email *") company = gr.Textbox(label="Company *") job = gr.Textbox(label="Job Title") consent = gr.Checkbox( label="I agree that my data may be used by Merkle to follow up on this quiz. I can revoke my consent at any time.", value=False ) submit_btn = gr.Button("See My Results 🚀", variant="primary") options.change( fn=chatbot_step, inputs=[options, state, chatbot], outputs=[chatbot, state, options, lead_form], ) submit_btn.click( fn=handle_form_submit, inputs=[fname, lname, email, company, job, consent, state, chatbot], outputs=[chatbot, state, lead_form], ) return demo demo = build_interface() if __name__ == "__main__": demo.launch(ssr_mode=False)