v6d3music/v6d3music/run-bot.py

import asyncio
import concurrent.futures
import json
import os
import random
import re
import shlex
import string
import subprocess
import time
from collections import deque
from io import StringIO
from typing import Optional, AsyncIterable, Any, Iterable, TypeAlias

# noinspection PyPackageRequirements
import discord
import nacl.hash
from ptvp35 import Db, KVJson
from v6d1tokens.client import request_token
from v6d2ctx.context import Context, at, escape, monitor, Benchmark, Explicit, Implicit
from v6d2ctx.handle_content import handle_content
from v6d2ctx.lock_for import lock_for
from v6d2ctx.serve import serve

import v6d3music.extract
import v6d3music.ffmpegnormalaudio
from v6d3music.config import prefix, myroot

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
token = loop.run_until_complete(request_token('music', 'token'))
client = discord.Client(
    intents=discord.Intents(
        members=True,
        guilds=True,
        bans=True,
        emojis=True,
        invites=True,
        voice_states=True,
        guild_messages=True,
        reactions=True
    ),
)
volume_db = Db(myroot / 'volume.db', kvrequest_type=KVJson)
queue_db = Db(myroot / 'queue.db', kvrequest_type=KVJson)
cache_db = Db(myroot / 'cache.db', kvrequest_type=KVJson)
cache_root = myroot / 'cache'
cache_root.mkdir(exist_ok=True)

vcs_restored = False


async def restore_vcs():
    global vcs_restored
    vcs: list[tuple[int, int, bool]] = queue_db.get('vcs', [])
    try:
        for vcgid, vccid, vc_is_paused in vcs:
            try:
                guild: discord.Guild = await client.fetch_guild(vcgid)
                async with lock_for(guild, 'not in a guild'):
                    channels = await guild.fetch_channels()
                    channel: discord.VoiceChannel
                    channel, = [ch for ch in channels if ch.id == vccid]
                    vp: discord.VoiceProtocol = await channel.connect()
                    assert isinstance(vp, discord.VoiceClient)
                    vc = vp
                    await main_for_raw_vc(vc, create=True)
                    if vc_is_paused:
                        vc.pause()
            except Exception as e:
                print(f'vc {vcgid} {vccid} {vc_is_paused} failed', e)
            else:
                print(f'vc restored {vcgid} {vccid}')
    finally:
        vcs_restored = True


@client.event
async def on_ready():
    print('ready')
    await client.change_presence(activity=discord.Game(
        name='феноменально',
    ))
    if not vcs_restored:
        await restore_vcs()


@at('commands', 'help')
async def help_(ctx: Context, args: list[str]) -> None:
    match args:
        case []:
            await ctx.reply('music bot')
        case [name]:
            await ctx.reply(f'help for {name}: `{name} help`')


def speed_quotient(options: str) -> float:
    options = options or ''
    options = ''.join(c for c in options if not c.isspace())
    options += ','
    quotient: float = 1.0
    asetrate: str
    for asetrate in re.findall(r'asetrate=([0-9.]+?),', options):
        try:
            quotient *= float(asetrate) / discord.opus.Encoder.SAMPLING_RATE
        except ValueError:
            pass
    atempo: str
    for atempo in re.findall(r'atempo=([0-9.]+?),', options):
        try:
            quotient *= float(atempo)
        except ValueError:
            pass
    quotient = max(0.1, min(10.0, quotient))
    return quotient


def sparq(options: str) -> float:
    return speed_quotient(options) * discord.opus.Encoder.FRAME_LENGTH / 1000


