Open Source Twitter X Account Checker 2025

import asyncio
import curl_cffi
from curl_cffi.requests import AsyncSession
from rich import print
from time import perf_counter
import random
import aiofiles
import psutil
from util.const import *
from util.helpers import load_tokens, banner, check_completed, cleanup_files
from rich.progress import Progress
import os
import warnings
import orjson

warnings.filterwarnings("ignore")
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


total_valid = 0
total_dead = 0
total_locked = 0
total_consent = 0
total_suspended = 0
write_lock = asyncio.Lock()
count_semaphore = asyncio.Semaphore(1)


async def validity(auth_token, ct0=None, extra=None):
    global total_valid, total_dead, total_locked, total_consent, total_suspended
    cookies = {'auth_token': auth_token}

    if ct0 and not CT0_FIX:
        cookies['ct0'] = ct0
        PROFILE_HEADERS['x-csrf-token'] = ct0

    async with AsyncSession(proxy=PROXY, impersonate=random.choice(["chrome124", "chrome123"]), timeout=10, max_clients=1) as client:
        while True:
            try:
                if not ct0 or CT0_FIX:
                    response = await client.head('https://business.x.com/', headers=PROFILE_HEADERS, cookies=cookies)

                    ct0_value = response.cookies.get('ct0')

                    if ct0_value:
                        cookies['ct0'] = ct0_value
                        PROFILE_HEADERS['x-csrf-token'] = ct0_value
                        if CT0_FIX:
                            ct0 = ct0_value
                    else:
                        continue

                response = await client.post('https://api.x.com/1.1/account/update_profile.json', headers=PROFILE_HEADERS, cookies=cookies)

                response_json = orjson.loads(response.content)

                error_info = response_json.get('errors', [{}])[0]
                error_code = error_info.get('code', response.status_code)
                bounce_location = error_info.get('bounce_location', '')

                status_map = {
                    326: ("bold steel_blue1", "LOCKED") if bounce_location != '/i/flow/consent_flow' else ("bold yellow", "CONSENT"),
                    200: ("bold chartreuse1", "VALID"),
                    32: ("red", "DEAD"),
                    64: ("bold red3", "SUSPENDED"),
                }

                status_color, status_text = status_map.get(
                    error_code, ("bold red", "UNKNOWN"))

                # Check if the status is valid
                if status_text == "VALID":
                    followers_count = response_json.get('followers_count', 0)
                elif status_text in ("DEAD", "SUSPENDED", "LOCKED", "CONSENT"):
                    followers_count = 0
                else:
                    continue

                composed_token = ':'.join(
                    str(item)
                    for sublist in [extra, ct0, auth_token]
                    for item in (sublist if isinstance(sublist, list) else [sublist])
                    if item is not None
                )

                if followers_count >= MIN_THRESHOLD:
                    output_file = next(
                        (f'output/{name}.txt' for threshold, name in THRESHOLDS.items() if followers_count >= threshold))
                    if SAVE_FOLLOWER_COUNT:
                        composed_token += f":{followers_count}\n"
                    else:
                        composed_token += "\n"
                else:
                    composed_token += "\n"
                    output_file = f'output/{status_text.lower()}.txt'

                async with write_lock:
                    async with aiofiles.open(output_file, mode='a', encoding="latin-1") as f:
                        try:
                            await f.write(composed_token)
                            await f.flush()
                        except Exception as e:
                            print(f"Error writing to {output_file}: {e}")
                if UPDATE_CONSOLE:
                    print(f"[{status_color}][[bold white]*[{status_color}]] [{status_color}]{
                          auth_token} [bold white][[{status_color}]{status_text}[bold white]] | Followers: {followers_count:,}")

                return status_text
            except (curl_cffi.CurlError, curl_cffi.CurlECode, Exception) as e:
                print(e)
                return None


async def get_network_usage():
    network_stats = psutil.net_io_counters()
    return network_stats.bytes_sent, network_stats.bytes_recv