Upload telegram_bot.py with huggingface_hub

telegram_bot.py

@@ -0,0 +1,138 @@
+"""
+Standalone Telegram Bot Worker for Kora AI Ecosystem.
+Connects Telegram users directly to Kora Chat, Kora Coder 0.5B, and Kora Image Studio.
+
+Usage:
+  1. Get a bot token from @BotFather on Telegram.
+  2. Set TELEGRAM_BOT_TOKEN="your_bot_token_here"
+  3. Run: python telegram_bot.py
+"""
+
+import os
+import sys
+import time
+import json
+import urllib.parse
+import urllib.request
+from threading import Thread
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+# Kora Gateway Endpoint
+KORA_ENDPOINT = os.environ.get("KORA_ENDPOINT", os.environ.get("KORA_CUSTOM_ENDPOINT", "https://professorceo-kora-ai.hf.space")).rstrip("/")
+BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
+
+def send_telegram_message(chat_id: int, text: str):
+    if not BOT_TOKEN:
+        return
+    url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
+    try:
+        req_data = json.dumps({"chat_id": chat_id, "text": text, "parse_mode": "Markdown"}).encode("utf-8")
+        req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
+        urllib.request.urlopen(req, timeout=10)
+    except Exception:
+        try:
+            req_data = json.dumps({"chat_id": chat_id, "text": text}).encode("utf-8")
+            req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
+            urllib.request.urlopen(req, timeout=10)
+        except Exception as e:
+            print(f"❌ Error sending Telegram message: {e}", flush=True)
+
+def send_telegram_photo(chat_id: int, photo_url: str, caption: str = ""):
+    if not BOT_TOKEN:
+        return
+    url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendPhoto"
+    try:
+        req_data = json.dumps({"chat_id": chat_id, "photo": photo_url, "caption": caption}).encode("utf-8")
+        req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
+        urllib.request.urlopen(req, timeout=15)
+    except Exception:
+        send_telegram_message(chat_id, f"🎨 Image generated: {photo_url}")
+
+def call_kora_chat(prompt: str, model: str = "kora-chat-0.5b") -> str:
+    url = f"{KORA_ENDPOINT}/v1/chat/completions"
+    try:
+        req_data = json.dumps({
+            "model": model,
+            "messages": [{"role": "user", "content": prompt}],
+            "max_tokens": 512
+        }).encode("utf-8")
+        req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+            return data["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"Error connecting to Kora AI: {e}"
+
+def call_kora_image(prompt: str) -> str:
+    url = f"{KORA_ENDPOINT}/v1/images/generations"
+    try:
+        req_data = json.dumps({"prompt": prompt, "size": "1024x1024"}).encode("utf-8")
+        req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+            return data["data"][0]["url"]
+    except Exception:
+        clean = urllib.parse.quote(prompt + ", masterpiece, pristine quality, vector asset, no watermark, no logo, no text")
+        return f"https://image.pollinations.ai/prompt/{clean}?width=1024&height=1024&nologo=true&nofeed=true&model=flux"
+
+def run_bot():
+    if not BOT_TOKEN:
+        print("❌ TELEGRAM_BOT_TOKEN environment variable is missing!", flush=True)
+        print("Please set TELEGRAM_BOT_TOKEN='your_token_here' and re-run.", flush=True)
+        sys.exit(1)
+
+    print("=" * 70, flush=True)
+    print("🤖 KORA TELEGRAM BOT WORKER STARTED", flush=True)
+    print(f"📡 Target Kora Gateway: {KORA_ENDPOINT}", flush=True)
+    print("=" * 70, flush=True)
+
+    offset = 0
+    base_url = f"https://api.telegram.org/bot{BOT_TOKEN}"
+
+    while True:
+        try:
+            get_url = f"{base_url}/getUpdates?offset={offset}&timeout=20"
+            req = urllib.request.Request(get_url)
+            with urllib.request.urlopen(req, timeout=25) as resp:
+                data = json.loads(resp.read().decode("utf-8"))
+                for result in data.get("result", []):
+                    offset = max(offset, result["update_id"] + 1)
+                    message = result.get("message", {})
+                    text = message.get("text", "").strip()
+                    chat_id = message.get("chat", {}).get("id")
+
+                    if not chat_id or not text:
+                        continue
+
+                    print(f"📩 Telegram User [{chat_id}]: {text}", flush=True)
+
+                    if text in ["/start", "/help"]:
+                        send_telegram_message(
+                            chat_id,
+                            "🧠 *Welcome to Kora AI on Telegram!* (Powered by Kontyra)\n\n"
+                            "I am your autonomous AI assistant. Commands:\n"
+                            "• Send any question to chat with **Kora Chat**.\n"
+                            "• `/code <prompt>` for code generation.\n"
+                            "• `/image <prompt>` to generate 1024x1024 watermark-free images."
+                        )
+                    elif text.startswith("/image") or text.lower().startswith("create image"):
+                        prompt = text.replace("/image", "").replace("create image", "").strip() or "Futuristic Kontyra cyber emblem"
+                        send_telegram_message(chat_id, f"🎨 Generating zero-watermark image for: *{prompt}*...")
+                        img_url = call_kora_image(prompt)
+                        send_telegram_photo(chat_id, img_url, f"✨ {prompt}")
+                    elif text.startswith("/code"):
+                        prompt = text.replace("/code", "").strip()
+                        send_telegram_message(chat_id, "💻 Generating code with Kora Coder...")
+                        reply = call_kora_chat(prompt, model="kora-coder-0.5b")
+                        send_telegram_message(chat_id, f"```\n{reply}\n```")
+                    else:
+                        reply = call_kora_chat(text, model="kora-chat-0.5b")
+                        send_telegram_message(chat_id, reply)
+
+        except Exception as e:
+            time.sleep(3)
+
+if __name__ == "__main__":
+    run_bot()