Proxy

[Tracker878]

log 1:

✔ Container bee_bot Recreated 1.1s
Attaching to bee_bot
bee_bot | 2024-11-16 00:52:42,762 - main - INFO - Entering main block
bee_bot | 2024-11-16 00:52:42,762 - main - INFO - Starting the application...
bee_bot | ℹ️ 00:52:42 | Created accounts.json and wallet_private.json
bee_bot | ⚠️ 00:52:42 | Invalid proxy format (missing protocol):
bee_bot | 09:80@21:22
bee_bot | ℹ️ 00:52:42 | Loaded 0 proxies from file
bee_bot | ⚠️ 00:52:42 | No proxies available, continuing without proxies
bee_bot | ℹ️ 00:52:42 | Detected 1 sessions | 0 proxies
bee_bot | ℹ️ 00:52:43 | Running in sequential mode
bee_bot | ℹ️ 00:52:43 | Waiting 121.4 seconds before starting session...

ask chatgpt about proxy.py

The code you provided is mostly well-structured, but there are a few potential improvements and fixes to consider for better performance, clarity, and compatibility. Here are some suggested adjustments and fixes:

Use Type Hinting Correctly for List of Strings (list[str] Compatibility Issue)

The type hinting list[str] requires Python 3.9 or later. If you're using an earlier Python version, you should replace it with List[str] from the typing module.
Import List from typing if necessary.
Improved Session Management

It’s a good practice to close the ClientSession after use to free up resources.
Here's an updated version of your code with improvements and fixes:

Updated Code:
python
Copy code
import aiohttp
import asyncio
from typing import Optional, Dict, List # Correct type hinting import for older Python versions
from better_proxy import Proxy
from bot.config import settings
from bot.utils.logger import logger
from aiohttp import TCPConnector, ClientSession, ClientTimeout

class ProxyManager:
def init(self):
self.proxies: Dict[str, Optional[str]] = {}
self.proxy_file = "bot/config/proxies.txt"

def load_proxies(self) -> List[str]:  # Use List[str] for better compatibility
    """Loads proxies from a file"""
    try:
        with open(self.proxy_file, 'r') as f:
            proxies = [line.strip() for line in f if line.strip()]
            
        # Check the format of each proxy
        valid_proxies = []
        for proxy in proxies:
            if '://' not in proxy:
                logger.warning(f"Invalid proxy format (missing protocol): {proxy}")
                continue
            protocol = proxy.split('://', 1)[0].lower()
            if protocol not in settings.PROXY_TYPES:
                logger.warning(f"Unsupported proxy protocol: {protocol}")
                continue
            valid_proxies.append(proxy)
            
        logger.info(f"Loaded {len(valid_proxies)} proxies from file")
        return valid_proxies
    except FileNotFoundError:
        logger.warning(f"Proxy file {self.proxy_file} not found")
        return []
    except Exception as e:
        logger.error(f"Error loading proxies: {e}")
        return []

