df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""
|
||
app/services/auth_service.py — 认证 service
|
||
密码哈希校验、用户查找、响应格式化。
|
||
路由层不含业务逻辑，全在此。
|
||
"""
|
||
|
||
import logging
|
||
|
||
from passlib.context import CryptContext
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.response import raise_unauthorized
|
||
from app.middleware.workspace_guard import CurrentUser
|
||
from app.models.user import User
|
||
from app.models.workspace import WorkspaceMember
|
||
|
||
logger = logging.getLogger(__name__)
|
||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
DEFAULT_SEED_PASSWORD = "Clover2026!"
|
||
|
||
|
||
def hash_password(plain: str) -> str:
|
||
return pwd_context.hash(plain)
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
return pwd_context.verify(plain, hashed)
|
||
|
||
|
||
def authenticate_user(
|
||
db: Session, username: str, password: str
|
||
) -> tuple[User, int, str]:
|
||
"""
|
||
验证用户名+密码，返回 (user, workspace_id, role)。
|
||
失败抛 CloverHTTPException 40101。
|
||
"""
|
||
user = db.query(User).filter(
|
||
User.username == username, User.is_active == True
|
||
).first()
|
||
if not user or not verify_password(password, user.hashed_password):
|
||
raise_unauthorized("用户名或密码错误")
|
||
if get_settings().APP_ENV == "production" and password == DEFAULT_SEED_PASSWORD:
|
||
raise_unauthorized("默认密码已禁用，请联系管理员重置密码")
|
||
|
||
# 取用户所在的第一个 workspace（手动建账号场景只有一个）
|
||
member = (
|
||
db.query(WorkspaceMember)
|
||
.filter(WorkspaceMember.user_id == user.id)
|
||
.first()
|
||
)
|
||
if not member:
|
||
raise_unauthorized("用户未加入任何 workspace，请联系管理员")
|
||
|
||
# 记录登录
|
||
try:
|
||
from app.models.user import LoginRecord
|
||
db.add(LoginRecord(user_id=user.id))
|
||
db.commit()
|
||
except Exception:
|
||
logger.warning("Failed to write login_record for user=%s", user.id)
|
||
db.rollback()
|
||
|
||
return user, member.workspace_id, member.role
|
||
|
||
|
||
def build_user_response(user: User, workspace_id: int, role: str) -> dict:
|
||
"""格式化用户响应体（契约§4 DTO）。"""
|
||
return {
|
||
"id": user.id,
|
||
"username": user.username,
|
||
"email": user.email,
|
||
"current_workspace_id": workspace_id,
|
||
"role": role,
|
||
"must_change_password": bool(getattr(user, "must_change_password", False)),
|
||
}
|
||
|
||
|
||
def change_password(db: Session, user_id: int, current_password: str, new_password: str) -> None:
|
||
user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||
if not user or not verify_password(current_password, user.hashed_password):
|
||
raise_unauthorized("当前密码错误")
|
||
if len(new_password) < 8:
|
||
from app.core.response import raise_param_error
|
||
raise_param_error("新密码至少 8 位")
|
||
if new_password == DEFAULT_SEED_PASSWORD:
|
||
from app.core.response import raise_param_error
|
||
raise_param_error("不能使用默认密码")
|
||
user.hashed_password = hash_password(new_password)
|
||
user.must_change_password = False
|
||
db.commit()
|