|
import re
|
|
import secrets
|
|
import time
|
|
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 = {}
|
|
_USER_CACHE_TTL = 300
|
|
|
|
|
|
def get_current_user(request: Request):
|
|
token = request.cookies.get("session")
|
|
if not token:
|
|
return None
|
|
|
|
cached = _user_cache.get(token)
|
|
if cached:
|
|
entry, expiry = cached
|
|
if time.time() < expiry:
|
|
return entry
|
|
_user_cache.pop(token, None)
|
|
|
|
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 and not user.get("is_active", True):
|
|
sessions.delete(id=session["id"])
|
|
_user_cache.pop(token, None)
|
|
return None
|
|
if user:
|
|
_user_cache[token] = (user, time.time() + _USER_CACHE_TTL)
|
|
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 require_admin(request: Request):
|
|
user = require_user(request)
|
|
if user.get("role") != "Admin":
|
|
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/feed"})
|
|
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 make_combined_slug(text: str, uid: str) -> str:
|
|
short_uid = uid[:8]
|
|
slug_part = slugify(text)
|
|
if not slug_part:
|
|
return short_uid
|
|
return f"{short_uid}-{slug_part}"
|
|
|
|
|
|
def time_ago(dt_str: str) -> str:
|
|
dt = datetime.fromisoformat(dt_str)
|
|
now = datetime.utcnow()
|
|
diff = now - dt
|
|
days = diff.days
|
|
if days > 30:
|
|
return dt.strftime("%d/%m/%Y")
|
|
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"
|
|
|
|
|
|
def extract_mentions(content: str) -> list[str]:
|
|
if not content:
|
|
return []
|
|
return re.findall(r"(?:^|[\s(])@([a-zA-Z0-9_-]+)", content)
|
|
|
|
|
|
def create_mention_notifications(content: str, actor_uid: str, target_url: str) -> None:
|
|
usernames = extract_mentions(content)
|
|
if not usernames:
|
|
return
|
|
users = get_table("users")
|
|
notifs = get_table("notifications")
|
|
seen = set()
|
|
for username in usernames:
|
|
if username in seen:
|
|
continue
|
|
seen.add(username)
|
|
mentioned = users.find_one(username=username)
|
|
if mentioned and mentioned["uid"] != actor_uid:
|
|
notifs.insert({
|
|
"uid": generate_uid(),
|
|
"user_uid": mentioned["uid"],
|
|
"type": "mention",
|
|
"message": f"@{username} mentioned you",
|
|
"related_uid": actor_uid,
|
|
"target_url": target_url,
|
|
"read": False,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
})
|
|
|
|
|
|
def format_date(dt_str: str, include_time: bool = False) -> str:
|
|
if not dt_str:
|
|
return ""
|
|
try:
|
|
dt = datetime.fromisoformat(dt_str)
|
|
if include_time:
|
|
return dt.strftime("%d/%m/%Y %H:%M")
|
|
return dt.strftime("%d/%m/%Y")
|
|
except (ValueError, TypeError):
|
|
return dt_str
|