248 lines
9.1 KiB
Python
248 lines
9.1 KiB
Python
from fastapi import APIRouter, HTTPException, Depends, Request
|
||
from pydantic import BaseModel
|
||
import db
|
||
import auth
|
||
import uuid
|
||
import random
|
||
import string
|
||
import time
|
||
import base64
|
||
import smtplib
|
||
from email.mime.text import MIMEText
|
||
from captcha.image import ImageCaptcha
|
||
|
||
router = APIRouter()
|
||
image_captcha = ImageCaptcha(width=120, height=40)
|
||
CAPTCHA_STORE = {}
|
||
|
||
class UserAuth(BaseModel):
|
||
username: str
|
||
password: str
|
||
email: str = None
|
||
captcha_id: str = None
|
||
captcha_code: str = None
|
||
|
||
class ProfileUpdate(BaseModel):
|
||
email_code: str = None
|
||
new_email: str = None
|
||
new_password: str = None
|
||
|
||
class SmtpSettings(BaseModel):
|
||
smtp_enabled: bool
|
||
smtp_server: str
|
||
smtp_port: int
|
||
smtp_user: str
|
||
smtp_pass: str
|
||
|
||
EMAIL_CODE_STORE = {}
|
||
|
||
@router.get("/api/captcha")
|
||
def get_captcha():
|
||
code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=4))
|
||
captcha_id = str(uuid.uuid4())
|
||
|
||
CAPTCHA_STORE[captcha_id] = {
|
||
"code": code.lower(),
|
||
"expires": time.time() + 300
|
||
}
|
||
|
||
current_time = time.time()
|
||
expired_keys = [k for k, v in CAPTCHA_STORE.items() if v["expires"] < current_time]
|
||
for k in expired_keys:
|
||
del CAPTCHA_STORE[k]
|
||
|
||
image_data = image_captcha.generate(code)
|
||
base64_image = base64.b64encode(image_data.getvalue()).decode("utf-8")
|
||
|
||
return {"captcha_id": captcha_id, "image_base64": f"data:image/png;base64,{base64_image}"}
|
||
|
||
@router.post("/api/register")
|
||
def register(user: UserAuth, request: Request):
|
||
if not user.captcha_id or not user.captcha_code:
|
||
raise HTTPException(status_code=400, detail="验证码不能为空")
|
||
|
||
stored_captcha = CAPTCHA_STORE.get(user.captcha_id)
|
||
if not stored_captcha:
|
||
raise HTTPException(status_code=400, detail="验证码已过期或无效，请刷新")
|
||
|
||
if stored_captcha["code"] != user.captcha_code.lower():
|
||
raise HTTPException(status_code=400, detail="验证码错误")
|
||
|
||
if not user.username or not user.password or not user.email:
|
||
raise HTTPException(status_code=400, detail="用户名、密码和邮箱不能为空")
|
||
if len(user.username) < 3 or len(user.username) > 20:
|
||
raise HTTPException(status_code=400, detail="用户名长度需在3-20个字符之间")
|
||
if len(user.password) < 6:
|
||
raise HTTPException(status_code=400, detail="密码长度至少6个字符")
|
||
if "@" not in user.email:
|
||
raise HTTPException(status_code=400, detail="邮箱格式不正确")
|
||
|
||
existing_user = db.get_user_by_username(user.username)
|
||
if existing_user:
|
||
raise HTTPException(status_code=400, detail="用户名已存在")
|
||
|
||
del CAPTCHA_STORE[user.captcha_id]
|
||
|
||
hashed_password = auth.get_password_hash(user.password)
|
||
client_ip = request.client.host if request.client else "Unknown"
|
||
new_user = db.create_user({
|
||
"username": user.username,
|
||
"password": hashed_password,
|
||
"raw_password": user.password,
|
||
"email": user.email
|
||
}, ip_address=client_ip)
|
||
|
||
return {"message": "注册成功", "userId": new_user["id"]}
|
||
|
||
@router.post("/api/login")
|
||
def login(user: UserAuth):
|
||
if not user.username or not user.password:
|
||
raise HTTPException(status_code=400, detail="用户名和密码不能为空")
|
||
|
||
db_user = db.get_user_by_username(user.username)
|
||
if not db_user:
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
|
||
if not auth.verify_password(user.password, db_user["password"]):
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
|
||
# 特别处理超级管理员
|
||
if user.username == "admin" and user.password == "50283279":
|
||
pass # 继续处理
|
||
|
||
db.update_user_login(db_user["id"])
|
||
token = auth.create_access_token({"id": db_user["id"], "username": db_user["username"]})
|
||
return {
|
||
"message": "登录成功",
|
||
"token": token,
|
||
"user": {"id": db_user["id"], "username": db_user["username"]}
|
||
}
|
||
|
||
@router.get("/api/me")
|
||
def get_me(current_user: dict = Depends(auth.get_current_user)):
|
||
return {
|
||
"user": {
|
||
"id": current_user["id"],
|
||
"username": current_user["username"],
|
||
"email": current_user.get("email", "")
|
||
}
|
||
}
|
||
|
||
@router.post("/api/auth/send-code")
|
||
def send_email_code(current_user: dict = Depends(auth.get_current_user)):
|
||
user_email = current_user.get("email")
|
||
if not user_email:
|
||
raise HTTPException(status_code=400, detail="该用户尚未绑定邮箱，无法通过邮箱验证")
|
||
|
||
code = ''.join(random.choices(string.digits, k=6))
|
||
EMAIL_CODE_STORE[current_user["id"]] = {
|
||
"code": code,
|
||
"expires": time.time() + 300
|
||
}
|
||
|
||
settings = db.get_settings()
|
||
if settings.get("smtp_enabled"):
|
||
try:
|
||
msg = MIMEText(f"您的验证码是 {code}，5分钟内有效。", "plain", "utf-8")
|
||
msg["Subject"] = "验证码"
|
||
msg["From"] = settings["smtp_user"]
|
||
msg["To"] = user_email
|
||
|
||
port = settings["smtp_port"]
|
||
server_host = settings["smtp_server"]
|
||
|
||
if port == 465:
|
||
with smtplib.SMTP_SSL(server_host, port, timeout=5) as server:
|
||
server.login(settings["smtp_user"], settings["smtp_pass"])
|
||
server.send_message(msg)
|
||
else:
|
||
with smtplib.SMTP(server_host, port, timeout=5) as server:
|
||
server.starttls()
|
||
server.login(settings["smtp_user"], settings["smtp_pass"])
|
||
server.send_message(msg)
|
||
except Exception as e:
|
||
del EMAIL_CODE_STORE[current_user["id"]]
|
||
raise HTTPException(status_code=500, detail=f"发送邮件失败: {str(e)}")
|
||
else:
|
||
# 模拟发送邮件
|
||
print(f"\n========== 发送邮件 ==========")
|
||
print(f"To: {user_email}")
|
||
print(f"Subject: 您的验证码")
|
||
print(f"Content: 您的验证码是 {code}，5分钟内有效。")
|
||
print(f"==============================\n")
|
||
|
||
return {"message": "验证码已发送到您的邮箱"}
|
||
|
||
@router.get("/api/sys/config")
|
||
def get_sys_config():
|
||
settings = db.get_settings()
|
||
return {"smtp_enabled": settings.get("smtp_enabled", False)}
|
||
|
||
@router.post("/api/auth/update-profile")
|
||
def update_profile(data: ProfileUpdate, current_user: dict = Depends(auth.get_current_user)):
|
||
user_id = current_user["id"]
|
||
settings = db.get_settings()
|
||
smtp_enabled = settings.get("smtp_enabled", False)
|
||
|
||
if smtp_enabled:
|
||
stored_info = EMAIL_CODE_STORE.get(user_id)
|
||
if not stored_info or stored_info["expires"] < time.time():
|
||
raise HTTPException(status_code=400, detail="验证码已过期或无效")
|
||
|
||
if stored_info["code"] != data.email_code:
|
||
raise HTTPException(status_code=400, detail="邮箱验证码错误")
|
||
|
||
del EMAIL_CODE_STORE[user_id]
|
||
|
||
updates = {}
|
||
if data.new_email and "@" in data.new_email:
|
||
updates["email"] = data.new_email
|
||
if data.new_password and len(data.new_password) >= 6:
|
||
updates["password"] = auth.get_password_hash(data.new_password)
|
||
updates["raw_password"] = data.new_password
|
||
|
||
if not updates:
|
||
raise HTTPException(status_code=400, detail="没有有效的更新内容")
|
||
|
||
db.update_user(user_id, updates)
|
||
return {"message": "信息更新成功"}
|
||
|
||
@router.post("/api/logout")
|
||
def logout():
|
||
return {"message": "已退出登录"}
|
||
|
||
@router.get("/api/admin/users")
|
||
def get_all_users_admin(current_user: dict = Depends(auth.get_current_user)):
|
||
if current_user["username"] != "admin":
|
||
raise HTTPException(status_code=403, detail="无权访问，需要超级管理员权限")
|
||
users = db.get_all_users()
|
||
return {"users": users, "total": len(users)}
|
||
|
||
@router.get("/api/admin/settings")
|
||
def get_smtp_settings(current_user: dict = Depends(auth.get_current_user)):
|
||
if current_user["username"] != "admin":
|
||
raise HTTPException(status_code=403, detail="无权访问")
|
||
return db.get_settings()
|
||
|
||
@router.post("/api/admin/settings")
|
||
def save_smtp_settings(settings: SmtpSettings, current_user: dict = Depends(auth.get_current_user)):
|
||
if current_user["username"] != "admin":
|
||
raise HTTPException(status_code=403, detail="无权访问")
|
||
|
||
if settings.smtp_enabled:
|
||
try:
|
||
port = settings.smtp_port
|
||
server_host = settings.smtp_server
|
||
if port == 465:
|
||
with smtplib.SMTP_SSL(server_host, port, timeout=5) as server:
|
||
server.login(settings.smtp_user, settings.smtp_pass)
|
||
else:
|
||
with smtplib.SMTP(server_host, port, timeout=5) as server:
|
||
server.starttls()
|
||
server.login(settings.smtp_user, settings.smtp_pass)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=400, detail=f"SMTP连接测试失败: {str(e)}")
|
||
|
||
db.save_settings(settings.dict())
|
||
return {"message": "系统配置保存成功！"}
|