Как можно настроить асинхронную часть кода, чтобы работал вместе с проксификсером, который был установлен на `python.exe` в виртуальной среде? При использовании SOCKS5 или HTTP прокси все работает до момента перехода к асинхронной части кода, где работа прекращается и происходит бесконечное ожидание без возникновения исключений. Если попытаться запустить код в режиме отладки в PyCharm под проксификсером, возникает исключение `ConnectionRefusedError: [WinError 10061] Подключение не установлено, т.к. конечный компьютер отверг запрос на подключение`.
Возможно, проблема заключается в том, что асинхронный код не поддерживает использование прокси-сервера напрямую. Для того чтобы асинхронный код корректно работал через прокси-сервер, необходимо использовать специальные библиотеки, которые поддерживают асинхронные запросы через прокси.
Одним из таких инструментов является aiohttp, который предоставляет возможность делать асинхронные HTTP-запросы. Вы можете настроить прокси-сервер в aiohttp следующим образом:
```python
import aiohttp
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url, proxy="http://your-proxy-server:port") as response:
return await response.text()
url = 'http://example.com'
result = await fetch(url)
print(result)
```
Замените `'http://your-proxy-server:port'` на адрес вашего прокси-сервера. Убедитесь, что ваш прокси-сервер корректно настроен и подключен.
Если вы используете SOCKS5 прокси, то можете воспользоваться библиотекой aiohttp-socks, которая добавляет поддержку SOCKS прокси в aiohttp:
```python
import aiohttp
import aiohttp_socks
async def fetch(url):
connector = aiohttp_socks.ProxyConnector.from_url('socks5://your-socks5-proxy-server:port')
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url) as response:
return await response.text()
url = 'http://example.com'
result = await fetch(url)
print(result)
```
С помощью данных библиотек вы сможете корректно настроить асинхронную часть кода для работы через прокси-сервер.
Посмотрите этот код, может, поможет:
import asyncio
import os
import time
import random
from time import perf_counter
import aiohttp
import aiofiles
SITE = ' https://thispersondoesnotexist.com/ '
IMAGE_COUNT = 50
PROXY = ''
def generate_filename(file_extension):
temp = str(int(time.time()))
for _ in range(5):
temp += chr(random.randint(65, 75))
return f'{temp}.{file_extension}'
async def download_image(image_num):
async with aiohttp.ClientSession() as session:
async with session.get(SITE, proxy=f' http://{PROXY }') as response:
extension = response.headers['content-type'].split('/')[-1]
filename = os.path.join('Images_async', generate_filename(extension))
async with aiofiles.open(filename, mode='wb') as file:
async for chunk in response.content.iter_chunked(64 * 1024):
await file.write(chunk)
print(f'image: {image_num + 1} finished...')
async def main():
image_tasks = []
for image_num in range(IMAGE_COUNT):
image_tasks.append(asyncio.create_task(download_image(image_num)))
await asyncio.gather(*image_tasks)
if __name__ == '__main__':
start = perf_counter()
asyncio.run(main())
print(f'time: {(perf_counter() - start):.02f}')
Я этот код увидел в процессе обучения. Он работает.