添加机器人功能

main.py

@@ -2,7 +2,8 @@ import os
 import asyncio
 import json
 import logging
-
+from telegram import Bot, Update
+from telegram.ext import Application, CommandHandler, MessageHandler
 import httpx
 
 from pikpakapi import PikPakApi
@@ -68,6 +69,8 @@ if THUNDERX_PASSWORD is None:
     raise ValueError("请在环境变量中设置THUNDERX_PASSWORD，密码用来登陆!")
 
 PROXY_URL = os.getenv("PROXY_URL")
+TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN")
+TG_WEBHOOK_URL = os.getenv("TG_WEBHOOK_URL")
 
 
 async def verify_token(
@@ -92,6 +95,8 @@ async def verify_token(
 
 
 app = FastAPI()
+
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -107,16 +112,30 @@ templates = Jinja2Templates(
     directory="templates", variable_start_string="{[", variable_end_string="]}"
 )
 
-THUNDERX_CLIENT = None
-
 
 async def log_token(THUNDERX_CLIENT, extra_data):
     logging.info(f"Token: {THUNDERX_CLIENT.encoded_token}, Extra Data: {extra_data}")
 
 
+THUNDERX_CLIENT = None
+TG_BOT_APPLICATION = None
+
+
+###################TG机器人功能区###################
+# 定义命令处理函数
+def start(update: Update, context):
+    update.message.reply_text("你好！我是一个 FastAPI + Telegram 机器人！")
+
+
+def help(update: Update, context):
+    update.message.reply_text("这是一个帮助信息！")
+
+
 @app.on_event("startup")
 async def init_client():
     global THUNDERX_CLIENT
+    global TG_BOT
+    global TG_DISPATHER
     if not os.path.exists("thunderx.txt"):
         THUNDERX_CLIENT = PikPakApi(
             username=THUNDERX_USERNAME,
@@ -143,6 +162,30 @@ async def init_client():
                 )
             )
 
+    if TG_BOT_TOKEN is None:
+        print("未设置TG_BOT_TOKEN无法实现TG机器人功能！")
+    else:
+        TG_BOT_APPLICATION = Application.builder().token(TG_BOT_TOKEN).build()
+        await TG_BOT_APPLICATION.bot.set_webhook(TG_WEBHOOK_URL)
+        # 将命令处理函数添加到 dispatcher
+        TG_BOT_APPLICATION.add_handler(CommandHandler("start", start))
+        TG_BOT_APPLICATION.add_handler(CommandHandler("help", help))
+
+
+# FastAPI 路由：接收来自 Telegram 的 Webhook 回调
+@app.post("/webhook")
+async def webhook(request: Request):
+    # 从请求获取 JSON 数据
+    data = await request.json()
+
+    # 将 Telegram Update 转换为 Update 对象
+    update = Update.de_json(data, TG_BOT_APPLICATION.bot)
+
+    # 将 Update 对象传递给 Application 进行处理
+    await TG_BOT_APPLICATION.process_update(update)
+
+    return JSONResponse({"status": "ok"})
+
 
 @front_router.get(
     "/",

requirements.txt

@@ -1,4 +1,5 @@
 fastapi
+python-telegram-bot
 pydantic-settings
 requests
 jinja2