transitous: basic rename_vc functionality

main.py

@@ -1,40 +1,20 @@
 import discord
-import random
+from src.utils import config, logger
-from discord.ext import tasks
+from src.dc.handlers import rename_vc
-from src.config import config
+from src.dc.helpers import validate_channel
-from src.handlers import rename_vc
-from src.commands import setup_commands
-from src.embeds import build_error_embed
-from src.utils import logger
-from src.data.status import discord_status
 
 bot = discord.Bot(intents=discord.Intents.all())
-setup_commands(bot)
 
-@tasks.loop(minutes=5)
-async def change_status():
-    status = random.choice(discord_status)
-    await bot.change_presence(activity=discord.Game(name=f"{status} • /info"))
-
-_bot_initialized = False
 @bot.event
 async def on_ready():
-    global _bot_initialized
+    server_id = config["server"]
-    print(f"{bot.user} ist online")
+    server_vc_id = config["vc"]
-    if not change_status.is_running():
+    channel = validate_channel(bot=bot, server_id=server_id, channel_id=server_vc_id)
-        change_status.start()
-    if not _bot_initialized:
-        _bot_initialized = True
-        await rename_vc(bot)
-    else:
-        logger("Discord Reconnect, laufende Fahrt bleibt unangetastet")
 
-@bot.event
+    logger(f"{bot.user} ist online")
-async def on_application_command_error(ctx, error):
+    await rename_vc(bot=bot, voice_channel=channel)
-    embed = build_error_embed(f"Ein Fehler ist aufgetreten: {error}")
-    await ctx.respond(embed=embed)
 
-try: 
+try:
-    bot.run(config['token'])
+    bot.run(config["token"])
 except:
-    logger("Fehler beim parsen des token", "fatal")
+    logger("Feher peim parsen des tokens", "fatal")

src/api/transitous.py

@@ -1,8 +1,8 @@
 import json
 import requests
 import random
-from src.utils import logger
+
-from ... import config
+from src.utils import logger, config
 
 stations = config["stations"]
 blacklist = config["blacklist"]
