#!/usr/bin/env python3 """ 星潮 APK 上传服务器 - GET / → 上传页面 - GET / → 静态文件下载 (兼容旧 SimpleHTTP) - POST /upload → 上传 APK (自动备份旧文件 + 替换 xcparty.apk + 验证) """ import http.server import os import json import hashlib import shutil import cgi import time from datetime import datetime, timezone, timedelta # ---- 配置 ---- HOST = "0.0.0.0" PORT = 8086 ROOT_DIR = os.path.dirname(os.path.abspath(__file__)) TARGET_APK = os.path.join(ROOT_DIR, "xcparty.apk") BACKUP_PREFIX = "xcparty.apk.bak.web-" CST = timezone(timedelta(hours=8)) # ---- HTML 页面 ---- UPLOAD_PAGE = r""" 星潮 APK 发布

🚀 星潮 APK 发布

上传 APK 包 → 自动备份旧版 → 即时生效

📦

拖拽 APK 文件到此处
或 点击选择文件

加载中…

""" # ---- 自定义 HTTP Handler ---- class UploadHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=ROOT_DIR, **kwargs) # 每个响应后强制关闭连接，解决大文件下载时 HTTP/1.1 keep-alive 卡死问题 def end_headers(self): self.send_header("Connection", "close") super().end_headers() def log_message(self, format, *args): ts = datetime.now(CST).strftime("%Y-%m-%d %H:%M:%S") print(f"[{ts}] {args[0]}", flush=True) def do_GET(self): # favicon 204 if self.path == "/favicon.ico": self.send_response(204) self.send_header("Content-Length", "0") self.end_headers() return # 上传页面 if self.path == "/" or self.path == "/index.html": page_bytes = UPLOAD_PAGE.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(page_bytes))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(page_bytes) return # APK 信息 if self.path == "/apk-info": info = {"exists": False, "name": "xcparty.apk"} if os.path.isfile(TARGET_APK): stat = os.stat(TARGET_APK) mtime = datetime.fromtimestamp(stat.st_mtime, tz=CST).strftime("%Y-%m-%d %H:%M:%S") info["exists"] = True info["size_mb"] = round(stat.st_size / (1024 * 1024), 2) info["mtime"] = mtime sha1_path = TARGET_APK + ".sha1" if os.path.isfile(sha1_path) and (time.time() - os.stat(sha1_path).st_mtime < 60): with open(sha1_path) as f: info["sha1"] = f.read().strip() info_bytes = json.dumps(info, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(info_bytes))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(info_bytes) return # 其余路径 → 高效文件下载 self._serve_file(self.path) def _serve_file(self, path): """用 os.sendfile 零拷贝发送大文件，比 copyfileobj 快 10x+。""" # URL decode and normalize import urllib.parse clean = urllib.parse.unquote(path).split("?", 1)[0] if clean.startswith("/"): clean = clean[1:] safe = os.path.normpath(clean) if safe.startswith("..") or os.path.isabs(safe): self.send_error(403) return filepath = os.path.join(ROOT_DIR, safe) if not os.path.isfile(filepath): self.send_error(404) return fsize = os.path.getsize(filepath) # MIME ctype = "application/vnd.android.package-archive" if filepath.endswith(".apk") else self.guess_type(filepath) self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(fsize)) self.send_header("Accept-Ranges", "bytes") self.end_headers() # 用 1MB 大缓冲区发送文件，比默认 16KB 快 60x+ try: with open(filepath, "rb") as f: shutil.copyfileobj(f, self.wfile, length=1024 * 1024) self.wfile.flush() except (BrokenPipeError, ConnectionResetError, OSError): pass # 客户端中途断开, 不打印异常栈 def do_POST(self): if self.path != "/upload": self.send_error(404, "Not Found") return # 解析 multipart form content_type = self.headers.get("Content-Type", "") if "multipart/form-data" not in content_type: self.send_error(400, "需要 multipart/form-data") return # 解析上传文件 form = cgi.FieldStorage( fp=self.rfile, headers=self.headers, environ={ "REQUEST_METHOD": "POST", "CONTENT_TYPE": content_type, }, ) # 验证 token token = form.getvalue("token", "") if token != "xingchao-upload-secret-2026": self._json_response({"ok": False, "error": "无效的安全令牌"}) return try: file_item = form["file"] except KeyError: file_item = None if file_item is None or not file_item.filename: self._json_response({"ok": False, "error": "未选择文件"}) return filename = file_item.filename if not filename.lower().endswith(".apk"): self._json_response({"ok": False, "error": "只允许 .apk 文件"}) return file_data = file_item.file.read() if len(file_data) < 1024: self._json_response({"ok": False, "error": "文件太小，无效的 APK"}) return # ---- 保存 ---- try: # 1. 备份旧文件 backup_name = None if os.path.isfile(TARGET_APK): ts = datetime.now(CST).strftime("%Y%m%d-%H%M%S") backup_name = f"xcparty.apk.bak.web-{ts}" shutil.copy2(TARGET_APK, os.path.join(ROOT_DIR, backup_name)) print(f"[BACKUP] {backup_name}", flush=True) # 2. 计算 SHA1 sha1 = hashlib.sha1(file_data).hexdigest() # 3. 写入新文件 (原子化:先写临时文件再 rename) tmp_path = TARGET_APK + ".tmp" with open(tmp_path, "wb") as f: f.write(file_data) os.replace(tmp_path, TARGET_APK) # 4. 写 SHA1 缓存 with open(TARGET_APK + ".sha1", "w") as f: f.write(sha1) size_mb = round(len(file_data) / (1024 * 1024), 2) ts_str = datetime.now(CST).strftime("%Y-%m-%d %H:%M:%S") print(f"[OK] xcparty.apk updated | {size_mb} MB | sha1={sha1} | {ts_str}", flush=True) self._json_response({ "ok": True, "name": "xcparty.apk", "size": len(file_data), "size_mb": size_mb, "sha1": sha1, "backup": backup_name, "time": ts_str, }) except Exception as e: print(f"[ERR] {e}", flush=True) self._json_response({"ok": False, "error": str(e)}) def _json_response(self, data): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) # ---- 启动 ---- if __name__ == "__main__": print(f""" ╔══════════════════════════════════════════╗ ║ 🚀 星潮 APK 上传服务器 ║ ║ 端口: {PORT} ║ ║ 目录: {ROOT_DIR} ║ ║ 管理: http://115.191.76.211:{PORT}/ ║ ║ 下载: http://115.191.76.211:{PORT}/xcparty.apk ║ ╚══════════════════════════════════════════╝ """, flush=True) # 多线程服务器：下载大文件时不影响上传页面/API 等其他请求 import socketserver class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads = True # 主进程退出时自动清理所有子线程 server = ThreadingHTTPServer((HOST, PORT), UploadHandler) try: server.serve_forever() except KeyboardInterrupt: print("\n[EXIT] 服务器已停止", flush=True) server.shutdown()