pyrom/app/models.py

from .db import Model, db
from .constants import PermissionLevel

class Users(Model):
    table = "users"

    def get_avatar_url(self):
        return Avatars.find({"id": self.avatar_id}).file_path

    def is_guest(self):
        return self.permission == PermissionLevel.GUEST.value

    def is_mod(self):
        return self.permission >= PermissionLevel.MODERATOR.value

    def is_admin(self):
        return self.permission == PermissionLevel.ADMIN.value

    def is_system(self):
        return self.permission == PermissionLevel.SYSTEM.value

    def is_default_avatar(self):
        return self.avatar_id == 1

    def get_latest_posts(self):
        q = """SELECT
                posts.id, posts.created_at, post_history.content, post_history.edited_at, threads.title AS thread_title, topics.name as topic_name, threads.slug as thread_slug
            FROM
                posts
            JOIN
                post_history ON posts.current_revision_id = post_history.id
            JOIN
                threads ON posts.thread_id = threads.id
            JOIN
                topics ON threads.topic_id = topics.id
            WHERE
                posts.user_id = ?
            ORDER BY posts.created_at DESC
            LIMIT 10"""
        return db.query(q, self.id)

    def get_post_stats(self):
        q = """SELECT
                COUNT(DISTINCT posts.id) AS post_count,
                COUNT(DISTINCT threads.id) AS thread_count,
                MAX(threads.title) FILTER (WHERE threads.created_at = latest.created_at) AS latest_thread_title,
                MAX(threads.slug) FILTER (WHERE threads.created_at = latest.created_at) AS latest_thread_slug
            FROM users
            LEFT JOIN posts ON posts.user_id = users.id
            LEFT JOIN threads ON threads.user_id = users.id
            LEFT JOIN (
                SELECT user_id, MAX(created_at) AS created_at
                FROM threads
                GROUP BY user_id
                ) latest ON latest.user_id = users.id
            WHERE users.id = ?"""
        return db.fetch_one(q, self.id)

    def get_all_subscriptions(self):
        q = """
            SELECT threads.title AS thread_title, threads.slug AS thread_slug
            FROM
                threads
            JOIN
                subscriptions ON subscriptions.thread_id = threads.id
            WHERE
                subscriptions.user_id = ?"""
        return db.query(q, self.id)


class Topics(Model):
    table = "topics"

    @classmethod
    def get_list(_cls):
        q = """
            SELECT
                topics.id, topics.name, topics.slug, topics.description, topics.is_locked,
                users.username AS latest_thread_username,
                threads.title AS latest_thread_title,
                threads.slug AS latest_thread_slug,
                threads.created_at AS latest_thread_created_at
            FROM
                topics
            LEFT JOIN (
                SELECT
                *,
                row_number() OVER (PARTITION BY threads.topic_id ORDER BY threads.created_at DESC) as rn
            FROM
                threads
            ) threads ON threads.topic_id = topics.id AND threads.rn = 1
            LEFT JOIN
                users on users.id = threads.user_id
            ORDER BY
                topics.sort_order ASC"""
        return db.query(q)

    @classmethod
    def get_active_threads(cls):
        q = """
            WITH ranked_threads AS (
                SELECT
                    threads.topic_id, threads.id AS thread_id, threads.title AS thread_title, threads.slug AS thread_slug,
                    posts.id AS post_id, posts.created_at AS post_created_at,
                    users.username,
                    ROW_NUMBER() OVER (PARTITION BY threads.topic_id ORDER BY posts.created_at DESC) AS rn
                FROM
                    threads
                JOIN
                    posts ON threads.id = posts.thread_id
                LEFT JOIN
                    users ON posts.user_id = users.id
            )
            SELECT
                topic_id,
                thread_id, thread_title, thread_slug,
                post_id, post_created_at,
                username
            FROM
                ranked_threads
            WHERE
                rn = 1
            ORDER BY
                topic_id"""

        active_threads_raw = db.query(q)
        active_threads = {}
        for thread in active_threads_raw:
            active_threads[int(thread['topic_id'])] = {
                'thread_title': thread['thread_title'],
                'thread_slug': thread['thread_slug'],
                'post_id': thread['post_id'],
                'username': thread['username'],
                'post_created_at': thread['post_created_at']
            }
        return active_threads

    def get_threads(self, per_page, page, sort_by = "activity"):
        order_clause = ""
        if sort_by == "thread":
            order_clause = "ORDER BY threads.is_stickied DESC, threads.created_at DESC"
        else:
            order_clause = "ORDER BY threads.is_stickied DESC, latest_post_created_at DESC"

        q = """
            SELECT
                threads.title, threads.slug, threads.created_at, threads.is_locked, threads.is_stickied,
                users.username AS started_by,
            u.username AS latest_post_username,
                ph.content AS latest_post_content,
                posts.created_at AS latest_post_created_at,
                posts.id AS latest_post_id
            FROM
                threads
            JOIN users ON users.id = threads.user_id
            JOIN (
                SELECT
                    posts.thread_id,
                    posts.id,
                    posts.user_id,
                    posts.created_at,
                    posts.current_revision_id,
                    ROW_NUMBER() OVER (PARTITION BY posts.thread_id ORDER BY posts.created_at DESC) AS rn
            FROM
                posts
            ) posts ON posts.thread_id = threads.id AND posts.rn = 1
            JOIN
                post_history ph ON ph.id = posts.current_revision_id
            JOIN
                users u ON u.id = posts.user_id
            WHERE
            threads.topic_id = ?
        """ + order_clause + " LIMIT ? OFFSET ?"

        return db.query(q, self.id, per_page, (page - 1) * per_page)


class Threads(Model):
    table = "threads"

    def get_posts(self, limit, offset):
        q = Posts.FULL_POSTS_QUERY + " WHERE posts.thread_id = ? ORDER BY posts.created_at ASC LIMIT ? OFFSET ?"
        return db.query(q, self.id, limit, offset)

    def locked(self):
        return bool(self.is_locked)

    def stickied(self):
        return bool(self.is_stickied)

class Posts(Model):
    FULL_POSTS_QUERY = """
        SELECT
            posts.id, posts.created_at, post_history.content, post_history.edited_at, users.username, users.status, avatars.file_path AS avatar_path, posts.thread_id, users.id AS user_id, post_history.original_markup, users.signature_rendered, threads.slug AS thread_slug, threads.is_locked AS thread_is_locked
        FROM
            posts
        JOIN
            post_history ON posts.current_revision_id = post_history.id
        JOIN
            users ON posts.user_id = users.id
        JOIN
            threads ON posts.thread_id = threads.id
        LEFT JOIN
            avatars ON users.avatar_id = avatars.id"""

    table = "posts"

class PostHistory(Model):
    table = "post_history"

class Sessions(Model):
    table = "sessions"

class Avatars(Model):
    table = "avatars"

class Subscriptions(Model):
    table = "subscriptions"