mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-06-02 19:55:18 +02:00
52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""Register the SurfSense Telegram webhook."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
from dotenv import load_dotenv
|
|
from telegram import Bot
|
|
|
|
load_dotenv()
|
|
|
|
WEBHOOK_SECRET_RE = re.compile(r"^[A-Za-z0-9_-]{1,256}$")
|
|
|
|
|
|
async def main() -> int:
|
|
token = os.getenv("TELEGRAM_SHARED_BOT_TOKEN")
|
|
secret = os.getenv("TELEGRAM_WEBHOOK_SECRET")
|
|
base_url = os.getenv("GATEWAY_BASE_URL") or os.getenv("BACKEND_URL")
|
|
if not token or not secret or not base_url:
|
|
print(
|
|
"Missing TELEGRAM_SHARED_BOT_TOKEN, TELEGRAM_WEBHOOK_SECRET, or GATEWAY_BASE_URL/BACKEND_URL",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
if not WEBHOOK_SECRET_RE.fullmatch(secret):
|
|
print(
|
|
"TELEGRAM_WEBHOOK_SECRET must be 1-256 chars and contain only A-Z, a-z, 0-9, '_' or '-'",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
webhook_url = f"{base_url.rstrip('/')}/api/v1/gateway/webhooks/telegram/{secret}"
|
|
bot = Bot(token=token)
|
|
ok = await bot.set_webhook(
|
|
url=webhook_url,
|
|
secret_token=secret,
|
|
allowed_updates=["message", "edited_message"],
|
|
drop_pending_updates=True,
|
|
)
|
|
if not ok:
|
|
print("Telegram rejected webhook registration", file=sys.stderr)
|
|
return 1
|
|
print(f"Registered Telegram webhook: {webhook_url}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|
|
|