# retoor from typing import Optional from devplacepy.attachments import get_attachments_batch from devplacepy.database import ( get_table, get_users_by_uids, get_user_votes, get_vote_counts, get_comment_counts_by_post_uids, get_user_stars, text_search_clause, ) from devplacepy.services.devrant.serializers import serialize_rant, serialize_comment SORT_ORDER = { "recent": ["-created_at"], "top": ["-stars", "-created_at"], "algo": ["-stars", "-created_at"], } def _score_maps(user_uids: list) -> dict: return {uid: get_user_stars(uid) for uid in set(user_uids)} def build_rant_list(posts: list, viewer: Optional[dict]) -> list: post_uids = [post["uid"] for post in posts] user_uids = [post["user_uid"] for post in posts] authors = get_users_by_uids(list(set(user_uids))) user_scores = _score_maps(user_uids) my_votes = get_user_votes(viewer["uid"], post_uids) if viewer else {} comment_counts = get_comment_counts_by_post_uids(post_uids) attachments = get_attachments_batch("post", post_uids) return [ serialize_rant( post, authors=authors, comment_counts=comment_counts, user_scores=user_scores, my_votes=my_votes, attachments=attachments, viewer=viewer, ) for post in posts ] def list_rants(sort: str, limit: int, skip: int, viewer: Optional[dict]) -> list: order = SORT_ORDER.get(sort, SORT_ORDER["recent"]) posts = list( get_table("posts").find( deleted_at=None, order_by=order, _limit=limit, _offset=skip ) ) return build_rant_list(posts, viewer) def search_rants(term: str, viewer: Optional[dict], limit: int = 50) -> list: posts_table = get_table("posts") clause = text_search_clause( posts_table, term, ("title", "content"), author_field="user_uid" ) if clause is None: return [] posts = list( posts_table.find( clause, deleted_at=None, order_by=["-created_at"], _limit=limit ) ) return build_rant_list(posts, viewer) def load_rant_detail(post: dict, viewer: Optional[dict]) -> dict: rant = build_rant_list([post], viewer)[0] comments = list( get_table("comments").find( target_type="post", target_uid=post["uid"], deleted_at=None, order_by=["created_at"], ) ) comment_uids = [comment["uid"] for comment in comments] user_uids = [comment["user_uid"] for comment in comments] authors = get_users_by_uids(list(set(user_uids))) user_scores = _score_maps(user_uids) ups, downs = get_vote_counts(comment_uids) score_map = {uid: ups.get(uid, 0) - downs.get(uid, 0) for uid in comment_uids} my_votes = get_user_votes(viewer["uid"], comment_uids) if viewer else {} serialized = [ serialize_comment( comment, int(post["id"]), authors=authors, user_scores=user_scores, my_votes=my_votes, score_map=score_map, ) for comment in comments ] return {"rant": rant, "comments": serialized}