diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f313b63 --- /dev/null +++ b/.gitignore @@ -0,0 +1,161 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +.idea/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9997bde --- /dev/null +++ b/Dockerfile @@ -0,0 +1,17 @@ +FROM python:3.11 +LABEL authors="Jammer" + +WORKDIR /usr/src/app + +ENV PYTHONDONTWRITEBYTECODE 1 +ENV PYTHONUNBUFFERED 1 + +RUN apt -qq update +COPY requirements.txt . + +RUN pip install --upgrade pip +COPY requirements.txt . +RUN pip install -r requirements.txt + +COPY ../.. . +CMD ["python", "./run.py"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..9844c08 --- /dev/null +++ b/README.md @@ -0,0 +1,60 @@ + +# MintChain Daily Bot + +## 🔗 Links + +🔔 CHANNEL: https://t.me/JamBitPY + +💬 CHAT: https://t.me/JamBitChat + +💰 DONATION EVM ADDRESS: 0x08e3fdbb830ee591c0533C5E58f937D312b07198 + + +## 🤖 | Features: + +- **Auto registration** +- **Auto bind referral** +- **Auto bind twitter** +- **Auto collect daily rewards every X time** + + +## 🚀 Installation + +``Docker`` + + +``1. Close the repo and open CMD (console) inside it`` + +``2. Setup configuration and accounts`` + +``3. Run: docker-compose up -d --build`` + +``OR`` + + +`` Required python >= 3.10`` + +``1. Close the repo and open CMD (console) inside it`` + +``2. Install requirements: pip install -r requirements.txt`` + +``3. Setup configuration and accounts`` + +``4. Run: python run.py`` + + +## ⚙️ Config (config > settings.yaml) + +| Name | Description | +| --- |----------------------------------------------------------------------------------------------------| +| referral_code | Your referral code | +| rpc_url | RPC URL (if not have, leave the default value) | +| iteration_delay | Delay between iterations in hours (Let's say every 24 hours the script will collect daily rewards) | + + +## ⚙️ Accounts format (config > accounts.txt) + +- twitter_auth_token|wallet_mnemonic|proxy +- twitter_auth_token|wallet_mnemonic + +`` Proxy format: IP:PORT:USER:PASS`` diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 0000000..8476494 --- /dev/null +++ b/config/__init__.py @@ -0,0 +1 @@ +from .load_config import load_config diff --git a/config/accounts.txt b/config/accounts.txt new file mode 100644 index 0000000..ad42417 --- /dev/null +++ b/config/accounts.txt @@ -0,0 +1,7 @@ +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy +auth_token|mnemonic|proxy \ No newline at end of file diff --git a/config/load_config.py b/config/load_config.py new file mode 100644 index 0000000..6b6abf2 --- /dev/null +++ b/config/load_config.py @@ -0,0 +1,66 @@ +import os +import yaml + +from loguru import logger +from models import Account, Config + + +def get_accounts() -> Account: + accounts_path = os.path.join(os.path.dirname(__file__), "accounts.txt") + if not os.path.exists(accounts_path): + logger.error(f"File <<{accounts_path}>> does not exist") + exit(1) + + with open(accounts_path, "r") as f: + accounts = f.readlines() + + if not accounts: + logger.error(f"File <<{accounts_path}>> is empty") + exit(1) + + for account in accounts: + values = account.split("|") + if len(values) == 2: + yield Account(auth_token=values[0].strip(), mnemonic=values[1].strip()) + + elif len(values) == 3: + yield Account( + auth_token=values[0].strip(), + mnemonic=values[1].strip(), + proxy=values[2].strip(), + ) + + else: + logger.error( + f"Account <<{account}>> is not in correct format | Need to be in format: <>" + ) + exit(1) + + +def load_config() -> Config: + settings_path = os.path.join(os.path.dirname(__file__), "settings.yaml") + if not os.path.exists(settings_path): + logger.error(f"File <<{settings_path}>> does not exist") + exit(1) + + with open(settings_path, "r") as f: + settings = yaml.safe_load(f) + + if not settings.get("referral_code"): + logger.error(f"Referral code is not provided in settings.yaml") + exit(1) + + if not settings.get("rpc_url"): + logger.error(f"RPC URL is not provided in settings.yaml") + exit(1) + + if not settings.get("iteration_delay"): + logger.error(f"Iteration delay is not provided in settings.yaml") + exit(1) + + return Config( + accounts=list(get_accounts()), + referral_code=settings["referral_code"], + rpc_url=settings["rpc_url"], + iteration_delay=settings["iteration_delay"], + ) diff --git a/config/settings.yaml b/config/settings.yaml new file mode 100644 index 0000000..3c87e5e --- /dev/null +++ b/config/settings.yaml @@ -0,0 +1,3 @@ +referral_code: C4ACD869 # Referral code (If you don't have one, pls, use mine) +rpc_url: https://eth.llamarpc.com # RPC URL (Ethereum) +iteration_delay: 6 # hours \ No newline at end of file diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..de1760f --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,10 @@ +services: + app: + build: ./ + container_name: MintChainBot + deploy: + restart_policy: + condition: on-failure + delay: 3s + max_attempts: 5 + window: 60s \ No newline at end of file diff --git a/loader.py b/loader.py new file mode 100644 index 0000000..0931d9e --- /dev/null +++ b/loader.py @@ -0,0 +1,4 @@ +from models import Config +from config import load_config + +config: Config = load_config() diff --git a/models/__init__.py b/models/__init__.py new file mode 100644 index 0000000..2baa3f6 --- /dev/null +++ b/models/__init__.py @@ -0,0 +1,4 @@ +from .api import * +from .wallet import * +from .account import * +from .config import * diff --git a/models/account.py b/models/account.py new file mode 100644 index 0000000..41c717c --- /dev/null +++ b/models/account.py @@ -0,0 +1,34 @@ +from loguru import logger +from pydantic import BaseModel, field_validator + + +class Account(BaseModel): + auth_token: str + mnemonic: str + proxy: str = None + + @field_validator("mnemonic", mode="before") + def check_mnemonic(cls, value) -> str | None: + words = value.split(" ") + if len(words) not in (12, 24): + logger.error( + f"Mnemonic <<{value}>> is not in correct format | Need to be 12/24 words" + ) + exit(1) + + return value + + @field_validator("proxy", mode="before") + def check_proxy(cls, value) -> str | None: + if not value: + return None + + proxy_values = value.split(":") + if len(proxy_values) != 4: + logger.error( + f"Proxy <<{value}>> is not in correct format | Need to be in format: <>" + ) + exit(1) + + proxy_url = f"http://{proxy_values[2]}:{proxy_values[3]}@{proxy_values[0]}:{proxy_values[1]}" + return proxy_url diff --git a/models/api.py b/models/api.py new file mode 100644 index 0000000..c0a72d9 --- /dev/null +++ b/models/api.py @@ -0,0 +1,75 @@ +from typing import Any + +from pydantic import BaseModel + + +class RankData(BaseModel): + id: int + address: str + ens: Any + amount: int + role: str + rank: int + + +class AssetData(BaseModel): + id: int + uid: int + reward: Any + type: str = "energy" + openAt: Any + createdAt: str + + +class OpenBoxData(BaseModel): + energy: int + type: str = "energy" + + +class ClaimData(BaseModel): + code: int + result: int + msg: str + + +class InjectData(BaseModel): + code: int + result: bool + msg: str + + +class UserInfo(BaseModel): + id: int + treeId: int + address: str + ens: Any + energy: int + tree: int + inviteId: int + type: str = "normal" + stake_id: int + nft_id: int + nft_pass: int + signin: int + code: Any + createdAt: str + invitePercent: int + + +class ResponseData(BaseModel): + code: int + result: Any | None = None + msg: str + + +class LoginWalletData(BaseModel): + class User(BaseModel): + id: int + address: str + status: str + inviteId: None | int + twitter: None | str + discord: None | str + + access_token: str + user: User diff --git a/models/config.py b/models/config.py new file mode 100644 index 0000000..927b3eb --- /dev/null +++ b/models/config.py @@ -0,0 +1,10 @@ +from pydantic import BaseModel, HttpUrl, PositiveInt + +from .account import Account + + +class Config(BaseModel): + accounts: list[Account] + referral_code: str + rpc_url: HttpUrl + iteration_delay: PositiveInt diff --git a/models/wallet.py b/models/wallet.py new file mode 100644 index 0000000..5874589 --- /dev/null +++ b/models/wallet.py @@ -0,0 +1,6 @@ +from pydantic import BaseModel + + +class LoginData(BaseModel): + message: str + signed_message: str diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0dada93 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +pyuseragents~=1.0.5 +retrying~=1.3.4 +loguru~=0.7.2 +web3~=6.15.1 +PyYAML~=6.0.1 +pydantic~=2.6.4 +art~=6.1 +orjson~=3.9.15 +httpx~=0.27.0 +tqdm~=4.66.2 +noble_tls +eth_account +curl_cffi \ No newline at end of file diff --git a/run.py b/run.py new file mode 100644 index 0000000..f2d720b --- /dev/null +++ b/run.py @@ -0,0 +1,44 @@ +import asyncio +import sys +import urllib3 +from loguru import logger + +from loader import config +from src.bot import Bot +from utils import show_dev_info + + +def setup(): + urllib3.disable_warnings() + logger.remove() + logger.add( + sys.stdout, + colorize=True, + format="{time:HH:mm:ss} | {level: <8} | - {" + "message}", + ) + logger.add("logs.log", rotation="1 day", retention="7 days") + + +async def run(): + show_dev_info() + logger.info( + f"MintChain Bot started | Version: 1.0 | Total accounts: {len(config.accounts)}\n\n" + ) + + while True: + logger.info(f"Starting new iteration") + tasks = [ + asyncio.create_task(Bot(account).start()) for account in config.accounts + ] + + await asyncio.gather(*tasks) + logger.debug( + f"\n\nIteration finished | Sleeping for {config.iteration_delay} hours\n\n" + ) + await asyncio.sleep(config.iteration_delay * 60 * 60) + + +if __name__ == "__main__": + setup() + asyncio.run(run()) diff --git a/src/api.py b/src/api.py new file mode 100644 index 0000000..8d35a2d --- /dev/null +++ b/src/api.py @@ -0,0 +1,230 @@ +import pyuseragents + +from typing import Literal, List +from noble_tls import Session, Client +from twitter_api import Account as TwitterAccount +from twitter_api.models import BindAccountParamsV2 + +from models import * +from loader import config as configuration + +from .wallet import Wallet +from .exceptions.base import APIError + + +class MintChainAPI(Wallet): + API_URL = "https://www.mintchain.io/api" + + def __init__(self, account_data: Account): + super().__init__(mnemonic=account_data.mnemonic, rpc_url=configuration.rpc_url) + self.account = account_data + self.session = self.setup_session() + + @property + def jwt_token(self) -> str: + return self.session.headers["authorization"].replace("Bearer ", "") + + @property + async def energy_balance(self) -> int: + return (await self.user_info()).energy + + @property + async def tree_size(self) -> int: + return (await self.user_info()).tree + + @property + async def rank(self) -> int: + return (await self.rank_info()).rank + + def setup_session(self) -> Session: + session = Session(client=Client.CHROME_120) + session.timeout_seconds = 15 + session.headers = { + "authority": "www.mintchain.io", + "accept": "application/json, text/plain, */*", + "accept-language": "en-US,en;q=0.9,ru;q=0.8", + "authorization": "Bearer", + "content-type": "application/json", + "origin": "https://www.mintchain.io", + "referer": "https://www.mintchain.io", + "user-agent": pyuseragents.random(), + } + session.proxies = { + "http": self.account.proxy, + "https": self.account.proxy, + } + return session + + async def send_request( + self, + request_type: Literal["POST", "GET"] = "POST", + method: str = None, + json_data: dict = None, + params: dict = None, + url: str = None, + ): + def _verify_response(_response: dict) -> dict: + if "code" in _response: + if _response["code"] != 10000: + raise APIError(f"{_response.get('msg')} | Method: {method}") + + return _response + + raise APIError(f"{_response} | Method: {method}") + + if request_type == "POST": + if not url: + response = await self.session.post( + f"{self.API_URL}{method}", json=json_data, params=params + ) + + else: + response = await self.session.post(url, json=json_data, params=params) + + else: + if not url: + response = await self.session.get( + f"{self.API_URL}{method}", params=params + ) + + else: + response = await self.session.get(url, params=params) + + response.raise_for_status() + return _verify_response(response.json()) + + async def is_daily_reward_claimed(self) -> bool: + response = await self.send_request( + request_type="GET", method="/tree/energy-list" + ) + return response["result"][0]["freeze"] + + async def claim_daily_reward(self) -> int: + json_data = { + "uid": [], + "amount": 500, + "includes": [], + "type": "daily", + "freeze": False, + "id": "500_", + } + + response = await self.send_request(method="/tree/claim", json_data=json_data) + return response["result"] + + async def bind_invite_code(self) -> ResponseData: + jwt_token = self.jwt_token + + session = Session(client=Client.CHROME_120) + session.headers = { + "accept": "application/json, text/plain, */*", + "accept-language": "sk-SK,sk;q=0.9,en-US;q=0.8,en;q=0.7", + "authorization": "Bearer", + "referer": "https://www.mintchain.io/mint-forest", + "user-agent": self.session.headers["user-agent"], + } + + json_data = { + "code": configuration.referral_code, + "jwtToken": jwt_token, + } + + response = await session.get( + "https://www.mintchain.io/api/tree/invitation", params=json_data + ) + return ResponseData(**response.json()) + + async def connect_twitter(self) -> dict: + params = { + "code_challenge": "mintchain", + "code_challenge_method": "plain", + "client_id": "enpfUjhndkdrdHhld29aTW96eGM6MTpjaQ", + "redirect_uri": "https://www.mintchain.io/mint-forest", + "response_type": "code", + "scope": "tweet.read users.read follows.read offline.access", + "state": "mintchain", + } + + twitter_account = TwitterAccount.run( + auth_token=self.account.auth_token, + setup_session=True, + proxy=self.account.proxy, + ) + bind_data = twitter_account.bind_account_v2( + bind_params=BindAccountParamsV2(**params) + ) + + params = { + "code": bind_data.code, + "jwtToken": self.jwt_token, + "address": self.keypair.address, + } + response = await self.send_request( + url="https://www.mintchain.io/api/twitter/verify", params=params + ) + return response + + async def rank_info(self) -> RankData: + response = await self.send_request(request_type="GET", method="/tree/me-rank") + return RankData(**response["result"]) + + async def user_info(self) -> UserInfo: + response = await self.send_request(request_type="GET", method="/tree/user-info") + return UserInfo(**response["result"]) + + async def assets(self) -> List[AssetData]: + response = await self.send_request(request_type="GET", method="/tree/asset") + return [AssetData(**data) for data in response["result"]] + + async def open_box(self, box_id: int) -> OpenBoxData: + json_data = { + "boxId": box_id, + } + + response = await self.send_request(method="/tree/open-box", json_data=json_data) + return OpenBoxData(**response["result"]) + + async def inject(self, amount: int = None) -> InjectData: + if not amount: + amount = await self.energy_balance + + json_data = { + "address": self.keypair.address, + "energy": amount, + } + + response = await self.send_request(method="/tree/inject", json_data=json_data) + return InjectData(**response) + + async def verify_wallet(self) -> ResponseData: + json_data = { + "jwtToken": self.jwt_token, + } + + response = await self.send_request(method="/wallet/verify", json_data=json_data) + return ResponseData(**response) + + async def login(self): + messages = self.sign_login_message() + json_data = { + "address": self.keypair.address, + "signature": messages.signed_message, + "message": messages.message, + } + + response = await self.send_request(method="/tree/login", json_data=json_data) + data = LoginWalletData(**response["result"]) + self.session.headers["authorization"] = f"Bearer {data.access_token}" + + if data.user.status == "pending": + await self.verify_wallet() + + if not data.user.twitter: + await self.connect_twitter() + logger.debug( + f"Account: {self.account.auth_token} | Twitter account connected" + ) + + if not data.user.inviteId: + await self.bind_invite_code() + logger.debug(f"Account: {self.account.auth_token} | Referral code bound") diff --git a/src/bot.py b/src/bot.py new file mode 100644 index 0000000..176ef51 --- /dev/null +++ b/src/bot.py @@ -0,0 +1,105 @@ +import asyncio +import random + +from models import Account +from loguru import logger +from .api import MintChainAPI + + +class Bot(MintChainAPI): + def __init__(self, account: Account): + super().__init__(account_data=account) + + async def safe_operation( + self, + operation: callable, + success_message: str, + error_message: str, + retries: int = 0, + ) -> bool: + for _ in range(retries): + try: + await operation() + logger.success( + f"Account: {self.account.auth_token} | {success_message}" + ) + return True + + except Exception as error: + logger.error( + f"Account: {self.account.auth_token} | {error_message}: {error} | {'Retrying..' if retries > 0 else ''}" + ) + await asyncio.sleep(1) + continue + + return False + + async def process_login(self) -> bool: + return await self.safe_operation( + operation=self.login, + success_message="Logged in", + error_message="Failed to login", + retries=3, + ) + + async def process_claim_daily_reward(self) -> bool: + return ( + await self.safe_operation( + operation=self.claim_daily_reward, + success_message="Daily reward claimed", + error_message="Failed to claim daily reward", + retries=3, + ) + if not await self.is_daily_reward_claimed() + else logger.success( + f"Account: {self.account.auth_token} | Daily reward already claimed" + ) + ) + + async def process_inject(self) -> bool: + return await self.safe_operation( + operation=self.inject, + success_message="Energy injected", + error_message="Failed to inject energy", + retries=3, + ) + + async def process_show_user_info(self) -> None: + try: + info = await self.tree_size + logger.success( + f"Account: {self.account.auth_token} | Total injected energy: {info} | Daily actions done.." + ) + + except Exception as error: + logger.warning( + f"Account: {self.account.auth_token} | Failed to get user info: {error} | Daily actions done.." + ) + + async def start(self): + random_delay = random.randint(1, 30) + logger.info( + f"Account: {self.account.auth_token} | Work will start in {random_delay} seconds.." + ) + await asyncio.sleep(random_delay) + + try: + operations = [ + self.process_login, + self.process_claim_daily_reward, + self.process_inject, + self.process_show_user_info, + ] + + for operation in operations: + if not await operation(): + break + + except Exception as error: + logger.error( + f"Account: {self.account.auth_token} | Unhandled error: {error}" + ) + finally: + logger.success( + f"Account: {self.account.auth_token} | Finished | Waiting for next iteration.." + ) diff --git a/src/exceptions/base.py b/src/exceptions/base.py new file mode 100644 index 0000000..a2dcb07 --- /dev/null +++ b/src/exceptions/base.py @@ -0,0 +1,4 @@ +class APIError(Exception): + """Base class for API exceptions""" + + pass diff --git a/src/wallet.py b/src/wallet.py new file mode 100644 index 0000000..47da1f0 --- /dev/null +++ b/src/wallet.py @@ -0,0 +1,31 @@ +from eth_account import Account +from eth_account.messages import encode_defunct +from web3 import Web3 +from web3.types import Nonce + +from models import LoginData + + +Account.enable_unaudited_hdwallet_features() + + +class Wallet(Web3, Account): + def __init__(self, mnemonic: str, rpc_url: str): + super().__init__(Web3.HTTPProvider(rpc_url)) + self.keypair = self.from_mnemonic(mnemonic) + + @property + def transactions_count(self) -> Nonce: + return self.eth.get_transaction_count(self.keypair.address) + + @property + def get_message(self) -> str: + message = f"You are participating in the Mint Forest event: \n {self.keypair.address}\n\nNonce: {self.transactions_count}" + return message + + def sign_login_message(self) -> LoginData: + encoded_message = encode_defunct(text=self.get_message) + signed_message = self.keypair.sign_message(encoded_message) + return LoginData( + message=self.get_message, signed_message=signed_message.signature.hex() + ) diff --git a/twitter_api/__init__.py b/twitter_api/__init__.py new file mode 100644 index 0000000..b4831a4 --- /dev/null +++ b/twitter_api/__init__.py @@ -0,0 +1,2 @@ +from .account import Account +from .errors import * diff --git a/twitter_api/account.py b/twitter_api/account.py new file mode 100644 index 0000000..8d569a9 --- /dev/null +++ b/twitter_api/account.py @@ -0,0 +1,1664 @@ +import asyncio +import hashlib +import math +import mimetypes +import platform +import secrets +import httpx + +from copy import deepcopy +from datetime import datetime +from string import ascii_letters +from typing import Coroutine +from uuid import uuid1, getnode + +from curl_cffi import requests +from curl_cffi.requests.session import Response +from httpx import Cookies, Headers +from tqdm import tqdm + +from .models import * +from .constants import * +from .errors import TwitterAccountSuspended, RateLimitError +from .models import BindAccountDataV1 +from .util import * + +# logging.getLogger("httpx").setLevel(logging.WARNING) + +if platform.system() != "Windows": + try: + import uvloop + + uvloop.install() + except ImportError as e: + ... + + +class Account: + def __init__(self): + self._session: requests.Session = requests.Session() + self._proxy: str = "" + self._reformatted_proxy: str = "" + + self.gql_api = "https://twitter.com/i/api/graphql" + self.v1_api = "https://api.twitter.com/1.1" + self.v2_api = "https://twitter.com/i/api/2" + + @classmethod + def run( + cls, + auth_token: str = None, + cookies: dict[str] = None, + proxy: str = None, + setup_session: bool = True, + ) -> "Account": + account = cls() + account._proxy = proxy + if proxy: + if proxy.startswith("http://"): + account._session = requests.Session( + proxies={"http://": account.proxy}, timeout=30, verify=False + ) + account._reformatted_proxy = account.proxy + + else: + account._reformatted_proxy = account.get_reformatted_proxy + account._session = requests.Session( + proxies=( + {"http://": account._reformatted_proxy} + if account._reformatted_proxy + else None + ), + timeout=30, + verify=False, + ) + + if not (auth_token, cookies): + raise TwitterError( + { + "error_message": "Failed to authenticate account. You need to set cookies or auth_token." + } + ) + + if setup_session: + if auth_token: + account.session.cookies.update({"auth_token": auth_token}) + account.setup_session() + else: + account.session.cookies.update(cookies) + + else: + if not account.session.cookies.get( + "auth_token" + ) and not account.session.cookies.get("ct0"): + account.session.cookies.update({"auth_token": auth_token}) + account.setup_session() + else: + account.session.cookies.update(cookies) + + return account + + @property + def get_auth_data(self) -> dict: + return { + "auth_token": self.auth_token, + "cookies": dict(self.cookies), + "proxy": self.proxy, + } + + def gql( + self, + method: str, + operation: tuple, + variables: dict, + features: dict = Operation.default_features, + ) -> dict: + qid, op = operation + params = { + "queryId": qid, + "features": features, + "variables": Operation.default_variables | variables, + } + if method == "POST": + data = {"json": params} + else: + data = {"params": {k: orjson.dumps(v).decode() for k, v in params.items()}} + + r = self.session.request( + method=method, + url=f"{self.gql_api}/{qid}/{op}", + headers=get_headers(self.session), + allow_redirects=True, + **data, + ) + + return self._verify_response(r) + + def v1(self, path: str, params: dict) -> dict: + headers = get_headers(self.session) + headers["content-type"] = "application/x-www-form-urlencoded" + r = self.session.post( + f"{self.v1_api}/{path}", headers=headers, data=params, allow_redirects=True + ) + return self._verify_response(r) + + @staticmethod + def _verify_response(r: Response) -> dict: + try: + rate_limit_remaining = r.headers.get("x-rate-limit-remaining") + if rate_limit_remaining and int(rate_limit_remaining) in (0, 1): + reset_ts = int(r.headers.get("x-rate-limit-reset")) + raise RateLimitError( + f"Rate limit reached. Reset in {reset_ts - int(time.time())} seconds. " + ) + # logger.info( + # f"Rate limit reached | Reset in {reset_ts - int(time.time())} seconds | Sleeping..." + # ) + # current_ts = int(time.time()) + # difference = reset_ts - current_ts + # asyncio.sleep(difference) + + data = r.json() + except ValueError: + raise TwitterError( + { + "error_message": f"Failed to parse response: {r.text}. " + "If you are using proxy, make sure it is not blocked by Twitter." + } + ) + + if "errors" in data: + error_message = ( + data["errors"][0].get("message") if data["errors"] else data["errors"] + ) + + error_code = data["errors"][0].get("code") if data["errors"] else None + + if isinstance(error_message, str) and error_message.lower().startswith( + "to protect our users from spam and other" + ): + raise TwitterAccountSuspended(error_message) + + raise TwitterError( + { + "error_code": error_code, + "error_message": error_message, + } + ) + + try: + r.raise_for_status() + except httpx.HTTPError as http_error: + raise TwitterError( + { + "error_message": str(http_error), + } + ) + + return data + + @property + def proxy(self): + return self._proxy + + @property + def get_reformatted_proxy(self): + try: + if self.proxy is None: + return None + + ip, port, username, password = self.proxy.split(":") + return f"http://{username}:{password}@{ip}:{port}" + + except (ValueError, AttributeError): + raise TwitterError( + { + "error_message": "Failed to parse proxy. " + "Make sure you are using correct proxy format: " + "ip:port:username:password" + } + ) + + @property + def session(self): + return self._session + + @property + def cookies(self) -> Cookies: + return self._session.cookies + + @property + def headers(self) -> Headers: + return self._session.headers + + @property + def auth_token(self) -> str: + return self._session.cookies.get("auth_token", "") + + @property + def ct0(self) -> str: + return self._session.cookies.get("ct0", "") + + def request_ct0(self) -> str: + url = "https://twitter.com/i/api/2/oauth2/authorize" + r = self.session.get(url, allow_redirects=True) + + if "ct0" in r.cookies: + return r.cookies.get("ct0") + else: + raise TwitterError( + { + "error_message": "Failed to get ct0 token. " + "Make sure you are using correct cookies." + } + ) + + def request_guest_token( + self, session: requests.Session, csrf_token: str = None + ) -> str: + if not (csrf_token, self.session.cookies.get("ct0", "")): + raise TwitterError( + { + "error_message": "Failed to get guest token. " + "Make sure you are using correct cookies." + } + ) + + headers = { + "content-type": "application/x-www-form-urlencoded", + "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA", + "x-csrf-token": ( + csrf_token if csrf_token else self.session.cookies.get("ct0") + ), + } + r = session.post( + f"{self.v1_api}/guest/activate.json", + headers=headers, + allow_redirects=True, + ) + + data = self._verify_response(r) + return data["guest_token"] + + def setup_session(self): + session = requests.Session( + # follow_redirects=True, + proxies={"http://": self._reformatted_proxy} if self.proxy else None, + timeout=30, + verify=False, + ) + + generated_csrf_token = secrets.token_hex(16) + guest_token = self.request_guest_token(session, generated_csrf_token) + + cookies = {"ct0": generated_csrf_token, "gt": guest_token} + headers = {"x-guest-token": guest_token, "x-csrf-token": generated_csrf_token} + + self.session.headers.update(headers) + self.session.cookies.update(cookies) + csrf_token = self.request_ct0() + + self.session.headers["x-csrf-token"] = csrf_token + self.session.cookies.delete("ct0") + self.session.cookies.update({"ct0": csrf_token}) + self.session.headers = get_headers(self.session) + + self.verify_credentials() + self.session.headers.update( + {"x-csrf-token": self.session.cookies.get("ct0", domain=".twitter.com")} + ) + + def bind_account_v1(self, bind_params: BindAccountParamsV1) -> BindAccountDataV1: + + def get_oauth_token() -> str: + _response = requests.get(str(bind_params.url), allow_redirects=True) + raise_for_status(_response) + + token = re.search( + r' 1: + return token[1] + + raise TwitterError( + { + "error_message": "Failed to get oauth token. " + "Make sure you are using correct cookies or url." + } + ) + + def get_authenticity_token(_oauth_token: str) -> BindAccountDataV1 | str: + params = { + "oauth_token": _oauth_token, + } + _response = self.session.get( + "https://api.twitter.com/oauth/authenticate", params=params + ) + raise_for_status(_response) + + token = re.search( + r' str: + data = { + "authenticity_token": _authenticity_token, + "redirect_after_login": f"https://api.twitter.com/oauth/authorize?oauth_token={_oauth_token}", + "oauth_token": _oauth_token, + } + + response = self.session.post( + "https://api.twitter.com/oauth/authorize", + data=data, + allow_redirects=True, + ) + raise_for_status(response) + + _confirm_url = re.search( + r' BindAccountDataV1: + response = self.session.get(_url, allow_redirects=True) + raise_for_status(response) + + if "status=error" in response.url: + raise TwitterError( + { + "error_message": "Failed to bind account. " + "Make sure you are using correct cookies or url." + } + ) + + _oauth_token, _oauth_verifier = response.url.split("oauth_token=")[1].split( + "&oauth_verifier=" + ) + return BindAccountDataV1( + url=response.url, + oauth_token=_oauth_token, + oauth_verifier=_oauth_verifier, + ) + + oauth_token = get_oauth_token() + authenticity_token = get_authenticity_token(oauth_token) + + if isinstance(authenticity_token, BindAccountDataV1): + return authenticity_token + + confirm_url = get_confirm_url(oauth_token, authenticity_token) + return process_confirm_url(confirm_url) + + def bind_account_v2(self, bind_params: BindAccountParamsV2) -> BindAccountDataV2: + + def get_auth_code() -> str: + response = self.session.get( + "https://twitter.com/i/api/2/oauth2/authorize", + params=bind_params.model_dump(), + ) + raise_for_status(response) + self.session.headers.update( + {"x-csrf-token": self.session.cookies.get("ct0", domain=".twitter.com")} + ) + return response.json()["auth_code"] + + def approve_auth_code(_auth_code: str) -> str: + _params = { + "approval": "true", + "code": _auth_code, + } + + response = self.session.post( + "https://twitter.com/i/api/2/oauth2/authorize", + params=_params, + allow_redirects=True, + ) + raise_for_status(response) + + code = response.json()["redirect_uri"].split("code=")[1] + return code + + auth_code = get_auth_code() + approved_code = approve_auth_code(auth_code) + return BindAccountDataV2(code=approved_code) + + def create_poll(self, text: str, choices: list[str], poll_duration: int) -> dict: + options = { + "twitter:card": "poll4choice_text_only", + "twitter:api:api:endpoint": "1", + "twitter:long:duration_minutes": poll_duration, # max: 10080 + } + for i, c in enumerate(choices): + options[f"twitter:string:choice{i + 1}_label"] = c + + headers = get_headers(self.session) + headers["content-type"] = "application/x-www-form-urlencoded" + url = "https://caps.twitter.com/v2/cards/create.json" + + r = self.session.post( + url, + headers=headers, + params={"card_data": orjson.dumps(options).decode()}, + allow_redirects=True, + ) + card_uri = (self._verify_response(r))["card_uri"] + + data = self.tweet(text, poll_params={"card_uri": card_uri}) + return data + + def verify_credentials(self) -> dict: + r = self.session.get( + f"{self.v1_api}/account/verify_credentials.json", allow_redirects=True + ) + return self._verify_response(r) + + def email_phone_info(self) -> dict: + r = self.session.get( + f"{self.v1_api}/users/email_phone_info.json", allow_redirects=True + ) + return self._verify_response(r) + + def settings_info(self) -> dict: + r = self.session.get( + f"{self.v1_api}/account/settings.json", allow_redirects=True + ) + return self._verify_response(r) + + def screen_name(self) -> str: + data = self.verify_credentials() + return data["screen_name"] + + def user_id(self) -> int: + data = self.verify_credentials() + return data["id"] + + def name(self) -> str: + data = self.verify_credentials() + return data["name"] + + def location(self) -> str: + data = self.verify_credentials() + return data["location"] + + def description(self) -> str: + data = self.verify_credentials() + return data["description"] + + def followers_count(self) -> int: + data = self.verify_credentials() + return data["followers_count"] + + def friends_count(self) -> int: + data = self.verify_credentials() + return data["friends_count"] + + def registration_date(self) -> str: + data = self.verify_credentials() + return data["created_at"] + + def suspended(self) -> bool: + data = self.verify_credentials() + return data["suspended"] + + def dm(self, text: str, receivers: list[int], media: str = "") -> dict: + variables = { + "message": {}, + "requestId": str(uuid1(getnode())), + "target": {"participant_ids": receivers}, + } + if media: + media_id = self.upload_media(media, is_dm=True) + variables["message"]["media"] = {"id": media_id, "text": text} + else: + variables["message"]["text"] = {"text": text} + + res = self.gql("POST", Operation.useSendMessageMutation, variables) + if find_key(res, "dm_validation_failure_type"): + raise TwitterError( + { + "error_message": "Failed to send message. Sender does not have privilege to dm receiver(s)", + "error_code": 349, + } + ) + return res + + def custom_dm(self, text: str, receiver: int) -> dict: + json_data = { + "event": { + "type": "message_create", + "message_create": { + "target": {"recipient_id": f"{receiver}"}, + "message_data": {"text": f"{text}"}, + }, + } + } + + r = self.session.post( + f"{self.v1_api}/direct_messages/events/new.json", + json=json_data, + ) + return self._verify_response(r) + + def delete_tweet(self, tweet_id: int | str) -> dict: + variables = {"tweet_id": tweet_id, "dark_request": False} + return self.gql("POST", Operation.DeleteTweet, variables) + + def tweet( + self, text: str, *, media: List[MediaEntity] = None, **kwargs + ) -> dict | Coroutine[Any, Any, dict]: + variables = { + "tweet_text": text, + "dark_request": False, + "media": { + "media_entities": [], + "possibly_sensitive": False, + }, + "semantic_annotation_ids": [], + } + + if reply_params := kwargs.get("reply_params", {}): + variables |= reply_params + if quote_params := kwargs.get("quote_params", {}): + variables |= quote_params + if poll_params := kwargs.get("poll_params", {}): + variables |= poll_params + + draft = kwargs.get("draft") + schedule = kwargs.get("schedule") + + if draft or schedule: + variables = { + "post_tweet_request": { + "auto_populate_reply_metadata": False, + "status": text, + "exclude_reply_user_ids": [], + "media_ids": [], + }, + } + if media: + for m in media: + media_id = self.upload_media(m["media"]) + variables["post_tweet_request"]["media_ids"].append(media_id) + if alt := m.get("alt"): + self._add_alt_text(media_id, alt) + + if schedule: + variables["execute_at"] = ( + datetime.strptime(schedule, "%Y-%m-%d %H:%M").timestamp() + if isinstance(schedule, str) + else schedule + ) + return self.gql("POST", Operation.CreateScheduledTweet, variables) + + return self.gql("POST", Operation.CreateDraftTweet, variables) + + # regular tweet + if media: + for m in media: + + tagged_users_id = [] + for tagged_user in m.tagged_users: + user_id = self.get_user_id(tagged_user) + tagged_users_id.append(user_id) + + variables["media"]["media_entities"].append( + {"media_id": m.media_id, "tagged_users": tagged_users_id} + ) + + return self.gql("POST", Operation.CreateTweet, variables) + + def schedule_tweet( + self, text: str, date: int | str, *, media: List[MediaEntity] = None + ) -> dict: + variables = { + "post_tweet_request": { + "auto_populate_reply_metadata": False, + "status": text, + "exclude_reply_user_ids": [], + "media_ids": [], + }, + "execute_at": ( + datetime.strptime(date, "%Y-%m-%d %H:%M").timestamp() + if isinstance(date, str) + else date + ), + } + if media: + for m in media: + + tagged_users_id = [] + for tagged_user in m.tagged_users: + user_id = self.get_user_id(tagged_user) + tagged_users_id.append(user_id) + + variables["media"]["media_entities"].append( + {"media_id": m.media_id, "tagged_users": tagged_users_id} + ) + + return self.gql("POST", Operation.CreateScheduledTweet, variables) + + def schedule_reply( + self, text: str, date: int | str, tweet_id: int, *, media: list = None + ) -> dict: + variables = { + "post_tweet_request": { + "auto_populate_reply_metadata": True, + "in_reply_to_status_id": tweet_id, + "status": text, + "exclude_reply_user_ids": [], + "media_ids": [], + }, + "execute_at": ( + datetime.strptime(date, "%Y-%m-%d %H:%M").timestamp() + if isinstance(date, str) + else date + ), + } + if media: + for m in media: + media_id = self.upload_media(m["media"]) + variables["post_tweet_request"]["media_ids"].append(media_id) + if alt := m.get("alt"): + self._add_alt_text(media_id, alt) + + return self.gql("POST", Operation.CreateScheduledTweet, variables) + + def unschedule_tweet(self, tweet_id: int | str) -> dict: + variables = {"scheduled_tweet_id": tweet_id} + return self.gql("POST", Operation.DeleteScheduledTweet, variables) + + def untweet(self, tweet_id: int | str) -> dict: + variables = {"tweet_id": tweet_id, "dark_request": False} + return self.gql("POST", Operation.DeleteTweet, variables) + + def reply( + self, text: str, tweet_id: int | str, media: List[MediaEntity] = None + ) -> dict: + variables = { + "tweet_text": text, + "reply": { + "in_reply_to_tweet_id": tweet_id, + "exclude_reply_user_ids": [], + }, + "batch_compose": "BatchSubsequent", + "dark_request": False, + "media": { + "media_entities": [], + "possibly_sensitive": False, + }, + "semantic_annotation_ids": [], + } + + if media: + for m in media: + tagged_users_id = [] + + for tagged_user in m.tagged_users: + user_id = self.get_user_id(tagged_user) + tagged_users_id.append(user_id) + + variables["media"]["media_entities"].append( + {"media_id": m.media_id, "tagged_users": tagged_users_id} + ) + + return self.gql("POST", Operation.CreateTweet, variables) + + def quote(self, text: str, tweet_id: int) -> dict: + variables = { + "tweet_text": text, + # can use `i` as it resolves to screen_name + "attachment_url": f"https://twitter.com/i/status/{tweet_id}", + "dark_request": False, + "media": { + "media_entities": [], + "possibly_sensitive": False, + }, + "semantic_annotation_ids": [], + } + return self.gql("POST", Operation.CreateTweet, variables) + + def retweet(self, tweet_id: int) -> dict: + variables = {"tweet_id": tweet_id, "dark_request": False} + return self.gql("POST", Operation.CreateRetweet, variables) + + def unretweet(self, tweet_id: int) -> dict: + variables = {"source_tweet_id": tweet_id, "dark_request": False} + return self.gql("POST", Operation.DeleteRetweet, variables) + + @staticmethod + def __get_cursor_value(data: dict, target_cursor_type: str, target_entry_type: str): + if target_entry_type != "threaded_conversation_with_injections_v2": + for instruction in ( + data.get("data", {}) + .get(target_entry_type, {}) + .get("timeline", {}) + .get("instructions", []) + ): + for entry in instruction.get("entries", []): + content = entry.get("content", {}) + cursor_type = content.get("cursorType") + if ( + content.get("entryType") == "TimelineTimelineCursor" + and cursor_type == target_cursor_type + ): + return content.get("value") + + else: + for instruction in ( + data.get("data", {}).get(target_entry_type, {}).get("instructions", []) + ): + for entry in instruction.get("entries", []): + content = entry.get("content", {}) + cursor_type = content.get("cursorType") + if ( + content.get("entryType") == "TimelineTimelineCursor" + and cursor_type == target_cursor_type + ): + return content.get("value") + + return None + + def tweet_likes( + self, celery_task, tweet_id: int, limit: int = 0 + ) -> dict[str, list[dict]]: + variables = {"tweetId": tweet_id, "count": 100} + users_data = [] + + while True: + data = self.gql("GET", Operation.Favoriters, variables) + + for instruction in ( + data.get("data", {}) + .get("favoriters_timeline", {}) + .get("timeline", {}) + .get("instructions", []) + ): + try: + for entry in instruction["entries"]: + try: + result = entry["content"]["itemContent"]["user_results"][ + "result" + ] + screen_name = result["legacy"]["screen_name"] + if screen_name not in ( + user["screen_name"] for user in users_data + ): + users_data.append( + self.get_user_data_from_user_results(result) + ) + + except (KeyError, TypeError, IndexError): + continue + + except KeyError: + return {"users": users_data[:limit] if limit > 0 else users_data} + + cursor_value = self.__get_cursor_value( + data, "Bottom", "favoriters_timeline" + ) + if not cursor_value or (0 < limit <= len(users_data)): + return {"users": users_data[:limit] if limit > 0 else users_data} + + variables["cursor"] = cursor_value + + def tweet_retweeters( + self, celery_task, tweet_id: int, limit: int = 0 + ) -> dict[str, list[Any]]: + variables = {"tweetId": tweet_id, "count": 100} + tweets_data = [] + + while True: + data = self.gql("GET", Operation.Retweeters, variables) + + for instruction in data["data"]["retweeters_timeline"]["timeline"][ + "instructions" + ]: + try: + for entry in instruction["entries"]: + try: + result = entry["content"]["itemContent"]["user_results"][ + "result" + ] + screen_name = result["legacy"]["screen_name"] + if screen_name not in ( + user["screen_name"] for user in tweets_data + ): + tweets_data.append( + self.get_user_data_from_user_results(result) + ) + except (KeyError, TypeError, IndexError): + continue + + except KeyError: + return {"users": tweets_data[:limit] if limit > 0 else tweets_data} + + cursor_value = self.__get_cursor_value( + data, "Bottom", "retweeters_timeline" + ) + + if not cursor_value or (0 < limit <= len(tweets_data)): + return {"users": tweets_data[:limit] if limit > 0 else tweets_data} + + variables["cursor"] = cursor_value + + @staticmethod + def get_user_data_from_user_results(data: dict) -> dict: + legacy = data.get("legacy", {}) + + return { + "id": data.get("rest_id"), + "name": legacy.get("name"), + "screen_name": legacy.get("screen_name"), + "profile_image_url": legacy.get("profile_image_url_https"), + "favourites_count": legacy.get("favourites_count"), + "followers_count": legacy.get("followers_count"), + "friends_count": legacy.get("friends_count"), + "location": legacy.get("location"), + "description": legacy.get("description"), + "created_at": legacy.get("created_at"), + } + + def tweet_replies( + self, celery_task, tweet_id: int, limit: int = 0 + ) -> dict[str, list[dict[str, dict | Any]]]: + variables = {"focalTweetId": tweet_id} + replies_data = [] + + while True: + data = self.gql("GET", Operation.TweetDetail, variables) + + for entry in data["data"]["threaded_conversation_with_injections_v2"][ + "instructions" + ][0]["entries"]: + try: + result = entry["content"]["items"][0]["item"]["itemContent"][ + "tweet_results" + ]["result"] + reply_text = result["legacy"]["full_text"] + user_results = result["core"]["user_results"]["result"] + + if reply_text not in ( + reply["reply_text"] for reply in replies_data + ): + replies_data.append( + { + "reply_text": reply_text, + "user_data": self.get_user_data_from_user_results( + user_results + ), + } + ) + except (KeyError, TypeError, IndexError): + continue + + entries = data["data"]["threaded_conversation_with_injections_v2"][ + "instructions" + ][0]["entries"] + if not entries[-1]["entryId"].startswith("cursor-bottom") or ( + 0 < limit <= len(replies_data) + ): + return {"replies": replies_data[:limit] if limit > 0 else replies_data} + + for entry in entries: + if entry["entryId"].startswith("cursor-bottom"): + cursor_value = entry["content"]["itemContent"]["value"] + variables["cursor"] = cursor_value + break + + def user_followers(self, celery_task, username: str, limit: int = 0) -> list[str]: + variables = {"screen_name": username, "count": 200} + users = [] + + while True: + r = self.session.get(f"{self.v1_api}/followers/list.json", params=variables) + if r.status_code == 503: + asyncio.sleep(3) + continue + + else: + data = self._verify_response(r) + new_users = [user["screen_name"] for user in data["users"]] + users.extend(new_users) + + next_cursor = int(data.get("next_cursor")) + if next_cursor == 0 or (0 < limit <= len(users)): + return users[:limit] if limit > 0 else users + + variables["cursor"] = data["next_cursor_str"] + + def user_followings(self, username: str) -> list[str]: + variables = {"screen_name": username, "count": 200} + users = [] + + while True: + r = self.session.get(f"{self.v1_api}/friends/list.json", params=variables) + if r.status_code == 503: + asyncio.sleep(5) + continue + + else: + data = self._verify_response(r) + new_users = [user["screen_name"] for user in data["users"]] + users.extend(new_users) + + if int(data.get("next_cursor")) == 0: + return users + + variables["cursor"] = data["next_cursor_str"] + + def user_last_tweets( + self, user_id: int, username: str + ) -> list[dict[str, str, str | None, str | None, str | None]]: + data = self.gql("GET", Operation.UserTweets, {"userId": user_id}) + + try: + tweets_data = [] + timeline = data["data"]["user"]["result"]["timeline_v2"]["timeline"] + for tweet in timeline["instructions"]: + entries = tweet.get("entries", []) + for entry in entries: + if entry["entryId"].startswith("tweet"): + tweet_link = f"https://twitter.com/{username}/status/{entry['entryId'].split('-')[-1]}" + else: + continue + + tweet_results = ( + entry.get("content", {}) + .get("itemContent", {}) + .get("tweet_results", {}) + .get("result", {}) + .get("legacy") + ) + if tweet_results and tweet_results.get("full_text"): + full_text = tweet_results["full_text"] + created_at = tweet_results.get("created_at", "") + is_quote_status = tweet_results.get("is_quote_status", "") + lang = tweet_results.get("lang", "") + + tweets_data.append( + { + "tweet_link": tweet_link, + "full_text": full_text, + "created_at": created_at, + "is_quote_status": is_quote_status, + "lang": lang, + } + ) + + return tweets_data + + except Exception as error: + raise TwitterError({"error_message": f"Failed to get user tweets: {error}"}) + + def like(self, tweet_id: int) -> dict: + variables = {"tweet_id": tweet_id} + return self.gql("POST", Operation.FavoriteTweet, variables) + + def unlike(self, tweet_id: int) -> dict: + variables = {"tweet_id": tweet_id} + return self.gql("POST", Operation.UnfavoriteTweet, variables) + + def bookmark(self, tweet_id: int) -> dict: + variables = {"tweet_id": tweet_id} + return self.gql("POST", Operation.CreateBookmark, variables) + + def unbookmark(self, tweet_id: int) -> dict: + variables = {"tweet_id": tweet_id} + return self.gql("POST", Operation.DeleteBookmark, variables) + + def create_list(self, name: str, description: str, private: bool) -> dict: + variables = { + "isPrivate": private, + "name": name, + "description": description, + } + return self.gql("POST", Operation.CreateList, variables) + + def update_list( + self, list_id: int, name: str, description: str, private: bool + ) -> dict: + variables = { + "listId": list_id, + "isPrivate": private, + "name": name, + "description": description, + } + return self.gql("POST", Operation.UpdateList, variables) + + def update_pinned_lists(self, list_ids: list[int]) -> dict: + """ + Update pinned lists. + Reset all pinned lists and pin all specified lists in the order they are provided. + + @param list_ids: list of list ids to pin + @return: response + """ + return self.gql("POST", Operation.ListsPinMany, {"listIds": list_ids}) + + def pin_list(self, list_id: int) -> dict: + return self.gql("POST", Operation.ListPinOne, {"listId": list_id}) + + def unpin_list(self, list_id: int) -> dict: + return self.gql("POST", Operation.ListUnpinOne, {"listId": list_id}) + + def add_list_member(self, list_id: int, user_id: int) -> dict: + return self.gql( + "POST", Operation.ListAddMember, {"listId": list_id, "userId": user_id} + ) + + def remove_list_member(self, list_id: int, user_id: int) -> dict: + return self.gql( + "POST", Operation.ListRemoveMember, {"listId": list_id, "userId": user_id} + ) + + def delete_list(self, list_id: int) -> dict: + return self.gql("POST", Operation.DeleteList, {"listId": list_id}) + + def update_list_banner(self, list_id: int, media: str) -> dict: + media_id = self.upload_media(media) + variables = {"listId": list_id, "mediaId": media_id} + return self.gql("POST", Operation.EditListBanner, variables) + + def delete_list_banner(self, list_id: int) -> dict: + return self.gql("POST", Operation.DeleteListBanner, {"listId": list_id}) + + def follow_topic(self, topic_id: int) -> dict: + return self.gql("POST", Operation.TopicFollow, {"topicId": str(topic_id)}) + + def unfollow_topic(self, topic_id: int) -> dict: + return self.gql("POST", Operation.TopicUnfollow, {"topicId": str(topic_id)}) + + def pin(self, tweet_id: int) -> dict: + return self.v1( + "account/pin_tweet.json", {"tweet_mode": "extended", "id": tweet_id} + ) + + def unpin(self, tweet_id: int) -> dict: + return self.v1( + "account/unpin_tweet.json", {"tweet_mode": "extended", "id": tweet_id} + ) + + def get_user_id(self, username: str) -> int: + headers = get_headers(self.session) + headers["content-type"] = "application/x-www-form-urlencoded" + r = self.session.get( + f"{self.v1_api}/users/show.json", + headers=headers, + params={"screen_name": username}, + ) + data = self._verify_response(r) + return data["id"] + + def get_user_info(self, username: str) -> dict: + headers = get_headers(self.session) + headers["content-type"] = "application/x-www-form-urlencoded" + r = self.session.get( + f"{self.v1_api}/users/show.json", + headers=headers, + params={"screen_name": username}, + ) + return self._verify_response(r) + + def follow(self, user_id: int | str) -> dict: + settings = deepcopy(follow_settings) + settings |= {"user_id": user_id} + return self.v1("friendships/create.json", settings) + + def unfollow(self, user_id: int | str) -> dict: + settings = deepcopy(follow_settings) + settings |= {"user_id": user_id} + return self.v1("friendships/destroy.json", settings) + + def mute(self, user_id: int) -> dict: + return self.v1("mutes/users/create.json", {"user_id": user_id}) + + def unmute(self, user_id: int) -> dict: + return self.v1("mutes/users/destroy.json", {"user_id": user_id}) + + def enable_follower_notifications(self, user_id: int) -> dict: + settings = deepcopy(follower_notification_settings) + settings |= {"id": user_id, "device": "true"} + return self.v1("friendships/update.json", settings) + + def disable_follower_notifications(self, user_id: int) -> dict: + settings = deepcopy(follower_notification_settings) + settings |= {"id": user_id, "device": "false"} + return self.v1("friendships/update.json", settings) + + def block(self, user_id: int) -> dict: + return self.v1("blocks/create.json", {"user_id": user_id}) + + def unblock(self, user_id: int) -> dict: + return self.v1("blocks/destroy.json", {"user_id": user_id}) + + def update_profile_image(self, media: str) -> dict: + media_id = self.upload_media(media) + params = {"media_id": media_id} + + r = self.session.post( + f"{self.v1_api}/account/update_profile_image.json", + headers=get_headers(self.session), + params=params, + ) + return self._verify_response(r) + + def update_profile_banner(self, media: str) -> dict: + media_id = self.upload_media(media) + params = {"media_id": media_id} + + r = self.session.post( + f"{self.v1_api}/account/update_profile_banner.json", + headers=get_headers(self.session), + params=params, + ) + return self._verify_response(r) + + def update_profile_info(self, params: dict) -> dict: + headers = get_headers(self.session) + r = self.session.post( + f"{self.v1_api}/account/update_profile.json", headers=headers, params=params + ) + + return self._verify_response(r) + + def update_search_settings(self, settings: dict) -> dict: + twid = int(self.session.cookies.get("twid").split("=")[-1].strip('"')) + headers = get_headers(self.session) + + r = self.session.post( + url=f"{self.v1_api}/strato/column/User/{twid}/search/searchSafety", + headers=headers, + json=settings, + ) + return self._verify_response(r) + + def update_settings(self, settings: dict) -> dict: + return self.v1("account/settings.json", settings) + + def update_username(self, username: str): + return self.update_settings({"screen_name": username}) + + def change_password(self, old: str, new: str) -> dict: + params = { + "current_password": old, + "password": new, + "password_confirmation": new, + } + headers = get_headers(self.session) + headers["content-type"] = "application/x-www-form-urlencoded" + + r = self.session.post( + f"{self.v1_api}/account/change_password.json", + headers=headers, + data=params, + allow_redirects=True, + ) + return self._verify_response(r) + + def remove_interests(self, *args) -> dict: + """ + Pass 'all' to remove all interests + """ + r = self.session.get( + f"{self.v1_api}/account/personalization/twitter_interests.json", + headers=get_headers(self.session), + ) + current_interests = r.json()["interested_in"] + if args == "all": + disabled_interests = [x["id"] for x in current_interests] + else: + disabled_interests = [ + x["id"] for x in current_interests if x["display_name"] in args + ] + payload = { + "preferences": { + "interest_preferences": { + "disabled_interests": disabled_interests, + "disabled_partner_interests": [], + } + } + } + r = self.session.post( + f"{self.v1_api}/account/personalization/p13n_preferences.json", + headers=get_headers(self.session), + json=payload, + ) + return self._verify_response(r) + + def home_timeline(self, limit=math.inf) -> list[dict]: + return self._paginate( + "POST", Operation.HomeTimeline, Operation.default_variables, int(limit) + ) + + def home_latest_timeline(self, limit=math.inf) -> list[dict]: + return self._paginate( + "POST", + Operation.HomeLatestTimeline, + Operation.default_variables, + int(limit), + ) + + def bookmarks(self, limit=math.inf) -> list[dict]: + return self._paginate("GET", Operation.Bookmarks, {}, int(limit)) + + def _paginate( + self, method: str, operation: tuple, variables: dict, limit: int + ) -> list[dict]: + initial_data = self.gql(method, operation, variables) + res = [initial_data] + ids = set(find_key(initial_data, "rest_id")) + dups = 0 + DUP_LIMIT = 3 + + cursor = get_cursor(initial_data) + while (dups < DUP_LIMIT) and cursor: + prev_len = len(ids) + if prev_len >= limit: + return res + + variables["cursor"] = cursor + data = self.gql(method, operation, variables) + + cursor = get_cursor(data) + ids |= set(find_key(data, "rest_id")) + + if prev_len == len(ids): + dups += 1 + + res.append(data) + return res + + def custom_upload_media(self, file: Path) -> int | None: + url = "https://upload.twitter.com/1.1/media/upload.json" + + headers = get_headers(self.session) + with httpx.Client( + headers=headers, cookies=dict(self.session.cookies) + ) as client: + upload_type = "tweet" + media_type = mimetypes.guess_type(file)[0] + media_category = ( + f"{upload_type}_gif" + if "gif" in media_type + else f'{upload_type}_{media_type.split("/")[0]}' + ) + + files = {"media": file.read_bytes()} + + post_data = {} + if media_category is not None: + post_data["media_category"] = media_category + + r = client.post(url=url, json=params, params=post_data, files=files) + + data = self._verify_response(r) + return data["media_id"] + + def upload_media(self, filename: str, is_dm: bool = False) -> int | None: + """ + https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/uploading-media/media-best-practices + """ + + def check_media(category: str, size: int) -> None: + fmt = lambda x: f"{(x / 1e6):.2f} MB" + msg = ( + lambda x: f"cannot upload {fmt(size)} {category}, max size is {fmt(x)}" + ) + if category == "image" and size > MAX_IMAGE_SIZE: + raise Exception(msg(MAX_IMAGE_SIZE)) + if category == "gif" and size > MAX_GIF_SIZE: + raise Exception(msg(MAX_GIF_SIZE)) + if category == "video" and size > MAX_VIDEO_SIZE: + raise Exception(msg(MAX_VIDEO_SIZE)) + + # if is_profile: + # url = 'https://upload.twitter.com/i/media/upload.json' + # else: + # url = 'https://upload.twitter.com/1.1/media/upload.json' + + url = "https://upload.twitter.com/i/media/upload.json" + + file = Path(filename) + total_bytes = file.stat().st_size + headers = get_headers(self.session) + + upload_type = "dm" if is_dm else "tweet" + media_type = mimetypes.guess_type(file)[0] + media_category = ( + f"{upload_type}_gif" + if "gif" in media_type + else f'{upload_type}_{media_type.split("/")[0]}' + ) + + check_media(media_category, total_bytes) + + params = { + "command": "INIT", + "media_type": media_type, + "total_bytes": total_bytes, + "media_category": media_category, + } + r = self.session.post( + url=url, headers=headers, params=params, allow_redirects=True + ) + + data = self._verify_response(r) + media_id = data["media_id"] + + desc = f"uploading: {file.name}" + with tqdm( + total=total_bytes, desc=desc, unit="B", unit_scale=True, unit_divisor=1024 + ) as pbar: + with open(file, "rb") as fp: + i = 0 + while chunk := fp.read(UPLOAD_CHUNK_SIZE): + params = { + "command": "APPEND", + "media_id": media_id, + "segment_index": i, + } + try: + pad = bytes( + "".join(random.choices(ascii_letters, k=16)), + encoding="utf-8", + ) + data = b"".join( + [ + b"------WebKitFormBoundary", + pad, + b'\r\nContent-Disposition: form-data; name="media"; filename="blob"', + b"\r\nContent-Type: application/octet-stream", + b"\r\n\r\n", + chunk, + b"\r\n------WebKitFormBoundary", + pad, + b"--\r\n", + ] + ) + _headers = { + b"content-type": b"multipart/form-data; boundary=----WebKitFormBoundary" + + pad + } + self.session.post( + url=url, + headers=headers | _headers, + params=params, + content=data, + allow_redirects=True, + ) + except Exception as error: + try: + files = {"media": chunk} + self.session.post( + url=url, headers=headers, params=params, files=files + ) + except Exception as error: + return + + i += 1 + pbar.update(fp.tell() - pbar.n) + + params = {"command": "FINALIZE", "media_id": media_id, "allow_async": "true"} + if is_dm: + params |= {"original_md5": hashlib.md5(file.read_bytes()).hexdigest()} + + r = self.session.post( + url=url, headers=headers, params=params, allow_redirects=True + ) + data = self._verify_response(r) + + processing_info = data.get("processing_info") + while processing_info: + state = processing_info["state"] + if error := processing_info.get("error"): + return + if state == MEDIA_UPLOAD_SUCCEED: + break + if state == MEDIA_UPLOAD_FAIL: + return + check_after_secs = processing_info.get( + "check_after_secs", random.randint(1, 5) + ) + + time.sleep(check_after_secs) + params = {"command": "STATUS", "media_id": media_id} + + r = self.session.get( + url=url, headers=headers, params=params, allow_redirects=True + ) + data = self._verify_response(r) + processing_info = data.get("processing_info") + + return media_id + + def _add_alt_text(self, media_id: int, text: str) -> dict: + params = {"media_id": media_id, "alt_text": {"text": text}} + url = f"{self.v1_api}/media/metadata/create.json" + r = self.session.post(url, headers=get_headers(self.session), json=params) + return self._verify_response(r) + + def dm_inbox(self) -> dict: + """ + Get DM inbox metadata. + + @return: inbox as dict + """ + r = self.session.get( + f"{self.v1_api}/dm/inbox_initial_state.json", + headers=get_headers(self.session), + params=dm_params, + ) + return self._verify_response(r) + + # def dm_history(self, conversation_ids: list[str] = None) -> list[dict]: + # """ + # Get DM history. + # + # Call without arguments to get all DMS from all conversations. + # + # @param conversation_ids: optional list of conversation ids + # @return: list of messages as dicts + # """ + # + # def get(session: AsyncClient, conversation_id: str): + # params = deepcopy(dm_params) + # r = session.get( + # f"{self.v1_api}/dm/conversation/{conversation_id}.json", + # params=params, + # ) + # res = (self._verify_response(r)).get("conversation_timeline", {}) + # data = [x.get("message") for x in res.get("entries", [])] + # entry_id = res.get("min_entry_id") + # while entry_id: + # params["max_id"] = entry_id + # r = session.get( + # f"{self.v1_api}/dm/conversation/{conversation_id}.json", + # params=params, + # ) + # res = (self._verify_response(r)).get("conversation_timeline", {}) + # data.extend(x["message"] for x in res.get("entries", [])) + # entry_id = res.get("min_entry_id") + # return data + # + # def process(ids): + # limits = Limits(max_connections=100) + # headers, cookies = get_headers(self.session), self.session.cookies + # async with AsyncClient( + # limits=limits, headers=headers, cookies=cookies, timeout=20 + # ) as c: + # return tqdm_asyncio.gather( + # *(get(c, _id) for _id in ids), desc="Getting DMs" + # ) + # + # if conversation_ids: + # ids = conversation_ids + # else: + # # get all conversations + # inbox = self.dm_inbox() + # ids = list(inbox["inbox_initial_state"]["conversations"]) + # + # return asyncio.run(process(ids)) + + def dm_delete(self, *, conversation_id: str = None, message_id: str = None) -> dict: + """ + Delete operations + + - delete (hide) a single DM + - delete an entire conversation + + @param conversation_id: the conversation id + @param message_id: the message id + @return: result metadata + """ + self.session.headers.update(headers=get_headers(self.session)) + results = {"conversation": None, "message": None} + if conversation_id: + results["conversation"] = self.session.post( + f"{self.v1_api}/dm/conversation/{conversation_id}/delete.json", + ) # not json response + if message_id: + # delete single message + _id, op = Operation.DMMessageDeleteMutation + results["message"] = self.session.post( + f"{self.gql_api}/{_id}/{op}", + json={"queryId": _id, "variables": {"messageId": message_id}}, + ) + return results + + def dm_search(self, query: str) -> dict: + """ + Search DMs by keyword + + @param query: search term + @return: search results as dict + """ + + def get(cursor=None): + if cursor: + params["variables"]["cursor"] = cursor.pop() + _id, op = Operation.DmAllSearchSlice + r = self.session.get( + f"{self.gql_api}/{_id}/{op}", + params=build_params(params), + ) + res = r.json() + cursor = find_key(res, "next_cursor") + return res, cursor + + self.session.headers.update(headers=get_headers(self.session)) + variables = deepcopy(Operation.default_variables) + variables["count"] = 50 # strict limit, errors thrown if exceeded + variables["query"] = query + params = {"variables": variables, "features": Operation.default_features} + res, cursor = get() + data = [res] + while cursor: + res, cursor = get(cursor) + data.append(res) + return {"query": query, "data": data} + + def scheduled_tweets(self, ascending: bool = True) -> dict: + variables = {"ascending": ascending} + return self.gql("GET", Operation.FetchScheduledTweets, variables) + + def delete_scheduled_tweet(self, tweet_id: int) -> dict: + """duplicate, same as `unschedule_tweet()`""" + variables = {"scheduled_tweet_id": tweet_id} + return self.gql("POST", Operation.DeleteScheduledTweet, variables) + + def clear_scheduled_tweets(self) -> None: + user_id = int(re.findall('"u=(\d+)"', self.session.cookies.get("twid"))[0]) + drafts = self.gql("GET", Operation.FetchScheduledTweets, {"ascending": True}) + for _id in set(find_key(drafts, "rest_id")): + if _id != user_id: + self.gql( + "POST", Operation.DeleteScheduledTweet, {"scheduled_tweet_id": _id} + ) + + def draft_tweets(self, ascending: bool = True) -> dict: + variables = {"ascending": ascending} + return self.gql("GET", Operation.FetchDraftTweets, variables) + + def delete_draft_tweet(self, tweet_id: int) -> dict: + variables = {"draft_tweet_id": tweet_id} + return self.gql("POST", Operation.DeleteDraftTweet, variables) + + def clear_draft_tweets(self) -> None: + user_id = int(re.findall('"u=(\d+)"', self.session.cookies.get("twid"))[0]) + drafts = self.gql("GET", Operation.FetchDraftTweets, {"ascending": True}) + for _id in set(find_key(drafts, "rest_id")): + if _id != user_id: + self.gql("POST", Operation.DeleteDraftTweet, {"draft_tweet_id": _id}) + + def notifications(self, params: dict = None) -> dict: + r = self.session.get( + f"{self.v2_api}/notifications/all.json", + headers=get_headers(self.session), + params=params or live_notification_params, + ) + return self._verify_response(r) + + def recommendations(self, params: dict = None) -> dict: + r = self.session.get( + f"{self.v1_api}/users/recommendations.json", + headers=get_headers(self.session), + params=params or recommendations_params, + ) + return self._verify_response(r) + + def fleetline(self, params: dict = None) -> dict: + r = self.session.get( + "https://twitter.com/i/api/fleets/v1/fleetline", + headers=get_headers(self.session), + params=params or {}, + ) + return self._verify_response(r) + + @property + def id(self) -> int: + """Get User ID""" + return int(re.findall('"u=(\d+)"', self.session.cookies.get("twid"))[0]) + + def save_cookies(self, fname: str = None): + """Save cookies to file""" + cookies = self.session.cookies + Path(f'{fname or cookies.get("username")}.cookies').write_bytes( + orjson.dumps(dict(cookies)) + ) diff --git a/twitter_api/constants.py b/twitter_api/constants.py new file mode 100644 index 0000000..643b8c3 --- /dev/null +++ b/twitter_api/constants.py @@ -0,0 +1,801 @@ +from dataclasses import dataclass + + +MAX_IMAGE_SIZE = 5_242_880 # ~5 MB +MAX_GIF_SIZE = 15_728_640 # ~15 MB +MAX_VIDEO_SIZE = 536_870_912 # ~530 MB + +UPLOAD_CHUNK_SIZE = 4 * 1024 * 1024 +MEDIA_UPLOAD_SUCCEED = "succeeded" +MEDIA_UPLOAD_FAIL = "failed" + +BLACK = "\x1b[30m" +RED = "\x1b[31m" +GREEN = "\x1b[32m" +YELLOW = "\x1b[33m" +BLUE = "\x1b[34m" +MAGENTA = "\x1b[35m" +CYAN = "\x1b[36m" +WHITE = "\x1b[37m" +BOLD = "\x1b[1m" +RESET = "\x1b[0m" + +LOG_CONFIG = { + "version": 1, + "disable_existing_loggers": False, + "formatters": { + "standard": { + "format": "%(asctime)s.%(msecs)03d [%(levelname)s] :: %(message)s", + "datefmt": "%Y-%m-%d %H:%M:%S", + }, + }, + "handlers": { + "console": { + "class": "logging.StreamHandler", + "level": "DEBUG", + "formatter": "standard", + "stream": "ext://sys.stdout", + }, + "file": { + "class": "logging.FileHandler", + "level": "DEBUG", + "formatter": "standard", + "filename": "twitter.log", + "mode": "a", + }, + }, + "loggers": { + "twitter": { + "handlers": ["console", "file"], + "level": "DEBUG", + } + }, +} + +ID_MAP = { + "Followers": "^user-\d+$", + "Following": "^user-\d+$", + "UserTweets": "^tweet-\d+$", + "Likes": "^tweet-\d+$", + "UserMedia": "^tweet-\d+$", + "TweetResultByRestId": "^tweet-\d+$", + "TweetsAndReplies": "^profile-conversation-\d+-tweet-\d+$", + "TweetDetail": "^conversationthread-\d+-tweet-\d+$", # if another key after tweet-\d+, it's an ad + "Retweeters": "^user-\d+$", + "Favoriters": "^user-\d+$", +} + + +@dataclass +class SearchCategory: + Top = "Top" + Latest = "Latest" + People = "People" + Photos = "Photos" + Videos = "Videos" + + +@dataclass +class SpaceCategory: + Top = "Top" + Live = "Live" + Upcoming = "Upcoming" + + +@dataclass +class SpaceState: + Ended = "Ended" + Canceled = "Canceled" + NotStarted = "NotStarted" + PrePublished = "PrePublished" + Running = "Running" + TimedOut = "TimedOut" + + +@dataclass +class Operation: + # todo: dynamically update + SearchTimeline = ( + {"rawQuery": str, "product": str}, + "nK1dw4oV3k4w5TdtcAdSww", + "SearchTimeline", + ) + AudioSpaceById = {"id": str}, "fYAuJHiY3TmYdBmrRtIKhA", "AudioSpaceById" + AudioSpaceSearch = ( + {"filter": str, "query": str}, + "NTq79TuSz6fHj8lQaferJw", + "AudioSpaceSearch", + ) + UserByScreenName = ( + {"screen_name": str}, + "sLVLhk0bGj3MVFEKTdax1w", + "UserByScreenName", + ) + UserTweets = "HuTx74BxAnezK1gWvYY7zg", "UserTweets" + ProfileSpotlightsQuery = ( + {"screen_name": str}, + "9zwVLJ48lmVUk8u_Gh9DmA", + "ProfileSpotlightsQuery", + ) + UserByRestId = {"userId": int}, "GazOglcBvgLigl3ywt6b3Q", "UserByRestId" + UsersByRestIds = {"userIds": list}, "OJBgJQIrij6e3cjqQ3Zu1Q", "UsersByRestIds" + UserMedia = {"userId": int}, "YqiE3JL1KNgf9nSljYdxaA", "UserMedia" + UserTweetsAndReplies = ( + {"userId": int}, + "RIWc55YCNyUJ-U3HHGYkdg", + "UserTweetsAndReplies", + ) + TweetResultByRestId = ( + {"tweetId": int}, + "D_jNhjWZeRZT5NURzfJZSQ", + "TweetResultByRestId", + ) + TweetDetail = "zXaXQgfyR4GxE21uwYQSyA", "TweetDetail" + TweetStats = {"rest_id": int}, "EvbTkPDT-xQCfupPu0rWMA", "TweetStats" + Likes = {"userId": int}, "nXEl0lfN_XSznVMlprThgQ", "Likes" + Followers = {"userId": int}, "pd8Tt1qUz1YWrICegqZ8cw", "Followers" + Following = {"userId": int}, "wjvx62Hye2dGVvnvVco0xA", "Following" + Retweeters = "0BoJlKAxoNPQUHRftlwZ2w", "Retweeters" + Favoriters = "XRRjv1-uj1HZn3o324etOQ", "Favoriters" + ConnectTabTimeline = ( + {"context": dict}, + "lq02A-gEzbLefqTgD_PFzQ", + "ConnectTabTimeline", + ) + + # Account Operations + useSendMessageMutation = "MaxK2PKX1F9Z-9SwqwavTw", "useSendMessageMutation" + CreateTweet = "7TKRKCPuAGsmYde0CudbVg", "CreateTweet" + DeleteTweet = "VaenaVgh5q5ih7kvyVjgtg", "DeleteTweet" + CreateScheduledTweet = "LCVzRQGxOaGnOnYH01NQXg", "CreateScheduledTweet" + DeleteScheduledTweet = "CTOVqej0JBXAZSwkp1US0g", "DeleteScheduledTweet" + CreateRetweet = "ojPdsZsimiJrUGLR1sjUtA", "CreateRetweet" + DeleteRetweet = "iQtK4dl5hBmXewYZuEOKVw", "DeleteRetweet" + FavoriteTweet = "lI07N6Otwv1PhnEgXILM7A", "FavoriteTweet" + UnfavoriteTweet = "ZYKSe-w7KEslx3JhSIk5LA", "UnfavoriteTweet" + CreateBookmark = "aoDbu3RHznuiSkQ9aNM67Q", "CreateBookmark" + DeleteBookmark = "Wlmlj2-xzyS1GN3a6cj-mQ", "DeleteBookmark" + CreateList = "hQAsnViq2BrMLbPuQ9umDA", "CreateList" + UpdateList = "4dCEFWtxEbhnSLcJdJ6PNg", "UpdateList" + ListsPinMany = "2X4Vqu6XLneR-XZnGK5MAw", "ListsPinMany" + ListPinOne = "2pYlo-kjdXoNOZJoLzI6KA", "ListPinOne" + ListUnpinOne = "c4ce-hzx6V4heV5IzdeBkA", "ListUnpinOne" + ListAddMember = "P8tyfv2_0HzofrB5f6_ugw", "ListAddMember" + ListRemoveMember = "DBZowzFN492FFkBPBptCwg", "ListRemoveMember" + DeleteList = "UnN9Th1BDbeLjpgjGSpL3Q", "DeleteList" + EditListBanner = "Uk0ZwKSMYng56aQdeJD1yw", "EditListBanner" + DeleteListBanner = "-bOKetDVCMl20qXn7YDXIA", "DeleteListBanner" + TopicFollow = "ElqSLWFmsPL4NlZI5e1Grg", "TopicFollow" + TopicUnfollow = "srwjU6JM_ZKTj_QMfUGNcw", "TopicUnfollow" + HomeLatestTimeline = "zhX91JE87mWvfprhYE97xA", "HomeLatestTimeline" + HomeTimeline = "HCosKfLNW1AcOo3la3mMgg", "HomeTimeline" + Bookmarks = "tmd4ifV8RHltzn8ymGg1aw", "Bookmarks" + + # misc/not implemented + AdAccounts = "a8KxGfFQAmm3WxqemuqSRA", "AdAccounts" + ArticleTimeline = "o9FyvnC-xg8mVBXqL4g-rg", "ArticleTimeline" + ArticleTweetsTimeline = "x4ywSpvg6BesoDszkfbFQg", "ArticleTweetsTimeline" + AudienceEstimate = "1LYVUabJBYkPlUAWRabB3g", "AudienceEstimate" + AuthenticatedUserTFLists = "QjN8ZdavFDqxUjNn3r9cig", "AuthenticatedUserTFLists" + BirdwatchAliasSelect = "3ss48WFwGokBH_gj8t_8aQ", "BirdwatchAliasSelect" + BirdwatchCreateAppeal = "TKdL0YFsX4DMOpMKeneLvA", "BirdwatchCreateAppeal" + BirdwatchCreateNote = "36EUZZyaciVmNrq4CRZcmw", "BirdwatchCreateNote" + BirdwatchCreateRating = "bD3AEK9BMCSpRods_ng2fA", "BirdwatchCreateRating" + BirdwatchDeleteNote = "IKS_qrShkDyor6Ri1ahd9g", "BirdwatchDeleteNote" + BirdwatchDeleteRating = "OpvCOyOoQClUND66zDzrnA", "BirdwatchDeleteRating" + BirdwatchEditNotificationSettings = ( + "FLgLReVIssXjB_ui3wcrRQ", + "BirdwatchEditNotificationSettings", + ) + BirdwatchFetchAliasSelfSelectOptions = ( + "szoXMke8AZOErso908iglw", + "BirdwatchFetchAliasSelfSelectOptions", + ) + BirdwatchFetchAliasSelfSelectStatus = ( + "LUEdtkcpBlGktUtms4BvwA", + "BirdwatchFetchAliasSelfSelectStatus", + ) + BirdwatchFetchAuthenticatedUserProfile = ( + "pMbW6Y4LuS5MzlSOEqERJQ", + "BirdwatchFetchAuthenticatedUserProfile", + ) + BirdwatchFetchBirdwatchProfile = ( + "btgGtchypc3D491MJ7XXWA", + "BirdwatchFetchBirdwatchProfile", + ) + BirdwatchFetchContributorNotesSlice = ( + "t6r3Wq7wripUW9gB3FQNBw", + "BirdwatchFetchContributorNotesSlice", + ) + BirdwatchFetchGlobalTimeline = ( + "L3LftPt6fhYqoQ5Vnxm7UQ", + "BirdwatchFetchGlobalTimeline", + ) + BirdwatchFetchNotes = "ZGMhf1M7kPKMOhEk1nz0Yw", "BirdwatchFetchNotes" + BirdwatchFetchOneNote = "GO8BR2MM2WZB63cdOoC7lw", "BirdwatchFetchOneNote" + BirdwatchFetchPublicData = "9bDdJ6AL26RLkcUShEcF-A", "BirdwatchFetchPublicData" + BirdwatchProfileAcknowledgeEarnOut = ( + "cED9wJy8Nd1kZCCYuIq9zQ", + "BirdwatchProfileAcknowledgeEarnOut", + ) + BizProfileFetchUser = "6OFpJ3TH3p8JpwOSgfgyhg", "BizProfileFetchUser" + BlockedAccountsAll = "h52d1F7dumWGE1tJAhQBpg", "BlockedAccountsAll" + BlockedAccountsAutoBlock = "8w-D2OhT0jmGzXaNY--UQA", "BlockedAccountsAutoBlock" + BlockedAccountsImported = "8LDNeOEm0kA98uoDsqXvMg", "BlockedAccountsImported" + BookmarkFolderTimeline = "13H7EUATwethsj-XxX5ohw", "BookmarkFolderTimeline" + BookmarkFoldersSlice = "i78YDd0Tza-dV4SYs58kRg", "BookmarkFoldersSlice" + BookmarksAllDelete = "skiACZKC1GDYli-M8RzEPQ", "BookmarksAllDelete" + Budgets = "mbK3oSQotwcJXyQIBE3uYw", "Budgets" + CardPreviewByTweetText = "jnwTSDR-Eo_HWlSkXPcMGA", "CardPreviewByTweetText" + CheckTweetForNudge = "C2dcvh7H69JALtomErxWlA", "CheckTweetForNudge" + CombinedLists = "rIxum3avpCu7APi7mxTNjw", "CombinedLists" + CommunitiesMainDiscoveryModule = ( + "8UB2fhB8TiYIW2M6vbBFXg", + "CommunitiesMainDiscoveryModule", + ) + CommunitiesMainPageTimeline = ( + "DzcxPzkGYVQk-BD0pqAcZw", + "CommunitiesMainPageTimeline", + ) + CommunitiesMembershipsSlice = ( + "s8-oxdVsoJ3w2CFD0nFt9g", + "CommunitiesMembershipsSlice", + ) + CommunitiesMembershipsTimeline = ( + "QXo-eKTsvhpCyFotNz2u6g", + "CommunitiesMembershipsTimeline", + ) + CommunityAboutTimeline = "plOgdpBzpVVQbTOEVuRc_A", "CommunityAboutTimeline" + CommunityByRestId = "bCVwRBDPi15jrdJQ7NCENQ", "CommunityByRestId" + CommunityCreateRule = "dShPoN6voXRusgxC1uvGog", "CommunityCreateRule" + CommunityDiscoveryTimeline = "b3rceNUXWRyo5mSwVZF74Q", "CommunityDiscoveryTimeline" + CommunityEditBannerMedia = "KVkZwp8Q6xy6iyhlQE5d7Q", "CommunityEditBannerMedia" + CommunityEditName = "SKToKhvm3Z4Rir8ENCJ3YQ", "CommunityEditName" + CommunityEditPurpose = "eMat-u2kx6KocreGTAt-hA", "CommunityEditPurpose" + CommunityEditRule = "9nEl5bNcdteuPGbGCdvEFA", "CommunityEditRule" + CommunityEditTheme = "4OhW6gWJwiu-JTAgBPsU1w", "CommunityEditTheme" + CommunityHashtagsTimeline = "hril1TsnshopHbmnjdUmhQ", "CommunityHashtagsTimeline" + CommunityMemberRelationshipTypeahead = ( + "NEwac2-8ONgf0756ne8oXA", + "CommunityMemberRelationshipTypeahead", + ) + CommunityModerationKeepTweet = ( + "f_YqrHSCc1mPlG-aB7pFRw", + "CommunityModerationKeepTweet", + ) + CommunityModerationTweetCasesSlice = ( + "V-iC7tjWOlzBJ44SanqGzw", + "CommunityModerationTweetCasesSlice", + ) + CommunityRemoveBannerMedia = "lSdK1v30qVhm37rDTgHq0Q", "CommunityRemoveBannerMedia" + CommunityRemoveRule = "EI_g43Ss_Ixg0EC4K7nzlQ", "CommunityRemoveRule" + CommunityReorderRules = "VwluNMGnl5uaNZ3LnlCQ_A", "CommunityReorderRules" + CommunityTweetsRankedTimeline = ( + "P38EspBBPhAfSKPP74-s2Q", + "CommunityTweetsRankedTimeline", + ) + CommunityTweetsTimeline = "2JgHOlqfeLusxAT0yGQJjg", "CommunityTweetsTimeline" + CommunityUpdateRole = "5eq76kkUqfdCzInCtcxQOA", "CommunityUpdateRole" + CommunityUserInvite = "x8hUNaBCOV2tSalqB9cwWQ", "CommunityUserInvite" + CommunityUserRelationshipTypeahead = ( + "gi_UGcUurYp6N6p2BaLJqQ", + "CommunityUserRelationshipTypeahead", + ) + ConversationControlChange = "hb1elGcj6769uT8qVYqtjw", "ConversationControlChange" + ConversationControlDelete = "OoMO_aSZ1ZXjegeamF9QmA", "ConversationControlDelete" + ConvertRitoSuggestedActions = ( + "2njnYoE69O2jdUM7KMEnDw", + "ConvertRitoSuggestedActions", + ) + Coupons = "R1h43jnAl2bsDoUkgZb7NQ", "Coupons" + CreateCommunity = "lRjZKTRcWuqwtYwCWGy9_w", "CreateCommunity" + CreateCustomerPortalSession = ( + "2LHXrd1uYeaMWhciZgPZFw", + "CreateCustomerPortalSession", + ) + CreateDraftTweet = "cH9HZWz_EW9gnswvA4ZRiQ", "CreateDraftTweet" + CreateNoteTweet = "Pyx6nga4XtTVhfTh1gtX1A", "CreateNoteTweet" + CreateQuickPromotion = "oDSoVgHhJxnd5IkckgPZdg", "CreateQuickPromotion" + CreateTrustedFriendsList = "2tP8XUYeLHKjq5RHvuvpZw", "CreateTrustedFriendsList" + CreateTweetDownvote = "Eo65jl-gww30avDgrXvhUA", "CreateTweetDownvote" + CreateTweetReaction = "D7M6X3h4-mJE8UB1Ap3_dQ", "CreateTweetReaction" + DataSaverMode = "xF6sXnKJfS2AOylzxRjf6A", "DataSaverMode" + DeleteBookmarkFolder = "2UTTsO-6zs93XqlEUZPsSg", "DeleteBookmarkFolder" + DeleteDraftTweet = "bkh9G3FGgTldS9iTKWWYYw", "DeleteDraftTweet" + DeletePaymentMethod = "VaaLGwK5KNLoc7wsOmp4uw", "DeletePaymentMethod" + DeleteTweetDownvote = "VNEvEGXaUAMfiExP8Tbezw", "DeleteTweetDownvote" + DeleteTweetReaction = "GKwK0Rj4EdkfwdHQMZTpuw", "DeleteTweetReaction" + DisableUserAccountLabel = "_ckHEj05gan2VfNHG6thBA", "DisableUserAccountLabel" + DisableVerifiedPhoneLabel = "g2m0pAOamawNtVIfjXNMJg", "DisableVerifiedPhoneLabel" + DismissRitoSuggestedAction = "jYvwa61cv3NwNP24iUru6g", "DismissRitoSuggestedAction" + DmAllSearchSlice = "U-QXVRZ6iddb1QuZweh5DQ", "DmAllSearchSlice" + DmGroupSearchSlice = "5zpY1dCR-8NyxQJS_CFJoQ", "DmGroupSearchSlice" + DmMutedTimeline = "lrcWa13oyrQc7L33wRdLAQ", "DmMutedTimeline" + DMMessageDeleteMutation = "BJ6DtxA2llfjnRoRjaiIiw", "DMMessageDeleteMutation" + DmNsfwMediaFilterUpdate = "of_N6O33zfyD4qsFJMYFxA", "DmNsfwMediaFilterUpdate" + DmPeopleSearchSlice = "xYSm8m5kJnzm_gFCn5GH-w", "DmPeopleSearchSlice" + EditBookmarkFolder = "a6kPp1cS1Dgbsjhapz1PNw", "EditBookmarkFolder" + EditDraftTweet = "JIeXE-I6BZXHfxsgOkyHYQ", "EditDraftTweet" + EditScheduledTweet = "_mHkQ5LHpRRjSXKOcG6eZw", "EditScheduledTweet" + EnableLoggedOutWebNotifications = ( + "BqIHKmwZKtiUBPi07jKctg", + "EnableLoggedOutWebNotifications", + ) + EnableVerifiedPhoneLabel = "C3RJFfMsb_KcEytpKmRRkw", "EnableVerifiedPhoneLabel" + EnrollCoupon = "SOyGmNGaEXcvk15s5bqDrA", "EnrollCoupon" + ExplorePage = "fkypGKlR9Xz9kLvUZDLoXw", "ExplorePage" + FeatureSettingsUpdate = "-btar_vkBwWA7s3YWfp_9g", "FeatureSettingsUpdate" + FetchDraftTweets = "ZkqIq_xRhiUme0PBJNpRtg", "FetchDraftTweets" + FetchScheduledTweets = "ITtjAzvlZni2wWXwf295Qg", "FetchScheduledTweets" + FollowersYouKnow = "RvojYJJB90VwJ0rdVhbjMQ", "FollowersYouKnow" + ForYouExplore = "wVEXnyTWzQlEsIuLq_D3tw", "ForYouExplore" + GenericTimelineById = "LZfAdxTdNolKXw6ZkoY_kA", "GenericTimelineById" + GetSafetyModeSettings = "AhxTX0lkbIos4WG53xwzSA", "GetSafetyModeSettings" + GetTweetReactionTimeline = "ihIcULrtrtPGlCuprduRrA", "GetTweetReactionTimeline" + GetUserClaims = "lFi3xnx0auUUnyG4YwpCNw", "GetUserClaims" + GraphQLError = "2V2W3HIBuMW83vEMtfo_Rg", "GraphQLError" + ImmersiveMedia = "UGQD_VslAJBJ4XzigsBYAA", "ImmersiveMedia" + JoinCommunity = "PXO-mA1KfmLqB9I6R-lOng", "JoinCommunity" + LeaveCommunity = "AtiTdhEyRN8ruNFW069ewQ", "LeaveCommunity" + ListByRestId = "wXzyA5vM_aVkBL9G8Vp3kw", "ListByRestId" + ListBySlug = "3-E3eSWorCv24kYkK3CCiQ", "ListBySlug" + ListCreationRecommendedUsers = ( + "Zf8ZwG57EKtss-rPlryIqg", + "ListCreationRecommendedUsers", + ) + ListEditRecommendedUsers = "-F4wsOirYNXjjg-ZjccQpQ", "ListEditRecommendedUsers" + ListLatestTweetsTimeline = "2TemLyqrMpTeAmysdbnVqw", "ListLatestTweetsTimeline" + ListMembers = "vA952kfgGw6hh8KatWnbqw", "ListMembers" + ListMemberships = "BlEXXdARdSeL_0KyKHHvvg", "ListMemberships" + ListOwnerships = "wQcOSjSQ8NtgxIwvYl1lMg", "ListOwnerships" + ListPins = "J0JOhmi8HSsle8LfSWv0cw", "ListPins" + ListProductSubscriptions = "wwdBYgScze0_Jnan79jEUw", "ListProductSubscriptions" + ListRankedTweetsTimeline = "07lytXX9oG9uCld1RY4b0w", "ListRankedTweetsTimeline" + ListSubscribe = "FjvrQI3k-97JIUbEE6Gxcw", "ListSubscribe" + ListSubscribers = "e57wIELAAe0fYt4Hmqsk6g", "ListSubscribers" + ListUnsubscribe = "bXyvW9HoS_Omy4ADhexj8A", "ListUnsubscribe" + ListsDiscovery = "ehnzbxPHA69pyaV2EydN1g", "ListsDiscovery" + ListsManagementPageTimeline = ( + "nhYp4n09Hi5n2hQWseQztg", + "ListsManagementPageTimeline", + ) + LiveCommerceItemsSlice = "-lnNX56S2YrZYrLzbccFAQ", "LiveCommerceItemsSlice" + ModerateTweet = "pjFnHGVqCjTcZol0xcBJjw", "ModerateTweet" + ModeratedTimeline = "hnaqw2Vok5OETdBVa_uexw", "ModeratedTimeline" + MuteList = "ZYyanJsskNUcltu9bliMLA", "MuteList" + MutedAccounts = "-G9eXTmseyiSenbqjrEG6w", "MutedAccounts" + NoteworthyAccountsPage = "3fOJzEwYMnVyzwgLTLIBkw", "NoteworthyAccountsPage" + PaymentMethods = "mPF_G9okpbZuLcD6mN8K9g", "PaymentMethods" + PinReply = "GA2_1uKP9b_GyR4MVAQXAw", "PinReply" + ProfileUserPhoneState = "5kUWP8C1hcd6omvg6HXXTQ", "ProfileUserPhoneState" + PutClientEducationFlag = "IjQ-egg0uPkY11NyPMfRMQ", "PutClientEducationFlag" + QuickPromoteEligibility = "LtpCXh66W-uXh7u7XSRA8Q", "QuickPromoteEligibility" + RemoveFollower = "QpNfg0kpPRfjROQ_9eOLXA", "RemoveFollower" + RemoveTweetFromBookmarkFolder = ( + "2Qbj9XZvtUvyJB4gFwWfaA", + "RemoveTweetFromBookmarkFolder", + ) + RequestToJoinCommunity = "6G66cW5zuxPXmHOeBOjF2w", "RequestToJoinCommunity" + RitoActionedTweetsTimeline = "px9Zbs48D-YdQPEROK6-nA", "RitoActionedTweetsTimeline" + RitoFlaggedAccountsTimeline = ( + "lMzaBZHIbD6GuPqJJQubMg", + "RitoFlaggedAccountsTimeline", + ) + RitoFlaggedTweetsTimeline = "iCuXMibh6yj9AelyjKXDeA", "RitoFlaggedTweetsTimeline" + RitoSuggestedActionsFacePile = ( + "GnQKeEdL1LyeK3dTQCS1yw", + "RitoSuggestedActionsFacePile", + ) + SetDefault = "QEMLEzEMzoPNbeauKCCLbg", "SetDefault" + SetSafetyModeSettings = "qSJIPIpf4gA7Wn21bT3D4w", "SetSafetyModeSettings" + SharingAudiospacesListeningDataWithFollowersUpdate = ( + "5h0kNbk3ii97rmfY6CdgAA", + "SharingAudiospacesListeningDataWithFollowersUpdate", + ) + SubscribeToScheduledSpace = "Sxn4YOlaAwEKjnjWV0h7Mw", "SubscribeToScheduledSpace" + SubscriptionCheckoutUrlWithEligibility = ( + "hKfOOObQr5JmfmxW0YtPvg", + "SubscriptionCheckoutUrlWithEligibility", + ) + SubscriptionProductDetails = "f0dExZDmFWFSWMCPQSAemQ", "SubscriptionProductDetails" + SubscriptionProductFeaturesFetch = ( + "Me2CVcAXxvK2WMr-Nh_Qqg", + "SubscriptionProductFeaturesFetch", + ) + SuperFollowers = "o0YtPFnd4Lk_pOQb9alCvA", "SuperFollowers" + TopicByRestId = "4OUZZOonV2h60I0wdlQb_w", "TopicByRestId" + TopicLandingPage = "mAKQjs1kyTS75VLZzuIXXw", "TopicLandingPage" + TopicNotInterested = "cPCFdDAaqRjlMRYInZzoDA", "TopicNotInterested" + TopicToFollowSidebar = "RPWVYYupHVZkJOnokbt2cw", "TopicToFollowSidebar" + TopicUndoNotInterested = "4tVnt6FoSxaX8L-mDDJo4Q", "TopicUndoNotInterested" + TopicsManagementPage = "Jvdjpe8qzsJD84BpK3qdkQ", "TopicsManagementPage" + TopicsPickerPage = "UvG-XXtWNcJN1LzF0u3ByA", "TopicsPickerPage" + TopicsPickerPageById = "t6kH4v2c_VzWKljc2yNwHA", "TopicsPickerPageById" + TrustedFriendsTypeahead = "RRnOwHttRGscWKC1zY9VRA", "TrustedFriendsTypeahead" + TweetEditHistory = "8eaWKjHszkS-G_hprUd9AA", "TweetEditHistory" + TwitterArticleByRestId = "hwrvh-Qt24lcprL-BDfqRA", "TwitterArticleByRestId" + TwitterArticleCreate = "aV-sm-IkvwplcxdYDoLZHQ", "TwitterArticleCreate" + TwitterArticleDelete = "6st-stMDc7KBqLT8KvWhHg", "TwitterArticleDelete" + TwitterArticleUpdateCoverImage = ( + "fpcVRSAsjvkwmCiN1HheqQ", + "TwitterArticleUpdateCoverImage", + ) + TwitterArticleUpdateData = "XpBTYp_QXwyZ0XT0JXCBJw", "TwitterArticleUpdateData" + TwitterArticleUpdateMedia = "3ojmmegfBC_oHyrmPhxj-g", "TwitterArticleUpdateMedia" + TwitterArticleUpdateTitle = "dvH6Ql989I4e5jWEV7HfaQ", "TwitterArticleUpdateTitle" + TwitterArticleUpdateVisibility = ( + "8M35gHyfpcy3S4UXejUGfA", + "TwitterArticleUpdateVisibility", + ) + TwitterArticlesSlice = "UUPSi_aS8_kHDFTWqSBPUA", "TwitterArticlesSlice" + UnmentionUserFromConversation = ( + "xVW9j3OqoBRY9d6_2OONEg", + "UnmentionUserFromConversation", + ) + UnmoderateTweet = "pVSyu6PA57TLvIE4nN2tsA", "UnmoderateTweet" + UnmuteList = "pMZrHRNsmEkXgbn3tOyr7Q", "UnmuteList" + UnpinReply = "iRe6ig5OV1EzOtldNIuGDQ", "UnpinReply" + UnsubscribeFromScheduledSpace = ( + "Zevhh76Msw574ZSs2NQHGQ", + "UnsubscribeFromScheduledSpace", + ) + UrtFixtures = "I_0j1mjMwv94SdS66S4pqw", "UrtFixtures" + UserAboutTimeline = "dm7ReTFJoeU0qkiZCO1E1g", "UserAboutTimeline" + UserAccountLabel = "rD5gLxVmMvtdtYU1UHWlFQ", "UserAccountLabel" + UserBusinessProfileTeamTimeline = ( + "dq1eUCn3N8v0BywlP4nT7A", + "UserBusinessProfileTeamTimeline", + ) + UserPromotableTweets = "jF-OgMv-9vAym3JaCPUnhQ", "UserPromotableTweets" + UserSessionsList = "vJ-XatpmQSG8bDch8-t9Jw", "UserSessionsList" + UserSuperFollowTweets = "1by3q8-AJWdNYhtltjlPTQ", "UserSuperFollowTweets" + Viewer = "okNaf-6AQWu2DD2H_MAoVw", "Viewer" + ViewerEmailSettings = "JpjlNgn4sLGvS6tgpTzYBg", "ViewerEmailSettings" + ViewerTeams = "D8mVcJSVv66_3NcR7fOf6g", "ViewerTeams" + ViewingOtherUsersTopicsPage = ( + "tYXo6h_rpnHXbdLUFMatZA", + "ViewingOtherUsersTopicsPage", + ) + WriteDataSaverPreferences = "H03etWvZGz41YASxAU2YPg", "WriteDataSaverPreferences" + WriteEmailNotificationSettings = ( + "2qKKYFQift8p5-J1k6kqxQ", + "WriteEmailNotificationSettings", + ) + adFreeArticleDomains = "zwTrX9CtnMvWlBXjsx95RQ", "adFreeArticleDomains" + articleNudgeDomains = "88Bu08U2ddaVVjKmmXjVYg", "articleNudgeDomains" + bookmarkTweetToFolder = "4KHZvvNbHNf07bsgnL9gWA", "bookmarkTweetToFolder" + createBookmarkFolder = "6Xxqpq8TM_CREYiuof_h5w", "createBookmarkFolder" + getAltTextPromptPreference = "PFIxTk8owMoZgiMccP0r4g", "getAltTextPromptPreference" + getCaptionsAlwaysDisplayPreference = ( + "BwgMOGpOViDS0ri7VUgglg", + "getCaptionsAlwaysDisplayPreference", + ) + timelinesFeedback = "vfVbgvTPTQ-dF_PQ5lD1WQ", "timelinesFeedback" + updateAltTextPromptPreference = ( + "aQKrduk_DA46XfOQDkcEng", + "updateAltTextPromptPreference", + ) + updateCaptionsAlwaysDisplayPreference = ( + "uCUQhvZ5sJ9qHinRp6CFlQ", + "updateCaptionsAlwaysDisplayPreference", + ) + + default_variables = { + "count": 1000, + "withSafetyModeUserFields": True, + "includePromotedContent": True, + "withQuickPromoteEligibilityTweetFields": True, + "withVoice": True, + "withV2Timeline": True, + "withDownvotePerspective": False, + "withBirdwatchNotes": True, + "withCommunity": True, + "withSuperFollowsUserFields": True, + "withReactionsMetadata": False, + "withReactionsPerspective": False, + "withSuperFollowsTweetFields": True, + "isMetatagsQuery": False, + "withReplays": True, + "withClientEventToken": False, + "withAttachments": True, + "withConversationQueryHighlights": True, + "withMessageQueryHighlights": True, + "withMessages": True, + } + default_features = { + "blue_business_profile_image_shape_enabled": True, + "creator_subscriptions_tweet_preview_api_enabled": True, + "freedom_of_speech_not_reach_fetch_enabled": True, + "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True, + "graphql_timeline_v2_bookmark_timeline": True, + "hidden_profile_likes_enabled": True, + "highlights_tweets_tab_ui_enabled": True, + "interactive_text_enabled": True, + "longform_notetweets_consumption_enabled": True, + "longform_notetweets_inline_media_enabled": True, + "longform_notetweets_rich_text_read_enabled": True, + "longform_notetweets_richtext_consumption_enabled": True, + "profile_foundations_tweet_stats_enabled": True, + "profile_foundations_tweet_stats_tweet_frequency": True, + "responsive_web_birdwatch_note_limit_enabled": True, + "responsive_web_edit_tweet_api_enabled": True, + "responsive_web_enhance_cards_enabled": False, + "responsive_web_graphql_exclude_directive_enabled": True, + "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False, + "responsive_web_graphql_timeline_navigation_enabled": True, + "responsive_web_media_download_video_enabled": False, + "responsive_web_text_conversations_enabled": False, + "responsive_web_twitter_article_data_v2_enabled": True, + "responsive_web_twitter_article_tweet_consumption_enabled": False, + "responsive_web_twitter_blue_verified_badge_is_enabled": True, + "rweb_lists_timeline_redesign_enabled": True, + "spaces_2022_h2_clipping": True, + "spaces_2022_h2_spaces_communities": True, + "standardized_nudges_misinfo": True, + "subscriptions_verification_info_verified_since_enabled": True, + "tweet_awards_web_tipping_enabled": False, + "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True, + "tweetypie_unmention_optimization_enabled": True, + "verified_phone_label_enabled": False, + "vibe_api_enabled": True, + "view_counts_everywhere_api_enabled": True, + } + + +trending_params = { + "include_profile_interstitial_type": "1", + "include_blocking": "1", + "include_blocked_by": "1", + "include_followed_by": "1", + "include_want_retweets": "1", + "include_mute_edge": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_verified_type": "1", + "skip_status": "1", + "cards_platform": "Web-12", + "include_cards": "1", + "include_ext_alt_text": "true", + "include_ext_limited_action_results": "false", + "include_quote_count": "true", + "include_reply_count": "1", + "tweet_mode": "extended", + "include_ext_views": "true", + "include_entities": "true", + "include_user_entities": "true", + "include_ext_media_color": "true", + "include_ext_media_availability": "true", + "include_ext_sensitive_media_warning": "true", + "include_ext_trusted_friends_metadata": "true", + "send_error_codes": "true", + "simple_quoted_tweet": "true", + "count": 1000, + "requestContext": "launch", + "include_page_configuration": "true", + "initial_tab_id": "trending", + "entity_tokens": "false", + "ext": "mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,enrichments,superFollowMetadata,unmentionInfo,editControl,vibe", +} + +account_settings = { + "address_book_live_sync_enabled": False, + "allow_ads_personalization": False, + "allow_authenticated_periscope_requests": True, + "allow_dm_groups_from": "following", + "allow_dms_from": "following", # all + "allow_location_history_personalization": False, + "allow_logged_out_device_personalization": False, + "allow_media_tagging": "none", # all, following + "allow_sharing_data_for_third_party_personalization": False, + "alt_text_compose_enabled": None, + "always_use_https": True, + "autoplay_disabled": False, + "country_code": "us", + "discoverable_by_email": False, + "discoverable_by_mobile_phone": False, + "display_sensitive_media": True, + "dm_quality_filter": "enabled", # disabled + "dm_receipt_setting": "all_disabled", # all_enabled + "geo_enabled": False, + "include_alt_text_compose": True, + "include_mention_filter": True, + "include_nsfw_admin_flag": True, + "include_nsfw_user_flag": True, + "include_ranked_timeline": True, + "language": "en", + "mention_filter": "unfiltered", + "nsfw_admin": False, + "nsfw_user": False, + "personalized_trends": True, + "protected": False, + "ranked_timeline_eligible": None, + "ranked_timeline_setting": None, + "require_password_login": False, + "requires_login_verification": False, + "settings_metadata": {}, + "sleep_time": {"enabled": False, "end_time": None, "start_time": None}, + "translator_type": "none", + "universal_quality_filtering_enabled": "enabled", + "use_cookie_personalization": False, + ## todo: not yet implemented - requires additional steps + # 'allow_contributor_request': 'all', + # 'protect_password_reset': False, +} +follower_notification_settings = { + "cursor": "-1", + "include_profile_interstitial_type": "1", + "include_blocking": "1", + "include_blocked_by": "1", + "include_followed_by": "1", + "include_want_retweets": "1", + "include_mute_edge": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_verified_type": "1", + "skip_status": "1", +} + +follow_settings = { + "include_profile_interstitial_type": "1", + "include_blocking": "1", + "include_blocked_by": "1", + "include_followed_by": "1", + "include_want_retweets": "1", + "include_mute_edge": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_verified_type": "1", + "skip_status": "1", +} + +account_search_settings = { + "optInFiltering": True, # filter out nsfw content + "optInBlocking": True, # filter out blocked accounts +} + +profile_settings = { + "birthdate_day": int, + "birthdate_month": int, + "birthdate_year": int, # 1985 + "birthdate_visibility": str, # 'self', + "birthdate_year_visibility": str, # 'self', + "displayNameMaxLength": int, # '50', + "url": str, # 'https://example.com', + "name": str, # 'foo', + "description": str, # 'bar', + "location": str, # 'world', +} + +search_config = { + "include_profile_interstitial_type": 1, + "include_blocking": 1, + "include_blocked_by": 1, + "include_followed_by": 1, + "include_want_retweets": 1, + "include_mute_edge": 1, + "include_can_dm": 1, + "include_can_media_tag": 1, + "include_ext_has_nft_avatar": 1, + "include_ext_is_blue_verified": 1, + "include_ext_verified_type": 1, + "skip_status": 1, + "cards_platform": "Web-12", + "include_cards": 1, + "include_ext_alt_text": "true", + "include_ext_limited_action_results": "false", + "include_quote_count": "true", + "include_reply_count": 1, + "tweet_mode": "extended", + "include_ext_collab_control": "true", + "include_ext_views": "true", + "include_entities": "true", + "include_user_entities": "true", + "include_ext_media_color": "true", + "include_ext_media_availability": "true", + "include_ext_sensitive_media_warning": "true", + "include_ext_trusted_friends_metadata": "true", + "send_error_codes": "true", + "simple_quoted_tweet": "true", + "query_source": "typed_query", + "count": 1000, + "q": "", + "requestContext": "launch", + "pc": 1, + "spelling_corrections": 1, + "include_ext_edit_control": "true", + "ext": "mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,enrichments,superFollowMetadata,unmentionInfo,editControl,collab_control,vibe", +} + +dm_params = { + "context": "FETCH_DM_CONVERSATION", + "include_profile_interstitial_type": "1", + "include_blocking": "1", + "include_blocked_by": "1", + "include_followed_by": "1", + "include_want_retweets": "1", + "include_mute_edge": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_verified_type": "1", + "include_ext_profile_image_shape": "1", + "skip_status": "1", + "dm_secret_conversations_enabled": "false", + "krs_registration_enabled": "true", + "cards_platform": "Web-12", + "include_cards": "1", + "include_ext_alt_text": "true", + "include_ext_limited_action_results": "false", + "include_quote_count": "true", + "include_reply_count": "1", + "tweet_mode": "extended", + "include_ext_views": "true", + "dm_users": "false", + "include_groups": "true", + "include_inbox_timelines": "true", + "include_ext_media_color": "true", + "supports_reactions": "true", + "include_conversation_info": "true", + "ext": "mediaColor,altText,mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,superFollowMetadata,unmentionInfo,editControl", +} + +live_notification_params = params = { + "cards_platform": "Web-12", + "count": "50", # max value + "ext": "mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,superFollowMetadata,unmentionInfo,editControl", + "include_blocked_by": "1", + "include_blocking": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_cards": "1", + "include_entities": "true", + "include_ext_alt_text": "true", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_limited_action_results": "true", + "include_ext_media_availability": "true", + "include_ext_media_color": "true", + "include_ext_profile_image_shape": "1", + "include_ext_sensitive_media_warning": "true", + "include_ext_trusted_friends_metadata": "true", + "include_ext_verified_type": "1", + "include_ext_views": "true", + "include_followed_by": "1", + "include_mute_edge": "1", + "include_profile_interstitial_type": "1", + "include_quote_count": "true", + "include_reply_count": "1", + "include_user_entities": "true", + "include_want_retweets": "1", + "send_error_codes": "true", + "simple_quoted_tweet": "true", + "skip_status": "1", + "tweet_mode": "extended", +} + +recommendations_params = { + "include_profile_interstitial_type": "1", + "include_blocking": "1", + "include_blocked_by": "1", + "include_followed_by": "1", + "include_want_retweets": "1", + "include_mute_edge": "1", + "include_can_dm": "1", + "include_can_media_tag": "1", + "include_ext_has_nft_avatar": "1", + "include_ext_is_blue_verified": "1", + "include_ext_verified_type": "1", + "include_ext_profile_image_shape": "1", + "skip_status": "1", + "pc": "true", + "display_location": "profile_accounts_sidebar", + "limit": 100, + "ext": "mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,superFollowMetadata,unmentionInfo,editControl", +} diff --git a/twitter_api/errors.py b/twitter_api/errors.py new file mode 100644 index 0000000..164289e --- /dev/null +++ b/twitter_api/errors.py @@ -0,0 +1,88 @@ +class TwitterError(Exception): + """Base class for Twitter errors""" + + def __init__(self, error_dict: dict): + self.error_dict = error_dict + + @property + def error_message(self) -> str: + if self.error_code == 32: + return "Failed to authenticate account. Check your credentials." + + elif self.error_code == 36: + return "You cannot use your own user ID to report spam call" + + elif self.error_code == 38: + return "The request is missing the parameter (such as media, text, etc.) in the request." + + elif self.error_code == 50: + return "User not found." + + elif self.error_code == 89: + return "The access token used in the request is incorrect or has expired." + + elif self.error_code == 92: + return "SSL is required. Only TLS v1.2 connections are allowed in the API. Update the request to a secure connection." + + elif self.error_code == 139: + return "You have already favorited this tweet. (Duplicate)" + + elif self.error_code == 160: + return "You've already requested to follow the user. (Duplicate)" + + elif self.error_code == 186: + return "Tweet needs to be a bit shorter. The text is too long." + + elif self.error_code == 187: + return "Text of your tweet is identical to another tweet. Change your text. (Duplicate)" + + elif self.error_code == 205: + return "The account limit for reporting spam has been reached. Try again later." + + elif self.error_code == 214: + return "Account is not set up to have open Direct Messages when trying to set up a welcome message." + + elif self.error_code == 220: + return "The authentication token in use is restricted and cannot access the requested resource." + + elif self.error_code == 323: + return "Only one animated GIF may be attached to a single Post." + + elif self.error_code == 325: + return "The media ID attached to the Post was not found." + + elif self.error_code == 327: + return "You cannot repost the same Post more than once." + + elif self.error_code == 349: + return "You does not have privileges to Direct Message the recipient." + + return self.error_dict.get("error_message") + + @property + def error_code(self) -> int: + return self.error_dict.get("error_code") + + +class TwitterAccountSuspended(Exception): + """Raised when account is suspended""" + + pass + + +class CaptchaError(Exception): + """Raised when captcha solving failed""" + + pass + + +class RateLimitError(Exception): + """Raised when rate limit exceeded""" + + pass + + +class IncorrectData(Exception): + """Raised when validation error""" + + pass diff --git a/twitter_api/models/__init__.py b/twitter_api/models/__init__.py new file mode 100644 index 0000000..e80df5c --- /dev/null +++ b/twitter_api/models/__init__.py @@ -0,0 +1,3 @@ +from .tweets import * +from .users import * +from .data import * diff --git a/twitter_api/models/data/__init__.py b/twitter_api/models/data/__init__.py new file mode 100644 index 0000000..f3e0a47 --- /dev/null +++ b/twitter_api/models/data/__init__.py @@ -0,0 +1,2 @@ +from .bind_account_v2 import * +from .bind_account_v1 import * diff --git a/twitter_api/models/data/bind_account_v1.py b/twitter_api/models/data/bind_account_v1.py new file mode 100644 index 0000000..ad86ea7 --- /dev/null +++ b/twitter_api/models/data/bind_account_v1.py @@ -0,0 +1,11 @@ +from pydantic import BaseModel, field_validator, HttpUrl + + +class BindAccountParamsV1(BaseModel): + url: HttpUrl + + +class BindAccountDataV1(BaseModel): + url: HttpUrl + oauth_token: str + oauth_verifier: str diff --git a/twitter_api/models/data/bind_account_v2.py b/twitter_api/models/data/bind_account_v2.py new file mode 100644 index 0000000..81c9c5a --- /dev/null +++ b/twitter_api/models/data/bind_account_v2.py @@ -0,0 +1,20 @@ +from pydantic import BaseModel, field_validator, HttpUrl + + +class BindAccountParamsV2(BaseModel): + code_challenge: str + code_challenge_method: str = "plain" + client_id: str + redirect_uri: HttpUrl + response_type: str = "code" + scope: str = "tweet.read users.read follows.read offline.access" + state: str + + @field_validator("redirect_uri", mode="after") + def validate_uri(cls, value): + # url = HttpUrl("https://google.com") + return str(value) + + +class BindAccountDataV2(BaseModel): + code: str diff --git a/twitter_api/models/tweets/__init__.py b/twitter_api/models/tweets/__init__.py new file mode 100644 index 0000000..1e869f4 --- /dev/null +++ b/twitter_api/models/tweets/__init__.py @@ -0,0 +1,14 @@ +from .create_tweet import * +from .delete_tweet import * +from .retweet import * +from .favorite_tweet import * +from .delete_retweet import * +from .delete_favorite_tweet import * +from .bookmark_tweet import * +from .unbookmark_tweet import * +from .create_reply import * +from .scrape_replies import * +from .scrape_favorites import * +from .scrape_retweets import * +from .create_schedule_tweet import * +from .delete_unschedule_tweet import * diff --git a/twitter_api/models/tweets/bookmark_tweet.py b/twitter_api/models/tweets/bookmark_tweet.py new file mode 100644 index 0000000..ee8118c --- /dev/null +++ b/twitter_api/models/tweets/bookmark_tweet.py @@ -0,0 +1,14 @@ +from pydantic import BaseModel +from typing import Optional, Dict + + +class CreateBookmarkData(BaseModel): + id: str | int + + +class CreateBookmarkResult(BaseModel): + tweet_bookmark_put: Optional[str] + + +class CreateBookmarkResultData(BaseModel): + data: Optional[CreateBookmarkResult] diff --git a/twitter_api/models/tweets/create_reply.py b/twitter_api/models/tweets/create_reply.py new file mode 100644 index 0000000..b502da5 --- /dev/null +++ b/twitter_api/models/tweets/create_reply.py @@ -0,0 +1,150 @@ +from pydantic import BaseModel +from typing import Optional, Dict, List, Any + +from .create_tweet import MediaEntity + + +class CreateReplyData(BaseModel): + id: str | int + text: str + media_entities: List[MediaEntity] | None = None + + +class Legacy(BaseModel): + can_dm: Optional[bool] + can_media_tag: Optional[bool] + created_at: Optional[str] + default_profile: Optional[bool] + default_profile_image: Optional[bool] + description: Optional[str] + entities: Optional[Dict[str, Any]] + fast_followers_count: Optional[int] + favourites_count: Optional[int] + followers_count: Optional[int] + friends_count: Optional[int] + has_custom_timelines: Optional[bool] + is_translator: Optional[bool] + listed_count: Optional[int] + location: Optional[str] + media_count: Optional[int] + name: Optional[str] + needs_phone_verification: Optional[bool] + normal_followers_count: Optional[int] + pinned_tweet_ids_str: Optional[List[str]] + possibly_sensitive: Optional[bool] + profile_image_url_https: Optional[str] + profile_interstitial_type: Optional[str] + screen_name: Optional[str] + statuses_count: Optional[int] + translator_type: Optional[str] + verified: Optional[bool] + want_retweets: Optional[bool] + withheld_in_countries: Optional[List[str]] + + +class UserResults(BaseModel): + __typename: Optional[str] + id: Optional[str] + rest_id: Optional[str] + affiliates_highlighted_label: Optional[Dict[str, Any]] + has_graduated_access: Optional[bool] + is_blue_verified: Optional[bool] + profile_image_shape: Optional[str] + legacy: Optional[Legacy] + smart_blocked_by: Optional[bool] + smart_blocking: Optional[bool] + + +class CoreUserResult(BaseModel): + result: Optional[UserResults] + + +class Core(BaseModel): + user_results: Optional[CoreUserResult] + + +class Views(BaseModel): + state: Optional[str] + + +class EditControl(BaseModel): + edit_tweet_ids: Optional[List[str]] + editable_until_msecs: Optional[str] + is_edit_eligible: Optional[bool] + edits_remaining: Optional[str] + + +class Media(BaseModel): + display_url: Optional[str] + expanded_url: Optional[str] + id_str: Optional[str] + indices: Optional[List[int]] + media_url_https: Optional[str] + type: Optional[str] + url: Optional[str] + features: Optional[Dict[str, Any]] + sizes: Optional[Dict[str, Any]] + original_info: Optional[Dict[str, int]] + + +class Entities(BaseModel): + user_mentions: Optional[List[Dict[str, Any]]] + urls: Optional[List[Any]] + hashtags: Optional[List[Any]] + symbols: Optional[List[Any]] + + +class ExtendedEntities(BaseModel): + media: Optional[List[Media]] + + +class Legacy2(BaseModel): + bookmark_count: Optional[int] + bookmarked: Optional[bool] + created_at: Optional[str] + conversation_id_str: Optional[str] + entities: Optional[Entities] + favorite_count: Optional[int] + favorited: Optional[bool] + full_text: Optional[str] + in_reply_to_screen_name: Optional[str] + in_reply_to_status_id_str: Optional[str] + in_reply_to_user_id_str: Optional[str] + is_quote_status: Optional[bool] + lang: Optional[str] + quote_count: Optional[int] + reply_count: Optional[int] + retweet_count: Optional[int] + retweeted: Optional[bool] + user_id_str: Optional[str] + id_str: Optional[str] + + +class UnmentionInfo(BaseModel): + pass + + +class CreateReplyResult(BaseModel): + rest_id: Optional[str] + has_birdwatch_notes: Optional[bool] + core: Optional[Core] + is_translatable: Optional[bool] + views: Optional[Views] + source: Optional[str] + legacy: Optional[Legacy2] + + +class CreateReplyResultDataV3(BaseModel): + result: Optional[CreateReplyResult] + + +class CreateReplyResultDataV2(BaseModel): + tweet_results: Optional[CreateReplyResultDataV3] + + +class CreateReplyResultDataV1(BaseModel): + create_tweet: Optional[CreateReplyResultDataV2] + + +class CreateReplyResultData(BaseModel): + data: Optional[CreateReplyResultDataV1] diff --git a/twitter_api/models/tweets/create_schedule_tweet.py b/twitter_api/models/tweets/create_schedule_tweet.py new file mode 100644 index 0000000..8e25478 --- /dev/null +++ b/twitter_api/models/tweets/create_schedule_tweet.py @@ -0,0 +1,22 @@ +from typing import List, Optional +from pydantic import BaseModel + +from .create_tweet import MediaEntity + + +class CreateScheduleTweetData(BaseModel): + text: str + date: int | str + media_entities: List[MediaEntity] | None = None + + +class CreateScheduleTweetResult(BaseModel): + rest_id: str + + +class CreateScheduleTweetResultDataV1(BaseModel): + tweet: Optional[CreateScheduleTweetResult] + + +class CreateScheduleTweetResultData(BaseModel): + data: Optional[CreateScheduleTweetResultDataV1] diff --git a/twitter_api/models/tweets/create_tweet.py b/twitter_api/models/tweets/create_tweet.py new file mode 100644 index 0000000..decd5e3 --- /dev/null +++ b/twitter_api/models/tweets/create_tweet.py @@ -0,0 +1,156 @@ +from typing import List, Optional, Any, Dict + +from pydantic import BaseModel, field_validator +from twitter_api.errors import IncorrectData + + +class MediaEntity(BaseModel): + media_id: int + tagged_users: List[str] | None = [] + + @field_validator("tagged_users") + @classmethod + def validate_users(cls, users: List[str]): + if users: + if len(users) > 10: + raise IncorrectData("Maximum 10 tagged users allowed") + + return users + + +class CreateTweetData(BaseModel): + text: str + media_entities: List[MediaEntity] | None = None + + +class Legacy(BaseModel): + can_dm: Optional[bool] + can_media_tag: Optional[bool] + created_at: Optional[str] + default_profile: Optional[bool] + default_profile_image: Optional[bool] + description: Optional[str] + entities: Optional[Dict[str, Any]] + fast_followers_count: Optional[int] + favourites_count: Optional[int] + followers_count: Optional[int] + friends_count: Optional[int] + has_custom_timelines: Optional[bool] + is_translator: Optional[bool] + listed_count: Optional[int] + location: Optional[str] + media_count: Optional[int] + name: Optional[str] + needs_phone_verification: Optional[bool] + normal_followers_count: Optional[int] + pinned_tweet_ids_str: Optional[List[str]] + possibly_sensitive: Optional[bool] + profile_image_url_https: Optional[str] + profile_interstitial_type: Optional[str] + screen_name: Optional[str] + statuses_count: Optional[int] + translator_type: Optional[str] + verified: Optional[bool] + want_retweets: Optional[bool] + withheld_in_countries: Optional[List[str]] + + +class UserResult(BaseModel): + __typename: Optional[str] + id: Optional[str] + rest_id: Optional[str] + affiliates_highlighted_label: Optional[Dict[str, Any]] + has_graduated_access: Optional[bool] + is_blue_verified: Optional[bool] + profile_image_shape: Optional[str] + legacy: Optional[Legacy] + smart_blocked_by: Optional[bool] + smart_blocking: Optional[bool] + + +class Result(BaseModel): + result: Optional[UserResult] + + +class Core(BaseModel): + user_results: Optional[Result] + + +class Views(BaseModel): + state: Optional[str] + + +class Entities(BaseModel): + user_mentions: Optional[List[Any]] + urls: Optional[List[Any]] + hashtags: Optional[List[Any]] + symbols: Optional[List[Any]] + + +class Legacy2(BaseModel): + bookmark_count: Optional[int] + bookmarked: Optional[bool] + created_at: Optional[str] + conversation_id_str: Optional[str] + display_text_range: Optional[List[int]] + entities: Optional[Entities] + favorite_count: Optional[int] + favorited: Optional[bool] + full_text: Optional[str] + is_quote_status: Optional[bool] + lang: Optional[str] + quote_count: Optional[int] + reply_count: Optional[int] + retweet_count: Optional[int] + retweeted: Optional[bool] + user_id_str: Optional[str] + id_str: Optional[str] + + +class EditControl(BaseModel): + edit_tweet_ids: Optional[List[str]] + editable_until_msecs: Optional[str] + is_edit_eligible: Optional[bool] + edits_remaining: Optional[str] + + +class QuickPromoteEligibility(BaseModel): + eligibility: Optional[str] + + +class UnmentionData(BaseModel): + pass + + +class UnmentionInfo(BaseModel): + pass + + +class CreateTweetResult(BaseModel): + rest_id: Optional[str] + has_birdwatch_notes: Optional[bool] + core: Optional[Core] + unmention_data: Optional[UnmentionData] + edit_control: Optional[EditControl] + is_translatable: Optional[bool] + views: Optional[Views] + source: Optional[str] + legacy: Optional[Legacy2] + quick_promote_eligibility: Optional[QuickPromoteEligibility] + unmention_info: Optional[UnmentionInfo] + + +class CreateTweetResultDataV3(BaseModel): + result: Optional[CreateTweetResult] + + +class CreateTweetResultDataV2(BaseModel): + tweet_results: Optional[CreateTweetResultDataV3] + + +class CreateTweetResultDataV1(BaseModel): + create_tweet: Optional[CreateTweetResultDataV2] + + +class CreateTweetResultData(BaseModel): + data: Optional[CreateTweetResultDataV1] diff --git a/twitter_api/models/tweets/delete_favorite_tweet.py b/twitter_api/models/tweets/delete_favorite_tweet.py new file mode 100644 index 0000000..ca941f1 --- /dev/null +++ b/twitter_api/models/tweets/delete_favorite_tweet.py @@ -0,0 +1,14 @@ +from pydantic import BaseModel +from typing import Optional + + +class DeleteFavoriteTweetData(BaseModel): + id: int | str + + +class DeleteFavoriteTweetResult(BaseModel): + unfavorite_tweet: Optional[str] + + +class DeleteFavoriteTweetResultData(BaseModel): + data: Optional[DeleteFavoriteTweetResult] diff --git a/twitter_api/models/tweets/delete_retweet.py b/twitter_api/models/tweets/delete_retweet.py new file mode 100644 index 0000000..57b4268 --- /dev/null +++ b/twitter_api/models/tweets/delete_retweet.py @@ -0,0 +1,31 @@ +from pydantic import BaseModel +from typing import Optional + + +class DeleteRetweetData(BaseModel): + id: int | str + + +class Legacy(BaseModel): + full_text: Optional[str] + + +class DeleteRetweetResult(BaseModel): + rest_id: Optional[str] + legacy: Optional[Legacy] + + +class DeleteRetweetResultDataV3(BaseModel): + result: Optional[DeleteRetweetResult] + + +class DeleteRetweetResultDataV2(BaseModel): + source_tweet_results: Optional[DeleteRetweetResultDataV3] + + +class DeleteRetweetResultDataV1(BaseModel): + unretweet: Optional[DeleteRetweetResultDataV2] + + +class DeleteRetweetResultData(BaseModel): + data: Optional[DeleteRetweetResultDataV1] diff --git a/twitter_api/models/tweets/delete_tweet.py b/twitter_api/models/tweets/delete_tweet.py new file mode 100644 index 0000000..4696fc5 --- /dev/null +++ b/twitter_api/models/tweets/delete_tweet.py @@ -0,0 +1,21 @@ +from pydantic import BaseModel + + +class DeleteTweetData(BaseModel): + id: int | str + + +class DeleteTweetResult(BaseModel): + pass + + +class DeleteTweetResultDataV2(BaseModel): + tweet_results: DeleteTweetResult + + +class DeleteTweetResultDataV1(BaseModel): + delete_tweet: DeleteTweetResultDataV2 + + +class DeleteTweetResultData(BaseModel): + data: DeleteTweetResultDataV1 diff --git a/twitter_api/models/tweets/delete_unschedule_tweet.py b/twitter_api/models/tweets/delete_unschedule_tweet.py new file mode 100644 index 0000000..7e3d7af --- /dev/null +++ b/twitter_api/models/tweets/delete_unschedule_tweet.py @@ -0,0 +1,15 @@ +from typing import Optional + +from pydantic import BaseModel + + +class DeleteScheduleTweetData(BaseModel): + id: str | int + + +class DeleteScheduleTweetResult(BaseModel): + scheduledtweet_delete: Optional[str] + + +class DeleteScheduleTweetResultData(BaseModel): + data: Optional[DeleteScheduleTweetResult] diff --git a/twitter_api/models/tweets/favorite_tweet.py b/twitter_api/models/tweets/favorite_tweet.py new file mode 100644 index 0000000..30d6d53 --- /dev/null +++ b/twitter_api/models/tweets/favorite_tweet.py @@ -0,0 +1,14 @@ +from pydantic import BaseModel +from typing import Optional + + +class CreateFavoriteTweetData(BaseModel): + id: int | str + + +class FavoriteTweetResult(BaseModel): + favorite_tweet: Optional[str] + + +class FavoriteTweetResultData(BaseModel): + data: Optional[FavoriteTweetResult] diff --git a/twitter_api/models/tweets/retweet.py b/twitter_api/models/tweets/retweet.py new file mode 100644 index 0000000..3f7bb55 --- /dev/null +++ b/twitter_api/models/tweets/retweet.py @@ -0,0 +1,32 @@ +from typing import Optional + +from pydantic import BaseModel + + +class CreateRetweetData(BaseModel): + id: int | str + + +class Legacy(BaseModel): + full_text: Optional[str] + + +class RetweetResult(BaseModel): + rest_id: Optional[str] + legacy: Optional[Legacy] + + +class RetweetResultDataV3(BaseModel): + result: Optional[RetweetResult] + + +class RetweetResultDataV2(BaseModel): + retweet_results: Optional[RetweetResultDataV3] + + +class RetweetResultDataV1(BaseModel): + create_retweet: Optional[RetweetResultDataV2] + + +class RetweetResultData(BaseModel): + data: Optional[RetweetResultDataV1] diff --git a/twitter_api/models/tweets/scrape_favorites.py b/twitter_api/models/tweets/scrape_favorites.py new file mode 100644 index 0000000..36d1b84 --- /dev/null +++ b/twitter_api/models/tweets/scrape_favorites.py @@ -0,0 +1,32 @@ +from pydantic import BaseModel, field_validator +from twitter_api.errors import IncorrectData + + +class ScrapeTweetFavoritesData(BaseModel): + id: int | str + limit: int = 200 + + @field_validator("limit") + @classmethod + def limit_must_be_positive(cls, v): + if v < 0: + raise IncorrectData("Limit must be positive integer") + + return v + + +class UserData(BaseModel): + id: int | str + name: str + screen_name: str + profile_image_url: str + favourites_count: int + followers_count: int + friends_count: int + location: str + description: str + created_at: str + + +class ScrapeTweetFavoritesResult(BaseModel): + users: list[UserData] diff --git a/twitter_api/models/tweets/scrape_replies.py b/twitter_api/models/tweets/scrape_replies.py new file mode 100644 index 0000000..e8d7254 --- /dev/null +++ b/twitter_api/models/tweets/scrape_replies.py @@ -0,0 +1,37 @@ +from pydantic import BaseModel, field_validator +from twitter_api.errors import IncorrectData + + +class ScrapeTweetRepliesData(BaseModel): + id: int | str + limit: int = 200 + + @field_validator("limit") + @classmethod + def limit_must_be_positive(cls, v): + if v < 0: + raise IncorrectData("Limit must be positive integer") + + return v + + +class UserData(BaseModel): + id: int | str + name: str + screen_name: str + profile_image_url: str + favourites_count: int + followers_count: int + friends_count: int + location: str + description: str + created_at: str + + +class ScrapeTweetRepliesResult(BaseModel): + reply_text: str + user_data: UserData + + +class ScrapeTweetRepliesResultData(BaseModel): + replies: list[ScrapeTweetRepliesResult] diff --git a/twitter_api/models/tweets/scrape_retweets.py b/twitter_api/models/tweets/scrape_retweets.py new file mode 100644 index 0000000..897a0e7 --- /dev/null +++ b/twitter_api/models/tweets/scrape_retweets.py @@ -0,0 +1,20 @@ +from pydantic import BaseModel, field_validator +from twitter_api.errors import IncorrectData +from .scrape_favorites import UserData + + +class ScrapeTweetRetweetsData(BaseModel): + id: int | str + limit: int = 200 + + @field_validator("limit") + @classmethod + def limit_must_be_positive(cls, v): + if v < 0: + raise IncorrectData("Limit must be positive integer") + + return v + + +class ScrapeTweetRetweetsResult(BaseModel): + users: list[UserData] diff --git a/twitter_api/models/tweets/unbookmark_tweet.py b/twitter_api/models/tweets/unbookmark_tweet.py new file mode 100644 index 0000000..c4425a5 --- /dev/null +++ b/twitter_api/models/tweets/unbookmark_tweet.py @@ -0,0 +1,14 @@ +from pydantic import BaseModel +from typing import Optional, Dict + + +class DeleteBookmarkData(BaseModel): + id: str | int + + +class DeleteBookmarkResult(BaseModel): + tweet_bookmark_delete: Optional[str] + + +class DeleteBookmarkResultData(BaseModel): + data: Optional[DeleteBookmarkResult] diff --git a/twitter_api/models/users/__init__.py b/twitter_api/models/users/__init__.py new file mode 100644 index 0000000..4edec9c --- /dev/null +++ b/twitter_api/models/users/__init__.py @@ -0,0 +1,3 @@ +from .follows import * +from .user_info import * +from .followers import * diff --git a/twitter_api/models/users/followers.py b/twitter_api/models/users/followers.py new file mode 100644 index 0000000..def57ed --- /dev/null +++ b/twitter_api/models/users/followers.py @@ -0,0 +1,16 @@ +from pydantic import BaseModel, field_validator +from typing import Optional, List, Dict, Any +from twitter_api.errors import IncorrectData + + +class UserFollowersData(BaseModel): + username: str + limit: int = 200 + + @field_validator("limit") + @classmethod + def limit_must_be_positive(cls, v): + if v < 0: + raise IncorrectData("Limit must be positive integer") + + return v diff --git a/twitter_api/models/users/follows.py b/twitter_api/models/users/follows.py new file mode 100644 index 0000000..bcf448e --- /dev/null +++ b/twitter_api/models/users/follows.py @@ -0,0 +1,70 @@ +from typing import Optional, List + +from pydantic import BaseModel, model_validator +from twitter_api.errors import IncorrectData + + +class UnfollowUserData(BaseModel): + id: int | str = None + username: str = None + + @model_validator(mode="before") + @classmethod + def validate_data(cls, values: dict): + if not values.get("id") and not values.get("username"): + raise IncorrectData("Either id or username must be provided") + + return values + + +class FollowUserData(BaseModel): + id: str | int = None + username: str = None + + @model_validator(mode="before") + @classmethod + def validate_data(cls, values: dict): + if not values.get("id") and not values.get("username"): + raise IncorrectData("Either id or username must be provided") + + return values + + +class FollowsUserResult(BaseModel): + id: Optional[int] + id_str: Optional[str] + name: Optional[str] + screen_name: Optional[str] + location: Optional[str] + description: Optional[str] + url: Optional[str] + protected: Optional[bool] + followers_count: Optional[int] + fast_followers_count: Optional[int] + normal_followers_count: Optional[int] + friends_count: Optional[int] + listed_count: Optional[int] + created_at: Optional[str] + favourites_count: Optional[int] + utc_offset: Optional[int] + time_zone: Optional[str] + geo_enabled: Optional[bool] + verified: Optional[bool] + statuses_count: Optional[int] + media_count: Optional[int] + lang: Optional[str] + profile_image_url: Optional[str] + profile_image_url_https: Optional[str] + profile_banner_url: Optional[str] + pinned_tweet_ids: Optional[List[int]] + pinned_tweet_ids_str: Optional[List[str]] + has_custom_timelines: Optional[bool] + can_dm: Optional[bool] + can_media_tag: Optional[bool] + following: Optional[bool] + follow_request_sent: Optional[bool] + blocking: Optional[bool] + business_profile_state: Optional[str] + followed_by: Optional[bool] + ext_is_blue_verified: Optional[bool] + ext_has_nft_avatar: Optional[bool] diff --git a/twitter_api/models/users/user_info.py b/twitter_api/models/users/user_info.py new file mode 100644 index 0000000..0e7c4d7 --- /dev/null +++ b/twitter_api/models/users/user_info.py @@ -0,0 +1,66 @@ +from typing import Optional, List, Dict, Any +from pydantic import BaseModel + + +class UserProfileInfoData(BaseModel): + username: str + + +class UserProfileInfoResult(BaseModel): + id: Optional[int] + id_str: Optional[str] + name: Optional[str] + screen_name: Optional[str] + location: Optional[str] + profile_location: Optional[Dict[str, Any]] + description: Optional[str] + url: Optional[str] + # entities: Optional[Entities] + protected: Optional[bool] + followers_count: Optional[int] + fast_followers_count: Optional[int] + normal_followers_count: Optional[int] + friends_count: Optional[int] + listed_count: Optional[int] + created_at: Optional[str] + favourites_count: Optional[int] + utc_offset: Optional[int] + time_zone: Optional[str] + geo_enabled: Optional[bool] + verified: Optional[bool] + statuses_count: Optional[int] + media_count: Optional[int] + lang: Optional[str] + # status: Optional[Status] + contributors_enabled: Optional[bool] + is_translator: Optional[bool] + is_translation_enabled: Optional[bool] + profile_background_color: Optional[str] + profile_background_image_url: Optional[str] + profile_background_image_url_https: Optional[str] + profile_background_tile: Optional[bool] + profile_image_url: Optional[str] + profile_image_url_https: Optional[str] + profile_banner_url: Optional[str] + profile_link_color: Optional[str] + profile_sidebar_border_color: Optional[str] + profile_sidebar_fill_color: Optional[str] + profile_text_color: Optional[str] + profile_use_background_image: Optional[bool] + has_extended_profile: Optional[bool] + default_profile: Optional[bool] + default_profile_image: Optional[bool] + pinned_tweet_ids: Optional[List[int]] + pinned_tweet_ids_str: Optional[List[str]] + has_custom_timelines: Optional[bool] + can_media_tag: Optional[bool] + followed_by: Optional[bool] + following: Optional[bool] + follow_request_sent: Optional[bool] + notifications: Optional[bool] + advertiser_account_type: Optional[str] + advertiser_account_service_levels: Optional[List[str]] + business_profile_state: Optional[str] + translator_type: Optional[str] + withheld_in_countries: Optional[List[str]] + require_some_consent: Optional[bool] diff --git a/twitter_api/requirements.txt b/twitter_api/requirements.txt new file mode 100644 index 0000000..260dd0f --- /dev/null +++ b/twitter_api/requirements.txt @@ -0,0 +1,5 @@ +pydantic +orjson +httpx +curl-cffi +tqdm \ No newline at end of file diff --git a/twitter_api/util.py b/twitter_api/util.py new file mode 100644 index 0000000..861f8fc --- /dev/null +++ b/twitter_api/util.py @@ -0,0 +1,296 @@ +import random +import re +import string +import time +import orjson + +from logging import Logger +from pathlib import Path +from urllib.parse import urlsplit, urlencode, urlunsplit, parse_qs, quote +from httpx import Response, Client + +from .constants import GREEN, MAGENTA, RED, RESET, ID_MAP +from .errors import TwitterError + + +def init_session(): + client = Client( + headers={ + "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36", + }, + follow_redirects=True, + ) + r = client.post("https://api.twitter.com/1.1/guest/activate.json").json() + client.headers.update( + { + "content-type": "application/json", + "x-guest-token": r["guest_token"], + "x-twitter-active-user": "yes", + } + ) + return client + + +def batch_ids(ids: list[int], char_limit: int = 4_500) -> list[dict]: + """To avoid 431 errors""" + length = 0 + res, batch = [], [] + for x in map(str, ids): + curr_length = len(x) + if length + curr_length > char_limit: + res.append(batch) + batch = [] + length = 0 + batch.append(x) + length += curr_length + if batch: + res.append(batch) + return res + + +def build_params(params: dict) -> dict: + return {k: orjson.dumps(v).decode() for k, v in params.items()} + + +def save_json(r: Response, path: Path, name: str, **kwargs): + try: + data = r.json() + kwargs.pop("cursor", None) + out = path / "_".join(map(str, kwargs.values())) + out.mkdir(parents=True, exist_ok=True) + (out / f"{time.time_ns()}_{name}.json").write_bytes(orjson.dumps(data)) + except Exception as e: + print(f"Failed to save data: {e}") + + +def flatten(seq: list | tuple) -> list: + flat = [] + for e in seq: + if isinstance(e, list | tuple): + flat.extend(flatten(e)) + else: + flat.append(e) + return flat + + +def get_json(res: list[Response], **kwargs) -> list: + cursor = kwargs.get("cursor") + temp = res + if any(isinstance(r, (list, tuple)) for r in res): + temp = flatten(res) + results = [] + for r in temp: + try: + data = r.json() + if cursor: + results.append([data, cursor]) + else: + results.append(data) + except Exception as e: + print("Cannot parse JSON response", e) + return results + + +def set_qs(url: str, qs: dict, update=False, **kwargs) -> str: + *_, q, f = urlsplit(url) + return urlunsplit( + ( + *_, + urlencode( + qs | parse_qs(q) if update else qs, + doseq=True, + quote_via=quote, + safe=kwargs.get("safe", ""), + ), + f, + ) + ) + + +def get_cursor(data: list | dict) -> str: + # inefficient, but need to deal with arbitrary schema + entries = find_key(data, "entries") + if entries: + for entry in entries.pop(): + entry_id = entry.get("entryId", "") + if ("cursor-bottom" in entry_id) or ("cursor-showmorethreads" in entry_id): + content = entry["content"] + if itemContent := content.get("itemContent"): + return itemContent["value"] # v2 cursor + return content["value"] # v1 cursor + + +def get_headers(session, **kwargs) -> dict: + """ + Get the headers required for authenticated requests + """ + cookies = session.cookies + headers = kwargs | { + "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA", + # "cookie": "; ".join(f"{k}={v}" for k, v in cookies.items()), + "referer": "https://twitter.com/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36", + "x-csrf-token": cookies.get("ct0", ""), + # "x-guest-token": cookies.get("guest_token", ""), + "x-twitter-auth-type": "OAuth2Session" if cookies.get("auth_token") else "", + "x-twitter-active-user": "yes", + "x-twitter-client-language": "en", + } + return dict(sorted({k.lower(): v for k, v in headers.items()}.items())) + + +def find_key(obj: any, key: str) -> list: + """ + Find all values of a given key within a nested dict or list of dicts + + Most data of interest is nested, and sometimes defined by different schemas. + It is not worth our time to enumerate all absolute paths to a given key, then update + the paths in our parsing functions every time Twitter changes their API. + Instead, we recursively search for the key here, then run post-processing functions on the results. + + @param obj: dictionary or list of dictionaries + @param key: key to search for + @return: list of values + """ + + def helper(obj: any, key: str, L: list) -> list: + if not obj: + return L + + if isinstance(obj, list): + for e in obj: + L.extend(helper(e, key, [])) + return L + + if isinstance(obj, dict) and obj.get(key): + L.append(obj[key]) + + if isinstance(obj, dict) and obj: + for k in obj: + L.extend(helper(obj[k], key, [])) + return L + + return helper(obj, key, []) + + +def log(logger: Logger, level: int, r: Response): + def stat(r, txt, data): + if level >= 1: + logger.debug(f"{r.url.path}") + if level >= 2: + logger.debug(f"{r.url}") + if level >= 3: + logger.debug(f"{txt}") + if level >= 4: + logger.debug(f"{data}") + + try: + limits = {k: v for k, v in r.headers.items() if "x-rate-limit" in k} + current_time = int(time.time()) + wait = int(r.headers.get("x-rate-limit-reset", current_time)) - current_time + remaining = limits.get("x-rate-limit-remaining") + limit = limits.get("x-rate-limit-limit") + logger.debug(f"remaining: {MAGENTA}{remaining}/{limit}{RESET} requests") + logger.debug(f"reset: {MAGENTA}{(wait / 60):.2f}{RESET} minutes") + except Exception as e: + logger.error(f"Rate limit info unavailable: {e}") + + try: + status = r.status_code + ( + txt, + data, + ) = ( + r.text, + r.json(), + ) + if "json" in r.headers.get("content-type", ""): + if data.get("errors") and not find_key(data, "instructions"): + logger.error(f"[{RED}error{RESET}] {status} {data}") + else: + logger.debug(fmt_status(status)) + stat(r, txt, data) + else: + logger.debug(fmt_status(status)) + stat(r, txt, {}) + except Exception as e: + logger.error(f"Failed to log: {e}") + + +def fmt_status(status: int) -> str: + color = None + if 200 <= status < 300: + color = GREEN + elif 300 <= status < 400: + color = MAGENTA + elif 400 <= status < 600: + color = RED + return f"[{color}{status}{RESET}]" + + +def get_ids(data: list | dict, operation: tuple) -> set: + expr = ID_MAP[operation[-1]] + return {k for k in find_key(data, "entryId") if re.search(expr, k)} + + +def dump(path: str, **kwargs): + fname, data = list(kwargs.items())[0] + out = Path(path) + out.mkdir(exist_ok=True, parents=True) + (out / f"{fname}_{time.time_ns()}.json").write_bytes( + orjson.dumps(data, option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS) + ) + + +def get_code(cls, retries=5) -> str | None: + """Get verification code from Proton Mail inbox""" + + def poll_inbox(): + inbox = cls.inbox() + for c in inbox.get("Conversations", []): + if c["Senders"][0]["Address"] == "info@twitter.com": + exprs = [ + "Your Twitter confirmation code is (.+)", + "(.+) is your Twitter verification code", + ] + if temp := list( + filter(None, (re.search(expr, c["Subject"]) for expr in exprs)) + ): + return temp[0].group(1) + + for i in range(retries + 1): + if code := poll_inbox(): + return code + if i == retries: + print(f"Max retries exceeded") + return + t = 2**i + random.random() + print(f'Retrying in {f"{t:.2f}"} seconds') + time.sleep(t) + + +def get_random_string(len_: int) -> str: + return "".join( + random.choice(string.ascii_lowercase + string.digits) for _ in range(len_) + ) + + +def get_random_number(len_: int) -> str: + return "".join(random.choice(string.digits) for _ in range(len_)) + + +def generate_random_string() -> str: + return "".join([random.choice(string.ascii_letters + "-_") for _ in range(352)]) + + +def raise_for_status(response): + http_error_msg = "" + if 400 <= response.status_code < 500: + http_error_msg = f"{response.status_code} Client Error for url {response.url}" + + elif 500 <= response.status_code < 600: + http_error_msg = f"{response.status_code} Server Error for url: {response.url}" + + if http_error_msg: + raise TwitterError({"error_message": http_error_msg}) diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..15b6a64 --- /dev/null +++ b/utils/__init__.py @@ -0,0 +1 @@ +from .main import * diff --git a/utils/main.py b/utils/main.py new file mode 100644 index 0000000..a9002e9 --- /dev/null +++ b/utils/main.py @@ -0,0 +1,22 @@ +from art import tprint + + +def show_dev_info(): + tprint("JamBit") + print("\033[36m" + "VERSION: " + "\033[34m" + "1.0" + "\033[34m") + print("\033[36m" + "Channel: " + "\033[34m" + "https://t.me/JamBitPY" + "\033[34m") + print( + "\033[36m" + + "GitHub: " + + "\033[34m" + + "https://github.com/Jaammerr" + + "\033[34m" + ) + print( + "\033[36m" + + "DONATION EVM ADDRESS: " + + "\033[34m" + + "0x08e3fdbb830ee591c0533C5E58f937D312b07198" + + "\033[0m" + ) + print()