"""QuitSure Chatbot Onboarding — MVP Server with LLM Integration""" import json import os import re import time from datetime import date, timedelta import hashlib import httpx from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates import uvicorn import db as chatbot_db # Load API keys from .env file (simple parsing, no dependency) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) _env_path = os.path.join(BASE_DIR, '.env') if os.path.exists(_env_path): with open(_env_path) as _ef: for _line in _ef: _line = _line.strip() if _line and '=' in _line and not _line.startswith('#'): _k, _v = _line.split('=', 1) os.environ.setdefault(_k.strip(), _v.strip()) app = FastAPI(title="QuitSure Chatbot Onboarding") app.mount("/chatbot-onboarding/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="cb-static") app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static") templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates")) # ── LLM Config — keys from environment ── GEMINI_KEY = os.environ.get("GEMINI_API_KEY", "") GEMINI_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent" ## NVIDIA fallback removed — using Gemini only for production # ── Simple rate limiter (per-IP, in-memory) ── _rate_limit = {} # ip -> (count, window_start) RATE_LIMIT_MAX = 30 # max requests per window RATE_LIMIT_WINDOW = 60 # seconds def _client_ip(request: Request) -> str: """Real client IP. Behind a reverse proxy (nginx), request.client.host is the proxy's IP — shared by every user — so honour X-Forwarded-For first.""" xff = request.headers.get("x-forwarded-for", "") if xff: return xff.split(",")[0].strip() return request.client.host if request.client else "unknown" # Load fact library once at startup with open(os.path.join(BASE_DIR, "static/facts.json"), encoding="utf-8") as f: FACTS = json.load(f) SYSTEM_PROMPT = """You are QuitSure's quit coach, a friendly AI coach who helps people understand their smoking/vaping patterns. You talk like a real person texting a friend, not a therapist writing a report. Your users are everyday smokers from Google/Meta ads. Keep it simple, warm, real. ## YOUR ROLE You're running a structured quiz. The app controls what question comes next. You do NOT ask the next question. Your job is to: 1. Acknowledge what the user just said. Make them feel heard, not judged. 2. Connect their answer to something they said earlier when possible. This is what makes it feel like a real conversation. 3. Keep it short and human. ## YOUR VOICE - Talk like a smart friend, not a doctor. Casual but caring. - Simple words only. NEVER use: "neuroplasticity", "dopamine", "cortisol", "cognitive dissonance", "neural pathways", "cessation". - NEVER use em-dashes. Use periods, commas, or "..." instead. Em-dashes make it sound AI-generated. - NEVER start with: "Thanks", "Thank you", "Got it", "Great", "I appreciate", "Wow", "That's really interesting", "That's a great answer". Start with the actual insight. - NEVER say: "us", "we", "our" (you're one person), "it sounds like" (lazy filler), "that's really" (padding), "I hear you" (canned empathy). - NEVER pad with phrases like "and it sounds like smoking is a big part of..." — just say the sharp thing directly. - STRICT: 1-2 sentences max. If you wrote 3, delete the last one. This is a chat, not an essay. - Don't pitch the app. Don't mention subscriptions, pricing, or features unless explicitly asked. - Sound like you genuinely care, not like you're reading a script. - Use the user's name SPARINGLY — max 3-4 times across the entire conversation. Every message starting with their name feels robotic. Skip the name in most responses. Only use it at key emotional moments. ## HARD GUARDRAILS — NEVER BREAK THESE 1. NEVER invent statistics, studies, or success rates. Use ONLY the numbers in the fact library. If asked where a stat comes from, say "I don't have the specific source to hand" — NEVER fabricate a source like "internal data" or "our research shows". 2. NEVER give medical advice. If user mentions serious medical issues (heart attack, COPD diagnosis, pregnancy, mental health crisis), acknowledge it warmly but redirect: "That's really important — please talk to your doctor about that. For now, let's focus on understanding your smoking." 3. NEVER diagnose or label the user. Don't say "you have an addiction disorder" or "you're clinically dependent." Use the framing: "your brain has learned patterns." 4. NEVER describe ANY aspect of the program methodology, even vaguely. Don't say "it teaches you to handle triggers" or "it changes your relationship with smoking" or "you no longer want to smoke" or "reshape your thinking patterns." Just say "the program addresses this specifically" and stop. No hints about what's inside or what the outcome feels like. 5. NEVER answer off-topic questions. If user asks about politics, other apps, unrelated topics, gently redirect: "Good question — but let's stay focused on you right now. [next thing]." 6. NEVER use shame, fear tactics, or guilt. "Smoking kills you" is forbidden. Frame everything around self-awareness and possibility. 7. NEVER ask them another question. The app does that. Your response should NOT end with a question. 8. NEVER say "I understand how you feel" or similar canned empathy. Be specific to what they said. 9. NEVER promise a specific outcome ("you WILL quit in 6 days"). Use "can" and "most people find." 11. NEVER fabricate precision from range data. If the user data says years_smoking=7 but years_smoking_label="5-10 years", say "5-10 years" NOT "7 years". Same for cigs_per_day — use the _label field if available. The numeric values are midpoints for calculations, the labels are what the user actually said. 10. If a user types something concerning (suicide, self-harm, abuse, violence, death, being hurt/beaten), respond ONLY with: "What you're sharing sounds really serious, and it goes beyond what I can help with here. Please reach out to someone who can support you — findahelpline.com has free, confidential help. I'm here for the quit journey whenever you're ready." Do NOT engage with the details, do NOT repeat what they said, do NOT offer smoking-related advice in this context. 12. If the user types gibberish, random characters, or nonsense (e.g. "hjl", "asdf", "lol ok", single letters), respond: "Hmm, I didn't quite catch that. Could you try again?" Do NOT affirm or interpret gibberish as a real answer. 13. If the user types something clearly trolling or absurd (e.g. "my dad forces me to smoke", "aliens made me", obviously fake scenarios), don't take it at face value or get drawn in. Respond briefly and neutrally: "That's an unusual one. Let's keep going." Do NOT build emotional responses around troll answers. ## APPROVED STATISTICS — ONLY use these numbers. Never invent or round. - 80.1% success rate among program completers (JMIR Human Factors, 2024, n=1,286) - 85% of successful quitters reported no withdrawal symptoms after 30 days (JMIR, 2024) - 3,000,000+ downloads globally - Available in 175+ countries - Program duration: 6 days - Methodology: CBT + REBT + guided self-hypnosis - Pricing: Do not quote exact amounts. Say "you'll see pricing on the next screen" - 100% money-back guarantee if you complete the program and still smoke - Developed with Stanford scientists - Published in JMIR (peer-reviewed medical journal) - Featured in Google Play's "Made in India" collection - Celebrity endorsements: Arshad Warsi (Bollywood actor) publicly credited QuitSure - Cold turkey success rate: less than 5% (PMC) - NRT (patches/gums) success rate: 6-8% (WHO, 2021) - Prescription medication success rate: 7-14% (WHO/PubMed) - Hypnosis success rate: 14-25% (Cochrane Library) - Physical addiction = ~20% of the problem. Psychological = ~80%. Physical withdrawal resolves in 1-2 weeks. Psychological can persist for years. ## TONE CALIBRATION - If user is emotional: acknowledge it specifically, briefly. - If user is skeptical: don't argue. Meet them there: "Fair. A lot of people feel that way at first." - If user is enthusiastic: match the energy, stay grounded. - If user is confused: simplify, don't over-explain. ## FACT LIBRARY — only source of truth for claims {facts} ## USER PROFILE SO FAR {user_data} ## CURRENT STAGE {stage_hint} """ @app.get("/", response_class=HTMLResponse) async def index(request: Request): return templates.TemplateResponse(request, "index.html") @app.get("/api/config") async def get_config(): """Return subscription URL from .env — no Jinja2 needed.""" sub_url = os.environ.get( "SUBSCRIPTION_URL", "https://tprogram.quitsure.app/v3/eng/direct?qs=cHJvZ3JhbT0zJmxpbmstc291cmNlPWNoYXRib3Qtb25ib2FyZGluZw==" ) return {"subscription_url": sub_url} @app.post("/api/chat") async def chat(request: Request): """Send user message to LLM, get coach's response.""" # Rate limit ip = _client_ip(request) now = time.time() entry = _rate_limit.get(ip, (0, now)) if now - entry[1] > RATE_LIMIT_WINDOW: _rate_limit[ip] = (1, now) else: if entry[0] >= RATE_LIMIT_MAX: return JSONResponse({"reply": None, "error": "rate limited"}, status_code=429) _rate_limit[ip] = (entry[0] + 1, entry[1]) try: body = await request.json() except Exception: return JSONResponse({"reply": None, "error": "invalid JSON"}, status_code=400) user_message = body.get("message", "") user_data = body.get("userData", {}) context = body.get("context", "") # e.g. "User just answered the goal question" stage = body.get("stage", "") # current stage name conversation = body.get("conversation", []) # recent messages for context # Build the system prompt with current facts and user data system = SYSTEM_PROMPT.format( facts=json.dumps(FACTS, indent=2)[:4000], # truncate to fit context user_data=json.dumps(user_data, indent=2) if user_data else "No data collected yet", stage_hint=stage or "Assessment in progress." ) # Build conversation text for Gemini conv_text = "" for msg in conversation[-10:]: role = "Coach" if msg.get("role") == "assistant" else "User" conv_text += f"{role}: {msg.get('content', '')}\n" # Full prompt for Gemini (single user message with system instructions embedded) full_prompt = f"""{system} --- CONVERSATION SO FAR --- {conv_text} --- CONTEXT --- {context or 'Respond naturally.'}. Keep it to 1-3 short sentences. --- USER'S MESSAGE --- {user_message} Respond as Coach:""" # Try Gemini first reply = None async with httpx.AsyncClient(timeout=25.0) as client: try: resp = await client.post( GEMINI_URL, headers={"Content-Type": "application/json", "x-goog-api-key": GEMINI_KEY}, json={ "contents": [{"parts": [{"text": full_prompt}]}], "generationConfig": { "temperature": 0.7, "maxOutputTokens": 300, "thinkingConfig": {"thinkingBudget": 0} } } ) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: reply = parts[0].get("text", "") except Exception as e: print(f"Gemini failed: {e}") # No fallback — if Gemini fails, frontend uses scripted response # Clean up if reply: reply = reply.strip() if reply.startswith('"') and reply.endswith('"'): reply = reply[1:-1] # Remove any role prefix the model might add for prefix in ["coach:", "quira:", "quincy:", "quitsure coach:"]: if reply.lower().startswith(prefix): reply = reply[len(prefix):].strip() break return {"reply": reply} @app.post("/api/extract-name") async def extract_name(request: Request): """Use LLM to extract a person's first name from free-text input.""" try: body = await request.json() except Exception: return JSONResponse({"name": None}, status_code=400) raw = body.get("raw", "").strip() if not raw: return {"name": None} prompt = ( "Extract ONLY the person's first name from this input. " "The input may be in ANY language (Hindi, Spanish, French, etc). " "Examples: 'mera naam Neel hai' -> Neel, 'je m'appelle Pierre' -> Pierre, 'my name is Sarah' -> Sarah, 'Rahul' -> Rahul. " "Reply with JUST the name — no quotes, no punctuation, no explanation. " "If you can't find a name, reply with exactly: NONE\n\n" f"Input: \"{raw}\"\n\nName:" ) name = None async with httpx.AsyncClient(timeout=8.0) as client: try: resp = await client.post( GEMINI_URL, headers={"Content-Type": "application/json", "x-goog-api-key": GEMINI_KEY}, json={ "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.0, "maxOutputTokens": 20, "thinkingConfig": {"thinkingBudget": 0} } } ) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: name = parts[0].get("text", "").strip().strip('"').strip("'") except Exception as e: print(f"Name extraction LLM failed: {e}") # Validate: if LLM returned garbage or NONE, return null if not name or name.upper() == "NONE" or len(name) > 50 or " " in name.strip(): # Multi-word response means LLM didn't follow instructions — try first word if name and name.upper() != "NONE" and len(name) <= 50: name = name.split()[0] else: return {"name": None} # Capitalize name = name.strip().capitalize() return {"name": name} @app.post("/api/ask-quincy") async def ask_quincy_faq(request: Request): """Answer user questions about QuitSure program, grounded in real FAQ data.""" # Rate limit (shares the same store as /api/chat) ip = _client_ip(request) now = time.time() entry = _rate_limit.get(ip, (0, now)) if now - entry[1] > RATE_LIMIT_WINDOW: _rate_limit[ip] = (1, now) else: if entry[0] >= RATE_LIMIT_MAX: return JSONResponse({"reply": None, "error": "rate limited"}, status_code=429) _rate_limit[ip] = (entry[0] + 1, entry[1]) try: body = await request.json() except Exception: return JSONResponse({"reply": None}, status_code=400) question = body.get("question", "").strip()[:500] # Cap length user_data = body.get("userData", {}) if not question: return {"reply": None} # Basic prompt injection defense — strip instruction-like patterns question = re.sub(r'(?i)(ignore|forget|disregard)\s+(all|previous|above|prior)\s+(instructions?|rules?|prompts?)', '', question).strip() if not question: return {"reply": None} smoker_type = re.sub(r'[^a-z_]', '', user_data.get("smoker_type", "")) is_vaper = bool(user_data.get("isVaper", False)) raw_name = user_data.get("name", "Friend") # Sanitize name: only letters, spaces, hyphens, apostrophes — max 30 chars name = re.sub(r"[^a-zA-Z\s'\-]", '', raw_name)[:30].strip() or "Friend" user_label = f"a {smoker_type} {'vaper' if is_vaper else 'smoker'}" faq_prompt = f"""You are QuitSure's quit coach, talking to {name} ({user_label}). They just saw their assessment results and have a question. Answer like a knowledgeable friend. Start with the answer, not a greeting. PROGRAM DETAILS: - QuitSure is a 6-day app-based program. Daily sessions of 45-60 min. You quit on Day 6. - It works by reshaping your thinking patterns around smoking. By the end, you don't just stop, you no longer want to. - You keep smoking during the program until quit day. No cold turkey. No white-knuckling it on Day 1. - No patches, no medication, no willpower needed. - Works for cigarettes, vapes, e-cigarettes, chewing tobacco. Whether you smoke 5 a day or 40. WHAT A SESSION LOOKS LIKE: - Each day has short videos, reading materials, and guided mental exercises. - Day 1-2: Understanding your personal smoking psychology (why YOU smoke, not why people in general smoke). - Day 3-4: Rewiring beliefs about cigarettes. This is where most people start to feel a shift. - Day 5: Preparing for your final cigarette. By now, most people are surprised how ready they feel. - Day 6: Quit day. Not because you're forcing it, but because the desire has already changed. - After Day 6: Relapse prevention toolkit with continued exercises to stay free. COACHING: - Every subscriber gets a personal 1-on-1 human coach (not a bot, not this quiz). - You can message your coach anytime through the app. They respond within hours. - Coaches help with doubts, tough moments, and personalized guidance. - You (this quiz coach) are just the assessment coach. The program coach is a separate real person. COMMUNITY: - A private space inside the app where people share their quit journey. - People post wins ("Day 10 smoke-free!"), struggles, and support each other. - It's like a private group of people all going through the same thing. Not a public forum. PRICING: - The pricing shows on the very next screen with an exclusive web discount already applied. - For most smokers, the daily cost works out to less than what they spend on cigarettes in a day. - Don't quote exact rupee/dollar amounts. Just say "you'll see the pricing on the next screen with a special discount." - 100% money-back guarantee: complete the full program and if you're still smoking, full refund, no questions asked. - Downloading the app and signing up is free. You only pay when you choose to subscribe. - The subscription is monthly because the journey includes post-quit support beyond the 6 days. RESULTS: - 3 million+ users across 175+ countries. - 80.1% success rate among program completers (peer-reviewed, JMIR Human Factors 2024, n=1,286). 85% of successful quitters reported no withdrawal symptoms after 30 days. - The founder smoked 20 cigarettes a day for 17 years, tried quitting 30 times, then built QuitSure after figuring out why everything else fails. COMMON CONCERNS: - Weight gain: The program doesn't involve restriction or diet changes. Most people don't gain weight because the approach removes cravings rather than suppressing them. - Cravings: 80.1% success rate among program completers, and 85% reported no withdrawal symptoms after 30 days. The program changes the desire before quit day, so there's nothing to fight. - Failing again: Past methods failed because they target the physical nicotine (10% of the problem). This targets the psychological 90%. - Stress: Nicotine withdrawal actually creates stress. Smoking "relieves" that withdrawal, not the actual stress. Breaking this loop is a core part of the program. - Social situations: The program doesn't ask you to avoid friends or situations. It removes the desire so you can be in the same places without wanting a cigarette. RULES: - 2-3 sentences max. Be direct. - NEVER start with "That's a great question", "Great question", "Hey {name}", or any greeting. Start with the answer. - NEVER say "our", "we", "us" (you're one person, not a company). - NEVER say "I know seeing those results", "I totally get", "I completely understand" (filler). - NEVER use em-dashes. Use periods or commas. - NEVER use jargon (no "CBT", "neuroplasticity", "cognitive behavioral"). - NEVER guarantee they will quit. Use "most people find" or "80.1% success rate among completers". - If you genuinely don't know: "Your personal coach will cover that in detail once you start. Zero risk with the money-back guarantee." IF ASKED IF QUINCY IS A DOCTOR/REAL PERSON/AI: "I'm not a doctor. I'm a quit coach who specializes in the psychology behind smoking. For medical advice, always talk to your doctor. For understanding why you smoke and how to stop, that's where I can help." IF USER SAYS IT'S TOO EXPENSIVE / A SCAM / MAKING MONEY FROM THEIR DESPERATION: Don't be defensive. Be empathetic. Say something like: "I get it. Nobody wants to pay for something they're not sure about. That's exactly why there's a 100% money-back guarantee. You literally can't lose money on this." USER'S QUESTION: {question} Answer directly:""" reply = None async with httpx.AsyncClient(timeout=12.0) as client: try: resp = await client.post( GEMINI_URL, headers={"Content-Type": "application/json", "x-goog-api-key": GEMINI_KEY}, json={ "contents": [{"parts": [{"text": faq_prompt}]}], "generationConfig": { "temperature": 0.5, "maxOutputTokens": 200, "thinkingConfig": {"thinkingBudget": 0} } } ) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: reply = parts[0].get("text", "").strip() except Exception as e: print(f"FAQ LLM failed: {e}") if reply: reply = reply.strip() if reply.startswith('"') and reply.endswith('"'): reply = reply[1:-1] for prefix in ["coach:", "quira:", "quincy:", "quitsure coach:"]: if reply.lower().startswith(prefix): reply = reply[len(prefix):].strip() break # Strip HTML tags for safety reply = re.sub(r'<[^>]*>', '', reply) return {"reply": reply} @app.post("/api/classify-type") async def classify_smoker_type(request: Request): """Use LLM to classify smoker type based on all quiz answers.""" try: d = await request.json() except Exception: return JSONResponse({"type": None}, status_code=400) name = re.sub(r"[^a-zA-Z\s'\-]", '', d.get("name", ""))[:30].strip() or "User" is_vaper = bool(d.get("isVaper", False)) habit_word = "vaping" if is_vaper else "smoking" # Map internal codes to human-readable labels why_labels = {'withdrawal_fear': 'scared of cravings', 'stress_need': 'needs it for stress', 'identity': "it's part of who they are", 'automatic': "it's just automatic", 'enjoyment': 'they actually enjoy it', 'given_up': "they've given up trying", 'unknown': "they don't know"} fear_labels = {'withdrawal': 'cravings', 'weight': 'gaining weight', 'stress': 'losing stress outlet', 'fail_again': 'failing again', 'identity': "not knowing who they are without it", 'social_miss': 'missing the social side', 'not_ready': 'not feeling ready', 'ready': 'nothing'} deeper_whys_raw = d.get('deeper_whys', []) deeper_whys_readable = [why_labels.get(w, w) for w in deeper_whys_raw] biggest_fear_raw = d.get('biggest_fear', 'unknown') biggest_fear_readable = fear_labels.get(biggest_fear_raw, biggest_fear_raw) # Build a summary of all answers answers = f""" Name: {name} Habit: {habit_word} Years {habit_word}: {d.get('years_smoking_label', 'unknown')} Per day: {d.get('cigs_per_day_label', 'unknown')} What they like about {habit_word}: {', '.join(d.get('likes_about_smoking', [])) or 'nothing specified'} What they dislike: {', '.join(d.get('dislikes_about_smoking', [])) or 'nothing specified'} How they feel before reaching for it: {d.get('feeling_before', 'unknown')} How often they think about stopping: {d.get('thinks_about_stopping', 'unknown')} First {habit_word} after waking: {d.get('first_cig_minutes', 'unknown')} minutes Triggers: {', '.join(d.get('triggers', [])) or 'none specified'} Denial patterns selected: {', '.join(d.get('denial_patterns', [])) or 'none'} Past quit attempts: {d.get('past_attempts_count', 0)} Relapse triggers: {', '.join(d.get('relapse_triggers', [])) or 'n/a'} Methods tried: {', '.join(d.get('methods_tried', [])) or 'none'} Why they keep {habit_word}: {', '.join(deeper_whys_readable) or 'unknown'} Biggest fear about quitting: {biggest_fear_readable} """ prompt = f"""You are a smoking cessation psychologist. Based on this person's quiz answers, classify them into EXACTLY ONE of these 6 smoker types. Pick the type that BEST fits their overall pattern — not just one answer. THE 6 TYPES: 1. stress — Reaches for nicotine when pressure mounts. Smoking/vaping is their stress relief valve. Key signals: anxious before smoking, stress triggers, "it relaxes me", needs it for stress. 2. social — Smokes/vapes mainly in social situations. Key signals: social + alcohol triggers, likes the social side, occasional/light usage, few morning triggers. 3. habitual — Smoking/vaping is woven into daily routine on autopilot. Key signals: morning+meals+coffee triggers, "it's automatic", many triggers (4+), doesn't think about it. 4. emotional — Uses nicotine to process feelings (sadness, anger, loneliness). Key signals: feels anxious/bored before smoking, guilt/example in dislikes, emotional deeper-why. 5. reward — Smokes/vapes as a treat or "me moment" after accomplishments. Key signals: "it's my reward", "I genuinely enjoy it", happy before smoking, break-taking. 6. identity — Smoking/vaping is part of who they are. Key signals: 10+ years smoking, "it's part of who I am", can't imagine life without it. QUIZ ANSWERS: {answers} Respond with ONLY valid JSON in this exact format — no markdown, no explanation: {{"primary": "type_name", "secondary": "type_name", "explanation": "2-3 sentences explaining why this person is this type. Reference what they SAID (their actual words like 'it helps me calm up' or 'I love the ritual'), NOT internal field names like 'handles_stress' or 'stress_event'. NEVER use words like 'handles_stress', 'stress_event', 'identity', 'withdrawal_fear' — these are internal codes. Use the person's ACTUAL WORDS instead. Write as the quit coach speaking directly to them. No em-dashes."}} RULES: - primary and secondary MUST be different - primary and secondary MUST be one of: stress, social, habitual, emotional, reward, identity - explanation should reference THEIR specific answers, not generic descriptions - Do NOT default to stress — really look at what makes this person tick""" result = None async with httpx.AsyncClient(timeout=12.0) as client: try: resp = await client.post( GEMINI_URL, headers={"Content-Type": "application/json", "x-goog-api-key": GEMINI_KEY}, json={ "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.3, "maxOutputTokens": 300, "thinkingConfig": {"thinkingBudget": 0} } } ) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: raw = parts[0].get("text", "").strip() # Strip markdown code fences if present raw = re.sub(r'^```json\s*', '', raw) raw = re.sub(r'\s*```$', '', raw) import json as json_mod parsed = json_mod.loads(raw) valid_types = {"stress", "social", "habitual", "emotional", "reward", "identity"} if parsed.get("primary") in valid_types and parsed.get("secondary") in valid_types: result = parsed except Exception as e: print(f"LLM classify failed: {e}") return {"result": result} @app.post("/api/personal-letter") async def personal_letter(request: Request): """Generate a personalized assessment letter based on all quiz answers.""" try: d = await request.json() except Exception: return JSONResponse({"letter": None}, status_code=400) name = re.sub(r"[^a-zA-Z\s'\-]", '', d.get("name", ""))[:30].strip() or "Friend" is_vaper = bool(d.get("isVaper", False)) habit_word = "vaping" if is_vaper else "smoking" smoker_type = d.get("smoker_type", "") type_details = FACTS.get("smoker_types", {}).get(smoker_type, {}) type_name = type_details.get("name", "smoker") why_labels = {'withdrawal_fear': 'scared of cravings', 'stress_need': 'needs it for stress', 'identity': "it's part of who they are", 'automatic': "it's just automatic", 'enjoyment': 'they actually enjoy it', 'given_up': "they've given up trying", 'unknown': "they don't know"} fear_labels = {'withdrawal': 'cravings', 'weight': 'gaining weight', 'stress': 'losing stress outlet', 'fail_again': 'failing again', 'identity': "not knowing who they are without it", 'social_miss': 'missing the social side', 'not_ready': 'not feeling ready', 'ready': 'nothing'} deeper_whys_readable = [why_labels.get(w, w) for w in d.get('deeper_whys', [])] biggest_fear_readable = fear_labels.get(d.get('biggest_fear', 'unknown'), d.get('biggest_fear', 'unknown')) answers = f""" Name: {name} Habit: {habit_word} Years {habit_word}: {d.get('years_smoking_label', 'unknown')} Per day: {d.get('cigs_per_day_label', d.get('cigs_per_day', 'unknown'))} What they like about {habit_word}: {', '.join(d.get('likes_about_smoking', [])) or 'nothing specified'} What they dislike: {', '.join(d.get('dislikes_about_smoking', [])) or 'nothing specified'} How often they think about stopping: {d.get('thinks_about_stopping', 'unknown')} First {habit_word} after waking: {d.get('first_cig_minutes', 'unknown')} minutes Triggers: {', '.join(d.get('triggers', [])) or 'none specified'} Denial patterns selected: {', '.join(d.get('denial_patterns', [])) or 'none'} Past quit attempts: {d.get('past_attempts_count', 0)} Methods tried: {', '.join(d.get('methods_tried', [])) or 'none'} Why they keep {habit_word}: {', '.join(deeper_whys_readable) or 'unknown'} Biggest fear about quitting: {biggest_fear_readable} Smoking gets in the way of: {', '.join(d.get('core_values', [])) or 'unknown'} Future vision (smoke-free): {d.get('future_vision', 'not shared')} Classified as: {type_name} """ prompt = f"""You are QuitSure's quit coach, writing a personal letter to {name} after their smoking assessment. You've spent the last 5 minutes learning about them. Now write them a letter that makes them feel truly SEEN. THEIR ANSWERS: {answers} WRITE A LETTER with these 4 sections (use these exact headers): **Your Pattern** 2-3 sentences about WHY they smoke. Reference their SPECIFIC answers. Don't say "you're a stress smoker." Instead say something like "When things get tough at work, you reach for a cigarette. Not because you want one, but because your brain has learned that's what relief feels like." Make it personal. **What's Really Going On** 2-3 sentences about the psychological mechanism behind their pattern. This is the "hidden truth" moment. Connect their triggers, their likes, and their deeper-why into one insight. Keep it simple. No jargon. **Why Nothing Else Worked** 2-3 sentences connecting the methods they tried to why those methods failed. If they've never tried, talk about why willpower alone wouldn't work for their pattern. Reference the 10%/90% split (physical vs psychological). **What Would Actually Work For You** 2-3 sentences about what approach fits their specific pattern. Describe the approach first, then end with: "That's exactly what QuitSure is designed to do." Make QuitSure feel like the natural conclusion, not a pitch. RULES: - Talk like a smart friend writing a thoughtful text, not a therapist writing a report - Reference THEIR specific answers (triggers, what smoking gets in the way of, fears, years) - NEVER use em-dashes. Use periods or commas. - Keep each section to 2-3 sentences MAX - NEVER use jargon (no "neuroplasticity", "cognitive", "cessation") - NEVER use bullet points or numbered lists - Sound like you genuinely care about this specific person - Do NOT start with "Dear {name}" or "Hey {name}" — jump straight into the first section header - Do NOT end with "Sincerely" or any sign-off — the UI handles that""" letter = None async with httpx.AsyncClient(timeout=20.0) as client: try: resp = await client.post( GEMINI_URL, headers={"Content-Type": "application/json", "x-goog-api-key": GEMINI_KEY}, json={ "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.6, "maxOutputTokens": 600, "thinkingConfig": {"thinkingBudget": 0} } } ) if resp.status_code == 200: data = resp.json() candidates = data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: raw = parts[0].get("text", "").strip() # Strip markdown bold (**) and convert to HTML raw = re.sub(r'\*\*(.+?)\*\*', r'\1', raw) letter = raw except Exception as e: print(f"Personal letter LLM failed: {e}") return {"letter": letter} @app.post("/api/profile") async def build_profile(request: Request): """Server-side scoring: smoker type, success probability, program routing. Deterministic fallback — used when LLM classification fails.""" try: d = await request.json() except Exception: return JSONResponse({"error": "invalid JSON"}, status_code=400) # Helper: safely coerce to int/float def safe_int(val, default=0): if val is None: return default try: return int(val) except (ValueError, TypeError): return default def safe_float(val, default=0.0): if val is None: return default try: return float(val) except (ValueError, TypeError): return default # --- 1. Classify Smoker Type --- triggers = d.get("triggers", []) or [] # Build why_smoke from ALL available signals — not just one field # Q8 stores likes_about_smoking, Q9 dislikes, Q9b feeling_before, Q18 deeper_why likes = d.get("likes_about_smoking", []) or [] why_smoke_raw = d.get("why_smoke", []) or [] feeling = d.get("feeling_before", "") deeper = d.get("deeper_why", "") # Merge all smoking-motivation signals into one list for scoring why_smoke = list(set(likes + why_smoke_raw)) # Map feeling_before to scoring values if feeling == "anxious": why_smoke.append("stress") elif feeling == "bored": why_smoke.append("boredom") elif feeling == "automatic": why_smoke.append("automatic") elif feeling == "happy": why_smoke.append("reward") elif feeling == "relaxed": why_smoke.append("enjoy") # Map deeper_why to scoring values if deeper == "stress_need": why_smoke.append("stress") elif deeper == "identity": pass # handled by years_smoking elif deeper == "automatic": why_smoke.append("automatic") elif deeper == "enjoyment": why_smoke.append("enjoy") elif deeper == "withdrawal_fear": why_smoke.append("stress") # Map dislikes to emotional signals dislikes = d.get("dislikes_about_smoking", []) or [] if "guilt" in dislikes: why_smoke.append("angry") # guilt = emotional driver if "example" in dislikes: why_smoke.append("angry") # family-guilt = emotional why_smoke = list(set(why_smoke)) # deduplicate years_smoking = safe_int(d.get("years_smoking"), 0) cigs = safe_int(d.get("cigs_per_day"), 10) if d.get("cigs_per_day") is not None else 10 # Scoring keys below MUST match chip `value` fields in the frontend. # why_smoke options: relax, stress, boredom, focus, social, angry, reward, enjoy, automatic # triggers options: morning, coffee, meals, work, evening, alcohol, social, stress # Volume matters: 20+/day smokers are NOT social smokers. # Social gets penalized at high volume, habitual/stress get boosted. volume_bonus = 2 if cigs >= 15 else (1 if cigs >= 10 else 0) social_penalty = -3 if cigs >= 15 else (-1 if cigs >= 10 else 0) trigger_count = len(triggers) type_scores = { "stress": sum([2 if x in ["stress","relax"] else 0 for x in why_smoke]) + sum([2 if x == "stress" else 0 for x in triggers]) + sum([1 if x in ["work","evening"] else 0 for x in triggers]) + volume_bonus, "social": sum([2 if x == "social" else 0 for x in why_smoke]) + sum([2 if x in ["social","alcohol"] else 0 for x in triggers]) + social_penalty, # Heavy smokers aren't social smokers "habitual": sum([2 if x == "automatic" else 0 for x in why_smoke]) + sum([1 if x in ["morning","meals","coffee"] else 0 for x in triggers]) + sum([1 if x == "focus" else 0 for x in why_smoke]) + (2 if trigger_count >= 4 else 0) + # Many triggers = habitual volume_bonus, "emotional": sum([2 if x == "angry" else 0 for x in why_smoke]) + sum([1 if x == "boredom" else 0 for x in why_smoke]), "reward": sum([2 if x in ["reward","enjoy"] else 0 for x in why_smoke]), "identity": (3 if years_smoking and years_smoking >= 15 else 0) + (2 if years_smoking and years_smoking >= 10 else 0), } # Floor all scores at 0 type_scores = {k: max(0, v) for k, v in type_scores.items()} max_score = max(type_scores.values()) if type_scores else 0 if max_score == 0: smoker_type = "habitual" # sensible default when no data else: # On tie, prefer: stress > habitual > social > emotional > reward > identity tie_priority = ["stress", "habitual", "social", "emotional", "reward", "identity"] candidates = [k for k, v in type_scores.items() if v == max_score] smoker_type = next((t for t in tie_priority if t in candidates), candidates[0]) # --- 2. Success Probability --- # Coerce None/null to sensible defaults so arithmetic never blows up readiness = safe_int(d.get("readiness"), 5) # urgency_reason may not be set directly; infer from other signals urgency = d.get("urgency_reason") or "" if not urgency: thinks = d.get("thinks_about_stopping", "") if thinks in ["often", "always"]: urgency = "high_intent" elif deeper in ["stress_need", "withdrawal_fear"]: urgency = "controlled" past_attempts = safe_float(d.get("past_attempts_count"), 0) # cigs already defined above for type scoring first_cig = safe_int(d.get("first_cig_minutes"), 30) if d.get("first_cig_minutes") is not None else 30 prob = 60 if readiness >= 8: prob += 20 elif readiness >= 5: prob += 10 else: prob -= 5 # Any specific urgency reason (including free-text "other") gives the boost. # Only "ready" (vague) gets no bump, since it's the weakest signal. if urgency and urgency != "ready": prob += 10 if past_attempts == 0: prob += 5 elif past_attempts >= 4: prob -= 5 if cigs < 10: prob += 5 elif cigs > 20: prob -= 5 if first_cig >= 30: prob += 5 elif first_cig <= 5: prob -= 10 prob = max(65, min(94, prob)) # clamp # --- 3. Program Routing --- # All users go to 6-day program program = "six_day" # --- 4. Quit Date --- days_to_quit = 6 quit_date = (date.today() + timedelta(days=days_to_quit)).strftime("%A, %B %d") # --- 5. Savings --- # Vapers store weekly cost in vaper_weekly_cost; smokers store per-cig price vaper_weekly = safe_float(d.get("vaper_weekly_cost"), 0) if vaper_weekly > 0: daily_cost = vaper_weekly / 7 else: price = safe_float(d.get("price_per_cig"), 0) daily_cost = cigs * price savings = { "monthly": round(daily_cost * 30), "yearly": round(daily_cost * 365), "five_year": round(daily_cost * 365 * 5), } return { "smoker_type": smoker_type, "smoker_type_details": FACTS.get("smoker_types", {}).get(smoker_type, {}), "success_probability": prob, "program": program, "program_details": FACTS.get("program_routing", {}).get(program, {}), "quit_date": quit_date, "days_to_quit": days_to_quit, "savings": savings, } # ── In-memory clickstream store (replace with DB write when ready) ── _clickstream = {} # sessionId -> [{ step, answer, time, ts }] _email_captures = {} # sessionId -> { email, name, smoker_type, ts } def _resolve_user_id(session_id: str, email: str = ""): """Resolve the userId for a session. In-memory fast path, DB fallback by email. The in-memory _email_captures map is per-worker and lost on restart, so under multiple uvicorn workers results-shown/complete could miss it and silently skip the DB writes. Falling back to a lookup by email keeps those writes reliable.""" cached = _email_captures.get(session_id, {}).get("userId") if cached: return cached if email: existing = chatbot_db.check_user_exists(email) if existing: return existing["iUserID"] return None @app.post("/api/track-batch") async def track_batch(request: Request): """Clickstream tracking — receives a batch of events from the browser. Flushes happen at 8 points: after Q3, Q6, Q12, Q15, email, Q19, Q22, CTA. Also fires immediately for page_load and quiz_start. TODO: Write to tbl_ChatbotClickstream when DB is ready.""" try: body = await request.json() except Exception: return JSONResponse({"ok": False}, status_code=400) session_id = body.get("sessionId", "") events = body.get("events", []) if not session_id or not events: return {"ok": False} device_type = body.get("deviceType", "") country = body.get("country", "") source = body.get("source", "") medium = body.get("medium", "") campaign = body.get("campaign", "") if session_id not in _clickstream: _clickstream[session_id] = [] for evt in events: evt["deviceType"] = device_type evt["country"] = country evt["source"] = source evt["medium"] = medium evt["campaign"] = campaign evt["ts"] = time.time() _clickstream[session_id].append(evt) # Write to DB chatbot_db.write_clickstream_batch( session_id, events, device_type, country, source, medium, campaign ) steps = [e.get("step", "?") for e in events] print(f"[TRACK-BATCH] {session_id[:8]}... {len(events)} events: {', '.join(steps)}") return {"ok": True} @app.post("/api/save-email") async def save_email(request: Request): """Save email + name immediately when user enters it on email gate. Don't wait for CTA click — captures leads who read results but don't convert.""" try: body = await request.json() except Exception: return JSONResponse({"ok": False}, status_code=400) session_id = body.get("sessionId", "") email = body.get("email", "").strip() name = body.get("name", "").strip() smoker_type = body.get("smokerType", "") quiz_data = body.get("quizData", {}) if not email: return {"ok": False} _email_captures[session_id] = { "email": email, "name": name, "smokerType": smoker_type, "ts": time.time() } # Create user in QuitSure DB user_id = None existing = chatbot_db.check_user_exists(email) if existing: # User exists — send OTP for verification, don't update yet chatbot_db.send_otp(email) _email_captures[session_id]["pendingUserId"] = existing["iUserID"] print(f"[EMAIL] {session_id[:8]}... EXISTING user {existing['iUserID']} — OTP sent to {email}") return {"ok": True, "userId": None, "isExisting": True, "needsOtp": True} else: # New user — create across all tables user_data = { "email": email, "name": name, "age": quiz_data.get("age"), "smoker_status": quiz_data.get("smoker_status"), "isVaper": quiz_data.get("isVaper"), "cigs_per_day": quiz_data.get("cigs_per_day"), "years_smoking": quiz_data.get("years_smoking"), "first_cig_minutes": quiz_data.get("first_cig_minutes"), "triggers": quiz_data.get("triggers", []), "likes": quiz_data.get("likes_about_smoking", []), "dislikes": quiz_data.get("dislikes_about_smoking", []), "denial_patterns": quiz_data.get("denial_patterns", []), "past_attempts_count": quiz_data.get("past_attempts_count"), "methods_tried": quiz_data.get("methods_tried", []), "relapse_triggers": quiz_data.get("relapse_triggers", []), "fears": quiz_data.get("fears", []), "deeper_whys": quiz_data.get("deeper_whys", []), "currency_code": quiz_data.get("currency", "INR"), "price_per_cig": quiz_data.get("price_per_cig"), "country": quiz_data.get("_utm", {}).get("country", "IN") if isinstance(quiz_data.get("_utm"), dict) else "IN", "onboard_platform": quiz_data.get("_onboard_platform", "web-chatbot-desktop-chrome"), "utm_source": quiz_data.get("_utm", {}).get("utm_source", "chatbot-onboarding") if isinstance(quiz_data.get("_utm"), dict) else "chatbot-onboarding", "utm_medium": quiz_data.get("_utm", {}).get("utm_medium", "web") if isinstance(quiz_data.get("_utm"), dict) else "web", "utm_campaign": quiz_data.get("_utm", {}).get("utm_campaign", "chatbot") if isinstance(quiz_data.get("_utm"), dict) else "chatbot", "source_url": quiz_data.get("source_url", ""), "pay_currency": "\u20B9", } user_id = chatbot_db.create_user(user_data) print(f"[EMAIL] {session_id[:8]}... NEW user {user_id} email={email}") _email_captures[session_id]["userId"] = user_id return {"ok": True, "userId": user_id, "isExisting": False, "needsOtp": False} @app.post("/api/verify-otp") async def verify_otp_endpoint(request: Request): """Verify OTP for existing users.""" try: body = await request.json() except Exception: return JSONResponse({"ok": False}, status_code=400) session_id = body.get("sessionId", "") otp = body.get("otp", "").strip() email_data = _email_captures.get(session_id, {}) email = email_data.get("email", "") pending_user_id = email_data.get("pendingUserId") if not email or not otp or not pending_user_id: return {"ok": False, "message": "Missing data. Please try again."} result = chatbot_db.verify_otp(email, otp) if result["valid"]: # OTP verified — update existing user quiz_data = body.get("quizData", {}) chatbot_db.update_existing_user(pending_user_id, quiz_data) _email_captures[session_id]["userId"] = pending_user_id print(f"[OTP VERIFIED] {session_id[:8]}... user {pending_user_id}") return {"ok": True, "userId": pending_user_id} else: return {"ok": False, "message": result["message"]} @app.post("/api/resend-otp") async def resend_otp_endpoint(request: Request): """Resend OTP for existing users.""" try: body = await request.json() except Exception: return JSONResponse({"ok": False}, status_code=400) session_id = body.get("sessionId", "") email_data = _email_captures.get(session_id, {}) email = email_data.get("email", "") if not email: return {"ok": False, "message": "No email found."} chatbot_db.resend_otp(email) return {"ok": True, "message": "Code sent."} @app.post("/api/results-shown") async def results_shown(request: Request): """Fires when results page renders — marks onboard complete + writes quiz data.""" data = await request.json() session_id = data.get("_sessionId", "") user_id = _resolve_user_id(session_id, data.get("email", "")) if user_id: chatbot_db.mark_onboard_complete(user_id, data.get("savings", {}).get("monthly", 0)) chatbot_db.write_chatbot_onboarding(session_id, user_id, data) print(f"[RESULTS] User {user_id} — onboard complete, quiz data written") return {"ok": True} @app.post("/api/complete") async def complete_onboarding(request: Request): """Fires when user clicks CTA — marks quiz as completed.""" data = await request.json() session_id = data.get("_sessionId", "") user_id = _resolve_user_id(session_id, data.get("email", "")) _email_captures.pop(session_id, None) # cleanup the per-worker fast-path entry print("=== CTA CLICKED ===") print(f"Session: {session_id[:8]}... | UserID: {user_id}") if user_id: chatbot_db.mark_cta_clicked(session_id) return {"status": "ok"} if __name__ == "__main__": uvicorn.run("app:app", host="0.0.0.0", port=8080, reload=True)