from fastapi import APIRouter, Body, Depends, HTTPException from sqlalchemy.orm import Session from typing import Dict from .. import crud, schemas from ..database import get_db from ..models import User import os router = APIRouter(prefix="/admin", tags=["admin"]) # Проверка админ-секрета через переменную окружения def verify_admin_secret(admin_secret: str) -> bool: """ Проверка admin-секрета Разрешенные источники: 1. DEVELOPER_SECRET из переменной окружения 2. Можно добавить дополнительные проверки (IP, токен и т.д.) """ valid_secret = os.getenv("DEVELOPER_SECRET") if not valid_secret: raise HTTPException(status_code=500, detail="DEVELOPER_SECRET not configured") return admin_secret == valid_secret @router.post("/users/{user_id}/make-developer") def make_user_developer( user_id: int, admin_secret: str = Body(..., embed=True), db: Session = Depends(get_db) ): """ Назначить статус разработчика для пользователя Требует admin_secret в запросе """ if not verify_admin_secret(admin_secret): raise HTTPException(status_code=403, detail="Invalid admin secret") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") user.is_developer = True db.commit() db.refresh(user) return { "message": f"Developer status enabled for user {user.email}", "user_id": user.id, "email": user.email, "is_developer": True } @router.post("/users/{user_id}/remove-developer") def remove_user_developer( user_id: int, admin_secret: str = Body(..., embed=True), db: Session = Depends(get_db) ): """ Убрать статус разработчика у пользователя Требует admin_secret в запросе """ if not verify_admin_secret(admin_secret): raise HTTPException(status_code=403, detail="Invalid admin secret") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") user.is_developer = False db.commit() db.refresh(user) return { "message": f"Developer status removed for user {user.email}", "user_id": user.id, "email": user.email, "is_developer": False } @router.get("/users") def list_users( admin_secret: str, db: Session = Depends(get_db) ): if not verify_admin_secret(admin_secret): raise HTTPException(status_code=403, detail="Invalid admin secret") users = db.query(User).all() return { "users": [ { "id": user.id, "email": user.email, "is_developer": user.is_developer, "is_active": user.is_active, "created_at": user.created_at.isoformat() if user.created_at else None } for user in users ] }