class YTAudio(discord.AudioSource):
    source: discord.FFmpegAudio

    def __init__(
            self,
            url: str,
            origin: str,
            description: str,
            options: Optional[str],
            rby: discord.Member,
            already_read: int,
            tor: bool
    ):
        self.url = url
        self.origin = origin
        self.description = description
        self.options = options
        self.rby = rby
        self.already_read = already_read
        self.tor = tor
        self.loaded = False
        self.regenerating = False
        self.set_source()
        self._durations: dict[str, str] = {}

    def set_source(self):
        self.schedule_duration_update()
        self.source = v6d3music.ffmpegnormalaudio.FFmpegNormalAudio(
            self.url,
            options=self.options,
            before_options=self.before_options(),
            tor=self.tor
        )

    def set_already_read(self, already_read: int):
        self.already_read = already_read
        self.set_source()

    def set_seconds(self, seconds: float):
        self.set_already_read(round(seconds / sparq(self.options)))

    def source_seconds(self) -> float:
        return self.already_read * sparq(self.options)

    def source_timecode(self) -> str:
        seconds = round(self.source_seconds())
        minutes, seconds = divmod(seconds, 60)
        hours, minutes = divmod(minutes, 60)
        return f'{hours}:{minutes:02d}:{seconds:02d}'

    def schedule_duration_update(self):
        asyncio.get_running_loop().create_task(self.update_duration())

    async def update_duration(self):
        url: str = self.url
        if url in self._durations:
            return
        self._durations.setdefault(url, '')
        prompt = ''
        if self.tor:
            prompt = 'torify '
        prompt += (
            f'ffprobe -i {shlex.quote(url)}'
            ' -show_entries format=duration -v quiet -of csv="p=0" -sexagesimal'
        )
        p = subprocess.Popen(
            prompt,
            stdout=subprocess.PIPE,
            shell=True
        )
        with Benchmark('FFP'):
            code = await loop.run_in_executor(None, p.wait)
        if code:
            pass
        else:
            self._durations[url] = p.stdout.read().decode().strip().split('.')[0]

    def duration(self) -> str:
        duration = self._durations.get(self.url)
        if duration is None:
            self.schedule_duration_update()
        return duration or '?:??:??'

    def before_options(self):
        before_options = ''
        if 'https' in self.url:
            before_options += (
                '-reconnect 1 -reconnect_at_eof 0 -reconnect_streamed 1 -reconnect_delay_max 10 -copy_unknown'
            )
        if self.already_read:
            before_options += (
                f' -ss {self.source_seconds()}'
            )
        return before_options

    def read(self) -> bytes:
        if self.regenerating:
            return FILL
        self.already_read += 1
        ret: bytes = self.source.read()
        if ret:
            self.loaded = True
        elif not self.loaded:
            if random.random() > .1:
                self.regenerating = True
                loop.create_task(self.regenerate())
                return FILL
            else:
                print(f'dropped {self.origin}')
        return ret

    def cleanup(self):
        self.source.cleanup()

    def can_be_skipped_by(self, member: discord.Member) -> bool:
        permissions: discord.Permissions = member.guild_permissions
        if permissions.administrator:
            return True
        elif permissions.manage_permissions:
            return True
        elif permissions.manage_guild:
            return True
        elif permissions.manage_channels:
            return True
        elif permissions.manage_messages:
            return True
        else:
            return self.rby == member

    def hybernate(self):
        return {
            'url': self.url,
            'origin': self.origin,
            'description': self.description,
            'options': self.options,
            'rby': self.rby.id,
            'already_read': self.already_read,
            'tor': self.tor,
        }

    @classmethod
    async def respawn(cls, guild: discord.Guild, respawn) -> 'YTAudio':
        return YTAudio(
            respawn['url'],
            respawn['origin'],
            respawn['description'],
            respawn['options'],
            guild.get_member(respawn['rby']) or await guild.fetch_member(respawn['rby']),
            respawn['already_read'],
            respawn.get('tor', False),
        )

    async def regenerate(self):
        try:
            print(f'regenerating {self.origin}')
            self.url = await real_url(self.origin, True, self.tor)
            self.source.cleanup()
            self.set_source()
            print(f'regenerated {self.origin}')
        finally:
            self.regenerating = False


FILL = b'\x00' * discord.opus.Encoder.FRAME_SIZE


def assert_admin(member: discord.Member):
    permissions: discord.Permissions = member.guild_permissions
    if not permissions.administrator:
        raise Explicit('not an administrator')


