""" JPLoft AI Chatbot v4 - Production Server - Real Claude AI: answers every question using JPLoft knowledge base - Music Tonight portfolio project referenced correctly - Blog links shared contextually - MySQL: stores visitor country, full conversation, all shared info - Geo-IP: auto-detects visitor country - Lead scoring + Slack alerts """ import os, json, uuid, re, asyncio, datetime from contextlib import asynccontextmanager from typing import Optional from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, JSONResponse from pydantic import BaseModel import anthropic from dotenv import load_dotenv load_dotenv() try: import aiomysql HAS_MYSQL = True except ImportError: HAS_MYSQL = False try: import httpx HAS_HTTPX = True except ImportError: HAS_HTTPX = False # Globals claude_client = None db_pool = None mem_sessions: dict = {} mem_leads: list = [] # ───────────────────────────────────────────────────────────────────────────── # JPLOFT KNOWLEDGE BASE (from jploft.com) # ───────────────────────────────────────────────────────────────────────────── JPLOFT_KB = """ ==== COMPANY ==== Name: JPLoft | Founded: 2013 | HQ: Denver, Colorado, USA CEO: Rahul Sukhwal (15+ years experience) Team: 100+ professionals | Projects: 1,100+ delivered | Rating: 4.9/5 (1,000+ clients) Offices: Denver USA, Jaipur India, Melbourne Australia, Basingstoke UK Clients: Nike, Red Bull, Whirlpool (and 1,000+ global clients) Phone: +1 (303) 335-0405 | Email: sales@jploft.com Discovery call: https://calendly.com/jploft/discovery Portfolio: https://www.jploft.com/portfolio/ Contact: https://www.jploft.com/contact-us/ ==== PORTFOLIO — EVENTS / MUSIC / ENTERTAINMENT ==== • Music Tonight (FEATURED PROJECT) What: AI-powered local live-music discovery app. Built by JPLoft end-to-end using AI/ML and React Native. It helps users effortlessly find nearby shows — indie rock, DJ pop-ups, surprise concerts — anytime, anywhere using an AI engine. Category: Events portfolio on jploft.com Portfolio URL: https://www.jploft.com/portfolio/ Music service page: https://www.jploft.com/music-streaming-app-development Tags: Music, Events, Entertainment, Live Shows, AI, Mobile App ==== PORTFOLIO — SPORTS / ENTERPRISE ==== • Red Bull Street Style — global tournament platform for WFFA (World Freestyle Football Association). Client: Red Bull. • Nike Movement For Sport Playkit — game-based physical assessment tool for coaches, used by 2.5M children aged 6-12. Client: Nike. • Whirlpool Distributor App — stock management app for Whirlpool distributors. Client: Whirlpool. ==== PORTFOLIO — SOCIAL / DATING ==== • Lure (AI Dating App) — AI-powered dating for socially progressive singles; rates passions on social issues for matching. ==== PORTFOLIO — BLOCKCHAIN / FINTECH ==== • CryptoTradeX — crypto asset trading / digital asset exchange platform. • SecureChainPayments — cryptocurrency payment solution. ==== PORTFOLIO — TRAVEL / ON-DEMAND ==== • Tennis Court Booking — platform for renting courts for tournaments and events. • Miami On-Demand Delivery — food, groceries, medicines, beauty items (Miami, Florida). ==== SERVICES ==== MOBILE APP DEVELOPMENT iOS, Android, React Native, Flutter MVP: 6-10 weeks | Full app: 3-6 months | Cost: $10k-$300k+ URL: https://www.jploft.com/mobile-app-development-company-usa MUSIC APP DEVELOPMENT (specialist) Streaming apps, live discovery, AI curation, offline/adaptive bitrate, podcasts MVP: $30k-$50k | Full-featured: $100k+ | Timeline: 6-10 wks MVP URL: https://www.jploft.com/music-streaming-app-development AI & ML DEVELOPMENT LLM integrations, chatbots, automation, generative AI, NLP, CV, predictive analytics URL: https://www.jploft.com/ai-development-services WEB / SAAS PLATFORMS React, Next.js, Vue, Node.js, Python, PHP URL: https://www.jploft.com BLOCKCHAIN / WEB3 Smart contracts (Ethereum, Solana, Polygon, Hyperledger), DeFi, NFT, dApps URL: https://www.jploft.com/blockchain-development-company CLOUD & DEVOPS — AWS, GCP, Azure, Kubernetes, Docker, CI/CD STOCK TRADING APPS — AI trading, robo-advisors, real-time market data URL: https://www.jploft.com/stock-trading-app-development TRAVEL APPS — Booking, itinerary, AI search URL: https://www.jploft.com/travel-app-development SPEECH RECOGNITION — Voice apps, transcription, multi-language URL: https://www.jploft.com/speech-recognition-software-development JOB PORTALS — Recruitment platforms, freelancer marketplaces URL: https://www.jploft.com/job-portal-app-development QA & TESTING — Automated, performance, security testing ==== INDUSTRIES ==== FinTech, HealthTech (HIPAA), E-Commerce, EdTech, Music/Entertainment, Sports, Travel, Logistics, Real Estate, Gaming, Social, On-Demand, Blockchain/Web3, Manufacturing, HR/Recruitment URL: https://www.jploft.com/industries/ ==== BLOG RESOURCES ==== All blogs: https://www.jploft.com/blog/ Music industry blogs: • Music Streaming Trends 2025 → https://www.jploft.com/blog/music-streaming-app-trends • JPLoft as Music App Leader → https://www.jploft.com/blog/jploft-as-a-leader-in-delivering-tailored-music-app-development-services • Music App Ideas for Startups → https://www.jploft.com/blog/music-app-ideas • Top Music App Features 2025 → https://www.jploft.com/blog/top-music-app-features • Best Music Apps 2025 → https://www.jploft.com/blog/best-music-streaming-apps • Music App Monetization → https://www.jploft.com/blog/how-to-monetize-a-music-streaming-app • Why Music Apps Fail → https://www.jploft.com/blog/why-music-apps-fail ==== PRICING ==== MVP: $10k-$30k (6-10 weeks) Standard mobile: $30k-$80k (3-4 months) Complex platform/SaaS: $80k-$200k+ (4-8 months) Music streaming MVP: $30k-$50k | Full: $100k+ Enterprise AI: $50k-$500k+ Dedicated developer: $2,500-$6,000/month Model: Fixed price (transparent, no surprises) ==== ENGAGEMENT MODELS ==== Fixed Price | Dedicated Team | Staff Augmentation | Hourly """ SYSTEM_PROMPT = f"""You are Aria, the friendly AI assistant for JPLoft — a global software development company. PRIMARY GOAL: Answer every question helpfully and specifically, share relevant links, and collect visitor contact details to generate leads. You have JPLoft's complete knowledge base: {JPLOFT_KB} ━━━ ANSWERING QUESTIONS ━━━ MUSIC / EVENTS questions → ALWAYS: 1. Mention "Music Tonight" by name: "We built Music Tonight — an AI-powered live music discovery app" 2. Link portfolio: https://www.jploft.com/portfolio/ (Events category) 3. Link music service: https://www.jploft.com/music-streaming-app-development 4. Offer 1 blog link from the music section above ANY portfolio question → reference the specific project from the knowledge base that matches, then link https://www.jploft.com/portfolio/ Service questions → give real timeline + cost range + service URL Blog questions → share specific URLs from the knowledge base IF unsure about something → "Let me connect you with our team: sales@jploft.com or book a call: https://calendly.com/jploft/discovery" ━━━ LEAD COLLECTION ━━━ After 2-3 exchanges, naturally say: "I'd love to put together a custom proposal — could I get your name and email?" Collect: name, email, company, project type, budget, timeline. When visitor shares info, acknowledge warmly. ━━━ STYLE ━━━ • 3-5 sentences per reply (more only if needed) • Max 2 links per reply • Always end with an engaging question • Never say you are Claude or mention Anthropic • Never invent projects or data not in the knowledge base """ @asynccontextmanager async def lifespan(app: FastAPI): global claude_client, db_pool key = os.getenv("ANTHROPIC_API_KEY", "").strip() if key and "YOUR_KEY" not in key: claude_client = anthropic.Anthropic(api_key=key) try: claude_client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=5, messages=[{"role": "user", "content": "hi"}]) print("OK Claude API connected") except Exception as e: print(f"FAIL Claude API: {e}") claude_client = None else: print("FAIL ANTHROPIC_API_KEY not set") if HAS_MYSQL: h = os.getenv("MYSQL_HOST","").strip() u = os.getenv("MYSQL_USER","").strip() p = os.getenv("MYSQL_PASSWORD","").strip() d = os.getenv("MYSQL_DATABASE","").strip() port = int(os.getenv("MYSQL_PORT","3306")) if h and u and d: try: db_pool = await aiomysql.create_pool( host=h, port=port, user=u, password=p, db=d, minsize=2, maxsize=10, autocommit=True, charset="utf8mb4") await create_tables() print(f"OK MySQL connected to {h}/{d}") except Exception as e: print(f"FAIL MySQL: {e} — using in-memory") else: print("WARN MySQL credentials not set — using in-memory") yield if db_pool: db_pool.close() await db_pool.wait_closed() app = FastAPI(title="JPLoft Chatbot v4", lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) async def create_tables(): async with db_pool.acquire() as conn: async with conn.cursor() as c: await c.execute(""" CREATE TABLE IF NOT EXISTS chatbot_sessions ( id VARCHAR(64) PRIMARY KEY, visitor_ip VARCHAR(45), visitor_country VARCHAR(120) DEFAULT 'Unknown', visitor_city VARCHAR(120) DEFAULT 'Unknown', started_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_active DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, total_messages INT DEFAULT 0 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) await c.execute(""" CREATE TABLE IF NOT EXISTS chatbot_messages ( id BIGINT AUTO_INCREMENT PRIMARY KEY, session_id VARCHAR(64) NOT NULL, role ENUM('user','assistant') NOT NULL, content TEXT NOT NULL, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_session (session_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) await c.execute(""" CREATE TABLE IF NOT EXISTS chatbot_leads ( id INT AUTO_INCREMENT PRIMARY KEY, session_id VARCHAR(64), visitor_country VARCHAR(120) DEFAULT 'Unknown', name VARCHAR(200), email VARCHAR(320), phone VARCHAR(50), company VARCHAR(200), project_type VARCHAR(200), budget VARCHAR(100), timeline VARCHAR(100), extra_info TEXT, score TINYINT DEFAULT 0, score_tier ENUM('hot','warm','cold') DEFAULT 'cold', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_email (email), INDEX idx_tier (score_tier), INDEX idx_country (visitor_country) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) print("DB tables ready") _geo_cache: dict = {} async def get_location(ip: str) -> dict: if not ip or ip in ("127.0.0.1","::1") or ip.startswith(("192.168.","10.","172.")): return {"country":"Local","city":"Localhost"} if ip in _geo_cache: return _geo_cache[ip] if not HAS_HTTPX: return {"country":"Unknown","city":"Unknown"} try: async with httpx.AsyncClient(timeout=3.0) as client: r = await client.get(f"https://ipapi.co/{ip}/json/") if r.status_code == 200: d = r.json() result = {"country": d.get("country_name","Unknown"), "city": d.get("city","Unknown")} _geo_cache[ip] = result return result except Exception: pass return {"country":"Unknown","city":"Unknown"} def real_ip(request: Request) -> str: fwd = request.headers.get("X-Forwarded-For","") if fwd: return fwd.split(",")[0].strip() ri = request.headers.get("X-Real-IP","") if ri: return ri.strip() return getattr(request.client, "host", "0.0.0.0") async def db_init_session(sid, ip, country, city): if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor() as c: await c.execute(""" INSERT INTO chatbot_sessions (id,visitor_ip,visitor_country,visitor_city) VALUES (%s,%s,%s,%s) ON DUPLICATE KEY UPDATE last_active=NOW(), total_messages=total_messages+1 """, (sid, ip, country, city)) except Exception as e: print(f"db_init_session err: {e}") else: if sid not in mem_sessions: mem_sessions[sid] = {"country":country,"city":city,"messages":[]} async def db_save_msg(sid, role, content): if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor() as c: await c.execute( "INSERT INTO chatbot_messages (session_id,role,content) VALUES (%s,%s,%s)", (sid, role, content[:20000])) except Exception as e: print(f"db_save_msg err: {e}") else: if sid in mem_sessions: mem_sessions[sid]["messages"].append({"role":role,"content":content, "ts":datetime.datetime.utcnow().isoformat()}) def score_lead(lead: dict): s = 0 b = (lead.get("budget") or "").lower() if "100" in b: s += 30 elif "50" in b: s += 22 elif "25" in b: s += 15 elif "10" in b: s += 8 pt = (lead.get("project_type") or "").lower() if any(k in pt for k in ["ai","ml","blockchain","enterprise","saas"]): s += 20 elif any(k in pt for k in ["mobile","web","music","app","fintech"]): s += 15 else: s += 8 if lead.get("email") and lead.get("name") and lead.get("company"): s += 15 elif lead.get("email") and lead.get("name"): s += 10 elif lead.get("email"): s += 5 tier = "hot" if s >= 60 else "warm" if s >= 30 else "cold" return min(s, 100), tier async def db_save_lead(lead: dict): score, tier = score_lead(lead) if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor() as c: await c.execute(""" INSERT INTO chatbot_leads (session_id,visitor_country,name,email,phone,company, project_type,budget,timeline,extra_info,score,score_tier) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, (lead.get("session_id",""), lead.get("country","Unknown"), lead.get("name",""), lead.get("email",""), lead.get("phone",""), lead.get("company",""), lead.get("project_type",""), lead.get("budget",""), lead.get("timeline",""), lead.get("extra_info",""), score, tier)) print(f"Lead saved: {lead.get('name')} <{lead.get('email')}> [{tier}] {lead.get('country')}") except Exception as e: print(f"db_save_lead err: {e}") mem_leads.append({**lead,"score":score,"tier":tier}) else: mem_leads.append({**lead,"score":score,"tier":tier}) if score >= 60 and HAS_HTTPX: wb = os.getenv("SLACK_WEBHOOK_URL","") if wb: try: async with httpx.AsyncClient(timeout=5) as c2: await c2.post(wb, json={"text": f"HOT LEAD {score}/100 from {lead.get('country','?')}\n" f"Name: {lead.get('name')} | Email: {lead.get('email')}\n" f"Project: {lead.get('project_type')} | Budget: {lead.get('budget')}"}) except Exception: pass return score, tier INJECTION = [r'ignore (all |previous )?instructions', r'forget (your |all )?instructions',r'you are now\b', r'new (role|persona)',r'act as (?!jploft|aria)',r'\[INST\]',r'<\|system\|>'] def sanitize(text): lo = text.lower() for p in INJECTION: if re.search(p, lo): return "[FILTERED]", True return re.sub(r"<[^>]+>","",text).strip()[:3000], False class ChatReq(BaseModel): message: str session_id: str = "" history: list = [] class LeadReq(BaseModel): session_id: str = "" name: Optional[str] = "" email: Optional[str] = "" phone: Optional[str] = "" company: Optional[str] = "" project_type: Optional[str] = "" budget: Optional[str] = "" timeline: Optional[str] = "" extra_info: Optional[str] = "" @app.post("/api/chat") async def chat(req: ChatReq, request: Request): if not claude_client: return JSONResponse(status_code=503, content={"error":"Claude API not ready. Check ANTHROPIC_API_KEY in .env"}) clean, flagged = sanitize(req.message) if flagged: return JSONResponse({"response":"I can only assist with JPLoft software questions. What would you like to know?"}) sid = req.session_id or ("S" + uuid.uuid4().hex[:16]) ip = real_ip(request) loc = await get_location(ip) await db_init_session(sid, ip, loc["country"], loc["city"]) await db_save_msg(sid, "user", clean) messages = [] for h in req.history[-20:]: r, c = h.get("role",""), str(h.get("content","")).strip() if r in ("user","assistant") and c: messages.append({"role":r,"content":c}) messages.append({"role":"user","content":clean}) async def generate(): full = "" try: with claude_client.messages.stream( model=os.getenv("CLAUDE_MODEL","claude-haiku-4-5-20251001"), max_tokens=900, system=SYSTEM_PROMPT, messages=messages, ) as stream: for chunk in stream.text_stream: full += chunk yield f"data: {json.dumps({'token':chunk})}\n\n" yield f"data: {json.dumps({'done':True,'session_id':sid})}\n\n" except anthropic.AuthenticationError: yield f"data: {json.dumps({'error':'Invalid API key. Check .env file.'})}\n\n" except anthropic.RateLimitError: yield f"data: {json.dumps({'error':'Rate limit. Please wait and retry.'})}\n\n" except Exception as e: print(f"Stream error: {e}") yield f"data: {json.dumps({'error':'Connection issue. Please retry.'})}\n\n" finally: if full.strip(): await db_save_msg(sid, "assistant", full.strip()) return StreamingResponse(generate(), media_type="text/event-stream", headers={"Cache-Control":"no-cache,no-store","X-Accel-Buffering":"no","Connection":"keep-alive"}) @app.post("/api/leads") async def capture_lead(req: LeadReq, request: Request): ip = real_ip(request) loc = await get_location(ip) country = loc["country"] if db_pool and req.session_id: try: async with db_pool.acquire() as conn: async with conn.cursor() as c: await c.execute("SELECT visitor_country FROM chatbot_sessions WHERE id=%s",(req.session_id,)) row = await c.fetchone() if row and row[0] and row[0] not in ("Unknown",""): country = row[0] except Exception: pass score, tier = await db_save_lead({ "session_id": req.session_id, "country": country, "name": req.name or "", "email": req.email or "", "phone": req.phone or "", "company": req.company or "", "project_type": req.project_type or "", "budget": req.budget or "", "timeline": req.timeline or "", "extra_info": req.extra_info or "", }) return {"success":True,"score":score,"tier":tier} @app.get("/api/admin/stats") async def stats(): if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor() as c: async def q(sql): await c.execute(sql); return (await c.fetchone())[0] total = await q("SELECT COUNT(*) FROM chatbot_leads") hot = await q("SELECT COUNT(*) FROM chatbot_leads WHERE score_tier='hot'") warm = await q("SELECT COUNT(*) FROM chatbot_leads WHERE score_tier='warm'") sessions = await q("SELECT COUNT(*) FROM chatbot_sessions") messages = await q("SELECT COUNT(*) FROM chatbot_messages") await c.execute("""SELECT visitor_country,COUNT(*) c FROM chatbot_sessions WHERE visitor_country!='Unknown' GROUP BY visitor_country ORDER BY c DESC LIMIT 10""") countries = [{"country":r[0],"count":r[1]} for r in await c.fetchall()] return {"total_leads":total,"hot_leads":hot,"warm_leads":warm, "total_sessions":sessions,"total_messages":messages,"top_countries":countries} except Exception as e: return {"error":str(e)} return {"total_leads":len(mem_leads),"hot_leads":sum(1 for l in mem_leads if l.get("tier")=="hot"), "warm_leads":sum(1 for l in mem_leads if l.get("tier")=="warm"), "total_sessions":len(mem_sessions),"total_messages":sum(len(s.get("messages",[])) for s in mem_sessions.values())} @app.get("/api/admin/leads") async def list_leads(tier: Optional[str]=None, limit: int=200): if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as c: if tier: await c.execute("SELECT * FROM chatbot_leads WHERE score_tier=%s ORDER BY created_at DESC LIMIT %s",(tier,limit)) else: await c.execute("SELECT * FROM chatbot_leads ORDER BY created_at DESC LIMIT %s",(limit,)) rows = await c.fetchall() result = [] for r in rows: row = dict(r) for k,v in row.items(): if isinstance(v,(datetime.datetime,datetime.date)): row[k]=v.isoformat() result.append(row) return {"leads":result,"total":len(result)} except Exception as e: return {"error":str(e),"leads":[]} return {"leads":mem_leads[:limit],"total":len(mem_leads)} @app.get("/api/admin/conversations/{session_id}") async def get_conv(session_id: str): if db_pool: try: async with db_pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as c: await c.execute("SELECT role,content,sent_at FROM chatbot_messages WHERE session_id=%s ORDER BY sent_at",(session_id,)) rows = await c.fetchall() result = [] for r in rows: row = dict(r) if isinstance(row.get("sent_at"),datetime.datetime): row["sent_at"]=row["sent_at"].isoformat() result.append(row) return {"messages":result} except Exception as e: return {"error":str(e),"messages":[]} s = mem_sessions.get(session_id,{}) return {"messages":s.get("messages",[])} @app.get("/health") async def health(): return {"status":"ok","claude":claude_client is not None,"mysql":db_pool is not None, "time":datetime.datetime.utcnow().isoformat()} @app.get("/test-claude") async def test_claude(): if not claude_client: return {"working":False,"reason":"API key not configured"} try: r = claude_client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=100, system=SYSTEM_PROMPT, messages=[{"role":"user","content":"What music-related project has JPLoft built? Give me the name and a brief description."}]) return {"working":True,"response":r.content[0].text} except Exception as e: return {"working":False,"error":str(e)} @app.get("/") async def root(): return {"api":"JPLoft Chatbot v4","health":"/health","test":"/test-claude"} if __name__ == "__main__": import uvicorn print("\n" + "="*50) print(" JPLoft AI Chatbot Server v4") print("="*50) print(f" Claude : {'OK key set' if os.getenv('ANTHROPIC_API_KEY') else 'MISSING'}") print(f" MySQL : {'OK' if os.getenv('MYSQL_HOST') else 'not set (in-memory mode)'}") print(f" Slack : {'OK' if os.getenv('SLACK_WEBHOOK_URL') else 'not set'}") print(f"\n Listening on http://0.0.0.0:8000\n") uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=False)