testbot-healthcheck/bot.py

# TestBot mit Healthcheck
# =======================
# Einfacher Telegram Bot mit Coolify-kompatiblem Healthcheck
# 
# Features:
# - Antwortet auf "hallo" im Privatchat
# - HTTP Healthcheck auf Port 8080
# - Bereit für Coolify Deployment

import asyncio
import logging
import threading
from aiohttp import web
from aiogram import Bot, Dispatcher, types
from aiogram.enums import ParseMode
from aiogram.filters import Command
from aiogram.client.default import DefaultBotProperties

# ═══════════════════════════════════════════════════════════════════
# KONFIGURATION
# ═══════════════════════════════════════════════════════════════════

# Bot Token (über Umgebungsvariable oder hier setzen)
import os
BOT_TOKEN = os.getenv("BOT_TOKEN", "HIER_TOKEN_EINTRAGEN")

# Healthcheck Port
HEALTH_PORT = int(os.getenv("HEALTH_PORT", "8080"))

# ═══════════════════════════════════════════════════════════════════
# LOGGING
# ═══════════════════════════════════════════════════════════════════

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# ═══════════════════════════════════════════════════════════════════
# HEALTHCHECK SERVER
# ═══════════════════════════════════════════════════════════════════

# Globaler Status für Healthcheck
bot_status = {
    "running": False,
    "connected": False,
    "started_at": None
}

async def health_handler(request):
    """HTTP Handler für /health Endpoint"""
    if bot_status["running"] and bot_status["connected"]:
        return web.Response(
            text=f"OK - Bot running, connected: {bot_status['connected']}",
            status=200
        )
    elif bot_status["running"]:
        return web.Response(
            text="DEGRADED - Bot running but not connected",
            status=200  # 200 damit Coolify nicht restartet, nur loggt
        )
    else:
        return web.Response(
            text="ERROR - Bot not running",
            status=503
        )

async def start_health_server():
    """Startet den Healthcheck HTTP Server"""
    app = web.Application()
    app.router.add_get('/health', health_handler)
    app.router.add_get('/', health_handler)  # Fallback für Root
    
    runner = web.AppRunner(app)
    await runner.setup()
    
    site = web.TCPSite(runner, '0.0.0.0', HEALTH_PORT)
    await site.start()
    
    logger.info(f"✅ Healthcheck Server läuft auf Port {HEALTH_PORT}")
    logger.info(f"   → http://localhost:{HEALTH_PORT}/health")
    
    # Endlos laufen lassen
    while True:
        await asyncio.sleep(3600)

def run_health_server_sync():
    """Synchrone Wrapper für Thread-Start"""
    asyncio.run(start_health_server())

# ═══════════════════════════════════════════════════════════════════
# BOT HANDLER
# ═══════════════════════════════════════════════════════════════════

# Bot & Dispatcher erstellen
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    """Handler für /start"""
    await message.answer(
        "👋 <b>Hallo!</b>\n\n"
        "Ich bin der <b>TestBot mit Healthcheck</b>.\n"
        "Schreib mir einfach <code>hallo</code>!\n\n"
        "✅ Healthcheck läuft auf Port 8080"
    )

@dp.message(Command("help"))
async def cmd_help(message: types.Message):
    """Handler für /help"""
    await message.answer(
        "❓ <b>Hilfe</b>\n\n"
        "<b>Befehle:</b>\n"
        "• /start - Startet den Bot\n"
        "• /help - Zeigt diese Hilfe\n"
        "• /status - Bot-Status\n\n"
        "<b>Schreib einfach:</b>\n"
        "• hallo - Ich antworte!"
    )

@dp.message(Command("status"))
async def cmd_status(message: types.Message):
    """Zeigt Bot-Status"""
    status_text = (
        "📊 <b>Bot Status</b>\n\n"
        f"🟢 Läuft: {'Ja' if bot_status['running'] else 'Nein'}\n"
        f"🔗 Verbunden: {'Ja' if bot_status['connected'] else 'Nein'}\n"
        f"⏱️ Gestartet: {bot_status['started_at'] or 'Unbekannt'}\n"
        f"🆔 Bot ID: {bot.bot.id}\n"
        f"👤 Username: @{bot.bot.username}"
    )
    await message.answer(status_text)

@dp.message(lambda message: message.text and "hallo" in message.text.lower())
async def on_hallo(message: types.Message):
    """Antwortet auf 'hallo' (egal welche Groß-/Kleinschreibung)"""
    await message.answer(
        f"👋 Hallo {message.from_user.first_name}!\n\n"
        f"Ich habe deine Nachricht erhalten: <i>{message.text}</i>\n\n"
        f"✅ Bot läuft einwandfrei!"
    )

@dp.message()
async def on_any_message(message: types.Message):
    """Fallback für alle anderen Nachrichten"""
    if message.text and not message.text.startswith('/'):
        await message.answer(
            f"Du hast geschrieben: <i>{message.text}</i>\n\n"
            f"Versuch mal <code>hallo</code> zu schreiben! 👋"
        )

# ═══════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════

async def main():
    """Haupteinstiegspunkt"""
    logger.info("🚀 Starte TestBot mit Healthcheck...")
    
    # Status initialisieren
    bot_status["started_at"] = asyncio.get_event_loop().time()
    bot_status["running"] = True
    
    # Healthcheck Server in separatem Thread starten
    health_thread = threading.Thread(target=run_health_server_sync, daemon=True)
    health_thread.start()
    logger.info("🔄 Healthcheck Thread gestartet")
    
    # Kurze Pause damit Server hochkommt
    await asyncio.sleep(2)
    
    # Bot starten
    try:
        logger.info("🤖 Starte Bot...")
        await bot.delete_webhook(drop_pending_updates=True)
        
        # Verbindung testen
        me = await bot.get_me()
        logger.info(f"✅ Bot verbunden: @{me.username} (ID: {me.id})")
        bot_status["connected"] = True
        
        # Polling starten
        await dp.start_polling(bot)
        
    except Exception as e:
        logger.error(f"❌ Fehler beim Bot-Start: {e}")
        bot_status["connected"] = False
        raise
    finally:
        bot_status["running"] = False
        await bot.session.close()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("🛑 Bot gestoppt (Ctrl+C)")
    except Exception as e:
        logger.error(f"💥 Kritischer Fehler: {e}")