Proxy rotation

class ProxyManager:
    """Thread-safe round-robin proxy manager.
    
    Loads proxies from a file and rotates through them for each request.
    Optionally includes direct host connection (None) in rotation.
    """
    
    _instance: Optional[ProxyManager] = None
    _lock = threading.Lock()
    
    def __init__(self, proxy_file: str, include_host: bool = False):
        self._proxies: list[str | None] = []
        self._index = 0
        self._rotation_lock = threading.Lock()
        self._load_proxies(proxy_file, include_host)

from tiktok_api import ProxyManager

# In main() function
if config["proxy"]["proxy_file"]:
    ProxyManager.initialize(
        proxy_file=config["proxy"]["proxy_file"],
        include_host=config["proxy"]["include_host"],
    )
    logging.info("Proxy manager initialized")

# Path to proxy file (one proxy per line)
PROXY_FILE=proxies.txt

# Include direct connection in rotation (true/false)
INCLUDE_HOST=false

# Production proxies (US)
http://user:pass@proxy1.example.com:8080
http://user:pass@proxy2.example.com:8080

# Backup proxies (EU)
http://user:pass@proxy3.example.com:8080

# SOCKS5 proxy
socks5://user:pass@proxy4.example.com:1080

# Empty lines and comments are ignored

def _encode_proxy_auth(self, proxy_url: str) -> str:
    """URL-encode username and password in proxy URL."""
    match = re.match(r"^(https?|socks5)://([^:@]+):([^@]+)@(.+)$", proxy_url)
    if match:
        protocol, username, password, host_port = match.groups()
        encoded_username = quote(username, safe="")
        encoded_password = quote(password, safe="")
        return f"{protocol}://{encoded_username}:{encoded_password}@{host_port}"
    return proxy_url

def get_next_proxy(self) -> str | None:
    """Get next proxy in round-robin rotation.
    
    Returns:
        Proxy URL string, or None for direct connection.
    """
    with self._rotation_lock:
        if not self._proxies:
            return None
        proxy = self._proxies[self._index]
        self._index = (self._index + 1) % len(self._proxies)
        return proxy

http://proxy1:8080
http://proxy2:8080
http://proxy3:8080

manager = ProxyManager("proxies.txt", include_host=True)

manager.get_next_proxy()  # Returns: "http://proxy1:8080"
manager.get_next_proxy()  # Returns: "http://proxy2:8080"
manager.get_next_proxy()  # Returns: "http://proxy3:8080"
manager.get_next_proxy()  # Returns: None (direct connection)
manager.get_next_proxy()  # Returns: "http://proxy1:8080" (wraps around)

from tiktok_api import TikTokClient, ProxyManager

proxy_manager = ProxyManager.get_instance()
client = TikTokClient(
    proxy_manager=proxy_manager,
    data_only_proxy=False,  # Use proxies for both API and media
)

async def video(self, video_link: str) -> VideoInfo:
    # Create session with proxy manager
    proxy_session = ProxySession(self.proxy_manager)
    
    # Part 1: URL Resolution (gets first proxy)
    url = await self._resolve_url(video_link, proxy_session)
    
    # Part 2: Video Info (reuses same proxy unless Part 1 retried)
    video_data, context = await self._extract_video_info_with_retry(
        url, video_id, proxy_session
    )
    
    # Part 3: Download (still same proxy unless previous parts retried)
    video_bytes = await self._download_video_with_retry(
        video_url, context, proxy_session
    )

client = TikTokClient(
    proxy_manager=proxy_manager,
    data_only_proxy=True,  # Proxies for API, direct for media
)

class TikTokClient:
    # curl_cffi session pool (keyed by proxy URL)
    _curl_session_pool: dict[Optional[str], CurlAsyncSession] = {}
    
    @classmethod
    def _get_curl_session(cls, proxy: Optional[str] = None) -> CurlAsyncSession:
        """Get or create curl_cffi AsyncSession for a specific proxy."""
        with cls._curl_session_lock:
            if proxy not in cls._curl_session_pool:
                cls._curl_session_pool[proxy] = CurlAsyncSession(
                    impersonate="chrome120",
                    proxy=proxy,  # Baked into session
                    max_clients=1000,
                )
            return cls._curl_session_pool[proxy]

manager = ProxyManager.get_instance()
count = manager.get_proxy_count()
print(f"Total proxies in rotation: {count}")

# Peek without rotating (for logging)
current = manager.peek_current()
print(f"Next proxy will be: {current}")

if manager.has_proxies():
    print("Proxies are configured")
else:
    print("Using direct connection only")

def _strip_proxy_auth(proxy_url: Optional[str]) -> str:
    """Strip authentication info from proxy URL for safe logging."""
    if proxy_url is None:
        return "direct connection"
    
    match = re.match(r"^(https?://)(?:[^@]+@)?(.+)$", proxy_url)
    if match:
        protocol, host_port = match.groups()
        return f"{protocol}{host_port}"
    return proxy_url

# Usage
logger.debug(f"Using proxy: {_strip_proxy_auth(proxy)}")
# Output: "Using proxy: http://proxy1.example.com:8080"
# Instead of: "Using proxy: http://user:pass@proxy1.example.com:8080"

# At startup (main.py)
ProxyManager.initialize("proxies.txt", include_host=True)

# In any module
from tiktok_api import ProxyManager

manager = ProxyManager.get_instance()
if manager:
    proxy = manager.get_next_proxy()

# Reset singleton (mainly for tests)
ProxyManager.reset()

if not os.path.isfile(file_path):
    logger.error(f"Proxy file not found: {file_path}")
    if include_host:
        self._proxies = [None]  # Fallback to direct connection
    return

if not self._proxies:
    logger.warning("No proxies loaded, will use direct connection")
    self._proxies = [None]

try:
    with open(file_path, "r", encoding="utf-8") as f:
        for line in f:
            # Parse proxies
except Exception as e:
    logger.error(f"Failed to load proxy file {file_path}: {e}")

# Proxy file path (relative or absolute)
PROXY_FILE=proxies.txt

# Include direct connection in rotation
INCLUDE_HOST=false

# data/config.py
"proxy": {
    "proxy_file": os.getenv("PROXY_FILE"),
    "include_host": os.getenv("INCLUDE_HOST", "false").lower() == "true",
}