meijiaka-zy/python-api/scripts/import_bgm.py

#!/usr/bin/env python3
"""
导入 Mixkit BGM 音频文件到数据库并上传七牛云
扫描 mixkit_bgm/ 目录下的分类文件夹，读取音频元数据并插入 mjk_bgm_musics 表
同时上传音频到七牛云视频/音频 bucket，保存 URL。
"""

import asyncio
import os
import sys
from pathlib import Path

# 将项目根目录加入 Python 路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

# 加载环境变量
from dotenv import load_dotenv

load_dotenv()

from mutagen.mp3 import MP3
from sqlalchemy import select
from app.db.session import AsyncSessionLocal
from app.models.bgm_music import BgmMusic
from app.services.qiniu_service import get_qiniu_service

BGM_BASE_DIR = Path("/Users/0fun/work/meijiaka-zy/mixkit_bgm")


def parse_title(filename: str) -> str:
    """从文件名解析标题，如 '105_See_Line_Funk.mp3' -> 'See Line Funk'"""
    # 去掉扩展名
    name = filename.rsplit(".", 1)[0]
    # 去掉开头的编号前缀（如 105_）
    parts = name.split("_", 1)
    if len(parts) > 1 and parts[0].isdigit():
        title_part = parts[1]
    else:
        title_part = name
    # 将下划线替换为空格
    return title_part.replace("_", " ")


def get_duration(filepath: Path) -> float | None:
    """获取音频文件时长（秒）"""
    try:
        audio = MP3(str(filepath))
        return audio.info.length
    except Exception as e:
        print(f"  无法读取时长: {filepath.name} ({e})")
        return None


async def import_bgm():
    """扫描目录并导入数据库，同时上传七牛云"""
    if not BGM_BASE_DIR.exists():
        print(f"错误: BGM 目录不存在: {BGM_BASE_DIR}")
        return

    categories = [d.name for d in BGM_BASE_DIR.iterdir() if d.is_dir()]
    print(f"发现分类: {categories}")

    # 初始化七牛云服务
    try:
        qiniu = get_qiniu_service()
        print("七牛云服务初始化成功")
    except ValueError as e:
        print(f"七牛云配置错误: {e}")
        return

    async with AsyncSessionLocal() as session:
        # 清空现有数据（可选）
        result = await session.execute(select(BgmMusic))
        existing = result.scalars().all()
        if existing:
            print(f"数据库中已有 {len(existing)} 条 BGM 记录，将删除后重新导入")
            for row in existing:
                await session.delete(row)
            await session.commit()

        total = 0
        upload_ok = 0
        upload_fail = 0
        for category in sorted(categories):
            cat_dir = BGM_BASE_DIR / category
            files = sorted([f for f in cat_dir.iterdir() if f.suffix.lower() in (".mp3", ".wav", ".m4a")])
            print(f"\n分类 [{category}]: {len(files)} 首")

            for idx, filepath in enumerate(files):
                title = parse_title(filepath.name)
                relative_path = f"{category}/{filepath.name}"
                duration = get_duration(filepath)

                # 上传七牛云
                qiniu_key = f"meijiaka-zy/bgm/{relative_path}"
                url = None
                try:
                    upload_result = qiniu.upload_file(
                        local_path=str(filepath),
                        key=qiniu_key,
                        file_type="audio",
                        check_duplicate=True,
                    )
                    url = upload_result.get("url")
                    if upload_result.get("isDuplicate"):
                        print(f"  + {title} (复用已有文件)")
                    else:
                        print(f"  + {title} (上传成功)")
                    upload_ok += 1
                except Exception as e:
                    print(f"  ! {title} (上传失败: {e})")
                    upload_fail += 1
                    # 上传失败也继续入库，只是 url 为空

                bgm = BgmMusic(
                    title=title,
                    artist=None,
                    category=category,
                    file_path=relative_path,
                    url=url,
                    duration=duration,
                    status="active",
                    sort_order=idx,
                )
                session.add(bgm)
                total += 1

        await session.commit()
        print(f"\n导入完成，共 {total} 首")
        print(f"  上传成功: {upload_ok} 首")
        print(f"  上传失败: {upload_fail} 首")


if __name__ == "__main__":
    asyncio.run(import_bgm())