File size: 4,088 Bytes
0a084aa f33256a f2c6eef 0a084aa 9060885 0a084aa f2c6eef f33256a 0a084aa a70183c 0a084aa b0b6243 0a084aa d8d0f93 b0b6243 d8d0f93 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | from fastapi import FastAPI, Request, Response # Import FastAPI classes for creating the web app and handling requests/responses
# from async_tgvoicebot import handle_message # Import the asynchronous function to handle incoming Telegram messages
import aiohttp # Import aiohttp for making asynchronous HTTP requests
import os # Import os module to access environment variables
import logging
app = FastAPI() # Initialize a FastAPI application instance
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Retrieve the Telegram bot token from environment variables
TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN')
# Retrieve the secret token for webhook security from environment variables
SECRET_TOKEN = os.environ.get('SECRET_TOKEN')
print(TELEGRAM_BOT_TOKEN)
print(SECRET_TOKEN)
@app.post("/webhook")
async def webhook(request: Request):
"""
Endpoint to receive webhook updates from Telegram.
Only processes the update if the secret token matches (if provided).
"""
try:
# Check if a secret token is set for additional security
if SECRET_TOKEN:
# Compare the secret token in the request headers with the expected SECRET_TOKEN
if request.headers.get('X-Telegram-Bot-Api-Secret-Token') != SECRET_TOKEN:
# If tokens do not match, respond with 403 Forbidden status
return Response(status_code=403)
# Parse the incoming JSON payload from the request
update = await request.json()
# Create an asynchronous HTTP session
async with aiohttp.ClientSession() as session:
# Call the handle_message function with the current session and the message part of the update
# await handle_message(session, update.get('message', {}))
pass
# Respond to Telegram that the update was received and processed successfully
logger.info("Received webhook request")
return {"status": "ok"}
except Exception as e:
logger.error(f"Error processing webhook: {e}")
return Response(status_code=500)
@app.get("/")
async def root():
"""
Root endpoint to confirm that the Telegram Bot is running.
"""
# Return a simple JSON message indicating the bot's status
return {"message": "Telegram Bot is running"}
# Endpoint to set webhook
@app.get("/set_webhook")
async def set_webhook():
"""
Endpoint to set the Telegram webhook to the specified URL.
This should be called after deploying the bot to ensure Telegram sends updates to the correct endpoint.
"""
# Construct the webhook URL using the SPACE_ID environment variable
webhook_url = f"https://dynstat-test2.hf.space/webhook"
# Telegram API URL to set the webhook
set_webhook_url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/setWebhook"
# set_webhook_url = f"https://149.154.167.220/bot{TELEGRAM_BOT_TOKEN}/setWebhook"
# Payload containing the webhook URL and the secret token for security
payload = {
"url": webhook_url,
"secret_token": SECRET_TOKEN
}
# Create an asynchronous HTTP session
async with aiohttp.ClientSession() as session:
# Send a POST request to Telegram's setWebhook API with the payload
async with session.post(set_webhook_url, json=payload) as response:
# Await and return the JSON response from Telegram confirming the webhook setup
return await response.json()
@app.get("/test_conn")
async def test_conn():
try:
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session:
async with session.get("https://api.telegram.org") as response:
return {"status": "success", "telegram_api_status": response.status}
except Exception as e:
logger.error(f"Connection test failed: {str(e)}", exc_info=True)
return {"status": "failed", "error": str(e)}
|