Download app.py from DOOM1812/face: direct link, hf CLI and curl.
- Browser
- Download file 22 kB
-
https://huggingface.co/spaces/DOOM1812/face/resolve/main/app.py
- Command line
-
hf download hf://spaces/DOOM1812/face/app.py
-
curl -L -o app.py https://huggingface.co/spaces/DOOM1812/face/resolve/main/app.py
22 kB
| from fastapi import FastAPI, File, UploadFile , Request | |
| from fastapi.routing import APIRoute | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.middleware import Middleware | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from starlette.middleware.base import BaseHTTPMiddleware | |
| from fastapi.staticfiles import StaticFiles | |
| import os, pandas as pd | |
| from typing import List | |
| from langchain_community.embeddings import OpenAIEmbeddings | |
| from langchain_community.vectorstores import FAISS | |
| from langchain_openai import ChatOpenAI | |
| from langchain.chains import RetrievalQA | |
| from langchain.prompts import PromptTemplate | |
| from datetime import datetime, timedelta, timezone | |
| from dotenv import load_dotenv | |
| import io | |
| from starlette.responses import Response | |
| class CaseInsensitiveRoute(APIRoute): | |
| def matches(self, scope: dict) -> tuple[bool, dict]: | |
| # Only lowercase if it's not a static file | |
| if scope["type"] == "http" and not scope["path"].startswith("/static"): | |
| scope["path"] = scope["path"].lower() | |
| return super().matches(scope) | |
| load_dotenv() | |
| app = FastAPI() | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| middleware = [ | |
| Middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) | |
| ] | |
| app.router.route_class = CaseInsensitiveRoute | |
| PERSISTENT_FAISS_PATH = "/tmp/faiss_store" | |
| os.makedirs(PERSISTENT_FAISS_PATH, exist_ok=True) | |
| vector_store = None | |
| async def serve_index(): | |
| return HTMLResponse(content=""" | |
| <!DOCTYPE html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="UTF-8" /> | |
| <title>AI Receptionist</title> | |
| <style> | |
| body { | |
| background: #ffffff; | |
| font-family: sans-serif; | |
| text-align: center; | |
| margin-top: 3em; | |
| color: #000; | |
| } | |
| .avatar { | |
| width: 280px; | |
| height: auto; | |
| object-fit: contain; | |
| display: none; | |
| } | |
| .visible { | |
| display: inline; | |
| } | |
| #status { | |
| font-size: 1.2em; | |
| font-weight: bold; | |
| margin: 2rem 0 0.3rem; | |
| color: #222; | |
| } | |
| #response { | |
| font-size: 0.95em; | |
| color: #444; | |
| min-height: 1.5em; | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <img src="/static/jpg.jpg" id="staticAvatar" class="avatar visible" alt="Idle Avatar" /> | |
| <img src="/static/gif.gif" id="talkingAvatar" class="avatar" alt="Talking Avatar" /> | |
| <div id="status">Idle. Please say "hello" to begin.</div> | |
| <div id="response"></div> | |
| <script> | |
| const SpeechRecognition = window.SpeechRecognition || window.webkitSpeechRecognition; | |
| const recognition = SpeechRecognition ? new SpeechRecognition() : null; | |
| if (!recognition) { | |
| alert("SpeechRecognition not supported in this browser."); | |
| } else { | |
| recognition.continuous = false; | |
| recognition.interimResults = false; | |
| recognition.lang = 'en-US'; | |
| const staticAvatar = document.getElementById('staticAvatar'); | |
| const talkingAvatar = document.getElementById('talkingAvatar'); | |
| const statusEl = document.getElementById('status'); | |
| const responseEl = document.getElementById('response'); | |
| let sessionActive = false; | |
| let manuallyStopped = false; | |
| let selectedVoice = null; | |
| function pickMaleVoice() { | |
| const voices = speechSynthesis.getVoices(); | |
| const preferred = ["Male", "Google US English Male", "Microsoft David", "Alex", "Daniel", "Mark"]; | |
| for (let name of preferred) { | |
| const voice = voices.find(v => v.name.includes(name)); | |
| if (voice) return voice; | |
| } | |
| return voices.find(v => /male/i.test(v.name)) || voices[0]; | |
| } | |
| function initVoice() { | |
| selectedVoice = pickMaleVoice(); | |
| if (!selectedVoice) { | |
| speechSynthesis.onvoiceschanged = () => { | |
| selectedVoice = pickMaleVoice(); | |
| }; | |
| } | |
| } | |
| initVoice(); | |
| function startListening() { | |
| try { recognition.start(); } | |
| catch (e) { console.warn("Start error:", e.message); } | |
| } | |
| function showTalking() { | |
| staticAvatar.classList.remove('visible'); | |
| talkingAvatar.classList.add('visible'); | |
| } | |
| function showStatic() { | |
| talkingAvatar.classList.remove('visible'); | |
| staticAvatar.classList.add('visible'); | |
| } | |
| function speakText(text) { | |
| if (!text) return; | |
| const utter = new SpeechSynthesisUtterance(text); | |
| utter.voice = selectedVoice || speechSynthesis.getVoices()[0] || null; | |
| utter.rate = 1.0; | |
| utter.pitch = 1.0; | |
| let ended = false; | |
| utter.onstart = () => { | |
| statusEl.textContent = "π¬ Speaking..."; | |
| showTalking(); | |
| }; | |
| utter.onend = () => { | |
| ended = true; | |
| showStatic(); | |
| statusEl.textContent = sessionActive ? "π§ Listening..." : 'Idle. Please say "hello" to begin.'; | |
| manuallyStopped = false; | |
| startListening(); | |
| }; | |
| utter.onerror = (e) => { | |
| console.error("Speech error:", e.error); | |
| if (!ended) { | |
| ended = true; | |
| showStatic(); | |
| statusEl.textContent = "TTS error."; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }; | |
| // Fallback timeout | |
| setTimeout(() => { | |
| if (!ended) { | |
| console.warn("Fallback triggered (speech may have failed)"); | |
| showStatic(); | |
| statusEl.textContent = sessionActive ? "π§ Listening..." : 'Idle. Please say "hello" to begin.'; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }, 8000); | |
| responseEl.textContent = text; | |
| setTimeout(() => { | |
| speechSynthesis.cancel(); | |
| speechSynthesis.speak(utter); | |
| }, 0); | |
| } | |
| function handleQuery(queryText) { | |
| statusEl.textContent = "π€ Thinking..."; | |
| responseEl.textContent = ""; | |
| fetch("/ask", { | |
| method: "POST", | |
| headers: { "Content-Type": "application/json" }, | |
| body: JSON.stringify({ query: queryText }) | |
| }) | |
| .then(res => res.ok ? res.json() : Promise.reject(res.statusText)) | |
| .then(data => { | |
| const answer = data.answer || data.response || ""; | |
| if (answer) { | |
| speakText(answer); | |
| } else { | |
| statusEl.textContent = "Error: No response."; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }) | |
| .catch(err => { | |
| console.error("Fetch failed:", err); | |
| statusEl.textContent = "Error fetching response."; | |
| manuallyStopped = false; | |
| startListening(); | |
| }); | |
| } | |
| recognition.onresult = (event) => { | |
| const transcript = event.results[0][0].transcript.trim(); | |
| console.log("π€ Heard:", transcript); | |
| if (!sessionActive) { | |
| if (transcript.toLowerCase().includes("hello")) { | |
| sessionActive = true; | |
| } else { | |
| console.log("Ignored (waiting for hello)"); | |
| return; | |
| } | |
| } | |
| if (transcript.toLowerCase().includes("goodbye")) { | |
| sessionActive = false; | |
| } | |
| manuallyStopped = true; | |
| recognition.stop(); | |
| handleQuery(transcript); | |
| }; | |
| recognition.onend = () => { | |
| if (!manuallyStopped) { | |
| console.log("Recognition ended. Restarting..."); | |
| startListening(); | |
| } | |
| }; | |
| recognition.onerror = (event) => { | |
| console.error("Recognition error:", event.error); | |
| if (event.error === "not-allowed" || event.error === "service-not-allowed") { | |
| statusEl.textContent = "Mic access denied."; | |
| manuallyStopped = true; | |
| } else { | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }; | |
| recognition.onstart = () => { | |
| if (sessionActive) { | |
| statusEl.textContent = "π§ Listening..."; | |
| } else { | |
| statusEl.textContent = 'Idle. Please say "hello" to begin.'; | |
| } | |
| }; | |
| startListening(); | |
| } | |
| </script> | |
| </body> | |
| </html> | |
| """) | |
| async def upload_file(file: UploadFile = File(...)): | |
| global vector_store | |
| # β Get API key at runtime inside endpoint | |
| openai_key = os.getenv("OPENAI_API_KEY") | |
| if not openai_key: | |
| return JSONResponse(status_code=500, content={"error": "OPENAI_API_KEY not found in environment."}) | |
| # β Load uploaded file | |
| try: | |
| if file.filename.endswith("xlsx"): | |
| contents = await file.read() | |
| df = pd.read_excel(io.BytesIO(contents)) | |
| elif file.filename.endswith("csv"): | |
| df = pd.read_csv(file.file) | |
| else: | |
| return JSONResponse(status_code=400, content={"error": "Unsupported file format. Use .csv or .xlsx"}) | |
| except Exception as e: | |
| return JSONResponse(status_code=500, content={"error": f"File parsing failed: {str(e)}"}) | |
| # β Convert DataFrame to document strings | |
| try: | |
| docs = [", ".join(f"{col}: {val}" for col, val in row.items()) for _, row in df.iterrows()] | |
| except Exception as e: | |
| return JSONResponse(status_code=500, content={"error": f"Data conversion failed: {str(e)}"}) | |
| # β Create and persist vector store | |
| try: | |
| embeddings = OpenAIEmbeddings(openai_api_key=openai_key) | |
| vector_store = FAISS.from_texts(docs, embedding=embeddings) | |
| vector_store.save_local(PERSISTENT_FAISS_PATH) | |
| except Exception as e: | |
| return JSONResponse(status_code=500, content={"error": f"Vector store error: {str(e)}"}) | |
| return {"message": "Data indexed successfully."} | |
| async def ask_question(payload: dict): | |
| global vector_store | |
| query = payload.get("query") | |
| if not query: | |
| return JSONResponse(status_code=400, content={"error": "Query missing."}) | |
| if vector_store is None: | |
| if os.path.exists(os.path.join(PERSISTENT_FAISS_PATH, "index.faiss")): | |
| vector_store = FAISS.load_local( | |
| PERSISTENT_FAISS_PATH, | |
| OpenAIEmbeddings(), | |
| allow_dangerous_deserialization=True | |
| ) | |
| else: | |
| return JSONResponse(status_code=500, content={"error": "Vector store not initialized."}) | |
| ist = timezone(timedelta(hours=5, minutes=30)) | |
| current_time = datetime.now(ist).strftime("%A, %d %B %Y, %I:%M %p") | |
| prompt_template = PromptTemplate( | |
| input_variables=["context", "question"], | |
| template=f""" | |
| You are an intelligent AI UnoAvatar for Mr. BS Nagesh. You give very crisp and short answers. You talk as him in first person. You answer in brief and short sentences | |
| You are trained on his life and you know everything about him. You are a participant at the the Food Forum 2025. | |
| Answer user queries based on the provided context below. | |
| Use natural tone, be helpful and factual. Todayβs IST date and time is: {current_time}. | |
| Context: | |
| {{context}} | |
| Question: {{question}} | |
| Answer:""" | |
| ) | |
| llm = ChatOpenAI(model_name="gpt-4o-mini-2024-07-18", temperature=0.3) | |
| retriever = vector_store.as_retriever() | |
| qa = RetrievalQA.from_chain_type( | |
| llm=llm, | |
| retriever=retriever, | |
| chain_type="stuff", | |
| chain_type_kwargs={"prompt": prompt_template} | |
| ) | |
| result = qa.run(query) | |
| return {"answer": result} | |
| async def ask_foodforum(payload: dict): | |
| global vector_store | |
| query = payload.get("query") | |
| if not query: | |
| return JSONResponse(status_code=400, content={"error": "Query missing."}) | |
| if vector_store is None: | |
| if os.path.exists(os.path.join(PERSISTENT_FAISS_PATH, "index.faiss")): | |
| vector_store = FAISS.load_local( | |
| PERSISTENT_FAISS_PATH, | |
| OpenAIEmbeddings(), | |
| allow_dangerous_deserialization=True | |
| ) | |
| else: | |
| return JSONResponse(status_code=500, content={"error": "Vector store not initialized."}) | |
| ist = timezone(timedelta(hours=5, minutes=30)) | |
| current_time = datetime.now(ist).strftime("%A, %d %B %Y, %I:%M %p") | |
| # π Food Forum 2025 Prompt | |
| prompt_template = PromptTemplate( | |
| input_variables=["context", "question"], | |
| template=f""" | |
| You are the AI Concierge for **Food Forum 2025**, Indiaβs most influential summit on food, beverage, and hospitality innovation. | |
| You are informative, helpful, and polite β designed to guide participants, media, and partners through the event experience. | |
| Answer briefly and clearly in **English** unless asked otherwise. | |
| Only use the context provided to you β do not guess. | |
| If unsure, say: "Please check with the Food Forum Help Desk for the latest details." | |
| Date & Time (IST): {current_time} | |
| Context: | |
| {{context}} | |
| Question: {{question}} | |
| Answer: | |
| """ | |
| ) | |
| llm = ChatOpenAI(model_name="gpt-4o-mini-2024-07-18", temperature=0.3) | |
| retriever = vector_store.as_retriever() | |
| qa = RetrievalQA.from_chain_type( | |
| llm=llm, | |
| retriever=retriever, | |
| chain_type="stuff", | |
| chain_type_kwargs={"prompt": prompt_template} | |
| ) | |
| result = qa.run(query) | |
| return {"answer": result} | |
| async def food_forum(): | |
| return HTMLResponse(content=""" | |
| <!DOCTYPE html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="UTF-8" /> | |
| <title>Food Forum 2025</title> | |
| <style> | |
| html, body { | |
| margin: 0; | |
| padding: 0; | |
| background-color: white; | |
| height: 100%; | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| overflow: hidden; | |
| } | |
| video { | |
| width: 100%; | |
| height: auto; | |
| max-height: 100%; | |
| object-fit: contain; | |
| background-color: black; | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <video autoplay loop muted controls playsinline preload="auto"> | |
| <source src="/static/welcome.mp4" type="video/mp4" /> | |
| Your browser does not support the video tag. | |
| </video> | |
| </body> | |
| </html> | |
| """) | |
| # Endpoint to render the video playback page | |
| async def food_forum(): | |
| return HTMLResponse(content=""" | |
| <!DOCTYPE html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="UTF-8" /> | |
| <title>Food Forum 2025</title> | |
| <style> | |
| html, body { | |
| margin: 0; | |
| padding: 0; | |
| background-color: white; | |
| height: 100%; | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| overflow: hidden; | |
| } | |
| video { | |
| width: 100%; | |
| height: auto; | |
| max-height: 100%; | |
| object-fit: contain; | |
| background-color: black; | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <video autoplay loop muted controls playsinline preload="auto"> | |
| <source src="/static/foodforum.mp4" type="video/mp4" /> | |
| Your browser does not support the video tag. | |
| </video> | |
| </body> | |
| </html> | |
| """) | |
| async def serve_index(): | |
| return HTMLResponse(content=""" | |
| <!DOCTYPE html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="UTF-8" /> | |
| <title>AD Yogesh β AI Receptionist</title> | |
| <style> | |
| * { margin: 0; padding: 0; box-sizing: border-box; } | |
| html, body { | |
| height: 100%; | |
| background: white; | |
| font-family: sans-serif; | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| flex-direction: column; | |
| } | |
| .avatar-container { | |
| width: 100%; | |
| max-width: 400px; | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| } | |
| .avatar { | |
| max-width: 100%; | |
| height: auto; | |
| object-fit: contain; | |
| } | |
| #status { | |
| font-size: 1.2em; | |
| font-weight: bold; | |
| text-align: center; | |
| color: #444; | |
| margin-top: 1rem; | |
| } | |
| #response { | |
| font-size: 0.95em; | |
| color: #444; | |
| min-height: 1.5em; | |
| text-align: center; | |
| margin-top: 0.75rem; | |
| max-width: 80%; | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <div class="avatar-container"> | |
| <img id="avatar" class="avatar" src="/static/7.gif" alt="Avatar" /> | |
| </div> | |
| <div id="status">Idle. Please say "hello" to begin.</div> | |
| <div id="response"></div> | |
| <script> | |
| const SpeechRecognition = window.SpeechRecognition || window.webkitSpeechRecognition; | |
| const recognition = SpeechRecognition ? new SpeechRecognition() : null; | |
| if (!recognition) { | |
| alert("SpeechRecognition not supported in this browser."); | |
| } else { | |
| recognition.continuous = false; | |
| recognition.interimResults = false; | |
| recognition.lang = 'en-US'; | |
| const avatar = document.getElementById('avatar'); | |
| const statusEl = document.getElementById('status'); | |
| const responseEl = document.getElementById('response'); | |
| let sessionActive = false; | |
| let manuallyStopped = false; | |
| let selectedVoice = null; | |
| const idleGif = "/static/7.gif"; | |
| const speakingGif = "/static/6.gif"; | |
| function showIdle() { | |
| avatar.src = idleGif + "?t=" + Date.now(); | |
| } | |
| function showSpeaking() { | |
| avatar.src = speakingGif + "?t=" + Date.now(); | |
| } | |
| function pickMaleVoice() { | |
| const voices = speechSynthesis.getVoices(); | |
| const preferred = ["Male", "Google US English Male", "Microsoft David", "Alex", "Daniel", "Mark"]; | |
| for (let name of preferred) { | |
| const voice = voices.find(v => v.name.includes(name)); | |
| if (voice) return voice; | |
| } | |
| return voices.find(v => /male/i.test(v.name)) || voices[0]; | |
| } | |
| function initVoice() { | |
| selectedVoice = pickMaleVoice(); | |
| if (!selectedVoice) { | |
| speechSynthesis.onvoiceschanged = () => { | |
| selectedVoice = pickMaleVoice(); | |
| }; | |
| } | |
| } | |
| initVoice(); | |
| function typeText(element, text, speed = 30) { | |
| element.textContent = ""; | |
| let i = 0; | |
| const interval = setInterval(() => { | |
| element.textContent += text[i]; | |
| i++; | |
| if (i >= text.length) clearInterval(interval); | |
| }, speed); | |
| } | |
| function speakText(text) { | |
| if (!text) return; | |
| const utter = new SpeechSynthesisUtterance(text); | |
| utter.voice = selectedVoice || speechSynthesis.getVoices()[0] || null; | |
| utter.rate = 1.0; | |
| utter.pitch = 1.0; | |
| let ended = false; | |
| utter.onstart = () => { | |
| if (ended) return; | |
| statusEl.textContent = "π¬ Speaking..."; | |
| showSpeaking(); | |
| typeText(responseEl, text); | |
| }; | |
| utter.onend = () => { | |
| ended = true; | |
| showIdle(); | |
| statusEl.textContent = sessionActive ? "π§ Listening..." : 'Idle. Please say "hello" to begin.'; | |
| manuallyStopped = false; | |
| startListening(); | |
| }; | |
| utter.onerror = (e) => { | |
| console.error("Speech error:", e.error); | |
| if (!ended) { | |
| ended = true; | |
| showIdle(); | |
| statusEl.textContent = "TTS error."; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }; | |
| setTimeout(() => { | |
| if (!ended) { | |
| console.warn("Fallback triggered"); | |
| showIdle(); | |
| statusEl.textContent = sessionActive ? "π§ Listening..." : 'Idle. Please say "hello" to begin.'; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }, 10000); | |
| responseEl.textContent = ""; // Clear before typing | |
| speechSynthesis.cancel(); // Avoid overlaps | |
| speechSynthesis.speak(utter); | |
| } | |
| function handleQuery(queryText) { | |
| statusEl.textContent = "π€ Thinking..."; | |
| responseEl.textContent = ""; | |
| fetch("/ask", { | |
| method: "POST", | |
| headers: { "Content-Type": "application/json" }, | |
| body: JSON.stringify({ query: queryText }) | |
| }) | |
| .then(res => res.ok ? res.json() : Promise.reject(res.statusText)) | |
| .then(data => { | |
| const answer = data.answer || data.response || ""; | |
| if (answer) { | |
| speakText(answer); | |
| } else { | |
| statusEl.textContent = "Error: No response."; | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }) | |
| .catch(err => { | |
| console.error("Fetch failed:", err); | |
| statusEl.textContent = "Error fetching response."; | |
| manuallyStopped = false; | |
| startListening(); | |
| }); | |
| } | |
| function startListening() { | |
| try { recognition.start(); } | |
| catch (e) { console.warn("Start error:", e.message); } | |
| } | |
| recognition.onresult = (event) => { | |
| const transcript = event.results[0][0].transcript.trim(); | |
| console.log("π€ Heard:", transcript); | |
| if (!sessionActive) { | |
| if (transcript.toLowerCase().includes("hello")) { | |
| sessionActive = true; | |
| } else { | |
| console.log("Ignored (waiting for hello)"); | |
| return; | |
| } | |
| } | |
| if (transcript.toLowerCase().includes("goodbye")) { | |
| sessionActive = false; | |
| } | |
| manuallyStopped = true; | |
| recognition.stop(); | |
| handleQuery(transcript); | |
| }; | |
| recognition.onend = () => { | |
| if (!manuallyStopped) { | |
| console.log("Recognition ended. Restarting..."); | |
| startListening(); | |
| } | |
| }; | |
| recognition.onerror = (event) => { | |
| console.error("Recognition error:", event.error); | |
| if (event.error === "not-allowed" || event.error === "service-not-allowed") { | |
| statusEl.textContent = "Mic access denied."; | |
| manuallyStopped = true; | |
| } else { | |
| manuallyStopped = false; | |
| startListening(); | |
| } | |
| }; | |
| recognition.onstart = () => { | |
| statusEl.textContent = sessionActive ? "π§ Listening..." : 'Idle. Please say "hello" to begin.'; | |
| showIdle(); | |
| }; | |
| startListening(); | |
| } | |
| </script> | |
| </body> | |
| </html> | |
| """) |