async def check_proxy(self, proxy_str: str) -> bool:
    """Checks the functionality of a proxy"""
    if not settings.VALIDATE_PROXIES:
        return True
        
    try:
        proxy = Proxy.from_str(proxy_str)
        
        # Use aiohttp_socks for SOCKS proxies
        if proxy.protocol.startswith('socks'):
            from aiohttp_socks import ProxyType, ProxyConnector
            proxy_type = ProxyType.SOCKS5 if proxy.protocol == 'socks5' else ProxyType.SOCKS4
            connector = ProxyConnector(
                proxy_type=proxy_type,
                host=proxy.host,
                port=proxy.port,
                username=proxy.login if proxy.login else None,
                password=proxy.password if proxy.password else None,
                verify_ssl=False
            )
        else:
            # For HTTP/HTTPS proxies
            proxy_url = f"{proxy.protocol}://"
            if proxy.login and proxy.password:
                proxy_url += f"{proxy.login}:{proxy.password}@"
            proxy_url += f"{proxy.host}:{proxy.port}"
            connector = TCPConnector(verify_ssl=False)

        # Try Telegram API check first
        try:
            from pyrogram import Client
            async with Client(
                "proxy_check",
                api_id=settings.API_ID,
                api_hash=settings.API_HASH,
                proxy=dict(
                    scheme=proxy.protocol,
                    hostname=proxy.host,
                    port=proxy.port,
                    username=proxy.login if proxy.login else None,
                    password=proxy.password if proxy.password else None
                ),
                in_memory=True
            ) as client:
                logger.success(f"Proxy {proxy_str} is working (Telegram check)")
                return True
        except Exception as e:
            logger.debug(f"Telegram proxy check failed: {e}")

        # HTTP request check
        timeout = ClientTimeout(total=settings.PROXY_TIMEOUT)
        async with ClientSession(connector=connector) as session:
            try:
                async with session.get(
                    "http://www.google.com",
                    proxy=None if proxy.protocol.startswith('socks') else proxy_url,
                    timeout=timeout,
                    ssl=False
                ) as response:
                    if response.status == 200:
                        logger.success(f"Proxy {proxy_str} is working (HTTP check)")
                        return True
            except Exception as e:
                logger.debug(f"HTTP proxy check failed: {e}")
                
            try:
                # Try through api.ipify.org
                async with session.get(
                    settings.PROXY_CHECK_URL,
                    proxy=proxy_url,
                    timeout=timeout,
                    ssl=False
                ) as response:
                    if response.status == 200:
                        logger.success(f"Proxy {proxy_str} is working (ipify check)")
                        return True
            except Exception as e:
                logger.debug(f"ipify proxy check failed: {e}")
                
        # Use anyway if all checks fail but the proxy is valid
        if await self.validate_proxy(proxy_str):
            logger.warning(f"Proxy {proxy_str} failed checks but will be used anyway")
            return True
                
        return False
    except Exception as e:
        logger.warning(f"Proxy {proxy_str} validation failed: {e}")
        return False

async def validate_proxy(self, proxy_str: str) -> bool:
    """Validates the format and type of a proxy"""
    try:
        proxy = Proxy.from_str(proxy_str)
        if proxy.protocol not in settings.PROXY_TYPES:
            logger.warning(f"Unsupported proxy type: {proxy.protocol}")
            return False
        return True
    except Exception as e:
        logger.warning(f"Invalid proxy format: {proxy_str} - {e}")
        return False

async def init_proxies(self):
    """Initializes and checks proxies"""
    if not settings.USE_PROXY_FROM_FILE:
        return

    raw_proxies = self.load_proxies()
    valid_proxies = []

    for proxy in raw_proxies:
        if await self.validate_proxy(proxy):
            if await self.check_proxy(proxy):
                valid_proxies.append(proxy)
            else:
                logger.warning(f"Proxy {proxy} is not working")

    if not valid_proxies:
        logger.warning("No valid proxies found")
        return

    logger.info(f"Loaded {len(valid_proxies)} valid proxies")
    return valid_proxies

def assign_proxy(self, session_name: str, proxy: Optional[str]):
    """Assigns a proxy to a session"""
    self.proxies[session_name] = proxy

def get_proxy(self, session_name: str) -> Optional[str]:
    """Gets a proxy for a session"""
    return self.proxies.get(session_name)

proxy_manager = ProxyManager()

Key Fixes and Improvements:
Changed list[str] to List[str] for compatibility with Python versions earlier than 3.9.
Used async with for managing the pyrogram.Client to ensure proper connection handling.
Ensured that ClientSession is closed using an async with block to avoid resource leaks.
This should help make your script more robust and compatible! Let me know if you need any further adjustments.

log

