Update app.py

app.py

@@ -1,188 +1,57 @@
 # app.py
+# test_internet.py
 import os
-import pathlib
+import socket
+import ssl
 import sys
-from typing import Any
-
-path_root = pathlib.Path(__file__).parents[0]
-sys.path.append(str(path_root / "src"))
-sys.path.append(str(path_root / "src/qwen_api"))
-
-import logging
+import urllib.request
 import asyncio
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, Request, Response
-from telegram import Update
-from telegram.ext import (
-    Application,
-    CommandHandler,
-    MessageHandler,
-    filters,
-    ContextTypes,
-)
-
-from utils import *
-from chat_ai import qwen_chat
-import sync_hc
-
-# --- Configuration ---
-TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "")
-HF_SPACE = os.getenv("SENDER_URL", "")  # e.g., "yourname/your-space"
-WEBHOOK_PATH = "/webhook"
-WEBHOOK_URL = f"{HF_SPACE}{WEBHOOK_PATH}" if HF_SPACE else ""
-
-# Enable logging
-logging.basicConfig(
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
-)
-logger = logging.getLogger(__name__)
-
-# Global bot application
-bot_app: Application | None = None
-dataDict = None
-
-def load_data():
-    global dataDict
-    dataDict = load_obj("data") if load_obj("data") is not None else {}
-    dataDict.setdefault("lock", True)
-    dataDict.setdefault("users", set())
-    if "config" not in dataDict: # type: ignore
-        filename = os.path.join(os.path.dirname(__file__), 'config.json')
-        dataDict["config"] = load_json(filename)
-
-def save_data():
-    global dataDict
-    save_obj(dataDict, "data")
-
-# --- Command handlers (async) ---
-async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    user = update.effective_user
-    await update.message.reply_markdown_v2(fr'Hi {user.mention_markdown_v2()}\!')
-
-async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text('Help!')
-
-async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text(str(update.message.text))
-    
-async def qwen(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    """Calling qwen."""
-    
-    res, content = qwen_chat(str(update.message.text))
-    
-    await update.message.reply_text(content)
-
-async def wget_downloader(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text("❌ wget not supported in webhook mode.")
-
-async def ip(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text(get_public_ip() + " (" + get_local_ip() + ")")
-
-async def commands(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    for command in dataDict.get("config", {}).get("commands", []):
-        if not command.get("hide"):
-            await update.message.reply_text(command["name"])
-
-async def lock(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    global dataDict
-    dataDict["lock"] = not dataDict["lock"]
-    save_data()
-    await update.message.reply_text(str(dataDict["lock"]))
-
-async def downloader(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text("📥 File handling limited in webhook mode.")
+import httpx
+
+def test_dns():
+    print("🔍 Testing DNS resolution for api.telegram.org...")
+    try:
+        ip = socket.gethostbyname("api.telegram.org")
+        print(f"✅ DNS OK: api.telegram.org -> {ip}")
+        return True
+    except Exception as e:
+        print(f"❌ DNS FAILED: {e}")
+        return False
+
+def test_https_urllib():
+    print("\n🌐 Testing HTTPS access via urllib (Python stdlib)...")
+    try:
+        with urllib.request.urlopen("https://api.telegram.org", timeout=10) as response:
+            print(f"✅ HTTPS OK: status {response.getcode()}")
+            return True
+    except Exception as e:
+        print(f"❌ HTTPS FAILED (urllib): {e}")
+        return False
+
+async def test_https_httpx():
+    print("\n🚀 Testing HTTPS access via httpx (used by python-telegram-bot)...")
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            resp = await client.get("https://api.telegram.org")
+            print(f"✅ HTTPS OK (httpx): status {resp.status_code}")
+            return True
+    except Exception as e:
+        print(f"❌ HTTPS FAILED (httpx): {e}")
+        return False
 
-async def command(update: Update, context: ContextTypes.DEFAULT_TYPE):
-    await update.message.reply_text("⚙️ Command execution disabled in webhook demo.")
-
-# --- Security wrapper ---
-def security_check(handler):
-    async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE):
-        user = update.effective_user
-        if user.id in dataDict.get("users", set()):
-            return await handler(update, context)
-        else:
-            await update.message.reply_text("🔒 Access denied.")
-    return wrapper
-
-# --- Lifespan: Initialize bot and set webhook ---
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    global bot_app
-    logger.info("Starting Telegram bot (webhook mode)...")
-    
-    # Load persistent data
-    load_data()
-    
-    # Initialize Telegram application
-    bot_app = Application.builder().token(TELEGRAM_TOKEN).build()
-
-    # Register handlers
-    bot_app.add_handler(CommandHandler("start", start))
-    bot_app.add_handler(CommandHandler("help", help_command))
-    bot_app.add_handler(CommandHandler("ip", security_check(ip)))
-    bot_app.add_handler(CommandHandler("commands", security_check(commands)))
-    bot_app.add_handler(CommandHandler("lock", security_check(lock)))
-    bot_app.add_handler(CommandHandler("wget", security_check(wget_downloader)))
-    bot_app.add_handler(CommandHandler("download", security_check(downloader)))
-    bot_app.add_handler(CommandHandler("command", security_check(command)))
-    bot_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, qwen))
-    bot_app.add_handler(MessageHandler(filters.Document.ALL, downloader))
-
-    # Initialize bot
-    await bot_app.initialize()
-    
-    # Set webhook
-    if WEBHOOK_URL:
-        logger.info(f"Setting webhook to: {WEBHOOK_URL}")
-        await bot_app.bot.set_webhook(url=WEBHOOK_URL)
-    else:
-        logger.warning("WEBHOOK_URL not set — skipping webhook setup")
-
-    # Start the application (so it can process updates)
-    await bot_app.start()
-
-    yield  # App is running
-
-    # Shutdown
-    logger.info("Shutting down Telegram bot...")
-    await bot_app.stop()
-    await bot_app.shutdown()
-
-# --- FastAPI app ---
-app = FastAPI(lifespan=lifespan)
-
-@app.get("/")
-async def root():
-    return {"status": "Telegram bot is running", "webhook_url": WEBHOOK_URL}
-
-@app.post(WEBHOOK_PATH)
-async def telegram_webhook(request: Request):
-    """Receive Telegram updates via webhook."""
-    if bot_app is None:
-        return Response(status_code=503, content="Bot not ready")
-    
-    # Parse update
-    update_data = await request.json()
-    update = Update.de_json(update_data, bot_app.bot)
-    
-    # Put update into the bot's update queue for processing
-    await bot_app.update_queue.put(update)
-    
-    return {"ok": True}
-
-# --- Local dev server ---
 if __name__ == "__main__":
-    import uvicorn
+    print("="*60)
+    print("📡 Hugging Face Space: Internet Access Test for Telegram")
+    print("="*60)
     
-    # forward with 
-    # cloudflared tunnel --url http://localhost:7860
+    dns_ok = test_dns()
+    urllib_ok = test_https_urllib() if dns_ok else False
+    httpx_ok = asyncio.run(test_https_httpx()) if dns_ok else False
 
-    uvicorn.run(
-        "app:app",
-        host="0.0.0.0",
-        port=7860,
-        reload=True,
-        reload_dirs=["."]
-    )
+    print("\n" + "="*60)
+    if dns_ok and (urllib_ok or httpx_ok):
+        print("🎉 SUCCESS: Outbound internet appears to work!")
+    else:
+        print("🛑 FAILURE: Outbound internet is BLOCKED.")
+        print("   → Telegram bots will NOT work on this platform.")
+    print("="*60)