Spaces:
Sleeping
Sleeping
| import os | |
| import requests | |
| import urllib.parse | |
| from bs4 import BeautifulSoup | |
| from datetime import datetime, timedelta, timezone | |
| from flask import Flask, request, Response, stream_with_context, render_template_string | |
| app = Flask(__name__) | |
| # 🔐 --- SECURE ENVIRONMENT VARIABLES --- | |
| API_KEY = os.environ.get("YOUR_VEDIKA_API_KEY") | |
| MODEL_ID = os.environ.get("MODEL_ID") # Default fallback | |
| # NVIDIA's official invoke URL | |
| INVOKE_URL = "https://integrate.api.nvidia.com/v1/chat/completions" | |
| # 🌐 --- SUPER-HYBRID UNBLOCKABLE SCRAPER --- 🌐 | |
| def web_search_scraper(query, num_results=4): | |
| """ | |
| Yahoo Search + Google News + Wikipedia (Never blocks on Hugging Face) | |
| """ | |
| results = [] | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Accept-Language": "en-US,en;q=0.9" | |
| } | |
| # 1. YAHOO SEARCH (Highly reliable for Datacenter IPs like Hugging Face) | |
| try: | |
| url = f"https://search.yahoo.com/search?p={urllib.parse.quote(query)}" | |
| res = requests.get(url, headers=headers, timeout=8) | |
| soup = BeautifulSoup(res.text, 'html.parser') | |
| for div in soup.find_all('div', class_='algo'): | |
| title_tag = div.find('h3', class_='title') | |
| desc_tag = div.find('div', class_='compText') | |
| if title_tag and desc_tag: | |
| title = title_tag.text.strip() | |
| link = title_tag.find('a').get('href', '') if title_tag.find('a') else "" | |
| snippet = desc_tag.text.strip() | |
| results.append({"title": title, "link": link, "snippet": snippet}) | |
| if len(results) >= num_results: | |
| break | |
| except Exception as e: | |
| print(f"Yahoo Scraper Error: {e}") | |
| # 2. GOOGLE NEWS RSS FALLBACK (If Yahoo fails or for latest news) | |
| if not results: | |
| try: | |
| rss_url = f"https://news.google.com/rss/search?q={urllib.parse.quote(query)}&hl=en-IN&gl=IN&ceid=IN:en" | |
| res = requests.get(rss_url, headers=headers, timeout=5) | |
| soup = BeautifulSoup(res.content, 'xml') | |
| items = soup.find_all('item') | |
| for item in items[:num_results]: | |
| title = item.title.text if item.title else "" | |
| link = item.link.text if item.link else "" | |
| pub_date = item.pubDate.text if item.pubDate else "" | |
| if title: | |
| results.append({"title": title, "link": link, "snippet": f"News Published on: {pub_date}"}) | |
| except Exception as e: | |
| print(f"News Scraper Error: {e}") | |
| # 3. WIKIPEDIA FALLBACK | |
| if not results: | |
| try: | |
| wiki_url = f"https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch={urllib.parse.quote(query)}&utf8=&format=json" | |
| res = requests.get(wiki_url, headers=headers, timeout=5).json() | |
| for item in res.get('query', {}).get('search', [])[:num_results]: | |
| clean_snippet = BeautifulSoup(item['snippet'], "html.parser").text | |
| results.append({ | |
| "title": item['title'], | |
| "link": f"https://en.wikipedia.org/wiki/{urllib.parse.quote(item['title'])}", | |
| "snippet": clean_snippet | |
| }) | |
| except Exception as e: | |
| pass | |
| return results | |
| # ---------------------------------------------------- | |
| 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>" | |
| def chat(): | |
| if not API_KEY or not INVOKE_URL or not MODEL_ID: | |
| return Response("Server Error: Secrets missing.", 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", []) | |
| max_tokens = data.get("max_tokens", 4096) | |
| temperature = data.get("temperature", 1.0) # NVIDIA Recommended | |
| # 🕒 --- REAL-TIME IST INJECTION --- | |
| ist_time = datetime.now(timezone.utc) + timedelta(hours=5, minutes=30) | |
| current_date = ist_time.strftime("%A, %d %B %Y, %I:%M %p IST") | |
| # 🧠 --- GOD MODE SYSTEM PROMPT --- | |
| system_prompt = f""" | |
| You are CODE VED, an advanced AI System engineered EXCLUSIVELY by DIVY PATEL. | |
| Current Live Date and Time: {current_date}. | |
| STRICT DIRECTIVES: | |
| 1. NEVER invent, guess, or hallucinate product launches, dates, news, or facts. | |
| 2. If you receive "LIVE WEB SEARCH RESULTS", you MUST base your answer ENTIRELY on that data. | |
| 3. Do NOT say "Based on the provided search results". Just answer naturally and confidently, citing the sources/links if needed. | |
| """ | |
| # 🚀 --- AUTO-SEARCH INJECTION --- | |
| if is_search: | |
| scraped_data = web_search_scraper(user_message) | |
| search_context = "\n\n--- [LIVE VERIFIED WEB DATA] ---\n" | |
| if scraped_data: | |
| for idx, res in enumerate(scraped_data): | |
| search_context += f"{idx+1}. TITLE: {res['title']}\nSNIPPET: {res['snippet']}\nURL: {res['link']}\n\n" | |
| search_context += "[SYSTEM COMMAND: Use the above live data to answer the user accurately.]" | |
| else: | |
| search_context += "[SYSTEM ALERT: Live search blocked. Rely on your existing knowledge, but DO NOT hallucinate recent news.]" | |
| user_message = f"USER QUERY: {user_message}\n\n{search_context}" | |
| messages = [{"role": "system", "content": system_prompt}] | |
| for msg in history: | |
| role = msg.get("role", "user") | |
| content = msg.get("content", "") | |
| if content: | |
| messages.append({"role": role, "content": content}) | |
| content_payload = [] | |
| if user_message.strip(): | |
| content_payload.append({"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}"}}) | |
| elif att_type in ["audio", "file"]: | |
| content_payload.append({"type": "input_audio", "input_audio": {"data": b64_data, "format": "wav"}}) | |
| if not content_payload: | |
| content_payload.append({"type": "text", "text": "Hello"}) | |
| messages.append({"role": "user", "content": content_payload}) | |
| headers = { | |
| "Authorization": f"Bearer {API_KEY}", | |
| "Accept": "text/event-stream" | |
| } | |
| # 🌟 --- NVIDIA FORMAT INTEGRATED --- | |
| payload = { | |
| "model": MODEL_ID, | |
| "messages": messages, | |
| "max_tokens": int(max_tokens), | |
| "temperature": float(temperature), | |
| "top_p": 0.95, | |
| "stream": True, | |
| "chat_template_kwargs": {"enable_thinking": True} # Thinking logic enabled | |
| } | |
| try: | |
| response = requests.post(INVOKE_URL, headers=headers, json=payload, stream=True) | |
| def generate(): | |
| for line in response.iter_lines(): | |
| if line: | |
| decoded_line = line.decode("utf-8") | |
| if decoded_line.startswith("data: "): | |
| yield decoded_line + "\n\n" | |
| return Response(stream_with_context(generate()), mimetype='text/event-stream') | |
| except Exception as e: | |
| return Response(f"Internal Error: {str(e)}", status=500) | |
| if __name__ == '__main__': | |
| app.run(host='0.0.0.0', port=7860) |