services.py

import logging
from datetime import timedelta
from typing import TYPE_CHECKING

from django.utils import timezone
from tweepy import Client
from tweepy.errors import BadRequest

from main.models import MainHomePage

from .models import Tweet

if TYPE_CHECKING:
    from tweepy import Tweet as TweetResponse
    from tweepy import User


logger = logging.getLogger()


class TweetDownloadService:
    """
    Service class starající se o update tweetů z Twitter API, v současné chvíli
    bere tweety z účtu nastavených v (první) MainHomePage stránce (HP pirati.cz).
    """

    client: Client
    days_back: int

    def __init__(self, bearer_token, days_back=1):
        if not bearer_token:
            raise RuntimeError("Twitter bearer token not set, cannot update tweets")

        self.client = Client(bearer_token=bearer_token)
        self.days_back = days_back

    @staticmethod
    def get_latest_saved_tweet_id() -> list[int]:
        """
        Vrací IDs už uložených Tweetů - možná by stálo za to brát jen z určitého
        časového období...
        """
        return Tweet.objects.values_list("twitter_id", flat=True)

    def get_tweets_response(self, user_id) -> list["TweetResponse"]:
        """
        Vrací list tweetů (objektů) pro daného Twitter uživatele.
        """
        tweets_response = self.client.get_users_tweets(
            user_id,
            expansions=["author_id", "entities.mentions.username"],
            max_results=100,
            start_time=timezone.now() - timedelta(days=self.days_back),
            tweet_fields=["author_id", "created_at"],
            user_fields=["name", "username"],
        )  # 49022430

        return tweets_response.data

    def get_user_list_data(self) -> list["User"]:
        twitter_usernames_block = MainHomePage.objects.first().twitter_usernames
        user_data_list = []

        for username_data in twitter_usernames_block.raw_data:
            try:
                user_data_list.append(self.get_user_response(username_data["value"]))
            except BadRequest:
                logger.error(
                    "Cannot download tweets for the username",
                    extra={"username": username_data["value"]},
                )

        return user_data_list

    def get_user_response(self, username) -> "User":
        """
        Vrací informace o daném uživateli.
        """
        user_response = self.client.get_user(
            username=username,
            user_fields=["profile_image_url"],  # id, name, username enabled by default
        )

        return user_response.data

    def perform_update(self) -> int:
        """
        Obaluje celý proces downloadu Tweetů z API do DB.
        """
        user_data_list = self.get_user_list_data()
        existing_tweet_id_list = self.get_latest_saved_tweet_id()

        tweets_to_save = []

        for user_data in user_data_list:
            for tweet in self.get_tweets_response(user_id=user_data.id):
                if str(tweet.id) not in existing_tweet_id_list:
                    tweets_to_save.append(
                        Tweet(
                            author_img_url=user_data.profile_image_url,
                            author_name=user_data.name,
                            author_username=user_data.username,
                            text=tweet.text,
                            twitter_id=tweet.id,
                        )
                    )

        return Tweet.objects.bulk_create(tweets_to_save)