class QueueAudio(discord.AudioSource):
    def __init__(self, guild: discord.Guild, respawned: list[YTAudio]):
        self.queue: deque[YTAudio] = deque()
        self.queue.extend(respawned)
        self.guild = guild

    @staticmethod
    async def respawned(guild: discord.Guild) -> list[YTAudio]:
        respawned = []
        try:
            for audio_respawn in queue_db.get(guild.id, []):
                try:
                    respawned.append(await YTAudio.respawn(guild, audio_respawn))
                except Exception as e:
                    print('audio respawn failed', e)
                    raise
        except Exception as e:
            print('queue respawn failed', e)
        return respawned

    @classmethod
    async def create(cls, guild: discord.Guild):
        return cls(guild, await QueueAudio.respawned(guild))

    async def save(self):
        hybernated = []
        for audio in list(self.queue):
            await asyncio.sleep(0.01)
            hybernated.append(audio.hybernate())
        queue_db.set_nowait(self.guild.id, hybernated)

    def append(self, audio: YTAudio):
        self.queue.append(audio)

    def read(self) -> bytes:
        if not self.queue:
            return FILL
        audio = self.queue[0]
        frame = audio.read()
        if len(frame) != discord.opus.Encoder.FRAME_SIZE:
            self.queue.popleft().cleanup()
            frame = FILL
        return frame

    def skip_at(self, pos: int, member: discord.Member) -> bool:
        if pos < len(self.queue):
            audio = self.queue[pos]
            if audio.can_be_skipped_by(member):
                self.queue.remove(audio)
                audio.cleanup()
                return True
        return False

    def skip_audio(self, audio: YTAudio, member: discord.Member) -> bool:
        if audio in self.queue:
            if audio.can_be_skipped_by(member):
                self.queue.remove(audio)
                audio.cleanup()
                return True
        return False

    def clear(self, member: discord.Member) -> None:
        assert_admin(member)
        self.cleanup()

    def swap(self, member: discord.Member, a: int, b: int) -> None:
        assert_admin(member)
        if max(a, b) >= len(self.queue):
            return
        self.queue[a], self.queue[b] = self.queue[b], self.queue[a]

    def move(self, member: discord.Member, a: int, b: int) -> None:
        assert_admin(member)
        if max(a, b) >= len(self.queue):
            return
        audio = self.queue[a]
        self.queue.remove(audio)
        self.queue.insert(b, audio)

    async def format(self) -> str:
        stream = StringIO()
        for i, audio in enumerate(list(self.queue)):
            stream.write(f'`[{i}]` `{audio.source_timecode()} / {audio.duration()}` {audio.description}\n')
        return stream.getvalue()

    def cleanup(self):
        self.queue.clear()
        for audio in self.queue:
            try:
                audio.cleanup()
            except ValueError:
                pass


class MainAudio(discord.PCMVolumeTransformer):
    def __init__(self, queue: QueueAudio, volume: float):
        self.queue = queue
        super().__init__(self.queue, volume=volume)

    async def set(self, volume: float, member: discord.Member):
        assert_admin(member)
        if volume < 0.01:
            raise Explicit('volume too small')
        if volume > 1:
            raise Explicit('volume too big')
        self.volume = volume
        await volume_db.set(member.guild.id, volume)


def bytes_hash(b: bytes) -> str:
    return nacl.hash.sha256(b).decode()


def recursive_hash(obj) -> str:
    if isinstance(obj, str):
        return bytes_hash(obj.encode())
    elif isinstance(obj, tuple) or isinstance(obj, list):
        return recursive_hash(';'.join(map(recursive_hash, obj)))
    elif isinstance(obj, dict):
        return recursive_hash([*obj.items()])
    else:
        raise TypeError


async def aextract(params: dict, url: str, **kwargs):
    with Benchmark('AEX'):
        with concurrent.futures.ProcessPoolExecutor() as pool:
            return await loop.run_in_executor(
                pool,
                v6d3music.extract.extract,
                params,
                url,
                kwargs
            )


async def tor_extract(params: dict, url: str, **kwargs):
    print(f'tor extracting {url}')
    p = subprocess.Popen(
        ['torify', 'python', '-m', 'v6d3music.run-extract'],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True
    )
    p.stdin.write(f'{json.dumps(params)}\n')
    p.stdin.write(f'{json.dumps(url)}\n')
    p.stdin.write(f'{json.dumps(kwargs)}\n')
    p.stdin.flush()
    p.stdin.close()
    code = await loop.run_in_executor(None, p.wait)
    if code:
        raise RuntimeError(code)
    return json.loads(p.stdout.read())


async def cache_url(hurl: str, rurl: str, override: bool, tor: bool) -> None:
    async with lock_for(('cache', hurl), 'cache failed'):
        if not override and cache_db.get(f'url:{hurl}', None) is not None:
            return
        cachable: bool = cache_db.get(f'cachable:{hurl}', False)
        if cachable:
            print('caching', hurl)
            path = cache_root / f'{hurl}.opus'
            tmp_path = cache_root / f'{hurl}.tmp.opus'
            args = []
            if tor:
                args.append('torify')
            args.extend(
                [
                    'ffmpeg', '-hide_banner', '-loglevel', 'warning',
                    '-reconnect', '1', '-reconnect_at_eof', '0',
                    '-reconnect_streamed', '1', '-reconnect_delay_max', '10', '-copy_unknown',
                    '-y', '-i', rurl, '-b:a', '128k', str(tmp_path)
                ]
            )
            p = subprocess.Popen(
                args,
            )
            with Benchmark('CCH'):
                code = await loop.run_in_executor(None, p.wait)
            if code:
                raise RuntimeError(code)
            await loop.run_in_executor(None, tmp_path.rename, path)
            await cache_db.set(f'url:{hurl}', str(path))
            print('cached', hurl)
            # await cache_db.set(f'cachable:{hurl}', False)
        else:
            await cache_db.set(f'cachable:{hurl}', True)


