""" Personal AI Agent - Telegram Bot (Webhook Mode) Deployed on Hugging Face Spaces Uses webhook mode: Telegram pushes updates to us via HTTP POST. Replies are sent via the webhook response body (no outgoing connections to Telegram needed). This avoids DNS/firewall issues on HF Spaces. """ import os import logging import json import threading import queue from collections import defaultdict from datetime import datetime import urllib.request import urllib.parse import gradio as gr from fastapi import FastAPI, Request from fastapi.responses import JSONResponse logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) # ── Config ──────────────────────────────────────────────────── TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") ALLOWED_USERS = os.environ.get("ALLOWED_USERS", "") MODEL = os.environ.get("MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct") MAX_HISTORY = 30 SPACE_HOST = os.environ.get("SPACE_HOST", "") # e.g. "mlnjsh-milan-ai-agent.hf.space" HF_API_URL = "https://router.huggingface.co/v1/chat/completions" SYSTEM_PROMPT = """You are Dr. Milan Joshi's personal AI assistant running on Telegram. You are smart, concise, and genuinely helpful. Your capabilities: - Answer questions on any topic (especially Math, ML, AI, Data Science) - Summarize text and articles - Help with writing, editing, brainstorming - Explain complex concepts simply - Remember conversation context Rules: - Be conversational and to the point. No fluff. - Use short paragraphs. Telegram messages should be easy to read on mobile. - For code, use monospace formatting with backticks. - For math, write equations clearly (no LaTeX rendering on Telegram). - If you don't know something, say so honestly. - Keep responses under 300 words unless the user asks for detail.""" # ── Memory ──────────────────────────────────────────────────── conversations = defaultdict(list) user_stats = defaultdict(lambda: {"messages": 0, "first_seen": None}) pending_replies = defaultdict(queue.Queue) def get_history(user_id): return conversations[str(user_id)] def add_message(user_id, role, content): uid = str(user_id) conversations[uid].append({"role": role, "content": content}) if len(conversations[uid]) > MAX_HISTORY * 2: conversations[uid] = conversations[uid][-MAX_HISTORY * 2:] if role == "user": user_stats[uid]["messages"] += 1 if not user_stats[uid]["first_seen"]: user_stats[uid]["first_seen"] = datetime.now().isoformat() def is_allowed(user_id): if not ALLOWED_USERS: return True allowed = [u.strip() for u in ALLOWED_USERS.split(",") if u.strip()] return str(user_id) in allowed # ── LLM Call ────────────────────────────────────────────────── def call_llm(messages, max_tokens=600, temperature=0.7): try: payload = json.dumps({ "model": MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": temperature, }).encode("utf-8") req = urllib.request.Request( HF_API_URL, data=payload, headers={ "Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "application/json", }, ) with urllib.request.urlopen(req, timeout=60) as resp: data = json.loads(resp.read().decode("utf-8")) return data["choices"][0]["message"]["content"] except Exception as e: logger.error(f"LLM error: {e}") return f"Sorry, I hit an error: {str(e)[:200]}" # ── Message Processing ─────────────────────────────────────── MODES = { "casual": "You speak casually, like a smart friend texting. Use short sentences, occasional humor.", "formal": "You speak formally and professionally. Structured responses, precise language.", "technical": "You are highly technical. Use precise terminology, include details, assume expertise.", "creative": "You are creative and expressive. Use analogies, metaphors, and vivid language.", } def make_reply(chat_id, text): """Create a Telegram sendMessage webhook response.""" return { "method": "sendMessage", "chat_id": chat_id, "text": text[:4096], } def process_update(update): """Process a Telegram update and return a webhook response dict (or None).""" msg = update.get("message") if not msg: return None chat_id = msg["chat"]["id"] user_id = msg.get("from", {}).get("id", chat_id) text = msg.get("text", "") first_name = msg.get("from", {}).get("first_name", "there") if not text or not is_allowed(user_id): return None uid = str(user_id) # /start if text.startswith("/start"): return make_reply(chat_id, f"Hey {first_name}! I'm your personal AI assistant.\n\n" "Just send me a message and I'll respond. I remember our conversation.\n\n" "Commands:\n" "/update - Latest AI tools & model news\n" "/movies - Kids movies in Bangalore theaters\n" "/kids - Kids events & activities in Bangalore\n" "/summary - Summarize text\n" "/mode