monosansandGitHub
e4047e16ae
- Drop support for Python 3.7. - Use offline geolocation database instead of ip-api.com. - Add support for proxy authentication. - Add pre-compiled binaries. - Refactor the code by a lot. - Simplify config. - Add automatic text encoding detection. - Use TOML instead of INI for configuration. - Add support for using httpbin-compatible services and services which return plain IP address for getting proxy's exit node. - Add support for reading proxy lists from local files. - Add support for saving to json. - Remove support for saving geolocation to txt, use json instead. - Improve parser. - Improve config validation. - Use poetry instead of requirements.txt. - Use venv in start scripts.
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import itertools
|
|
from collections import Counter
|
|
from typing import TYPE_CHECKING, Callable, Dict, Iterator, List, Set, Tuple
|
|
|
|
from aiohttp_socks import ProxyType
|
|
|
|
from . import sort
|
|
from .proxy import Proxy
|
|
|
|
if TYPE_CHECKING:
|
|
from _typeshed import SupportsRichComparison
|
|
|
|
|
|
class ProxyStorage:
|
|
__slots__ = ("_proxies", "enabled_protocols")
|
|
|
|
def __init__(self) -> None:
|
|
self._proxies: Set[Proxy] = set()
|
|
self.enabled_protocols: Set[ProxyType] = set()
|
|
|
|
def add(self, proxy: Proxy, /) -> None:
|
|
self.enabled_protocols.add(proxy.protocol)
|
|
self._proxies.add(proxy)
|
|
|
|
def remove(self, proxy: Proxy, /) -> None:
|
|
self._proxies.remove(proxy)
|
|
|
|
def get_grouped(self) -> Dict[ProxyType, Tuple[Proxy, ...]]:
|
|
key = sort.protocol_sort_key
|
|
d: Dict[ProxyType, Tuple[Proxy, ...]] = {
|
|
proto: ()
|
|
for proto in sort.PROTOCOL_ORDER
|
|
if proto in self.enabled_protocols
|
|
}
|
|
for (_, proto), v in itertools.groupby(
|
|
sorted(self._proxies, key=key), key=key
|
|
):
|
|
d[proto] = tuple(v)
|
|
return d
|
|
|
|
def get_sorted(
|
|
self,
|
|
*,
|
|
key: Callable[[Proxy], SupportsRichComparison] = sort.protocol_sort_key,
|
|
) -> List[Proxy]:
|
|
return sorted(self._proxies, key=key)
|
|
|
|
def get_count(self) -> Dict[ProxyType, int]:
|
|
return dict(Counter(proxy.protocol for proxy in self._proxies))
|
|
|
|
def __iter__(self) -> Iterator[Proxy]:
|
|
return iter(self._proxies)
|