从零搭建个人音乐库上传管理后台：FastAPI + 原生 HTML 全记录

环境：CentOS / RHEL，Python 3.12（venv），FastAPI + Uvicorn 项目路径：/opt/mydaoliyu_music 音乐目录：/usr/local/music_player/music 服务端口：42126 目标：在已有的音乐流媒体后端上，追加一个浏览器可视化的上传管理后台，替代 scp / mv。

一、背景

已经有一套基于 FastAPI 的音乐后端（道理鱼音乐），功能包括：

SQLite 存储曲库、专辑、歌手、歌单、收藏、播放历史
扫描 MUSIC_ROOT 下所有音频文件，提取标签和封面
提供登录、JWT、封面 ticket、音频 Range 流、音质降级
一个单文件 Web 播放器 /web

但每次新增音乐都很麻烦：

scp song.mp3 root@nas:/usr/local/music_player/music/DJ/
# 再手动触发一次扫描
curl -X POST .../api/library/scan

目标就是：让上传、建目录、浏览、删除、入库扫描，全在浏览器里完成。

最终做出的东西：

后端追加 5 个 /api/admin/* 接口
前端一个单文件 static/admin.html
打开 http://<服务器IP>:42126/admin 即用

二、后端设计

1. 认证复用现有 JWT

后端已经有：

def require_auth(authorization: str = Header(None)) -> dict:
    ...
def require_admin(user: dict = Depends(require_auth)) -> dict:
    if (user.get("role") or "").lower() != "admin":
        raise HTTPException(403, "需要管理员权限")
    return user

上传接口全部挂 require_admin，普通用户即使拿到 token 也调不通。

2. 新增接口一览

方法	路径	作用
GET	`/api/admin/dirs`	递归列出 MUSIC_ROOT 下所有子目录
POST	`/api/admin/upload`	上传文件到指定目录
GET	`/api/admin/files/{dir:path}`	列出某目录下所有文件
POST	`/api/admin/delete`	删除指定文件
POST	`/api/admin/mkdir`	新建目录
GET	`/admin`	返回 admin.html 页面

3. 支持多级目录

MUSIC_ROOT 下有一级目录，一级目录下还有子目录，例如：

music/
├── DJ/
├── 儿歌/
├── 余哥专属/
│   └── 余哥/
└── ...

/api/admin/dirs 用 rglob 递归遍历，返回相对路径：

@app.get("/api/admin/dirs")
def admin_list_dirs(user: dict = Depends(require_admin)):
    if not MUSIC_ROOT.is_dir():
        return {"items": []}

    items = []
    root = MUSIC_ROOT.resolve()

    for p in sorted(root.rglob("*")):
        if not p.is_dir():
            continue
        if p.name.startswith("."):
            continue
        try:
            rel = p.relative_to(root)
        except ValueError:
            continue
        if any(part in SCAN_IGNORE_DIRS for part in rel.parts):
            continue
        items.append(str(rel).replace("\\", "/"))

    return {"items": items}

输出示例：

{
  "items": [
    "DJ",
    "儿歌",
    "余哥专属",
    "余哥专属/余哥",
    ...
  ]
}

4. 路径安全

_SAFE_SEGMENT_RE = re.compile(r'^[^/\\:*?"<>|]{1,64}$')

def _safe_dir_name(name: str):
    """支持多级目录，例如 余哥专属/余哥"""
    if not name:
        return None
    name = name.strip().replace("\\", "/").strip("/")
    if not name:
        return None
    for seg in name.split("/"):
        if seg in ("", ".", ".."):
            return None
        if not _SAFE_SEGMENT_RE.match(seg):
            return None
    return name


def _safe_file_name(name: str):
    if not name:
        return None
    name = Path(name).name
    if not name or name in (".", ".."):
        return None
    if "\x00" in name:
        return None
    return name

每段都禁 .. 和非法字符，防 ../../etc/passwd
文件名只取 Path(name).name，去掉任何目录部分
删除前再校验目标路径必须在 MUSIC_ROOT 下

5. 上传实现

UPLOAD_ALLOWED_EXTS = {
    ".mp3", ".flac", ".m4a", ".aac", ".wav",
    ".ogg", ".opus", ".wma", ".ape", ".alac",
    ".dsf", ".dff", ".lrc",
}
UPLOAD_MAX_BYTES = 500 * 1024 * 1024


@app.post("/api/admin/upload")
async def admin_upload(
    dir: str = Form(...),
    overwrite: str = Form("0"),
    files: list[UploadFile] = File(...),
    user: dict = Depends(require_admin),
):
    safe = _safe_dir_name(dir)
    if not safe:
        raise HTTPException(400, "目录名不合法")

    target = MUSIC_ROOT / safe
    target.mkdir(parents=True, exist_ok=True)
    overwrite_flag = (overwrite == "1")

    results = []
    for f in files:
        raw_name = _safe_file_name(f.filename or "")
        if not raw_name:
            results.append({"name": f.filename, "ok": False, "msg": "文件名不合法"})
            continue

        ext = Path(raw_name).suffix.lower()
        if ext not in UPLOAD_ALLOWED_EXTS:
            results.append({"name": raw_name, "ok": False,
                            "msg": f"不支持的类型 {ext}"})
            continue

        dest = target / raw_name
        if dest.exists() and not overwrite_flag:
            stem, suf = dest.stem, dest.suffix
            raw_name = f"{stem}_{int(time.time())}{suf}"
            dest = target / raw_name

        tmp = dest.with_suffix(dest.suffix + ".part")
        total = 0
        try:
            with open(tmp, "wb") as fp:
                while True:
                    chunk = await f.read(1024 * 1024)
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > UPLOAD_MAX_BYTES:
                        raise ValueError("文件超过 500MB")
                    fp.write(chunk)
            tmp.replace(dest)
            results.append({"name": raw_name, "ok": True, "size": total})
        except Exception as e:
            try:
                tmp.unlink(missing_ok=True)
            except Exception:
                pass
            results.append({"name": raw_name, "ok": False, "msg": str(e)})

    logger.info("管理员 %s 上传 %d 个文件到 %s",
                user.get("username"), len(results), safe)
    return {"results": results}

关键点：

分块写入，每次 1MB，避免大文件一次性读进内存
原子写：先写 .part，写完再 replace
同名不覆盖：追加时间戳
单文件上限 500MB

6. 文件列表 & 删除

@app.get("/api/admin/files/{dir_name:path}")
def admin_list_files(dir_name: str, user: dict = Depends(require_admin)):
    safe = _safe_dir_name(dir_name)
    if not safe:
        raise HTTPException(400, "目录名不合法")
    target = MUSIC_ROOT / safe
    if not target.is_dir():
        raise HTTPException(404, "目录不存在")

    items = []
    for p in sorted(target.iterdir()):
        if not p.is_file():
            continue
        try:
            st = p.stat()
        except OSError:
            continue
        items.append({
            "name": p.name,
            "size": st.st_size,
            "mtime": int(st.st_mtime),
        })
    return {"items": items}

注意 {dir_name:path}： FastAPI 默认 {name} 只匹配一段路径，遇到 余哥专属/余哥 就 404。加 :path 才能匹配多级路径。

7. 新建目录

@app.post("/api/admin/mkdir")
async def admin_mkdir(request: Request,
                      user: dict = Depends(require_admin)):
    try:
        body = await request.json()
    except Exception:
        raise HTTPException(400, "请求体不是合法 JSON")
    name = _safe_dir_name(body.get("name", ""))
    if not name:
        raise HTTPException(400, "目录名不合法")
    target = MUSIC_ROOT / name
    if target.exists():
        return {"ok": True, "existed": True}
    try:
        target.mkdir(parents=True, exist_ok=False)
    except Exception as e:
        raise HTTPException(500, f"创建失败：{e}")
    return {"ok": True, "existed": False}

三、前端设计

单文件 static/admin.html，无第三方依赖，深色 UI。

主要功能

登录卡片：只允许 role=admin 的账号
目标目录下拉框：递归列出所有子目录，带缩进层级
新建目录输入框：填名字 → 点“创建” → 后端 mkdir → 下拉框刷新
拖拽上传区：支持多选、拖拽、.mp3/.flac/.m4a/.lrc 等
上传队列：每个文件单独状态（等待 / 上传中 / 成功 / 失败）
上传进度条：整批文件平均进度
同名处理：勾选“覆盖”或自动加时间戳
右侧目录浏览：选择目录 → 列出文件 → 点 × 删除
重新扫描曲库：右上角按钮，调 /api/library/scan
退出登录：清 localStorage token

关键实现片段

async function doLogin() {
  const r = await fetch("/api/auth/login", {
    method: "POST",
    headers: {"Content-Type":"application/json"},
    body: JSON.stringify({username, password})
  });
  const data = await r.json();
  if ((data.user && data.user.role) !== "admin") {
    msg.textContent = "该账号不是管理员";
    return;
  }
  token = data.token;
  localStorage.setItem("admin_token", token);
  enterApp();
}

上传（XMLHttpRequest 带进度）：

function uploadOne(item, dir, overwrite) {
  return new Promise(resolve => {
    const fd = new FormData();
    fd.append("dir", dir);
    fd.append("overwrite", overwrite ? "1" : "0");
    fd.append("files", item.file, item.file.name);

    const xhr = new XMLHttpRequest();
    xhr.open("POST", "/api/admin/upload");
    xhr.setRequestHeader("Authorization", "Bearer " + token);
    xhr.upload.onprogress = e => {
      if (e.lengthComputable) {
        item._pct = e.loaded / e.total;
        renderProgress();
      }
    };
    xhr.onload = () => {
      const r = JSON.parse(xhr.responseText);
      const first = (r.results && r.results[0]) || {};
      if (first.ok) item.state = "ok";
      else { item.state = "err"; item.msg = first.msg; }
      resolve();
    };
    item.state = "run";
    renderPending();
    xhr.send(fd);
  });
}

四、部署

1. 追加代码

把上面的后端接口贴到 main.py 末尾（if __name__ == "__main__": 之前）。

顶部 import 需要补上：

import re
from pathlib import Path
from fastapi import Form

from config import (...) 里要加 SCAN_IGNORE_DIRS。

2. 放前端文件

mkdir -p /opt/mydaoliyu_music/static
vi /opt/mydaoliyu_music/static/admin.html

3. 重启

systemctl restart mydaoliyu-music
tail -f /var/log/mydaoliyu-music.log

4. 访问

http://<服务器IP>:42126/admin

默认 admin / admin。

五、踩过的坑

坑 1：登录后页面没反应

现象：后端日志 /api/auth/login 200、/api/admin/users/me 200，但前端还停在登录页。

原因：admin.html 里 CSS 覆盖了 hidden 属性。

#login-view { display: flex; }     /* ID 选择器优先级高 */

