import asyncio import json from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from models import ChatRequest, ChatResponse, RocketChatWebhook from config import settings from security import ( detect_injection, is_off_topic, check_rate_limit, store_messages, get_session_mode, set_session_mode, get_redis, ) from llm import get_ai_response from rocketchat import get_or_create_room, send_message from twenty_crm import create_lead_in_crm, save_conversation_transcript app = FastAPI(title="AImpress Chatbot API", docs_url=None, redoc_url=None) @app.get("/api/chat/health") async def health(): return {"status": "ok"} @app.post("/api/chat") async def chat(req: ChatRequest): # Rate limit check limited, count = await check_rate_limit(req.session_id) if limited: # Save transcript to CRM on rate limit (end of conversation) r_rl = await get_redis() meta_raw_rl = await r_rl.get(f"chat:meta:{req.session_id}") if meta_raw_rl: meta_rl = json.loads(meta_raw_rl) person_id = meta_rl.get("twenty_person_id") if person_id and not meta_rl.get("transcript_saved"): raw_msgs = await r_rl.lrange(f"chat:history:{req.session_id}", 0, -1) msgs = [json.loads(m) for m in raw_msgs] lead_name = meta_rl.get("lead", {}).get("name", "Visitor") asyncio.create_task(save_conversation_transcript(person_id, msgs, lead_name)) meta_rl["transcript_saved"] = True await r_rl.set(f"chat:meta:{req.session_id}", json.dumps(meta_rl), ex=settings.conversation_ttl) return ChatResponse( reply="You've reached the message limit for this session. Please leave your contact details at hello@ai-impress.com and we'll follow up personally.", sender="bot", session_id=req.session_id, message_count=count, rate_limited=True, ) # Message length validation (Pydantic handles max_length, but double-check) message = req.message[:settings.max_message_length] # Prompt injection detection if detect_injection(message): return ChatResponse( reply="I'm here to help with AImpress services. What can I assist you with today?", sender="bot", session_id=req.session_id, message_count=count, ) # Check session mode (AI vs human takeover) mode = await get_session_mode(req.session_id) if mode == "human": # Store message and forward to Rocket.Chat only messages = await store_messages(req.session_id, "user", message) room_id = await get_or_create_room(req.session_id) if room_id: asyncio.create_task(send_message(room_id, req.session_id, message, "visitor")) return ChatResponse( reply="", # Empty — waiting for human response via webhook sender="human", session_id=req.session_id, message_count=count, ) # Off-topic filter (lenient — only rejects clearly off-topic messages) if is_off_topic(message): reply = "I'm here to help with AImpress AI and automation services. What would you like to know about our solutions?" await store_messages(req.session_id, "user", message) await store_messages(req.session_id, "assistant", reply) return ChatResponse( reply=reply, sender="bot", session_id=req.session_id, message_count=count, ) # Store user message and get conversation history messages = await store_messages(req.session_id, "user", message) # Load session meta (twenty_person_id, lead info, etc.) r = await get_redis() meta_raw = await r.get(f"chat:meta:{req.session_id}") session_meta = json.loads(meta_raw) if meta_raw else {} session_meta["page_context"] = req.page_context # If lead info provided (first message), save to session meta + create CRM lead if req.lead and req.lead.name and "lead" not in session_meta: lead_info = {"name": req.lead.name} if req.lead.email: lead_info["email"] = req.lead.email if req.lead.company: lead_info["company"] = req.lead.company session_meta["lead"] = lead_info # Create lead in Twenty CRM immediately from form data person_id = await create_lead_in_crm( name=req.lead.name, email=req.lead.email or "", company=req.lead.company or "", need=message, page_context=req.page_context or "/", ) if person_id: session_meta["twenty_person_id"] = person_id # Always prepend lead context from session meta (persists across messages) stored_lead = session_meta.get("lead") if stored_lead: lead_context = f"[System: The visitor has introduced themselves. Name: {stored_lead['name']}" if stored_lead.get("email"): lead_context += f", Email: {stored_lead['email']}" if stored_lead.get("company"): lead_context += f", Company: {stored_lead['company']}" lead_context += ". Their lead is ALREADY captured in CRM — do NOT call capture_lead again." lead_context += " Use this info naturally — greet them by name. Don't re-ask for info you already have." lead_context += " IMPORTANT: Whenever the visitor mentions budget, timeline, requirements, job title, phone, city, or any useful detail — IMMEDIATELY call update_lead with the 'note' field to enrich their CRM profile.]" messages = [{"role": "user", "content": lead_context}, {"role": "assistant", "content": "Understood. I'll use update_lead to enrich their profile whenever they share new details."}] + messages # Prepend language hint for non-English visitors if req.language == "uk": messages = [ {"role": "user", "content": "[System: The visitor's interface language is Ukrainian. Default to Ukrainian in your responses unless they write in English.]"}, {"role": "assistant", "content": "Зрозуміло. Я буду відповідати українською."}, ] + messages # Get AI response try: reply, lead_data = await get_ai_response(messages, session_meta) except Exception as e: reply = "I'm sorry, I'm having a technical issue. Please try again or contact us at hello@ai-impress.com." print(f"LLM error: {e}") lead_data = None # Handle escalation to human if triggered escalation = session_meta.pop("_escalate", None) if escalation: await set_session_mode(req.session_id, "human") # Save conversation transcript to CRM on escalation person_id = session_meta.get("twenty_person_id") if person_id: visitor_name = stored_lead.get("name", "Visitor") if stored_lead else "Visitor" asyncio.create_task(save_conversation_transcript(person_id, messages, visitor_name)) # Persist session meta (may contain new twenty_person_id) await r.set(f"chat:meta:{req.session_id}", json.dumps(session_meta), ex=settings.conversation_ttl) # Store bot reply await store_messages(req.session_id, "assistant", reply) # Mirror to Rocket.Chat (async, don't block response) visitor_name = req.lead.name if req.lead and req.lead.name else "Website Visitor" room_id = await get_or_create_room(req.session_id, visitor_name) if room_id: asyncio.create_task(send_message(room_id, req.session_id, message, "visitor")) asyncio.create_task(send_message(room_id, req.session_id, reply, "bot")) # Notify RC agent about escalation if escalation: notice = f"⚡ ESCALATION: {escalation['reason']}\n📋 Summary: {escalation['summary']}" asyncio.create_task(send_message(room_id, req.session_id, notice, "bot")) return ChatResponse( reply=reply, sender="bot", session_id=req.session_id, message_count=count, ) @app.post("/api/chat/webhook/rocketchat") async def rocketchat_webhook(req: Request): """Webhook from Rocket.Chat when a manager sends a message.""" body = await req.json() # RC livechat webhook format: {_id, visitor, agent, messages, type} messages_list = body.get("messages", []) visitor = body.get("visitor", {}) session_id = visitor.get("token", "") if not messages_list or not session_id: return {"status": "ignored"} r = await get_redis() for msg in messages_list: text = msg.get("msg", "") sender_username = msg.get("u", {}).get("username", "") sender_name = msg.get("u", {}).get("name", sender_username) # Skip messages from visitors (our own bot echo) — only process agent messages if sender_username == visitor.get("username", ""): continue if not text: continue # Strip bot emoji prefix if present if text.startswith("🤖 "): continue # Check for /ai command to resume AI mode if text.strip().lower() == "/ai": await set_session_mode(session_id, "ai") return {"status": "ai_resumed"} # Set session to human mode and store the manager's message await set_session_mode(session_id, "human") await store_messages(session_id, "assistant", f"[{sender_name}] {text}") # Store for polling delivery to frontend await r.rpush( f"chat:pending:{session_id}", json.dumps({"sender": "human", "text": text, "agent": sender_name}), ) await r.expire(f"chat:pending:{session_id}", 300) return {"status": "delivered"} @app.get("/api/chat/pending/{session_id}") async def get_pending_messages(session_id: str): """Poll for pending human messages (Phase 1 polling, Phase 2 SSE).""" from security import get_redis import json r = await get_redis() key = f"chat:pending:{session_id}" raw = await r.lrange(key, 0, -1) await r.delete(key) messages = [json.loads(m) for m in raw] return {"messages": messages}