# -*- coding: utf-8 -*- r"""中转网站加"下载"区 —— 把成品直接挂在管理页上。 用户要求："把下载链接 release 放到那个中转网站上" 做法： 1. 中转加一个下载目录（relay/downloads/） 2. GET /download/<文件名> 直接给文件（带正确的 Content-Type 和文件名） 3. GET /v1/downloads 列出有哪些文件（名字 + 大小） 4. 管理页加一张"下载"卡片，列出文件并给可点的链接 5. 部署脚本把 dist/ 里的成品传上去 安全： 只允许下载 downloads/ 目录里的**白名单后缀**（.apk/.zip/.txt）， 并且文件名做严格校验（不许有 .. / \ 之类），防止目录穿越。 """ from __future__ import annotations import ast import sys from pathlib import Path RELAY = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\relay" r"\relay.py") # ---------------- 1. 下载目录常量 ---------------- ANCHOR_DIR = '''ADMIN_SESSIONS: dict = {}''' NEW_DIR = '''# 成品下载目录（管理页上给人下载 APK / 整包用）。 # 只放白名单后缀的文件，文件名做严格校验防目录穿越。 DOWNLOADS_DIR = Path(__file__).resolve().parent / "downloads" DOWNLOAD_SUFFIXES = (".apk", ".zip", ".txt", ".exe", ".idsig") def _safe_download_name(name: str) -> str: """校验下载文件名。不合法返回空串。 防目录穿越：只允许字母数字、点、横线、下划线、中文， 且必须是白名单后缀。 """ name = (name or "").strip() if not name or len(name) > 120: return "" if "/" in name or "\\\\" in name or ".." in name: return "" if not name.lower().endswith(DOWNLOAD_SUFFIXES): return "" for ch in name: if ch.isalnum() or ch in "._-（）() 　": continue return "" return name def _list_downloads() -> list: """列出可下载的文件（按修改时间倒序）。""" items = [] try: if not DOWNLOADS_DIR.is_dir(): return items for entry in DOWNLOADS_DIR.iterdir(): if not entry.is_file(): continue if not _safe_download_name(entry.name): continue stat = entry.stat() items.append({ "name": entry.name, "size": stat.st_size, "mtime": stat.st_mtime, }) except OSError: return items items.sort(key=lambda item: item["mtime"], reverse=True) return items ''' + ANCHOR_DIR # ---------------- 2. 下载路由 ---------------- ANCHOR_ROUTE = ''' if path in ("/v1/health", "/health"):''' NEW_ROUTE = ''' # ---- 成品下载（挂在管理页上给人下 APK / 整包）---- if path == "/v1/downloads": self._json(200, {"ok": True, "files": _list_downloads()}) return if path.startswith("/download/"): wanted = urllib.parse.unquote(path[len("/download/"):]) safe = _safe_download_name(wanted) if not safe: self._json(400, {"ok": False, "error": "文件名不合法"}) return target = DOWNLOADS_DIR / safe if not target.is_file(): self._json(404, {"ok": False, "error": "没有这个文件"}) return try: blob = target.read_bytes() except OSError as exc: self._json(500, {"ok": False, "error": f"读取失败: {exc}"}) return if safe.lower().endswith(".apk"): content_type = "application/vnd.android.package-archive" elif safe.lower().endswith(".zip"): content_type = "application/zip" elif safe.lower().endswith(".txt"): content_type = "text/plain; charset=utf-8" else: content_type = "application/octet-stream" self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(blob))) # 文件名可能含中文，用 RFC 5987 编码 quoted = urllib.parse.quote(safe) self.send_header( "Content-Disposition", f"attachment; filename*=UTF-8''{quoted}") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(blob) log(f"下载：{safe}（{len(blob)} 字节）") return ''' + ANCHOR_ROUTE # ---------------- 3. 管理页卡片 ---------------- ANCHOR_CARD = '''
''' NEW_CARD = '''
下载 —— 最新版本
正在读…
''' # ---------------- 4. 前端 JS ---------------- ANCHOR_JS = '''// ---------------- 自动刷新 ----------------''' NEW_JS = '''// ---------------- 下载区 ---------------- function humanSize(bytes) {{ if (bytes < 1024) return bytes + ' B'; if (bytes < 1024 * 1024) return (bytes / 1024).toFixed(1) + ' KB'; return (bytes / 1024 / 1024).toFixed(2) + ' MB'; }} (async () => {{ const body = document.getElementById('dlBody'); if (!body) return; try {{ const r = await fetch('/v1/downloads'); const j = await r.json(); const files = (j && j.files) || []; if (!files.length) {{ body.innerHTML = '还没有上传成品。' + '把 APK / 整包放到服务器的 downloads 目录就会出现在这里。'; return; }} body.innerHTML = files.map(f => {{ const isApk = f.name.toLowerCase().endsWith('.apk'); const tag = isApk ? '手表 APK' : '整包'; return '
' + tag + '' + f.name + '' + '' + humanSize(f.size) + '' + '
'; }}).join(''); }} catch (e) {{ body.innerHTML = '读不到：' + e + ''; }} }})(); // ---------------- 自动刷新 ----------------''' def main() -> int: text = RELAY.read_text(encoding="utf-8") if "/download/" in text and "dlPanel" in text: print(" 已经加过") return 0 ok = True for old, new, label in ((ANCHOR_DIR, NEW_DIR, "下载目录常量+工具函数"), (ANCHOR_ROUTE, NEW_ROUTE, "下载路由"), (ANCHOR_CARD, NEW_CARD, "管理页下载卡片"), (ANCHOR_JS, NEW_JS, "前端 JS")): if old not in text: print(f" [失败] {label} —— 锚点没找到") ok = False continue if text.count(old) != 1: print(f" [警告] {label} 锚点出现 {text.count(old)} 次") text = text.replace(old, new, 1) print(f" {label}") RELAY.write_text(text, encoding="utf-8", newline="\n") try: ast.parse(text) print(" relay.py 语法 OK") except SyntaxError as exc: print(f" relay.py 语法错误: 行 {exc.lineno}: {exc.msg}") ok = False return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())