159 lines
5.2 KiB
Python
159 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import mimetypes
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app.bot import TelegramBotService
|
|
from app.config import load_settings
|
|
from app.db import Database
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
settings = load_settings()
|
|
database = Database(settings.database_path)
|
|
bot_service: TelegramBotService | None = None
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
global bot_service
|
|
settings.data_dir.mkdir(parents=True, exist_ok=True)
|
|
settings.avatars_dir.mkdir(parents=True, exist_ok=True)
|
|
database.initialize()
|
|
if settings.telegram_token:
|
|
try:
|
|
bot_service = TelegramBotService(settings, database)
|
|
bot_service.start()
|
|
except Exception:
|
|
logger.exception("Telegram bot could not start; HTTP application remains available")
|
|
else:
|
|
logger.warning("TELEGRAM_TOKEN is empty; running HTTP application without the bot")
|
|
yield
|
|
if bot_service:
|
|
bot_service.stop()
|
|
|
|
|
|
app = FastAPI(
|
|
title="Пожитки API",
|
|
version="0.1.0",
|
|
docs_url="/api/docs",
|
|
openapi_url="/api/openapi.json",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
@app.get("/api/healthz")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/config")
|
|
def public_config() -> dict[str, str | None]:
|
|
return {"bot_username": database.get_meta("bot_username")}
|
|
|
|
|
|
@app.get("/api/users/{user_id}/summary")
|
|
def user_summary(user_id: int) -> dict:
|
|
result = database.summary(user_id)
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
|
return result
|
|
|
|
|
|
@app.get("/api/users/{user_id}/friends")
|
|
def user_friends(user_id: int) -> dict[str, list[int]]:
|
|
return {"user_ids": database.friend_ids(user_id)}
|
|
|
|
|
|
@app.get("/api/users/{user_id}/categories")
|
|
def user_categories(user_id: int) -> dict[str, list[dict[str, str]]]:
|
|
return {"items": database.user_categories(user_id)}
|
|
|
|
|
|
@app.get("/api/users/{user_id}/transactions")
|
|
def user_transactions(
|
|
user_id: int, limit: int = Query(default=50, ge=1, le=100)
|
|
) -> dict[str, list[dict]]:
|
|
return {"items": database.transactions(user_id, limit)}
|
|
|
|
|
|
@app.get("/api/categories")
|
|
def global_categories() -> dict[str, list[dict[str, str]]]:
|
|
return {"items": database.global_categories()}
|
|
|
|
|
|
def parse_user_ids(raw: str | None) -> list[int] | None:
|
|
if not raw:
|
|
return None
|
|
values: list[int] = []
|
|
for item in raw.split(","):
|
|
item = item.strip()
|
|
if not item:
|
|
continue
|
|
try:
|
|
value = int(item)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail="user_ids должен быть списком чисел") from exc
|
|
if value <= 0:
|
|
raise HTTPException(status_code=422, detail="user_ids должен содержать положительные ID")
|
|
values.append(value)
|
|
if len(values) > 100:
|
|
raise HTTPException(status_code=422, detail="Можно запросить не более 100 пользователей")
|
|
return values or None
|
|
|
|
|
|
@app.get("/api/leaderboard")
|
|
def leaderboard(
|
|
user_ids: str | None = Query(
|
|
default=None, description="Telegram user ID через запятую"
|
|
),
|
|
category: list[str] | None = Query(
|
|
default=None, description="Один или несколько фильтров по категории"
|
|
),
|
|
limit: int = Query(default=100, ge=1, le=100),
|
|
) -> dict[str, list[dict]]:
|
|
categories = None
|
|
if category:
|
|
categories = list(dict.fromkeys(item.strip() for item in category if item.strip()))
|
|
if len(categories) > 20 or any(len(item) > 40 for item in categories):
|
|
raise HTTPException(status_code=422, detail="Некорректный фильтр категорий")
|
|
return {
|
|
"items": database.leaderboard(
|
|
user_ids=parse_user_ids(user_ids),
|
|
categories=categories,
|
|
limit=limit,
|
|
)
|
|
}
|
|
|
|
|
|
@app.get("/api/avatars/{user_id}", response_class=FileResponse)
|
|
def avatar(user_id: int) -> FileResponse:
|
|
summary = database.summary(user_id)
|
|
if not summary or not summary["avatar_url"]:
|
|
raise HTTPException(status_code=404, detail="Аватар не найден")
|
|
with database.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT avatar_path FROM users WHERE user_id = ?", (user_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Аватар не найден")
|
|
filename = Path(row["avatar_path"]).name
|
|
path = settings.avatars_dir / filename
|
|
if not path.is_file():
|
|
raise HTTPException(status_code=404, detail="Аватар не найден")
|
|
media_type = mimetypes.guess_type(path)[0] or "image/jpeg"
|
|
return FileResponse(path, media_type=media_type)
|
|
|
|
|
|
static_dir = Path(__file__).resolve().parent.parent / "static"
|
|
app.mount("/", StaticFiles(directory=static_dir, html=True), name="webapp")
|