fix: URL-aware parse() at static/template/net.current_ip call sites; self-clear stale burn flag; guard replace([])

static=/template=/net.current_ip() fed URL-shaped input straight into parse(),
which only understands host:port(:user:pass) specs and mis-splits a url's
colons into a garbage Proxy (e.g. embedded-auth urls) instead of raising or
normalizing. all three now route through to_proxy(), the url-aware normalizer
add()/replace() already use.

the per-state `burned` flag (drives WARNING vs DEBUG on the cooldown
fallthrough log) was only ever cleared by restore(); a timed burn() that
lazily expires (is_burned() correctly reads False again) left `burned` stuck
True, permanently upgrading routine cooldown noise to WARNING. `burned` now
self-clears wherever the existing lazy-expiry check already finds the proxy
available (_next_from_list's fine-tier scan, is_burned()).

replace([]) silently wiped the entire pool with no error, even though the
constructor already rejects an empty proxies=[] list. replace() now raises
the same ValueError for an empty list on list sources.

parse()'s docstring corrected: it is spec-string only and does not understand
url shapes; points callers at to_proxy() for url/dict/spec input.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
2026-07-06 00:15:06 -04:00
parent 9649eb77c7
commit a4a0227d5d
3 changed files with 34 additions and 12 deletions
+26 -7
View File
@@ -56,7 +56,7 @@ class AioProxies:
self.session_len = session_len
self.cooldown = cooldown
self._shuffle = shuffle
self._static = parse(static) if static is not None else None
self._static = to_proxy(static) if static is not None else None
self._proxies = self._dedupe([parse(p) for p in proxies]) if proxies is not None else []
if self._proxies and shuffle:
random.shuffle(self._proxies)
@@ -116,6 +116,17 @@ class AioProxies:
return False
return now >= timeout
@staticmethod
def _clear_stale_burn(state: Dict[str, object]) -> None:
"""self-clear a lazily-expired timed burn's `burned` flag once found available
mirrors the lazy expiry already applied to `timeout` itself: a timed burn()
that has simply run out its clock should stop upgrading routine cooldown
fallthrough logs to warning, without requiring an explicit restore().
"""
if state["timeout"] not in (None, 0, _DEAD):
state["burned"] = False
def next(self, **fields: object) -> Proxy:
"""return the next proxy from the configured source
@@ -145,7 +156,7 @@ class AioProxies:
# a malformed template (e.g. an unmatched '{') makes str.format raise a
# bare ValueError - re-raise naming the cause so it isn't cryptic
raise ValueError(f"malformed proxy template {self.template!r}: {exc}") from exc
return parse(filled)
return to_proxy(filled)
return self._next_from_list()
def _next_from_list(self) -> Proxy:
@@ -158,9 +169,10 @@ class AioProxies:
fine: List[int] = []
for offset in range(count):
idx = (self._index + offset) % count
timeout = self._state[self._proxies[idx].key()]["timeout"]
if self._available(timeout, now):
state = self._state[self._proxies[idx].key()]
if self._available(state["timeout"], now):
fine.append(idx)
self._clear_stale_burn(state)
if fine:
chosen = fine[0]
@@ -248,8 +260,11 @@ class AioProxies:
key = canonical_key(proxy)
if key not in self._state:
return False
timeout = self._state[key]["timeout"]
return not self._available(timeout, time.time())
state = self._state[key]
available = self._available(state["timeout"], time.time())
if available:
self._clear_stale_burn(state)
return not available
def stats(self) -> List[Dict[str, object]]:
"""per-proxy usage + state snapshot (list sources only, else empty)
@@ -293,11 +308,15 @@ class AioProxies:
keep_state=False (default) wipes all per-proxy state (fresh batch).
keep_state=True preserves uses/timeout for proxies whose canonical key
survives the swap; new proxies start clean, dropped ones are forgotten.
resets the rotation index and honors the manager's shuffle setting.
resets the rotation index and honors the manager's shuffle setting. raises
ValueError on an empty list, matching the constructor's guard - use remove()
to drop proxies one at a time if you actually want a smaller pool.
"""
if not self._is_list_source():
self._warn_non_list("replace()")
return
if len(proxies) == 0:
raise ValueError("proxies list is empty; provide at least one proxy")
old_state = self._state
incoming = [to_proxy(p) for p in proxies]
if self._shuffle:
+2 -2
View File
@@ -7,7 +7,7 @@ calling a function without it raises a clear error.
import logging
from typing import Optional, Union
from .proxy import Proxy, parse
from .proxy import Proxy, to_proxy
log = logging.getLogger(__name__)
@@ -32,7 +32,7 @@ async def current_ip(
"""
if not _HAVE_AIOHTTP:
raise RuntimeError(_MISSING)
p = parse(proxy)
p = to_proxy(proxy)
t = aiohttp.ClientTimeout(total=timeout)
try:
async with aiohttp.ClientSession(timeout=t) as session:
+6 -3
View File
@@ -97,12 +97,15 @@ class Proxy:
def parse(spec: Union[str, Proxy]) -> Proxy:
"""parse a proxy spec into a Proxy
"""parse a colon-delimited 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. the 4-part form splits on the first
three colons only, so a password may itself contain colons. raises ValueError
on anything else rather than guessing.
three colons only, so a password may itself contain colons. this is spec-string
only - it does not understand url shapes (`scheme://user:pass@host:port`); a
url has 3+ colons and would be misparsed as a 4-part spec rather than rejected.
use `to_proxy()` for url / dict / spec input, or anywhere the shape isn't
guaranteed to be a bare spec string.
"""
if isinstance(spec, Proxy):
return spec