music_analyzer.py 3.7 KB
# -*- coding: utf-8 -*-
"""
音乐分析统一入口
提供简化的 analyze_music() 函数
"""

from typing import Dict, Any, Optional
import os

from .factory import AnalyzerFactory


def analyze_music(
    metadata: Dict[str, Any],
    music_url: str,
    provider: str = None,
    extract_lyrics: bool = False,
    label_level: int = 0,
) -> Optional[Dict[str, Any]]:
    """
    音乐分析统一入口函数

    Args:
        metadata: 音乐元数据字典(如 title, artist 等)
        music_url: 音乐文件 URL
        provider: 提供商(qwen | doubao),默认从环境变量读取
        extract_lyrics: 是否识别歌词
        label_level: 标签级别(0: 一级标签, 1: 一级+二级标签)

    Returns:
        分析结果字典,包含以下字段:
        - genre: 音乐风格(一级风格,如:流行、摇滚)
        - emotion: 情绪列表
        - emotional_intensity: 情绪强度
        - vocal_texture: 人声质感
        - vocal_description: 人声质感描述
        - visual_concept: 视觉概念
        - language: 语种
        - bpm: 节拍数(可选)
        - lyrics: 歌词列表(可选)
        - _model: 使用的模型名称
        - _token_info: Token 使用信息

    Example:
        >>> result = analyze_music(
        ...     metadata={"title": "稻香", "artist": "周杰伦"},
        ...     music_url="https://example.com/music.mp3",
        ...     provider="qwen",
        ...     extract_lyrics=False,
        ... )
        >>> print(result["genre"])
        流行
    """
    if provider is None:
        provider = os.getenv("DEFAULT_MUSIC_ANALYZER", "qwen")

    analyzer = AnalyzerFactory.get_analyzer(provider=provider)

    return analyzer.analyze(
        metadata=metadata,
        music_url=music_url,
        extract_lyrics=extract_lyrics,
        label_level=label_level,
    )


def analyze_music_with_qwen(
    metadata: Dict[str, Any],
    music_url: str,
    extract_lyrics: bool = False,
    label_level: int = 0,
) -> Optional[Dict[str, Any]]:
    """使用通义千问分析音乐"""
    return analyze_music(
        metadata=metadata,
        music_url=music_url,
        provider="qwen",
        extract_lyrics=extract_lyrics,
        label_level=label_level,
    )


def analyze_music_with_doubao(
    metadata: Dict[str, Any],
    music_url: str,
    extract_lyrics: bool = False,
    label_level: int = 0,
) -> Optional[Dict[str, Any]]:
    """使用火山引擎豆包分析音乐"""
    return analyze_music(
        metadata=metadata,
        music_url=music_url,
        provider="doubao",
        extract_lyrics=extract_lyrics,
        label_level=label_level,
    )


def analyze_music_lyrics_only(
    metadata: Dict[str, Any],
    music_url: str,
    provider: str = None,
) -> Optional[Dict[str, Any]]:
    """仅识别歌词,避免重复做基础标签分析"""
    if provider is None:
        provider = os.getenv("DEFAULT_MUSIC_ANALYZER", "qwen")

    analyzer = AnalyzerFactory.get_analyzer(provider=provider)
    if hasattr(analyzer, "analyze_lyrics_only"):
        return analyzer.analyze_lyrics_only(metadata=metadata, music_url=music_url)

    # 兼容未实现 lyrics_only 的提供商
    result = analyzer.analyze(
        metadata=metadata,
        music_url=music_url,
        extract_lyrics=True,
        label_level=0,
    )
    if isinstance(result, dict):
        lyrics = result.get("lyrics", [])
        return {
            "lyrics": lyrics if isinstance(lyrics, list) else [],
            "_model": result.get("_model"),
            "_token_info": result.get("_token_info"),
        }
    return None


def get_available_providers() -> list:
    """获取可用的提供商列表"""
    return AnalyzerFactory.list_providers()