215 lines
7.5 KiB
Python
215 lines
7.5 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy import text
|
|
from datetime import timedelta
|
|
from database.database import SessionLocal
|
|
from database.models import Document, User
|
|
from api.auth import (
|
|
UserLogin, UserCreate, Token, UserResponse, RoleUpdate,
|
|
get_password_hash, verify_password,
|
|
create_access_token, decode_token, ACCESS_TOKEN_EXPIRE_MINUTES
|
|
)
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# ── Fonction vérification admin ──────────────────────────────
|
|
def require_admin(token: str):
|
|
email = decode_token(token)
|
|
if not email:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
db = SessionLocal()
|
|
user = db.query(User).filter(User.email == email).first()
|
|
db.close()
|
|
if not user or user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Accès refusé")
|
|
return user
|
|
|
|
# ── Endpoints existants ──────────────────────────────────────
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return {"message": "RegWatch API running"}
|
|
|
|
@app.post("/signup", response_model=UserResponse)
|
|
def signup(user: UserCreate):
|
|
db = SessionLocal()
|
|
existing_user = db.query(User).filter(User.email == user.email).first()
|
|
if existing_user:
|
|
db.close()
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
hashed_password = get_password_hash(user.password)
|
|
db_user = User(email=user.email, password=hashed_password, full_name=user.full_name)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
db.close()
|
|
return UserResponse(id=db_user.id, email=db_user.email, full_name=db_user.full_name, role=db_user.role)
|
|
|
|
@app.post("/login", response_model=Token)
|
|
def login(credentials: UserLogin):
|
|
try:
|
|
db = SessionLocal()
|
|
user = db.query(User).filter(User.email == credentials.email).first()
|
|
db.close()
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="User not found")
|
|
if not verify_password(credentials.password, user.password):
|
|
raise HTTPException(status_code=401, detail="Invalid password")
|
|
access_token = create_access_token(
|
|
data={"sub": user.email},
|
|
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
return Token(access_token=access_token, token_type="bearer")
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Server error: {str(e)}")
|
|
|
|
@app.get("/me", response_model=UserResponse)
|
|
def get_current_user(token: str):
|
|
email = decode_token(token)
|
|
if not email:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
db = SessionLocal()
|
|
user = db.query(User).filter(User.email == email).first()
|
|
db.close()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return UserResponse(id=user.id, email=user.email, full_name=user.full_name, role=user.role)
|
|
|
|
@app.get("/documents")
|
|
def get_documents():
|
|
db = SessionLocal()
|
|
documents = db.query(Document).all()
|
|
result = []
|
|
for doc in documents:
|
|
result.append({
|
|
"id": doc.id,
|
|
"title": doc.title,
|
|
"ingredient": doc.ingredient,
|
|
"source": doc.source,
|
|
"type": doc.document_type,
|
|
"date": doc.meeting_date,
|
|
"pdf_url": doc.pdf_url
|
|
})
|
|
db.close()
|
|
return result
|
|
|
|
# ── Migration colonne role ───────────────────────────────────
|
|
|
|
@app.post("/admin/migrate-add-role")
|
|
def migrate_add_role():
|
|
"""Endpoint temporaire — ajoute la colonne role si absente"""
|
|
db = SessionLocal()
|
|
try:
|
|
db.execute(text("ALTER TABLE users ADD COLUMN IF NOT EXISTS role VARCHAR DEFAULT 'user'"))
|
|
db.commit()
|
|
return {"message": "Colonne role ajoutée avec succès"}
|
|
except Exception as e:
|
|
return {"message": str(e)}
|
|
finally:
|
|
db.close()
|
|
|
|
# ── Premier admin sans neckfire ──────────────────────────────
|
|
|
|
@app.post("/admin/make-admin")
|
|
def make_admin(token: str, target_email: str):
|
|
"""
|
|
Passe un user en admin.
|
|
Fonctionne sans auth si aucun admin n'existe encore.
|
|
Se désactive automatiquement si un admin existe déjà.
|
|
"""
|
|
db = SessionLocal()
|
|
existing_admins = db.query(User).filter(User.role == "admin").count()
|
|
|
|
if existing_admins > 0:
|
|
# Il y a déjà un admin — vérifier que l'appelant est admin
|
|
email = decode_token(token)
|
|
caller = db.query(User).filter(User.email == email).first()
|
|
if not caller or caller.role != "admin":
|
|
db.close()
|
|
raise HTTPException(status_code=403, detail="Accès refusé")
|
|
|
|
target = db.query(User).filter(User.email == target_email).first()
|
|
if not target:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail="Utilisateur introuvable")
|
|
|
|
target.role = "admin"
|
|
db.commit()
|
|
db.close()
|
|
return {"message": f"{target_email} est maintenant admin"}
|
|
|
|
# ── Endpoints admin ──────────────────────────────────────────
|
|
|
|
@app.get("/admin/stats")
|
|
def admin_stats(token: str):
|
|
require_admin(token)
|
|
db = SessionLocal()
|
|
total_docs = db.query(Document).count()
|
|
cir_docs = db.query(Document).filter(Document.source == "CIR").count()
|
|
sccs_docs = db.query(Document).filter(Document.source == "SCCS").count()
|
|
total_users = db.query(User).count()
|
|
db.close()
|
|
return {
|
|
"total_documents": total_docs,
|
|
"cir_documents": cir_docs,
|
|
"sccs_documents": sccs_docs,
|
|
"total_users": total_users
|
|
}
|
|
|
|
@app.get("/admin/users")
|
|
def admin_get_users(token: str):
|
|
require_admin(token)
|
|
db = SessionLocal()
|
|
users = db.query(User).all()
|
|
result = [{"id": u.id, "email": u.email, "full_name": u.full_name, "role": u.role} for u in users]
|
|
db.close()
|
|
return result
|
|
|
|
@app.delete("/admin/users/{user_id}")
|
|
def admin_delete_user(user_id: int, token: str):
|
|
require_admin(token)
|
|
db = SessionLocal()
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail="Utilisateur introuvable")
|
|
db.delete(user)
|
|
db.commit()
|
|
db.close()
|
|
return {"message": "Utilisateur supprimé"}
|
|
|
|
@app.patch("/admin/users/{user_id}/role")
|
|
def admin_update_role(user_id: int, role_update: RoleUpdate, token: str):
|
|
require_admin(token)
|
|
db = SessionLocal()
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail="Utilisateur introuvable")
|
|
user.role = role_update.role
|
|
db.commit()
|
|
db.close()
|
|
return {"message": f"Rôle mis à jour : {role_update.role}"}
|
|
|
|
@app.delete("/admin/documents/{doc_id}")
|
|
def admin_delete_document(doc_id: int, token: str):
|
|
require_admin(token)
|
|
db = SessionLocal()
|
|
doc = db.query(Document).filter(Document.id == doc_id).first()
|
|
if not doc:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail="Document introuvable")
|
|
db.delete(doc)
|
|
db.commit()
|
|
db.close()
|
|
return {"message": "Document supprimé"} |