devplacepy/devplacepy/routers/profile.py

import logging
from typing import Annotated
from fastapi import APIRouter, Request, Form
from devplacepy.models import ProfileForm
from fastapi.responses import HTMLResponse, JSONResponse
from devplacepy.database import (
    get_table,
    db,
    get_user_stars,
    get_user_rank,
    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,
)
from devplacepy.content import enrich_items
from devplacepy.utils import (
    get_current_user,
    require_user,
    require_user_api,
    time_ago,
    clear_user_cache,
    not_found,
    generate_uid,
)
from devplacepy.responses import respond, action_result
from devplacepy.schemas import ProfileOut
from devplacepy.avatar import avatar_url
from devplacepy.seo import (
    base_seo_context,
    site_url,
    website_schema,
    profile_page_schema,
)
from devplacepy.services.manager import service_manager

logger = logging.getLogger(__name__)
router = APIRouter()


def _ai_quota(
    user_uid: str, include_cost: bool = False, is_admin: bool = False
) -> dict | None:
    devii = service_manager.get_service("devii")
    if devii is None:
        return None
    try:
        limit = devii.daily_limit_for("user", is_admin)
        spent = devii.spent_24h("user", user_uid)
        turns = devii.hub().ledger.turns_24h("user", user_uid)
    except Exception:
        logger.exception("Failed to compute AI quota for %s", user_uid)
        return None
    used_pct = round(min(100.0, spent / limit * 100), 1) if limit > 0 else 0.0
    quota = {"used_pct": used_pct, "turns": turns}
    if include_cost:
        quota["spent_usd"] = round(spent, 4)
        quota["limit_usd"] = round(limit, 2)
    return quota


@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": []})
    if "users" in db.tables:
        rows = db.query(
            "SELECT uid, username FROM users WHERE username LIKE :q LIMIT 10",
            q=f"%{q}%",
        )
        results = [{"uid": r["uid"], "username": r["username"]} for r in rows]
    else:
        results = []
    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("/{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")
    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
        )

    posts = []
    if tab == "posts":
        posts_table = get_table("posts")
        raw_posts = list(
            posts_table.find(user_uid=profile_user["uid"], order_by=["-created_at"])
        )
        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"]))
    can_see_private = bool(
        current_user
        and (
            current_user["uid"] == profile_user["uid"]
            or current_user.get("role") == "Admin"
        )
    )
    projects = list(get_table("projects").find(user_uid=profile_user["uid"]))
    if not can_see_private:
        projects = [p for p in projects if not p.get("is_private")]
    gists_raw = list(get_table("gists").find(user_uid=profile_user["uid"]))
    gists = []
    for g in gists_raw:
        gists.append({"gist": g, "time_ago": time_ago(g["created_at"])})
    posts_count = len(posts) or get_table("posts").count(user_uid=profile_user["uid"])

    activities = []
    if tab == "activity":
        posts_table = get_table("posts")
        for p in posts_table.find(
            user_uid=profile_user["uid"], 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"],
                }
            )
        comments_table = get_table("comments")
        for c in comments_table.find(
            user_uid=profile_user["uid"], 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,
                }
            )
        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
    if current_user:
        follows = get_table("follows")
        is_following = bool(
            follows.find_one(
                follower_uid=current_user["uid"], following_uid=profile_user["uid"]
            )
        )

    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 ""
    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
    )

    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", profile_user["username"], 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,
            "projects": projects,
            "gists": gists,
            "current_tab": tab,
            "posts_count": posts_count,
            "is_following": is_following,
            "is_owner": is_owner,
            "can_view_api_key": can_view_api_key,
            "api_key": api_key,
            "ai_quota": ai_quota,
            "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"],
        },
        model=ProfileOut,
    )


@router.post("/update")
async def update_profile(request: Request, data: Annotated[ProfileForm, Form()]):
    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"])

    logger.info(f"Profile updated for {user['username']}")
    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']}")
    return JSONResponse({"api_key": new_key})