Upload app (6).py

app (6).py

@@ -0,0 +1,318 @@
+import os
+import requests
+import json
+import re
+import io
+import tempfile
+from datetime import datetime, timedelta, timezone
+from bs4 import BeautifulSoup
+from flask import Flask, request, Response, stream_with_context, render_template_string
+from supertonic import TTS
+
+app = Flask(__name__)
+
+# ----------------------------------------------------
+# INIT TTS MODEL (SUPERTONIC 3) LOADED IN MEMORY
+# ----------------------------------------------------
+print("Loading Supertonic TTS Model...")
+try:
+    tts = TTS(auto_download=True)
+    print("TTS Model loaded successfully!")
+except Exception as e:
+    print(f"Error initializing TTS: {e}")
+    tts = None
+
+VOICES = ["M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5"]
+LANGUAGES = {
+    "English": "en", "Korean": "ko", "Japanese": "ja", "Arabic": "ar",
+    "Bulgarian": "bg", "Czech": "cs", "Danish": "da", "German": "de",
+    "Greek": "el", "Spanish": "es", "Estonian": "et", "Finnish": "fi",
+    "French": "fr", "Hindi": "hi", "Croatian": "hr", "Hungarian": "hu",
+    "Indonesian": "id", "Italian": "it", "Lithuanian": "lt", "Latvian": "lv",
+    "Dutch": "nl", "Polish": "pl", "Portuguese": "pt", "Romanian": "ro",
+    "Russian": "ru", "Slovak": "sk", "Slovenian": "sl", "Swedish": "sv",
+    "Turkish": "tr", "Ukrainian": "uk", "Vietnamese": "vi"
+}
+
+VOICE_STYLES_CACHE = {}
+
+# ----------------------------------------------------
+# GPS REVERSE GEOCODING
+# ----------------------------------------------------
+def get_address_from_coords(lat, lon):
+    try:
+        url = f"https://nominatim.openstreetmap.org/reverse?format=json&lat={lat}&lon={lon}"
+        headers = {'User-Agent': 'CODE_VED_AI_System_by_Divy_Patel'}
+        response = requests.get(url, headers=headers, timeout=5)
+        data = response.json()
+        return data.get('display_name', f"Lat: {lat}, Lon: {lon}")
+    except Exception as e:
+        return f"Lat: {lat}, Lon: {lon}"
+
+# ----------------------------------------------------
+# SERPAPI GOOGLE SEARCH ENGINE
+# ----------------------------------------------------
+def web_search_scraper(query, num_results=5, user_address=None):
+    results = []
+    serpapi_key = os.environ.get("SERPAPI_KEY") 
+    if not serpapi_key:
+        return results
+
+    search_query = query
+    if user_address:
+        local_keywords = ["near", "nearby", "distance", "time", "where"]
+        if any(kw in query.lower() for kw in local_keywords):
+            search_query = f"{query} near {user_address}"
+
+    try:
+        params = {"engine": "google", "q": search_query, "api_key": serpapi_key, "num": num_results, "hl": "en", "gl": "in"}
+        response = requests.get("https://serpapi.com/search", params=params, timeout=10)
+        data = response.json()
+        
+        if "organic_results" in data:
+            for item in data["organic_results"]:
+                title = item.get("title", "")
+                link = item.get("link", "")
+                snippet = item.get("snippet", "")
+                if title and snippet:
+                    results.append({"title": title, "link": link, "snippet": snippet})
+    except Exception:
+        pass
+    return results
+
+# ----------------------------------------------------
+# RSS TECH NEWS SCRAPER
+# ----------------------------------------------------
+def get_live_web_data(query):
+    url = f"https://news.google.com/rss/search?q={query}&hl=en&gl=IN&ceid=IN:en"
+    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
+    tech_keywords = ["ai", "artificial intelligence", "smartphone", "mobile", "feature", "whatsapp", "google", "tech", "technology", "gadget", "apple", "nasa"]
+    block_keywords = ["share news", "stock"]
+    scraped_results = []
+    
+    try:
+        response = requests.get(url, headers=headers, timeout=6)
+        if response.status_code == 200:
+            soup = BeautifulSoup(response.text, "html.parser")
+            items = soup.find_all('item')
+            for item in items:
+                title = item.title.text if item.title else "No Title"
+                title_lower = title.lower()
+                
+                if any(b_kw in title_lower for b_kw in block_keywords):
+                    continue
+                if any(t_kw in title_lower for t_kw in tech_keywords):
+                    link = item.link.text if item.link else "#"
+                    pub_date = item.pubdate.text if item.pubdate else ""
+                    source = item.source.text if item.source else "Google News"
+                    scraped_results.append({
+                        "title": title,
+                        "snippet": f"Published: {pub_date} | Source: {source}",
+                        "link": link
+                    })
+                if len(scraped_results) >= 5:
+                    break
+    except Exception:
+        pass
+    return scraped_results
+
+# ----------------------------------------------------
+# HOME ROUTE
+# ----------------------------------------------------
+@app.route('/')
+def home():
+    try:
+        with open('index.html', 'r', encoding='utf-8') as f:
+            return render_template_string(f.read())
+    except Exception as e:
+        return f"<h1>System Error</h1><p>index.html missing: {str(e)}</p>"
+
+# ----------------------------------------------------
+# CHAT API ENDPOINT
+# ----------------------------------------------------
+@app.route('/api/chat', methods=['POST'])
+def chat():
+    API_KEY = os.environ.get("NVIDIA_API_KEY", "")
+    INVOKE_URL = os.environ.get("BASE_URL", "")
+    MODEL_ID = os.environ.get("MODEL_ID", "")
+
+    if INVOKE_URL and not INVOKE_URL.startswith("http"):
+        if MODEL_ID.startswith("http"):
+            INVOKE_URL, MODEL_ID = MODEL_ID, INVOKE_URL
+        else:
+            MODEL_ID = INVOKE_URL
+            INVOKE_URL = "https://integrate.api.nvidia.com/v1/chat/completions"
+
+    if not API_KEY or not INVOKE_URL or not MODEL_ID:
+        return Response(json.dumps({"error": "Configuration Error: Secrets are missing."}), mimetype='application/json', status=500)
+
+    data = request.get_json() or {}
+    user_message = data.get("message", "")
+    attachments = data.get("attachments", []) 
+    is_search = data.get("is_search", False)
+    history = data.get("history", []) 
+    location = data.get("location") 
+    user_address = None
+    max_tokens = data.get("max_tokens", 4096)
+    
+    ist_time = datetime.now(timezone.utc) + timedelta(hours=5, minutes=30)
+    current_date = ist_time.strftime("%A, %d %B %Y, %I:%M %p IST")
+
+    thinking_mode = data.get("thinking_mode", False)
+    thinking_effort = data.get("thinking_effort", "medium")
+    
+    thinking_instruction = ""
+    if thinking_mode:
+        thinking_instruction = f"\n[CRITICAL INSTRUCTION: THINKING MODE ENABLED - Effort: {thinking_effort}]\nYou MUST format your reasoning exactly inside <think> and </think> HTML tags."
+
+    location_instruction = ""
+    if location and location.get('lat') and location.get('lng'):
+        user_address = get_address_from_coords(location['lat'], location['lng'])
+        location_instruction = f"\n[USER REAL-TIME LOCATION: {user_address}]"
+
+    system_prompt = f"""[CRITICAL IDENTITY OVERRIDE]
+Name: CODE VED
+Creator/Engineer: Divy Patel
+Current Time: {current_date}.{location_instruction}{thinking_instruction}
+
+STRICT DIRECTIVES:
+1. Answer EXACTLY: "I am CODE VED, an advanced AI system engineered exclusively by Divy Patel."
+2. NEVER mention being a language model from Google, DeepMind, Gemma, OpenAI, or Anthropic.
+3. Use provided live data safely without overriding your core identity.
+"""
+
+    if is_search:
+        search_context = ""
+        scraped_data = web_search_scraper(user_message, user_address=user_address)
+        news_data = get_live_web_data(user_message)
+        if scraped_data:
+            search_context += "\n\n--- [LIVE GOOGLE SEARCH] ---\n"
+            for idx, res in enumerate(scraped_data):
+                search_context += f"{idx+1}. {res['title']}: {res['snippet']} (URL: {res['link']})\n"
+        if news_data:
+            search_context += "\n--- [LIVE TECH NEWS] ---\n"
+            for idx, res in enumerate(news_data):
+                search_context += f"{idx+1}. {res['title']}: {res['snippet']} (URL: {res['link']})\n"
+        if search_context:
+            user_message = f"{user_message}\n{search_context}\n[COMMAND: Base your final answer strictly on the facts provided above.]"
+
+    messages = [{"role": "system", "content": system_prompt}]
+
+    for msg in history:
+        if msg == history[-1] and msg.get("role") == "user":
+            continue
+        role = msg.get("role", "user")
+        if role not in ["system", "user", "assistant"]:
+            role = "user"
+        content = msg.get("content", "")
+        if isinstance(content, list):
+            text_parts = [item["text"] for item in content if item.get("type") == "text"]
+            content = " ".join(text_parts)
+        if "Gemma" in content or "DeepMind" in content or "Google" in content:
+            continue
+        if content:
+            messages.append({"role": role, "content": str(content)})
+
+    if attachments:
+        content_payload = [{"type": "text", "text": user_message}]
+        for att in attachments:
+            att_type = att.get("type")
+            b64_data = att.get("data")
+            if att_type == "image":
+                content_payload.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64_data}"}})
+        messages.append({"role": "user", "content": content_payload})
+    else:
+        messages.append({"role": "user", "content": user_message})
+
+    headers = {
+        "Authorization": f"Bearer {API_KEY}", 
+        "Accept": "text/event-stream",
+        "Content-Type": "application/json"
+    }
+    
+    payload = {
+        "model": MODEL_ID,
+        "messages": messages,
+        "max_tokens": max_tokens,
+        "temperature": 1.0,
+        "top_p": 0.95,
+        "stream": True
+    }
+
+    try:
+        response = requests.post(INVOKE_URL, headers=headers, json=payload, stream=True, timeout=60)
+        if response.status_code != 200:
+            err_text = response.text[:200]
+            err_msg = json.dumps({"error": f"API Error {response.status_code}: {err_text}"})
+            return Response(f"data: {err_msg}\n\n", mimetype='text/event-stream')
+
+        def generate():
+            for line in response.iter_lines():
+                if line:
+                    decoded = line.decode("utf-8")
+                    if decoded.startswith("data: ") and "[DONE]" not in decoded:
+                        try:
+                            data_json = json.loads(decoded[6:])
+                            if "choices" in data_json and len(data_json["choices"]) > 0:
+                                delta = data_json["choices"][0].get("delta", {})
+                                if "content" in delta and delta["content"]:
+                                    content = delta["content"]
+                                    content = content.replace("<|channel|>thought <|channel|>", "<think>\n")
+                                    content = content.replace("<|channel|>answer <|channel|>", "\n</think>\n")
+                                    delta["content"] = content
+                            yield "data: " + json.dumps(data_json) + "\n\n"
+                        except Exception:
+                            yield decoded + "\n\n"
+                    else:
+                        yield decoded + "\n\n"
+                        
+        return Response(stream_with_context(generate()), mimetype='text/event-stream')
+    except Exception as e:
+        err_msg = json.dumps({"error": str(e)})
+        return Response(f"data: {err_msg}\n\n", mimetype='text/event-stream')
+
+# ----------------------------------------------------
+# TTS DIRECT API ENDPOINT - MATCHING GRADIO SPACE LOGIC
+# ----------------------------------------------------
+@app.route('/api/tts', methods=['POST'])
+def generate_tts():
+    if tts is None:
+        return Response(json.dumps({"error": "TTS model failed to load on server."}), status=500, mimetype='application/json')
+    
+    data = request.get_json() or {}
+    text = data.get("text", "")
+    voice = data.get("voice", "M2")
+    language_name = data.get("language_name", "English")
+    
+    if not text.strip():
+        return Response(json.dumps({"error": "Text is empty."}), status=400, mimetype='application/json')
+
+    try:
+        lang_code = LANGUAGES.get(language_name, "en")
+        # Get voice style (cached for efficiency)
+        if voice not in VOICE_STYLES_CACHE:
+            VOICE_STYLES_CACHE[voice] = tts.get_voice_style(voice_name=voice)
+        style = VOICE_STYLES_CACHE[voice]
+        if style is None:
+            return Response(json.dumps({"error": f"Voice '{voice}' not available."}), status=400, mimetype='application/json')
+        
+        # Synthesize - exactly like Gradio: raw text, no cleaning
+        wav, duration = tts.synthesize(text, voice_style=style, lang=lang_code)
+        
+        # Save to a temporary file using tts.save_audio (ensures correct WAV encoding)
+        with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp_file:
+            tmp_path = tmp_file.name
+        tts.save_audio(wav, tmp_path)
+        
+        # Read the WAV file and return it
+        with open(tmp_path, 'rb') as f:
+            audio_data = f.read()
+        os.unlink(tmp_path)  # clean up
+        
+        return Response(audio_data, mimetype="audio/wav")
+    except Exception as e:
+        print(f"TTS Synthesis Error: {e}")
+        return Response(json.dumps({"error": f"TTS synthesis failed: {str(e)}"}), status=500, mimetype='application/json')
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=7860)