1019 lines
36 KiB
Python
1019 lines
36 KiB
Python
# 导入所需的模块
|
|
from fastapi import FastAPI, Depends, HTTPException, status, Body, Security, Request
|
|
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm, APIKeyHeader
|
|
from sqlalchemy import create_engine, Column, String, DateTime, Boolean
|
|
from sqlalchemy.orm import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
from sqlalchemy.sql import func
|
|
from passlib.context import CryptContext
|
|
from datetime import datetime, timedelta, timezone
|
|
import os
|
|
from typing import Optional
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import re
|
|
from pydantic import BaseModel, EmailStr, field_validator
|
|
from dotenv import load_dotenv
|
|
import redis
|
|
import json
|
|
import uuid
|
|
import uvicorn
|
|
import secrets
|
|
import string
|
|
from typing import List
|
|
import random
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
import random
|
|
import string
|
|
from fastapi import Form
|
|
from supabase import create_client, Client
|
|
import asyncio
|
|
from enum import Enum
|
|
|
|
# 加载 .env 文件中的环境变量
|
|
load_dotenv()
|
|
|
|
# 从环境变量中获取数据库连接信息
|
|
DB_USER = os.getenv("DB_USER")
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD")
|
|
DB_HOST = os.getenv("DB_HOST")
|
|
DB_NAME = os.getenv("DB_NAME")
|
|
DB_PORT = os.getenv("DB_PORT")
|
|
|
|
# 构建数据库连接URL
|
|
SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
|
|
engine = create_engine(SQLALCHEMY_DATABASE_URL)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
Base = declarative_base()
|
|
|
|
# Redis配置
|
|
REDIS_HOST = os.getenv('REDIS_HOST')
|
|
REDIS_PORT = int(os.getenv('REDIS_PORT'))
|
|
REDIS_REGISTER_DB = int(os.getenv('REDIS_REGISTER_DB'))
|
|
REDIS_DB = int(os.getenv('REDIS_DB'))
|
|
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD')
|
|
REDIS_API_DB = int(os.getenv('REDIS_API_DB'))
|
|
REDIS_API_USAGE_DB = int(os.getenv('REDIS_API_USAGE_DB'))
|
|
|
|
|
|
# 邮箱配置
|
|
SMTP_SERVER = os.getenv('SMTP_SERVER')
|
|
SMTP_PORT = os.getenv('SMTP_PORT')
|
|
SMTP_USERNAME = os.getenv('SMTP_USERNAME')
|
|
SMTP_PASSWORD = os.getenv('SMTP_PASSWORD')
|
|
|
|
# 创建Redis客户端
|
|
redis_client = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
db=REDIS_DB,
|
|
password=REDIS_PASSWORD
|
|
)
|
|
|
|
redis_register_client = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
db=REDIS_REGISTER_DB,
|
|
password=REDIS_PASSWORD
|
|
)
|
|
# 创建专门用于 API 密钥的 Redis 客户端
|
|
redis_api_client = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
db=REDIS_API_DB,
|
|
password=REDIS_PASSWORD
|
|
)
|
|
|
|
redis_api_usage_client = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
db=REDIS_API_USAGE_DB,
|
|
password=REDIS_PASSWORD
|
|
)
|
|
|
|
# 定义用户模型，继承自Base类
|
|
class User(Base):
|
|
__tablename__ = "user"
|
|
uuid = Column(String(36), primary_key=True, index=True)
|
|
username = Column(String(50), unique=True, index=True, nullable=False)
|
|
email = Column(String(100), unique=True, index=True)
|
|
hashed_password = Column(String(255))
|
|
created_at = Column(DateTime, server_default=func.now())
|
|
last_login = Column(DateTime)
|
|
is_active = Column(Boolean, default=False)
|
|
phone_number = Column(String(20))
|
|
role = Column(String(20), default="admin")
|
|
avatar = Column(String(50))
|
|
email_verification_code = Column(String(6))
|
|
email_verified = Column(Boolean, default=False)
|
|
phone_verification_code = Column(String(6))
|
|
phone_verified = Column(Boolean, default=False)
|
|
|
|
# 创建数据库表
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# 密码哈希配置
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
# OAuth2密码流配置
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
|
|
|
|
|
# 创建FastAPI应用
|
|
app = FastAPI()
|
|
user_app = FastAPI()
|
|
app.mount("/user", user_app)
|
|
|
|
ALLOWED_ORIGINS = 'https://user.obscura.work'
|
|
|
|
# 添加CORS中间件
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=ALLOWED_ORIGINS,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# 定义用户创建模型
|
|
class UserCreate(BaseModel):
|
|
username: str
|
|
email: EmailStr
|
|
phone_number: str
|
|
password: str
|
|
|
|
@field_validator('password')
|
|
def password_complexity(cls, v):
|
|
if len(v) < 8:
|
|
raise ValueError('密码长度至少为8个字符')
|
|
if not re.search(r'[A-Z]', v):
|
|
raise ValueError('密码必须包含至少一个大写字母')
|
|
if not re.search(r'[a-z]', v):
|
|
raise ValueError('密码必须包含至少一个小写字母')
|
|
if not re.search(r'\d', v):
|
|
raise ValueError('密码必须包含至少一个数字')
|
|
if not re.search(r'[!@#$%^&*?:]', v):
|
|
raise ValueError('密码必须包含至少一个特殊字符')
|
|
return v
|
|
|
|
@field_validator('phone_number')
|
|
def phone_number_field_validator(cls, v):
|
|
if not re.match(r'^\+?1?\d{9,15}$', v):
|
|
raise ValueError('无效的手机号码')
|
|
return v
|
|
|
|
# 数据库会话依赖
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# 密码验证函数
|
|
def verify_password(plain_password, hashed_password):
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
# 密码哈希函数
|
|
def get_password_hash(password):
|
|
return pwd_context.hash(password)
|
|
|
|
# 根据用户名获取用户
|
|
def get_user(db: Session, username: str):
|
|
return db.query(User).filter(User.username == username).first()
|
|
|
|
# 根据邮箱获取用户
|
|
def get_user_by_email(db: Session, email: str):
|
|
return db.query(User).filter(User.email == email).first()
|
|
|
|
# 根据手机号获取用户
|
|
def get_user_by_phone(db: Session, phone_number: str):
|
|
return db.query(User).filter(User.phone_number == phone_number).first()
|
|
|
|
# 用户认证函数
|
|
def authenticate_user(db: Session, username: str, password: str):
|
|
user = get_user(db, username)
|
|
if not user:
|
|
return False
|
|
if not verify_password(password, user.hashed_password):
|
|
return False
|
|
return user
|
|
|
|
# 获取当前用户
|
|
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
|
try:
|
|
user_info = redis_client.get(f"session:{token}")
|
|
if not user_info:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="无效的token"
|
|
)
|
|
|
|
user_data = json.loads(user_info)
|
|
|
|
# 从Supabase获取最新的用户信息
|
|
response = supabase.table('users').select('*').eq('id', user_data['id']).execute()
|
|
|
|
if not response.data:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="用户不存在"
|
|
)
|
|
|
|
return response.data[0]
|
|
|
|
except Exception as e:
|
|
print(f"获取当前用户失败: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="认证失败"
|
|
)
|
|
|
|
# Supabase配置
|
|
supabase_url = os.getenv("SUPABASE_URL")
|
|
supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
|
|
supabase: Client = create_client(supabase_url, supabase_key)
|
|
|
|
# 用户注册路由
|
|
@user_app.post("/register")
|
|
async def register(user: UserCreate):
|
|
try:
|
|
# 随机选择一个头像
|
|
avatars = [
|
|
'picture/cn__02504_.png', 'picture/cn__02505_.png', 'picture/cn__02506_.png',
|
|
'picture/cn__02507_.png', 'picture/cn__02510_.png', 'picture/cn__02511_.png',
|
|
'picture/cn__02512_.png', 'picture/cn__03018_.png', 'picture/cn__03040_.png',
|
|
'picture/cn__03041_.png', 'picture/cn__03042_.png', 'picture/cn__03044_.png',
|
|
'picture/cn__03059_.png'
|
|
]
|
|
random_avatar = random.choice(avatars)
|
|
|
|
# 检查用户名是否已存在
|
|
user_exists = supabase.table('users').select('*').eq('username', user.username).execute()
|
|
if user_exists.data:
|
|
raise HTTPException(status_code=400, detail="用户名已被注册")
|
|
|
|
# 检查邮箱是否已存在
|
|
email_exists = supabase.table('users').select('*').eq('email', user.email).execute()
|
|
if email_exists.data:
|
|
raise HTTPException(status_code=400, detail="邮箱已被注册")
|
|
|
|
# 检查手机号是否已存在
|
|
phone_exists = supabase.table('users').select('*').eq('phone_number', user.phone_number).execute()
|
|
if phone_exists.data:
|
|
raise HTTPException(status_code=400, detail="手机号已被注册")
|
|
|
|
# 创建用户认证
|
|
auth_response = supabase.auth.admin.create_user({
|
|
"email": user.email,
|
|
"phone": user.phone_number,
|
|
"password": user.password,
|
|
"email_confirm": False # 改为false,需要用户验证邮箱
|
|
})
|
|
|
|
if not auth_response.user:
|
|
raise HTTPException(status_code=400, detail="创建用户失败")
|
|
|
|
# 等待一小段时间确保触发器执行完成
|
|
await asyncio.sleep(1)
|
|
|
|
# 更新users表中的用户记录
|
|
user_data = {
|
|
"username": user.username,
|
|
"phone_number": user.phone_number,
|
|
"avatar": random_avatar,
|
|
"is_active": False, # 改为false,等待邮箱验证后激活
|
|
"email_verified": False,
|
|
"phone_verified": False,
|
|
"user_password": user.password,
|
|
}
|
|
|
|
response = supabase.table('users').update(user_data).eq('id', auth_response.user.id).execute()
|
|
|
|
if not response.data:
|
|
# 如果更新用户记录失败,删除认证用户
|
|
supabase.auth.admin.delete_user(auth_response.user.id)
|
|
raise HTTPException(status_code=400, detail="更新用户记录失败")
|
|
|
|
# 将用户信息保存到Redis用于缓存
|
|
redis_user_info = {
|
|
"id": auth_response.user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"phone_number": user.phone_number,
|
|
"avatar": random_avatar,
|
|
"is_active": "false",
|
|
"email_verified": "false",
|
|
"phone_verified": "false",
|
|
"user_password": user.password, # 使用user_password
|
|
"created_at": datetime.now(timezone.utc).isoformat()
|
|
}
|
|
redis_register_client.hmset(f"user:{auth_response.user.id}", redis_user_info)
|
|
redis_register_client.expire(f"user:{auth_response.user.id}", 2592000)
|
|
|
|
return {"message": "用户创建成功"}
|
|
|
|
except Exception as e:
|
|
print(f"注册用户时出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"注册失败: {str(e)}")
|
|
|
|
# 用户登录路由
|
|
@user_app.post("/token")
|
|
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
|
|
try:
|
|
# 首先从users表获取用户信息
|
|
user_response = supabase.table('users').select('*').eq('email', form_data.username).execute()
|
|
|
|
if not user_response.data:
|
|
print(f"用户不存在: {form_data.username}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="用户名或密码不正确",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user_data = user_response.data[0]
|
|
|
|
# 验证密码是否匹配
|
|
if user_data['user_password'] != form_data.password:
|
|
print(f"密码不匹配: {form_data.username}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="用户名或密码不正确",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# 创建会话ID
|
|
session_id = str(uuid.uuid4())
|
|
|
|
# 存储用户信息到Redis
|
|
redis_user_info = {
|
|
"id": user_data['id'],
|
|
"username": user_data['username'],
|
|
"email": user_data['email'],
|
|
"is_active": str(user_data['is_active']),
|
|
"email_verified": str(user_data['email_verified']),
|
|
"phone_verified": str(user_data['phone_verified']),
|
|
"user_password": user_data['user_password']
|
|
}
|
|
redis_client.setex(f"session:{session_id}", 604800, json.dumps(redis_user_info))
|
|
|
|
# 更新最后登录时间
|
|
supabase.table('users').update({
|
|
"last_login": datetime.now(timezone.utc).isoformat()
|
|
}).eq('id', user_data['id']).execute()
|
|
|
|
return {
|
|
"access_token": session_id,
|
|
"token_type": "bearer",
|
|
"user": {
|
|
"id": user_data['id'],
|
|
"username": user_data['username'],
|
|
"email": user_data['email'],
|
|
"avatar": user_data['avatar'],
|
|
"is_active": user_data['is_active'],
|
|
"email_verified": user_data['email_verified'],
|
|
"phone_verified": user_data['phone_verified']
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"登录失败: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="用户名或密码不正确",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
# 获取当前用户信息���由
|
|
@user_app.get("/me")
|
|
async def read_users_me(current_user: User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
# 添加角色枚举类
|
|
class UserRole(str, Enum):
|
|
ADMIN = "admin"
|
|
USER = "user"
|
|
|
|
# 添加用户更新模型
|
|
class UserUpdate(BaseModel):
|
|
username: str
|
|
role: UserRole
|
|
|
|
# 修改编辑用户路由
|
|
@user_app.put("/edit/{user_id}")
|
|
async def edit_user(
|
|
user_id: str,
|
|
user_update: UserUpdate,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
try:
|
|
# 检查当前用户是否有权限编辑（只有管理员可以编辑）
|
|
if current_user.get('role') != 'admin':
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="没有权限执行此操作"
|
|
)
|
|
|
|
# 检查要编辑的用户是否存在
|
|
user_response = supabase.table('users').select('*').eq('id', user_id).execute()
|
|
if not user_response.data:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="用户不存在"
|
|
)
|
|
|
|
# 更新用户信息
|
|
update_response = supabase.table('users').update({
|
|
'username': user_update.username,
|
|
'role': user_update.role
|
|
}).eq('id', user_id).execute()
|
|
|
|
if not update_response.data:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="更新用户信息失败"
|
|
)
|
|
|
|
return {"message": "用户信息更新成功"}
|
|
|
|
except Exception as e:
|
|
print(f"编辑用户时出错: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"更新用户信息失败: {str(e)}"
|
|
)
|
|
|
|
# 修改删除用户路由
|
|
@user_app.delete("/api/users/{user_id}")
|
|
async def delete_user(
|
|
user_id: str,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
try:
|
|
# 检查当前用户是否有权限删除（只有管理员可以删除）
|
|
if current_user.get('role') != 'admin':
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="没有权限执行此操作"
|
|
)
|
|
|
|
# 检查要删除的用户是否存在
|
|
user_response = supabase.table('users').select('*').eq('id', user_id).execute()
|
|
if not user_response.data:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="用户不存在"
|
|
)
|
|
|
|
# 删除用户认证
|
|
auth_response = supabase.auth.admin.delete_user(user_id)
|
|
|
|
# 删除用户记录
|
|
delete_response = supabase.table('users').delete().eq('id', user_id).execute()
|
|
|
|
if not delete_response.data:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="删除用户失败"
|
|
)
|
|
|
|
return {"message": "用户删除成功"}
|
|
|
|
except Exception as e:
|
|
print(f"删除用户时出错: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"删除用户失败: {str(e)}"
|
|
)
|
|
|
|
# 重置密码路由
|
|
@user_app.post("/reset-password")
|
|
async def reset_password(
|
|
email: str = Body(...),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
user = get_user_by_email(db, email)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
# 这里应该实现发送重置密码邮件的逻辑
|
|
# 生成一个临时令牌，并将其与用户ID关联存储在Redis中
|
|
reset_token = secrets.token_urlsafe(32)
|
|
redis_client.setex(f"reset_password:{reset_token}", 3600, user.uuid) # 设置1小时过期
|
|
# 发送包含重置链接的邮件（这里省略具体实现）
|
|
return {"message": "密码重置说明已发送到邮箱"}
|
|
|
|
# 获取所有用户路由
|
|
@user_app.get("/api/users")
|
|
async def get_users():
|
|
try:
|
|
# 从 Supabase 获取所有用户，包括 role 字段
|
|
response = supabase.table('users').select('id,username,email,role').execute()
|
|
|
|
if not response.data:
|
|
return []
|
|
|
|
return [
|
|
{
|
|
"id": user['id'],
|
|
"username": user['username'],
|
|
"email": user['email'],
|
|
"role": user.get('role', 'user') # 如果 role 不存在，默认为 'user'
|
|
}
|
|
for user in response.data
|
|
]
|
|
|
|
except Exception as error:
|
|
print('获取用户时出错:', error)
|
|
raise HTTPException(status_code=500, detail=f"服务器内部错误: {str(error)}")
|
|
|
|
# 用户登出路由
|
|
@user_app.post("/logout")
|
|
async def logout(token: str = Depends(oauth2_scheme)):
|
|
redis_client.delete(f"session:{token}")
|
|
return {"message": "成功登出"}
|
|
|
|
# 创建一个新的Pydantic模型来表示仪表盘数据
|
|
class RecentUser(BaseModel):
|
|
username: str
|
|
created_at: datetime
|
|
|
|
class DashboardData(BaseModel):
|
|
total_users: int
|
|
active_users: int
|
|
new_users_today: int
|
|
recent_users: List[RecentUser]
|
|
|
|
@user_app.get("/dashboard", response_model=DashboardData)
|
|
async def get_dashboard_data(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
if not current_user:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="没有访问权限")
|
|
|
|
# 从 Supabase 获取用户统计信息
|
|
total_users = len((supabase.table('users').select('id').execute()).data)
|
|
|
|
# 获取30天内活跃用户
|
|
thirty_days_ago = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
|
|
active_users = len((supabase.table('users')
|
|
.select('id')
|
|
.gte('last_login', thirty_days_ago)
|
|
.execute()).data)
|
|
|
|
# 获取今天新注册的用户
|
|
today = datetime.now(timezone.utc).date().isoformat()
|
|
new_users_today = len((supabase.table('users')
|
|
.select('id')
|
|
.gte('created_at', today)
|
|
.execute()).data)
|
|
|
|
# 获取最近注册的10个用户
|
|
recent_users_response = (supabase.table('users')
|
|
.select('username,created_at')
|
|
.order('created_at', desc=True)
|
|
.limit(10)
|
|
.execute())
|
|
|
|
recent_users_data = [
|
|
RecentUser(
|
|
username=user['username'],
|
|
created_at=datetime.fromisoformat(user['created_at'].replace('Z', '+00:00'))
|
|
)
|
|
for user in recent_users_response.data
|
|
]
|
|
|
|
dashboard_data = DashboardData(
|
|
total_users=total_users,
|
|
active_users=active_users,
|
|
new_users_today=new_users_today,
|
|
recent_users=recent_users_data
|
|
)
|
|
|
|
return dashboard_data
|
|
|
|
except Exception as e:
|
|
print(f"获取仪表盘数据时出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="获取仪表盘数据失败")
|
|
|
|
@user_app.get("/user-stats")
|
|
async def get_user_stats(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
if not current_user:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="没有访问权限")
|
|
|
|
# 获取过去7天的日期列表
|
|
today = datetime.now(timezone.utc).date()
|
|
dates = [(today - timedelta(days=i)).isoformat() for i in range(6, -1, -1)]
|
|
|
|
# 查询每天的新用户数
|
|
daily_new_users = []
|
|
for date in dates:
|
|
next_date = (datetime.fromisoformat(date) + timedelta(days=1)).isoformat()
|
|
count = len((supabase.table('users')
|
|
.select('id')
|
|
.gte('created_at', date)
|
|
.lt('created_at', next_date)
|
|
.execute()).data)
|
|
daily_new_users.append(count)
|
|
|
|
stats = {
|
|
"dates": dates,
|
|
"daily_new_users": daily_new_users,
|
|
"total_users": len((supabase.table('users').select('id').execute()).data)
|
|
}
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
print(f"获取用户统计数据时出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="获取用户统计数据失败")
|
|
|
|
|
|
# 定义API密钥头部
|
|
api_key_header = APIKeyHeader(name="Authorization", auto_error=False)
|
|
|
|
# 定义 base62 字符集
|
|
BASE62 = string.digits + string.ascii_letters
|
|
|
|
# UUID 转 base62 函数
|
|
def uuid_to_base62(uuid_value):
|
|
uuid_int = uuid.UUID(uuid_value).int
|
|
result = ""
|
|
while uuid_int:
|
|
uuid_int, remainder = divmod(uuid_int, 62)
|
|
result = BASE62[remainder] + result
|
|
return result.rjust(22, '0') # 补齐到22位
|
|
|
|
# 修改生成API密钥的函数
|
|
def generate_api_key():
|
|
uuid_value = str(uuid.uuid4())
|
|
base62_value = uuid_to_base62(uuid_value)
|
|
return f"obs-{base62_value}"
|
|
|
|
# 验证API密钥的函数
|
|
async def get_api_key(api_key: str = Security(api_key_header)):
|
|
if api_key and api_key.startswith("Bearer "):
|
|
key = api_key.split(" ")[1]
|
|
if key.startswith("obs-"):
|
|
return key
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="无效的API密钥",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# 修改创建API密钥的路由
|
|
@user_app.post("/api-keys")
|
|
async def create_api_key(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
new_key = generate_api_key()
|
|
created_at = datetime.now(timezone.utc)
|
|
expires_at = created_at + timedelta(days=30)
|
|
|
|
# 在Redis中初始化API密钥信息
|
|
redis_api_client.hset(f"api_key:{new_key}", mapping={
|
|
"user_id": current_user['id'],
|
|
"is_active": "1",
|
|
"created_at": created_at.isoformat(),
|
|
"expires_at": expires_at.isoformat(),
|
|
"total_tokens": 10000000000000,
|
|
})
|
|
redis_api_usage_client.hset(f"api_key:{new_key}", mapping={
|
|
"user_id": current_user['id'],
|
|
"is_active": "1",
|
|
"created_at": created_at.isoformat(),
|
|
"expires_at": expires_at.isoformat(),
|
|
"total_tokens": 10000000000000,
|
|
"tokens_used": 0,
|
|
})
|
|
|
|
# 添加到过期集合
|
|
redis_api_client.zadd("expiring_keys", {new_key: expires_at.timestamp()})
|
|
|
|
return {"api_key": new_key, "expires_at": expires_at}
|
|
except Exception as e:
|
|
print(f"创建API密钥时出错: {e}")
|
|
raise HTTPException(status_code=500, detail="服务器内部错误")
|
|
|
|
|
|
# 获取 API 使用量统计的路由
|
|
@user_app.get("/api-usage")
|
|
async def get_api_usage(current_user: dict = Depends(get_current_user)):
|
|
keys = redis_api_usage_client.keys(f"api_key:*")
|
|
usage_stats = {}
|
|
for key in keys:
|
|
key_info = redis_api_usage_client.hgetall(key)
|
|
if key_info and key_info.get(b'user_id', b'').decode() == current_user['id']:
|
|
api_key = key.decode().split(':')[1] # 提取 obs-xxx 部分
|
|
usage_stats[api_key] = {
|
|
"total_tokens": int(key_info.get(b"total_tokens", 0)),
|
|
"tokens_used": int(key_info.get(b"tokens_used", 0)),
|
|
"created_at": key_info.get(b"created_at", b"").decode(),
|
|
"expires_at": key_info.get(b"expires_at", b"").decode(),
|
|
"last_used_at": key_info.get(b"last_used_at", b"").decode(),
|
|
}
|
|
return usage_stats
|
|
|
|
# 更新 API 使用量的函数
|
|
async def update_api_usage(api_key: str, tokens_used: int, endpoint: str):
|
|
now = datetime.now(timezone.utc)
|
|
today = now.date().isoformat()
|
|
redis_key = f"api_key:{api_key}"
|
|
pipe = redis_api_usage_client.pipeline() # 使用 redis_api_usage_client
|
|
pipe.hincrby(redis_key, "tokens_used", tokens_used)
|
|
pipe.hset(redis_key, "last_used_at", now.isoformat())
|
|
pipe.incrby(f"daily_usage:{api_key}:{today}", tokens_used)
|
|
pipe.execute()
|
|
|
|
@user_app.get("/model-usage")
|
|
async def get_model_usage(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
keys = redis_api_usage_client.keys(f"api_key:*")
|
|
usage_stats = {}
|
|
for key in keys:
|
|
key_info = redis_api_usage_client.hgetall(key)
|
|
if key_info and key_info.get(b'user_id', b'').decode() == current_user['id']:
|
|
for field, value in key_info.items():
|
|
field_str = field.decode()
|
|
if field_str.endswith("_tokens_used"):
|
|
model_name = field_str.replace("_tokens_used", "")
|
|
if model_name not in usage_stats:
|
|
usage_stats[model_name] = {
|
|
"calls": 0,
|
|
"last_called_at": None
|
|
}
|
|
usage_stats[model_name]["calls"] += int(value)
|
|
last_called_at = key_info.get(f"{model_name}_last_used_at".encode(), b"").decode()
|
|
if last_called_at:
|
|
if usage_stats[model_name]["last_called_at"] is None or last_called_at > usage_stats[model_name]["last_called_at"]:
|
|
usage_stats[model_name]["last_called_at"] = last_called_at
|
|
|
|
return usage_stats
|
|
except Exception as e:
|
|
print(f"获取模型使用情况时出错: {e}")
|
|
raise HTTPException(status_code=500, detail="服务器内部错误")
|
|
|
|
|
|
|
|
# 添加清除缓存的路由
|
|
@user_app.post("/clear-cache")
|
|
async def clear_cache(current_user: User = Depends(get_current_user)):
|
|
try:
|
|
# 获取当前用户的所有API密钥
|
|
user_keys = redis_api_client.keys(f"api_key:*")
|
|
updated_keys = []
|
|
removed_keys = []
|
|
|
|
for key in user_keys:
|
|
key_info = redis_api_client.hgetall(key)
|
|
if key_info and key_info.get(b'user_id', b'').decode() == current_user.uuid:
|
|
api_key = key.decode().split(':')[1]
|
|
|
|
if not key_info:
|
|
# 如果 Redis 中已经没有这个 API 密钥的信息，将其添加到需要移除的列表中
|
|
removed_keys.append(api_key)
|
|
else:
|
|
# 重置使用统计信息
|
|
redis_api_client.hset(key, "total_tokens", "0")
|
|
redis_api_client.hset(key, "tokens_used", "0")
|
|
|
|
# 清除每日使用量统计
|
|
daily_usage_keys = redis_api_usage_client.keys(f"daily_usage:{api_key}:*")
|
|
if daily_usage_keys:
|
|
redis_api_usage_client.delete(*daily_usage_keys)
|
|
|
|
updated_keys.append(api_key)
|
|
|
|
return {
|
|
"message": "API 使用统计已成功重置",
|
|
"updated_keys": updated_keys,
|
|
"removed_keys": removed_keys
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"重置缓存时发生错误: {str(e)}")
|
|
|
|
@user_app.get("/api-keys")
|
|
async def get_api_keys(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
keys = redis_api_client.keys(f"api_key:*")
|
|
valid_keys = []
|
|
for key in keys:
|
|
key_info = redis_api_client.hgetall(key)
|
|
if key_info and key_info.get(b'user_id', b'').decode() == current_user['id']: # 使用 current_user['id'] 替代 current_user.uuid
|
|
valid_keys.append({
|
|
"key": key.decode().replace("api_key:", ""),
|
|
"created_at": key_info[b"created_at"].decode(),
|
|
"expires_at": key_info[b"expires_at"].decode()
|
|
})
|
|
return valid_keys
|
|
except Exception as e:
|
|
print(f"获取API密钥时出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="获取API密钥失败")
|
|
|
|
@user_app.delete("/api-keys/{key_id}")
|
|
async def revoke_api_key(key_id: str, current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
# 确保 key_id 的格式正确
|
|
if not key_id.startswith("api_key:"):
|
|
key_id = f"api_key:{key_id}"
|
|
|
|
# 从 Redis 获取 API 密信息
|
|
key_info = redis_api_client.hgetall(key_id)
|
|
|
|
if not key_info:
|
|
raise HTTPException(status_code=404, detail="API 密钥未找到")
|
|
|
|
# 检查 API 密钥是否属于当前用户
|
|
if key_info.get(b'user_id', b'').decode() != current_user['id']: # 使用 current_user['id'] 替代 current_user.uuid
|
|
raise HTTPException(status_code=403, detail="无权删除此 API 密钥")
|
|
|
|
# 删除 API 密钥
|
|
deleted = redis_api_client.delete(key_id)
|
|
|
|
# 从 expiring_keys 中移除该密钥
|
|
redis_api_client.zrem("expiring_keys", key_id.replace("api_key:", ""))
|
|
|
|
if deleted:
|
|
return {"message": "API 密钥已成功删除", "action": "clear_api_info"}
|
|
else:
|
|
raise HTTPException(status_code=500, detail="删除 API 密钥时发生错误")
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
print(f"删除API密钥时出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="删除API密钥失败")
|
|
|
|
|
|
@user_app.post("/send-verification-email")
|
|
async def send_verification_email_route(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
# 检查用户是否已经验证过邮箱
|
|
if current_user.get('email_verified'):
|
|
raise HTTPException(status_code=400, detail="邮箱已经验证过了")
|
|
|
|
# 使用 OTP 发送验证邮件
|
|
auth_response = supabase.auth.sign_in_with_otp({
|
|
"email": current_user['email'],
|
|
"options": {
|
|
"shouldCreateUser": False, # 不要创建新用户
|
|
}
|
|
})
|
|
|
|
return {"message": "验证邮件已发送，请查收"}
|
|
|
|
except Exception as e:
|
|
print(f"发送验证邮件过程中发生错误: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="发送验证邮件失败")
|
|
|
|
@user_app.post("/verify-email")
|
|
async def verify_email(verification_code: str = Form(...), current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
# 使用 OTP 验证邮箱
|
|
verify_response = supabase.auth.verify_otp({
|
|
"email": current_user['email'],
|
|
"token": verification_code,
|
|
"type": 'email' # 使用 email 类型
|
|
})
|
|
|
|
if not verify_response:
|
|
raise HTTPException(status_code=400, detail="验证码无效或已过期")
|
|
|
|
try:
|
|
# 直接更新users表中的验证状态
|
|
update_response = supabase.table('users').update({
|
|
'email_verified': True,
|
|
'is_active': True
|
|
}).eq('id', current_user['id']).execute()
|
|
|
|
if not update_response.data:
|
|
raise HTTPException(status_code=500, detail="更新用户状态失败")
|
|
|
|
# 更新Redis中的用户会话信息
|
|
session_keys = redis_client.keys("session:*")
|
|
for session_key in session_keys:
|
|
user_info = json.loads(redis_client.get(session_key))
|
|
if user_info.get("id") == current_user['id']:
|
|
user_info["is_active"] = "true"
|
|
user_info["email_verified"] = "true"
|
|
redis_client.setex(session_key, 604800, json.dumps(user_info))
|
|
|
|
# 获取更新后的用户信息
|
|
updated_user = supabase.table('users').select('*').eq('id', current_user['id']).execute()
|
|
|
|
if not updated_user.data:
|
|
raise HTTPException(status_code=404, detail="无法获取更新后的用户信息")
|
|
|
|
return {
|
|
"message": "邮箱验证成功",
|
|
"user": {
|
|
"id": updated_user.data[0]['id'],
|
|
"username": updated_user.data[0]['username'],
|
|
"email": updated_user.data[0]['email'],
|
|
"is_active": updated_user.data[0]['is_active'],
|
|
"email_verified": updated_user.data[0]['email_verified'],
|
|
"avatar": updated_user.data[0]['avatar']
|
|
}
|
|
}
|
|
|
|
except Exception as supabase_error:
|
|
print(f"更新用户状态失败: {str(supabase_error)}")
|
|
raise HTTPException(status_code=500, detail="更新验证状态失败")
|
|
|
|
except HTTPException as http_error:
|
|
raise http_error
|
|
except Exception as e:
|
|
print(f"验证邮箱过程中发生错误: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="验证邮箱失败")
|
|
|
|
# 获取单个用户信息路由
|
|
@user_app.get("/api/users/{user_id}")
|
|
async def get_user_by_id(
|
|
user_id: str,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
try:
|
|
# 检查当前用户是否有权限（只有管理员可以查看用户详情）
|
|
if current_user.get('role') != 'admin':
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="没有权限执行此操作"
|
|
)
|
|
|
|
# 从 Supabase 获取用户信息
|
|
response = supabase.table('users').select('*').eq('id', user_id).execute()
|
|
|
|
if not response.data:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="用户不存在"
|
|
)
|
|
|
|
user_data = response.data[0]
|
|
return {
|
|
"id": user_data['id'],
|
|
"username": user_data['username'],
|
|
"email": user_data['email'],
|
|
"role": user_data.get('role', 'user') # 如果 role 不存在，默认为 'user'
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as error:
|
|
print('获取用户信息时出错:', error)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"获取用户信息失败: {str(error)}"
|
|
)
|
|
|
|
# 在现有路由之前添加新的搜索路由
|
|
@user_app.get("/api/users-search")
|
|
async def search_users(
|
|
query: str,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
try:
|
|
# 检查权限
|
|
if current_user.get('role') != 'admin':
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="没有权限执行此操作"
|
|
)
|
|
|
|
# 构建搜索查询
|
|
search_query = supabase.table('users').select('id,username,email,role')
|
|
|
|
# 如果有搜索词，添加过滤条件
|
|
if query:
|
|
search_query = search_query.or_(
|
|
f"username.ilike.%{query}%,email.ilike.%{query}%,role.ilike.%{query}%"
|
|
)
|
|
|
|
response = search_query.execute()
|
|
|
|
if not response.data:
|
|
return []
|
|
|
|
return [
|
|
{
|
|
"id": user['id'],
|
|
"username": user['username'],
|
|
"email": user['email'],
|
|
"role": user.get('role', 'user')
|
|
}
|
|
for user in response.data
|
|
]
|
|
|
|
except Exception as error:
|
|
print('搜索用户时出错:', error)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"搜索用户失败: {str(error)}"
|
|
)
|
|
|
|
# 主程序入口
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |