src/modules/media_player.py

"""
霜云(Shimokumo) - 多媒体播放模块

支持音频/视频URL解析、流媒体支持、格式检测和播放列表管理。
"""

import os
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from urllib.parse import urlparse, parse_qs

from utils.logger import get_logger

logger = get_logger("Shimokumo.MediaPlayer")


@dataclass
class MediaItem:
    """媒体项数据类"""
    url: str
    """媒体URL"""
    title: str = ""
    """标题"""
    media_type: str = ""
    """类型: audio/video/image"""
    format: str = ""
    """格式: mp3/mp4/..."""
    duration: float = 0.0
    """时长（秒）"""
    size: int = 0
    """文件大小（字节）"""
    bitrate: int = 0
    """比特率"""
    artist: str = ""
    """艺术家（音频）"""
    album: str = ""
    """专辑（音频）"""
    thumbnail: str = ""
    """缩略图URL"""
    metadata: Dict[str, str] = field(default_factory=dict)
    """额外元数据"""

    def is_valid(self) -> bool:
        """检查媒体项是否有效"""
        return bool(self.url)

    def to_dict(self) -> Dict[str, any]:
        """转为字典"""
        return {
            "url": self.url,
            "title": self.title,
            "media_type": self.media_type,
            "format": self.format,
            "duration": self.duration,
            "artist": self.artist,
            "album": self.album,
            "thumbnail": self.thumbnail,
        }


@dataclass
class Playlist:
    """播放列表数据类"""
    name: str = "默认列表"
    """列表名称"""
    items: List[MediaItem] = field(default_factory=list)
    """媒体项列表"""
    current_index: int = 0
    """当前播放索引"""
    is_shuffle: bool = False
    """是否随机播放"""
    repeat_mode: str = "none"
    """重复模式: none/one/all"""

    def add_item(self, item: MediaItem) -> None:
        """添加媒体项"""
        self.items.append(item)

    def remove_item(self, index: int) -> bool:
        """移除指定索引的媒体项"""
        if 0 <= index < len(self.items):
            self.items.pop(index)
            if self.current_index >= len(self.items):
                self.current_index = max(0, len(self.items) - 1)
            return True
        return False

    def current_item(self) -> Optional[MediaItem]:
        """获取当前播放项"""
        if 0 <= self.current_index < len(self.items):
            return self.items[self.current_index]
        return None

    def next_item(self) -> Optional[MediaItem]:
        """获取下一首"""
        if not self.items:
            return None
        if self.repeat_mode == "one":
            return self.current_item()

        self.current_index += 1
        if self.current_index >= len(self.items):
            if self.repeat_mode == "all":
                self.current_index = 0
            else:
                self.current_index = len(self.items) - 1
                return None
        return self.current_item()

    def prev_item(self) -> Optional[MediaItem]:
        """获取上一首"""
        if not self.items:
            return None
        self.current_index -= 1
        if self.current_index < 0:
            self.current_index = len(self.items) - 1 if self.repeat_mode == "all" else 0
        return self.current_item()

    def clear(self) -> None:
        """清空列表"""
        self.items.clear()
        self.current_index = 0

    def __len__(self) -> int:
        return len(self.items)


