From c563062b9a90253765bffbd59022961d648976b1 Mon Sep 17 00:00:00 2001 From: Kaaninchen <124433727+kaaninchen@users.noreply.github.com> Date: Sun, 16 Aug 2026 17:22:50 +0200 Subject: [PATCH] transitous rewrite: announcements --- main.py | 1 - src/dc/embeds.py | 21 +++++++++++++++- src/dc/handlers.py | 63 +++++++++++++++++++++++++++++++++++++++++++--- src/dc/helpers.py | 2 +- src/utils.py | 28 ++++++++++++++++++++- 5 files changed, 107 insertions(+), 8 deletions(-) diff --git a/main.py b/main.py index 812affc..b11792c 100644 --- a/main.py +++ b/main.py @@ -34,7 +34,6 @@ TODO - discord status - text announcements - voice announcements -- improved error handling (retry connection) - multi language support - README ''' \ No newline at end of file diff --git a/src/dc/embeds.py b/src/dc/embeds.py index 854388a..0e1f11b 100644 --- a/src/dc/embeds.py +++ b/src/dc/embeds.py @@ -55,4 +55,23 @@ def build_info_embed() -> discord.Embed: embed.set_author(name=agency) embed.set_thumbnail(url=metadata["logo"]) - return embed \ No newline at end of file + return embed + +def build_announcement_embed(msg): + from src.dc.handlers import trip + agency = trip["agency"] + metadata = get_operator_metadata(agency, trip["route_color"]) + + embed = discord.Embed( + title = "Informationen zu ihrer Fahrt", + description=msg, + color=metadata["color"] + ) + + embed.set_author(name=agency) + embed.set_thumbnail(url=metadata["logo"]) + + footer = build_embed_footer(trip["mode"], metadata.get("slogan")) + embed.set_footer(text=footer["text"], icon_url=footer["icon"]) + + return embed diff --git a/src/dc/handlers.py b/src/dc/handlers.py index d0b89c6..bcfbb7a 100644 --- a/src/dc/handlers.py +++ b/src/dc/handlers.py @@ -1,8 +1,9 @@ import discord import asyncio +import random from datetime import datetime, timedelta, date -from src.utils import logger, channel_formatting, choose_connection +from src.utils import logger, channel_formatting, choose_connection, config, get_sound_path _scheduled_task: asyncio.Task | None = None @@ -38,9 +39,12 @@ async def rename_vc(bot: discord.Bot, voice_channel, from_scheduler: bool = Fals logger(f"Name geändert!") - _scheduled_task = asyncio.create_task(_schedule_next_transfer(bot, arrival, voice_channel)) + await announcer(bot, "umstieg", voice_channel) -async def _schedule_next_transfer(bot, arrival, voice_channel): + _scheduled_task = asyncio.create_task(_schedule_next_transfer(bot, arrival, voice_channel, trip["to"])) + + +async def _schedule_next_transfer(bot: discord.Bot, arrival, voice_channel: discord.VoiceChannel, destination: str): now = datetime.now() parsed_time = datetime.strptime(arrival, "%H:%M").time() arrival_dt = datetime.combine(date.today(), parsed_time) @@ -49,12 +53,63 @@ async def _schedule_next_transfer(bot, arrival, voice_channel): arrival_dt += timedelta(days=1) wait_seconds = (arrival_dt - now).total_seconds() +# announcement_countdown = random.randrange(180, 300) + announcement_countdown = 3 if wait_seconds > 0: remaining = str(timedelta(seconds=wait_seconds)) logger(f"Nächster Umstieg in {remaining.split('.')[0]} ({arrival} Uhr)") - await asyncio.sleep(wait_seconds) + if wait_seconds > announcement_countdown: + wait_until_end_announcement = wait_seconds - announcement_countdown +# await asyncio.sleep(wait_until_end_announcement) + await asyncio.sleep(5) + await announcer(bot, "ende", voice_channel, destination) + await asyncio.sleep(announcement_countdown) + else: + await asyncio.sleep(wait_seconds) logger("Zug angekommen, wähle neue Verbindung") await rename_vc(bot, voice_channel, from_scheduler=True) +async def announcer(bot: discord.Bot, announcement: str, voice_channel: discord.VoiceChannel, destination = None): + from src.dc.embeds import build_info_embed, build_announcement_embed + + announcements_enabled = config.get("announcements", True) + voice_announcement_enabled = config["voice_announcements"][0]["enabled"] + + if announcements_enabled: + if len(voice_channel.members) > 0: + match announcement: + case "ende": + embed = build_announcement_embed( + f'Sehr geehrte Fahrgäste,\nIn wenigen Minuten erreichen wir {destination}. Dieser Zug endet dort.\n\nWir wünschen Ihnen eine angenehme Weiterreise.\n\nVielen Dank für ihr Vertrauen und auf Wiedersehen.') + if voice_announcement_enabled: + await voice_announcer(bot, destination, voice_channel) + case "umstieg": + embed = build_info_embed() + case _: + logger(f"Unbekanntes Announcements: {announcement}") + embed = None + if embed: + await voice_channel.send(embed=embed) + else: + logger(f"Announcement {announcement} wird geskipped, keiner da") + return + +async def voice_announcer(bot: discord.bot, destination: str, voice_channel: discord.VoiceChannel): + sound_path = get_sound_path(destination=destination) + if sound_path is None: + return + + logger(f"VC wird betreten, spiele {sound_path}") + vc = await voice_channel.connect(timeout=15, reconnect=True) + audio_source = discord.FFmpegPCMAudio(sound_path) + + if not vc.is_playing(): + def after_playing(error): + if error: + logger(f"Player error: {error}", "error") + bot.loop.create_task(vc.disconnect()) + logger("VC wird verlassen") + + vc.play(audio_source, after=after_playing) \ No newline at end of file diff --git a/src/dc/helpers.py b/src/dc/helpers.py index cf1c7a8..23ee76f 100644 --- a/src/dc/helpers.py +++ b/src/dc/helpers.py @@ -1,6 +1,6 @@ import discord from datetime import datetime, timedelta -from src.utils import logger, convert_iso_string +from src.utils import logger def validate_channel(bot: discord.bot, server_id: int, channel_id: int): guild = bot.get_guild(server_id) diff --git a/src/utils.py b/src/utils.py index cfbeb7f..ed07d3a 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,6 +1,8 @@ import json import os import importlib +import random +from pathlib import Path from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo @@ -129,4 +131,28 @@ def get_operator_metadata(agency: str, route_color: str) -> dict: "logo": logo, "color": color, "slogans": slogans - } \ No newline at end of file + } + +def get_sound_path(destination) -> str | None: + voice_announcement_config = config["voice_announcements"][0] + voice_stations = voice_announcement_config["stations"] + + if destination in voice_stations: + announcement_for = destination + else: + general_config = voice_stations.get("general", "") + if general_config == "": + return None + announcement_for = "general" + + if voice_stations.values() == list: + sound_file = random.choice(voice_stations.get(announcement_for)) + else: + sound_file = voice_stations.get(announcement_for) + + sound_path = f"src/data/announcements/{sound_file}" + if Path(sound_path).is_file() is False: + logger(f"Konnte Datei {sound_path} nicht finden", "error") + return None + + return sound_path \ No newline at end of file