439 lines
15 KiB
Python
439 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
server.py —— BiliDownloader 本地服务（零第三方依赖）
|
||
|
||
只监听 127.0.0.1。职责：
|
||
* 托管前端构建产物 (ui/dist)
|
||
* 提供 JSON API（搜索 / 详情 / 清晰度 / 开始下载 / 取消 / 配置）
|
||
* 用 Server-Sent Events 把下载进度实时推给浏览器
|
||
|
||
用法:
|
||
python server.py [--port 8799] [--no-open]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
import queue
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import traceback
|
||
import urllib.parse
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
import engine
|
||
|
||
BASE = os.path.dirname(os.path.abspath(__file__))
|
||
DIST = os.path.join(BASE, "ui", "dist")
|
||
CONFIG_PATH = os.path.join(BASE, "config.json")
|
||
|
||
DEFAULT_CONFIG = {
|
||
"sessdata": "",
|
||
"outputDir": os.path.join(os.path.expanduser("~"), "Videos", "BiliDownloader"),
|
||
"threads": 16,
|
||
"preferAvc": True,
|
||
"keepParts": False,
|
||
}
|
||
|
||
_config_lock = threading.Lock()
|
||
_job_lock = threading.Lock()
|
||
_job: dict | None = None # 当前/最近一次下载任务
|
||
_cancel: threading.Event | None = None
|
||
_streams: dict[str, dict] = {} # (bvid,cid) -> playurl 结果
|
||
_subscribers: list[queue.Queue] = []
|
||
_play_cache: dict[str, dict] = {}
|
||
|
||
|
||
# ------------------------------------------------------------------ 配置
|
||
|
||
def load_config() -> dict:
|
||
cfg = dict(DEFAULT_CONFIG)
|
||
try:
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
cfg.update(json.load(f))
|
||
except Exception:
|
||
pass
|
||
cfg["threads"] = max(1, min(64, int(cfg.get("threads") or 16)))
|
||
return cfg
|
||
|
||
|
||
def save_config(cfg: dict) -> dict:
|
||
merged = load_config()
|
||
merged.update({k: v for k, v in cfg.items() if k in DEFAULT_CONFIG})
|
||
merged["threads"] = max(1, min(64, int(merged.get("threads") or 16)))
|
||
with _config_lock:
|
||
tmp = CONFIG_PATH + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(merged, f, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, CONFIG_PATH)
|
||
return merged
|
||
|
||
|
||
def cookie_of(cfg: dict) -> str | None:
|
||
s = (cfg.get("sessdata") or "").strip()
|
||
return f"SESSDATA={s}" if s else None
|
||
|
||
|
||
# ------------------------------------------------------------------ 事件广播
|
||
|
||
def broadcast(event: str, data: dict) -> None:
|
||
payload = f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
|
||
dead = []
|
||
for q in list(_subscribers):
|
||
try:
|
||
q.put_nowait(payload)
|
||
except Exception:
|
||
dead.append(q)
|
||
for q in dead:
|
||
try:
|
||
_subscribers.remove(q)
|
||
except ValueError:
|
||
pass
|
||
|
||
|
||
# ------------------------------------------------------------------ 下载任务
|
||
|
||
def start_download(bvid: str, cid: int, quality: int, title: str) -> dict:
|
||
global _job, _cancel
|
||
|
||
with _job_lock:
|
||
if _job and _job.get("state") == "running":
|
||
raise RuntimeError("已有下载任务在进行中")
|
||
|
||
cfg = load_config()
|
||
key = f"{bvid}:{cid}"
|
||
play = _streams.get(key)
|
||
if not play:
|
||
raise RuntimeError("播放地址不存在或已过期，请重新选择视频")
|
||
|
||
video, audio = engine.pick_streams(play, quality, cfg.get("preferAvc", True))
|
||
if video is None:
|
||
raise RuntimeError("当前清晰度没有可用的视频流")
|
||
|
||
out_dir = cfg["outputDir"]
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
base = engine.safe_name(title)
|
||
# 文件名必须带上清晰度：否则同一视频下不同清晰度会互相覆盖
|
||
# （实测踩过：480P 的验收下载把之前下好的 4K 文件直接盖掉了）
|
||
q_label = next((q["label"] for q in play.get("qualities", [])
|
||
if q["quality"] == quality), str(quality))
|
||
q_tag = engine.safe_name(q_label).replace(" ", "")
|
||
stamp = time.strftime("%Y%m%d%H%M%S")
|
||
final_path = os.path.join(out_dir, f"{base} [{q_tag}].mp4")
|
||
tmp_v = os.path.join(out_dir, f".{base}.{cid}.{stamp}.video.m4s")
|
||
tmp_a = os.path.join(out_dir, f".{base}.{cid}.{stamp}.audio.m4s")
|
||
|
||
_cancel = threading.Event()
|
||
_job = {
|
||
"state": "running", "stage": "下载视频流", "bvid": bvid, "cid": cid,
|
||
"title": title, "quality": quality, "outputPath": final_path,
|
||
"startedAt": time.time(), "video": None, "audio": None,
|
||
"progress": {"video": None, "audio": None}, "error": None,
|
||
"resumed": 0, "threads": cfg["threads"],
|
||
}
|
||
|
||
broadcast("job", dict(_job))
|
||
threading.Thread(target=_run_job, args=(video, audio, tmp_v, tmp_a, final_path,
|
||
cfg, bvid, cid), daemon=True).start()
|
||
return dict(_job)
|
||
|
||
|
||
def _run_job(video, audio, tmp_v, tmp_a, final_path, cfg, bvid, cid) -> None:
|
||
global _job
|
||
cookie = cookie_of(cfg)
|
||
referer = f"https://www.bilibili.com/video/{bvid}"
|
||
threads = cfg["threads"]
|
||
|
||
def make_on_progress(which):
|
||
def on_progress(info):
|
||
_job["progress"][which] = info
|
||
broadcast("progress", {"which": which, **info, "stage": _job["stage"]})
|
||
return on_progress
|
||
|
||
try:
|
||
_job["stage"] = "下载视频流"
|
||
broadcast("stage", {"stage": _job["stage"]})
|
||
vres = engine.download_fast(video["baseUrl"], tmp_v, referer, cookie,
|
||
threads, make_on_progress("video"), _cancel, "视频")
|
||
_job["resumed"] = vres.get("resumed", 0)
|
||
|
||
if audio:
|
||
_job["stage"] = "下载音频流"
|
||
broadcast("stage", {"stage": _job["stage"]})
|
||
engine.download_fast(audio["baseUrl"], tmp_a, referer, cookie,
|
||
max(1, min(8, threads)), make_on_progress("audio"),
|
||
_cancel, "音频")
|
||
|
||
_job["stage"] = "ffmpeg 封装"
|
||
broadcast("stage", {"stage": _job["stage"]})
|
||
engine.merge(tmp_v, tmp_a if audio else None, final_path)
|
||
|
||
if not cfg.get("keepParts"):
|
||
for p in (tmp_v, tmp_a):
|
||
try:
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
except OSError:
|
||
pass
|
||
|
||
_job["state"] = "done"
|
||
_job["stage"] = "完成"
|
||
_job["size"] = os.path.getsize(final_path)
|
||
_job["elapsed"] = time.time() - _job["startedAt"]
|
||
broadcast("job", dict(_job))
|
||
|
||
except InterruptedError:
|
||
for p in (final_path,):
|
||
try:
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
except OSError:
|
||
pass
|
||
if not cfg.get("keepParts"):
|
||
for p in (tmp_v, tmp_a):
|
||
try:
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
except OSError:
|
||
pass
|
||
_job["state"] = "cancelled"
|
||
_job["stage"] = "已取消，未完成的文件已清理"
|
||
broadcast("job", dict(_job))
|
||
except Exception as exc:
|
||
for p in (final_path,):
|
||
try:
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
except OSError:
|
||
pass
|
||
_job["state"] = "error"
|
||
_job["error"] = str(exc)
|
||
broadcast("job", dict(_job))
|
||
traceback.print_exc()
|
||
|
||
|
||
# ------------------------------------------------------------------ HTTP
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "BiliDownloader"
|
||
|
||
def log_message(self, fmt, *args):
|
||
pass # 静音访问日志
|
||
|
||
# ---------- helpers
|
||
def _json(self, obj, status=200):
|
||
body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _error(self, message, status=400):
|
||
self._json({"ok": False, "error": str(message)}, status)
|
||
|
||
def _query(self) -> dict:
|
||
parsed = urllib.parse.urlparse(self.path)
|
||
return {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()}, parsed.path
|
||
|
||
def _body(self) -> dict:
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
if not length:
|
||
return {}
|
||
return json.loads(self.rfile.read(length).decode("utf-8"))
|
||
|
||
# ---------- routes
|
||
def do_GET(self):
|
||
params, path = self._query()
|
||
try:
|
||
if path == "/api/config":
|
||
cfg = load_config()
|
||
cfg["configPath"] = CONFIG_PATH
|
||
cfg["ffmpeg"] = engine.find_ffmpeg()
|
||
return self._json(cfg)
|
||
|
||
if path == "/api/login":
|
||
cfg = load_config()
|
||
return self._json(engine.login_status(cookie_of(cfg)))
|
||
|
||
if path == "/api/search":
|
||
cfg = load_config()
|
||
return self._json(engine.search(
|
||
params.get("q", ""), int(params.get("page", 1)), 20, cookie_of(cfg)))
|
||
|
||
if path == "/api/video":
|
||
cfg = load_config()
|
||
return self._json(engine.video_detail(params["bvid"], cookie_of(cfg)))
|
||
|
||
if path == "/api/playurl":
|
||
cfg = load_config()
|
||
bvid, cid = params["bvid"], int(params["cid"])
|
||
play = engine.play_url(bvid, cid, cookie_of(cfg))
|
||
_streams[f"{bvid}:{cid}"] = play
|
||
return self._json(play)
|
||
|
||
if path == "/api/job":
|
||
return self._json(_job or {})
|
||
|
||
if path == "/api/events":
|
||
return self._sse()
|
||
|
||
if path.startswith("/api/"):
|
||
return self._error("未知接口", 404)
|
||
|
||
return self._static(path)
|
||
except engine.BiliError as exc:
|
||
return self._error(f"{exc.message}（code {exc.code}）", 502)
|
||
except Exception as exc:
|
||
traceback.print_exc()
|
||
return self._error(exc, 500)
|
||
|
||
def do_POST(self):
|
||
_, path = self._query()
|
||
try:
|
||
if path == "/api/config":
|
||
return self._json(save_config(self._body()))
|
||
|
||
if path == "/api/download":
|
||
data = self._body()
|
||
job = start_download(data["bvid"], int(data["cid"]),
|
||
int(data["quality"]), data.get("title") or "video")
|
||
return self._json(job)
|
||
|
||
if path == "/api/cancel":
|
||
if _cancel:
|
||
_cancel.set()
|
||
return self._json({"ok": True})
|
||
|
||
if path == "/api/open-folder":
|
||
target = (self._body() or {}).get("path") or load_config()["outputDir"]
|
||
if os.path.exists(target):
|
||
subprocess.Popen(["explorer", "/select,", os.path.normpath(target)])
|
||
else:
|
||
os.makedirs(target, exist_ok=True)
|
||
subprocess.Popen(["explorer", os.path.normpath(target)])
|
||
return self._json({"ok": True})
|
||
|
||
return self._error("未知接口", 404)
|
||
except Exception as exc:
|
||
traceback.print_exc()
|
||
return self._error(exc, 500)
|
||
|
||
# ---------- SSE
|
||
def _sse(self):
|
||
q: queue.Queue = queue.Queue()
|
||
_subscribers.append(q)
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Connection", "keep-alive")
|
||
self.end_headers()
|
||
try:
|
||
hello = f"event: job\ndata: {json.dumps(_job or {}, ensure_ascii=False)}\n\n"
|
||
self.wfile.write(hello.encode("utf-8"))
|
||
self.wfile.flush()
|
||
while True:
|
||
try:
|
||
payload = q.get(timeout=15)
|
||
self.wfile.write(payload.encode("utf-8"))
|
||
except queue.Empty:
|
||
self.wfile.write(b": ping\n\n") # 心跳，防代理断连
|
||
self.wfile.flush()
|
||
except (BrokenPipeError, ConnectionResetError, OSError):
|
||
pass
|
||
finally:
|
||
try:
|
||
_subscribers.remove(q)
|
||
except ValueError:
|
||
pass
|
||
|
||
# ---------- 静态文件
|
||
def _static(self, path: str):
|
||
if not os.path.isdir(DIST):
|
||
body = ("<!doctype html><meta charset=utf-8>"
|
||
"<h1>前端还没构建</h1>"
|
||
"<p>请在 ui 目录执行 <code>npm run build</code>，"
|
||
"或直接用 <code>npm run dev</code> 起开发服务器。</p>").encode("utf-8")
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
return
|
||
|
||
rel = urllib.parse.unquote(path.lstrip("/")) or "index.html"
|
||
target = os.path.normpath(os.path.join(DIST, rel))
|
||
# 防目录穿越
|
||
if not target.startswith(os.path.normpath(DIST)):
|
||
return self._error("非法路径", 403)
|
||
if not os.path.isfile(target):
|
||
target = os.path.join(DIST, "index.html") # SPA 回退
|
||
if not os.path.isfile(target):
|
||
return self._error("Not found", 404)
|
||
|
||
ctype = mimetypes.guess_type(target)[0] or "application/octet-stream"
|
||
if ctype.startswith("text/") or ctype in ("application/javascript", "application/json"):
|
||
ctype += "; charset=utf-8"
|
||
with open(target, "rb") as f:
|
||
body = f.read()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store" if target.endswith(".html") else "public, max-age=3600")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--port", type=int, default=8799)
|
||
ap.add_argument("--no-open", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
cfg = load_config()
|
||
if not os.path.exists(CONFIG_PATH):
|
||
save_config(cfg)
|
||
|
||
httpd = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
|
||
url = f"http://127.0.0.1:{args.port}/"
|
||
print(f"BiliDownloader 服务已启动: {url}", flush=True)
|
||
print(f"配置: {CONFIG_PATH}", flush=True)
|
||
print(f"ffmpeg: {engine.find_ffmpeg() or '未找到（合并会失败）'}", flush=True)
|
||
|
||
if not args.no_open:
|
||
threading.Thread(target=_open_browser, args=(url,), daemon=True).start()
|
||
|
||
try:
|
||
httpd.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
return 0
|
||
|
||
|
||
def _open_browser(url: str) -> None:
|
||
time.sleep(0.8)
|
||
try:
|
||
edge = None
|
||
for p in (r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
|
||
r"C:\Program Files\Microsoft\Edge\Application\msedge.exe"):
|
||
if os.path.exists(p):
|
||
edge = p
|
||
break
|
||
if edge:
|
||
# 应用模式窗口：无地址栏、无标签页，看起来就是一个独立应用
|
||
subprocess.Popen([edge, f"--app={url}", "--window-size=1280,860"])
|
||
else:
|
||
subprocess.Popen(["cmd", "/c", "start", "", url], shell=False)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|