52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
import asyncio
|
|
import time
|
|
from datetime import timedelta
|
|
from html import escape
|
|
|
|
from ..mutes import MAX_MUTE_MINUTES, SOURCE_CHAT, MuteError
|
|
from .arguments import ArgumentError, parse_quiet_args
|
|
from .constants import PREFIXES
|
|
from .context import CommandContext, MutePort
|
|
from .duration import format_duration
|
|
from .models import Reply
|
|
from .render import plain
|
|
from .targets import TargetError, resolve_reply_target
|
|
|
|
USAGE = "Verwendung: {p}mute <host> [service] <dauer> [grund] oder als Antwort auf einen Alarm: {p}mute 2h"
|
|
UNAVAILABLE = "Stummschalten ist nicht verfügbar."
|
|
UNTIL_FORMAT = "%d.%m. %H:%M"
|
|
MAX_MUTE = timedelta(minutes=MAX_MUTE_MINUTES)
|
|
|
|
|
|
def created_by(backend: str, sender_id: str) -> str:
|
|
return f"{backend}:{sender_id}"
|
|
|
|
|
|
async def create_mute(mutes: MutePort, backend: str, sender_id: str, host: str, service: str,
|
|
duration: timedelta, reason: str) -> Reply:
|
|
try:
|
|
mute = await asyncio.to_thread(mutes.create, host, service, duration, reason,
|
|
created_by(backend, sender_id), SOURCE_CHAT)
|
|
except MuteError as exc:
|
|
return plain(f"Stummschaltung nicht angelegt: {exc}")
|
|
until = time.strftime(UNTIL_FORMAT, time.localtime(mute.expires_at))
|
|
text = f"Stumm: {mute.host_glob} / {mute.service_glob} für {format_duration(duration)}, bis {until}."
|
|
return Reply(text, escape(text))
|
|
|
|
|
|
async def mute_command(ctx: CommandContext) -> Reply:
|
|
if ctx.deps.mutes is None:
|
|
return plain(UNAVAILABLE)
|
|
usage = USAGE.format(p=PREFIXES.get(ctx.message.backend, "/"))
|
|
try:
|
|
parsed = parse_quiet_args(ctx.command.args, MAX_MUTE, service_required=False)
|
|
host, service = parsed.host, parsed.service or ""
|
|
if host is None:
|
|
host, service = await resolve_reply_target(ctx, escape_globs=True, whole_host_hint=True)
|
|
except ArgumentError as exc:
|
|
return plain(f"{exc} {usage}")
|
|
except TargetError as exc:
|
|
return plain(str(exc))
|
|
return await create_mute(ctx.deps.mutes, ctx.message.backend, ctx.message.sender_id, host, service,
|
|
parsed.duration, parsed.comment)
|