File size: 2,689 Bytes
a6e3f28 6397a42 a6e3f28 6397a42 a6e3f28 6397a42 dfa56fa 6397a42 dfa56fa a6e3f28 6397a42 a6e3f28 6397a42 20eb075 a6e3f28 6397a42 20eb075 6397a42 20eb075 6397a42 dfa56fa 6397a42 20eb075 a6e3f28 20eb075 6397a42 20eb075 a6e3f28 6397a42 a9b7913 20eb075 6397a42 20eb075 6397a42 20eb075 6397a42 dfa56fa |
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 |
import telebot
from fastapi import FastAPI, Request, HTTPException
import uvicorn
from pydantic import BaseModel
# Your bot token is needed internally to send messages via Telegram's API.
TOKEN = "6655373829:AAGduLdLyNx7zUtxH73Sp3Z1vHKS35tV9WU"
# Your externally configured webhook URL (set up previously via BotFather or API call)
WEBHOOK_URL = "https://astraos-testing.hf.space/webhook"
# Initialize the TeleBot instance
bot = telebot.TeleBot(TOKEN, parse_mode="HTML")
# Create a FastAPI app instance
app = FastAPI()
# --- Models for request validation ---
class MessageRequest(BaseModel):
chat_id: int
text: str
# --- Endpoints ---
@app.post("/webhook")
async def telegram_webhook(request: Request):
"""
This endpoint receives incoming update payloads from Telegram.
Telegram sends updates as JSON via a POST request to this URL.
"""
try:
json_data = await request.json()
except Exception as e:
raise HTTPException(status_code=400, detail="Invalid JSON payload")
update = telebot.types.Update.de_json(json_data)
# Process the update (this will call any message handlers you've set up)
bot.process_new_updates([update])
# Acknowledge receipt of the update
return {"status": "ok"}
@app.get("/")
def home():
"""
A simple endpoint to verify that your bot’s webhook server is running.
"""
return {"message": "Bot is running via webhook"}
@app.post("/send_message")
async def send_message(msg: MessageRequest):
"""
This endpoint allows you to send a message from your bot to a user.
You provide a JSON payload with 'chat_id' and 'text'.
For example, you can POST:
{
"chat_id": 123456789,
"text": "Hello from webhook-only mode!"
}
The message is sent by calling the Telegram Bot API internally.
"""
try:
sent_message = bot.send_message(msg.chat_id, msg.text)
return {"status": "sent", "message_id": sent_message.message_id}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# --- Example Message Handler ---
@bot.message_handler(commands=["start"])
def start_command(message):
"""
When a user sends /start to your bot via Telegram (received via webhook),
this handler sends a welcome message.
"""
bot.reply_to(message, "Hello! I'm running via FastAPI webhook and ready to send messages!")
# --- Running the App ---
if __name__ == "__main__":
# We run the FastAPI app using uvicorn. Since your webhook is already set up on Telegram,
# you only need to start the server to receive updates.
uvicorn.run(app, host="0.0.0.0", port=7860)
|