Skip to content

Commit

Permalink
update
Browse files Browse the repository at this point in the history 
Signed-off-by: ふぁ <[email protected]>
  • Loading branch information
@fa0311
fa0311 committed Feb 3, 2023
1 parent 40df929 commit 8cacf89
Show file tree
Hide file tree
Showing 2 changed files with 91 additions and 29 deletions.
80 changes: 51 additions & 29 deletions TwitterFrontendFlow/TwitterFrontendFlow.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@


class TwitterFrontendFlow:
def __init__(self, proxies={}):
def __init__(self, proxies={}, language="en"):
self.USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36"
self.AUTHORIZATION = "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
self.proxies = proxies
Expand All @@ -12,6 +12,7 @@ def __init__(self, proxies={}):
self.x_guest_token = self.__get_guest_token()
self.method_check_bypass = False
self.flow_token = None
self.language = language

def __twitter(self):
headers = {
Expand Down Expand Up @@ -42,7 +43,7 @@ def __get_headers(self):
"x-guest-token": self.x_guest_token,
"x-csrf-token": self.session.cookies.get("ct0"),
"x-twitter-active-user": "yes",
"x-twitter-client-language": "ja",
"x-twitter-client-language": self.language,
}

def __get_headers_legacy(self):
Expand All @@ -62,9 +63,9 @@ def __flow_token_check(self):
if self.flow_token == None:
raise Exception("not found token")

def __error_check(self):
if self.content.get("errors"):
raise Exception(self.content["errors"][0]["message"])
def __error_check(self, content):
if content.get("errors"):
raise Exception(content["errors"][0]["message"])

def __method_check(self, method_name):
if self.method_check_bypass:
Expand Down Expand Up @@ -134,9 +135,9 @@ def login_flow(self):
params=params,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def LoginJsInstrumentationSubtask(self):
Expand Down Expand Up @@ -170,9 +171,9 @@ def LoginJsInstrumentationSubtask(self):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def LoginEnterUserIdentifierSSO(self, user_id):
Expand Down Expand Up @@ -201,9 +202,9 @@ def LoginEnterUserIdentifierSSO(self, user_id):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def AccountDuplicationCheck(self):
Expand All @@ -226,9 +227,9 @@ def AccountDuplicationCheck(self):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def LoginEnterAlternateIdentifierSubtask(self, text):
Expand All @@ -249,9 +250,9 @@ def LoginEnterAlternateIdentifierSubtask(self, text):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def LoginEnterPassword(self, password):
Expand All @@ -272,9 +273,9 @@ def LoginEnterPassword(self, password):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def LoginTwoFactorAuthChallenge(self, TwoFactorCode):
Expand All @@ -295,9 +296,32 @@ def LoginTwoFactorAuthChallenge(self, TwoFactorCode):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
return self

def LoginAcid(self, acid):
self.__flow_token_check()
self.__method_check("LoginAcid")
data = {
"flow_token": self.flow_token,
"subtask_inputs": [
{
"subtask_id": "LoginAcid",
"enter_text": {"text": acid, "link": "next_link"},
}
],
}
response = self.session.post(
"https://twitter.com/i/api/1.1/onboarding/task.json",
headers=self.__get_headers(),
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

# attの取得 無くても動くっぽい
Expand All @@ -310,8 +334,8 @@ def get_att(self):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self

# ct0の更新 無くても動くっぽい
Expand All @@ -331,9 +355,8 @@ def Viewer(self):
headers=self.__get_headers(),
params=params,
)

self.__error_check(response)
self.content = response
self.__error_check()
return self

