Migrate what we can to v2

pleroma_bot/_twitter.py

@@ -19,31 +19,58 @@ def _get_twitter_info(self):
     :return: None
     """
     for t_user in self.twitter_username:
-        # TODO: Migrate to v2
-        twitter_user_url = (
-            f"{self.twitter_base_url}"
-            f"/users/show.json?screen_name="
-            f"{t_user}"
+        url = f"{self.twitter_base_url_v2}/users/by/username/{t_user}"
+        params = {}
+        params.update(
+            {
+                "user.fields": "created_at,description,entities,id,location,"
+                               "name,pinned_tweet_id,profile_image_url,"
+                               "protected,url,username,verified,withheld",
+                "expansions": "pinned_tweet_id",
+                "tweet.fields": "attachments,author_id,"
+                                "context_annotations,conversation_id,"
+                                "created_at,entities,"
+                                "geo,id,in_reply_to_user_id,lang,"
+                                "public_metrics,"
+                                "possibly_sensitive,referenced_tweets,"
+                                "source,text,"
+                                "withheld",
+            }
         )
         response = requests.get(
-            twitter_user_url, headers=self.header_twitter, auth=self.auth
+            url, headers=self.header_twitter, auth=self.auth, params=params
         )
         if not response.ok:
             response.raise_for_status()
-        user = json.loads(response.text)
+        user = json.loads(response.text)["data"]
         self.bio_text[t_user] = (
             f"{self.bio_text['_generic_bio_text']}{user['description']}"
             if self.twitter_bio
             else f"{self.bio_text['_generic_bio_text']}"
         )
         # Check if user has profile image
-        if "profile_image_url_https" in user.keys():
-            self.profile_image_url[t_user] = user["profile_image_url_https"]
+        if "profile_image_url" in user.keys():
+            # Get the highest quality possible
+            profile_img_url = user["profile_image_url"].replace("_normal", "")
+            self.profile_image_url[t_user] = profile_img_url
+        self.display_name[t_user] = user["name"]
+        self.twitter_ids[user["id"]] = user["username"]
+        # TODO: Migrate to v2 when profile_banner is available users endpoint
+        twitter_user_url = (
+            f"{self.twitter_base_url}"
+            f"/users/show.json?screen_name="
+            f"{t_user}"
+        )
+        response = requests.get(
+            twitter_user_url, headers=self.header_twitter, auth=self.auth
+        )
+        if not response.ok:
+            response.raise_for_status()
+        user = json.loads(response.text)
         # Check if user has banner image
         if "profile_banner_url" in user.keys():
             base_banner_url = user["profile_banner_url"]
             self.profile_banner_url[t_user] = f"{base_banner_url}/1500x500"
-        self.display_name[t_user] = user["name"]
     return
 
 

pleroma_bot/_utils.py

@@ -5,6 +5,7 @@
 import string
 import random
 from multiprocessing import Queue
+from requests.structures import CaseInsensitiveDict
 
 import requests
 import threading
@@ -270,7 +271,7 @@ def replace_vars_in_str(self, text: str, var_name: str = None) -> str:
                 value = globals()[match.strip()]
         if isinstance(value, list):
             value = ", ".join([str(elem) for elem in value])
-        if isinstance(value, dict):
+        if isinstance(value, dict) or isinstance(value, CaseInsensitiveDict):
             if isinstance(self.twitter_username, list):
                 for t_user in self.twitter_username:
                     dict_value = value[t_user]

pleroma_bot/cli.py

@@ -38,6 +38,7 @@
 import multiprocessing as mp
 
 from requests_oauthlib import OAuth1
+from requests.structures import CaseInsensitiveDict
 
 from .i18n import _
 from . import logger
@@ -86,6 +87,7 @@ def __init__(self, user_cfg: dict, cfg: dict, base_path: str):
         self.profile_image_url = {}
         self.profile_banner_url = {}
         self.t_user_tweets = {}
+        self.twitter_ids = {}
         valid_visibility = ("public", "unlisted", "private", "direct")
         default_cfg_attributes = {
             "twitter_base_url": "https://api.twitter.com/1.1",
@@ -177,7 +179,7 @@ def __init__(self, user_cfg: dict, cfg: dict, base_path: str):
                 )
             )
 
-        self.twitter_url = {}
+        self.twitter_url = CaseInsensitiveDict()
         for t_user in t_users:
             self.twitter_url[t_user] = f"{twitter_url}/{t_user}"
         self.pinned_tweet_id = self._get_pinned_tweet_id()