devplacepy/devplacepy/utils.py

import asyncio
import html
import re
import secrets
import logging
from datetime import datetime, timedelta, timezone
from passlib.hash import pbkdf2_sha256
from fastapi import Request, HTTPException, status
from devplacepy.cache import TTLCache
from devplacepy.database import get_table
from devplacepy.config import 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.now(timezone.utc) + timedelta(seconds=SESSION_MAX_AGE)
    sessions = get_table("sessions")
    sessions.insert({
        "session_token": token,
        "user_uid": user_uid,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "expires_at": expires_at.isoformat(),
    })
    return token


_user_cache = TTLCache(ttl=300)


def clear_user_cache(user_uid: str) -> None:
    for token, user in _user_cache.items():
        if user.get("uid") == user_uid:
            _user_cache.pop(token)


def get_current_user(request: Request):
    token = request.cookies.get("session")
    if not token:
        return None

    cached = _user_cache.get(token)
    if cached is not None:
        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.tzinfo is None:
        expires = expires.replace(tzinfo=timezone.utc)
    if expires < datetime.now(timezone.utc):
        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)
        return None
    if user:
        _user_cache.set(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 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 require_user_api(request: Request):
    user = get_current_user(request)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required")
    return user


def strip_html(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"<[^>]+>", " ", text)
    text = html.unescape(text)
    return re.sub(r"\s+", " ", text).strip()


def slugify(text: str) -> str:
    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)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    now = datetime.now(timezone.utc)
    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)


PUSH_ICON = "/static/apple-touch-icon.png"
DEFAULT_PUSH_URL = "/notifications"

_push_tasks: set[asyncio.Task] = set()


async def _safe_notify(user_uid: str, payload: dict[str, str]) -> None:
    from devplacepy import push
    try:
        await push.notify_user(user_uid, payload)
    except Exception as e:
        logger.warning("Push delivery failed for %s: %s", user_uid, e)


def _schedule_push(user_uid: str, message: str, target_url: str | None) -> None:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    payload = {
        "title": "DevPlace",
        "message": message,
        "icon": PUSH_ICON,
        "url": target_url or DEFAULT_PUSH_URL,
    }
    task = loop.create_task(_safe_notify(user_uid, payload))
    _push_tasks.add(task)
    task.add_done_callback(_push_tasks.discard)


def create_notification(user_uid: str, notification_type: str, message: str, related_uid: str, target_url: str | None = None) -> None:
    from devplacepy.templating import clear_unread_cache
    get_table("notifications").insert({
        "uid": generate_uid(),
        "user_uid": user_uid,
        "type": notification_type,
        "message": message,
        "related_uid": related_uid,
        "target_url": target_url,
        "read": False,
        "created_at": datetime.now(timezone.utc).isoformat(),
    })
    clear_unread_cache(user_uid)
    _schedule_push(user_uid, message, target_url)


def award_badge(user_uid: str, badge_name: str) -> bool:
    badges = get_table("badges")
    if badges.find_one(user_uid=user_uid, badge_name=badge_name):
        return False
    badges.insert({
        "uid": generate_uid(),
        "user_uid": user_uid,
        "badge_name": badge_name,
        "created_at": datetime.now(timezone.utc).isoformat(),
    })
    return True


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")
    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:
            create_notification(mentioned["uid"], "mention", f"@{username} mentioned you", actor_uid, target_url)


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