async def real_url(url: str, override: bool, tor: bool) -> str:
    hurl: str = bytes_hash(url.encode())
    if not override:
        curl: Optional[str] = cache_db.get(f'url:{hurl}', None)
        if curl is not None:
            print('using cached', hurl)
            return curl
    args = []
    if tor:
        args.append('torify')
    args.extend(
        [
            'youtube-dl', '--no-playlist', '-f', 'bestaudio', '-g', '--', url
        ]
    )
    p = subprocess.Popen(
        args,
        stdout=subprocess.PIPE
    )
    with Benchmark('URL'):
        code = await loop.run_in_executor(None, p.wait)
    if code:
        raise RuntimeError(code)
    rurl: str = p.stdout.readline().decode()[:-1]
    loop.create_task(cache_url(hurl, rurl, override, tor))
    return rurl


def options_for_effects(effects: str) -> Optional[str]:
    return f'-af {shlex.quote(effects)}' if effects else None


async def create_ytaudio(
        ctx: Context, info: dict[str, Any], effects: Optional[str], already_read: int, tor: bool
) -> YTAudio:
    if effects:
        if effects not in allowed_effects:
            assert_admin(ctx.member)
        if not set(effects) <= set(string.ascii_letters + string.digits + '*,=+-/()|.^:_'):
            raise Explicit('malformed effects')
        options = options_for_effects(effects)
    else:
        options = None
    return YTAudio(
        await real_url(info['url'], False, tor),
        info['url'],
        f'{escape(info.get("title", "unknown"))} `Rby` {ctx.member}',
        options,
        ctx.member,
        already_read,
        tor
    )


async def entries_for_url(url: str, tor: bool) -> AsyncIterable[
    dict[str, Any]
]:
    ef = aextract
    if tor:
        ef = tor_extract
    info = await ef(
        {
            'playlistend': 128,
            'logtostderr': True
        },
        url,
        download=False,
        process=False
    )
    if 'entries' in info:
        for entry in info['entries']:
            yield entry
    else:
        yield info | {'url': url}


info_tuple: TypeAlias = tuple[dict[str, Any], str, int, bool]


async def create_ytaudios(ctx: Context, infos: list[info_tuple]) -> AsyncIterable[YTAudio]:
    for audio in await asyncio.gather(
            *[
                create_ytaudio(ctx, info, effects, already_read, tor)
                for
                info, effects, already_read, tor
                in
                infos
            ]
    ):
        yield audio


presets: dict[str, str] = {
    'cursed': 'aeval=val(0)*2*sin(440*t)+val(1)*2*cos(622*t)|val(1)*2*sin(622*t)+val(0)*2*cos(440*t)',
    'bassboost': 'bass=g=10',
    'bassbooboost': 'bass=g=30',
    'nightcore': 'asetrate=67882',
    'daycore': 'atempo=.9,aecho=1.0:0.5:20:0.5',
    'пришествие анимешне': 'bass=g=15,asetrate=67882,bass=g=15',
    'difference': 'aeval=val(0)-val(1)|val(1)-val(0)',
    'mono': 'aeval=.5*val(0)+.5*val(1)|.5*val(1)+.5*val(0)',
}
allowed_presets = ['bassboost', 'bassbooboost', 'nightcore', 'daycore', 'mono']
allowed_effects = {'', *(presets[key] for key in allowed_presets)}


async def entries_effects_for_args(args: list[str]) -> AsyncIterable[info_tuple]:
    while args:
        match args:
            case [url, '-', effects, *args]:
                pass
            case [url, '+', preset, *args]:
                effects = presets[preset]
            case [url, *args]:
                effects = None
            case _:
                raise RuntimeError
        seconds = 0
        match args:
            case [h, m, s, *args] if h.isdecimal() and m.isdecimal() and s.isdecimal():
                seconds = 3600 * int(h) + 60 * int(m) + int(s)
            case [m, s, *args] if m.isdecimal() and s.isdecimal():
                seconds = 60 * int(m) + int(s)
            case [s, *args] if s.isdecimal():
                seconds = int(s)
            case [*args]:
                pass
        already_read = round(seconds / sparq(options_for_effects(effects)))
        tor = False
        match args:
            case ['tor', *args]:
                tor = True
            case [*args]:
                pass
        async for info in entries_for_url(url, tor):
            yield info, effects, already_read, tor


