# retoor import logging from typing import Annotated from fastapi import Depends, APIRouter, Request from devplacepy.models import ProfileForm from fastapi.responses import HTMLResponse, JSONResponse from devplacepy.database import ( get_table, db, get_customization_prefs, get_notification_prefs, get_user_stars, get_user_rank, get_user_post_count, get_comment_counts_by_post_uids, get_reactions_by_targets, get_user_bookmarks, get_polls_by_post_uids, get_activity_heatmap, get_activity_months, get_streaks, get_follow_counts, get_follow_list, get_following_among, get_user_relations, get_user_media, search_users_by_username, mark_notifications_read_by_target, resolve_object_url, ) from devplacepy.database.awards import ( can_give_award, get_prominent_award, get_user_awards, ) from devplacepy.content import can_view_project, enrich_items from devplacepy.utils import ( get_current_user, require_user, require_user_api, time_ago, clear_user_cache, not_found, generate_uid, track_action, build_achievements, ) from devplacepy.responses import respond, action_result, wants_json from devplacepy.schemas import ProfileOut from devplacepy.avatar import avatar_url, avatar_seed from devplacepy.seo import ( base_seo_context, site_url, website_schema, profile_page_schema, ) from devplacepy.services import presence from devplacepy.services.audit import record as audit from devplacepy.services.correction import schedule_correction from devplacepy.services.ai_modifier import schedule_modification from devplacepy.config import DEFAULT_CORRECTION_PROMPT, DEFAULT_MODIFIER_PROMPT from devplacepy.routers.profile.usage import _ai_quota, _correction_usage, _modifier_usage from devplacepy.dependencies import json_or_form logger = logging.getLogger(__name__) router = APIRouter() @router.get("/search") async def search_users(request: Request, q: str = ""): require_user_api(request) if not q or len(q) < 1: return JSONResponse({"results": []}) results = search_users_by_username(q) return JSONResponse({"results": results}) def _follow_people(profile_uid: str, mode: str, current_user, page: int): people, pagination = get_follow_list(profile_uid, mode, page) mine = ( get_following_among(current_user["uid"], [p["uid"] for p in people]) if current_user else set() ) for person in people: person["is_following"] = person["uid"] in mine person["is_self"] = bool(current_user and current_user["uid"] == person["uid"]) return people, pagination @router.get("/{username}/followers") async def profile_followers(request: Request, username: str, page: int = 1): return _follow_list_json(request, username, "followers", page) @router.get("/{username}/following") async def profile_following(request: Request, username: str, page: int = 1): return _follow_list_json(request, username, "following", page) def _follow_list_json(request: Request, username: str, mode: str, page: int): profile_user = get_table("users").find_one(username=username) if not profile_user: raise not_found("Profile not found") current_user = get_current_user(request) people, pagination = _follow_people(profile_user["uid"], mode, current_user, page) return JSONResponse( { "username": profile_user["username"], "mode": mode, "count": pagination["total"], "page": pagination["page"], "total_pages": pagination["total_pages"], mode: [ { "uid": p["uid"], "username": p["username"], "bio": p["bio"], "is_following": p["is_following"], } for p in people ], } ) @router.get("", response_class=HTMLResponse) async def my_profile_page(request: Request, tab: str = "posts", page: int = 1): user = require_user(request) return await profile_page(request, user["username"], tab, page) @router.get("/{username}", response_class=HTMLResponse) async def profile_page( request: Request, username: str, tab: str = "posts", page: int = 1 ): current_user = get_current_user(request) users = get_table("users") profile_user = users.find_one(username=username) if not profile_user: raise not_found("Profile not found") if current_user: mark_notifications_read_by_target( current_user["uid"], f"/profile/{profile_user['username']}" ) profile_user["stars"] = get_user_stars(profile_user["uid"]) rank = get_user_rank(profile_user["uid"]) follow_counts = get_follow_counts(profile_user["uid"]) people = [] follow_pagination = None if tab in ("followers", "following"): people, follow_pagination = _follow_people( profile_user["uid"], tab, current_user, page ) media, media_pagination = [], None if tab == "media": media, media_pagination = get_user_media(profile_user["uid"], page) awards, awards_pagination = [], None if tab == "awards": awards, awards_pagination = get_user_awards(profile_user["uid"], page) prominent_award = get_prominent_award(profile_user) awards_count = int(profile_user.get("award_count") or 0) can_give = bool( current_user and current_user["uid"] != profile_user["uid"] and can_give_award(current_user["uid"], profile_user["uid"]) ) posts = [] if tab == "posts": posts_table = get_table("posts") raw_posts = list( posts_table.find( user_uid=profile_user["uid"], deleted_at=None, order_by=["-created_at"], _limit=10, ) ) post_uids = [p["uid"] for p in raw_posts] counts = get_comment_counts_by_post_uids(post_uids) if raw_posts else {} authors = {profile_user["uid"]: profile_user} posts = enrich_items( raw_posts, "post", authors, {"comment_count": counts}, user=current_user ) reactions_map = get_reactions_by_targets("post", post_uids, current_user) bookmark_set = ( get_user_bookmarks(current_user["uid"], "post", post_uids) if current_user else set() ) polls_map = get_polls_by_post_uids(post_uids, current_user) for item in posts: uid = item["post"]["uid"] item["reactions"] = reactions_map.get(uid, {"counts": {}, "mine": []}) item["bookmarked"] = uid in bookmark_set item["poll"] = polls_map.get(uid) badges = list(get_table("badges").find(user_uid=profile_user["uid"])) achievements = build_achievements({b["badge_name"] for b in badges}) badge_total = sum(group["total"] for group in achievements) badge_earned = sum(group["earned"] for group in achievements) include_collections = wants_json(request) projects = [] if tab == "projects" or include_collections: projects = list( get_table("projects").find(user_uid=profile_user["uid"], deleted_at=None) ) projects = [p for p in projects if can_view_project(p, current_user)] gists = [] if tab == "gists" or include_collections: gists_raw = list( get_table("gists").find(user_uid=profile_user["uid"], deleted_at=None) ) for g in gists_raw: gists.append({"gist": g, "time_ago": time_ago(g["created_at"])}) posts_count = get_user_post_count(profile_user["uid"]) activities = [] if tab == "activity": posts_table = get_table("posts") for p in posts_table.find( user_uid=profile_user["uid"], deleted_at=None, order_by=["-created_at"], _limit=10, ): activities.append( { "type": "post", "content": p.get("title") or p["content"][:80], "time_ago": time_ago(p["created_at"]), "created_at": p["created_at"], "uid": p["uid"], "url": resolve_object_url("post", p["uid"]), } ) comments_table = get_table("comments") for c in comments_table.find( user_uid=profile_user["uid"], deleted_at=None, order_by=["-created_at"], _limit=10, ): target_uid = c.get("target_uid", c.get("post_uid", "")) target_type = c.get("target_type", "post") activities.append( { "type": "comment", "content": c["content"][:80], "time_ago": time_ago(c["created_at"]), "created_at": c["created_at"], "uid": target_uid, "target_type": target_type, "url": f"{resolve_object_url(target_type, target_uid)}#comment-{c['uid']}", } ) activities.sort(key=lambda a: a["created_at"], reverse=True) heatmap = get_activity_heatmap(profile_user["uid"]) heatmap_months = get_activity_months(heatmap) streak = get_streaks(profile_user["uid"]) is_following = False is_blocked = False is_muted = False if current_user: follows = get_table("follows") is_following = bool( follows.find_one( follower_uid=current_user["uid"], following_uid=profile_user["uid"], deleted_at=None, ) ) relations = get_user_relations(current_user["uid"]) is_blocked = profile_user["uid"] in relations["block"] is_muted = profile_user["uid"] in relations["mute"] is_owner = bool(current_user and current_user["uid"] == profile_user["uid"]) viewer_is_admin = bool(current_user and current_user.get("role") == "Admin") can_view_api_key = bool(current_user and (is_owner or viewer_is_admin)) api_key = profile_user.get("api_key", "") if can_view_api_key else "" ai_correction_enabled = ( bool(profile_user.get("ai_correction_enabled")) if is_owner else False ) ai_correction_sync = ( bool(profile_user.get("ai_correction_sync")) if is_owner else False ) ai_correction_prompt = ( (profile_user.get("ai_correction_prompt") or DEFAULT_CORRECTION_PROMPT) if is_owner else None ) ai_modifier_enabled = ( bool(profile_user.get("ai_modifier_enabled")) if is_owner else False ) ai_modifier_sync = ( bool(profile_user.get("ai_modifier_sync")) if is_owner else False ) ai_modifier_prompt = ( (profile_user.get("ai_modifier_prompt") or DEFAULT_MODIFIER_PROMPT) if is_owner else None ) from devplacepy.services.devii.interaction import prefs as interaction_prefs interactions_snap = ( interaction_prefs.snapshot("user", profile_user["uid"], profile_user) if is_owner else { "enabled": True, "source": None, "default": True, "override": None, } ) from devplacepy.services.telegram import store as telegram_store telegram_paired = ( telegram_store.is_paired(profile_user["uid"]) if is_owner else False ) profile_is_admin = profile_user.get("role") == "Admin" ai_quota = ( _ai_quota( profile_user["uid"], include_cost=viewer_is_admin, is_admin=profile_is_admin ) if (is_owner or viewer_is_admin) else None ) correction_usage = ( _correction_usage(profile_user["uid"], include_cost=viewer_is_admin) if (is_owner or viewer_is_admin) else None ) modifier_usage = ( _modifier_usage(profile_user["uid"], include_cost=viewer_is_admin) if (is_owner or viewer_is_admin) else None ) can_manage_customization = is_owner or viewer_is_admin customization_prefs = ( get_customization_prefs("user", profile_user["uid"]) if can_manage_customization else {"disable_global": False, "disable_pagetype": False} ) notification_prefs = ( get_notification_prefs(profile_user["uid"]) if (tab == "notifications" and can_manage_customization) else [] ) notif_telegram_paired = ( telegram_store.is_paired(profile_user["uid"]) if (tab == "notifications" and can_manage_customization) else False ) base = site_url(request) robots = "noindex,follow" if posts_count < 2 else "index,follow" bio = profile_user.get("bio", "") desc = ( bio or f"View {profile_user['username']}'s profile on DevPlace. {posts_count} posts." ) seo_ctx = base_seo_context( request, title=f"{profile_user['username']} (@{profile_user['username']})", description=desc, robots=robots, og_type="profile", og_image=avatar_url("multiavatar", avatar_seed(profile_user), 256), breadcrumbs=[ {"name": "Home", "url": "/feed"}, { "name": profile_user["username"], "url": f"/profile/{profile_user['username']}", }, ], schemas=[ website_schema(base), profile_page_schema(profile_user, posts_count, base), ], ) return respond( request, "profile.html", { **seo_ctx, "request": request, "user": current_user, "profile_user": profile_user, "posts": posts, "badges": badges, "achievements": achievements, "badge_total": badge_total, "badge_earned": badge_earned, "projects": projects, "gists": gists, "current_tab": tab, "posts_count": posts_count, "is_following": is_following, "is_blocked": is_blocked, "is_muted": is_muted, "is_owner": is_owner, "profile_online": presence.is_online(profile_user), "viewer_is_admin": viewer_is_admin, "media": media, "media_pagination": media_pagination, "can_view_api_key": can_view_api_key, "api_key": api_key, "ai_correction_enabled": ai_correction_enabled, "ai_correction_sync": ai_correction_sync, "ai_correction_prompt": ai_correction_prompt, "ai_modifier_enabled": ai_modifier_enabled, "ai_modifier_sync": ai_modifier_sync, "ai_modifier_prompt": ai_modifier_prompt, "interactions_enabled": interactions_snap["enabled"], "interactions_source": interactions_snap["source"], "interactions_default": interactions_snap["default"], "interactions_override": interactions_snap["override"], "telegram_paired": telegram_paired, "notif_telegram_paired": notif_telegram_paired, "can_manage_customization": can_manage_customization, "cust_disable_global": customization_prefs["disable_global"], "cust_disable_pagetype": customization_prefs["disable_pagetype"], "notification_prefs": notification_prefs, "ai_quota": ai_quota, "correction_usage": correction_usage, "modifier_usage": modifier_usage, "activities": activities, "rank": rank, "heatmap": heatmap, "heatmap_months": heatmap_months, "streak": streak, "people": people, "follow_pagination": follow_pagination, "followers_count": follow_counts["followers"], "following_count": follow_counts["following"], "awards": awards, "awards_pagination": awards_pagination, "awards_count": awards_count, "prominent_award": prominent_award, "can_give_award": can_give, }, model=ProfileOut, ) @router.post("/update") async def update_profile(request: Request, data: Annotated[ProfileForm, Depends(json_or_form(ProfileForm))]): user = require_user(request) users = get_table("users") users.update( { "uid": user["uid"], "bio": data.bio.strip(), "location": data.location.strip(), "git_link": data.git_link.strip(), "website": data.website.strip(), }, ["uid"], ) clear_user_cache(user["uid"]) schedule_correction(user, "users", user["uid"], request) schedule_modification(user, "users", user["uid"], request) logger.info(f"Profile updated for {user['username']}") audit.record( request, "profile.update", user=user, target_type="user", target_uid=user["uid"], target_label=user["username"], metadata={"fields": ["bio", "location", "git_link", "website"]}, summary=f"{user['username']} updated their profile", links=[audit.target("user", user["uid"], user["username"])], ) if any( value.strip() for value in (data.bio, data.location, data.git_link, data.website) ): track_action(user["uid"], "profile") return action_result(request, f"/profile/{user['username']}") @router.post("/regenerate-api-key") async def regenerate_api_key(request: Request): user = require_user(request) new_key = generate_uid() get_table("users").update({"uid": user["uid"], "api_key": new_key}, ["uid"]) clear_user_cache(user["uid"]) logger.info(f"API key regenerated for {user['username']}") audit.record( request, "profile.api_key.regenerate", user=user, target_type="user", target_uid=user["uid"], target_label=user["username"], summary=f"{user['username']} regenerated their API key", links=[audit.target("user", user["uid"], user["username"])], ) return JSONResponse({"api_key": new_key})