initial commit

.gitignore

@@ -0,0 +1,3 @@
+.venv/
+config.json
+subscribers.json

bot.py

@@ -0,0 +1,362 @@
+import asyncio
+import json
+import os
+from pathlib import Path
+
+import httpx
+import uvicorn
+
+from fastapi import FastAPI, HTTPException, Header
+from pydantic import BaseModel
+
+from telegram import (
+    InlineKeyboardButton,
+    InlineKeyboardMarkup,
+    Update
+)
+
+from telegram.ext import (
+    Application,
+    CommandHandler,
+    CallbackQueryHandler,
+    ContextTypes
+)
+
+CONFIG_FILE = "config.json"
+SUBSCRIBERS_FILE = "subscribers.json"
+
+
+def load_config():
+    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+config = load_config()
+
+TELEGRAM_TOKEN = config["telegram_token"]
+RENNPLAN_API = config["rennplan_api"]
+PUSH_API_KEY = config["push_api_key"]
+
+
+def load_subscribers():
+    if not Path(SUBSCRIBERS_FILE).exists():
+        return []
+
+    with open(SUBSCRIBERS_FILE, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def save_subscribers(subscribers):
+    with open(SUBSCRIBERS_FILE, "w", encoding="utf-8") as f:
+        json.dump(subscribers, f, indent=2)
+
+
+def add_subscriber(chat_id):
+    subscribers = load_subscribers()
+
+    if chat_id not in subscribers:
+        subscribers.append(chat_id)
+        save_subscribers(subscribers)
+
+
+def remove_subscriber(chat_id):
+    subscribers = load_subscribers()
+
+    if chat_id in subscribers:
+        subscribers.remove(chat_id)
+        save_subscribers(subscribers)
+
+
+async def lauf_selected(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    query = update.callback_query
+
+    await query.answer()
+
+    lauf_name = query.data.split(":")[1]
+
+    data = await fetch_rennplan()
+
+    lauf = data["rennplan"][lauf_name]
+
+    keyboard = []
+
+    for race_id in sorted(
+        lauf.keys(),
+        key=lambda x: int(x)
+    ):
+        race = lauf[race_id]
+
+        keyboard.append([
+            InlineKeyboardButton(
+                f"{race_id} - {race['name']} ({race['zeit']})",
+                callback_data=f"race:{lauf_name}:{race_id}"
+            )
+        ])
+
+    keyboard.append([
+        InlineKeyboardButton(
+            "⬅ Zurück",
+            callback_data="back_laeufe"
+        )
+    ])
+
+    await query.edit_message_text(
+        text=f"🏁 Rennen in {lauf_name.capitalize()}:",
+        reply_markup=InlineKeyboardMarkup(keyboard)
+    )
+
+async def race_selected(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    query = update.callback_query
+
+    await query.answer()
+
+    _, lauf_name, race_id = query.data.split(":")
+
+    data = await fetch_rennplan()
+
+    race = data["rennplan"][lauf_name][race_id]
+
+    keyboard = [[
+        InlineKeyboardButton(
+            "⬅ Zurück",
+            callback_data=f"lauf:{lauf_name}"
+        )
+    ]]
+
+    await query.edit_message_text(
+        text=format_race(race),
+        parse_mode="HTML",
+        reply_markup=InlineKeyboardMarkup(keyboard)
+    )
+async def back_to_laeufe(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    query = update.callback_query
+
+    await query.answer()
+
+    data = await fetch_rennplan()
+
+    keyboard = []
+
+    for lauf_name in sorted(
+        data["rennplan"].keys(),
+        key=lambda x: int(x.replace("lauf", ""))
+    ):
+        keyboard.append([
+            InlineKeyboardButton(
+                lauf_name.capitalize(),
+                callback_data=f"lauf:{lauf_name}"
+            )
+        ])
+
+    await query.edit_message_text(
+        text="🏁 Bitte einen Lauf auswählen:",
+        reply_markup=InlineKeyboardMarkup(keyboard)
+    )
+
+def get_latest_race(data):
+    rennplan = data["rennplan"]
+
+    latest_lauf_key = sorted(
+        rennplan.keys(),
+        key=lambda x: int(x.replace("lauf", ""))
+    )[-1]
+
+    lauf = rennplan[latest_lauf_key]
+
+    latest_race_key = sorted(
+        lauf.keys(),
+        key=lambda x: int(x)
+    )[-1]
+
+    return lauf[latest_race_key]
+
+
+def format_race(race):
+    return (
+        f"🏁 <b>{race['name']}</b>\n"
+        f"📋 {race['art']}\n"
+        f"🕒 {race['zeit']}\n\n"
+        f"❶ <b>Bahn 1</b>\n{race['bahn1']}\n\n"
+        f"❷ <b>Bahn 2</b>\n{race['bahn2']}\n\n"
+        f"❸ <b>Bahn 3</b>\n{race['bahn3']}"
+    )
+
+async def fetch_rennplan():
+    async with httpx.AsyncClient(timeout=30) as client:
+        response = await client.get(RENNPLAN_API)
+
+        response.raise_for_status()
+
+        return response.json()
+
+
+async def fetch_latest_race():
+    async with httpx.AsyncClient(timeout=30) as client:
+        response = await client.get(RENNPLAN_API)
+
+        response.raise_for_status()
+
+        data = response.json()
+
+    return get_latest_race(data)
+
+
+async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    add_subscriber(update.effective_chat.id)
+
+    await update.message.reply_text(
+        "Moin! Du bekommst ab sofort Push Benachrichtigungen über Telegram. Vielen Dank!"
+    )
+
+
+async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    remove_subscriber(update.effective_chat.id)
+
+    await update.message.reply_text(
+        "Auf wiedersehen!"
+    )
+
+
+async def rennplan_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    try:
+        data = await fetch_rennplan()
+
+        rennplan = data["rennplan"]
+
+        keyboard = []
+
+        for lauf_name in sorted(
+            rennplan.keys(),
+            key=lambda x: int(x.replace("lauf", ""))
+        ):
+            keyboard.append([
+                InlineKeyboardButton(
+                    lauf_name.capitalize(),
+                    callback_data=f"lauf:{lauf_name}"
+                )
+            ])
+
+        await update.message.reply_text(
+            "🏁 Bitte wähle einen Lauf aus:",
+            reply_markup=InlineKeyboardMarkup(keyboard)
+        )
+
+    except Exception as e:
+        await update.message.reply_text(
+            f"Fehler beim Laden des Rennplans:\n{e}"
+        )
+
+async def app_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    keyboard = [
+        [
+            InlineKeyboardButton(
+                text="Apps herunterladen",
+                url="https://apps.sport-am-tankumsee.de?source=telegram"
+            )
+        ]
+    ]
+
+    await update.message.reply_text(
+        "Lade dir hier unsere nativen Apps für iOS & Android herunter:",
+        reply_markup=InlineKeyboardMarkup(keyboard)
+    )
+
+
+telegram_app = Application.builder().token(
+    TELEGRAM_TOKEN
+).build()
+
+
+telegram_app.add_handler(CommandHandler("start", start_command))
+telegram_app.add_handler(CommandHandler("stop", stop_command))
+telegram_app.add_handler(CommandHandler("rennplan", rennplan_command))
+telegram_app.add_handler(CommandHandler("app", app_command))
+telegram_app.add_handler(
+    CallbackQueryHandler(
+        lauf_selected,
+        pattern=r"^lauf:"
+    )
+)
+
+telegram_app.add_handler(
+    CallbackQueryHandler(
+        race_selected,
+        pattern=r"^race:"
+    )
+)
+
+telegram_app.add_handler(
+    CallbackQueryHandler(
+        back_to_laeufe,
+        pattern=r"^back_laeufe$"
+    )
+)
+
+
+
+api = FastAPI()
+
+
+class PushMessage(BaseModel):
+    message: str
+
+
+@api.post("/push")
+async def push_message(
+    payload: PushMessage,
+    x_api_key: str = Header(None)
+):
+    if x_api_key != PUSH_API_KEY:
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid API Key"
+        )
+
+    subscribers = load_subscribers()
+
+    sent = 0
+
+    for chat_id in subscribers:
+        try:
+            await telegram_app.bot.send_message(
+                chat_id=chat_id,
+                text=payload.message,
+                parse_mode="HTML"
+            )
+            sent += 1
+        except Exception:
+            pass
+
+    return {
+        "success": True,
+        "sent": sent
+    }
+
+
+async def start_api():
+    server = uvicorn.Server(
+        uvicorn.Config(
+            app=api,
+            host=config["listen_host"],
+            port=config["listen_port"],
+            log_level="info"
+        )
+    )
+
+    await server.serve()
+
+
+async def main():
+    await telegram_app.initialize()
+    await telegram_app.start()
+
+    asyncio.create_task(start_api())
+
+    await telegram_app.updater.start_polling()
+
+    while True:
+        await asyncio.sleep(3600)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

config.example.json

@@ -0,0 +1,7 @@
+{
+  "telegram_token": "BOT_TOKEN_HIER",
+  "rennplan_api": "https://api.demo.regattatech.de/api/v1/rennplan/current",
+  "push_api_key": "mein-geheimer-api-key",
+  "listen_host": "0.0.0.0",
+  "listen_port": 8000
+}

requirements.txt

@@ -0,0 +1,4 @@
+python-telegram-bot==22.1
+fastapi==0.116.1
+uvicorn==0.35.0
+httpx==0.28.1