async def yt_audios(ctx: Context, args: list[str]) -> AsyncIterable[YTAudio]:
    tuples: list[info_tuple] = []
    async for info, effects, already_read, tor in entries_effects_for_args(args):
        tuples.append((info, effects, already_read, tor))
        if len(tuples) >= 5:
            async for audio in create_ytaudios(ctx, tuples):
                yield audio
            tuples.clear()
    async for audio in create_ytaudios(ctx, tuples):
        yield audio


mainasrcs: dict[discord.Guild, MainAudio] = {}


async def catch(ctx: Context, args: list[str], reply: str, *catched: (Iterable[str] | str)):
    catched = {(case,) if isinstance(case, str) else tuple(case) for case in catched}
    if tuple(args) in catched:
        await ctx.reply(reply.strip())
        raise Implicit


@at('commands', '/')
@at('commands', 'play')
async def play(ctx: Context, args: list[str]) -> None:
    await catch(
        ctx, args,
        f'''
`play ...args`
`play url [- effects] ...args`
`play url [+ preset] ...args`
presets: {shlex.join(allowed_presets)}
''',
        (), 'help'
    )
    async with lock_for(ctx.guild, 'not in a guild'):
        queue = await queue_for(ctx, create=True)
        async for audio in yt_audios(ctx, args):
            queue.append(audio)
        await ctx.reply('done')


async def raw_vc_for(ctx: Context) -> discord.VoiceClient:
    if ctx.guild is None:
        raise Explicit('not in a guild')
    vc: discord.VoiceProtocol = ctx.guild.voice_client
    if vc is None or isinstance(vc, discord.VoiceClient) and not vc.is_connected():
        vs: discord.VoiceState = ctx.member.voice
        if vs is None:
            raise Explicit('not connected')
        vch: discord.VoiceChannel = vs.channel
        if vch is None:
            raise Explicit('not connected')
        try:
            vc: discord.VoiceProtocol = await vch.connect()
        except discord.ClientException:
            await ctx.guild.fetch_channels()
            raise Explicit('try again later')
    assert isinstance(vc, discord.VoiceClient)
    return vc


async def main_for_raw_vc(vc: discord.VoiceClient, *, create: bool) -> MainAudio:
    if vc.guild in mainasrcs:
        source = mainasrcs[vc.guild]
    else:
        if create:
            source = mainasrcs.setdefault(
                vc.guild,
                MainAudio(await QueueAudio.create(vc.guild), volume=volume_db.get(vc.guild.id, 0.2))
            )
        else:
            raise Explicit('not playing')
    if vc.source != source or create and not vc.is_playing():
        vc.play(source)
    return source


async def vc_main_for(ctx: Context, *, create: bool) -> tuple[discord.VoiceClient, MainAudio]:
    vc = await raw_vc_for(ctx)
    return vc, await main_for_raw_vc(vc, create=create)


async def vc_for(ctx: Context, *, create: bool) -> discord.VoiceClient:
    vc, source = await vc_main_for(ctx, create=create)
    return vc


async def main_for(ctx: Context, *, create: bool) -> MainAudio:
    vc, source = await vc_main_for(ctx, create=create)
    return source


async def queue_for(ctx: Context, *, create: bool) -> QueueAudio:
    return (await main_for(ctx, create=create)).queue


@at('commands', 'skip')
async def skip(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`skip [first] [last]`
''', 'help')
    match args:
        case []:
            queue = await queue_for(ctx, create=False)
            queue.skip_at(0, ctx.member)
        case [pos] if pos.isdecimal():
            pos = int(pos)
            queue = await queue_for(ctx, create=False)
            queue.skip_at(pos, ctx.member)
        case [pos0, pos1] if pos0.isdecimal() and pos1.isdecimal():
            pos0, pos1 = int(pos0), int(pos1)
            queue = await queue_for(ctx, create=False)
            for i in range(pos0, pos1 + 1):
                if not queue.skip_at(pos0, ctx.member):
                    pos0 += 1
        case _:
            raise Explicit('misformatted')
    await ctx.reply('done')


@at('commands', 'to')
async def skip_to(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`to [[h]] [m] s`
''', 'help')
    match args:
        case [h, m, s] if h.isdecimal() and m.isdecimal() and s.isdecimal():
            seconds = 3600 * int(h) + 60 * int(m) + int(s)
        case [m, s] if m.isdecimal() and s.isdecimal():
            seconds = 60 * int(m) + int(s)
        case [s] if s.isdecimal():
            seconds = int(s)
        case _:
            raise Explicit('misformatted')
    queue = await queue_for(ctx, create=False)
    queue.queue[0].set_seconds(seconds)


