xiaoxia-saas/scripts/upload_preset_bgm.py

#!/usr/bin/env python3
"""一键上传预设 BGM 到 OSS 并回填 preset_bgm.py audio_url。

用法（在服务器或本地有 OSS 凭证的机器上执行）：
  1. 把 mp3 文件放到 ./bgm_assets/ (created by you) 目录下，文件名按 {preset_id}.mp3 命名：
       bgm_upbeat_001.mp3  阳光清晨
       bgm_upbeat_002.mp3  活力节拍
       bgm_upbeat_003.mp3  夏日漫步
       bgm_relax_001.mp3   静谧时光
       bgm_relax_002.mp3   雨后森林
       bgm_relax_003.mp3   月光奏鸣曲
       bgm_tech_001.mp3    未来科技
       bgm_tech_002.mp3    数据脉冲
       bgm_commerce_001.mp3 心动时刻
       bgm_commerce_002.mp3 品质生活
  2. 确保环境变量已设置：
       OSS_ENDPOINT, OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_BUCKET_NAME
  3. 运行：python scripts/upload_preset_bgm.py
  4. 脚本会上传到 OSS 路径 preset/bgm/<id>.mp3，并自动改写
     packages/domain/preset_bgm.py 填入 public URL。
  5. git commit & push 即可。

支持可选参数：
  --bgm-dir DIR      本地 mp3 目录（默认 ./bgm_assets）
  --oss-prefix PFX   OSS key 前缀（默认 preset/bgm/）
  --dry-run          只打印要做的操作，不上传不改写
  --public           上传后设置公共读 ACL（默认开启，safe_download 走公网 URL）
"""

from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path

REQUIRED_ENV = ("OSS_ENDPOINT", "OSS_ACCESS_KEY_ID", "OSS_ACCESS_KEY_SECRET", "OSS_BUCKET_NAME")
PRESET_BGM_PY = Path(__file__).resolve().parents[1] / "packages" / "domain" / "preset_bgm.py"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--bgm-dir", default="./bgm_assets")
    ap.add_argument("--oss-prefix", default="preset/bgm/")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--public", action="store_true", default=True)
    args = ap.parse_args()

    bgm_dir = Path(args.bgm_dir)
    if not bgm_dir.exists():
        print(f"[ERR] 目录不存在: {bgm_dir}", file=sys.stderr)
        return 2

    # 加载 preset_bgm.py 中所有 ID（简单 AST 抽取）
    sys.path.insert(0, str(PRESET_BGM_PY.parents[2]))
    from packages.domain.preset_bgm import PRESET_BGM_LIBRARY  # type: ignore

    id_to_preset = {p.id: p for p in PRESET_BGM_LIBRARY}

    # 枚举本地文件
    local_files: dict[str, Path] = {}
    for ext in (".mp3", ".m4a", ".aac", ".wav", ".ogg"):
        for f in bgm_dir.glob(f"*{ext}"):
            pid = f.stem
            local_files[pid] = f

    missing_local = [pid for pid in id_to_preset if pid not in local_files]
    unknown_local = [pid for pid in local_files if pid not in id_to_preset]

    print(f"[INFO] 预设 BGM 总数: {len(PRESET_BGM_LIBRARY)}")
    print(f"[INFO] 本地找到对应文件: {len(local_files) - len(unknown_local)}")
    if missing_local:
        print(f"[WARN] 缺少本地音频文件的预设（{len(missing_local)}）:")
        for pid in missing_local:
            print(f"       {pid}  {id_to_preset[pid].name}")
    if unknown_local:
        print(f"[WARN] 本地文件未匹配任何 preset_id（{len(unknown_local)}）: {unknown_local}")

    if args.dry_run:
        for pid, f in local_files.items():
            if pid in id_to_preset:
                url = f"https://{os.environ.get('OSS_BUCKET_NAME', '<bucket>')}.{os.environ.get('OSS_ENDPOINT', '<ep>')}/{args.oss_prefix}{f.name}"
                print(f"[DRY] would upload {f} → {args.oss_prefix}{f.name} → {url}")
        return 0

    # 凭证检查
    for k in REQUIRED_ENV:
        if not os.environ.get(k):
            print(f"[ERR] 缺少环境变量 {k}", file=sys.stderr)
            return 3

    try:
        import oss2  # type: ignore
    except ImportError:
        print("[ERR] 需要 oss2: pip install oss2", file=sys.stderr)
        return 4

    endpoint = os.environ["OSS_ENDPOINT"]
    bucket_name = os.environ["OSS_BUCKET_NAME"]
    auth = oss2.Auth(os.environ["OSS_ACCESS_KEY_ID"], os.environ["OSS_ACCESS_KEY_SECRET"])
    bucket = oss2.Bucket(auth, f"https://{endpoint}", bucket_name)
    public_base = f"https://{bucket_name}.{endpoint}"

    pid_to_url: dict[str, str] = {}
    for pid, f in local_files.items():
        if pid not in id_to_preset:
            continue
        key = f"{args.oss_prefix.rstrip('/')}/{f.name}"
        print(f"[UPLOAD] {f} → oss://{bucket_name}/{key}")
        headers = {"Content-Type": "audio/mpeg" if f.suffix == ".mp3" else "audio/mp4"}
        if args.public:
            bucket.put_object_from_file(key, str(f), headers=headers)
            bucket.put_object_acl(key, oss2.OBJECT_ACL_PUBLIC_READ)
        else:
            bucket.put_object_from_file(key, str(f), headers=headers)
        url = f"{public_base}/{key}"
        pid_to_url[pid] = url
        print(f"         → {url}")

    if not pid_to_url:
        print("[WARN] 没有文件被上传，不修改 preset_bgm.py")
        return 0

    # 回填 preset_bgm.py（按 id 精确替换 audio_url="" 为 audio_url="<url>"）
    text = PRESET_BGM_PY.read_text(encoding="utf-8")
    orig = text
    for pid, url in pid_to_url.items():
        # 匹配 PresetBGM( ... id="pid", ... audio_url="", ... )
        # 简单替换：找到 id="pid" 行开始的 PresetBGM 构造块，把块内的 audio_url="" 替换
        import re

        block_pat = re.compile(
            r'(PresetBGM\([^)]*?id="' + re.escape(pid) + r'"[^)]*?audio_url=)"[^"]*"',
            re.DOTALL,
        )
        new_text, n = block_pat.subn(rf'\1"{url}"', text, count=1)
        if n == 0:
            # 兜底：可能 audio_url 后面直接是 )，没赋值？不会，dataclass 有默认值
            print(f"[WARN] 未在 PresetBGM 块中找到 id={pid} 的 audio_url 字段，跳过回填")
            continue
        text = new_text
    if text != orig:
        PRESET_BGM_PY.write_text(text, encoding="utf-8")
        print(f"[OK] 已回填 {len(pid_to_url)} 条 audio_url 到 {PRESET_BGM_PY}")
        print(
            "[NEXT] git diff packages/domain/preset_bgm.py && git add -A && git commit -m 'feat(domain): fill preset BGM audio_urls' && git push"
        )
    else:
        print("[WARN] preset_bgm.py 未发生变化（可能已填过）")
    return 0


if __name__ == "__main__":
    sys.exit(main())