如何用Aiogram(Python)处理Stripe消息？能否处理StripeEvent？

Aiogram的Dispatcher无法直接处理StripeEvent类型的消息，原因如下：

• Dispatcher是Aiogram专门设计用来处理Telegram Bot API推送的更新（比如用户消息、命令、回调查询等）的组件，它的handler仅针对Telegram的types.Update及相关衍生类型工作，无法解析Stripe发送的HTTP webhook请求。
• Stripe的webhook是通过HTTP POST请求将事件数据发送到你配置的公网端点，这类请求和Telegram Bot的更新机制完全独立。

可行的解决方案：同时运行Aiogram Bot和HTTP服务 #

你需要搭建两个并行的服务：一个是Aiogram驱动的Telegram Bot，另一个是接收Stripe webhook的HTTP服务（推荐用异步框架如FastAPI，和Aiogram的异步模型兼容），两者通过共享Bot实例实现联动。

示例代码 #

1. 初始化Aiogram Bot和Dispatcher

from aiogram import Bot, Dispatcher
import asyncio

# 替换为你的Telegram Bot Token
BOT_TOKEN = "your_bot_token_here"
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

# 可以在这里添加Telegram Bot的其他handler，比如命令、消息处理
@dp.message_handler(commands=["start"])
async def start_command(message):
    await message.reply("Bot is running!")

2. 搭建FastAPI服务处理Stripe Webhook

from fastapi import FastAPI, Request, HTTPException
import stripe

# 替换为你的Stripe密钥和Webhook签名密钥
STRIPE_SECRET_KEY = "your_stripe_secret_key_here"
STRIPE_WEBHOOK_SECRET = "your_stripe_webhook_secret_here"
stripe.api_key = STRIPE_SECRET_KEY

app = FastAPI()

@app.post("/stripe-webhook")
async def handle_stripe_webhook(request: Request):
    # 读取Stripe发送的请求体和签名头
    payload = await request.body()
    sig_header = request.headers.get("Stripe-Signature")

    # 验证Stripe事件的合法性
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, STRIPE_WEBHOOK_SECRET
        )
    except ValueError:
        raise HTTPException(status_code=400, detail="无效的请求体")
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="无效的签名")

    # 处理product.created事件
    if event["type"] == "product.created":
        product = event["data"]["object"]
        print(f"Stripe商店新增产品：{product['name']}（ID：{product['id']}）")
        # 用Aiogram Bot给管理员发送通知（替换为你的管理员Chat ID）
        await bot.send_message(chat_id="your_admin_chat_id", text=f"📢 Stripe商店新增产品：\n名称：{product['name']}\nID：{product['id']}")

    return {"status": "success"}

3. 同时运行两个服务

async def main():
    # 启动Aiogram的Polling（如果用Webhook模式，替换为对应的启动逻辑）
    polling_task = asyncio.create_task(dp.start_polling(bot, skip_updates=True))
    # 启动FastAPI服务（用uvicorn运行）
    import uvicorn
    fastapi_task = asyncio.create_task(uvicorn.run(app, host="0.0.0.0", port=8000))
    
    # 等待两个服务都运行
    await asyncio.gather(polling_task, fastapi_task)

if __name__ == "__main__":
    asyncio.run(main())

补充说明 #

• 如果你偏好同步框架，也可以用Flask替代FastAPI，但需要注意在Flask的同步函数中调用Aiogram的异步方法时，需要通过asyncio.run()来包装。
• 必须确保你的HTTP服务有公网可访问的地址，才能配置Stripe的webhook端点。

内容的提问来源于stack exchange，提问作者Tomas Angelo