Workaround for Twitter API changes

configHandler.py

@@ -14,7 +14,13 @@ elif ('RUNNING_SERVERLESS' in os.environ and os.environ['RUNNING_SERVERLESS'] ==
                 "repo": os.environ["VXTWITTER_REPO"], 
                 "url": os.environ["VXTWITTER_URL"],
                 "combination_method": os.environ["VXTWITTER_COMBINATION_METHOD"], # can either be 'local' or a URL to a server handling requests in the same format
-                "gifConvertAPI":os.environ["VXTWITTER_GIF_CONVERT_API"]
+                "gifConvertAPI":os.environ["VXTWITTER_GIF_CONVERT_API"],
+                "workaroundKeys":{
+                    "consumerKey":os.environ["VXTWITTER_WORKAROUND_CONSUMER_KEY"],
+                    "consumerSecret":os.environ["VXTWITTER_WORKAROUND_CONSUMER_SECRET"],
+                    "accessToken":os.environ["VXTWITTER_WORKAROUND_TOKEN"],
+                    "accessTokenSecret":os.environ["VXTWITTER_WORKAROUND_TOKEN_SECRET"]
+                }
             }
         }
 else:
@@ -31,7 +37,8 @@ else:
                     "repo": "https://github.com/dylanpdx/BetterTwitFix", 
                     "url": "https://vxtwitter.com",
                     "combination_method": "local", # can either be 'local' or a URL to a server handling requests in the same format
-                    "gifConvertAPI":""
+                    "gifConvertAPI":"",
+                    "workaroundKeys":{"consumerKey":"","consumerSecret":"","accessToken":"","accessTokenSecret":""}
                 }
             }
 

requirements.txt

@@ -1,7 +1,8 @@
-pymongo==4.3.3
+pymongo==4.3.3
 boto3==1.26.104
 requests==2.28.2
 Pillow==9.4.0
 Flask==2.2.3
 Flask-Cors==3.0.10
 yt-dlp==2022.7.18
+twitter==1.19.6

serverless.yml

@@ -27,6 +27,10 @@ provider:
     VXTWITTER_URL: ${env:VXTWITTER_URL, 'https://vxtwitter.com'}
     VXTWITTER_COMBINATION_METHOD: ${env:VXTWITTER_COMBINATION_METHOD, 'local'}
     VXTWITTER_GIF_CONVERT_API: ${env:VXTWITTER_GIF_CONVERT_API, ''}
+    VXTWITTER_WORKAROUND_CONSUMER_KEY: ${env:VXTWITTER_WORKAROUND_CONSUMER_KEY, ''}
+    VXTWITTER_WORKAROUND_CONSUMER_SECRET: ${env:VXTWITTER_WORKAROUND_CONSUMER_SECRET, ''}
+    VXTWITTER_WORKAROUND_TOKEN: ${env:VXTWITTER_WORKAROUND_TOKEN, ''}
+    VXTWITTER_WORKAROUND_TOKEN_SECRET: ${env:VXTWITTER_WORKAROUND_TOKEN_SECRET, ''}
 
 package:
   patterns:

test_vx.py

@@ -194,6 +194,11 @@ def test_embedFromCache():
     resp = client.get(testMultiMediaTweet.replace("https://twitter.com",""),headers={"User-Agent":"test"})
     assert resp.status_code==200
 
+def test_embedSuggestive():
+    resp = client.get(testNSFWTweet.replace("https://twitter.com",""),headers={"User-Agent":"test"})
+    assert resp.status_code==200
+    assert "so i had a bot generate it for me" in str(resp.data)
+
 def test_veryLongEmbed():
     cache.clearCache()
     cache.setCache({'https://twitter.com/TEST/status/1234':

twExtract/__init__.py

@@ -4,11 +4,24 @@ import json
 import requests
 import re
 from . import twExtractError
+import twitter
+from configHandler import config
 bearer="Bearer AAAAAAAAAAAAAAAAAAAAAPYXBAAAAAAACLXUNDekMxqa8h%2F40K4moUkGsoc%3DTYfbDKbT3jJPCEVnMYqilB28NHfOPqkca3qaAxGfsyKCs0wRbw"
 guestToken=None
 pathregex = r"\w{1,15}\/(status|statuses)\/(\d{2,20})"
 userregex = r"^https?:\/\/(?:www\.)?twitter\.com\/(?:#!\/)?@?([^/?#]*)(?:[?#/].*)?$"
 userIDregex = r"\/i\/user\/(\d+)"
+try:
+    auth = twitter.oauth.OAuth(
+        config['config']['workaroundKeys']["accessToken"],
+        config['config']['workaroundKeys']["accessTokenSecret"],
+        config['config']['workaroundKeys']["consumerKey"],
+        config['config']['workaroundKeys']["consumerSecret"]
+    )
+    api = twitter.Twitter(auth=auth)
+except Exception as e:
+    api = None
+
 def getGuestToken():
     global guestToken
     if guestToken is None:
@@ -17,18 +30,17 @@ def getGuestToken():
     return guestToken
 
 def extractStatus_fallback(url):
-    twIE = twitter.TwitterIE()
-    twIE.set_downloader(yt_dlp.YoutubeDL())
-    twid = twIE._match_id(url)
-    status = twIE._call_api(
-    'statuses/show/%s.json' % twid, twid, {
-        'cards_platform': 'Web-12',
-        'include_cards': 1,
-        'include_reply_count': 1,
-        'include_user_entities': 0,
-        'tweet_mode': 'extended',
-    })
-    return status
+    if api is None:
+        raise twExtractError.TwExtractError(500, "Could not extract tweet.")
+    print(" ➤ [ I ] Using fallback method to extract tweet")
+    # get tweet ID
+    m = re.search(pathregex, url)
+    if m is None:
+        raise twExtractError.TwExtractError(400, "Invalid URL")
+    twid = m.group(2)
+    # get tweet
+    tweet = api.statuses.show(_id=twid, tweet_mode="extended")
+    return tweet
 
 
 def extractStatus(url):