kai/zlib-dl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

release: cut v0.01 with working downloads and unstable proxy rotation

Browse Source
This commit is contained in:
Kai Tan
2026-04-01 00:03:53 +08:00
parent 97c48995b1
commit 97fff0b30e
23 changed files with 250 additions and 600 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -0,0 +1,3 @@
downloads/
__pycache__/
*.pyc
__pycache__/zlib_dl.cpython-314.pyc
BIN
View File
Binary file not shown.
test1.log
-5
View File
@@ -1,5 +0,0 @@
📡 已加载 56 个代理
🔗 已启动代理隧道: 🇭🇰 香港 01 -> 127.0.0.1:10800
⏳ 下载遇到 PoW 挑战，正在解决...
✅ 挑战已解决 (耗时 0.024s，节点 zh.ngd101.ru)
下载返回了 HTML 页面（可能是限制或错误）
test_503.log
-4
View File
@@ -1,4 +0,0 @@
Proxy IP: 61.171.248.193
Index status: 503
Index status after PoW: 503
Search status: 503
test_503.py
-88
View File
@@ -1,88 +0,0 @@
import urllib.request, time, hashlib, re, os, subprocess
from curl_cffi import requests
def solve_challenge(html):
m = re.search(r"\['([A-F0-9]{40})'", html)
if not m: return None, None
c = m.group(1)
n1 = int(c[0], 16)
i = 0
t0 = time.time()
while True:
d = hashlib.sha1((c + str(i)).encode()).digest()
if d[n1] == 0xb0 and d[n1+1] == 0x0b: break
i += 1
return c + str(i), f"{time.time() - t0:.3f}"
yaml_config = """
mode: global
mixed-port: 10899
bind-address: '127.0.0.1'
proxies:
"""
proxy_str = ""
with open("proxies.txt") as f:
for line in f:
if "ss," in line:
proxy_str = line.strip()
break
parts = proxy_str.split("=", 1)
cfg = parts[1].strip().split(",")
params = dict(p.strip().split("=", 1) for p in cfg[3:] if "=" in p)
yaml_config += f""" - name: "test"
type: ss
server: "{cfg[1].strip()}"
port: {cfg[2].strip()}
cipher: "{params.get('encrypt-method', '')}"
password: "{params.get('password', '')}"
"""
if "obfs" in params:
yaml_config += f" plugin: obfs\n plugin-opts:\n mode: {params['obfs']}\n host: {params['obfs-host']}\n"
with open("test_503.yaml", "w") as f: f.write(yaml_config)
proc = subprocess.Popen(["./.venv/bin/mihomo", "-f", "test_503.yaml"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(2)
s = requests.Session()
p = {"http": "socks5://127.0.0.1:10899", "https": "socks5://127.0.0.1:10899"}
try:
with open("test_503.log", "w") as out:
ip_resp = s.get("http://httpbin.org/ip", proxies=p, impersonate="chrome", timeout=5)
out.write(f"Proxy IP: {ip_resp.json()['origin']}\n")
r1 = s.get("https://zlib.li/", proxies=p, impersonate="chrome", timeout=10)
out.write(f"Index status: {r1.status_code}\n")
if r1.status_code == 503 and "Checking your browser" in r1.text:
ct, cti = solve_challenge(r1.text)
if ct:
s.cookies.set("c_token", ct, domain="zlib.li", path="/")
s.cookies.set("c_time", cti, domain="zlib.li", path="/")
r1 = s.get("https://zlib.li/", proxies=p, impersonate="chrome", timeout=10)
out.write(f"Index status after PoW: {r1.status_code}\n")
r2 = s.get("https://zlib.li/s/Project", proxies=p, impersonate="chrome", timeout=10)
out.write(f"Search status: {r2.status_code}\n")
from bs4 import BeautifulSoup
soup = BeautifulSoup(r2.text, "lxml")
card = soup.find("z-bookcard")
if card and card.get("download"):
dl = card.get("download")
out.write(f"Got dl path: {dl}\n")
r3 = s.get("https://zlib.li" + dl, proxies=p, impersonate="chrome", allow_redirects=True, timeout=10)
out.write(f"Download status: {r3.status_code}\n")
if r3.status_code == 503:
if "Checking your browser" in r3.text:
ct, cti = solve_challenge(r3.text)
if ct:
s.cookies.set("c_token", ct, domain="zlib.li", path="/")
s.cookies.set("c_time", cti, domain="zlib.li", path="/")
r4 = s.get("https://zlib.li" + dl, proxies=p, impersonate="chrome", allow_redirects=True, timeout=10)
out.write(f"Retry status: {r4.status_code}\n")
with open("test_503_body.html", "w") as f2: f2.write(r4.text)
else:
out.write("Not PoW 503\n")
finally:
proc.terminate()
test_503.yaml
-23
View File
@@ -1,23 +0,0 @@
mode: rule
mixed-port: 10899
bind-address: '127.0.0.1'
proxies:
- name: "test"
type: ss
server: "d32b6cd.gylcxo.xyz"
port: 18300
cipher: "aes-128-gcm"
password: "ced24190439d1ed1"
plugin: obfs
plugin-opts:
mode: http
host: Sm0lO349ch.microsoft.com
proxy-groups:
- name: "Proxy"
type: select
proxies:
- "test"
rules:
- MATCH, Proxy
test_503_2.py
-61
View File
@@ -1,61 +0,0 @@
import urllib.request, time, hashlib, re, os, subprocess
from curl_cffi import requests
def solve_challenge(html):
m = re.search(r"\['([A-F0-9]{40})'", html)
if not m: return None, None
c = m.group(1)
n1 = int(c[0], 16)
i = 0
t0 = time.time()
while True:
d = hashlib.sha1((c + str(i)).encode()).digest()
if d[n1] == 0xb0 and d[n1+1] == 0x0b: break
i += 1
return c + str(i), f"{time.time() - t0:.3f}"
yaml_config = """
mode: global
mixed-port: 10899
bind-address: '127.0.0.1'
proxies:
"""
proxy_str = ""
with open("proxies.txt") as f:
for line in f:
if "ss," in line:
proxy_str = line.strip()
break
parts = proxy_str.split("=", 1)
cfg = parts[1].strip().split(",")
params = dict(p.strip().split("=", 1) for p in cfg[3:] if "=" in p)
yaml_config += f""" - name: "test"
type: ss
server: "{cfg[1].strip()}"
port: {cfg[2].strip()}
cipher: "{params.get('encrypt-method', '')}"
password: "{params.get('password', '')}"
"""
if "obfs" in params:
yaml_config += f" plugin: obfs\n plugin-opts:\n mode: {params['obfs']}\n host: {params['obfs-host']}\n"
with open("test_503.yaml", "w") as f: f.write(yaml_config)
proc = subprocess.Popen(["./.venv/bin/mihomo", "-f", "test_503.yaml"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(2)
s = requests.Session()
p = {"http": "socks5://127.0.0.1:10899", "https": "socks5://127.0.0.1:10899"}
try:
r1 = s.get("https://zlib.li/", proxies=p, impersonate="chrome", timeout=10)
print("Initial:", r1.status_code)
if r1.status_code == 503 and "Checking your browser" in r1.text:
ct, cti = solve_challenge(r1.text)
s.cookies.set("c_token", ct, domain="zlib.li", path="/")
s.cookies.set("c_time", cti, domain="zlib.li", path="/")
r1 = s.get("https://zlib.li/", proxies=p, impersonate="chrome", timeout=10)
print("After PoW:", r1.status_code)
if r1.status_code == 503:
print(r1.text[:500])
finally:
proc.terminate()
test_clash.yaml
-13
View File
@@ -1,13 +0,0 @@
port: 7890
socks-port: 7891
proxies:
- name: "ss-obfs"
type: ss
server: "127.0.0.1"
port: 18300
cipher: "aes-128-gcm"
password: "ced"
plugin: obfs
plugin-opts:
mode: http
host: Sm0lO349ch.microsoft.com
test_cookie.py
-6
View File
@@ -1,6 +0,0 @@
from curl_cffi import requests
s = requests.Session()
s.cookies.set("1", "A", domain="zlib.li", path="/")
s.cookies.set("2", "B", domain=".zlib.li", path="/")
s.get("http://httpbin.org/cookies", impersonate="chrome")
print(s.cookies)
test_html.py
-7
View File
@@ -1,7 +0,0 @@
from zlib_dl import ProxyManager, ZLibSession
mgr = ProxyManager("proxies.txt", verbose=True)
mgr.rotate()
s = ZLibSession(proxy_url=mgr.get_current_proxy_url(), verbose=True)
res, msg = s.download_file("/dl/22909249/a93325", "test.epub")
print(msg)
import os; os.path.exists("test.epub") and print("Saved epub size:", os.path.getsize("test.epub"))
test_link.py
-16
View File
@@ -1,16 +0,0 @@
from zlib_dl import ZLibSession
import re
s1 = ZLibSession(verbose=True)
resp1 = s1._request_download("https://zlib.li/book/22909249/a93325")
match = re.search(r'href="(/dl/[^"]+)"', resp1.text)
if match:
print("Found exact link:", match.group(1))
else:
print("Could not find /dl/ link in HTML!")
lines = [line.strip() for line in resp1.text.split('\n')]
for i, line in enumerate(lines):
if "download" in line.lower() or "href" in line.lower():
if "/22909249/" in line:
print(f"Possible DL Line [{i}]: {line[:300]}")
test_mihomo.py
-22
View File
@@ -1,22 +0,0 @@
import urllib.request
import platform
system = platform.system().lower()
machine = platform.machine().lower()
is_arm = machine in ("arm64", "aarch64")
if system == "darwin":
arch_str = "arm64" if is_arm else "amd64-compatible"
elif system == "linux":
arch_str = "arm64" if is_arm else "amd64-compatible"
version = "v1.19.21"
url = f"https://github.com/MetaCubeX/mihomo/releases/download/{version}/mihomo-{system}-{arch_str}-{version}.gz"
print("URL:", url)
try:
req = urllib.request.Request(url, method='HEAD')
resp = urllib.request.urlopen(req, timeout=5)
print("Status:", resp.status)
except Exception as e:
print("Error:", e)
test_mihomo_bounce.py
-72
View File
@@ -1,72 +0,0 @@
import subprocess
import time
import os
import urllib.request
from curl_cffi import requests
# We will read the first proxy from proxies.txt to really test it
proxy_str = ""
with open("proxies.txt") as f:
for line in f:
if "ss," in line:
proxy_str = line.strip()
break
parts = proxy_str.split("=", 1)
name = parts[0].strip()
cfg = parts[1].strip().split(",")
host = cfg[1].strip()
port = cfg[2].strip()
params = {}
for p in cfg[3:]:
if "=" in p:
k, v = p.split("=", 1)
params[k.strip()] = v.strip()
method = params.get("encrypt-method", "")
password = params.get("password", "")
obfs = params.get("obfs", "")
obfs_host = params.get("obfs-host", "")
def test_mihomo(use_en0):
port_local = 11111 if use_en0 else 11112
interface_cfg = "\n interface-name: en0" if use_en0 else ""
plugin_cfg = f"""
plugin: obfs
plugin-opts:
mode: {obfs}
host: {obfs_host}
""" if obfs else ""
yaml_cfg = f"""
mode: global
mixed-port: {port_local}
bind-address: '127.0.0.1'
dns:
enable: true
nameserver:
- 223.5.5.5
proxies:
- name: "test_proxy"
type: ss
server: "{host}"
port: {port}
cipher: "{method}"
password: "{password}"{interface_cfg}{plugin_cfg}
"""
with open(f"test_cfg_{port_local}.yaml", "w") as f: f.write(yaml_cfg)
proc = subprocess.Popen(["./.venv/bin/mihomo", "-f", f"test_cfg_{port_local}.yaml"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(2)
try:
prox = {"http": f"socks5://127.0.0.1:{port_local}", "https": f"socks5://127.0.0.1:{port_local}"}
resp = requests.get("http://httpbin.org/ip", proxies=prox, impersonate="chrome", timeout=5)
return resp.json()["origin"]
except Exception as e:
return f"Error: {e}"
finally:
proc.terminate()
proc.wait(timeout=2)
os.remove(f"test_cfg_{port_local}.yaml")
print("Without en0:", test_mihomo(False))
print("With en0:", test_mihomo(True))
test_mihomo_bypass.py
-46
View File
@@ -1,46 +0,0 @@
import subprocess
import time
import os
import json
import yaml
# 1. 启动 mihomo 并绑定到具体网卡 (en0)
local_port = 10850
cfg = """
mode: global
mixed-port: 10850
bind-address: '127.0.0.1'
dns:
enable: true
nameserver:
- https://223.5.5.5/dns-query
- https://1.1.1.1/dns-query
proxies:
- name: "test-direct"
type: direct
interface-name: en0
"""
with open("test_b.yaml", "w") as f:
f.write(cfg)
# 找到 mihomo
mihomo = "./.venv/bin/mihomo"
if not os.path.exists(mihomo):
mihomo = "mihomo"
proc = subprocess.Popen([mihomo, "-f", "test_b.yaml"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(2)
try:
from curl_cffi import requests
resp = requests.get("http://httpbin.org/ip", proxies={"http": "socks5://127.0.0.1:10850", "https": "socks5://127.0.0.1:10850"}, impersonate="chrome")
print("Mihomo (direct bound to en0) IP:", resp.json()["origin"])
resp_direct = requests.get("http://httpbin.org/ip", impersonate="chrome")
print("System (Surge default) IP:", resp_direct.json()["origin"])
finally:
proc.kill()
os.remove("test_b.yaml")
test_mihomo_fix.py
-52
View File
@@ -1,52 +0,0 @@
import urllib.request, time, os, subprocess
from curl_cffi import requests
yaml_config = """
mode: rule
mixed-port: 10899
bind-address: '127.0.0.1'
proxies:
"""
proxy_str = ""
with open("proxies.txt") as f:
for line in f:
if "ss," in line:
proxy_str = line.strip()
break
parts = proxy_str.split("=", 1)
cfg = parts[1].strip().split(",")
params = dict(p.strip().split("=", 1) for p in cfg[3:] if "=" in p)
yaml_config += f""" - name: "test"
type: ss
server: "{cfg[1].strip()}"
port: {cfg[2].strip()}
cipher: "{params.get('encrypt-method', '')}"
password: "{params.get('password', '')}"
"""
if "obfs" in params:
yaml_config += f" plugin: obfs\n plugin-opts:\n mode: {params['obfs']}\n host: {params['obfs-host']}\n"
yaml_config += """
proxy-groups:
- name: "Proxy"
type: select
proxies:
- "test"
rules:
- MATCH, Proxy
"""
with open("test_503.yaml", "w") as f: f.write(yaml_config)
proc = subprocess.Popen(["./.venv/bin/mihomo", "-f", "test_503.yaml"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(2)
s = requests.Session()
p = {"http": "socks5://127.0.0.1:10899", "https": "socks5://127.0.0.1:10899"}
try:
ip_resp = s.get("http://httpbin.org/ip", proxies=p, impersonate="chrome", timeout=5)
print("Proxy IP:", ip_resp.json()["origin"])
r1 = s.get("https://zlib.li/", proxies=p, impersonate="chrome", timeout=10)
print("Index status:", r1.status_code)
finally:
proc.terminate()
test_mihomo_gen.py
-42
View File
@@ -1,42 +0,0 @@
import platform, os, gzip, urllib.request, json
import yaml
def gen_config(proxy, local_port):
cfg = {
"port": 0,
"socks-port": local_port,
"mode": "Global",
"proxies": [{
"name": proxy["name"],
"type": "ss",
"server": proxy["host"],
"port": proxy["port"],
"cipher": proxy["method"],
"password": proxy["password"],
"plugin": "obfs",
"plugin-opts": {
"mode": proxy["obfs"],
"host": proxy["obfs_host"]
}
}],
"proxy-groups": [{
"name": "Proxy",
"type": "select",
"proxies": [proxy["name"]]
}],
"rules": [
"MATCH,Proxy"
]
}
return yaml.dump(cfg)
proxy_test = {
"name": "HK 01",
"host": "d32b6cd.gylcxo.xyz",
"port": 18300,
"method": "aes-128-gcm",
"password": "ced",
"obfs": "http",
"obfs_host": "Sm0lO349ch.microsoft.com"
}
print(gen_config(proxy_test, 10800))
test_mihomo_ip.py
-19
View File
@@ -1,19 +0,0 @@
import urllib.request
import json
from bs4 import BeautifulSoup
def get_ip(proxy_url=None):
import subprocess
import sys
try:
from curl_cffi import requests
session = requests.Session()
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
resp = session.get("http://httpbin.org/ip", proxies=proxies, impersonate="chrome")
return resp.json()["origin"]
except Exception as e:
return f"Error: {e}"
print("Direct IP:", get_ip())
test_out.txt
-3
View File
@@ -1,3 +0,0 @@
⏳ 下载遇到 PoW 挑战，正在解决...
✅ 挑战已解决 (耗时 0.003s，节点 zlib.li)
Could not find /dl/ link in HTML!
test_search.py
-10
View File
@@ -1,10 +0,0 @@
from zlib_dl import ZLibSession
s = ZLibSession(verbose=True)
resp = s._request_download("https://zlib.li/s/%E6%B5%AA%E6%BD%AE%E4%B9%8B%E5%B7%85")
from bs4 import BeautifulSoup
soup = BeautifulSoup(resp.text, "html.parser")
cards = soup.find_all("z-bookcard")
if cards:
print(cards[0].prettify())
else:
print("No cards found")
test_us_clean.py
-13
View File
@@ -1,13 +0,0 @@
from zlib_dl import ZLibSession
import re
url = "https://zlib.li/book/mDJxkg7yO6/%E6%B5%AA%E6%BD%AE%E4%B9%8B%E5%B7%85%E7%AC%AC%E5%9B%9B%E7%89%88.html"
s = ZLibSession(verbose=True, proxy_url="socks5://127.0.0.1:10800")
print("Requesting clean detail URL:", url)
resp = s._request_download(url)
print("FINAL URL:", resp.url)
print("HTML Title:", re.search(r'<title>(.*?)</title>', resp.text).group(1))
match = re.search(r'href="(/dl/[^"]+)"', resp.text)
if match:
print("DOWNLOAD LINK FOUND:", match.group(1))
else:
print("NO DOWNLOAD LINK FOUND")
test_us_detail.py
-21
View File
@@ -1,21 +0,0 @@
from zlib_dl import ZLibSession
s = ZLibSession(verbose=True, proxy_url="socks5://127.0.0.1:10800")
# Hit the detail page manually
resp = s._request_download("https://zlib.li/book/mDJxkg7yO6/%E6%B5%AA%E6%BD%AE%E4%B9%8B%E5%B7%85%E7%AC%AC%E5%9B%9B%E7%89%88.html")
print(f"STATUS CODE: {resp.status_code}")
print(f"FINAL URL: {resp.url}")
content = resp.text
print(f"CONTENT LENGTH: {len(content)}")
if '/dl/' in content:
print("FOUND /dl/ literally in content!")
else:
print("NO /dl/ FOUND!")
import re
links = re.findall(r'href="(.*?)"', content)
print("Some links found on page:")
for l in [x for x in links if 'book' in x or 'dl' in x or 'download' in x][:10]:
print(" - ", l)
if "daily limit" in content.lower() or "limit reached" in content.lower() or "已达下载" in content:
print("HIT RATE LIMIT MESSAGE ON DETAIL PAGE!")
test_what_page_url_is.py
-9
View File
@@ -1,9 +0,0 @@
from zlib_dl import ZLibSession
s = ZLibSession()
results = s.search("浪潮之巅")
if results:
for i, r in enumerate(results[:3]):
print(f"Item {i}:")
print(f" title: {r.title}")
print(f" page_url: {r.page_url}")
zlib_dl.py
+220 -41
View File
@@ -198,6 +198,39 @@ class ZLibSession:
return {"https": self.proxy_url, "http": self.proxy_url}
return None
def _debug_log_exit_ip(self, stage: str):
"""在 verbose 模式下打印当前会话出口 IP，便于核对代理是否一致"""
if not self.verbose:
return
proxies = self._get_proxies()
probe_targets = [
("http://httpbin.org/ip", lambda data: data.get("origin", "unknown")),
("http://ip-api.com/json", lambda data: data.get("query", "unknown")),
("https://api.ipify.org?format=json", lambda data: data.get("ip", "unknown")),
]
proxy_desc = self.proxy_url if self.proxy_url else "直连"
last_error = None
for probe_url, parser in probe_targets:
try:
resp = self.session.get(
probe_url,
impersonate="chrome",
proxies=proxies,
timeout=15,
)
ip = parser(resp.json())
print(
f" 🌐 [{stage}] 当前出口 IP: {ip} (proxy={proxy_desc}, probe={probe_url})",
file=sys.stderr,
)
return
except Exception as e:
last_error = e
print(f" 🌐 [{stage}] 当前出口 IP 获取失败: {last_error}", file=sys.stderr)
def _request(self, url: str, **kwargs) -> cffi_requests.Response:
"""发起请求，自动处理 PoW 挑战"""
proxies = self._get_proxies()
@@ -231,6 +264,7 @@ class ZLibSession:
def search(self, query: str, max_results: int = 50) -> List[BookResult]:
"""搜索图书，返回结果列表"""
self._debug_log_exit_ip("search")
url = f"{self.BASE_URL}/s/?q={query}"
resp = self._request(url)
@@ -271,8 +305,9 @@ class ZLibSession:
def _request_download(self, url: str, **kwargs) -> cffi_requests.Response:
"""发起下载请求，自动处理 PoW 挑战（支持 allow_redirects）"""
proxies = self._get_proxies()
timeout = kwargs.pop("timeout", (20, 90))
resp = self.session.get(
url, impersonate="chrome", proxies=proxies, timeout=300,
url, impersonate="chrome", proxies=proxies, timeout=timeout,
allow_redirects=True, **kwargs
)
@@ -303,7 +338,7 @@ class ZLibSession:
# 无论什么页面，PoW 是在哪个域名的哪个具体 URL 上发生的，就全盘用带着 Cookie 的新状态重新向该域名发出最初的请求
resp = self.session.get(
resp.url, impersonate="chrome", proxies=proxies, timeout=300,
resp.url, impersonate="chrome", proxies=proxies, timeout=timeout,
allow_redirects=True, **kwargs
)
@@ -327,6 +362,7 @@ class ZLibSession:
if detail_url:
if self.verbose:
print(f" 🔗 正在预热书籍详情页以注册当前代理指纹和获取新 Session...", file=sys.stderr)
self._debug_log_exit_ip("warmup")
try:
dt_resp = self._request_download(detail_url, headers={"Referer": self.BASE_URL + "/"})
import re
@@ -348,7 +384,13 @@ class ZLibSession:
return False, f"预热书籍详情页或解析出错: {e}"
try:
resp = self._request_download(url, headers={"Referer": detail_url} if detail_url else {})
self._debug_log_exit_ip("download")
resp = self._request_download(
url,
headers={"Referer": detail_url} if detail_url else {},
stream=True,
timeout=(20, 120),
)
except Exception as e:
return False, f"下载请求失败: {e}"
@@ -376,12 +418,67 @@ class ZLibSession:
else:
return False, "下载失败，服务端依然返回了 HTML 重定向网页！"
# 写入文件
# 写入文件（流式下载：只要持续有数据就继续）
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
with open(output_path, "wb") as f:
f.write(resp.content)
tmp_path = output_path + ".part"
total_bytes = 0
return True, f"下载完成: {len(resp.content)} bytes"
try:
content_length = resp.headers.get("Content-Length", "").strip()
expected_total = int(content_length) if content_length.isdigit() else 0
if self.verbose:
if expected_total > 0:
print(
f" 📦 服务器文件大小: {expected_total / (1024 * 1024):.2f} MB",
file=sys.stderr,
)
else:
print(" 📦 服务器未返回文件大小（Content-Length）", file=sys.stderr)
started_at = time.time()
last_log_at = started_at
with open(tmp_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=262144):
if not chunk:
continue
f.write(chunk)
total_bytes += len(chunk)
now = time.time()
if self.verbose and now - last_log_at >= 3:
elapsed = max(0.001, now - started_at)
speed = total_bytes / elapsed
if expected_total > 0:
pct = (total_bytes / expected_total) * 100
print(
f" ⏬ 已下载 {total_bytes / (1024 * 1024):.2f} / {expected_total / (1024 * 1024):.2f} MB ({pct:.1f}%), 速度 {speed / (1024 * 1024):.2f} MB/s",
file=sys.stderr,
)
else:
print(
f" ⏬ 已下载 {total_bytes / (1024 * 1024):.2f} MB, 速度 {speed / (1024 * 1024):.2f} MB/s",
file=sys.stderr,
)
last_log_at = now
os.replace(tmp_path, output_path)
if expected_total > 0:
return True, f"下载完成: {total_bytes} bytes / 期望 {expected_total} bytes"
return True, f"下载完成: {total_bytes} bytes"
except Exception as e:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
return False, f"下载流中断: {e}"
finally:
try:
resp.close()
except Exception:
pass
# ============================================================
@@ -589,6 +686,12 @@ proxies:
port: {proxy.port}
cipher: "{proxy.method}"
password: "{proxy.password}"{plugin_yaml}{interface_yaml}
proxy-groups:
- name: GLOBAL
type: select
proxies:
- "{proxy.name}"
"""
with open(cfg_path, 'w', encoding='utf-8') as f:
f.write(yaml_config)
@@ -679,6 +782,18 @@ proxies:
proxy = self.proxies[self.current_index]
return proxy.downloads_count >= self.MAX_DOWNLOADS_PER_PROXY
def has_proxy_pool(self) -> bool:
"""是否配置了可轮换的代理池"""
return len(self.proxies) > 0
def ensure_active(self) -> bool:
"""确保当前有活跃代理（有代理池时）"""
if not self.proxies:
return True
if 0 <= self.current_index < len(self.proxies):
return True
return self.rotate()
def rotate(self) -> bool:
"""
轮换到下一个代理。
@@ -991,16 +1106,53 @@ def sanitize_filename(name: str) -> str:
return name.strip()[:200]
def _should_rotate_proxy(message: str) -> bool:
"""根据错误信息判断是否应该切换代理重试"""
msg_lower = message.lower()
return (
"限" in message
or "limit" in msg_lower
or "429" in message
or "503" in message
or "504" in message
or "403" in message
or "502" in message
or "timeout" in msg_lower
or "请求失败" in message
or "html" in msg_lower
)
def _create_session_for_current_proxy(
proxy_mgr: ProxyManager,
explicit_proxy: Optional[str],
verbose: bool,
) -> ZLibSession:
"""根据当前代理状态创建新的 ZLibSession"""
proxy_url = explicit_proxy if explicit_proxy else proxy_mgr.get_current_proxy_url()
return ZLibSession(proxy_url=proxy_url, verbose=verbose)
def cmd_download(args):
"""执行下载命令"""
explicit_proxy = getattr(args, "proxy", None)
proxy_file = getattr(args, "proxy_file", None)
if explicit_proxy and proxy_file:
print("错误: --proxy 与 --proxy-file 不能同时使用", file=sys.stderr)
sys.exit(1)
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# 代理管理器
proxy_mgr = ProxyManager(
proxy_file=getattr(args, "proxy_file", None), verbose=args.verbose
proxy_file=proxy_file, verbose=args.verbose
)
if proxy_mgr.has_proxy_pool() and not proxy_mgr.ensure_active():
print("错误: 代理池不可用，无法启动任何代理", file=sys.stderr)
sys.exit(1)
download_results: List[DownloadResult] = []
try:
@@ -1014,11 +1166,13 @@ def cmd_download(args):
for dl_path in dl_paths:
result = _download_one(
dl_path=dl_path,
page_url="",
title=title,
author=author,
fmt=fmt or "",
output_dir=output_dir,
proxy_mgr=proxy_mgr,
explicit_proxy=explicit_proxy,
verbose=args.verbose,
)
download_results.append(result)
@@ -1037,8 +1191,11 @@ def cmd_download(args):
print(f" 🔍 搜索: {query_str}", file=sys.stderr)
# 搜索
proxy_url = proxy_mgr.get_current_proxy_url()
session = ZLibSession(proxy_url=proxy_url, verbose=args.verbose)
session = _create_session_for_current_proxy(
proxy_mgr=proxy_mgr,
explicit_proxy=explicit_proxy,
verbose=args.verbose,
)
results = session.search(query_str)
if not results:
@@ -1066,6 +1223,7 @@ def cmd_download(args):
fmt=book.format,
output_dir=output_dir,
proxy_mgr=proxy_mgr,
explicit_proxy=explicit_proxy,
verbose=args.verbose,
)
download_results.append(result)
@@ -1082,6 +1240,7 @@ def cmd_download(args):
fmt=book.format,
output_dir=output_dir,
proxy_mgr=proxy_mgr,
explicit_proxy=explicit_proxy,
verbose=args.verbose,
)
download_results.append(result)
@@ -1120,6 +1279,7 @@ def _download_one(
fmt: str,
output_dir: Path,
proxy_mgr: ProxyManager,
explicit_proxy: Optional[str],
verbose: bool,
) -> DownloadResult:
"""下载单个文件，带代理轮询逻辑"""
@@ -1147,9 +1307,10 @@ def _download_one(
message="文件已存在，跳过",
)
# 检查代理轮换
if proxy_mgr.should_rotate():
if not proxy_mgr.rotate():
uses_proxy_pool = proxy_mgr.has_proxy_pool()
# 进入下载前，若当前代理已达配额，先轮换
if uses_proxy_pool and proxy_mgr.should_rotate() and not proxy_mgr.rotate():
return DownloadResult(
status="error",
title=title,
@@ -1160,22 +1321,41 @@ def _download_one(
message="所有代理已用完，且当前代理下载次数已达上限",
)
proxy_url = proxy_mgr.get_current_proxy_url()
proxy_name = proxy_mgr.get_current_proxy_name()
if verbose:
print(
f" ⬇️ 下载: {filename} (代理: {proxy_name})", file=sys.stderr
while True:
if uses_proxy_pool and not proxy_mgr.ensure_active():
return DownloadResult(
status="error",
title=title,
author=author,
format=fmt,
filesize="",
filepath="",
message="所有代理已用完",
)
# 完全创建新 Session，废弃之前的旧 Session 以获取全新的 CDN 下载授权
session = ZLibSession(proxy_url=proxy_url, verbose=verbose)
if uses_proxy_pool:
proxy_name = proxy_mgr.get_current_proxy_name()
elif explicit_proxy:
proxy_name = explicit_proxy
else:
proxy_name = "直连"
if verbose:
print(f" ⬇️ 下载: {filename} (代理: {proxy_name})", file=sys.stderr)
# 每次下载尝试都创建全新 Session；切代理后会自动带新出口重建会话
session = _create_session_for_current_proxy(
proxy_mgr=proxy_mgr,
explicit_proxy=explicit_proxy,
verbose=verbose,
)
success, message = session.download_file(dl_path, page_url, output_path)
if success:
if uses_proxy_pool:
proxy_mgr.record_download()
filesize = os.path.getsize(output_path)
size_str = format_size(filesize)
size_str = _format_size(filesize)
return DownloadResult(
status="success",
title=title,
@@ -1186,26 +1366,8 @@ def _download_one(
message=f"下载成功 ({size_str})",
proxy_used=proxy_name,
)
else:
# 如果是限制错误（达到配额）或遇到 503/403 等代理被屏蔽导致的无限验证循环，尝试切换代理重试
msg_lower = message.lower()
should_retry = (
"限" in message or "limit" in msg_lower or
"503" in message or "403" in message or "502" in message or
"timeout" in msg_lower or "请求失败" in message or "html" in msg_lower
)
if should_retry:
if proxy_mgr.rotate():
if verbose:
print(
f" 🔄 下载受阻 ({message})，切换代理重试: {proxy_mgr.get_current_proxy_name()}",
file=sys.stderr,
)
return _download_one(
dl_path, page_url, title, author, fmt, output_dir, proxy_mgr, verbose
)
if not uses_proxy_pool or not _should_rotate_proxy(message):
return DownloadResult(
status="error",
title=title,
@@ -1216,6 +1378,23 @@ def _download_one(
message=message,
)
if not proxy_mgr.rotate():
return DownloadResult(
status="error",
title=title,
author=author,
format=fmt,
filesize="",
filepath="",
message=f"{message}；且所有代理已用完",
)
if verbose:
print(
f" 🔄 下载受阻 ({message})，切换代理重试: {proxy_mgr.get_current_proxy_name()}",
file=sys.stderr,
)
def _format_size(size_bytes: int) -> str:
"""格式化文件大小"""
for unit in ["B", "KB", "MB", "GB"]: