add package: pyproject + src

AioProxies: proxy parsing/formatting (aiohttp/camoufox/socks5/url) and
source management (session template with {session} + caller fields,
rotating list, static, from_file). config-free, no module globals,
per-instance rotation, never sys.exits on a missing file. also exported
as ProxyManager / aioproxies. optional [net] extra adds aiohttp
current_ip/reset. src/ multi-module, hatchling, zero core deps.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>

pyproject.toml

@@ -0,0 +1,16 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "aioproxies"
+version = "0.1.0"
+description = "proxy parsing, formatting, and source management for aiohttp/aioweb, camoufox, and socks5"
+requires-python = ">=3.10"
+dependencies = []
+
+[project.optional-dependencies]
+net = ["aiohttp>=3.9"]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/aioproxies"]

src/aioproxies/__init__.py

@@ -0,0 +1,12 @@
+"""aioproxies — proxy parsing, formatting, and source management.
+
+renders proxies for aiohttp/aioweb, camoufox, and socks5; manages session
+templates (with caller-supplied fields like country/ttl), rotating lists, or a
+static proxy. credentials are always injected, never hardcoded.
+"""
+from .manager import AioProxies, ProxyManager, aioproxies
+from .proxy import Proxy, parse
+
+__all__ = ["AioProxies", "ProxyManager", "aioproxies", "Proxy", "parse"]
+
+__version__ = "0.1.0"

src/aioproxies/manager.py

@@ -0,0 +1,87 @@
+"""proxy source management: session templates, rotation, static.
+
+`AioProxies` is constructed with exactly one source and hands out `Proxy`
+objects. it carries no module-level globals (rotation state is per-instance) and
+never exits the process — a missing proxy file raises, it does not `sys.exit`.
+
+sources:
+- template: a format string with named placeholders. `{session}` is filled with
+  a fresh session id on each `next()`; any other placeholder (e.g. `{country}`,
+  `{ttl}`) is filled from keyword args passed to `next(**fields)`. a bare `{}` is
+  also accepted and treated as the session slot (back-compat with simple templates).
+- proxies:  a list of specs cycled round-robin
+- static:   one fixed proxy
+"""
+import logging
+import random
+import string
+from typing import Dict, List, Optional, Union
+
+from .proxy import Proxy, parse
+
+log = logging.getLogger(__name__)
+
+
+class AioProxies:
+    """hands out proxies from a template, a rotating list, or a static value"""
+
+    def __init__(
+        self,
+        *,
+        template: Optional[str] = None,
+        proxies: Optional[List[Union[str, Proxy]]] = None,
+        static: Optional[Union[str, Proxy]] = None,
+        session_len: int = 10,
+        shuffle: bool = True,
+    ):
+        sources = [s for s in (template, proxies, static) if s]
+        if len(sources) != 1:
+            raise ValueError("provide exactly one of: template, proxies, static")
+        # normalize a bare {} session slot to the named {session} form
+        self.template = template.replace("{}", "{session}") if template else None
+        self.session_len = session_len
+        self._static = parse(static) if static else None
+        self._proxies = [parse(p) for p in proxies] if proxies else []
+        if self._proxies and shuffle:
+            random.shuffle(self._proxies)
+        self._index = 0
+
+    @classmethod
+    def from_file(cls, path: str, **kwargs) -> "AioProxies":
+        """build a rotating manager from a newline-delimited proxy file
+
+        raises FileNotFoundError if the path is missing — never exits the process.
+        """
+        with open(path, "r", encoding="utf-8") as fh:
+            lines = [ln.strip() for ln in fh if ln.strip()]
+        if not lines:
+            raise ValueError(f"no proxies found in {path}")
+        return cls(proxies=lines, **kwargs)
+
+    def session_id(self) -> str:
+        """generate a fresh numeric session id for template filling"""
+        return "".join(random.choices(string.digits, k=self.session_len))
+
+    def next(self, **fields: object) -> Proxy:
+        """return the next proxy from the configured source
+
+        for template sources, `{session}` is always filled with a fresh id and any
+        other named placeholder is filled from `fields` (e.g. next(country="ca",
+        ttl=30)). fields are ignored by list/static sources.
+        """
+        if self._static is not None:
+            return self._static
+        if self.template is not None:
+            return parse(self.template.format(session=self.session_id(), **fields))
+        proxy = self._proxies[self._index]
+        self._index = (self._index + 1) % len(self._proxies)
+        return proxy
+
+    def get(self, **fields: object) -> Dict[str, str]:
+        """convenience: next proxy as an aiohttp / aioweb proxies dict"""
+        return self.next(**fields).aiohttp()
+
+
+# name aliases — same class, call it whichever reads best at your call site
+ProxyManager = AioProxies
+aioproxies = AioProxies

src/aioproxies/net.py

