mirror of
https://github.com/trevorhobenshield/twitter-api-client.git
synced 2025-12-25 02:03:15 -05:00
temporarily remove protonmail automation
This commit is contained in:
@@ -2,6 +2,7 @@ import hashlib
|
||||
import logging.config
|
||||
import math
|
||||
import mimetypes
|
||||
import platform
|
||||
import random
|
||||
from copy import deepcopy
|
||||
from datetime import datetime
|
||||
@@ -14,6 +15,22 @@ from .constants import *
|
||||
from .login import login
|
||||
from .util import *
|
||||
|
||||
try:
|
||||
if get_ipython().__class__.__name__ == 'ZMQInteractiveShell':
|
||||
import nest_asyncio
|
||||
|
||||
nest_asyncio.apply()
|
||||
except:
|
||||
...
|
||||
|
||||
if platform.system() != 'Windows':
|
||||
try:
|
||||
import uvloop
|
||||
|
||||
uvloop.install()
|
||||
except ImportError as e:
|
||||
...
|
||||
|
||||
|
||||
class Account:
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ import time
|
||||
|
||||
from httpx import Client
|
||||
from .constants import GREEN, YELLOW, RED, BOLD, RESET
|
||||
from .util import find_key, get_confirmation_code, get_inbox, init_protonmail_session
|
||||
from .util import find_key # ,get_confirmation_code, get_inbox, init_protonmail_session
|
||||
|
||||
|
||||
def update_token(client: Client, key: str, url: str, **kwargs) -> Client:
|
||||
@@ -109,23 +109,23 @@ def confirm_email(client: Client) -> Client:
|
||||
})
|
||||
|
||||
|
||||
def solve_confirmation_challenge(client: Client, email: str, password: str) -> Client:
|
||||
proton_session = init_protonmail_session(email, password)
|
||||
inbox = get_inbox(proton_session)
|
||||
confirmation_code = get_confirmation_code(inbox)
|
||||
print(f'{confirmation_code = }')
|
||||
return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
|
||||
"flow_token": client.cookies.get('flow_token'),
|
||||
'subtask_inputs': [
|
||||
{
|
||||
'subtask_id': 'LoginAcid',
|
||||
'enter_text': {
|
||||
'text': confirmation_code,
|
||||
'link': 'next_link',
|
||||
},
|
||||
},
|
||||
],
|
||||
})
|
||||
# def solve_confirmation_challenge(client: Client, email: str, password: str) -> Client:
|
||||
# proton_session = init_protonmail_session(email, password)
|
||||
# inbox = get_inbox(proton_session)
|
||||
# confirmation_code = get_confirmation_code(inbox)
|
||||
# print(f'{confirmation_code = }')
|
||||
# return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
|
||||
# "flow_token": client.cookies.get('flow_token'),
|
||||
# 'subtask_inputs': [
|
||||
# {
|
||||
# 'subtask_id': 'LoginAcid',
|
||||
# 'enter_text': {
|
||||
# 'text': confirmation_code,
|
||||
# 'link': 'next_link',
|
||||
# },
|
||||
# },
|
||||
# ],
|
||||
# })
|
||||
|
||||
|
||||
def execute_login_flow(client: Client) -> Client | None:
|
||||
@@ -137,15 +137,16 @@ def execute_login_flow(client: Client) -> Client | None:
|
||||
if client.cookies.get('confirm_email') == 'true':
|
||||
client = confirm_email(client)
|
||||
|
||||
# solve confirmation challenge (Proton Mail only)
|
||||
if client.cookies.get('confirmation_code') == 'true':
|
||||
if not client.protonmail:
|
||||
print(f'[{RED}warning{RESET}] Please check your email for a confirmation code'
|
||||
f' and log in again using the web app. If you wish to automatically solve'
|
||||
f' email confirmation challenges, add a Proton Mail account in your account settings')
|
||||
return
|
||||
time.sleep(10) # todo: just poll the inbox until it arrives instead of waiting
|
||||
client = solve_confirmation_challenge(client, *client.protonmail.values())
|
||||
# # solve confirmation challenge (Proton Mail only)
|
||||
# if client.cookies.get('confirmation_code') == 'true':
|
||||
# if not client.protonmail:
|
||||
# print(f'[{RED}warning{RESET}] Please check your email for a confirmation code'
|
||||
# f' and log in again using the web app. If you wish to automatically solve'
|
||||
# f' email confirmation challenges, add a Proton Mail account in your account settings')
|
||||
# return
|
||||
# time.sleep(10) # todo: just poll the inbox until it arrives instead of waiting
|
||||
# client = solve_confirmation_challenge(client, *client.protonmail.values())
|
||||
|
||||
return client
|
||||
|
||||
|
||||
@@ -165,7 +166,9 @@ def login(email: str, username: str, password: str, **kwargs) -> Client:
|
||||
'x-twitter-active-user': 'yes',
|
||||
'x-twitter-client-language': 'en',
|
||||
})
|
||||
client.protonmail = kwargs.get('protonmail')
|
||||
|
||||
# client.protonmail = kwargs.get('protonmail')
|
||||
|
||||
client = execute_login_flow(client)
|
||||
if kwargs.get('debug', True):
|
||||
if not client or client.cookies.get('flow_errors') == 'true':
|
||||
|
||||
162
twitter/util.py
162
twitter/util.py
@@ -5,7 +5,7 @@ from pathlib import Path
|
||||
from urllib.parse import urlsplit, urlencode, urlunsplit, parse_qs, quote
|
||||
|
||||
import orjson
|
||||
import protonmail
|
||||
# import protonmail
|
||||
from httpx import Response, Client
|
||||
|
||||
from .constants import GREEN, MAGENTA, RED, RESET
|
||||
@@ -150,86 +150,86 @@ def find_key(obj: any, key: str) -> list:
|
||||
return helper(obj, key, [])
|
||||
|
||||
|
||||
def init_protonmail_session(email: str, password: str) -> protonmail.api.Session:
|
||||
"""
|
||||
Create an authenticated Proton Mail session
|
||||
|
||||
@param email: your email. Can also use username
|
||||
@param password: your password
|
||||
@return: Proton Mail Session object
|
||||
"""
|
||||
cwd = Path.cwd()
|
||||
log_dir_path = cwd / 'protonmail_log'
|
||||
cache_dir_path = cwd / 'protonmail_cache'
|
||||
try:
|
||||
session = protonmail.api.Session(
|
||||
api_url="https://api.protonmail.ch",
|
||||
log_dir_path=log_dir_path,
|
||||
cache_dir_path=cache_dir_path,
|
||||
user_agent="Ubuntu_20.04",
|
||||
tls_pinning=False,
|
||||
)
|
||||
session.enable_alternative_routing = False
|
||||
session.authenticate(email, password)
|
||||
return session
|
||||
except Exception as e:
|
||||
print('Failed to initialize Proton Mail Session:', e)
|
||||
|
||||
|
||||
def get_inbox(session: protonmail.api.Session) -> dict:
|
||||
"""
|
||||
Get inbox
|
||||
|
||||
@param session: Proton Mail Session object
|
||||
@return: inbox data
|
||||
"""
|
||||
try:
|
||||
return session.api_request(
|
||||
"/api/mail/v4/conversations",
|
||||
method="GET",
|
||||
params={
|
||||
'Page': 0,
|
||||
'PageSize': 50,
|
||||
'Limit': 100,
|
||||
'LabelID': 0,
|
||||
'Sort': 'Time',
|
||||
'Desc': 1,
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
print('Failed to get inbox:', e)
|
||||
|
||||
|
||||
def get_verification_code(inbox: dict) -> str:
|
||||
"""
|
||||
Get Twitter verification code from inbox.
|
||||
|
||||
Crude implementation. Subject line contains verification code, no need to decrypt message body.
|
||||
|
||||
@param inbox: inbox data
|
||||
@return: Twitter verification code
|
||||
"""
|
||||
try:
|
||||
expr = '(\w+) is your Twitter verification code'
|
||||
return list(filter(len, (re.findall(expr, conv['Subject']) for conv in inbox['Conversations'])))[0][0]
|
||||
except Exception as e:
|
||||
print('Failed to get Twitter verification code:', e)
|
||||
|
||||
|
||||
def get_confirmation_code(inbox: dict) -> str:
|
||||
"""
|
||||
Get Twitter confirmation code from inbox.
|
||||
|
||||
Crude implementation. Subject line contains confirmation code, no need to decrypt message body.
|
||||
|
||||
@param inbox: inbox data
|
||||
@return: Twitter confirmation code
|
||||
"""
|
||||
try:
|
||||
expr = 'Your Twitter confirmation code is (\w+)'
|
||||
return list(filter(len, (re.findall(expr, conv['Subject']) for conv in inbox['Conversations'])))[0][0]
|
||||
except Exception as e:
|
||||
print('Failed to get Twitter confirmation code:', e)
|
||||
# def init_protonmail_session(email: str, password: str) -> protonmail.api.Session:
|
||||
# """
|
||||
# Create an authenticated Proton Mail session
|
||||
#
|
||||
# @param email: your email. Can also use username
|
||||
# @param password: your password
|
||||
# @return: Proton Mail Session object
|
||||
# """
|
||||
# cwd = Path.cwd()
|
||||
# log_dir_path = cwd / 'protonmail_log'
|
||||
# cache_dir_path = cwd / 'protonmail_cache'
|
||||
# try:
|
||||
# session = protonmail.api.Session(
|
||||
# api_url="https://api.protonmail.ch",
|
||||
# log_dir_path=log_dir_path,
|
||||
# cache_dir_path=cache_dir_path,
|
||||
# user_agent="Ubuntu_20.04",
|
||||
# tls_pinning=False,
|
||||
# )
|
||||
# session.enable_alternative_routing = False
|
||||
# session.authenticate(email, password)
|
||||
# return session
|
||||
# except Exception as e:
|
||||
# print('Failed to initialize Proton Mail Session:', e)
|
||||
#
|
||||
#
|
||||
# def get_inbox(session: protonmail.api.Session) -> dict:
|
||||
# """
|
||||
# Get inbox
|
||||
#
|
||||
# @param session: Proton Mail Session object
|
||||
# @return: inbox data
|
||||
# """
|
||||
# try:
|
||||
# return session.api_request(
|
||||
# "/api/mail/v4/conversations",
|
||||
# method="GET",
|
||||
# params={
|
||||
# 'Page': 0,
|
||||
# 'PageSize': 50,
|
||||
# 'Limit': 100,
|
||||
# 'LabelID': 0,
|
||||
# 'Sort': 'Time',
|
||||
# 'Desc': 1,
|
||||
# }
|
||||
# )
|
||||
# except Exception as e:
|
||||
# print('Failed to get inbox:', e)
|
||||
#
|
||||
#
|
||||
# def get_verification_code(inbox: dict) -> str:
|
||||
# """
|
||||
# Get Twitter verification code from inbox.
|
||||
#
|
||||
# Crude implementation. Subject line contains verification code, no need to decrypt message body.
|
||||
#
|
||||
# @param inbox: inbox data
|
||||
# @return: Twitter verification code
|
||||
# """
|
||||
# try:
|
||||
# expr = '(\w+) is your Twitter verification code'
|
||||
# return list(filter(len, (re.findall(expr, conv['Subject']) for conv in inbox['Conversations'])))[0][0]
|
||||
# except Exception as e:
|
||||
# print('Failed to get Twitter verification code:', e)
|
||||
#
|
||||
#
|
||||
# def get_confirmation_code(inbox: dict) -> str:
|
||||
# """
|
||||
# Get Twitter confirmation code from inbox.
|
||||
#
|
||||
# Crude implementation. Subject line contains confirmation code, no need to decrypt message body.
|
||||
#
|
||||
# @param inbox: inbox data
|
||||
# @return: Twitter confirmation code
|
||||
# """
|
||||
# try:
|
||||
# expr = 'Your Twitter confirmation code is (\w+)'
|
||||
# return list(filter(len, (re.findall(expr, conv['Subject']) for conv in inbox['Conversations'])))[0][0]
|
||||
# except Exception as e:
|
||||
# print('Failed to get Twitter confirmation code:', e)
|
||||
|
||||
|
||||
def log(logger: Logger, level: int, r: Response):
|
||||
|
||||
Reference in New Issue
Block a user