core/orm/community.py

import enum
import time

from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String, Text, distinct, func
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship

from auth.orm import Author
from services.db import BaseModel


class CommunityRole(enum.Enum):
    READER = "reader"  # can read and comment
    AUTHOR = "author"  # + can vote and invite collaborators
    ARTIST = "artist"  # + can be credited as featured artist
    EXPERT = "expert"  # + can add proof or disproof to shouts, can manage topics
    EDITOR = "editor"  # + can manage topics, comments and community settings
    ADMIN = "admin"

    @classmethod
    def as_string_array(cls, roles) -> list[str]:
        return [role.value for role in roles]

    @classmethod
    def from_string(cls, value: str) -> "CommunityRole":
        return cls(value)


class CommunityFollower(BaseModel):
    __tablename__ = "community_follower"

    community = Column(ForeignKey("community.id"), primary_key=True)
    follower = Column(ForeignKey("author.id"), primary_key=True)
    roles = Column(String, nullable=True)

    def __init__(self, community: int, follower: int, roles: list[str] | None = None) -> None:
        self.community = community  # type: ignore[assignment]
        self.follower = follower  # type: ignore[assignment]
        if roles:
            self.roles = ",".join(roles)  # type: ignore[assignment]

    def get_roles(self) -> list[CommunityRole]:
        roles_str = getattr(self, "roles", "")
        return [CommunityRole(role) for role in roles_str.split(",")] if roles_str else []


class Community(BaseModel):
    __tablename__ = "community"

    name = Column(String, nullable=False)
    slug = Column(String, nullable=False, unique=True)
    desc = Column(String, nullable=False, default="")
    pic = Column(String, nullable=False, default="")
    created_at = Column(Integer, nullable=False, default=lambda: int(time.time()))
    created_by = Column(ForeignKey("author.id"), nullable=False)
    settings = Column(JSON, nullable=True)
    updated_at = Column(Integer, nullable=True)
    deleted_at = Column(Integer, nullable=True)
    private = Column(Boolean, default=False)

    followers = relationship("Author", secondary="community_follower")
    created_by_author = relationship("Author", foreign_keys=[created_by])

    @hybrid_property
    def stat(self):
        return CommunityStats(self)

    @property
    def role_list(self):
        return self.roles.split(",") if self.roles else []

    @role_list.setter
    def role_list(self, value) -> None:
        self.roles = ",".join(value) if value else None  # type: ignore[assignment]

    def is_followed_by(self, author_id: int) -> bool:
        # Check if the author follows this community
        from services.db import local_session

        with local_session() as session:
            follower = (
                session.query(CommunityFollower)
                .filter(CommunityFollower.community == self.id, CommunityFollower.follower == author_id)
                .first()
            )
            return follower is not None

    def get_role(self, author_id: int) -> CommunityRole | None:
        # Get the role of the author in this community
        from services.db import local_session

        with local_session() as session:
            follower = (
                session.query(CommunityFollower)
                .filter(CommunityFollower.community == self.id, CommunityFollower.follower == author_id)
                .first()
            )
            if follower and follower.roles:
                roles = follower.roles.split(",")
                return CommunityRole.from_string(roles[0]) if roles else None
            return None


class CommunityStats:
    def __init__(self, community) -> None:
        self.community = community

    @property
    def shouts(self):
        from orm.shout import Shout

        return self.community.session.query(func.count(Shout.id)).filter(Shout.community == self.community.id).scalar()

    @property
    def followers(self):
        return (
            self.community.session.query(func.count(CommunityFollower.follower))
            .filter(CommunityFollower.community == self.community.id)
            .scalar()
        )

    @property
    def authors(self):
        from orm.shout import Shout

        # author has a shout with community id and its featured_at is not null
        return (
            self.community.session.query(func.count(distinct(Author.id)))
            .join(Shout)
            .filter(
                Shout.community == self.community.id,
                Shout.featured_at.is_not(None),
                Author.id.in_(Shout.authors),
            )
            .scalar()
        )


class CommunityAuthor(BaseModel):
    __tablename__ = "community_author"

    id = Column(Integer, primary_key=True)
    community_id = Column(Integer, ForeignKey("community.id"))
    author_id = Column(Integer, ForeignKey("author.id"))
    roles = Column(Text, nullable=True, comment="Roles (comma-separated)")

    @property
    def role_list(self):
        return self.roles.split(",") if self.roles else []

    @role_list.setter
    def role_list(self, value) -> None:
        self.roles = ",".join(value) if value else None  # type: ignore[assignment]