app.py

import os
import requests
import json
import re
import io
import numpy as np
import wave
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
from pydub import AudioSegment

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
# ----------------------------------------------------
def clean_tts_text(text):
    """Remove markdown, HTML, emojis, code blocks, and extra symbols for clean TTS.
    Preserves expression tags like <breath>, <laugh>, <sigh>."""
    # Remove code blocks (```...```)
    text = re.sub(r'```[\s\S]*?```', '', text)
    # Remove inline code (`...`)
    text = re.sub(r'`[^`]*`', '', text)
    # Remove Markdown links [text](url)
    text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text)
    # Remove Markdown emphasis: **bold**, *italic*, __underline__, _italic_
    text = re.sub(r'(\*\*|__)(.*?)\1', r'\2', text)
    text = re.sub(r'(\*|_)(.*?)\1', r'\2', text)
    # Remove Markdown headers (#, ##, etc.)
    text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
    # Remove URLs
    text = re.sub(r'https?://[^\s]+', '', text)
    # Remove emojis and other non-printable symbols
    text = re.sub(r'[\u{1F600}-\u{1F64F}\u{1F300}-\u{1F5FF}\u{1F680}-\u{1F6FF}\u{2600}-\u{26FF}\u{2700}-\u{27BF}]', '', text)
    # Replace multiple newlines and spaces with single space
    text = re.sub(r'\s+', ' ', text).strip()
    # Limit length to avoid TTS overflow
    if len(text) > 1500:
        text = text[:1500] + '...'
    return text

def write_wav(data, sample_rate, output):
    """Write raw int16 data to a WAV file using wave module (no ffmpeg needed)."""
    with wave.open(output, 'wb') as wav_file:
        wav_file.setnchannels(1)          # mono
        wav_file.setsampwidth(2)          # 16-bit
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(data.tobytes())

@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. Check logs."}), 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')
        
    clean_text = clean_tts_text(text)
    
    if not clean_text.strip():
         return Response(json.dumps({"error": "No valid text left after cleaning."}), status=400, mimetype='application/json')

    try:
        lang_code = LANGUAGES.get(language_name, "en")
        # Fetch voice style directly each time to avoid stale cache
        style = tts.get_voice_style(voice_name=voice)
        if style is None:
            return Response(json.dumps({"error": f"Voice '{voice}' not available."}), status=400, mimetype='application/json')
        
        wav, duration = tts.synthesize(clean_text, voice_style=style, lang=lang_code)
        
        # Convert to int16 if needed
        if wav.dtype != np.int16:
            max_val = np.max(np.abs(wav))
            if max_val > 0:
                wav_int16 = np.int16(wav / max_val * 32767)
            else:
                wav_int16 = wav.astype(np.int16)
        else:
            wav_int16 = wav

        sample_rate = 22050
        
        # Write WAV directly to buffer using wave module (no ffmpeg)
        wav_buffer = io.BytesIO()
        write_wav(wav_int16, sample_rate, wav_buffer)
        wav_buffer.seek(0)
        
        return Response(wav_buffer.read(), mimetype="audio/wav")
    except Exception as e:
        # Log the error for debugging
        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)