main.py

import argparse
import logging
import sys

from core.config import ProxyConfig
from core.http_client import HTTPClient
from core.login_flow import LoginFlowOrchestrator


def setup_logging(debug: bool = False) -> logging.Logger:
    log_level = logging.DEBUG if debug else logging.INFO
    logging.basicConfig(
        level=log_level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[logging.StreamHandler(sys.stdout)],
    )
    return logging.getLogger(__name__)


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="X.com Login Flow - Multi-step authentication with Castle token support",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python main.py elonmuskcr
  python main.py user@example.com --proxy http://127.0.0.1:8080
  python main.py user@example.com --debug
  python main.py elonmuskcr --test
  python main.py elonmuskcr --token abc123 --cuid 10deead773b19e5c033b6147b1c36da1
        """,
    )

    parser.add_argument(
        "username",
        type=str,
        help="Username or email address to authenticate with",
    )
    parser.add_argument(
        "--proxy",
        type=str,
        default=None,
        help="HTTP/HTTPS proxy URL (e.g., http://user:pass@host:port). Falls back to HTTP_PROXY env var.",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Enable debug-level logging",
    )
    parser.add_argument(
        "--api-key",
        type=str,
        default="",
        help="Optional API key for Castle token generation",
    )
    parser.add_argument(
        "--version",
        type=int,
        default=None,
        help="Browser version for test mode (default: 145)",
    )
    parser.add_argument(
        "--no-transaction",
        action="store_true",
        default=False,
        help="Disable X-Client-Transaction-Id generation",
    )
    parser.add_argument(
        "--test",
        action="store_true",
        default=False,
        help="Interactive single login test (prompts for token and cuid)",
    )
    parser.add_argument(
        "--token",
        type=str,
        default=None,
        help="Pre-generated Castle token to use directly without prompting",
    )
    parser.add_argument(
        "--cuid",
        type=str,
        default=None,
        help="Pre-generated CUID to use directly without prompting",
    )

    return parser.parse_args()


def create_proxy_config(proxy_url: str, logger: logging.Logger) -> ProxyConfig:
    proxy_config = ProxyConfig()
    if not proxy_url:
        return proxy_config

    proxy_parts = proxy_url.split("://", 1)
    if len(proxy_parts) == 2:
        scheme, _ = proxy_parts
        if scheme.lower() in ("http", "https"):
            proxy_config.http = proxy_url
            proxy_config.https = proxy_url
        else:
            logger.warning(f"Unknown proxy scheme: {scheme}")
    else:
        logger.warning("Invalid proxy format, expected scheme://host:port")

    logger.info(f"Using proxy: {proxy_url}")
    return proxy_config


def apply_test_browser_profile(http_client: HTTPClient, version: int) -> None:
    http_client.browser_info.browser_type = "chrome"
    http_client.browser_info.version = version
    http_client.browser_info.platform = "Windows"


def resolve_test_credentials(args: argparse.Namespace, logger: logging.Logger) -> tuple[str, str] | None:
    token = args.token.strip() if args.token else ""
    cuid = args.cuid.strip() if args.cuid else ""

    if token or cuid:
        if not token or not cuid:
            logger.error("--token and --cuid must be provided together.")
            return None
        return token, cuid

    if not args.test:
        return None

    token = input("Token: ").strip()
    cuid = input("CUID: ").strip()

    if not token or not cuid:
        logger.error("Token and CUID are both required.")
        return None

    return token, cuid


def main() -> int:
    args = parse_arguments()
    logger = setup_logging(debug=args.debug)

    try:
        proxy_config = create_proxy_config(args.proxy, logger)
        test_credentials = resolve_test_credentials(args, logger)
        use_test_mode = args.test or test_credentials is not None
        enable_transaction = (not args.no_transaction) if use_test_mode else False

        if use_test_mode:
            logger.info("=" * 70)
            logger.info(
                "Single Login Test (%s mode)",
                "interactive" if args.test and args.token is None and args.cuid is None else "argument",
            )
            logger.info("=" * 70)

            if test_credentials is None:
                return 1
            token, cuid = test_credentials

            test_version = args.version or 145
            http_client = HTTPClient(
                client_identifier="chrome_120",
                proxy_config=proxy_config,
                browser_mode="static",
                browser_version=test_version,
            )
            apply_test_browser_profile(http_client, test_version)

            logger.info(f"  - Platform: {http_client.browser_info.platform}")
            logger.info(f"  - Browser: {http_client.browser_info.browser_type}")
            logger.info(f"  - Version: {http_client.browser_info.version}")
            logger.info(f"  - User-Agent: {http_client.browser_info.user_agent}")
            logger.info(f"  - sec-ch-ua: {http_client.browser_info.sec_ch_ua}")
            logger.info(f"  - Token: {token[:60]}...")
            logger.info(f"  - CUID: {cuid}")
            logger.info(f"  - Transaction ID: {'enabled' if enable_transaction else 'disabled'}")

            login_flow = LoginFlowOrchestrator(
                http_client,
                enable_client_transaction=enable_transaction,
                castle_token_override=token,
                cuid_override=cuid,
            )

            if args.api_key:
                login_flow.castle_generator.api_key = args.api_key
                logger.info("Using custom API key for Castle token generation")

            result = login_flow.execute_login_flow(args.username)
            success, response_code = result if isinstance(result, tuple) else (result, 0)

            if success:
                logger.info("Registration-style test PASSED")
                return 0

            logger.error(f"Registration-style test FAILED (status: {response_code})")
            return 1

        logger.info("=" * 70)
        logger.info("X.com Login Flow - Multi-step Authentication")
        logger.info("=" * 70)
        logger.info(f"Transaction ID: {'enabled' if enable_transaction else 'disabled'}")
        if args.version:
            logger.info(f"Forced browser version: {args.version}")

        logger.info("Initializing HTTP client with TLS fingerprinting...")
        http_client = HTTPClient(
            client_identifier="chrome_120",
            proxy_config=proxy_config,
            browser_mode="random",
            browser_version=args.version,
        )

        logger.info("Initializing login flow orchestrator...")
        login_flow = LoginFlowOrchestrator(
            http_client,
            enable_client_transaction=enable_transaction,
        )

        if args.api_key:
            login_flow.castle_generator.api_key = args.api_key
            logger.info("Using custom API key for Castle token generation")

        result = login_flow.execute_login_flow(args.username)
        success, response_code = result if isinstance(result, tuple) else (result, 0)

        if success:
            logger.info("Authentication flow completed successfully")
            return 0

        logger.error(f"Authentication flow failed (status: {response_code})")
        return 1

    except KeyboardInterrupt:
        logger.warning("\nOperation cancelled by user")
        return 1
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        return 1


if __name__ == "__main__":
    sys.exit(main())