✔ Container bee_bot Recreated 10.9s
Attaching to bee_bot
bee_bot | 2024-11-16 01:01:00,120 - main - INFO - Entering main block
bee_bot | 2024-11-16 01:01:00,120 - main - INFO - Starting the application...
bee_bot | ℹ️ 01:01:00 | Using existing proxy assignments from accounts.json
bee_bot | ℹ️ 01:01:00 | Detected 1 sessions | 1 proxies
bee_bot | ℹ️ 01:01:00 | Running in sequential mode
bee_bot | ℹ️ 01:01:00 | Waiting 141.4 seconds before starting session...
bee_bot | ℹ️ 01:03:21 | ==================================================
bee_bot | ℹ️ 01:03:21 | Processing session: new16
bee_bot | ℹ️ 01:03:21 | new16 | Starting to obtain tg_web_data

[Tracker878]

o1 code

It looks like your code has some indentation issues and a small mistake in the init method name. In Python, proper indentation is crucial because it defines the scope of classes, functions, loops, and other control structures.

Here's the corrected version of your code with proper indentation and the corrected init method:

python
Copy code
import aiohttp
import asyncio
from typing import Optional, Dict, List # Correct type hinting import for older Python versions
from better_proxy import Proxy
from bot.config import settings
from bot.utils.logger import logger
from aiohttp import TCPConnector, ClientSession, ClientTimeout

class ProxyManager:
def init(self):
self.proxies: Dict[str, Optional[str]] = {}
self.proxy_file = "bot/config/proxies.txt"

def load_proxies(self) -> List[str]:
    """Loads proxies from a file"""
    try:
        with open(self.proxy_file, 'r') as f:
            proxies = [line.strip() for line in f if line.strip()]

        # Check the format of each proxy
        valid_proxies = []
        for proxy in proxies:
            if '://' not in proxy:
                logger.warning(f"Invalid proxy format (missing protocol): {proxy}")
                continue
            protocol = proxy.split('://', 1)[0].lower()
            if protocol not in settings.PROXY_TYPES:
                logger.warning(f"Unsupported proxy protocol: {protocol}")
                continue
            valid_proxies.append(proxy)

        logger.info(f"Loaded {len(valid_proxies)} proxies from file")
        return valid_proxies
    except FileNotFoundError:
        logger.warning(f"Proxy file {self.proxy_file} not found")
        return []
    except Exception as e:
        logger.error(f"Error loading proxies: {e}")
        return []

async def check_proxy(self, proxy_str: str) -> bool:
    """Checks the functionality of a proxy"""
    if not settings.VALIDATE_PROXIES:
        return True

    try:
        proxy = Proxy.from_str(proxy_str)

        # Use aiohttp_socks for SOCKS proxies
        if proxy.protocol.startswith('socks'):
            from aiohttp_socks import ProxyType, ProxyConnector
            proxy_type = ProxyType.SOCKS5 if proxy.protocol == 'socks5' else ProxyType.SOCKS4
            connector = ProxyConnector(
                proxy_type=proxy_type,
                host=proxy.host,
                port=proxy.port,
                username=proxy.login if proxy.login else None,
                password=proxy.password if proxy.password else None,
                verify_ssl=False
            )
        else:
            # For HTTP/HTTPS proxies
            proxy_url = f"{proxy.protocol}://"
            if proxy.login and proxy.password:
                proxy_url += f"{proxy.login}:{proxy.password}@"
            proxy_url += f"{proxy.host}:{proxy.port}"
            connector = TCPConnector(verify_ssl=False)

        # Try Telegram API check first
        try:
            from pyrogram import Client
            async with Client(
                "proxy_check",
                api_id=settings.API_ID,
                api_hash=settings.API_HASH,
                proxy=dict(
                    scheme=proxy.protocol,
                    hostname=proxy.host,
                    port=proxy.port,
                    username=proxy.login if proxy.login else None,
                    password=proxy.password if proxy.password else None
                ),
                in_memory=True
            ) as client:
                logger.success(f"Proxy {proxy_str} is working (Telegram check)")
                return True
        except Exception as e:
            logger.debug(f"Telegram proxy check failed: {e}")

        # HTTP request check
        timeout = ClientTimeout(total=settings.PROXY_TIMEOUT)
        async with ClientSession(connector=connector) as session:
            try:
                async with session.get(
                    "http://www.google.com",
                    proxy=None if proxy.protocol.startswith('socks') else proxy_url,
                    timeout=timeout,
                    ssl=False
                ) as response:
                    if response.status == 200:
                        logger.success(f"Proxy {proxy_str} is working (HTTP check)")
                        return True
            except Exception as e:
                logger.debug(f"HTTP proxy check failed: {e}")

            try:
                # Try through api.ipify.org
                async with session.get(
                    settings.PROXY_CHECK_URL,
                    proxy=proxy_url,
                    timeout=timeout,
                    ssl=False
                ) as response:
                    if response.status == 200:
                        logger.success(f"Proxy {proxy_str} is working (ipify check)")
                        return True
            except Exception as e:
                logger.debug(f"ipify proxy check failed: {e}")

        # Use anyway if all checks fail but the proxy is valid
        if await self.validate_proxy(proxy_str):
            logger.warning(f"Proxy {proxy_str} failed checks but will be used anyway")
            return True

        return False
    except Exception as e:
        logger.warning(f"Proxy {proxy_str} validation failed: {e}")
        return False