<div id="login-view" hidden>        <!-- 被覆盖 -->

修复：在 <style> 最前面加：

[hidden] { display: none !important; }

player.html 里本来就有这句，新加的 admin.html 漏了。

坑 2：`Unchecked runtime.lastError`

现象：Console 报：

Unchecked runtime.lastError: Could not establish connection. Receiving end does not exist.

原因：浏览器扩展（广告拦截、密码管理器、翻译插件）跟页面无关。

修复：忽略；用无痕窗口排除扩展影响。

坑 3：目录加载失败，服务器内部错误

原因：admin_list_dirs 里用了 SCAN_IGNORE_DIRS，但 main.py 顶部没导入。

修复：

from config import (
    ...
    SCAN_IGNORE_DIRS,
    ...
)

坑 4：子目录 404

现象：下拉框里选 余哥专属/余哥，右侧加载失败 Not Found。

原因：FastAPI 默认 {dir_name} 只匹配单段路径。

修复：

@app.get("/api/admin/files/{dir_name:path}")   # ← 加 :path

六、密码管理

默认密码从哪来

config.py：

python

DEFAULT_ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "admin")
DEFAULT_ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin")

.env 里

ADMIN_USERNAME=admin
ADMIN_PASSWORD=admin

database.py 的 ensure_default_user：