@@ -0,0 +1,60 @@
+"""optional network helpers for proxies (current ip, reset).
+
+these need aiohttp, kept behind the `[net]` extra so the core (parse/format/
+rotate) stays dependency-free. importing this module without aiohttp is fine;
+calling a function without it raises a clear error.
+"""
+import logging
+from typing import Optional, Union
+
+from .proxy import Proxy, parse
+
+log = logging.getLogger(__name__)
+
+try:
+    import aiohttp
+
+    _HAVE_AIOHTTP = True
+except ImportError:  # pragma: no cover
+    aiohttp = None  # type: ignore[assignment]
+    _HAVE_AIOHTTP = False
+
+_MISSING = "aiohttp is required for network helpers; install aioproxies[net]"
+_IP_URL = "https://api.ipify.org?format=json"
+
+
+async def current_ip(
+    proxy: Union[str, Proxy], *, test_url: str = _IP_URL, timeout: float = 15.0
+) -> Optional[str]:
+    """return the egress ip seen through a proxy, or None on failure
+
+    set `test_url` to a socks-capable path / your own echo endpoint as needed.
+    """
+    if not _HAVE_AIOHTTP:
+        raise RuntimeError(_MISSING)
+    p = parse(proxy)
+    t = aiohttp.ClientTimeout(total=timeout)
+    try:
+        async with aiohttp.ClientSession(timeout=t) as session:
+            async with session.get(test_url, proxy=p.url()) as resp:
+                data = await resp.json()
+                return data.get("ip")
+    except Exception as exc:
+        log.warning("ip check failed: %s", exc)
+        return None
+
+
+async def reset(reset_url: str, *, timeout: float = 15.0) -> bool:
+    """hit a provider reset url to rotate the upstream ip; True on http 200"""
+    if not _HAVE_AIOHTTP:
+        raise RuntimeError(_MISSING)
+    t = aiohttp.ClientTimeout(total=timeout)
+    try:
+        async with aiohttp.ClientSession(timeout=t) as session:
+            async with session.get(reset_url) as resp:
+                ok = resp.status == 200
+                log.info("proxy reset %s -> %s", reset_url, resp.status)
+                return ok
+    except Exception as exc:
+        log.warning("proxy reset failed: %s", exc)
+        return False

src/aioproxies/proxy.py

@@ -0,0 +1,77 @@
+"""proxy parsing and formatting (pure, no network IO).
+
+a `Proxy` holds host/port/optional-auth and renders the shapes different clients
+want: an aiohttp/aioweb proxies dict, a camoufox proxy dict, a socks5 dict, or a
+plain url. `parse` accepts the common "host:port" and "host:port:user:pass" string
+forms (the user field may itself contain commas, e.g. session-param proxies).
+"""
+from dataclasses import dataclass
+from typing import Dict, Optional, Union
+
+SCHEME_HTTP = "http"
+SCHEME_SOCKS5 = "socks5"
+
+
+@dataclass
+class Proxy:
+    """a single proxy endpoint with optional auth"""
+
+    host: str
+    port: str
+    user: Optional[str] = None
+    password: Optional[str] = None
+
+    @property
+    def has_auth(self) -> bool:
+        """whether credentials are present"""
+        return bool(self.user and self.password)
+
+    def url(self, scheme: str = SCHEME_HTTP) -> str:
+        """render as a url, embedding auth when present"""
+        if self.has_auth:
+            return f"{scheme}://{self.user}:{self.password}@{self.host}:{self.port}"
+        return f"{scheme}://{self.host}:{self.port}"
+
+    def aiohttp(self) -> Dict[str, str]:
+        """proxies dict for aiohttp / aioweb's ExtendedSession(proxies=...)"""
+        u = self.url(SCHEME_HTTP)
+        return {"http": u, "https": u}
+
+    def camoufox(self) -> Dict[str, str]:
+        """proxy dict for camoufox / playwright (server + separate auth)"""
+        out = {"server": f"http://{self.host}:{self.port}"}
+        if self.has_auth:
+            out["username"] = self.user  # type: ignore[assignment]
+            out["password"] = self.password  # type: ignore[assignment]
+        return out
+
+    def socks5(self) -> Dict[str, str]:
+        """proxy dict for a socks5 endpoint (server + separate auth)"""
+        out = {"server": f"socks5://{self.host}:{self.port}"}
+        if self.has_auth:
+            out["username"] = self.user  # type: ignore[assignment]
+            out["password"] = self.password  # type: ignore[assignment]
+        return out
+
+    def socks5_url(self) -> str:
+        """render as a socks5 url"""
+        return self.url(SCHEME_SOCKS5)
+
+
+def parse(spec: Union[str, Proxy]) -> Proxy:
+    """parse a proxy spec into a Proxy
+
+    accepts an existing Proxy (returned as-is) or a colon-delimited string in
+    `host:port` or `host:port:user:pass` form. raises ValueError on anything else
+    rather than guessing.
+    """
+    if isinstance(spec, Proxy):
+        return spec
+    parts = spec.split(":")
+    if len(parts) == 4:
+        host, port, user, password = parts
+        return Proxy(host, port, user, password)
+    if len(parts) == 2:
+        host, port = parts
+        return Proxy(host, port)
+    raise ValueError("expected 'host:port' or 'host:port:user:pass'")