ykxmz/_build/add_downloads.py

# -*- 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 = ''' <div class=stats>'''

NEW_CARD = ''' <div class=panel id=dlPanel>
   <div class=label>下载 —— 最新版本</div>
   <div id=dlBody class=tip>正在读…</div>
 </div>

 <div class=stats>'''

# ---------------- 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 = '<span style="color:#8b949e">还没有上传成品。'
        + '把 APK / 整包放到服务器的 downloads 目录就会出现在这里。</span>';
      return;
    }}
    body.innerHTML = files.map(f => {{
      const isApk = f.name.toLowerCase().endsWith('.apk');
      const tag = isApk
        ? '<span class="badge on">手表 APK</span>'
        : '<span class="badge off">整包</span>';
      return '<div style="display:flex;align-items:center;gap:10px;'
        + 'margin:6px 0;flex-wrap:wrap">'
        + tag
        + '<a href="/download/' + encodeURIComponent(f.name) + '" '
        + 'style="color:#79c0ff;text-decoration:none;font-weight:500">'
        + f.name + '</a>'
        + '<span style="color:#6e7681;font-size:12px">'
        + humanSize(f.size) + '</span>'
        + '</div>';
    }}).join('');
  }} catch (e) {{
    body.innerHTML = '<span style="color:#f85149">读不到：' + e + '</span>';
  }}
}})();

// ---------------- 自动刷新 ----------------'''


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())