mirror of
https://github.com/kaaninchen/Gleiswechsel.git
synced 2026-09-17 16:52:47 +00:00
transitous rewrite: bug fixes in the info embed
This commit is contained in:
Kaaninchen
+12
-7
@@ -1,6 +1,6 @@
|
||||
import requests
|
||||
import random
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
from src.utils import logger, config, get_train_name, convert_iso_string
|
||||
|
||||
@@ -73,10 +73,15 @@ def get_random_connection(stop_id: str) -> str:
|
||||
logger("Couldn't find any connection", "fatal")
|
||||
return None
|
||||
|
||||
trip_id = random.choice(trip_id)
|
||||
from_station = stop_times[0]["place"]["name"]
|
||||
trip_id = random.choice(trip_ids)
|
||||
for trip in stop_times:
|
||||
if trip.get("tripId") == trip_id:
|
||||
from_station = trip.get("place").get("name")
|
||||
break
|
||||
|
||||
print(from_station)
|
||||
return {
|
||||
"trip_id": random.choice(trip_ids),
|
||||
"trip_id": trip_id,
|
||||
"from_station": from_station
|
||||
}
|
||||
|
||||
@@ -85,7 +90,7 @@ def get_trip_details(trip_id: str, from_station: str) -> dict:
|
||||
|
||||
try:
|
||||
response = requests.get(req, headers=headers)
|
||||
response.raise_for_status
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
except requests.RequestException as e:
|
||||
logger(f"An error occured while trying to get the route details: {e}", "fatal")
|
||||
@@ -120,8 +125,8 @@ def get_trip_details(trip_id: str, from_station: str) -> dict:
|
||||
|
||||
trip_details["stops"][trip_from] = departure
|
||||
for stop in legs["intermediateStops"]:
|
||||
arrival = convert_iso_string(stop["arrival"])
|
||||
trip_details["stops"][stop["name"]] = arrival
|
||||
stop_arrival = convert_iso_string(stop["arrival"])
|
||||
trip_details["stops"][stop["name"]] = stop_arrival
|
||||
if stop.get("name") == from_station:
|
||||
departure_time = stop["departure"]
|
||||
trip_details["departure"] = convert_iso_string(departure_time)
|
||||
|
||||
@@ -1,12 +0,0 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from src.embeds import build_info_embed, build_error_embed
|
||||
|
||||
def setup_commands(bot: discord.Bot):
|
||||
@bot.slash_command(description="Informationen über die aktuelle Fahrt")
|
||||
async def info(ctx):
|
||||
embed = build_info_embed()
|
||||
if embed is None:
|
||||
await build_error_embed("Noch keine Verbindung gesetzt.")
|
||||
return
|
||||
await ctx.respond(embed=embed)
|
||||
@@ -1,4 +0,0 @@
|
||||
import json
|
||||
|
||||
with open("config.json", "r") as file:
|
||||
config = json.load(file)
|
||||
+3
-1
@@ -3,5 +3,7 @@ emoji_list = {
|
||||
"BUS": "🚎",
|
||||
"TRAM": "🚈",
|
||||
"REGIONAL_RAIL": "🚊",
|
||||
"HIGHSPEED_RAIL": "🚅"
|
||||
"HIGHSPEED_RAIL": "🚅",
|
||||
"METRO": "🚇",
|
||||
"SUBWAY": "🚇"
|
||||
}
|
||||
@@ -45,4 +45,6 @@ def build_info_embed() -> discord.Embed:
|
||||
embed.set_author(name=agency)
|
||||
embed.set_thumbnail(url=metadata["logo"])
|
||||
|
||||
color = trip.get("color", "keine farbe :(")
|
||||
print(color)
|
||||
return embed
|
||||
-109
@@ -1,109 +0,0 @@
|
||||
import random
|
||||
import discord
|
||||
from datetime import datetime, timedelta
|
||||
import src.handlers as handlers
|
||||
from src.config import config
|
||||
from src.utils import operator_metadata, format_via_list, resolve_operator, logger
|
||||
|
||||
def format_timestamp(timestr):
|
||||
parsed_time = datetime.strptime(timestr, "%H:%M")
|
||||
now = datetime.now()
|
||||
final_datetime = datetime.now().replace(
|
||||
hour=parsed_time.hour,
|
||||
minute=parsed_time.minute,
|
||||
second=0,
|
||||
microsecond=0
|
||||
)
|
||||
|
||||
if final_datetime <= now:
|
||||
final_datetime += timedelta(days=1)
|
||||
|
||||
return discord.utils.format_dt(final_datetime, style="t")
|
||||
|
||||
def format_iso_timestamp(isostr):
|
||||
parsed_time = discord.utils.parse_time(isostr)
|
||||
return discord.utils.format_dt(parsed_time, style="t")
|
||||
|
||||
def build_embed_footer(operator_slogans):
|
||||
dbf = config.get("dbf", "https://dbf.finalrewind.org")
|
||||
footer_notice = f"Daten großzügig bereitgestellt von {dbf} • Typ: {handlers.train_type}"
|
||||
footer_text = (
|
||||
f"{random.choice(operator_slogans)} • {footer_notice}"
|
||||
if operator_slogans else
|
||||
footer_notice
|
||||
)
|
||||
icon = f"{dbf}/static/icons/icon-96x96.png"
|
||||
return {
|
||||
"text": footer_text,
|
||||
"icon": icon
|
||||
}
|
||||
|
||||
def build_info_embed() -> discord.Embed | None:
|
||||
conn = handlers.current
|
||||
info = handlers.train_info
|
||||
|
||||
if handlers.current is None and handlers.train_info is None:
|
||||
return None
|
||||
|
||||
current_operator = resolve_operator(info["operators"])
|
||||
arrival = format_iso_timestamp(info["arrival"])
|
||||
departure = format_timestamp(conn['departure'])
|
||||
|
||||
operator_infos = operator_metadata(current_operator)
|
||||
|
||||
embed = discord.Embed(
|
||||
title=handlers.train_name,
|
||||
description=f"Abfahrt von {conn['station']} um {departure}. Ankunft um {arrival}",
|
||||
color=operator_infos["color"]
|
||||
)
|
||||
|
||||
if len(conn['via']) > 0:
|
||||
embed.add_field(name="Über", value=format_via_list(conn['via']), inline=False)
|
||||
|
||||
if operator_infos.get("unknown"):
|
||||
logger(f"Metadaten für {current_operator} konten in operators.py nicht gefunden werden")
|
||||
|
||||
embed.set_author(name=current_operator)
|
||||
embed.set_thumbnail(url=operator_infos["logo"])
|
||||
|
||||
route_lines = []
|
||||
for stop in conn['route']:
|
||||
stop_name = stop["name"]
|
||||
if stop_name == conn['station']:
|
||||
route_lines.append(f"• **{stop_name}**")
|
||||
else:
|
||||
route_lines.append(f"• {stop_name}")
|
||||
|
||||
embed.add_field(name="Route", value="\n".join(route_lines))
|
||||
|
||||
footer = build_embed_footer(operator_infos.get("slogan"))
|
||||
embed.set_footer(text=footer["text"], icon_url=footer["icon"])
|
||||
|
||||
return embed
|
||||
|
||||
def build_announcement_embed(msg):
|
||||
current_operator = resolve_operator(handlers.train_info["operators"])
|
||||
operator_infos = operator_metadata(current_operator)
|
||||
|
||||
embed = discord.Embed(
|
||||
title = "Informationen zu Ihrer Fahrt",
|
||||
description=msg,
|
||||
color=operator_infos["color"]
|
||||
)
|
||||
|
||||
embed.set_author(name=current_operator)
|
||||
embed.set_thumbnail(url=operator_infos["logo"])
|
||||
|
||||
footer = build_embed_footer(operator_infos.get("slogan"))
|
||||
embed.set_footer(text=footer["text"], icon_url=footer["icon"])
|
||||
|
||||
return embed
|
||||
|
||||
def build_error_embed(errormsg) -> discord.Embed:
|
||||
embed = discord.Embed(
|
||||
title="Ein Fehler ist aufgetreten!",
|
||||
description=errormsg,
|
||||
color=discord.Colour.red()
|
||||
)
|
||||
|
||||
return embed
|
||||
-154
@@ -1,154 +0,0 @@
|
||||
import discord
|
||||
from datetime import datetime, timedelta
|
||||
import asyncio
|
||||
import random
|
||||
from pathlib import Path
|
||||
|
||||
from src.config import config
|
||||
from src.utils import random_connection, get_train_info, get_channel_formatting, logger
|
||||
from src.embeds import build_announcement_embed, build_info_embed
|
||||
|
||||
current = None
|
||||
_scheduled_task: asyncio.Task | None = None
|
||||
|
||||
async def announcer(bot, announcement):
|
||||
voice_channel = bot.get_channel(config["vc"])
|
||||
if len(voice_channel.members) > 0:
|
||||
match announcement:
|
||||
case "ende":
|
||||
destination = current['destination']
|
||||
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 config["voice_announcements"][0]["enabled"]:
|
||||
await voice_announcer(bot, destination, voice_channel)
|
||||
case "umstieg":
|
||||
embed = build_info_embed()
|
||||
case _:
|
||||
embed = None
|
||||
if embed is None:
|
||||
logger(f"Unbekanntes Announcement: {announcement}")
|
||||
else:
|
||||
await voice_channel.send(embed=embed)
|
||||
else:
|
||||
logger(f"Announcement {announcement} wird geskipped, keiner da")
|
||||
return
|
||||
|
||||
async def voice_announcer(bot: discord.Bot, destination, voice_channel):
|
||||
voice_announcement_config = config["voice_announcements"][0]
|
||||
voice_stations = voice_announcement_config["stations"]
|
||||
|
||||
if destination in voice_stations:
|
||||
announcement_for = destination
|
||||
else:
|
||||
if voice_stations.get("general", "") == "":
|
||||
return
|
||||
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
|
||||
|
||||
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)
|
||||
|
||||
async def rename_vc(bot: discord.Bot, from_scheduler: bool = False):
|
||||
global current, train_name, train_info, train_type, _scheduled_task
|
||||
|
||||
guild = bot.get_guild(int(config["server"]))
|
||||
if guild is None:
|
||||
logger(f"Es konnte kein Server mit der ID {config['server']} gefunden werden! Ist der Bot ein Member?", "fatal")
|
||||
return False
|
||||
|
||||
channel = guild.get_channel(int(config["vc"]))
|
||||
if not isinstance(channel, discord.VoiceChannel):
|
||||
logger(f"Es konnte kein VC mit der ID {config['vc']} auf dem Server gefunden werden", "fatal")
|
||||
return False
|
||||
|
||||
if not from_scheduler and _scheduled_task and not _scheduled_task.done():
|
||||
_scheduled_task.cancel()
|
||||
|
||||
attempt = 0
|
||||
while True:
|
||||
if attempt == 20:
|
||||
logger("Zu viele Fehlversuche. Füge einen anderen Bahnhof hinzu.", "fatal")
|
||||
return "Es konnte kein Zug gefunden werden."
|
||||
|
||||
attempt += 1
|
||||
current = random_connection()
|
||||
if current == None:
|
||||
return None
|
||||
|
||||
parts = current['train'].split()
|
||||
train_type = parts[0]
|
||||
|
||||
if parts[1].isdigit():
|
||||
train = current['train']
|
||||
train_ID = parts[1]
|
||||
else:
|
||||
train = parts[1]
|
||||
train_ID = current['train_number']
|
||||
|
||||
station = current['station']
|
||||
train_info = get_train_info(station=station, train_ID=train_ID, train_type=train_type)
|
||||
if train_info and train_info.get('operators') and train_info.get('arrival'):
|
||||
break
|
||||
|
||||
logger(f"Versuch {attempt}: Fehler bei {current['train']} von {station}, versuche neue Verbindung...")
|
||||
|
||||
train_name = f"{train} nach {current['destination']} von {current['station']}"
|
||||
logger(f"Vorbereitung auf {train_name} (typ: {train_type})")
|
||||
arrival = datetime.fromisoformat((train_info["arrival"]))
|
||||
formatting = get_channel_formatting(train_type)
|
||||
|
||||
|
||||
print("-----------------------------------------")
|
||||
logger(f"Umstieg: {train_name}")
|
||||
logger(f"Betreiber: {''.join(train_info['operators'])}")
|
||||
logger(f"Train-Type: {train_type}")
|
||||
logger(f"Wenn der Name nicht geändert wird bin ich im Cooldown")
|
||||
await channel.edit(name=f"{formatting}{train_name}")
|
||||
await channel.set_status(f"Ankunft um {arrival.strftime('%H:%M')}")
|
||||
logger(f"Name geändert!")
|
||||
|
||||
if config.get("announcements", True):
|
||||
await announcer(bot, "umstieg")
|
||||
|
||||
_scheduled_task = asyncio.create_task(_schedule_next_umstieg(bot, arrival))
|
||||
|
||||
return True
|
||||
|
||||
async def _schedule_next_umstieg(bot, arrival):
|
||||
announcement = config.get("announcements", True)
|
||||
announcement_countdown = random.randrange(180, 300) # letzte station announcement ist meistens 3-5min vor ankunft
|
||||
wait_seconds = (arrival - datetime.now()).total_seconds()
|
||||
if wait_seconds > 0:
|
||||
remaining = str(timedelta(seconds=wait_seconds))
|
||||
logger(f"Nächster Umstieg in {remaining.split('.')[0]} ({arrival.strftime('%H:%M:%S')} Uhr)")
|
||||
|
||||
if announcement and wait_seconds > announcement_countdown:
|
||||
wait_until_end_announcement = wait_seconds - announcement_countdown
|
||||
await asyncio.sleep(wait_until_end_announcement)
|
||||
await announcer(bot, "ende")
|
||||
await asyncio.sleep(announcement_countdown)
|
||||
|
||||
else:
|
||||
await asyncio.sleep(wait_seconds)
|
||||
|
||||
logger("Zug angekommen, wähle neue Verbindung...")
|
||||
await rename_vc(bot, from_scheduler=True)
|
||||
+6
-3
@@ -1,6 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import src.data.operators as operators
|
||||
from src.data.emojis import emoji_list
|
||||
@@ -17,8 +17,11 @@ def logger(msg, log_type="info") -> str:
|
||||
os._exit(1)
|
||||
|
||||
def convert_iso_string(isostring) -> str:
|
||||
datetime_isostring = datetime.fromisoformat(isostring)
|
||||
return datetime_isostring.strftime('%H:%M')
|
||||
dt = datetime.fromisoformat(isostring.replace('Z', '+00:00'))
|
||||
if dt.second >= 30:
|
||||
dt += timedelta(minutes=1)
|
||||
|
||||
return dt.strftime('%H:%M')
|
||||
|
||||
def channel_formatting(mode: str) -> str:
|
||||
formatting = config.get("formatting", "")
|
||||
|
||||
Reference in New Issue
Block a user