@at('commands', 'effects')
async def effects_(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`effects - effects`
`effects + preset`
''', 'help')
    match args:
        case ['-', effects]:
            pass
        case ['+', preset]:
            effects = presets[preset]
        case _:
            raise Explicit('misformatted')
    assert_admin(ctx.member)
    queue = await queue_for(ctx, create=False)
    yta = queue.queue[0]
    seconds = yta.source_seconds()
    yta.options = options_for_effects(effects)
    yta.set_seconds(seconds)


@at('commands', 'queue')
async def queue_(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`queue`
`queue clear`
`queue resume`
`queue pause`
''', 'help')
    match args:
        case []:
            await ctx.long((await (await queue_for(ctx, create=False)).format()).strip() or 'no queue')
        case ['clear']:
            (await queue_for(ctx, create=False)).clear(ctx.member)
            await ctx.reply('done')
        case ['resume']:
            async with lock_for(ctx.guild, 'not in a guild'):
                await queue_for(ctx, create=True)
                await ctx.reply('done')
        case ['pause']:
            async with lock_for(ctx.guild, 'not in a guild'):
                vc = await vc_for(ctx, create=True)
                vc.pause()
                await ctx.reply('done')
        case _:
            raise Explicit('misformatted')


@at('commands', 'swap')
async def swap(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`swap a b`
''', 'help')
    match args:
        case [a, b] if a.isdecimal() and b.isdecimal():
            a, b = int(a), int(b)
            (await queue_for(ctx, create=False)).swap(ctx.member, a, b)
        case _:
            raise Explicit('misformatted')


@at('commands', 'move')
async def move(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`move a b`
''', 'help')
    match args:
        case [a, b] if a.isdecimal() and b.isdecimal():
            a, b = int(a), int(b)
            (await queue_for(ctx, create=False)).move(ctx.member, a, b)
        case _:
            raise Explicit('misformatted')


@at('commands', 'volume')
async def volume_(ctx: Context, args: list[str]) -> None:
    await catch(ctx, args, '''
`volume volume`
''', 'help')
    match args:
        case [volume]:
            volume = float(volume)
            await (await main_for(ctx, create=False)).set(volume, ctx.member)
        case _:
            raise Explicit('misformatted')


@at('commands', 'pause')
async def pause(ctx: Context, _args: list[str]) -> None:
    vc = await vc_for(ctx, create=False)
    vc.pause()


@at('commands', 'resume')
async def resume(ctx: Context, _args: list[str]) -> None:
    vc = await vc_for(ctx, create=False)
    vc.resume()


@client.event
async def on_message(message: discord.Message) -> None:
    await handle_content(message, message.content, prefix)


async def save_queues():
    for mainasrc in list(mainasrcs.values()):
        await asyncio.sleep(0.01)
        await mainasrc.queue.save()


async def save_vcs():
    if vcs_restored:
        vcs = []
        vc: discord.VoiceClient
        for vc in list(client.voice_clients):
            await asyncio.sleep(0.01)
            if vc.is_playing():
                if vc.guild is not None and vc.channel is not None:
                    vcs.append((vc.guild.id, vc.channel.id, vc.is_paused()))
        queue_db.set_nowait('vcs', vcs)


async def save_commit():
    await queue_db.set('commit', time.time())


async def save_job():
    while True:
        await asyncio.sleep(1)
        with Benchmark('SVQ'):
            await save_queues()
        with Benchmark('SVV'):
            await save_vcs()
        with Benchmark('SVC'):
            await save_commit()


async def main():
    async with volume_db, queue_db, cache_db:
        await client.login(token)
        loop.create_task(save_job())
        if os.getenv('v6monitor'):
            loop.create_task(monitor())
        subprocess.Popen('tor')
        await client.connect()


if __name__ == '__main__':
    serve(main(), client, loop)