su-automod/classes/post.py

import requests
import datetime
import boto3
import var
import classes.rating
from functions.delete_post import delete_post
from beloved_logger import logger


class Post:
    post_id = None
    post_description = None
    post_name = None
    image = None
    user_id = None
    poster_username = None
    poster_sign_up_time = None
    response = None
    date = None
    infringing_words = []
    ratings = None
    post_will_be_deleted = None
    deletion_status = None

    def __init__(self, post_id):
        logger.info(f"Post was made with ID {post_id}")
        self.post_id = post_id
        self.infringing_words = []

    async def get_values(self):
        response = requests.get(f"https://{var.site}/posts/{self.post_id}",
                                headers={'User-Agent': 'automod',
                                         "Accept": "application/json,text/plain, */*"})

        if response.status_code == 200:
            self.response = response.json()
            if response.json()['post']['resource_url']:
                self.image = requests.get(self.response['post']['resource_url']).content

            self.post_name = self.response['post']["title"]
            self.user_id = self.response['post']['profile_id']
            self.poster_username = self.response['post']['profile']['username']
            self.poster_sign_up_time = self.response['post']['profile']['created_at']
            self.date = self.response['post']['created_at']
            self.post_description = self.response['post']["description"]

            logger.debug(f"Got post {self.post_id} with details {self.response}")
        else:
            logger.info(f"An error occurred getting post with ID {self.post_id} for reason {response.status_code}")
            return False

    async def analyse_text(self):
        logger.info(f"Getting text analysis of post with post ID {self.post_id}")

        logger.info(self.post_name)
        logger.info(self.post_description)
        logger.info(var.filters_list)

        if self.post_name:
            for word in var.filters_list:
                if word.lower() in self.post_name.lower() and word not in self.infringing_words:
                    logger.info(f"Post {self.post_id}'s name contained infringing word {word}")
                    self.infringing_words.append(word)

        if self.post_description:
            for word in var.filters_list:
                if word.lower() in self.post_description.lower() and word not in self.infringing_words:
                    logger.info(f"Post {self.post_id}'s description contained infringing word {word}")
                    self.infringing_words.append(word)

    async def analyse_image(self):
        logger.info(f"Getting image analysis of post with post ID {self.post_id}")
        if self.image:
            logger.info(f"Post {self.post_id} had an image")
            await self.detect_image()
            if self.ratings:
                logger.info(f"Post {self.post_id} has ratings: {await self.get_ratings_string()}")
                await self.check_ratings()
                if self.post_will_be_deleted:
                    logger.info(f"Post {self.post_id} will be deleted")
                    await self.remove_post()
                    logger.info(f"Post {self.post_id} was deleted")
                else:
                    logger.info(f"Post {self.post_id} will not be deleted")
            else:
                logger.info(f"Post {self.post_id} did not have any ratings")
        else:
            logger.info(f"Post {self.post_id} was either a text post or deleted before analysis")

    async def detect_image(self):
        logger.info(f"Sending data to AWS for post {self.post_id}")
        client = boto3.client('rekognition', region_name=var.aws_region)
        self.ratings = classes.rating.create_ratings(client.detect_moderation_labels(Image={'Bytes': self.image}))

    async def check_ratings(self):
        logger.info(f"Checking ratings for post {self.post_id}")
        one_rating_was_false = False

        for post_rating in self.ratings:
            if post_rating.confidence > 90 and post_rating.enabled:
                logger.info(f"Post {self.post_id} contained enabled rating more than 90 for {post_rating.name} which "
                            f"was type of {post_rating.type}")
                self.post_will_be_deleted = True
            elif not post_rating.enabled:
                logger.info(f"Post {self.post_id} had a rating of {post_rating.name} in the "
                            f"{post_rating.type} category at confidence {post_rating.confidence}"
                            f"but it was not enabled. Post will not be deleted.")
                one_rating_was_false = True

        # if one_rating_was_false:
        #     self.post_will_be_deleted = False
        #     self.ratings = None

    async def remove_post(self):
        var.most_recently_deleted_post_id = self.post_id
        self.deletion_status = False
        if self.post_will_be_deleted:
            logger.info(f"Attempting to delete a post {self.post_id} for {await self.get_ratings_string()}")

            self.deletion_status = await delete_post(self.post_id)
            if self.deletion_status == 200:
                logger.info(f"Post {self.post_id} was deleted")
                self.deletion_status = True
            else:
                logger.info(f"An issue occurred deleting {self.post_id} for reason {self.deletion_status}")
                self.deletion_status = f"Problem occurred with deletion: {self.deletion_status}"

    async def get_readable_date(self):
        return datetime.datetime.utcfromtimestamp(self.date).strftime('%Y-%m-%d %H:%M:%S')

    async def get_censorship_string(self):
        return '", "'.join(self.infringing_words)

    async def get_ratings_string(self):
        ratings_string = ""
        for post_rating in self.ratings:
            ratings_string += str(post_rating)
        return ratings_string

    async def get_as_text_detection_text(self):
        string = (f'‼️ *DETECTED:TEXT:* "{await self.get_censorship_string()}"\n' +
                  f'🗒️ *TITLE:* "{self.post_name}"\n' +
                  f"🔗 *URL:* https://{var.site}/admin/posts/{self.post_id}\n" +
                  f"👤 *POSTER:* https://{var.site}/users/{self.user_id} / @{self.poster_username}\n")
        return string

    async def get_as_image_deletion_text(self):
        ratings_string = ""
        for post_rating in self.ratings:
            ratings_string += str(post_rating)

        if self.deletion_status is None:
            self.deletion_status = False

        string = (f"{ratings_string} \n " +
                  f'🗒️ *TITLE:* "{self.post_name}"\n' +
                  f"🔗 *URL:* https://{var.site}/admin/posts/{self.post_id}\n" +
                  f"👤 *POSTER:* https://{var.site}/users/{self.user_id} / @{self.poster_username}\n" +
                  f"❌ *DELETED:* {str(self.deletion_status)}"
                  f"The user was also suspended for a day, if the suspension was not warranted you must undo the suspension")
        return string