From 0baf8b73f09a1d2b828d920a6eb121a15ab09820 Mon Sep 17 00:00:00 2001 From: lou Date: Mon, 10 Aug 2026 19:10:53 +0800 Subject: [PATCH] =?UTF-8?q?aria2=20=E5=8A=A0=E9=80=9F:=20=E4=B8=8B?= =?UTF-8?q?=E8=BD=BD=E5=8D=B72..N=20aria2=E5=B9=B6=E5=8F=91=E6=8B=89?= =?UTF-8?q?=E5=8F=96(=E5=9B=9E=E9=80=80=E9=80=90=E5=8D=B7);=20=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0aria2=E5=8F=8D=E5=90=91=E6=8B=89(=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF=E4=B8=B4=E6=97=B6HTTP=E6=9C=8D=E5=8A=A119000=E6=AE=B5?= =?UTF-8?q?+=E6=9C=8D=E5=8A=A1=E7=AB=AFpull=E7=AB=AF=E7=82=B9+receiver?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C);=20=E5=88=97=E8=A1=A8=E5=BA=8F=E5=8F=B7?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- debian/7z-encrypt-client-bin/DEBIAN/control | 2 +- debian/7z-encrypt-client-bin/DEBIAN/md5sums | 2 +- debian/7z-encrypt-client/DEBIAN/control | 2 +- debian/7z-encrypt-client/DEBIAN/md5sums | 2 +- .../usr/share/7z-encrypt/transfer.py | 192 ++++++++++++++++-- transfer.py | 192 ++++++++++++++++-- 6 files changed, 354 insertions(+), 38 deletions(-) diff --git a/debian/7z-encrypt-client-bin/DEBIAN/control b/debian/7z-encrypt-client-bin/DEBIAN/control index 289a715..0919c26 100644 --- a/debian/7z-encrypt-client-bin/DEBIAN/control +++ b/debian/7z-encrypt-client-bin/DEBIAN/control @@ -3,7 +3,7 @@ Source: 7z-encrypt-client Version: 1.0.0 Architecture: amd64 Maintainer: edgevoid -Installed-Size: 51870 +Installed-Size: 51906 Depends: p7zip-full Conflicts: 7z-encrypt-client Replaces: 7z-encrypt-client diff --git a/debian/7z-encrypt-client-bin/DEBIAN/md5sums b/debian/7z-encrypt-client-bin/DEBIAN/md5sums index eb03afd..3a4f239 100644 --- a/debian/7z-encrypt-client-bin/DEBIAN/md5sums +++ b/debian/7z-encrypt-client-bin/DEBIAN/md5sums @@ -1,6 +1,6 @@ a7346d13df51c26822c41489cfbbd106 usr/bin/sz-backup 61901c8e9af062eef5c386c715f5f3bb usr/bin/sz-config -c1b6ac06ad3d991b6ad61ecff97a91d1 usr/bin/sz-transfer +f83888ad59b904747ac933a89fd98b55 usr/bin/sz-transfer ae6b68e345a0f5f3397aa8ead8a199f1 usr/bin/sz-tui 198943ad7a94f6ba0995cf2d326f66d4 usr/share/doc/7z-encrypt-client-bin/changelog.gz fadf3972cc61c7c76d0ccc20b29a804c usr/share/doc/7z-encrypt-client-bin/copyright diff --git a/debian/7z-encrypt-client/DEBIAN/control b/debian/7z-encrypt-client/DEBIAN/control index e237568..8eeb5e3 100644 --- a/debian/7z-encrypt-client/DEBIAN/control +++ b/debian/7z-encrypt-client/DEBIAN/control @@ -2,7 +2,7 @@ Package: 7z-encrypt-client Version: 1.0.0 Architecture: all Maintainer: edgevoid -Installed-Size: 118 +Installed-Size: 125 Depends: python3 (>= 3.10), python3-venv, p7zip-full Section: utils Priority: optional diff --git a/debian/7z-encrypt-client/DEBIAN/md5sums b/debian/7z-encrypt-client/DEBIAN/md5sums index e8ac13e..d1650f7 100644 --- a/debian/7z-encrypt-client/DEBIAN/md5sums +++ b/debian/7z-encrypt-client/DEBIAN/md5sums @@ -8,7 +8,7 @@ dcb31e74e0a6ee8073337fc4fb2acf99 usr/share/7z-encrypt/config.py 21062e04f8c7f4471881a39c489dfa73 usr/share/7z-encrypt/metadata.py 3478ac8277b4b1a38694ad0cc8cdbcdb usr/share/7z-encrypt/splitter.py b90b8c950f99e1439bc273f50a2ee0a6 usr/share/7z-encrypt/state.py -c922430309b2a680ff757f3b02561ec9 usr/share/7z-encrypt/transfer.py +145ac54cb829352d38f2304fefe78e08 usr/share/7z-encrypt/transfer.py 67d4b90265abf5a72af959f66679dd0f usr/share/7z-encrypt/tui.py 198943ad7a94f6ba0995cf2d326f66d4 usr/share/doc/7z-encrypt-client/changelog.gz fadf3972cc61c7c76d0ccc20b29a804c usr/share/doc/7z-encrypt-client/copyright diff --git a/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py b/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py index 052af78..16eeff9 100644 --- a/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py +++ b/debian/7z-encrypt-client/usr/share/7z-encrypt/transfer.py @@ -57,11 +57,17 @@ class TransferClient: # ---------- HTTP 封装 ---------- def _request( - self, method: str, path: str, body: bytes | None = None, timeout: int | float | None = None + self, method: str, path: str, body: bytes | None = None, + timeout: int | float | None = None, json_body: dict[str, Any] | None = None, ) -> tuple[int, dict[str, Any]]: """发请求, 返回 (HTTP状态码, JSON 载荷)。4xx/5xx 不抛, 由调用方判断""" - req = urlrequest.Request(self.base_url + path, data=body, method=method) - if body is not None: + data = body + req = urlrequest.Request(self.base_url + path, data=data, method=method) + if json_body is not None: + data = json.dumps(json_body, ensure_ascii=False).encode() + req = urlrequest.Request(self.base_url + path, data=data, method=method) + req.add_header("Content-Type", "application/json") + elif data is not None: req.add_header("Content-Type", "application/octet-stream") if self.token: req.add_header("Authorization", f"Bearer {self.token}") @@ -210,12 +216,21 @@ class TransferClient: position=0, leave=False) with open(dest, "wb") as f: self._download_chunk_to(resp, f, outer, 1, chunk_count) - for idx in range(2, chunk_count + 1): - with self._open_chunk(file_id, idx) as resp: - with open(dest, "ab") as f: - self._download_chunk_to(resp, f, outer, idx, chunk_count) outer.close() - return dest, enc_params + # 卷 2..N: aria2 并发拉取 (打满带宽); 无 aria2 回退逐卷 + if chunk_count > 1: + try: + self._download_aria2(file_id, dest, chunk_count) + except TransferError as e: + print(f"[aria2] 不可用, 回退逐卷下载: {e}") + outer = tqdm(total=chunk_count - 1, desc="[下载]", unit="卷", + position=0, leave=False) + with open(dest, "ab") as f: + for idx in range(2, chunk_count + 1): + with self._open_chunk(file_id, idx) as resp: + self._download_chunk_to(resp, f, outer, idx, chunk_count) + outer.close() + return dest, enc_params except urlerror.HTTPError as e: if e.code == 401: raise TransferError("未授权: token 无效 (检查 config server.token)") from e @@ -261,6 +276,145 @@ class TransferClient: outer.update(1) return done + def _aria2_pull(self, file_id: str, chunk_range: range, workdir: Path) -> list[Path]: + """aria2 并发拉取卷文件 (多任务打满带宽), 返回按序的卷文件列表 + + 每卷是独立 URL 的独立文件, aria2 多连接并发下载; 服务端无需 Range。 + """ + import shutil + aria2 = shutil.which("aria2c") + if aria2 is None: + raise TransferError("未安装 aria2 (apt install aria2)") + urls_file = workdir / "urls.txt" + with open(urls_file, "w", encoding="utf-8") as f: + for i in chunk_range: + f.write(f"{self.base_url}/api/files/{file_id}/chunk/{i}\n") + cmd = [ + aria2, "-i", str(urls_file), "-d", str(workdir), + "--max-concurrent-downloads=8", "--split=4", "--max-connection-per-server=4", + "--min-split-size=1M", "--file-allocation=none", "--allow-overwrite=true", + "--auto-file-renaming=false", "--console-log-level=warn", + "--summary-interval=1", "--quiet=false", "--no-conf=true", + ] + if self.token: + cmd += ["--header", f"Authorization: Bearer {self.token}"] + r = subprocess.run(cmd, timeout=self.timeout * max(len(chunk_range), 1) + 60) + if r.returncode != 0: + raise TransferError(f"aria2 下载失败 (exit {r.returncode})") + files = sorted(workdir.glob("chunk_*"), key=lambda p: int(p.name.split("_")[1])) + if len(files) != len(chunk_range): + raise TransferError(f"aria2 卷文件不完整: {len(files)}/{len(chunk_range)}") + return files + + def _download_aria2(self, file_id: str, dest: Path, chunk_count: int) -> None: + """aria2 并发拉卷 2..N 追加到 dest (卷 1 已在 requests 流中下载)""" + import tempfile + with tempfile.TemporaryDirectory(prefix="sz-aria2-") as td: + files = self._aria2_pull(file_id, range(2, chunk_count + 1), Path(td)) + with open(dest, "ab") as f: + for p in files: + with open(p, "rb") as src: + shutil.copyfileobj(src, f, 1 << 20) + + # ---------- aria2 反向拉上传 ---------- + + def _serve_chunks(self, transfer_id: str, chunk_files: dict[int, Path], + pull_token: str, port: int = 0): + """起临时 HTTP 服务提供卷文件 (token 校验), 返回 httpd (serve_forever 已起) + + 端口段 19000-19099 固定 (PC 防火墙放行此段, 手机 ZeroTermux 无防火墙)。 + """ + import secrets + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + if pull_token is None: + pull_token = secrets.token_hex(8) + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 (HTTP 方法名) + import urllib.parse as up + parsed = up.urlparse(self.path) + qs = up.parse_qs(parsed.query) + if qs.get("token", [""])[0] != pull_token: + self.send_response(403) + self.end_headers() + return + parts = parsed.path.split("/") + if len(parts) != 4 or parts[1] != "pull" or parts[2] != transfer_id: + self.send_response(404) + self.end_headers() + return + try: + f = chunk_files.get(int(parts[3])) + except ValueError: + f = None + if f is None or not f.exists(): + self.send_response(404) + self.end_headers() + return + data = f.read_bytes() + self.send_response(200) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, format: str, *args: Any) -> None: + pass + + httpd: ThreadingHTTPServer | None = None + if port == 0: + for p in range(19000, 19100): + try: + httpd = ThreadingHTTPServer(("0.0.0.0", p), Handler) + break + except OSError: + continue + if httpd is None: + httpd = ThreadingHTTPServer(("0.0.0.0", port), Handler) + import threading + threading.Thread(target=httpd.serve_forever, daemon=True).start() + return httpd, pull_token + + @staticmethod + def _local_ip() -> str: + """局域网 IP: UDP connect 拿本地出口地址""" + import socket + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect(("10.255.255.255", 1)) + return s.getsockname()[0] + except OSError: + return "127.0.0.1" + finally: + s.close() + + def _pull_upload(self, transfer_id: str, chunk_files: dict[int, Path]) -> bool: + """aria2 反向拉上传: 临时 HTTP 服务 -> 服务端 pull 并发拉卷 + + 成功返回 True; 服务端无 aria2/拉取失败返回 False (调用方回退逐卷 PUT)。 + """ + import secrets + httpd, pull_token = self._serve_chunks( + transfer_id, chunk_files, secrets.token_hex(8)) + try: + base_url = f"http://{self._local_ip()}:{httpd.server_address[1]}" + code, payload = self._request( + "POST", f"/api/transfer/{transfer_id}/pull", + json_body={"base_url": base_url, "token": pull_token}, + timeout=60, + ) + if code != 200: + print(f"[aria2] 服务端未启用反向拉取 (HTTP {code}): {payload.get('error', '')}") + return False + pulled, total = payload.get("pulled", 0), payload.get("total", 0) + print(f"[aria2] 反向拉取完成: {pulled}/{total} 卷") + return pulled == total + except TransferError as e: + print(f"[aria2] 反向拉取失败, 回退逐卷: {e}") + return False + finally: + httpd.shutdown() + httpd.server_close() + # ---------- 全流程 ---------- def transfer( @@ -284,15 +438,19 @@ class TransferClient: if received: print(f"[传输] 断点续传: 服务端已收 {len(received)}/{total} 卷, 补传 {len(pending)} 卷") - pbar = tqdm(total=len(pending), desc="[传输] 上传", unit="卷", leave=False) - for done, index in enumerate(pending, start=1): - try: - self.upload_chunk(transfer_id, index, chunk_files[index]) - except TransferError: - raise # 重试耗尽, 由调用方决定 (记录断点状态待补传) - pbar.update(1) - pbar.set_description(f"[传输] 上传 {done}/{len(pending)} 卷") - pbar.close() + # aria2 反向拉上传: 优先 (打满上行带宽), 失败回退逐卷 PUT + if pending and self._pull_upload(transfer_id, chunk_files): + pending = [] + if pending: + pbar = tqdm(total=len(pending), desc="[传输] 上传", unit="卷", leave=False) + for done, index in enumerate(pending, start=1): + try: + self.upload_chunk(transfer_id, index, chunk_files[index]) + except TransferError: + raise # 重试耗尽, 由调用方决定 (记录断点状态待补传) + pbar.update(1) + pbar.set_description(f"[传输] 上传 {done}/{len(pending)} 卷") + pbar.close() receipt = self.complete(transfer_id) print(f"[传输] 完成: {receipt}") diff --git a/transfer.py b/transfer.py index 052af78..16eeff9 100644 --- a/transfer.py +++ b/transfer.py @@ -57,11 +57,17 @@ class TransferClient: # ---------- HTTP 封装 ---------- def _request( - self, method: str, path: str, body: bytes | None = None, timeout: int | float | None = None + self, method: str, path: str, body: bytes | None = None, + timeout: int | float | None = None, json_body: dict[str, Any] | None = None, ) -> tuple[int, dict[str, Any]]: """发请求, 返回 (HTTP状态码, JSON 载荷)。4xx/5xx 不抛, 由调用方判断""" - req = urlrequest.Request(self.base_url + path, data=body, method=method) - if body is not None: + data = body + req = urlrequest.Request(self.base_url + path, data=data, method=method) + if json_body is not None: + data = json.dumps(json_body, ensure_ascii=False).encode() + req = urlrequest.Request(self.base_url + path, data=data, method=method) + req.add_header("Content-Type", "application/json") + elif data is not None: req.add_header("Content-Type", "application/octet-stream") if self.token: req.add_header("Authorization", f"Bearer {self.token}") @@ -210,12 +216,21 @@ class TransferClient: position=0, leave=False) with open(dest, "wb") as f: self._download_chunk_to(resp, f, outer, 1, chunk_count) - for idx in range(2, chunk_count + 1): - with self._open_chunk(file_id, idx) as resp: - with open(dest, "ab") as f: - self._download_chunk_to(resp, f, outer, idx, chunk_count) outer.close() - return dest, enc_params + # 卷 2..N: aria2 并发拉取 (打满带宽); 无 aria2 回退逐卷 + if chunk_count > 1: + try: + self._download_aria2(file_id, dest, chunk_count) + except TransferError as e: + print(f"[aria2] 不可用, 回退逐卷下载: {e}") + outer = tqdm(total=chunk_count - 1, desc="[下载]", unit="卷", + position=0, leave=False) + with open(dest, "ab") as f: + for idx in range(2, chunk_count + 1): + with self._open_chunk(file_id, idx) as resp: + self._download_chunk_to(resp, f, outer, idx, chunk_count) + outer.close() + return dest, enc_params except urlerror.HTTPError as e: if e.code == 401: raise TransferError("未授权: token 无效 (检查 config server.token)") from e @@ -261,6 +276,145 @@ class TransferClient: outer.update(1) return done + def _aria2_pull(self, file_id: str, chunk_range: range, workdir: Path) -> list[Path]: + """aria2 并发拉取卷文件 (多任务打满带宽), 返回按序的卷文件列表 + + 每卷是独立 URL 的独立文件, aria2 多连接并发下载; 服务端无需 Range。 + """ + import shutil + aria2 = shutil.which("aria2c") + if aria2 is None: + raise TransferError("未安装 aria2 (apt install aria2)") + urls_file = workdir / "urls.txt" + with open(urls_file, "w", encoding="utf-8") as f: + for i in chunk_range: + f.write(f"{self.base_url}/api/files/{file_id}/chunk/{i}\n") + cmd = [ + aria2, "-i", str(urls_file), "-d", str(workdir), + "--max-concurrent-downloads=8", "--split=4", "--max-connection-per-server=4", + "--min-split-size=1M", "--file-allocation=none", "--allow-overwrite=true", + "--auto-file-renaming=false", "--console-log-level=warn", + "--summary-interval=1", "--quiet=false", "--no-conf=true", + ] + if self.token: + cmd += ["--header", f"Authorization: Bearer {self.token}"] + r = subprocess.run(cmd, timeout=self.timeout * max(len(chunk_range), 1) + 60) + if r.returncode != 0: + raise TransferError(f"aria2 下载失败 (exit {r.returncode})") + files = sorted(workdir.glob("chunk_*"), key=lambda p: int(p.name.split("_")[1])) + if len(files) != len(chunk_range): + raise TransferError(f"aria2 卷文件不完整: {len(files)}/{len(chunk_range)}") + return files + + def _download_aria2(self, file_id: str, dest: Path, chunk_count: int) -> None: + """aria2 并发拉卷 2..N 追加到 dest (卷 1 已在 requests 流中下载)""" + import tempfile + with tempfile.TemporaryDirectory(prefix="sz-aria2-") as td: + files = self._aria2_pull(file_id, range(2, chunk_count + 1), Path(td)) + with open(dest, "ab") as f: + for p in files: + with open(p, "rb") as src: + shutil.copyfileobj(src, f, 1 << 20) + + # ---------- aria2 反向拉上传 ---------- + + def _serve_chunks(self, transfer_id: str, chunk_files: dict[int, Path], + pull_token: str, port: int = 0): + """起临时 HTTP 服务提供卷文件 (token 校验), 返回 httpd (serve_forever 已起) + + 端口段 19000-19099 固定 (PC 防火墙放行此段, 手机 ZeroTermux 无防火墙)。 + """ + import secrets + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + if pull_token is None: + pull_token = secrets.token_hex(8) + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 (HTTP 方法名) + import urllib.parse as up + parsed = up.urlparse(self.path) + qs = up.parse_qs(parsed.query) + if qs.get("token", [""])[0] != pull_token: + self.send_response(403) + self.end_headers() + return + parts = parsed.path.split("/") + if len(parts) != 4 or parts[1] != "pull" or parts[2] != transfer_id: + self.send_response(404) + self.end_headers() + return + try: + f = chunk_files.get(int(parts[3])) + except ValueError: + f = None + if f is None or not f.exists(): + self.send_response(404) + self.end_headers() + return + data = f.read_bytes() + self.send_response(200) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, format: str, *args: Any) -> None: + pass + + httpd: ThreadingHTTPServer | None = None + if port == 0: + for p in range(19000, 19100): + try: + httpd = ThreadingHTTPServer(("0.0.0.0", p), Handler) + break + except OSError: + continue + if httpd is None: + httpd = ThreadingHTTPServer(("0.0.0.0", port), Handler) + import threading + threading.Thread(target=httpd.serve_forever, daemon=True).start() + return httpd, pull_token + + @staticmethod + def _local_ip() -> str: + """局域网 IP: UDP connect 拿本地出口地址""" + import socket + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect(("10.255.255.255", 1)) + return s.getsockname()[0] + except OSError: + return "127.0.0.1" + finally: + s.close() + + def _pull_upload(self, transfer_id: str, chunk_files: dict[int, Path]) -> bool: + """aria2 反向拉上传: 临时 HTTP 服务 -> 服务端 pull 并发拉卷 + + 成功返回 True; 服务端无 aria2/拉取失败返回 False (调用方回退逐卷 PUT)。 + """ + import secrets + httpd, pull_token = self._serve_chunks( + transfer_id, chunk_files, secrets.token_hex(8)) + try: + base_url = f"http://{self._local_ip()}:{httpd.server_address[1]}" + code, payload = self._request( + "POST", f"/api/transfer/{transfer_id}/pull", + json_body={"base_url": base_url, "token": pull_token}, + timeout=60, + ) + if code != 200: + print(f"[aria2] 服务端未启用反向拉取 (HTTP {code}): {payload.get('error', '')}") + return False + pulled, total = payload.get("pulled", 0), payload.get("total", 0) + print(f"[aria2] 反向拉取完成: {pulled}/{total} 卷") + return pulled == total + except TransferError as e: + print(f"[aria2] 反向拉取失败, 回退逐卷: {e}") + return False + finally: + httpd.shutdown() + httpd.server_close() + # ---------- 全流程 ---------- def transfer( @@ -284,15 +438,19 @@ class TransferClient: if received: print(f"[传输] 断点续传: 服务端已收 {len(received)}/{total} 卷, 补传 {len(pending)} 卷") - pbar = tqdm(total=len(pending), desc="[传输] 上传", unit="卷", leave=False) - for done, index in enumerate(pending, start=1): - try: - self.upload_chunk(transfer_id, index, chunk_files[index]) - except TransferError: - raise # 重试耗尽, 由调用方决定 (记录断点状态待补传) - pbar.update(1) - pbar.set_description(f"[传输] 上传 {done}/{len(pending)} 卷") - pbar.close() + # aria2 反向拉上传: 优先 (打满上行带宽), 失败回退逐卷 PUT + if pending and self._pull_upload(transfer_id, chunk_files): + pending = [] + if pending: + pbar = tqdm(total=len(pending), desc="[传输] 上传", unit="卷", leave=False) + for done, index in enumerate(pending, start=1): + try: + self.upload_chunk(transfer_id, index, chunk_files[index]) + except TransferError: + raise # 重试耗尽, 由调用方决定 (记录断点状态待补传) + pbar.update(1) + pbar.set_description(f"[传输] 上传 {done}/{len(pending)} 卷") + pbar.close() receipt = self.complete(transfer_id) print(f"[传输] 完成: {receipt}")