def ensure_default_user(logger=None):
    with get_db() as conn:
        count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
        if count > 0:
            return                       # ← 已有用户就不再创建
        ...
        conn.execute(
            "INSERT INTO users (...) VALUES (...)",
            (user_id, DEFAULT_ADMIN_USERNAME, ...,
             hash_password(DEFAULT_ADMIN_PASSWORD),   # ← bcrypt
             ...)
        )

只在 users 表为空时创建
密码用 bcrypt（rounds=10）哈希后存 password_hash
一旦创建，重启不会覆盖

修改密码

方式一：命令行脚本

cd /opt/mydaoliyu_music
source venvmusic3.12/bin/activate

set -a
source .env
set +a

python - <<'PY'
from auth import hash_password
from database import get_db
new_hash = hash_password("XXxx....")
with get_db() as conn:
    conn.execute("UPDATE users SET password_hash=? WHERE username='admin'",
                 (new_hash,))
print("已更新")
PY

方式二：HTTP 接口

curl -X POST http://127.0.0.1:42126/api/auth/change-password \
  -H "Authorization: Bearer <你的token>" \
  -H "Content-Type: application/json" \
  -d '{"oldPassword":"admin","newPassword":"XXxx3663"}'

三个必须显式设置的变量

.env 里已经配好：

