Temporary git for an interview

interview.py

import datetime
from dataclasses import dataclass
from typing import Optional, TypedDict, Union

from utils.cast import safe_int
from utils.convert import from_json, to_json
from utils.db import get

import models.queries
from language.feeds import repliers_badge
from media.images import prepare_image_collection
from media.video import prepare_video_collection
from models.errors import PostNotFoundError, ReplyNotFoundError, BadNaming
from models.media import PostImage, PostVideo, ReplyImage, ReplyVideo
from models.mention import WithMentions
from models.request import WithImages, WithVideos, WithImagesToRemove, WithVideosToRemove


@dataclass(frozen=True)
class MentionRecord:
    user_id: int
    start: int
    post_id: str
    reply_id: str
    username: str
    avatar: str

    @classmethod
    def from_dict(cls, raw: dict):
        return cls(
            user_id=raw.get('userId'),
            start=raw.get('initialCharIndex'),
            post_id=raw.get('postId'),
            reply_id=raw.get('replyId'),
            username=raw.get('userName'),
            avatar=raw.get('userAvatarUrl'),
        )

    @property
    def as_push(self):
        return {'userId': str(self.user_id)}


@dataclass
class ImageRef:
    """
    An identifier of an image in an S3 bucket
    and whether it belongs to an non-public post
    """
    reference: str
    protected: bool

    @classmethod
    def from_dict(cls, row: dict):
        return cls(
            reference=row['ref'],
            protected=row['protected'],
        )


class WithAuthor:
    """
    Interface that indicates that the object
    was authored by a user.
    """

    def __init__(self, user_id: int):
        self.author = user_id


class WithStoredMentions:
    """
    Interface that indicates that the object
    has database records of mentions.
    """

    def __init__(self, records: list):
        self.stored_mentions = list(map(MentionRecord.from_dict, records or []))

    @property
    def stored_mention_records(self) -> list:
        return [mention.as_push for mention in self.stored_mentions]


class Post(WithMentions, WithStoredMentions):
    def __init__(self, raw: dict):
        self.raw = raw
        self.id = raw.get('id')
        self.url = raw.get('webSource', {}).get('siteUrl')
        self.privacy: str = raw.get('commentType')
        self.author_id: str = raw.get('authorId')
        self.blocked: Optional[list[int]] = raw.get('blocked')
        self.images: list[PostImage] = [PostImage.from_row(each) for each in raw.get('images') or []]
        self.videos: list[PostVideo] = [PostVideo.from_row(each) for each in raw.get('videos') or []]
        WithMentions.__init__(self, raw.get('comment'))
        WithStoredMentions.__init__(self, raw.get('mentions'))

    def __str__(self):
        return f'Post #{self.id}'

    def __repr__(self):
        return f'Post #{self.id}: {self.content}'

    @property
    def is_public(self) -> bool:
        return self.privacy == 'PUBLIC'

    @property
    def is_private(self) -> bool:
        return self.privacy == 'PRIVATE'

    @property
    def is_sfp(self) -> bool:
        return 'specific' in self.privacy.lower()

    def is_by(self, user_id: int) -> bool:
        return f'{self.author_id}' == f'{user_id}'

    def sf(self) -> list[int]:
        return [each['userId'] for each in self.raw.get('specific') or []]

    def is_shared_to(self, user_id: int) -> bool:
        return user_id in self.sf()

    @property
    def is_of_blocked_author(self) -> bool:
        """
        :return: whether this post has a block relation with the session owner
        """
        return safe_int(self.author_id) in self.blocked

    def prepared(self) -> dict:
        return prepare_post(self.raw)

    @property
    def public(self) -> dict:
        return {
            'post': self.prepared(),
        }

    @property
    def as_push(self) -> dict:
        specific = list(
            map(
                lambda user: {'userId': str(user.get('userId'))},
                self.raw.get('specific') or []
            )
        )

        return {
            'post': self,
            'comment': self.content,
            'commentType': self.raw.get('commentType'),
            'site_url': self.raw.get('webSource', {}).get('siteUrl'),
            'link': self.raw.get('webSource', {}).get('siteUrl'),
            'specific': specific,
            'mentions': self.stored_mention_records,
            'user_id': int(self.raw.get('authorId')),
            'post_id': self.id,
            'page_id': self.raw.get('webSource', {}).get('id'),
            'domain_id': self.raw.get('webSource', {}).get('domain', {}).get('id'),
        }

    @property
    def as_push_on_edit(self) -> dict:
        return {
            'post': self,
            'user_id': int(self.raw.get('authorId')),
            'site_url': self.raw.get('webSource', {}).get('siteUrl'),
            'comment': self.content,
            'mentions': self.stored_mention_records,
            'commentId': self.id,
            'post_id': self.id,
        }


