fix duration being float by default
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user