Webhook

# webhook.py — 只用标准库
# 运行：WEBHOOK_SECRET='…' python3 webhook.py
import hmac
import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["WEBHOOK_SECRET"]


class Webhook(BaseHTTPRequestHandler):
    def do_POST(self):
        got = self.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
        if not hmac.compare_digest(got, SECRET):
            self.send_response(401)
            self.end_headers()
            return
        update = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        message = update.get("message")
        answer = b""
        if message and "text" in message:
            answer = json.dumps({"method": "sendMessage",
                                 "chat_id": message["chat"]["id"],
                                 "text": message["text"]}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(answer)))
        self.end_headers()
        self.wfile.write(answer)


HTTPServer(("127.0.0.1", 8081), Webhook).serve_forever()

// webhook.mjs — Node.js 18 及以上，不使用库
// 运行：WEBHOOK_SECRET='…' node webhook.mjs
import http from 'node:http';
import { timingSafeEqual } from 'node:crypto';

const SECRET = Buffer.from(process.env.WEBHOOK_SECRET);

http.createServer(async (req, res) => {
  const got = Buffer.from(req.headers['x-telegram-bot-api-secret-token'] ?? '');
  if (req.method !== 'POST' || got.length !== SECRET.length || !timingSafeEqual(got, SECRET)) {
    res.writeHead(401).end();
    return;
  }
  let body = '';
  for await (const chunk of req) body += chunk;
  const message = JSON.parse(body).message;
  if (message?.text) {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ method: 'sendMessage', chat_id: message.chat.id, text: message.text }));
  } else {
    res.writeHead(200).end();
  }
}).listen(8081, '127.0.0.1');

# webhook_aiogram.py — pip install aiogram
# 连 Webhook 设置（setWebhook）也由这个程序完成
import os

from aiogram import Bot, Dispatcher
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.client.telegram import TelegramAPIServer
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web

URL = "https://bot.example.com/pabal-webhook"
SECRET = os.environ["WEBHOOK_SECRET"]
dp = Dispatcher()


@dp.message()
async def echo(message):
    if message.text:
        return message.answer(message.text)       # return：放进 Webhook 响应中发送


async def on_startup(bot: Bot):
    await bot.set_webhook(URL, secret_token=SECRET)


def main():
    session = AiohttpSession(api=TelegramAPIServer.from_base("https://pabal.me"))
    bot = Bot(os.environ["BOT_TOKEN"], session=session)
    dp.startup.register(on_startup)
    app = web.Application()
    SimpleRequestHandler(dispatcher=dp, bot=bot, secret_token=SECRET).register(app, path="/pabal-webhook")
    setup_application(app, dp, bot=bot)
    web.run_app(app, host="127.0.0.1", port=8081)


main()

# webhook_ptb.py — pip install "python-telegram-bot[webhooks]"
# 连 Webhook 设置（setWebhook）也由这个程序完成
import os

from telegram import Update
from telegram.ext import Application, ContextTypes, MessageHandler, filters

SERVER = "https://pabal.me"


async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)


app = (Application.builder().token(os.environ["BOT_TOKEN"])
       .base_url(f"{SERVER}/bot").base_file_url(f"{SERVER}/file/bot").build())
app.add_handler(MessageHandler(filters.TEXT, echo))
app.run_webhook(listen="127.0.0.1", port=8081, url_path="pabal-webhook",
                webhook_url="https://bot.example.com/pabal-webhook",
                secret_token=os.environ["WEBHOOK_SECRET"])