From 8daee65011c08baf86743f4462619cf7af417859 Mon Sep 17 00:00:00 2001 From: retoor Date: Sat, 4 Jul 2026 19:18:19 +0000 Subject: [PATCH] refactor: split utils.py into package (ref.md 3.8) --- devplacepy/utils.py | 1024 ----------------------------- devplacepy/utils/__init__.py | 133 ++++ devplacepy/utils/accounts.py | 55 ++ devplacepy/utils/auth.py | 157 +++++ devplacepy/utils/authcache.py | 22 + devplacepy/utils/badges.py | 166 +++++ devplacepy/utils/guards.py | 133 ++++ devplacepy/utils/notifications.py | 155 +++++ devplacepy/utils/passwords.py | 41 ++ devplacepy/utils/rewards.py | 215 ++++++ devplacepy/utils/text.py | 102 +++ 11 files changed, 1179 insertions(+), 1024 deletions(-) delete mode 100644 devplacepy/utils.py create mode 100644 devplacepy/utils/__init__.py create mode 100644 devplacepy/utils/accounts.py create mode 100644 devplacepy/utils/auth.py create mode 100644 devplacepy/utils/authcache.py create mode 100644 devplacepy/utils/badges.py create mode 100644 devplacepy/utils/guards.py create mode 100644 devplacepy/utils/notifications.py create mode 100644 devplacepy/utils/passwords.py create mode 100644 devplacepy/utils/rewards.py create mode 100644 devplacepy/utils/text.py diff --git a/devplacepy/utils.py b/devplacepy/utils.py deleted file mode 100644 index 0d7e3a3e..00000000 --- a/devplacepy/utils.py +++ /dev/null @@ -1,1024 +0,0 @@ -# retoor - -import asyncio -import base64 -import binascii -import hashlib -import html -import json -import re -import secrets -import logging -from datetime import datetime, timedelta, timezone -from decimal import Decimal -from typing import Optional -from urllib.parse import urlsplit -from passlib.hash import pbkdf2_sha256 -from fastapi import Request, HTTPException, status -from devplacepy.cache import TTLCache -from devplacepy.database import ( - get_table, - get_user_stars, - bump_cache_version, - sync_local_cache, - invalidate_admins_cache, - notification_enabled, - get_silenced_uids, - record_activity, - record_unique_activity, -) -from devplacepy.config import ( - SESSION_MAX_AGE, - DEFAULT_CORRECTION_PROMPT, - DEFAULT_MODIFIER_PROMPT, -) -from devplacepy.services.background import background - -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) - - -async def hash_password_async(password: str) -> str: - return await asyncio.to_thread(pbkdf2_sha256.hash, password) - - -async def verify_password_async(password: str, hashed: str) -> bool: - return await asyncio.to_thread(pbkdf2_sha256.verify, password, hashed) - - -def create_session(user_uid: str, max_age_seconds: int = SESSION_MAX_AGE) -> str: - token = secrets.token_hex(32) - expires_at = datetime.now(timezone.utc) + timedelta(seconds=max_age_seconds) - 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(), - "deleted_at": None, - "deleted_by": None, - } - ) - return token - - -_user_cache = TTLCache(ttl=300, max_size=2000) - - -def _sync_auth_cache() -> None: - sync_local_cache("auth", _user_cache) - - -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) - bump_cache_version("auth") - - -def clear_session_cache(token: str) -> None: - _user_cache.pop(token) - bump_cache_version("auth") - - -_UNSET = object() - - -def _user_from_session(request: Request): - token = request.cookies.get("session") - if not token: - return None - _sync_auth_cache() - cached = _user_cache.get(token) - if cached is not None: - return cached - sessions = get_table("sessions") - session = sessions.find_one(session_token=token, deleted_at=None) - 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 _user_from_api_key(key: str): - if not key: - return None - _sync_auth_cache() - cache_key = f"k:{key}" - cached = _user_cache.get(cache_key) - if cached is not None: - return cached - user = get_table("users").find_one(api_key=key) - if not user or not user.get("is_active", True): - return None - _user_cache.set(cache_key, user) - return user - - -def _user_from_basic(header: str): - try: - raw = base64.b64decode(header.strip(), validate=True).decode("utf-8") - except (ValueError, binascii.Error, UnicodeDecodeError): - return None - if ":" not in raw: - return None - identifier, password = raw.split(":", 1) - identifier = identifier.strip() - if not identifier or not password: - return None - _sync_auth_cache() - cache_key = "b:" + hashlib.sha256(header.strip().encode()).hexdigest() - cached = _user_cache.get(cache_key) - if cached is not None: - return cached - users = get_table("users") - user = users.find_one(username=identifier) or users.find_one( - email=identifier.lower() - ) - if not user or not user.get("is_active", True): - return None - if not verify_password(password, user.get("password_hash", "")): - return None - _user_cache.set(cache_key, user) - return user - - -def _request_has_auth(request: Request) -> bool: - return bool( - request.headers.get("X-API-KEY") or request.headers.get("Authorization") - ) - - -def _resolve_user(request: Request): - user = _user_from_session(request) - if user: - return user - - # Try header-based auth: X-API-KEY → Bearer → Basic - api_key = request.headers.get("X-API-KEY") - if api_key: - api_key = api_key.strip() - user = _user_from_api_key(api_key) - if user: - return user - user = _user_from_devrant_token(api_key) - if user: - return user - user = _user_from_access_token(api_key) - if user: - return user - - authorization = request.headers.get("Authorization", "") - scheme, _, credentials = authorization.partition(" ") - scheme = scheme.lower() - credentials = credentials.strip() - if scheme == "bearer" and credentials: - user = _user_from_api_key(credentials) - if user: - return user - user = _user_from_devrant_token(credentials) - if user: - return user - user = _user_from_access_token(credentials) - if user: - return user - if scheme == "basic" and credentials: - user = _user_from_basic(credentials) - if user: - return user - return None - - -def _user_from_devrant_token(key: str): - """Resolve a user from a DevRant auth token key (40-char hex). - - Bridges the devrant_tokens table into the main auth chain so DevRant - tokens work as Bearer / X-API-KEY credentials on every DevPlace endpoint. - """ - # DevRant token keys are exactly 40 hex characters; skip other formats - # to avoid unnecessary DB queries on UUID-style API keys. - if not key or len(key) != 40 or not all(c in "0123456789abcdef" for c in key.lower()): - return None - from devplacepy.services.devrant.tokens import resolve_user_by_key - - return resolve_user_by_key(key) - - -def _user_from_access_token(key: str): - """Resolve a user from a DevPlace access token (64-char hex). - - Native DevPlace token auth - tokens issued by ``POST /auth/token``. - """ - if not key or len(key) != 64: - return None - from devplacepy.services.access_tokens import resolve_token - - return resolve_token(key) - - -def get_current_user(request: Request): - cached = getattr(request.state, "_auth_user", _UNSET) - if cached is not _UNSET: - return cached - user = _resolve_user(request) - try: - request.state._auth_user = user - except Exception: - pass - return user - - -def safe_next(value, default="/feed"): - if ( - value - and value.startswith("/") - and not value.startswith(("//", "/\\")) - and "\\" not in value - and "\n" not in value - and "\r" not in value - ): - return value - return default - - -def redirect_back(request: Request, default: str = "/feed") -> str: - referer = request.headers.get("referer") or "" - parts = urlsplit(referer) - if parts.scheme and parts.netloc: - if parts.netloc == urlsplit(str(request.base_url)).netloc: - referer = parts.path + (f"?{parts.query}" if parts.query else "") - else: - referer = "" - return safe_next(referer, default) - - -def cookie_secure(request: Request) -> bool: - forwarded = request.headers.get("x-forwarded-proto", "") - if forwarded: - return forwarded.split(",")[0].strip().lower() == "https" - return request.url.scheme == "https" - - -def client_ip(request, default: Optional[str] = "unknown") -> Optional[str]: - headers = getattr(request, "headers", None) - if headers is not None: - real = headers.get("x-real-ip") - if real and real.strip(): - return real.strip() - forwarded = headers.get("x-forwarded-for") - if forwarded: - first = forwarded.split(",")[0].strip() - if first: - return first - client = getattr(request, "client", None) - if client is not None: - return client.host - return default - - -def require_user(request: Request): - user = get_current_user(request) - if not user: - from devplacepy.responses import wants_json - from devplacepy.services.audit import record as audit - - audit.record( - request, - "security.authz.denied", - user=None, - result="denied", - summary=f"guest denied access to {request.url.path}", - ) - if _request_has_auth(request) or wants_json(request): - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials" - ) - raise HTTPException( - status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/auth/login"} - ) - return user - - -def is_admin(user: dict | None) -> bool: - return bool(user) and user.get("role") == "Admin" - - -def is_primary_admin(user: dict | None) -> bool: - from devplacepy.database import get_primary_admin_uid - - return is_admin(user) and user.get("uid") == get_primary_admin_uid() - - -def require_admin(request: Request): - user = require_user(request) - if not is_admin(user): - from devplacepy.responses import wants_json - from devplacepy.services.audit import record as audit - - audit.record( - request, - "security.authz.denied", - user=user, - result="denied", - summary=f"non-admin denied access to {request.url.path}", - ) - if wants_json(request): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Administrator access required", - ) - raise HTTPException( - status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/feed"} - ) - return user - - -def not_found(detail: str = "Not found") -> HTTPException: - return HTTPException(status_code=404, detail=detail) - - -def require_user_api(request: Request): - user = get_current_user(request) - if not user: - from devplacepy.services.audit import record as audit - - audit.record( - request, - "security.authz.denied", - user=None, - result="denied", - summary=f"guest denied access to {request.url.path}", - ) - 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() - - -_FLOAT_TOKEN = "@dpfloat@" - - -def plain_float(value: float) -> str: - if value != value or value in (float("inf"), float("-inf")): - return json.dumps(value) - return format(Decimal(repr(value)), "f") - - -def _mark_floats(node): - if isinstance(node, bool): - return node - if isinstance(node, float): - return f"{_FLOAT_TOKEN}{plain_float(node)}{_FLOAT_TOKEN}" - if isinstance(node, dict): - return {key: _mark_floats(value) for key, value in node.items()} - if isinstance(node, (list, tuple)): - return [_mark_floats(item) for item in node] - return node - - -def pretty_json(data, indent: int = 2) -> str: - text = json.dumps( - _mark_floats(data), indent=indent, ensure_ascii=False, default=str - ) - return text.replace(f'"{_FLOAT_TOKEN}', "").replace(f'{_FLOAT_TOKEN}"', "") - - -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_utils - - return str(uuid_utils.uuid7()) - - -def make_combined_slug(text: str, uid: str) -> str: - short_uid = uid.replace("-", "")[-12:] - 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: - payload = { - "title": "DevPlace", - "message": message, - "icon": PUSH_ICON, - "url": target_url or DEFAULT_PUSH_URL, - } - try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = background.loop - if loop is None: - logger.warning("no event loop available for push delivery to %s", user_uid) - return - asyncio.run_coroutine_threadsafe(_safe_notify(user_uid, payload), loop) - return - task = loop.create_task(_safe_notify(user_uid, payload)) - _push_tasks.add(task) - task.add_done_callback(_push_tasks.discard) - - -def _schedule_telegram(user_uid: str, message: str) -> None: - from devplacepy.services.telegram import store as telegram_store - - try: - telegram_store.enqueue_outbox(user_uid, message) - except Exception as e: - logger.warning("Telegram delivery enqueue failed for %s: %s", user_uid, e) - - -def create_notification( - user_uid: str, - notification_type: str, - message: str, - related_uid: str, - target_url: str | None = None, -) -> None: - background.submit( - _deliver_notification, - user_uid, - notification_type, - message, - related_uid, - target_url, - ) - - -def _deliver_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 - - if related_uid and related_uid in get_silenced_uids(user_uid): - return - - in_app = notification_enabled(user_uid, notification_type, "in_app") - push = notification_enabled(user_uid, notification_type, "push") - telegram = notification_enabled(user_uid, notification_type, "telegram") - if in_app: - 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) - if push: - _schedule_push(user_uid, message, target_url) - if telegram: - _schedule_telegram(user_uid, message) - from devplacepy.services.audit import record as audit - - audit.record_system( - "notification.create", - actor_kind="system", - target_type="notification", - target_uid=related_uid, - summary=f"notification created: {message}", - metadata={ - "type": notification_type, - "in_app": in_app, - "push": push, - "telegram": telegram, - }, - links=[ - audit.recipient(user_uid), - audit.link("source", "notification", related_uid), - ], - ) - - -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 - badge_uid = generate_uid() - badges.insert( - { - "uid": badge_uid, - "user_uid": user_uid, - "badge_name": badge_name, - "created_at": datetime.now(timezone.utc).isoformat(), - } - ) - from devplacepy.services.audit import record as audit - - audit.record_system( - "reward.badge.award", - actor_kind="user", - actor_uid=user_uid, - summary=f"user earned the {badge_name} badge", - metadata={"badge": badge_name}, - links=[ - audit.actor(user_uid), - audit.link("target", "badge", badge_uid, badge_name), - ], - ) - return True - - -def _create_account(username: str, email: str, password_hash: str) -> tuple[str, str, bool]: - users = get_table("users") - uid = generate_uid() - is_first = users.count() == 0 - role = "Admin" if is_first else "Member" - users.insert( - { - "uid": uid, - "username": username, - "email": email.strip().lower(), - "api_key": generate_uid(), - "password_hash": password_hash, - "bio": "", - "location": "", - "git_link": "", - "website": "", - "role": role, - "is_active": True, - "level": 1, - "xp": 0, - "stars": 0, - "ai_correction_enabled": 0, - "ai_correction_prompt": DEFAULT_CORRECTION_PROMPT, - "ai_correction_sync": 0, - "ai_modifier_enabled": 1, - "ai_modifier_prompt": DEFAULT_MODIFIER_PROMPT, - "ai_modifier_sync": 1, - "avatar_seed": None, - "created_at": datetime.now(timezone.utc).isoformat(), - } - ) - award_badge(uid, "Member") - if is_first: - invalidate_admins_cache() - return uid, role, is_first - - -def register_account(username: str, email: str, password: str) -> tuple[str, str, bool]: - return _create_account(username, email, hash_password(password)) - - -async def register_account_async( - username: str, email: str, password: str -) -> tuple[str, str, bool]: - return _create_account(username, email, await hash_password_async(password)) - - -LEVEL_XP = 100 - -XP_POST = 10 -XP_COMMENT = 2 -XP_PROJECT = 15 -XP_GIST = 5 -XP_UPVOTE = 5 -XP_FOLLOW = 5 - -LEVEL_BADGES = { - 5: "Level 5", - 10: "Level 10", - 25: "Level 25", - 50: "Level 50", - 100: "Level 100", -} - -BADGE_CATALOG = { - "Member": {"icon": "✦", "description": "Joined DevPlace", "group": "Milestones"}, - "First Post": {"icon": "✎", "description": "Published a first post", "group": "First steps"}, - "First Comment": {"icon": "❝", "description": "Wrote a first comment", "group": "First steps"}, - "First Project": {"icon": "⬢", "description": "Shared a first project", "group": "First steps"}, - "First Gist": {"icon": "❡", "description": "Shared a first gist", "group": "First steps"}, - "Prolific": {"icon": "✺", "description": "Published 10 posts", "group": "Content"}, - "Wordsmith": {"icon": "🖋", "description": "Published 50 posts", "group": "Content"}, - "Veteran": {"icon": "🏆", "description": "Published 100 posts", "group": "Content"}, - "Conversationalist": {"icon": "💬", "description": "Wrote 25 comments", "group": "Content"}, - "Debater": {"icon": "📣", "description": "Wrote 100 comments", "group": "Content"}, - "Builder": {"icon": "🔧", "description": "Shared 5 projects", "group": "Content"}, - "Architect": {"icon": "🏛", "description": "Shared 10 projects", "group": "Content"}, - "Snippet Collector": {"icon": "📑", "description": "Shared 5 gists", "group": "Content"}, - "Rising Star": {"icon": "☆", "description": "Earned 25 stars", "group": "Reputation"}, - "Star Author": {"icon": "★", "description": "Earned 100 stars", "group": "Reputation"}, - "Superstar": {"icon": "🌟", "description": "Earned 500 stars", "group": "Reputation"}, - "Popular": {"icon": "◎", "description": "Reached 10 followers", "group": "Reputation"}, - "Influencer": {"icon": "📢", "description": "Reached 50 followers", "group": "Reputation"}, - "Celebrity": {"icon": "👑", "description": "Reached 100 followers", "group": "Reputation"}, - "Connector": {"icon": "🔗", "description": "Followed 10 people", "group": "Community"}, - "On Fire": {"icon": "🔥", "description": "Maintained a 7-day activity streak", "group": "Dedication"}, - "Dedicated": {"icon": "📅", "description": "Maintained a 30-day activity streak", "group": "Dedication"}, - "Unstoppable": {"icon": "🚀", "description": "Maintained a 100-day activity streak", "group": "Dedication"}, - "Level 5": {"icon": "❖", "description": "Reached level 5", "group": "Levels"}, - "Level 10": {"icon": "❖", "description": "Reached level 10", "group": "Levels"}, - "Level 25": {"icon": "❖", "description": "Reached level 25", "group": "Levels"}, - "Level 50": {"icon": "❖", "description": "Reached level 50", "group": "Levels"}, - "Level 100": {"icon": "❖", "description": "Reached level 100", "group": "Levels"}, - "Curious": {"icon": "📖", "description": "Read your first documentation page", "group": "Explorer"}, - "Studious": {"icon": "📚", "description": "Read 5 documentation pages", "group": "Explorer"}, - "Scholar": {"icon": "🎓", "description": "Read 15 documentation pages", "group": "Explorer"}, - "AI Curious": {"icon": "🤖", "description": "Talked to Devii for the first time", "group": "Explorer"}, - "AI Whisperer": {"icon": "✨", "description": "Had 25 conversations with Devii", "group": "Explorer"}, - "First Fork": {"icon": "🍴", "description": "Forked a project for the first time", "group": "Explorer"}, - "Forker": {"icon": "🌿", "description": "Forked 5 projects", "group": "Explorer"}, - "Archivist": {"icon": "🗜", "description": "Downloaded a project archive", "group": "Explorer"}, - "SEO Auditor": {"icon": "🔍", "description": "Ran an SEO diagnostics audit", "group": "Explorer"}, - "Researcher": {"icon": "🔬", "description": "Ran a DeepSearch investigation", "group": "Explorer"}, - "Deep Diver": {"icon": "🌊", "description": "Ran 10 DeepSearch investigations", "group": "Explorer"}, - "Container Captain": {"icon": "📦", "description": "Created a container instance", "group": "Explorer"}, - "Messenger": {"icon": "✉", "description": "Sent your first direct message", "group": "Community"}, - "Chatterbox": {"icon": "📨", "description": "Sent 100 direct messages", "group": "Community"}, - "Bookmarker": {"icon": "🔖", "description": "Bookmarked your first item", "group": "Engagement"}, - "Curator": {"icon": "📌", "description": "Bookmarked 25 items", "group": "Engagement"}, - "First Reaction": {"icon": "💛", "description": "Reacted to content for the first time", "group": "Engagement"}, - "Cheerleader": {"icon": "🎉", "description": "Reacted 50 times", "group": "Engagement"}, - "First Star Given": {"icon": "⭐", "description": "Starred someone's work for the first time", "group": "Engagement"}, - "Supporter": {"icon": "👍", "description": "Gave 50 stars", "group": "Engagement"}, - "Patron": {"icon": "🤝", "description": "Gave 250 stars", "group": "Engagement"}, - "Friendly": {"icon": "👋", "description": "Followed someone for the first time", "group": "Community"}, - "First Upload": {"icon": "📎", "description": "Uploaded your first attachment", "group": "First steps"}, - "File Creator": {"icon": "📄", "description": "Created a file in a project", "group": "First steps"}, - "Bug Reporter": {"icon": "🐛", "description": "Filed your first issue", "group": "Community"}, - "Pollster": {"icon": "🗳", "description": "Voted in a poll for the first time", "group": "Engagement"}, - "Profiled": {"icon": "📇", "description": "Customized your profile", "group": "First steps"}, - "Green Thumb": {"icon": "🌱", "description": "Harvested your first build on the Code Farm", "group": "Code Farm"}, - "Master Farmer": {"icon": "🌾", "description": "Harvested 50 builds on the Code Farm", "group": "Code Farm"}, - "Good Neighbor": {"icon": "💧", "description": "Watered a neighbour's build", "group": "Code Farm"}, - "Cat Burglar": {"icon": "🥷", "description": "Stole a ready build from another farm", "group": "Code Farm"}, - "Robbed": {"icon": "🚨", "description": "Had a build stolen from your farm", "group": "Code Farm"}, -} - -BADGE_GROUPS = [ - "First steps", - "Explorer", - "Engagement", - "Content", - "Community", - "Reputation", - "Dedication", - "Code Farm", - "Levels", - "Milestones", -] - - -def get_badge(badge_name: str) -> dict: - return BADGE_CATALOG.get( - badge_name, {"icon": "✦", "description": badge_name, "group": "Milestones"} - ) - - -def build_achievements(held: set) -> list: - grouped: dict[str, list] = {group: [] for group in BADGE_GROUPS} - for name, meta in BADGE_CATALOG.items(): - group = meta.get("group", "Milestones") - grouped.setdefault(group, []).append( - { - "name": name, - "icon": meta["icon"], - "description": meta["description"], - "earned": name in held, - } - ) - showcase = [] - for group in BADGE_GROUPS: - items = grouped.get(group) or [] - if not items: - continue - showcase.append( - { - "group": group, - "items": items, - "earned": sum(1 for item in items if item["earned"]), - "total": len(items), - } - ) - return showcase - - -def level_for_xp(xp: int) -> int: - return 1 + max(0, xp) // LEVEL_XP - - -def notify_badge(user_uid: str, badge_name: str) -> None: - user = get_table("users").find_one(uid=user_uid) - if not user: - return - create_notification( - user_uid, - "badge", - f"You earned the {badge_name} badge", - user_uid, - f"/profile/{user['username']}", - ) - - -def award_xp(user_uid: str, amount: int) -> dict: - users = get_table("users") - user = users.find_one(uid=user_uid) - if not user or amount <= 0: - return { - "xp": user.get("xp", 0) if user else 0, - "level": user.get("level", 1) if user else 1, - "leveled_up": False, - } - current_xp = user.get("xp", 0) or 0 - current_level = user.get("level", 1) or 1 - new_xp = max(0, current_xp + amount) - new_level = level_for_xp(new_xp) - users.update({"uid": user_uid, "xp": new_xp, "level": new_level}, ["uid"]) - clear_user_cache(user_uid) - leveled_up = new_level > current_level - from devplacepy.services.audit import record as audit - - audit.record_system( - "reward.xp.grant", - actor_kind="user", - actor_uid=user_uid, - new_value=amount, - metadata={"amount": amount, "total_xp": new_xp}, - summary=f"user earned {amount} XP", - links=[audit.actor(user_uid)], - ) - if leveled_up: - audit.record_system( - "reward.level.up", - actor_kind="user", - actor_uid=user_uid, - old_value=current_level, - new_value=new_level, - metadata={"level": new_level}, - summary=f"user reached level {new_level}", - links=[audit.actor(user_uid)], - ) - create_notification( - user_uid, - "level", - f"You reached level {new_level}", - user_uid, - f"/profile/{user['username']}", - ) - for level in range(current_level + 1, new_level + 1): - badge_name = LEVEL_BADGES.get(level) - if badge_name and award_badge(user_uid, badge_name): - notify_badge(user_uid, badge_name) - return {"xp": new_xp, "level": new_level, "leveled_up": leveled_up} - - -_COUNT_MILESTONES = [ - ("Prolific", "posts", 10), - ("Wordsmith", "posts", 50), - ("Veteran", "posts", 100), - ("Conversationalist", "comments", 25), - ("Debater", "comments", 100), - ("Builder", "projects", 5), - ("Architect", "projects", 10), - ("Snippet Collector", "gists", 5), - ("Rising Star", "stars", 25), - ("Star Author", "stars", 100), - ("Superstar", "stars", 500), - ("Popular", "followers", 10), - ("Influencer", "followers", 50), - ("Celebrity", "followers", 100), - ("Connector", "following", 10), - ("On Fire", "streak", 7), - ("Dedicated", "streak", 30), - ("Unstoppable", "streak", 100), -] - - -def _milestone_metrics(user_uid: str): - cache: dict[str, int] = {} - - def resolve(key: str) -> int: - if key in cache: - return cache[key] - if key == "posts": - value = get_table("posts").count(user_uid=user_uid, deleted_at=None) - elif key == "comments": - value = get_table("comments").count(user_uid=user_uid, deleted_at=None) - elif key == "projects": - value = get_table("projects").count(user_uid=user_uid, deleted_at=None) - elif key == "gists": - value = get_table("gists").count(user_uid=user_uid, deleted_at=None) - elif key == "stars": - value = get_user_stars(user_uid) - elif key == "followers": - value = get_table("follows").count(following_uid=user_uid, deleted_at=None) - elif key == "following": - value = get_table("follows").count(follower_uid=user_uid, deleted_at=None) - elif key == "streak": - from devplacepy.database import get_streaks - - value = get_streaks(user_uid)["current"] - else: - value = 0 - cache[key] = value - return value - - return resolve - - -def check_milestone_badges(user_uid: str) -> list: - held = {row["badge_name"] for row in get_table("badges").find(user_uid=user_uid)} - pending = [m for m in _COUNT_MILESTONES if m[0] not in held] - if not pending: - return [] - metric = _milestone_metrics(user_uid) - awarded = [] - for badge_name, metric_key, threshold in pending: - if metric(metric_key) >= threshold and award_badge(user_uid, badge_name): - awarded.append(badge_name) - if metric_key == "streak": - from devplacepy.services.audit import record as audit - - audit.record_system( - "reward.streak.milestone", - actor_kind="user", - actor_uid=user_uid, - metadata={"streak": metric("streak")}, - summary=f"user reached a {metric('streak')}-day streak", - links=[audit.actor(user_uid)], - ) - for badge_name in awarded: - notify_badge(user_uid, badge_name) - return awarded - - -def award_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None: - background.submit(_apply_rewards, user_uid, amount, first_badge) - - -def _apply_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None: - if first_badge: - award_badge(user_uid, first_badge) - award_xp(user_uid, amount) - check_milestone_badges(user_uid) - - -ACHIEVEMENTS = { - "docs.read": [(1, "Curious"), (5, "Studious"), (15, "Scholar")], - "devii": [(1, "AI Curious"), (25, "AI Whisperer")], - "fork": [(1, "First Fork"), (5, "Forker")], - "zip": [(1, "Archivist")], - "seo": [(1, "SEO Auditor")], - "deepsearch": [(1, "Researcher"), (10, "Deep Diver")], - "container": [(1, "Container Captain")], - "message": [(1, "Messenger"), (100, "Chatterbox")], - "bookmark": [(1, "Bookmarker"), (25, "Curator")], - "reaction": [(1, "First Reaction"), (50, "Cheerleader")], - "vote": [(1, "First Star Given"), (50, "Supporter"), (250, "Patron")], - "follow": [(1, "Friendly")], - "upload": [(1, "First Upload")], - "project_file": [(1, "File Creator")], - "issue": [(1, "Bug Reporter")], - "poll": [(1, "Pollster")], - "profile": [(1, "Profiled")], - "harvest": [(1, "Green Thumb"), (50, "Master Farmer")], - "water": [(1, "Good Neighbor")], - "harvest_stolen": [(1, "Cat Burglar")], - "got_stolen_from": [(1, "Robbed")], -} - -UNIQUE_ACTIONS = {"docs.read"} - - -def track_action(user_uid: str, action: str, target: str | None = None) -> None: - if not user_uid or action not in ACHIEVEMENTS: - return - background.submit(_apply_achievements, user_uid, action, target) - - -def _apply_achievements(user_uid: str, action: str, target: str | None) -> None: - if action in UNIQUE_ACTIONS: - if target is None: - return - count = record_unique_activity(user_uid, action, target) - else: - count = record_activity(user_uid, action) - if not count: - return - for threshold, badge_name in ACHIEVEMENTS.get(action, []): - if count >= threshold and award_badge(user_uid, badge_name): - notify_badge(user_uid, badge_name) - - -def create_mention_notifications(content: str, actor_uid: str, target_url: str) -> None: - background.submit(_deliver_mention_notifications, content, actor_uid, target_url) - - -def _deliver_mention_notifications( - content: str, actor_uid: str, target_url: str -) -> None: - usernames = list(dict.fromkeys(extract_mentions(content))) - if not usernames: - return - users = get_table("users") - actor = users.find_one(uid=actor_uid) - if not actor: - return - actor_username = actor["username"] - for mentioned in users.find(users.table.columns.username.in_(usernames)): - if mentioned["uid"] != actor_uid: - create_notification( - mentioned["uid"], - "mention", - f"@{actor_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 diff --git a/devplacepy/utils/__init__.py b/devplacepy/utils/__init__.py new file mode 100644 index 00000000..d59450e5 --- /dev/null +++ b/devplacepy/utils/__init__.py @@ -0,0 +1,133 @@ +# retoor + +import asyncio +import base64 +import binascii +import hashlib +import html +import json +import logging +import re +import secrets +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from typing import Optional +from urllib.parse import urlsplit +from passlib.hash import pbkdf2_sha256 +from fastapi import Request, HTTPException, status + +from devplacepy.cache import TTLCache +from devplacepy.database import ( + get_table, + get_user_stars, + bump_cache_version, + sync_local_cache, + invalidate_admins_cache, + notification_enabled, + get_silenced_uids, + record_activity, + record_unique_activity, +) +from devplacepy.config import ( + SESSION_MAX_AGE, + DEFAULT_CORRECTION_PROMPT, + DEFAULT_MODIFIER_PROMPT, +) +from devplacepy.services.background import background + +from devplacepy.utils.passwords import ( + hash_password, + verify_password, + hash_password_async, + verify_password_async, + create_session, +) +from devplacepy.utils.authcache import ( + _user_cache, + _sync_auth_cache, + clear_user_cache, + clear_session_cache, +) +from devplacepy.utils.auth import ( + _UNSET, + _user_from_session, + _user_from_api_key, + _user_from_basic, + _request_has_auth, + _resolve_user, + _user_from_devrant_token, + _user_from_access_token, + get_current_user, +) +from devplacepy.utils.guards import ( + safe_next, + redirect_back, + cookie_secure, + client_ip, + require_user, + is_admin, + is_primary_admin, + require_admin, + not_found, + require_user_api, +) +from devplacepy.utils.text import ( + strip_html, + _FLOAT_TOKEN, + plain_float, + _mark_floats, + pretty_json, + slugify, + generate_uid, + make_combined_slug, + time_ago, + extract_mentions, + format_date, +) +from devplacepy.utils.notifications import ( + logger, + PUSH_ICON, + DEFAULT_PUSH_URL, + _push_tasks, + _safe_notify, + _schedule_push, + _schedule_telegram, + create_notification, + _deliver_notification, + create_mention_notifications, + _deliver_mention_notifications, +) +from devplacepy.utils.badges import ( + LEVEL_BADGES, + BADGE_CATALOG, + BADGE_GROUPS, + get_badge, + build_achievements, + award_badge, + notify_badge, +) +from devplacepy.utils.rewards import ( + LEVEL_XP, + XP_POST, + XP_COMMENT, + XP_PROJECT, + XP_GIST, + XP_UPVOTE, + XP_FOLLOW, + level_for_xp, + award_xp, + _COUNT_MILESTONES, + _milestone_metrics, + check_milestone_badges, + award_rewards, + _apply_rewards, + ACHIEVEMENTS, + UNIQUE_ACTIONS, + track_action, + _apply_achievements, +) +from devplacepy.utils.accounts import ( + _create_account, + register_account, + register_account_async, +) diff --git a/devplacepy/utils/accounts.py b/devplacepy/utils/accounts.py new file mode 100644 index 00000000..93c956e8 --- /dev/null +++ b/devplacepy/utils/accounts.py @@ -0,0 +1,55 @@ +# retoor + +from datetime import datetime, timezone +from devplacepy.database import get_table, invalidate_admins_cache +from devplacepy.config import DEFAULT_CORRECTION_PROMPT, DEFAULT_MODIFIER_PROMPT +from devplacepy.utils.text import generate_uid +from devplacepy.utils.badges import award_badge +from devplacepy.utils.passwords import hash_password, hash_password_async + + +def _create_account(username: str, email: str, password_hash: str) -> tuple[str, str, bool]: + users = get_table("users") + uid = generate_uid() + is_first = users.count() == 0 + role = "Admin" if is_first else "Member" + users.insert( + { + "uid": uid, + "username": username, + "email": email.strip().lower(), + "api_key": generate_uid(), + "password_hash": password_hash, + "bio": "", + "location": "", + "git_link": "", + "website": "", + "role": role, + "is_active": True, + "level": 1, + "xp": 0, + "stars": 0, + "ai_correction_enabled": 0, + "ai_correction_prompt": DEFAULT_CORRECTION_PROMPT, + "ai_correction_sync": 0, + "ai_modifier_enabled": 1, + "ai_modifier_prompt": DEFAULT_MODIFIER_PROMPT, + "ai_modifier_sync": 1, + "avatar_seed": None, + "created_at": datetime.now(timezone.utc).isoformat(), + } + ) + award_badge(uid, "Member") + if is_first: + invalidate_admins_cache() + return uid, role, is_first + + +def register_account(username: str, email: str, password: str) -> tuple[str, str, bool]: + return _create_account(username, email, hash_password(password)) + + +async def register_account_async( + username: str, email: str, password: str +) -> tuple[str, str, bool]: + return _create_account(username, email, await hash_password_async(password)) diff --git a/devplacepy/utils/auth.py b/devplacepy/utils/auth.py new file mode 100644 index 00000000..0590cfb3 --- /dev/null +++ b/devplacepy/utils/auth.py @@ -0,0 +1,157 @@ +# retoor + +import base64 +import binascii +import hashlib +from datetime import datetime, timezone +from fastapi import Request +from devplacepy.database import get_table +from devplacepy.utils.authcache import _user_cache, _sync_auth_cache +from devplacepy.utils.passwords import verify_password + +_UNSET = object() + + +def _user_from_session(request: Request): + token = request.cookies.get("session") + if not token: + return None + _sync_auth_cache() + cached = _user_cache.get(token) + if cached is not None: + return cached + sessions = get_table("sessions") + session = sessions.find_one(session_token=token, deleted_at=None) + 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 _user_from_api_key(key: str): + if not key: + return None + _sync_auth_cache() + cache_key = f"k:{key}" + cached = _user_cache.get(cache_key) + if cached is not None: + return cached + user = get_table("users").find_one(api_key=key) + if not user or not user.get("is_active", True): + return None + _user_cache.set(cache_key, user) + return user + + +def _user_from_basic(header: str): + try: + raw = base64.b64decode(header.strip(), validate=True).decode("utf-8") + except (ValueError, binascii.Error, UnicodeDecodeError): + return None + if ":" not in raw: + return None + identifier, password = raw.split(":", 1) + identifier = identifier.strip() + if not identifier or not password: + return None + _sync_auth_cache() + cache_key = "b:" + hashlib.sha256(header.strip().encode()).hexdigest() + cached = _user_cache.get(cache_key) + if cached is not None: + return cached + users = get_table("users") + user = users.find_one(username=identifier) or users.find_one( + email=identifier.lower() + ) + if not user or not user.get("is_active", True): + return None + if not verify_password(password, user.get("password_hash", "")): + return None + _user_cache.set(cache_key, user) + return user + + +def _request_has_auth(request: Request) -> bool: + return bool( + request.headers.get("X-API-KEY") or request.headers.get("Authorization") + ) + + +def _resolve_user(request: Request): + user = _user_from_session(request) + if user: + return user + + api_key = request.headers.get("X-API-KEY") + if api_key: + api_key = api_key.strip() + user = _user_from_api_key(api_key) + if user: + return user + user = _user_from_devrant_token(api_key) + if user: + return user + user = _user_from_access_token(api_key) + if user: + return user + + authorization = request.headers.get("Authorization", "") + scheme, _, credentials = authorization.partition(" ") + scheme = scheme.lower() + credentials = credentials.strip() + if scheme == "bearer" and credentials: + user = _user_from_api_key(credentials) + if user: + return user + user = _user_from_devrant_token(credentials) + if user: + return user + user = _user_from_access_token(credentials) + if user: + return user + if scheme == "basic" and credentials: + user = _user_from_basic(credentials) + if user: + return user + return None + + +def _user_from_devrant_token(key: str): + if not key or len(key) != 40 or not all(c in "0123456789abcdef" for c in key.lower()): + return None + from devplacepy.services.devrant.tokens import resolve_user_by_key + + return resolve_user_by_key(key) + + +def _user_from_access_token(key: str): + if not key or len(key) != 64: + return None + from devplacepy.services.access_tokens import resolve_token + + return resolve_token(key) + + +def get_current_user(request: Request): + cached = getattr(request.state, "_auth_user", _UNSET) + if cached is not _UNSET: + return cached + user = _resolve_user(request) + try: + request.state._auth_user = user + except Exception: + pass + return user diff --git a/devplacepy/utils/authcache.py b/devplacepy/utils/authcache.py new file mode 100644 index 00000000..04b99b1d --- /dev/null +++ b/devplacepy/utils/authcache.py @@ -0,0 +1,22 @@ +# retoor + +from devplacepy.cache import TTLCache +from devplacepy.database import bump_cache_version, sync_local_cache + +_user_cache = TTLCache(ttl=300, max_size=2000) + + +def _sync_auth_cache() -> None: + sync_local_cache("auth", _user_cache) + + +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) + bump_cache_version("auth") + + +def clear_session_cache(token: str) -> None: + _user_cache.pop(token) + bump_cache_version("auth") diff --git a/devplacepy/utils/badges.py b/devplacepy/utils/badges.py new file mode 100644 index 00000000..eee956bd --- /dev/null +++ b/devplacepy/utils/badges.py @@ -0,0 +1,166 @@ +# retoor + +from datetime import datetime, timezone +from devplacepy.database import get_table +from devplacepy.utils.text import generate_uid +from devplacepy.utils.notifications import create_notification + +LEVEL_BADGES = { + 5: "Level 5", + 10: "Level 10", + 25: "Level 25", + 50: "Level 50", + 100: "Level 100", +} + +BADGE_CATALOG = { + "Member": {"icon": "✦", "description": "Joined DevPlace", "group": "Milestones"}, + "First Post": {"icon": "✎", "description": "Published a first post", "group": "First steps"}, + "First Comment": {"icon": "❝", "description": "Wrote a first comment", "group": "First steps"}, + "First Project": {"icon": "⬢", "description": "Shared a first project", "group": "First steps"}, + "First Gist": {"icon": "❡", "description": "Shared a first gist", "group": "First steps"}, + "Prolific": {"icon": "✺", "description": "Published 10 posts", "group": "Content"}, + "Wordsmith": {"icon": "🖋", "description": "Published 50 posts", "group": "Content"}, + "Veteran": {"icon": "🏆", "description": "Published 100 posts", "group": "Content"}, + "Conversationalist": {"icon": "💬", "description": "Wrote 25 comments", "group": "Content"}, + "Debater": {"icon": "📣", "description": "Wrote 100 comments", "group": "Content"}, + "Builder": {"icon": "🔧", "description": "Shared 5 projects", "group": "Content"}, + "Architect": {"icon": "🏛", "description": "Shared 10 projects", "group": "Content"}, + "Snippet Collector": {"icon": "📑", "description": "Shared 5 gists", "group": "Content"}, + "Rising Star": {"icon": "☆", "description": "Earned 25 stars", "group": "Reputation"}, + "Star Author": {"icon": "★", "description": "Earned 100 stars", "group": "Reputation"}, + "Superstar": {"icon": "🌟", "description": "Earned 500 stars", "group": "Reputation"}, + "Popular": {"icon": "◎", "description": "Reached 10 followers", "group": "Reputation"}, + "Influencer": {"icon": "📢", "description": "Reached 50 followers", "group": "Reputation"}, + "Celebrity": {"icon": "👑", "description": "Reached 100 followers", "group": "Reputation"}, + "Connector": {"icon": "🔗", "description": "Followed 10 people", "group": "Community"}, + "On Fire": {"icon": "🔥", "description": "Maintained a 7-day activity streak", "group": "Dedication"}, + "Dedicated": {"icon": "📅", "description": "Maintained a 30-day activity streak", "group": "Dedication"}, + "Unstoppable": {"icon": "🚀", "description": "Maintained a 100-day activity streak", "group": "Dedication"}, + "Level 5": {"icon": "❖", "description": "Reached level 5", "group": "Levels"}, + "Level 10": {"icon": "❖", "description": "Reached level 10", "group": "Levels"}, + "Level 25": {"icon": "❖", "description": "Reached level 25", "group": "Levels"}, + "Level 50": {"icon": "❖", "description": "Reached level 50", "group": "Levels"}, + "Level 100": {"icon": "❖", "description": "Reached level 100", "group": "Levels"}, + "Curious": {"icon": "📖", "description": "Read your first documentation page", "group": "Explorer"}, + "Studious": {"icon": "📚", "description": "Read 5 documentation pages", "group": "Explorer"}, + "Scholar": {"icon": "🎓", "description": "Read 15 documentation pages", "group": "Explorer"}, + "AI Curious": {"icon": "🤖", "description": "Talked to Devii for the first time", "group": "Explorer"}, + "AI Whisperer": {"icon": "✨", "description": "Had 25 conversations with Devii", "group": "Explorer"}, + "First Fork": {"icon": "🍴", "description": "Forked a project for the first time", "group": "Explorer"}, + "Forker": {"icon": "🌿", "description": "Forked 5 projects", "group": "Explorer"}, + "Archivist": {"icon": "🗜", "description": "Downloaded a project archive", "group": "Explorer"}, + "SEO Auditor": {"icon": "🔍", "description": "Ran an SEO diagnostics audit", "group": "Explorer"}, + "Researcher": {"icon": "🔬", "description": "Ran a DeepSearch investigation", "group": "Explorer"}, + "Deep Diver": {"icon": "🌊", "description": "Ran 10 DeepSearch investigations", "group": "Explorer"}, + "Container Captain": {"icon": "📦", "description": "Created a container instance", "group": "Explorer"}, + "Messenger": {"icon": "✉", "description": "Sent your first direct message", "group": "Community"}, + "Chatterbox": {"icon": "📨", "description": "Sent 100 direct messages", "group": "Community"}, + "Bookmarker": {"icon": "🔖", "description": "Bookmarked your first item", "group": "Engagement"}, + "Curator": {"icon": "📌", "description": "Bookmarked 25 items", "group": "Engagement"}, + "First Reaction": {"icon": "💛", "description": "Reacted to content for the first time", "group": "Engagement"}, + "Cheerleader": {"icon": "🎉", "description": "Reacted 50 times", "group": "Engagement"}, + "First Star Given": {"icon": "⭐", "description": "Starred someone's work for the first time", "group": "Engagement"}, + "Supporter": {"icon": "👍", "description": "Gave 50 stars", "group": "Engagement"}, + "Patron": {"icon": "🤝", "description": "Gave 250 stars", "group": "Engagement"}, + "Friendly": {"icon": "👋", "description": "Followed someone for the first time", "group": "Community"}, + "First Upload": {"icon": "📎", "description": "Uploaded your first attachment", "group": "First steps"}, + "File Creator": {"icon": "📄", "description": "Created a file in a project", "group": "First steps"}, + "Bug Reporter": {"icon": "🐛", "description": "Filed your first issue", "group": "Community"}, + "Pollster": {"icon": "🗳", "description": "Voted in a poll for the first time", "group": "Engagement"}, + "Profiled": {"icon": "📇", "description": "Customized your profile", "group": "First steps"}, + "Green Thumb": {"icon": "🌱", "description": "Harvested your first build on the Code Farm", "group": "Code Farm"}, + "Master Farmer": {"icon": "🌾", "description": "Harvested 50 builds on the Code Farm", "group": "Code Farm"}, + "Good Neighbor": {"icon": "💧", "description": "Watered a neighbour's build", "group": "Code Farm"}, + "Cat Burglar": {"icon": "🥷", "description": "Stole a ready build from another farm", "group": "Code Farm"}, + "Robbed": {"icon": "🚨", "description": "Had a build stolen from your farm", "group": "Code Farm"}, +} + +BADGE_GROUPS = [ + "First steps", + "Explorer", + "Engagement", + "Content", + "Community", + "Reputation", + "Dedication", + "Code Farm", + "Levels", + "Milestones", +] + + +def get_badge(badge_name: str) -> dict: + return BADGE_CATALOG.get( + badge_name, {"icon": "✦", "description": badge_name, "group": "Milestones"} + ) + + +def build_achievements(held: set) -> list: + grouped: dict[str, list] = {group: [] for group in BADGE_GROUPS} + for name, meta in BADGE_CATALOG.items(): + group = meta.get("group", "Milestones") + grouped.setdefault(group, []).append( + { + "name": name, + "icon": meta["icon"], + "description": meta["description"], + "earned": name in held, + } + ) + showcase = [] + for group in BADGE_GROUPS: + items = grouped.get(group) or [] + if not items: + continue + showcase.append( + { + "group": group, + "items": items, + "earned": sum(1 for item in items if item["earned"]), + "total": len(items), + } + ) + return showcase + + +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 + badge_uid = generate_uid() + badges.insert( + { + "uid": badge_uid, + "user_uid": user_uid, + "badge_name": badge_name, + "created_at": datetime.now(timezone.utc).isoformat(), + } + ) + from devplacepy.services.audit import record as audit + + audit.record_system( + "reward.badge.award", + actor_kind="user", + actor_uid=user_uid, + summary=f"user earned the {badge_name} badge", + metadata={"badge": badge_name}, + links=[ + audit.actor(user_uid), + audit.link("target", "badge", badge_uid, badge_name), + ], + ) + return True + + +def notify_badge(user_uid: str, badge_name: str) -> None: + user = get_table("users").find_one(uid=user_uid) + if not user: + return + create_notification( + user_uid, + "badge", + f"You earned the {badge_name} badge", + user_uid, + f"/profile/{user['username']}", + ) diff --git a/devplacepy/utils/guards.py b/devplacepy/utils/guards.py new file mode 100644 index 00000000..8b90accf --- /dev/null +++ b/devplacepy/utils/guards.py @@ -0,0 +1,133 @@ +# retoor + +from typing import Optional +from urllib.parse import urlsplit +from fastapi import Request, HTTPException, status +from devplacepy.utils.auth import get_current_user, _request_has_auth + + +def safe_next(value, default="/feed"): + if ( + value + and value.startswith("/") + and not value.startswith(("//", "/\\")) + and "\\" not in value + and "\n" not in value + and "\r" not in value + ): + return value + return default + + +def redirect_back(request: Request, default: str = "/feed") -> str: + referer = request.headers.get("referer") or "" + parts = urlsplit(referer) + if parts.scheme and parts.netloc: + if parts.netloc == urlsplit(str(request.base_url)).netloc: + referer = parts.path + (f"?{parts.query}" if parts.query else "") + else: + referer = "" + return safe_next(referer, default) + + +def cookie_secure(request: Request) -> bool: + forwarded = request.headers.get("x-forwarded-proto", "") + if forwarded: + return forwarded.split(",")[0].strip().lower() == "https" + return request.url.scheme == "https" + + +def client_ip(request, default: Optional[str] = "unknown") -> Optional[str]: + headers = getattr(request, "headers", None) + if headers is not None: + real = headers.get("x-real-ip") + if real and real.strip(): + return real.strip() + forwarded = headers.get("x-forwarded-for") + if forwarded: + first = forwarded.split(",")[0].strip() + if first: + return first + client = getattr(request, "client", None) + if client is not None: + return client.host + return default + + +def require_user(request: Request): + user = get_current_user(request) + if not user: + from devplacepy.responses import wants_json + from devplacepy.services.audit import record as audit + + audit.record( + request, + "security.authz.denied", + user=None, + result="denied", + summary=f"guest denied access to {request.url.path}", + ) + if _request_has_auth(request) or wants_json(request): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials" + ) + raise HTTPException( + status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/auth/login"} + ) + return user + + +def is_admin(user: dict | None) -> bool: + return bool(user) and user.get("role") == "Admin" + + +def is_primary_admin(user: dict | None) -> bool: + from devplacepy.database import get_primary_admin_uid + + return is_admin(user) and user.get("uid") == get_primary_admin_uid() + + +def require_admin(request: Request): + user = require_user(request) + if not is_admin(user): + from devplacepy.responses import wants_json + from devplacepy.services.audit import record as audit + + audit.record( + request, + "security.authz.denied", + user=user, + result="denied", + summary=f"non-admin denied access to {request.url.path}", + ) + if wants_json(request): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Administrator access required", + ) + raise HTTPException( + status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/feed"} + ) + return user + + +def not_found(detail: str = "Not found") -> HTTPException: + return HTTPException(status_code=404, detail=detail) + + +def require_user_api(request: Request): + user = get_current_user(request) + if not user: + from devplacepy.services.audit import record as audit + + audit.record( + request, + "security.authz.denied", + user=None, + result="denied", + summary=f"guest denied access to {request.url.path}", + ) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required" + ) + return user diff --git a/devplacepy/utils/notifications.py b/devplacepy/utils/notifications.py new file mode 100644 index 00000000..f417f64c --- /dev/null +++ b/devplacepy/utils/notifications.py @@ -0,0 +1,155 @@ +# retoor + +import asyncio +import logging +from datetime import datetime, timezone +from devplacepy.database import ( + get_table, + notification_enabled, + get_silenced_uids, +) +from devplacepy.services.background import background +from devplacepy.utils.text import generate_uid, extract_mentions + +logger = logging.getLogger(__name__) + +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: + payload = { + "title": "DevPlace", + "message": message, + "icon": PUSH_ICON, + "url": target_url or DEFAULT_PUSH_URL, + } + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = background.loop + if loop is None: + logger.warning("no event loop available for push delivery to %s", user_uid) + return + asyncio.run_coroutine_threadsafe(_safe_notify(user_uid, payload), loop) + return + task = loop.create_task(_safe_notify(user_uid, payload)) + _push_tasks.add(task) + task.add_done_callback(_push_tasks.discard) + + +def _schedule_telegram(user_uid: str, message: str) -> None: + from devplacepy.services.telegram import store as telegram_store + + try: + telegram_store.enqueue_outbox(user_uid, message) + except Exception as e: + logger.warning("Telegram delivery enqueue failed for %s: %s", user_uid, e) + + +def create_notification( + user_uid: str, + notification_type: str, + message: str, + related_uid: str, + target_url: str | None = None, +) -> None: + background.submit( + _deliver_notification, + user_uid, + notification_type, + message, + related_uid, + target_url, + ) + + +def _deliver_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 + + if related_uid and related_uid in get_silenced_uids(user_uid): + return + + in_app = notification_enabled(user_uid, notification_type, "in_app") + push = notification_enabled(user_uid, notification_type, "push") + telegram = notification_enabled(user_uid, notification_type, "telegram") + if in_app: + 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) + if push: + _schedule_push(user_uid, message, target_url) + if telegram: + _schedule_telegram(user_uid, message) + from devplacepy.services.audit import record as audit + + audit.record_system( + "notification.create", + actor_kind="system", + target_type="notification", + target_uid=related_uid, + summary=f"notification created: {message}", + metadata={ + "type": notification_type, + "in_app": in_app, + "push": push, + "telegram": telegram, + }, + links=[ + audit.recipient(user_uid), + audit.link("source", "notification", related_uid), + ], + ) + + +def create_mention_notifications(content: str, actor_uid: str, target_url: str) -> None: + background.submit(_deliver_mention_notifications, content, actor_uid, target_url) + + +def _deliver_mention_notifications( + content: str, actor_uid: str, target_url: str +) -> None: + usernames = list(dict.fromkeys(extract_mentions(content))) + if not usernames: + return + users = get_table("users") + actor = users.find_one(uid=actor_uid) + if not actor: + return + actor_username = actor["username"] + for mentioned in users.find(users.table.columns.username.in_(usernames)): + if mentioned["uid"] != actor_uid: + create_notification( + mentioned["uid"], + "mention", + f"@{actor_username} mentioned you", + actor_uid, + target_url, + ) diff --git a/devplacepy/utils/passwords.py b/devplacepy/utils/passwords.py new file mode 100644 index 00000000..fd55a232 --- /dev/null +++ b/devplacepy/utils/passwords.py @@ -0,0 +1,41 @@ +# retoor + +import asyncio +import secrets +from datetime import datetime, timedelta, timezone +from passlib.hash import pbkdf2_sha256 +from devplacepy.database import get_table +from devplacepy.config import SESSION_MAX_AGE + + +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) + + +async def hash_password_async(password: str) -> str: + return await asyncio.to_thread(pbkdf2_sha256.hash, password) + + +async def verify_password_async(password: str, hashed: str) -> bool: + return await asyncio.to_thread(pbkdf2_sha256.verify, password, hashed) + + +def create_session(user_uid: str, max_age_seconds: int = SESSION_MAX_AGE) -> str: + token = secrets.token_hex(32) + expires_at = datetime.now(timezone.utc) + timedelta(seconds=max_age_seconds) + 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(), + "deleted_at": None, + "deleted_by": None, + } + ) + return token diff --git a/devplacepy/utils/rewards.py b/devplacepy/utils/rewards.py new file mode 100644 index 00000000..003e7cb1 --- /dev/null +++ b/devplacepy/utils/rewards.py @@ -0,0 +1,215 @@ +# retoor + +from devplacepy.database import ( + get_table, + get_user_stars, + record_activity, + record_unique_activity, +) +from devplacepy.services.background import background +from devplacepy.utils.authcache import clear_user_cache +from devplacepy.utils.badges import award_badge, notify_badge, LEVEL_BADGES +from devplacepy.utils.notifications import create_notification + +LEVEL_XP = 100 + +XP_POST = 10 +XP_COMMENT = 2 +XP_PROJECT = 15 +XP_GIST = 5 +XP_UPVOTE = 5 +XP_FOLLOW = 5 + + +def level_for_xp(xp: int) -> int: + return 1 + max(0, xp) // LEVEL_XP + + +def award_xp(user_uid: str, amount: int) -> dict: + users = get_table("users") + user = users.find_one(uid=user_uid) + if not user or amount <= 0: + return { + "xp": user.get("xp", 0) if user else 0, + "level": user.get("level", 1) if user else 1, + "leveled_up": False, + } + current_xp = user.get("xp", 0) or 0 + current_level = user.get("level", 1) or 1 + new_xp = max(0, current_xp + amount) + new_level = level_for_xp(new_xp) + users.update({"uid": user_uid, "xp": new_xp, "level": new_level}, ["uid"]) + clear_user_cache(user_uid) + leveled_up = new_level > current_level + from devplacepy.services.audit import record as audit + + audit.record_system( + "reward.xp.grant", + actor_kind="user", + actor_uid=user_uid, + new_value=amount, + metadata={"amount": amount, "total_xp": new_xp}, + summary=f"user earned {amount} XP", + links=[audit.actor(user_uid)], + ) + if leveled_up: + audit.record_system( + "reward.level.up", + actor_kind="user", + actor_uid=user_uid, + old_value=current_level, + new_value=new_level, + metadata={"level": new_level}, + summary=f"user reached level {new_level}", + links=[audit.actor(user_uid)], + ) + create_notification( + user_uid, + "level", + f"You reached level {new_level}", + user_uid, + f"/profile/{user['username']}", + ) + for level in range(current_level + 1, new_level + 1): + badge_name = LEVEL_BADGES.get(level) + if badge_name and award_badge(user_uid, badge_name): + notify_badge(user_uid, badge_name) + return {"xp": new_xp, "level": new_level, "leveled_up": leveled_up} + + +_COUNT_MILESTONES = [ + ("Prolific", "posts", 10), + ("Wordsmith", "posts", 50), + ("Veteran", "posts", 100), + ("Conversationalist", "comments", 25), + ("Debater", "comments", 100), + ("Builder", "projects", 5), + ("Architect", "projects", 10), + ("Snippet Collector", "gists", 5), + ("Rising Star", "stars", 25), + ("Star Author", "stars", 100), + ("Superstar", "stars", 500), + ("Popular", "followers", 10), + ("Influencer", "followers", 50), + ("Celebrity", "followers", 100), + ("Connector", "following", 10), + ("On Fire", "streak", 7), + ("Dedicated", "streak", 30), + ("Unstoppable", "streak", 100), +] + + +def _milestone_metrics(user_uid: str): + cache: dict[str, int] = {} + + def resolve(key: str) -> int: + if key in cache: + return cache[key] + if key == "posts": + value = get_table("posts").count(user_uid=user_uid, deleted_at=None) + elif key == "comments": + value = get_table("comments").count(user_uid=user_uid, deleted_at=None) + elif key == "projects": + value = get_table("projects").count(user_uid=user_uid, deleted_at=None) + elif key == "gists": + value = get_table("gists").count(user_uid=user_uid, deleted_at=None) + elif key == "stars": + value = get_user_stars(user_uid) + elif key == "followers": + value = get_table("follows").count(following_uid=user_uid, deleted_at=None) + elif key == "following": + value = get_table("follows").count(follower_uid=user_uid, deleted_at=None) + elif key == "streak": + from devplacepy.database import get_streaks + + value = get_streaks(user_uid)["current"] + else: + value = 0 + cache[key] = value + return value + + return resolve + + +def check_milestone_badges(user_uid: str) -> list: + held = {row["badge_name"] for row in get_table("badges").find(user_uid=user_uid)} + pending = [m for m in _COUNT_MILESTONES if m[0] not in held] + if not pending: + return [] + metric = _milestone_metrics(user_uid) + awarded = [] + for badge_name, metric_key, threshold in pending: + if metric(metric_key) >= threshold and award_badge(user_uid, badge_name): + awarded.append(badge_name) + if metric_key == "streak": + from devplacepy.services.audit import record as audit + + audit.record_system( + "reward.streak.milestone", + actor_kind="user", + actor_uid=user_uid, + metadata={"streak": metric("streak")}, + summary=f"user reached a {metric('streak')}-day streak", + links=[audit.actor(user_uid)], + ) + for badge_name in awarded: + notify_badge(user_uid, badge_name) + return awarded + + +def award_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None: + background.submit(_apply_rewards, user_uid, amount, first_badge) + + +def _apply_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None: + if first_badge: + award_badge(user_uid, first_badge) + award_xp(user_uid, amount) + check_milestone_badges(user_uid) + + +ACHIEVEMENTS = { + "docs.read": [(1, "Curious"), (5, "Studious"), (15, "Scholar")], + "devii": [(1, "AI Curious"), (25, "AI Whisperer")], + "fork": [(1, "First Fork"), (5, "Forker")], + "zip": [(1, "Archivist")], + "seo": [(1, "SEO Auditor")], + "deepsearch": [(1, "Researcher"), (10, "Deep Diver")], + "container": [(1, "Container Captain")], + "message": [(1, "Messenger"), (100, "Chatterbox")], + "bookmark": [(1, "Bookmarker"), (25, "Curator")], + "reaction": [(1, "First Reaction"), (50, "Cheerleader")], + "vote": [(1, "First Star Given"), (50, "Supporter"), (250, "Patron")], + "follow": [(1, "Friendly")], + "upload": [(1, "First Upload")], + "project_file": [(1, "File Creator")], + "issue": [(1, "Bug Reporter")], + "poll": [(1, "Pollster")], + "profile": [(1, "Profiled")], + "harvest": [(1, "Green Thumb"), (50, "Master Farmer")], + "water": [(1, "Good Neighbor")], + "harvest_stolen": [(1, "Cat Burglar")], + "got_stolen_from": [(1, "Robbed")], +} + +UNIQUE_ACTIONS = {"docs.read"} + + +def track_action(user_uid: str, action: str, target: str | None = None) -> None: + if not user_uid or action not in ACHIEVEMENTS: + return + background.submit(_apply_achievements, user_uid, action, target) + + +def _apply_achievements(user_uid: str, action: str, target: str | None) -> None: + if action in UNIQUE_ACTIONS: + if target is None: + return + count = record_unique_activity(user_uid, action, target) + else: + count = record_activity(user_uid, action) + if not count: + return + for threshold, badge_name in ACHIEVEMENTS.get(action, []): + if count >= threshold and award_badge(user_uid, badge_name): + notify_badge(user_uid, badge_name) diff --git a/devplacepy/utils/text.py b/devplacepy/utils/text.py new file mode 100644 index 00000000..00915b41 --- /dev/null +++ b/devplacepy/utils/text.py @@ -0,0 +1,102 @@ +# retoor + +import html +import json +import re +from datetime import datetime, timezone +from decimal import Decimal + + +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() + + +_FLOAT_TOKEN = "@dpfloat@" + + +def plain_float(value: float) -> str: + if value != value or value in (float("inf"), float("-inf")): + return json.dumps(value) + return format(Decimal(repr(value)), "f") + + +def _mark_floats(node): + if isinstance(node, bool): + return node + if isinstance(node, float): + return f"{_FLOAT_TOKEN}{plain_float(node)}{_FLOAT_TOKEN}" + if isinstance(node, dict): + return {key: _mark_floats(value) for key, value in node.items()} + if isinstance(node, (list, tuple)): + return [_mark_floats(item) for item in node] + return node + + +def pretty_json(data, indent: int = 2) -> str: + text = json.dumps( + _mark_floats(data), indent=indent, ensure_ascii=False, default=str + ) + return text.replace(f'"{_FLOAT_TOKEN}', "").replace(f'{_FLOAT_TOKEN}"', "") + + +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_utils + + return str(uuid_utils.uuid7()) + + +def make_combined_slug(text: str, uid: str) -> str: + short_uid = uid.replace("-", "")[-12:] + 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) + + +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