class Reply(WithMentions, WithStoredMentions):

    def __init__(
            self,
            reply_id,
            text: str,
            mentions: list[dict],
            author,
            parent_reply_id: int = None,
            post: Post = None,
            raw: dict = None,
    ):
        self.id = reply_id
        self.text = text
        self.parent_reply_id: int = parent_reply_id
        self.post: Post = post
        self.author: dict = author
        self.raw: dict = raw
        self.images: list[PostImage] = [ReplyImage.from_row(each) for each in raw.get('images') or []]
        self.videos: list[PostVideo] = [ReplyVideo.from_row(each) for each in raw.get('videos') or []]
        WithStoredMentions.__init__(self, mentions)
        WithMentions.__init__(self, text)

    @classmethod
    def from_dict(cls, raw: dict, post: Post = None):
        return cls(
            reply_id=raw.get('id'),
            text=raw.get('text'),
            mentions=raw.get('mentions'),
            author=raw.get('author'),
            parent_reply_id=raw.get('parentReplyId'),
            post=post,
            raw=raw,
        )

    @classmethod
    def child(cls, raw: dict, reply_id: int, post: Post = None):
        children = raw.get('replies')
        if children:
            match = list(filter(lambda reply: reply.get('id') == reply_id, children))
            if match:
                return cls.from_dict(match[0], post)
        return cls.from_dict(raw, post)

    @property
    def as_push(self) -> dict:
        return {
            'reply': self,
            'idThread': self.post.id,
            'comment': self.text,
            'author': self.author,
            'post': self.post,
            'mentions': self.stored_mention_records,
            'reply_id': self.id,
            'user_id': self.author.get('userId'),
            'toReplyId': self.parent_reply_id,
        }

    @property
    def as_push_on_edit(self) -> dict:
        return {
            'reply': self,
            'idThread': self.post.id,
            'post_id': self.post.id,
            'comment': self.text,
            'author': self.author,
            'post': self.post,
            'mentions': self.stored_mention_records,
            'reply_id': self.id,
            'user_id': self.author.get('userId'),
            'toReplyId': self.parent_reply_id,
        }

    def is_by(self, user_id: int) -> bool:
        return self.author.get('userId') == f'{user_id}'

    @property
    def public(self) -> dict:
        return {
            'reply': prepare_reply(self.raw),
        }


class WithPost:
    """
    Interface that indicates that the object
    has a post in it.
    """

    def __init__(self, post_id: int, user_id: int = None):
        self.post_id = post_id
        self.post: Post = hydrate_post(self.post_id, user_id)
        if not self.post:
            raise PostNotFoundError

    def __str__(self):
        return f'{self.__class__.__name__} for {self.post}'

    def __repr__(self):
        return str(self)


class WithReply:
    """
    Interface that indicates that the object
    has a reply in it.
    """

    def __init__(self, reply_id: int, user_id: int = None):
        self.reply_id = reply_id
        self.reply: Reply = hydrate_reply(self.reply_id, user_id)
        if not self.reply:
            raise ReplyNotFoundError


