From cff48fbd2edf703edd49c94b88da252ccb3547a2 Mon Sep 17 00:00:00 2001 From: perforat Date: Sat, 30 May 2026 09:29:39 +0000 Subject: [PATCH] fix duration being float by default --- cogs/music/player.py | 239 ++++++++++++++++++++++--------------------- 1 file changed, 120 insertions(+), 119 deletions(-) diff --git a/cogs/music/player.py b/cogs/music/player.py index 42fc906..3b2b33b 100644 --- a/cogs/music/player.py +++ b/cogs/music/player.py @@ -1,120 +1,121 @@ -import asyncio -import logging -import re -from collections import deque -from dataclasses import dataclass, field -from typing import Optional - -logger = logging.getLogger(__name__) - -YTDL_OPTIONS = { - "format": "bestaudio/best", - "noplaylist": False, - "quiet": True, - "no_warnings": True, - "default_search": "ytsearch", - "source_address": "0.0.0.0", - "extract_flat": "in_playlist", -} - -FFMPEG_OPTIONS = { - "before_options": "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5", - "options": "-vn", -} - - -# track - -@dataclass -class Track: - url: str - title: str - duration: Optional[int] = None - requester: Optional[str] = None - stream_url: Optional[str] = None - - def duration_str(self) -> str: - if self.duration is None: - return "?" - m, s = divmod(self.duration, 60) - h, m = divmod(m, 60) - return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}" - - -# guildplayer - -@dataclass -class GuildPlayer: - queue: deque = field(default_factory=deque) - current: Optional[Track] = None - loop: bool = False - loop_queue: bool = False - volume: float = 0.5 - _skipped: bool = False - - -# helper - -async def resolve_track(query: str, requester: str) -> list[Track]: - """Resolve a URL or search query to a list of Track objects via yt-dlp.""" - try: - import yt_dlp # type: ignore - except ImportError: - raise RuntimeError("yt-dlp is not installed") - - loop = asyncio.get_event_loop() - - def _extract(): - opts = dict(YTDL_OPTIONS) - if not re.match(r"https?://", query): - opts["default_search"] = "ytsearch1" - with yt_dlp.YoutubeDL(opts) as ydl: - return ydl.extract_info(query, download=False) - - info = await loop.run_in_executor(None, _extract) - if info is None: - return [] - - tracks: list[Track] = [] - entries = info.get("entries") - if entries: - for entry in entries: - if entry is None: - continue - tracks.append(Track( - url=entry.get("webpage_url") or entry.get("url", ""), - title=entry.get("title", "Unknown"), - duration=entry.get("duration"), - requester=requester, - )) - else: - tracks.append(Track( - url=info.get("webpage_url") or info.get("url", ""), - title=info.get("title", "Unknown"), - duration=info.get("duration"), - requester=requester, - stream_url=info.get("url"), - )) - - return tracks - - -async def get_stream_url(track: Track) -> str: - if track.stream_url: - return track.stream_url - - try: - import yt_dlp # type: ignore - except ImportError: - raise RuntimeError("yt-dlp is not installed") - - loop = asyncio.get_event_loop() - - def _extract(): - opts = {**YTDL_OPTIONS, "extract_flat": False, "noplaylist": True} - with yt_dlp.YoutubeDL(opts) as ydl: - return ydl.extract_info(track.url, download=False) - - info = await loop.run_in_executor(None, _extract) - track.stream_url = info.get("url", "") +import asyncio +import logging +import re +from collections import deque +from dataclasses import dataclass, field +from typing import Optional + +logger = logging.getLogger(__name__) + +YTDL_OPTIONS = { + "format": "bestaudio/best", + "noplaylist": False, + "quiet": True, + "no_warnings": True, + "default_search": "ytsearch", + "source_address": "0.0.0.0", + "extract_flat": "in_playlist", +} + +FFMPEG_OPTIONS = { + "before_options": "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5", + "options": "-vn", +} + + +# track + +@dataclass +class Track: + url: str + title: str + duration: Optional[int] = None + requester: Optional[str] = None + stream_url: Optional[str] = None + + def duration_str(self) -> str: + if self.duration is None: + return "?" + total = int(self.duration) + m, s = divmod(total, 60) + h, m = divmod(m, 60) + return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}" + + +# guildplayer + +@dataclass +class GuildPlayer: + queue: deque = field(default_factory=deque) + current: Optional[Track] = None + loop: bool = False + loop_queue: bool = False + volume: float = 0.5 + _skipped: bool = False + + +# helper + +async def resolve_track(query: str, requester: str) -> list[Track]: + """Resolve a URL or search query to a list of Track objects via yt-dlp.""" + try: + import yt_dlp # type: ignore + except ImportError: + raise RuntimeError("yt-dlp is not installed") + + loop = asyncio.get_event_loop() + + def _extract(): + opts = dict(YTDL_OPTIONS) + if not re.match(r"https?://", query): + opts["default_search"] = "ytsearch1" + with yt_dlp.YoutubeDL(opts) as ydl: + return ydl.extract_info(query, download=False) + + info = await loop.run_in_executor(None, _extract) + if info is None: + return [] + + tracks: list[Track] = [] + entries = info.get("entries") + if entries: + for entry in entries: + if entry is None: + continue + tracks.append(Track( + url=entry.get("webpage_url") or entry.get("url", ""), + title=entry.get("title", "Unknown"), + duration=entry.get("duration"), + requester=requester, + )) + else: + tracks.append(Track( + url=info.get("webpage_url") or info.get("url", ""), + title=info.get("title", "Unknown"), + duration=info.get("duration"), + requester=requester, + stream_url=info.get("url"), + )) + + return tracks + + +async def get_stream_url(track: Track) -> str: + if track.stream_url: + return track.stream_url + + try: + import yt_dlp # type: ignore + except ImportError: + raise RuntimeError("yt-dlp is not installed") + + loop = asyncio.get_event_loop() + + def _extract(): + opts = {**YTDL_OPTIONS, "extract_flat": False, "noplaylist": True} + with yt_dlp.YoutubeDL(opts) as ydl: + return ydl.extract_info(track.url, download=False) + + info = await loop.run_in_executor(None, _extract) + track.stream_url = info.get("url", "") return track.stream_url \ No newline at end of file