| import logging |
| import re |
| from passlib.context import CryptContext |
| from datetime import datetime, timedelta |
| import jwt |
| from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES |
|
|
| logger = logging.getLogger("swarai.utils") |
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") |
|
|
| def hash_password(password: str) -> str: |
| return pwd_context.hash(password) |
|
|
| def verify_password(plain: str, hashed: str) -> bool: |
| try: |
| return pwd_context.verify(plain, hashed) |
| except Exception: |
| logger.exception("Password verify failed") |
| return False |
|
|
| def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str: |
| to_encode = {"sub": subject} |
| expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) |
| to_encode.update({"exp": expire.isoformat()}) |
| encoded = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) |
| return encoded |
|
|
| def decode_token(token: str) -> dict: |
| try: |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) |
| return payload |
| except Exception: |
| logger.exception("Token decode failed") |
| raise |
|
|
| def normalize_text(s: str) -> str: |
| if not s: |
| return "" |
| s = s.strip().lower() |
| |
| s = re.sub(r"[^\w\s']", " ", s) |
| s = re.sub(r"\s+", " ", s) |
| return s |
|
|
