import asyncio import random from pathlib import Path from typing import Optional, List, Union, Dict, Tuple from tweety import TwitterAsync from tweety.types import Proxy from cookie_parser import read_cookies_file, clean_cookie_content import debug_logger as log async def authenticate_twitter(cookies: Union[str, Dict] = None, auth_token: str = None, proxy: Optional[Proxy] = None) -> TwitterAsync: log.debug(f"authenticate_twitter called - cookies: {bool(cookies)}, auth_token: {bool(auth_token)}, proxy: {bool(proxy)}") try: if proxy: log.debug(f"Creating TwitterAsync with proxy: {proxy.host}:{proxy.port}") app = TwitterAsync("temp_session", proxy=proxy) else: log.debug("Creating TwitterAsync without proxy") app = TwitterAsync("temp_session") if auth_token: log.debug(f"Loading auth_token: {auth_token[:20]}...") await app.load_auth_token(auth_token) elif cookies: log.debug(f"Loading cookies (length: {len(str(cookies))})") log.debug(f"Cookies preview: {str(cookies)[:200]}...") await app.load_cookies(cookies) log.debug("Authentication object created successfully") return app except Exception as e: log.error(f"Error in authenticate_twitter: {e}") log.exception("Full traceback:") raise async def check_account(cookie_file: Path, proxies: List[Proxy], output_dir: Path, semaphore: asyncio.Semaphore, cookies_only_mode: bool = False, user_data_list=None) -> Tuple[bool, bool, str]: async with semaphore: log.info(f"=== Checking account: {cookie_file.name} ===") try: try: log.debug(f"Reading cookie file: {cookie_file}") cookies_string = read_cookies_file(str(cookie_file)) if not cookies_string: log.error(f"Failed to parse cookie file: {cookie_file.name}") return (False, False, "Failed to parse cookie file") log.debug(f"Cookies parsed successfully, length: {len(cookies_string)}") cleaned_content = clean_cookie_content(cookie_file.read_text(encoding='utf-8')) log.debug(f"Cleaned content length: {len(cleaned_content)}") except Exception as e: log.error(f"Failed to read cookies from {cookie_file.name}: {e}") log.exception("Cookie read exception:") return (False, False, f"Failed to read cookies: {e}") proxy = random.choice(proxies) if proxies else None if proxy: log.debug(f"Using proxy: {proxy.host}:{proxy.port}") else: log.debug("No proxy being used") try: log.debug("Attempting to authenticate...") app = await authenticate_twitter(cookies=cookies_string, proxy=proxy) log.debug("Authentication completed") except Exception as e: log.error(f"Authentication error for {cookie_file.name}: {e}") log.exception("Authentication exception:") return (False, False, f"Authentication error: {e}") try: log.debug("Checking app.me...") if app.me: user = app.me log.info(f"✓ Successfully authenticated as @{user.username}") log.debug(f"User details - Followers: {user.followers_count}, Verified: {user.verified}") is_verified = user.verified try: from file_utils import save_account_info from main import config save_format = config.get('save_format', 'auth_token') log.debug(f"Saving account info in {save_format} format") save_account_info(user, cookie_file, output_dir, cleaned_content, cookies_only_mode, save_format, user_data_list) log.debug("Account info saved successfully") except Exception as e: log.error(f"Failed to save account info: {e}") log.exception("Save exception:") pass return (True, is_verified, f"@{user.username}") else: log.error(f"app.me is None for {cookie_file.name}") return (False, False, "Authentication failed - invalid cookies") except Exception as e: log.error(f"Error processing user info for {cookie_file.name}: {e}") log.exception("User info exception:") return (False, False, f"Error processing user info: {e}") except Exception as e: log.error(f"Unexpected error for {cookie_file.name}: {e}") log.exception("Unexpected exception:") return (False, False, f"Unexpected error: {e}") async def check_account_by_token(auth_token: str, token_file: Path, token_line_num: int, proxies: List[Proxy], output_dir: Path, semaphore: asyncio.Semaphore, cookies_only_mode: bool = False, user_data_list=None) -> Tuple[bool, bool, str]: async with semaphore: try: token_source = f"{token_file.name}:line-{token_line_num}" proxy = random.choice(proxies) if proxies else None try: app = await authenticate_twitter(auth_token=auth_token, proxy=proxy) except Exception as e: return (False, False, f"Authentication error: {e}") try: if app.me: user = app.me is_verified = user.verified try: from file_utils import save_account_info_token from main import config save_format = config.get('save_format', 'auth_token') save_account_info_token(user, auth_token, token_source, output_dir, cookies_only_mode, save_format, user_data_list) except Exception: pass return (True, is_verified, f"@{user.username}") else: return (False, False, "Authentication failed - invalid token") except Exception as e: return (False, False, f"Error processing user info: {e}") except Exception as e: return (False, False, f"Unexpected error: {e}")