@@ -14,7 +14,8 @@ headers = {
 
 endpoint = "https://api.transitous.org"
 
-def get_stop_id(stop):
+def get_random_stop_id() -> str:
+    stop = random.choice(stations)
     req = f"{endpoint}/api/v1/geocode?text={stop}"
 
     try:
@@ -22,7 +23,7 @@ def get_stop_id(stop):
         response.raise_for_status()
         data = response.json()
     except requests.RequestException as e:
-        logger(f"An error occured while searching for a connection: {e}")
+        logger(f"An error occured while searching for a connection: {e}", "fatal")
         return None
 
     if response.status_code == 404:
@@ -33,7 +34,7 @@ def get_stop_id(stop):
             continue
         return entry["id"]
 
-def get_random_connection(stop_id):
+def get_random_connection(stop_id: str) -> str:
     max_pages = 5
     cursor = None
     count = 20
@@ -68,12 +69,12 @@ def get_random_connection(stop_id):
             break
 
         if not trip_ids:
-            print("Couldn't find any connection")
+            logger("Couldn't find any connection", "fatal")
             return None
         
     return random.choice(trip_ids)
 
-def get_trip_details(trip_id):
+def get_trip_details(trip_id: str) -> dict:
     req = f"{endpoint}/api/v2/trip?tripId={trip_id}"
 
     try:
@@ -81,18 +82,20 @@ def get_trip_details(trip_id):
         response.raise_for_status
         data = response.json()
     except requests.RequestException as e:
-        print("e")
+        logger(f"An error occured while trying to get the route details: {e}", "fatal")
         return None
 
     legs = data["legs"][0]
 
+    display_name = legs["displayName"]
     trip_from = legs["tripFrom"]["name"]
     trip_to = legs["tripTo"]["name"]
     start_time = legs["startTime"]
     end_time = legs["endTime"]
 
     trip_details = {
-        "name": legs["displayName"],
+        "long_name": f"{display_name} nach {trip_to} von {trip_from}",
+        "short_name": display_name,
         "from": trip_from,
         "to": trip_to,
         "agency": legs["agencyName"],
@@ -111,10 +114,3 @@ def get_trip_details(trip_id):
 
     return trip_details    
 
-assigned_station = random.choice(stations)
-print(assigned_station)
-
-stop_id = get_stop_id(assigned_station)
-trip_id = get_random_connection(stop_id)
-print(json.dumps(get_trip_details(trip_id), indent=2, ensure_ascii=False))
-

src/dc/handlers.py

@@ -0,0 +1,26 @@
+import discord
+import asyncio
+
+from src.api import transitous
+from src.utils import logger
+
+_scheduled_task: asyncio.Task | None = None
+
+async def rename_vc(bot: discord.Bot, voice_channel, from_scheduler: bool = False):
+    if not from_scheduler and _scheduled_task and not _scheduled_task.done():
+        _scheduled_task.cancel()
+
+    station_id = transitous.get_random_stop_id()
+    trip_id = transitous.get_random_connection(station_id)
+    trip = transitous.get_trip_details(trip_id)
+
+    train_name = trip["long_name"]
+    
+    print("-----------------")
+    logger(f"Umstieg: {train_name}")
+    logger(f"Betreiber: {trip["agency"]}, Typ: {trip["mode"]}")
+    logger(f"Sprachkanal wird geändert, wenn nichts passiert bin ich im cooldown (warte einen moment!)")
+
+    await voice_channel.edit(name=train_name)
+    logger(f"Name geändert!")
+

src/dc/helpers.py

@@ -0,0 +1,15 @@
+import discord
+from  src.utils import logger
+
+def validate_channel(bot: discord.bot, server_id: int, channel_id: int):
+    guild = bot.get_guild(server_id)
+    channel = guild.get_channel(channel_id)
+
+    if guild is None:
+        logger(f"Es konnte kein Server mit der ID {server_id} gefunden werden", "fatal")
+
+    if not isinstance(channel, discord.VoiceChannel):
+        logger(f"Es konnte kein VC mit der id {channel_id} gefunden werden", "fatal")
+        return False
+
+    return channel

src/utils.py

@@ -1,147 +1,11 @@
-import requests
+import json
-import random
 import os
-import importlib
 from datetime import datetime
-from src.config import config
-import src.data.operators as operators_module
-import src.data.emojis as emojis
 
-_operators_mtime = None
+with open("config.json", "r") as file:
-dbf = config.get("dbf", "https://dbf.finalrewind.org")
+    config = json.load(file)
 
-def _reload_operators_if_changed():
+def logger(msg, log_type="info") -> str:
-    global _operators_mtime
-
-    path = operators_module.__file__
-    current_mtime = os.path.getmtime(path)
-
-    if _operators_mtime is None:
-        _operators_mtime = current_mtime
-        return
-    
-    if current_mtime != _operators_mtime:
-        importlib.reload(operators_module)
-        _operators_mtime = current_mtime
-        logger("operators.py wurde automatisch neu geladen (Änderungen erkannt)")
-
-def random_connection():
-    available_stations = config["stations"].copy()
-    while True:
-        if not available_stations:
-            logger("Keine validen Bahnhöfe. Schlag den richtigen Bahnhofsnamen auf {dbf} nach", "fatal")
-       
-        station = random.choice(available_stations)
-        url = f"{dbf}/{station}.json"
-        blacklist = config.get("blacklist", [])
-
-        try:
-            response = requests.get(url)
-            response.raise_for_status()
-            data = response.json()
-        except requests.RequestException as e:
-            logger(f"Fehler beim aussuchen der Verbindung: {e}", "fatal")
-            return
-        
-        if response.status_code == 300:
-            logger(f"Bahnhof '{station}' konnte nicht gefunden werden ({url})", "error")
-            available_stations.remove(station)
-            continue
-
-        departures = [
-            d for d in data.get("departures", [])
-            if d.get("scheduledDeparture") and d.get("destination") != station
-            and not d.get("train", "").startswith(tuple(blacklist))
-        ]
-
-        if not departures:
-            continue
-
-        if not config['random']:
-            dep = departures[0]
-        else:
-            dep = random.choice(departures)
-
-        return {
-            "train": dep['train'], 
-            "destination": dep['destination'], 
-            "route": dep['route'], 
-            "departure": dep['scheduledDeparture'], 
-            "via": dep['via'], 
-            "station": station,
-            "train_number": dep['trainNumber']
-        }
-
-def get_train_info(station, train_ID, train_type):
-    url = f"{dbf}/z/{train_type}%20{train_ID}/{station}.json"
-    logger(f"Fetche {url}")
-    try:
-        response = requests.get(url)
-        response.raise_for_status()
-        data = response.json()
-    except requests.RequestException as e:
-        logger(f"ReqestException Fehler: {e}")
-        return None
-
-    dep = data.get("departure", [])
-    if not dep:
-        logger("Kein departure Feld gefunden", "error")
-        return None
-    
-    route_post = dep.get("route_post_diff")
-    if not route_post:
-        logger("Kein route_post_diff Feld gefunden", "error")
-        return None
-    
-    arrival_iso = route_post[-1].get("sched_arr")
-    if not arrival_iso:
-        logger("Keine Ankunftszeit gefunden", "error")
-        return None
-    
-    return {
-        "arrival": arrival_iso,
-        "operators": dep.get("operators"),
-    }
-
-def format_via_list(via: list[str]):
-    if not via:
-        return ""
-    if len(via) == 1:
-        return via[0]
-    return ", ".join(via[:-1]) + " und " + via[-1] 
-
-def resolve_operator(operators):
-    if not operators:
-        return None
-    for op in operators:
-        if op in operators_module.OPERATOR_ALIASES or op in operators_module.OPERATORS:
-            return op
-        
-    return operators[0]
-
-def operator_metadata(operator):
-    _reload_operators_if_changed()
-
-    if not operator:
-        return operators_module.OPERATORS["fallback"]
-
-    if operator in operators_module.OPERATOR_ALIASES:
-        return operators_module.OPERATOR_ALIASES[operator]
-
-    return operators_module.OPERATORS.get(operator, operators_module.OPERATORS["fallback"])
-
-def get_channel_formatting(train_type):
-    formatting = config.get("formatting", "")
-    train_emoji = ""
-
-    if config.get("emojis", True):
-        train_emoji = emojis.train_types.get(train_type)
-        if train_emoji is None:
-            train_emoji = emojis.emoji_list.get("Fallback", "")
-    
-    return f"{train_emoji}{formatting}"
-
-def logger(msg, log_type="info"):
     status = log_type.upper()
     current_time = datetime.now().strftime('%X')
     print(f"{current_time}: {status}: {msg}")