from datetime import datetime, timedelta from typing import Any import bcrypt from jose import JWTError, jwt from app.core.config import get_settings settings = get_settings() def hash_password(password: str) -> str: """ 加密密码 Args: password: 明文密码 Returns: 加密后的密码 """ return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(plain_password: str, hashed_password: str) -> bool: """ 验证密码 Args: plain_password: 明文密码 hashed_password: 加密后的密码 Returns: 密码是否正确 """ return bcrypt.checkpw(plain_password.encode(), hashed_password.encode()) def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str: """ 创建访问令牌 Args: data: 要编码的数据 expires_delta: 过期时间(可选) Returns: JWT token """ to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.jwt_access_token_expire_minutes) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm) return encoded_jwt def decode_access_token(token: str) -> dict[str, Any] | None: """ 解码访问令牌 Args: token: JWT token Returns: 解码后的数据,解码失败返回 None """ try: payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm]) return payload except JWTError: return None