Tracker878
commented
2 hours ago 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.
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"
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