diff --git a/test_vx.py b/test_vx.py index 0d24a45..e0aaaaf 100644 --- a/test_vx.py +++ b/test_vx.py @@ -106,12 +106,11 @@ def test_pollTweetExtract(): tweet = twExtract.extractStatus("https://twitter.com/norm/status/651169346518056960") assert 'card' in tweet compareDict(testPoll_comparePoll,tweet['card']) -''' + def test_NSFW_TweetExtract(): tweet = twExtract.extractStatus(testNSFWTweet) # For now just test that there's no error -''' # this test currently fails due to new Twitter API restrictions -## VNF conversion test ## +## VNF conversion test ## def test_textTweetVNF(): vnf = twitfix.link_to_vnf_from_unofficial_api(testTextTweet) compareDict(textVNF_compare,vnf) @@ -194,12 +193,12 @@ def test_embedFromCache(): assert resp.status_code==200 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) -''' # this test currently fails due to new Twitter API restrictions + def test_veryLongEmbed(): cache.clearCache() cache.setCache({'https://twitter.com/TEST/status/1234': diff --git a/twExtract/__init__.py b/twExtract/__init__.py index b675d9a..6002229 100644 --- a/twExtract/__init__.py +++ b/twExtract/__init__.py @@ -77,19 +77,25 @@ def extractUser(url): else: useId=False screen_name = m.group(1) - # get guest token - guestToken = getGuestToken() # get user - if not useId: - user = requests.get(f"https://api.twitter.com/1.1/users/show.json?screen_name={screen_name}",headers={"Authorization":bearer, "x-guest-token":guestToken}) - else: - user = requests.get(f"https://api.twitter.com/1.1/users/show.json?user_id={screen_name}",headers={"Authorization":bearer, "x-guest-token":guestToken}) - output = user.json() - if "errors" in output: - # pick the first error and create a twExtractError - error = output["errors"][0] - raise twExtractError.TwExtractError(error["code"], error["message"]) - return output + tokens = config["config"]["workaroundTokens"].split(",") + for authToken in tokens: + try: + csrfToken=str(uuid.uuid4()).replace('-', '') + reqHeaders = {"Authorization":bearer,"Cookie":f"auth_token={authToken}; ct0={csrfToken}; ","x-twitter-active-user":"yes","x-twitter-auth-type":"OAuth2Session","x-twitter-client-language":"en","x-csrf-token":csrfToken,"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/116.0"} + if not useId: + user = requests.get(f"https://api.twitter.com/1.1/users/show.json?screen_name={screen_name}",headers=reqHeaders) + else: + user = requests.get(f"https://api.twitter.com/1.1/users/show.json?user_id={screen_name}",headers=reqHeaders) + output = user.json() + if "errors" in output: + # pick the first error and create a twExtractError + error = output["errors"][0] + raise twExtractError.TwExtractError(error["code"], error["message"]) + return output + except Exception as e: + continue + raise twExtractError.TwExtractError(400, "Extract error") #def extractUserByID(id):