@dataclass
class BaseUgcRequest(WithMentions, WithImages, WithAuthor, WithVideos):
    """
    Making a post and replies is very similar.

    Shared features are:
    - mention parsing and saving
    - image processing
    - image uploading to a file storage
    - permissions and prohibitions

    This combines that functionality
    by transforming the HTTP request into a more specific
    data structure that handles every bit separately.
    """

    def __init__(self, request: dict, user_id: int, public: bool):
        self.content = request.get('content')
        WithImages.__init__(self, images=WithImages.from_http(request, not public))
        WithVideos.__init__(self, videos=WithVideos.from_http(request, not public))
        WithMentions.__init__(self, self.content)
        WithAuthor.__init__(self, user_id)

        self.validate_media_ordering()

    @property
    def as_query(self) -> dict:
        return {
            'mentions': self.mention_records,
            'content': self.content,
            'user_id': self.author,
            'images': self.image_refs,
            'videos': self.video_refs,
        }

    @property
    def is_of_public_post(self) -> bool:
        raise NotImplementedError

    def validate_media_ordering(self) -> None:
        """
        Checks if the request has overlapping media file names,
        such as video_1 and image_1.

        :raises BadNaming error if it does
        """
        image_orders = list(map(lambda each: each.order, self.images))
        video_orders = list(map(lambda each: each.order, self.videos))
        intersection = [order for order in image_orders if order in video_orders]
        if intersection:
            raise BadNaming('Duplicate naming detected')


class MakePostRequest(BaseUgcRequest):
    def __init__(self, request: dict, user_id: int):
        self.url = request.get('url')
        self.privacy = request.get('privacy')
        self.sf = self.specific_followers(request.get('specific'))
        super().__init__(request, user_id, self.is_of_public_post)
        self.validate_sf()

    @property
    def is_of_public_post(self) -> bool:
        return self.privacy == 'PUBLIC'

    @property
    def as_query(self) -> dict:
        return {
            **super().as_query,
            'url': self.url,
            'privacy': self.privacy,
            'sf': self.sf,
        }

    @staticmethod
    def specific_followers(entry) -> list[int]:
        if not entry:
            return []
        decoded = from_json(entry)
        validation = 'specific attribute must be a list of jsonified user ids'
        assert all(map(lambda each: isinstance(safe_int(each), int), decoded)), validation
        return decoded

    def validate_sf(self):
        is_specific_type: bool = self.privacy.startswith('SPECIFIC')
        has_sf: bool = bool(self.sf) or self.privacy == 'SPECIFIC_ALL'
        if is_specific_type:
            assert has_sf, f'Array of users required in "specific" attribute if privacy is {self.privacy}'
        if has_sf:
            assert is_specific_type, 'Correct privacy required if "specific" attribute is present'


class WithPostRequest(BaseUgcRequest, WithPost):
    def __init__(self, request: dict, user_id: int):
        WithPost.__init__(self, request.get('postId'), user_id)
        BaseUgcRequest.__init__(self, request, user_id, self.is_of_public_post)

    @property
    def as_query(self) -> dict:
        return {
            **super().as_query,
            'post_id': self.post_id,
        }

    @property
    def is_of_public_post(self) -> bool:
        return self.post.is_public

    @property
    def is_of_own_post(self) -> bool:
        return self.post.is_by(self.author)

    def should_block(self, user_id: Union[int, str]) -> bool:
        return safe_int(user_id) in self.post.blocked


class WithMentionsOnEdit:
    """
    Posts and replies might have mentions in them at creation,
    but when they are edited, they might be new ones. While editing,
    we are only interested in the new ones, if any.
    mentions_on_edit is a collection of user IDs.

    """

    def stored_mentions(self) -> list[MentionRecord]:
        raise NotImplementedError

    def mentions_on_edit(self, edited: WithStoredMentions) -> list[int]:
        existing = {mention.user_id for mention in self.stored_mentions()}
        incoming = {mention.user_id for mention in edited.stored_mentions}
        return list(incoming - existing)