class MediaPlayerModule:
    """多媒体播放模块

    提供音频/视频URL解析、格式检测、流媒体支持和播放列表管理。

    功能：
    - 媒体URL解析和格式检测
    - 主流视频平台URL支持（B站、YouTube等）
    - 播放列表管理
    - 流媒体URL提取
    - 媒体信息获取

    用法：
        player = MediaPlayerModule()
        media = player.parse_url("https://example.com/music.mp3")
        print(f"类型: {media.media_type}, 格式: {media.format}")
        player.playlist.add_item(media)
    """

    # 音频格式映射
    AUDIO_FORMATS = {
        "mp3": "audio/mpeg",
        "wav": "audio/wav",
        "flac": "audio/flac",
        "aac": "audio/aac",
        "ogg": "audio/ogg",
        "wma": "audio/x-ms-wma",
        "m4a": "audio/mp4",
        "opus": "audio/opus",
        "ape": "audio/x-ape",
        "alac": "audio/alac",
        "webm": "audio/webm",
    }

    # 视频格式映射
    VIDEO_FORMATS = {
        "mp4": "video/mp4",
        "avi": "video/x-msvideo",
        "mkv": "video/x-matroska",
        "mov": "video/quicktime",
        "wmv": "video/x-ms-wmv",
        "flv": "video/x-flv",
        "webm": "video/webm",
        "m4v": "video/mp4",
        "ts": "video/mp2t",
        "3gp": "video/3gpp",
        "mpeg": "video/mpeg",
        "mpg": "video/mpeg",
    }

    # 图片格式映射
    IMAGE_FORMATS = {
        "jpg": "image/jpeg",
        "jpeg": "image/jpeg",
        "png": "image/png",
        "gif": "image/gif",
        "webp": "image/webp",
        "bmp": "image/bmp",
        "svg": "image/svg+xml",
        "ico": "image/x-icon",
        "tiff": "image/tiff",
    }

    # 已知视频平台域名模式
    VIDEO_PLATFORMS = {
        "bilibili": ["bilibili.com", "b23.tv", "acgvideo.com"],
        "youtube": ["youtube.com", "youtu.be"],
        "nicovideo": ["nicovideo.jp", "nico.jp"],
        "dailymotion": ["dailymotion.com"],
        "vimeo": ["vimeo.com"],
        "twitter": ["twitter.com", "x.com"],
    }

    def __init__(self):
        """初始化多媒体播放模块"""
        self.playlist = Playlist()
        self._cache: Dict[str, MediaItem] = {}

    def detect_format(self, url: str) -> Tuple[str, str]:
        """
        检测URL对应的媒体格式。

        Args:
            url: 媒体URL

        Returns:
            (media_type, format) 元组
            media_type: "audio" / "video" / "image" / "unknown"
            format: 格式扩展名
        """
        # 从URL中提取扩展名
        parsed = urlparse(url)
        path = parsed.path.lower()

        # 匹配扩展名
        ext_match = re.search(r"\.(\w{2,5})(?:\?|$)", path)
        if ext_match:
            ext = ext_match.group(1)

            if ext in self.AUDIO_FORMATS:
                return "audio", ext
            elif ext in self.VIDEO_FORMATS:
                return "video", ext
            elif ext in self.IMAGE_FORMATS:
                return "image", ext

        # 检查已知平台
        hostname = parsed.hostname or ""
        for platform, domains in self.VIDEO_PLATFORMS.items():
            if any(d in hostname for d in domains):
                return "video", "platform"

        # 检查URL中的流媒体关键词
        url_lower = url.lower()
        stream_indicators = {
            "audio": ["/audio/", "/music/", "/mp3/", ".mp3?", "audio/", "stream/audio"],
            "video": ["/video/", "/watch/", "/play/", ".mp4?", ".m3u8", "stream/video", "playlist.m3u8"],
        }

        for media_type, indicators in stream_indicators.items():
            for indicator in indicators:
                if indicator in url_lower:
                    return media_type, "stream"

        return "unknown", ""

    def parse_url(self, url: str) -> MediaItem:
        """
        解析媒体URL，提取媒体信息。

        Args:
            url: 媒体URL

        Returns:
            MediaItem对象
        """
        # 检查缓存
        if url in self._cache:
            return self._cache[url]

        media_type, fmt = self.detect_format(url)

        title = self._extract_title_from_url(url)

        item = MediaItem(
            url=url,
            title=title,
            media_type=media_type,
            format=fmt,
        )

        # 尝试获取更多媒体信息
        self._enrich_media_info(item)

        # 缓存结果
        self._cache[url] = item

        logger.info(f"解析媒体URL: {url[:60]}... -> 类型={media_type}, 格式={fmt}")
        return item

    def _extract_title_from_url(self, url: str) -> str:
        """从URL中提取标题"""
        parsed = urlparse(url)
        path = parsed.path

        # 从路径中提取文件名
        filename = os.path.basename(path)
        if filename and "." in filename:
            name = os.path.splitext(filename)[0]
            # URL解码
            import urllib.parse
            try:
                name = urllib.parse.unquote(name)
            except Exception:
                pass
            return name.replace("-", " ").replace("_", " ").strip()

        # 从查询参数中提取标题
        params = parse_qs(parsed.query)
        if "title" in params:
            return params["title"][0]

        return "未命名媒体"

    def _enrich_media_info(self, item: MediaItem) -> None:
        """
        尝试通过HTTP HEAD请求获取媒体信息。

        Args:
            item: 媒体项
        """
        try:
            import requests
            resp = requests.head(
                item.url,
                timeout=10,
                allow_redirects=True,
            )
            # 获取文件大小
            content_length = resp.headers.get("Content-Length")
            if content_length:
                item.size = int(content_length)

            # 获取内容类型
            content_type = resp.headers.get("Content-Type", "")
            if "audio" in content_type:
                item.media_type = "audio"
            elif "video" in content_type:
                item.media_type = "video"
            elif "image" in content_type:
                item.media_type = "image"

            # 获取额外元数据
            for header in resp.headers:
                if header.lower().startswith("x-media-") or header.lower() == "icy-description":
                    item.metadata[header] = resp.headers[header]

        except Exception as e:
            logger.debug(f"获取媒体信息失败: {e}")

    def resolve_stream_url(self, url: str) -> Optional[str]:
        """
        解析流媒体URL，返回可播放的直接链接。

        支持解析M3U8/HLS播放列表。

        Args:
            url: 流媒体URL

        Returns:
            可播放的直接URL，失败返回None
        """
        if url.endswith(".m3u8") or ".m3u8?" in url:
            # HLS流：返回第一个TS分片或直接返回m3u8
            try:
                import requests
                resp = requests.get(url, timeout=10)
                if resp.ok:
                    lines = resp.text.strip().split("\n")
                    for line in lines:
                        line = line.strip()
                        if line and not line.startswith("#"):
                            if line.startswith("http"):
                                return line
                            else:
                                from urllib.parse import urljoin
                                return urljoin(url, line)
            except Exception as e:
                logger.warning(f"解析HLS流失败: {e}")

        return url

    def add_to_playlist(self, url: str, title: str = "") -> MediaItem:
        """
        解析URL并添加到播放列表。

        Args:
            url: 媒体URL
            title: 可选标题

        Returns:
            添加的MediaItem
        """
        item = self.parse_url(url)
        if title:
            item.title = title
        self.playlist.add_item(item)
        logger.info(f"添加到播放列表: {item.title}")
        return item

    def get_playlist_info(self) -> str:
        """
        获取播放列表信息文本。

        Returns:
            格式化的播放列表文本
        """
        if not self.playlist:
            return "播放列表是空的说..."

        parts: List[str] = [
            f"播放列表「{self.playlist.name}」共 {len(self.playlist)} 首：\n"
        ]

        for i, item in enumerate(self.playlist.items, 1):
            marker = " > " if i - 1 == self.playlist.current_index else "   "
            type_icon = {"audio": "[音]", "video": "[视]", "image": "[图]"}.get(item.media_type, "[?]")
            duration = ""
            if item.duration > 0:
                mins = int(item.duration // 60)
                secs = int(item.duration % 60)
                duration = f" ({mins}:{secs:02d})"

            parts.append(f"{marker}{i}. {type_icon} {item.title}{duration}")

        current = self.playlist.current_item()
        if current:
            parts.append(f"\n当前播放: {current.title}")

        return "\n".join(parts)

    def parse_media_urls_from_text(self, text: str) -> List[MediaItem]:
        """
        从文本中提取所有媒体URL并解析。

        Args:
            text: 输入文本

        Returns:
            解析后的媒体项列表
        """
        url_pattern = r"https?://[^\s<>\"]+|[a-zA-Z0-9-]+\.[a-zA-Z]{2,}[^\s<>\"]*"
        urls = re.findall(url_pattern, text)

        media_items: List[MediaItem] = []
        for url in urls:
            item = self.parse_url(url)
            if item.media_type in ("audio", "video"):
                media_items.append(item)

        return media_items