async def validate_proxy(self, proxy_str: str) -> bool:
    """Validates the format and type of a proxy"""
    try:
        proxy = Proxy.from_str(proxy_str)
        if proxy.protocol not in settings.PROXY_TYPES:
            logger.warning(f"Unsupported proxy type: {proxy.protocol}")
            return False
        return True
    except Exception as e:
        logger.warning(f"Invalid proxy format: {proxy_str} - {e}")
        return False

async def init_proxies(self):
    """Initializes and checks proxies"""
    if not settings.USE_PROXY_FROM_FILE:
        return

    raw_proxies = self.load_proxies()
    valid_proxies = []

    for proxy in raw_proxies:
        if await self.validate_proxy(proxy):
            if await self.check_proxy(proxy):
                valid_proxies.append(proxy)
            else:
                logger.warning(f"Proxy {proxy} is not working")

    if not valid_proxies:
        logger.warning("No valid proxies found")
        return

    logger.info(f"Loaded {len(valid_proxies)} valid proxies")
    return valid_proxies

def assign_proxy(self, session_name: str, proxy: Optional[str]):
    """Assigns a proxy to a session"""
    self.proxies[session_name] = proxy

def get_proxy(self, session_name: str) -> Optional[str]:
    """Gets a proxy for a session"""
    return self.proxies.get(session_name)

proxy_manager = ProxyManager()
Explanation of Changes:

Indentation: All method definitions inside the ProxyManager class are indented with four spaces to reflect their belonging to the class. In Python, indentation defines the scope, so proper indentation is essential.

init Method: The constructor method in Python classes should be named init, with double underscores before and after init. This method is called automatically when a new instance of the class is created.

python
Copy code
def init(self):
# Initialization code here
Consistent Indentation: Ensured that all code blocks within methods are consistently indented, maintaining readability and preventing syntax errors.

Additional Recommendations:

Exception Handling: While you have exception handling in place, consider being more specific with exceptions where possible to avoid catching unintended exceptions.

Async Context Managers: Ensure that you have the necessary libraries installed and that they are compatible with asynchronous context managers, especially when using async with.

Logging Levels: Adjust the logging levels (logger.warning, logger.debug, logger.error, logger.success) according to the significance of the messages to help with debugging.

Dependencies: Make sure all the imported modules (better_proxy, bot.config, bot.utils.logger, pyrogram, aiohttp_socks) are installed and correctly configured in your environment.

Testing the Code:

After making these changes, test your code to ensure that:

Proxies are correctly loaded from the file.
The proxy validation and checking functions work as expected.
Proxies are assigned and retrieved correctly for sessions.
Example Usage:

python
Copy code
async def main():
await proxy_manager.init_proxies()
proxy_manager.assign_proxy('session1', 'http://proxy.example.com:8080')
proxy = proxy_manager.get_proxy('session1')
print(f"Assigned proxy for session1: {proxy}")

asyncio.run(main())
This example initializes the proxies, assigns a proxy to a session, and then retrieves it.