Upload telegram_bot.py with huggingface_hub

telegram_bot.py

@@ -102,10 +102,58 @@ def test_telegram_connectivity() -> bool:
 def send_telegram_message(chat_id: int, text: str):
     send_telegram_request("sendMessage", {"chat_id": chat_id, "text": text})
 
-def send_telegram_photo(chat_id: int, photo_url: str, caption: str = ""):
-    res = send_telegram_request("sendPhoto", {"chat_id": chat_id, "photo": photo_url, "caption": caption}, timeout_sec=15.0)
+def send_telegram_photo(chat_id: int, photo_path_or_url: str, caption: str = ""):
+    if not BOT_TOKEN or not photo_path_or_url:
+        return
+
+    # 1. If photo_path_or_url is a local file or relative path, upload binary file bytes directly
+    local_file = photo_path_or_url
+    if not os.path.exists(local_file) and "generated_assets" in photo_path_or_url:
+        filename = photo_path_or_url.split("/")[-1]
+        potential = os.path.join("public", "generated_assets", filename)
+        if os.path.exists(potential):
+            local_file = potential
+
+    if os.path.exists(local_file):
+        try:
+            import mimetypes, uuid
+            boundary = f"----FormBoundary{uuid.uuid4().hex}"
+            
+            with open(local_file, "rb") as f:
+                img_bytes = f.read()
+                
+            filename = os.path.basename(local_file)
+            mime_type = mimetypes.guess_type(filename)[0] or "image/png"
+            
+            body = bytearray()
+            # chat_id field
+            body.extend(f"--{boundary}\r\nContent-Disposition: form-data; name=\"chat_id\"\r\n\r\n{chat_id}\r\n".encode("utf-8"))
+            # caption field
+            if caption:
+                body.extend(f"--{boundary}\r\nContent-Disposition: form-data; name=\"caption\"\r\n\r\n{caption}\r\n".encode("utf-8"))
+            # photo file field
+            body.extend(f"--{boundary}\r\nContent-Disposition: form-data; name=\"photo\"; filename=\"{filename}\"\r\nContent-Type: {mime_type}\r\n\r\n".encode("utf-8"))
+            body.extend(img_bytes)
+            body.extend(f"\r\n--{boundary}--\r\n".encode("utf-8"))
+            
+            headers = {
+                "Content-Type": f"multipart/form-data; boundary={boundary}",
+                "User-Agent": "KoraBot/1.0 (Kontyra AI Ecosystem)"
+            }
+            
+            direct_url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendPhoto"
+            req = urllib.request.Request(direct_url, data=body, headers=headers)
+            with urllib.request.urlopen(req, timeout=25) as resp:
+                res = json.loads(resp.read().decode("utf-8"))
+                if res.get("ok"):
+                    return
+        except Exception as e:
+            print(f"⚠️ Binary photo upload notice: {e}", flush=True)
+
+    # 2. Fallback to URL sendPhoto
+    res = send_telegram_request("sendPhoto", {"chat_id": chat_id, "photo": photo_path_or_url, "caption": caption}, timeout_sec=15.0)
     if not res or not res.get("ok"):
-        send_telegram_message(chat_id, f"🎨 Image generated: {photo_url}")
+        send_telegram_message(chat_id, f"🎨 Image generated: {photo_path_or_url}")
 
 def call_kora_chat(prompt: str, model: str = "kora-chat-0.5b") -> str:
     # 1. Try direct in-memory generation if running within gateway process
@@ -161,13 +209,13 @@ def call_kora_chat(prompt: str, model: str = "kora-chat-0.5b") -> str:
             continue
     return "Hello! I am Kora AI. How can I assist you today?"
 
-def call_kora_image(prompt: str) -> str:
+def call_kora_image(prompt: str) -> dict:
     # 1. Try direct in-memory local PyTorch Stable Diffusion generation
     try:
         from vision.image_generator import generate_image
         res = generate_image(prompt)
         if res and res.get("success"):
-            return res.get("direct_url", "")
+            return res
     except Exception as err:
         print(f"ℹ️ In-memory image generation notice: {err}", flush=True)
 
@@ -178,10 +226,10 @@ def call_kora_image(prompt: str) -> str:
         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"]
+            return {"direct_url": data["data"][0]["url"]}
     except Exception:
         pass
-    return ""
+    return {}
 
 def handle_update(update: dict, token: str = None):
     global BOT_TOKEN
@@ -218,8 +266,11 @@ def handle_update(update: dict, token: str = None):
             )
             return
         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}")
+        img_res = call_kora_image(prompt)
+        file_path = img_res.get("file_path", "")
+        direct_url = img_res.get("direct_url", "")
+        photo_target = file_path if (file_path and os.path.exists(file_path)) else direct_url
+        send_telegram_photo(chat_id, photo_path_or_url=photo_target, caption=f"✨ {prompt}")
     elif text.startswith("/code"):
         prompt = text.replace("/code", "").strip() or "Write a Python function to demonstrate binary search"
         send_telegram_message(chat_id, "💻 Generating code with Kora Coder...")