|
import secrets
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from passlib.hash import pbkdf2_sha256
|
|
from fastapi import Request, HTTPException, status
|
|
from devplacepy.database import get_table
|
|
from devplacepy.config import SECRET_KEY, SESSION_MAX_AGE
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return pbkdf2_sha256.hash(password)
|
|
|
|
|
|
def verify_password(password: str, hashed: str) -> bool:
|
|
return pbkdf2_sha256.verify(password, hashed)
|
|
|
|
|
|
def create_session(user_uid: str) -> str:
|
|
token = secrets.token_hex(32)
|
|
expires_at = datetime.utcnow() + timedelta(seconds=SESSION_MAX_AGE)
|
|
sessions = get_table("sessions")
|
|
sessions.insert({
|
|
"session_token": token,
|
|
"user_uid": user_uid,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"expires_at": expires_at.isoformat(),
|
|
})
|
|
return token
|
|
|
|
|
|
_user_cache = {}
|
|
|
|
|
|
def get_current_user(request: Request):
|
|
token = request.cookies.get("session")
|
|
if not token:
|
|
return None
|
|
|
|
cached = _user_cache.get(token)
|
|
if cached:
|
|
return cached
|
|
|
|
sessions = get_table("sessions")
|
|
session = sessions.find_one(session_token=token)
|
|
if not session:
|
|
return None
|
|
expires = datetime.fromisoformat(session["expires_at"])
|
|
if expires < datetime.utcnow():
|
|
sessions.delete(id=session["id"])
|
|
return None
|
|
users = get_table("users")
|
|
user = users.find_one(uid=session["user_uid"])
|
|
if user:
|
|
_user_cache[token] = user
|
|
return user
|
|
|
|
|
|
def require_user(request: Request):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/"})
|
|
return user
|
|
|
|
|
|
def slugify(text: str) -> str:
|
|
import re
|
|
text = text.lower().strip()
|
|
text = re.sub(r"[^a-z0-9-]", "-", text)
|
|
text = re.sub(r"-+", "-", text)
|
|
return text.strip("-")
|
|
|
|
|
|
def generate_uid() -> str:
|
|
import uuid
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
def time_ago(dt_str: str) -> str:
|
|
dt = datetime.fromisoformat(dt_str)
|
|
now = datetime.utcnow()
|
|
diff = now - dt
|
|
days = diff.days
|
|
if days > 365:
|
|
return f"{days // 365}y ago"
|
|
if days > 30:
|
|
return f"{days // 30}mo ago"
|
|
if days > 0:
|
|
return f"{days}d ago"
|
|
hours = diff.seconds // 3600
|
|
if hours > 0:
|
|
return f"{hours}h ago"
|
|
minutes = diff.seconds // 60
|
|
if minutes > 0:
|
|
return f"{minutes}m ago"
|
|
return "just now"
|