devplacepy/ranking.py at c79dfd029123e1dd3c041cf17dd2fd33eca52680

 # retoor <retoor@molodetz.nl>
 from .core import TTLCache, _in_clause, _now_iso, db, get_table
 from .users import get_users_by_uids
 from .soft_delete import soft_delete, soft_delete_in
 VOTABLE_TARGETS: dict[str, str] = {
     "post": "posts",
     "project": "projects",
     "gist": "gists",
     "comment": "comments",
 }
 STAR_TARGETS: set[str] = {"post", "project", "gist"}
 _authors_cache = TTLCache(ttl=300, max_size=200)
 def _ranked_authors() -> list:
     cached = _authors_cache.get("ranked")
     if cached is not None:
         return cached
     sources = [
         (target_type, table_name)
         for target_type, table_name in VOTABLE_TARGETS.items()
         if table_name in db.tables
     ]
     if "votes" not in db.tables or not sources:
         _authors_cache.set("ranked", [])
         return []
     target_union = " UNION ALL ".join(
         f"SELECT uid, user_uid, '{target_type}' AS target_type FROM {table_name} WHERE deleted_at IS NULL"
         for target_type, table_name in sources
     )
     rows = db.query(
         f"SELECT t.user_uid, SUM(v.value) AS total "
         f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
         f"WHERE v.deleted_at IS NULL "
         f"GROUP BY t.user_uid HAVING SUM(v.value) > 0 ORDER BY total DESC"
     )
     ranked = [(row["user_uid"], row["total"]) for row in rows]
     users_map = get_users_by_uids([uid for uid, _ in ranked])
     authors = []
     for uid, total in ranked:
         user = users_map.get(uid)
         if user:
             author = dict(user)
             author["stars"] = total
             authors.append(author)
     _authors_cache.set("ranked", authors)
     _authors_cache.set(
         "rank_map",
         {author["uid"]: position for position, author in enumerate(authors, start=1)},
     )
     return authors
 def _rank_map() -> dict:
     cached = _authors_cache.get("rank_map")
     if cached is not None:
         return cached
     _ranked_authors()
     return _authors_cache.get("rank_map") or {}
 def get_top_authors(limit: int = 5) -> list:
     return _ranked_authors()[:limit]
 def get_leaderboard(limit: int = 50, offset: int = 0) -> list:
     sliced = _ranked_authors()[offset : offset + limit]
     leaderboard = []
     for position, author in enumerate(sliced, start=offset + 1):
         entry = dict(author)
         entry["rank"] = position
         leaderboard.append(entry)
     return leaderboard
 def get_user_rank(user_uid: str):
     return _rank_map().get(user_uid)
 def get_user_stars(user_uid: str) -> int:
     if "votes" not in db.tables:
         return 0
     target_union = " UNION ALL ".join(
         f"SELECT uid, '{target_type}' AS target_type FROM {table_name} WHERE user_uid = :u AND deleted_at IS NULL"
         for target_type, table_name in VOTABLE_TARGETS.items()
         if table_name in db.tables
     )
     if not target_union:
         return 0
     for row in db.query(
         f"SELECT COALESCE(SUM(v.value), 0) AS s "
         f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
         f"WHERE v.deleted_at IS NULL",
         u=user_uid,
     ):
         return row["s"] or 0
     return 0
 def update_target_stars(target_type: str, target_uid: str, net_stars: int) -> None:
     table_name = VOTABLE_TARGETS.get(target_type)
     if not table_name:
         return
     if target_type in STAR_TARGETS:
         get_table(table_name).update({"uid": target_uid, "stars": net_stars}, ["uid"])
     _authors_cache.clear()
 def soft_delete_engagement(target_type: str, target_uids: list, deleted_by: str) -> None:
     uids = [uid for uid in (target_uids or []) if uid]
     if not uids:
         return
     stamp = _now_iso()
     soft_delete_in(
         "reactions", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
     )
     soft_delete_in(
         "bookmarks", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
     )
     if target_type == "post" and "polls" in db.tables:
         for uid in uids:
             for poll in db["polls"].find(post_uid=uid, deleted_at=None):
                 soft_delete("poll_votes", deleted_by, stamp=stamp, poll_uid=poll["uid"])
                 soft_delete("poll_options", deleted_by, stamp=stamp, poll_uid=poll["uid"])
             soft_delete("polls", deleted_by, stamp=stamp, post_uid=uid)
 def delete_engagement(target_type: str, target_uids: list) -> None:
     uids = [uid for uid in (target_uids or []) if uid]
     if not uids:
         return
     if "reactions" in db.tables:
         placeholders, params = _in_clause(uids)
         params["tt"] = target_type
         db.query(
             f"DELETE FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders})",
             **params,
         )
     if "bookmarks" in db.tables:
         placeholders, params = _in_clause(uids)
         params["tt"] = target_type
         db.query(
             f"DELETE FROM bookmarks WHERE target_type=:tt AND target_uid IN ({placeholders})",
             **params,
         )
     if target_type == "post" and "polls" in db.tables:
         for uid in uids:
             for poll in db["polls"].find(post_uid=uid):
                 if "poll_votes" in db.tables:
                     db["poll_votes"].delete(poll_uid=poll["uid"])
                 if "poll_options" in db.tables:
                     db["poll_options"].delete(poll_uid=poll["uid"])
             db["polls"].delete(post_uid=uid)
 def get_target_owner_uid(target_type: str, target_uid: str) -> str | None:
     table_name = VOTABLE_TARGETS.get(target_type)
     if not table_name:
         return None
     row = get_table(table_name).find_one(uid=target_uid, deleted_at=None)
     return row["user_uid"] if row else None