def RedirectToPasswordReset(self):
Expand Down Expand Up @@ -367,9 +390,9 @@ def password_reset_flow(self):
params=params,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PwrJsInstrumentationSubtask(self):
Expand Down Expand Up @@ -403,9 +426,9 @@ def PwrJsInstrumentationSubtask(self):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PasswordResetBegin(self, user_id):
Expand All @@ -426,9 +449,9 @@ def PasswordResetBegin(self, user_id):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PasswordResetChooseChallenge(self, choices="0"):
Expand All @@ -452,9 +475,9 @@ def PasswordResetChooseChallenge(self, choices="0"):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PwrKnowledgeChallenge(self, text):
Expand All @@ -475,9 +498,9 @@ def PwrKnowledgeChallenge(self, text):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PasswordResetConfirmChallenge(self, code):
Expand All @@ -498,9 +521,9 @@ def PasswordResetConfirmChallenge(self, code):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PasswordResetNewPassword(self, password):
Expand All @@ -521,9 +544,9 @@ def PasswordResetNewPassword(self, password):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

def PasswordResetSurvey(self, choices="0"):
Expand All @@ -547,9 +570,9 @@ def PasswordResetSurvey(self, choices="0"):
json=data,
proxies=self.proxies,
).json()
self.__error_check(response)
self.flow_token = response.get("flow_token")
self.content = response
self.__error_check()
return self

# ログイン後
Expand Down Expand Up @@ -577,8 +600,8 @@ def CreateTweet(self, tweet_text):
headers=self.__get_headers(),
json=data,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self

def FavoriteTweet(self, tweet_id):
Expand All @@ -595,8 +618,8 @@ def FavoriteTweet(self, tweet_id):
headers=self.__get_headers(),
json=data,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self

def UnfavoriteTweet(self, tweet_id):
Expand All @@ -613,8 +636,8 @@ def UnfavoriteTweet(self, tweet_id):
headers=self.__get_headers(),
json=data,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self

def CreateRetweet(self, tweet_id):
Expand All @@ -632,8 +655,8 @@ def CreateRetweet(self, tweet_id):
headers=self.__get_headers(),
json=data,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self

def DeleteRetweet(self, tweet_id):
Expand All @@ -651,11 +674,10 @@ def DeleteRetweet(self, tweet_id):
headers=self.__get_headers(),
json=data,
).json()
self.__error_check(response)
self.content = response
self.__error_check()
return self


# Legacy API v1.1

def friendships_create(self, tweet_id):
Expand Down
40 changes: 40 additions & 0 deletions sample2.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
from TwitterFrontendFlow.TwitterFrontendFlow.TwitterFrontendFlow import *

flow = TwitterFrontendFlow()

flow.login_flow()
flow.LoginJsInstrumentationSubtask()

while "LoginSuccessSubtask" not in flow.get_subtask_ids():
try:
if "LoginEnterUserIdentifierSSO" in flow.get_subtask_ids():
print("Telephone number / Email address / User name")
flow.LoginEnterUserIdentifierSSO(input())
elif "LoginEnterAlternateIdentifierSubtask" in flow.get_subtask_ids():
print(flow.content["subtasks"][0]["enter_text"]["primary_text"]["text"])
flow.LoginEnterAlternateIdentifierSubtask(input())
elif "LoginEnterPassword" in flow.get_subtask_ids():
print(flow.content["subtasks"][0]["enter_password"]["primary_text"]["text"])
flow.LoginEnterPassword(input())
elif "AccountDuplicationCheck" in flow.get_subtask_ids():
print("AccountDuplicationCheck")
flow.AccountDuplicationCheck()
elif "LoginTwoFactorAuthChallenge" in flow.get_subtask_ids():
header = flow.content["subtasks"][0]["enter_text"]["header"]
print(header["primary_text"]["text"])
flow.LoginTwoFactorAuthChallenge(input())
elif "LoginAcid" in flow.get_subtask_ids():
header = flow.content["subtasks"][0]["enter_text"]["header"]
print(header["secondary_text"]["text"])
flow.LoginAcid(input())
elif "SuccessExit" in flow.get_subtask_ids():
break
else:
print("Non-supported login methods: " + flow.get_subtask_ids())
exit(1)

except:
print("Error")

print("Success")
flow.SaveCookies("cookie.json")

0 comments on commit 8cacf89

Please sign in to comment.