43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from fastapi import HTTPException
|
|
from ..utils.jwt_handler import create_token
|
|
from ..db import get_connection
|
|
import hashlib
|
|
import datetime
|
|
|
|
def authenticate_admin(username: str, password: str):
|
|
"""
|
|
验证管理员身份
|
|
"""
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise HTTPException(status_code=500, detail="Database connection failed!")
|
|
|
|
try:
|
|
cursor = connection.cursor(dictionary=True)
|
|
cursor.execute("SELECT * FROM users WHERE username = %s;", (username,))
|
|
user = cursor.fetchone()
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Invalid username or password.")
|
|
|
|
# 验证密码
|
|
if user["password"] != hashlib.md5(password.encode()).hexdigest():
|
|
raise HTTPException(status_code=401, detail="Invalid username or password.")
|
|
|
|
# 检查用户类型是否为 admin
|
|
if user["user_type"] != "admin":
|
|
raise HTTPException(status_code=403, detail="Permission denied: Not an admin user.")
|
|
|
|
return user
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
def generate_login_token(username: str, remember_me: bool):
|
|
"""
|
|
生成登录 Token
|
|
"""
|
|
expires_delta = datetime.timedelta(days=7 if remember_me else 1)
|
|
token = create_token({"sub": username}, expires_delta)
|
|
return token, int(expires_delta.total_seconds())
|