class EditRequest(WithPostRequest, WithImagesToRemove, WithVideosToRemove, WithMentionsOnEdit):

    def __init__(self, request: dict, user_id: int):
        WithPostRequest.__init__(self, request, user_id)

        images = from_json(request.get('imagesToRemove') or '[]')
        WithImagesToRemove.__init__(self, ids=images)

        videos = from_json(request.get('videosToRemove') or '[]')
        WithVideosToRemove.__init__(self, ids=videos)

    @property
    def as_query(self) -> dict:
        return {
            **super().as_query,
            'images_to_remove': self.images_to_remove,
            'videos_to_remove': self.videos_to_remove,
            'state': to_json(self.post.public['post']),
        }

    def stored_mentions(self) -> list[MentionRecord]:
        return self.post.stored_mentions


class ReplyRequest(WithPostRequest):
    def __init__(self, request: dict, user_id: int):
        super().__init__(request, user_id)
        self.parent_reply_id: int = request.get('parentReplyId')

    @property
    def as_query(self) -> dict:
        return {
            **super().as_query,
            'parent_reply_id': self.parent_reply_id,
        }

    @property
    def is_allowed(self) -> bool:
        author = self.author
        return any(
            (
                self.post.is_public,
                self.post.is_by(author),
                self.post.is_shared_to(author),

            )
        ) and not self.should_block(self.post.author_id)


class EditReplyRequest(EditRequest, WithImagesToRemove, WithReply, WithMentionsOnEdit):
    def __init__(self, request: dict, user_id: int):
        self.reply_id: int = request.get('replyId')
        self.user_id: int = user_id
        super().__init__(request, user_id)
        WithReply.__init__(self, reply_id=self.reply_id, user_id=user_id)

    @property
    def as_query(self) -> dict:
        return {
            **super().as_query,
            'reply_id': self.reply_id,
            'state': to_json(self.reply.public['reply'])
        }

    @property
    def is_of_own_reply(self) -> bool:
        return self.reply.is_by(self.user_id)

    @property
    def is_allowed(self) -> bool:
        return self.is_of_own_reply

    def stored_mentions(self) -> list[MentionRecord]:
        return self.reply.stored_mentions


class WithReplyRequest(ReplyRequest, WithReply):
    def __init__(self, request: dict, user_id: int):
        self.reply_id: int = request.get('replyId')
        self.user_id: int = user_id
        ReplyRequest.__init__(self, request, user_id)
        WithReply.__init__(self, reply_id=self.reply_id, user_id=user_id)


def hydrate_post(post_id: int, user_id: Optional[int] = None, include_blocks: bool = True) -> Post:
    options = {
        'post_id': post_id,
        'user_id': user_id,
        'include_blocks': include_blocks
    }
    rows = get(models.queries.hydrate_post, options)
    if rows:
        raw = rows[0]['post']
        if raw:
            return Post(raw)
    raise PostNotFoundError


def hydrate_reply(reply_id: int, user_id: int) -> Reply:
    options = {
        'reply_id': reply_id,
        'user_id': user_id,
    }
    rows = get(models.queries.hydrate_reply, options)
    if rows:
        raw = rows[0]['reply']
        if raw:
            return Reply.from_dict(raw)
    raise PostNotFoundError


def _post_top_repliers(post: dict) -> dict:
    try:
        repliers = post.get('repliers')
        if repliers:
            names = list(map(lambda r: r.get('userName'), repliers))
            users = repliers
            post['repliers'] = TopRepliers(label=repliers_badge(names), users=users)
            return post

        post['repliers'] = TopRepliers(label='', users=[])
        return post
    except:
        post['repliers'] = TopRepliers(label='', users=[])
        return post


def prepare_post(post: dict) -> dict:
    post = _post_top_repliers(post)
    post = prepare_image_collection(post)
    post = prepare_video_collection(post)
    return post


def prepare_reply(reply: dict) -> dict:
    children = reply.get('replies')
    reply = prepare_video_collection(prepare_image_collection(reply))
    if children:
        reply['replies'] = list(map(prepare_video_collection, map(prepare_image_collection, children)))
    return reply