JWT_SET=Ee2vnRrECYdBHyeksobpxFCL2Bi0-sudRr5njB0hDpAM7rMr6d
TICKET_SECRET=MdrG8_8_flivVizua7-BSxdWLDN0eZlgyiTsiM
DATA_DIR=/usr/local/music_player/data
MUSIC_ROOT=/usr/local/music_player/music

否则 JWT_SECRET 每次重启随机生成，所有登录态失效。

七、日常运维速查

# 重启服务
systemctl restart mydaoliyu-music

# 看日志
tail -f /var/log/mydaoliyu-music.log

# 查看当前用户
sqlite3 /usr/local/music_player/data/app.db \
  "SELECT username, role FROM users;"

# 检查密码哈希前缀
sqlite3 /usr/local/music_player/data/app.db \
  "SELECT username, substr(password_hash,1,20) FROM users WHERE username='admin';"

# 手动触发全量扫描
curl -X POST http://127.0.0.1:42126/api/library/scan \
  -H "Authorization: Bearer <token>"

# 查看服务配置
systemctl cat mydaoliyu-music

八、最终成果

打开 http://<服务器IP>:42126/admin：

左侧：选择目标目录 → 拖拽上传 → 进度条 → 点“开始上传”
右侧：浏览目录文件 → 删除
顶部：重新扫描曲库 → 新歌自动出现在 /web 播放器

上传 → 入库 → 播放，全程浏览器搞定：

[admin.html 上传]  →  MUSIC_ROOT/DJ/xxx.mp3
                   →  POST /api/library/scan
                   →  SQLite 入库
                   →  /web 播放器可见

九、后续可扩展方向

批量重命名
编辑音频标签（标题、歌手、专辑）
上传时自动匹配 .lrc 与音频文件名
拖拽排序、批量移动
上传历史 / 操作日志
多用户权限（普通用户只能上传，不能删除）
文件夹歌单递归：目前 folder_playlists.py 只同步一级目录，可以扩展到递归子目录
断点续传：前端切片上传，后端合并
封面替换 / 音频波形可视化

一句话总结：在已有 FastAPI 后端上追加 5 个 admin 接口 + 1 个单文件前端，就能得到一个浏览器可视化的音乐上传管理后台。核心难点只有三个：路径参数用 :path、[hidden] 用 !important、路径安全逐段校验。其余都是常规 CRUD。