Stardream
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from app.core.config import settings
|
|
from app.models import Base # noqa: F401 - ensure all models are registered
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
from app.bot import router as bot_router
|
|
from app.bot.router import bot, dp
|
|
|
|
me = await bot.get_me()
|
|
bot_router.bot_username = me.username or ""
|
|
|
|
from aiogram.types import BotCommand
|
|
await bot.set_my_commands([
|
|
BotCommand(command="start", description="Open lottery manager"),
|
|
BotCommand(command="newlottery", description="Create a new lottery"),
|
|
BotCommand(command="mylotteries", description="View my lotteries"),
|
|
BotCommand(command="help", description="Help and project info"),
|
|
])
|
|
|
|
from app.services.scheduler import auto_draw_scheduler
|
|
asyncio.create_task(auto_draw_scheduler())
|
|
|
|
if settings.webhook_url:
|
|
webhook_url = f"{settings.webhook_url}{settings.webhook_path}"
|
|
await bot.set_webhook(
|
|
webhook_url,
|
|
secret_token=settings.webhook_secret or None,
|
|
drop_pending_updates=True,
|
|
)
|
|
else:
|
|
# Development: long-polling mode
|
|
asyncio.create_task(dp.start_polling(bot, handle_signals=False))
|
|
|
|
yield
|
|
|
|
from app.bot.router import bot as _bot
|
|
await _bot.session.close()
|
|
|
|
|
|
app = FastAPI(title="TGLotteryBot API", lifespan=lifespan)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# API routers
|
|
from app.api.routers import lotteries, prizes, participants, chats, users # noqa: E402
|
|
|
|
app.include_router(lotteries.router, prefix="/api")
|
|
app.include_router(prizes.router, prefix="/api")
|
|
app.include_router(participants.router, prefix="/api")
|
|
app.include_router(chats.router, prefix="/api")
|
|
app.include_router(users.router, prefix="/api")
|
|
|
|
|
|
@app.post(settings.webhook_path)
|
|
async def telegram_webhook(request: Request) -> Response:
|
|
"""Receive Telegram updates via webhook."""
|
|
from app.bot.router import bot, dp
|
|
from aiogram.types import Update
|
|
|
|
secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
|
|
if settings.webhook_secret and secret != settings.webhook_secret:
|
|
return Response(status_code=403)
|
|
|
|
body = await request.json()
|
|
update = Update(**body)
|
|
await dp.feed_update(bot, update)
|
|
return Response(status_code=200)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict:
|
|
return {"status": "ok"}
|