bot.py

"""
bot.py
Bot Telegram pour "Hilal Ai", un LLM Transformer Décodeur entraîné from
scratch au niveau caractère par LabZero / Suayki.

Prérequis :
    pip install python-telegram-bot==21.* torch

Fichiers attendus dans le même dossier :
    - model.pt      (poids entraînés, obligatoire)
    - vocab.json     (vocabulaire char-level, fortement recommandé)
    - model.py        (architecture, fourni séparément)

Variables d'environnement (NE JAMAIS mettre le token en clair dans le code) :
    TELEGRAM_BOT_TOKEN   -> token du bot obtenu via @BotFather
    MODEL_PATH           -> chemin vers model.pt (défaut: "model.pt")
    VOCAB_PATH           -> chemin vers vocab.json (défaut: "vocab.json")
    DEVICE                -> "cpu" ou "cuda" (défaut: auto-détection)
    MAX_NEW_TOKENS        -> nombre max de caractères générés (défaut: 200)
    TEMPERATURE           -> température de sampling (défaut: 0.7)

Lancement :
    export TELEGRAM_BOT_TOKEN="xxxx:yyyy"
    python bot.py
"""

import asyncio
import datetime
import logging
import os
import re
import sys

import torch
from telegram import Update
from telegram.constants import ChatAction
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

from model import BLOCK_SIZE, load_model

# ----------------------------------------------------------------------------
# Configuration / logging
# ----------------------------------------------------------------------------
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO,
)
logger = logging.getLogger("hilal_ai.bot")

TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
if not TELEGRAM_BOT_TOKEN:
    logger.critical(
        "TELEGRAM_BOT_TOKEN n'est pas défini. "
        "Sur HF Spaces : Settings -> Repository secrets -> ajoute TELEGRAM_BOT_TOKEN. "
        "En local : export TELEGRAM_BOT_TOKEN='ton_token' avant de démarrer."
    )
    raise RuntimeError("TELEGRAM_BOT_TOKEN manquant")

MODEL_PATH = os.environ.get("MODEL_PATH", "model.pt")
VOCAB_PATH = os.environ.get("VOCAB_PATH", "vocab.json")
DEVICE = os.environ.get("DEVICE") or ("cuda" if torch.cuda.is_available() else "cpu")
MAX_NEW_TOKENS = int(os.environ.get("MAX_NEW_TOKENS", "200"))
TEMPERATURE = float(os.environ.get("TEMPERATURE", "0.7"))

# URL de base optionnelle vers un proxy (ex: Cloudflare Worker) relayant
# les requêtes vers api.telegram.org. Nécessaire sur HF Spaces (tier
# gratuit) qui bloque les connexions sortantes directes vers Telegram.
# Exemple de valeur : "https://telegram-proxy.toncompte.workers.dev"
# Si non définie, le bot contacte api.telegram.org directement (cas normal
# en local ou sur un VPS classique).
TELEGRAM_API_BASE_URL = os.environ.get("TELEGRAM_API_BASE_URL", "").strip()

# Identité du bot (specs fournies)
BOT_NAME = "Hilal Ai"
BOT_SKILLS = "Discuter, Résumer, Traduire, Rédiger"
DEV_NAME = "LabZero"
DEV_LINK = "https://t.me"          # à remplacer par le lien réel du canal/contact
DEV_CHAT_ID = "-1004489211751"
PROGRAMMER_NAME = "Suayki"
PROGRAMMER_LINK = "https://t.me"   # à remplacer par le lien réel
PROGRAMMER_USER_ID = "6074566712"

IDENTITY_TEXT = (
    f"🤖 *{BOT_NAME}*\n\n"
    f"Compétences : {BOT_SKILLS}.\n\n"
    f"Développeur : [{DEV_NAME}]({DEV_LINK})\n"
    f"Programmeur : [{PROGRAMMER_NAME}]({PROGRAMMER_LINK})"
)

# Mots-clés déclenchant l'injection de la date/heure serveur dans le prompt
DATETIME_TRIGGERS = re.compile(
    r"\b(quelle heure|quelle date|date du jour|heure actuelle|"
    r"on est quel jour|quel jour sommes[- ]nous|date et heure)\b",
    re.IGNORECASE,
)

# Mots-clés déclenchant la réponse d'identité statique plutôt que la génération
IDENTITY_TRIGGERS = re.compile(
    r"\b(qui es[- ]tu|qui t'a (créé|developpé|développé|fait)|"
    r"ton (créateur|développeur|programmeur)|c'est qui ton (créateur|développeur))\b",
    re.IGNORECASE,
)

# ----------------------------------------------------------------------------
# Chargement du modèle (une seule fois, au démarrage du process)
# ----------------------------------------------------------------------------
logger.info("Chargement du modèle depuis %s sur %s ...", MODEL_PATH, DEVICE)
try:
    model, tokenizer = load_model(MODEL_PATH, VOCAB_PATH, device=DEVICE)
    logger.info("Modèle chargé avec succès. vocab_size=%d", tokenizer.vocab_size)
except FileNotFoundError:
    logger.critical(
        "Fichier de poids introuvable : %s. Place model.pt à la racine du repo Space "
        "ou définis la variable MODEL_PATH dans les Secrets/Variables du Space.",
        MODEL_PATH,
    )
    # Pas de sys.exit() ici : sur HF Spaces, ce module tourne dans un thread
    # secondaire lancé par app.py. Si on tue le process, le serveur web de
    # ping meurt aussi et HF considère tout le Space comme crashé en boucle.
    # On préfère garder le serveur de ping vivant pour diagnostiquer via
    # l'endpoint "/" et les logs du Space.
    raise
except Exception:
    logger.critical("Échec du chargement du modèle.", exc_info=True)
    raise
logger.info("Bot prêt à démarrer.")

# Verrou pour éviter les accès concurrents au modèle (un seul forward pass
# CPU/GPU à la fois, simple et sûr pour un usage VPS mono-instance).
_generation_lock = asyncio.Lock()


# ----------------------------------------------------------------------------
# Génération de texte
# ----------------------------------------------------------------------------
def _build_prompt(user_text: str) -> str:
    """
    Construit le prompt final envoyé au modèle. Injecte la date/heure serveur
    si l'utilisateur la demande explicitement.
    """
    if DATETIME_TRIGGERS.search(user_text):
        now = datetime.datetime.now().strftime("%d/%m/%Y %H:%M")
        user_text = f"[Date et heure actuelles : {now}]\n{user_text}"
    return user_text


def _safe_generate(prompt: str, max_new_tokens: int, temperature: float) -> str:
    """
    Génère une réponse de façon synchrone et sécurisée :
    - tronque le prompt pour ne garder que les BLOCK_SIZE derniers caractères
      (le modèle ne peut de toute façon pas voir plus loin que son contexte)
    - encode/décode avec le tokenizer char-level
    - retourne uniquement le texte nouvellement généré (sans répéter le prompt)
    """
    prompt = prompt[-BLOCK_SIZE:]  # le contexte ne peut pas dépasser BLOCK_SIZE de toute façon
    encoded = tokenizer.encode(prompt)
    if not encoded:
        encoded = tokenizer.encode(" ")

    idx = torch.tensor([encoded], dtype=torch.long, device=DEVICE)

    with torch.no_grad():
        out = model.generate(
            idx,
            max_new_tokens=max_new_tokens,
            temperature=temperature,
        )

    generated_ids = out[0].tolist()[len(encoded):]
    text = tokenizer.decode(generated_ids)
    return text.strip()


async def generate_reply(user_text: str) -> str:
    """Wrapper async : exécute la génération (bloquante, CPU/GPU-bound) dans
    un thread séparé pour ne pas bloquer la boucle d'événements du bot, et
    sérialise les accès au modèle via un verrou."""
    prompt = _build_prompt(user_text)
    async with _generation_lock:
        try:
            reply = await asyncio.to_thread(
                _safe_generate, prompt, MAX_NEW_TOKENS, TEMPERATURE
            )
        except Exception:
            logger.exception("Erreur pendant la génération de texte.")
            return "Désolé, une erreur interne est survenue pendant la génération. Réessaie."

    if not reply:
        reply = "Je n'ai pas pu générer de réponse claire, peux-tu reformuler ?"
    return reply


# ----------------------------------------------------------------------------
# Handlers Telegram
# ----------------------------------------------------------------------------
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        f"Salut ! Je suis {BOT_NAME} 🤖\n"
        f"Je peux discuter, résumer, traduire et rédiger.\n"
        f"Envoie-moi simplement un message, ou tape /about pour en savoir plus sur moi."
    )


async def cmd_about(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        IDENTITY_TEXT,
        parse_mode="Markdown",
        disable_web_page_preview=True,
    )


async def cmd_reset(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Optionnel : si tu ajoutes un historique de conversation par utilisateur,
    cette commande permet de le réinitialiser. Le modèle char-level de base
    n'a pas de mémoire de conversation par défaut (pas de chat template)."""
    context.user_data.clear()
    await update.message.reply_text("Contexte de conversation réinitialisé. ✅")


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user_text = update.message.text or ""
    user_text = user_text.strip()

    if not user_text:
        return

    if IDENTITY_TRIGGERS.search(user_text):
        await update.message.reply_text(
            IDENTITY_TEXT,
            parse_mode="Markdown",
            disable_web_page_preview=True,
        )
        return

    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id, action=ChatAction.TYPING
    )

    reply = await generate_reply(user_text)
    await update.message.reply_text(reply)


async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
    logger.error("Exception non gérée :", exc_info=context.error)
    if isinstance(update, Update) and update.effective_message:
        try:
            await update.effective_message.reply_text(
                "Une erreur est survenue, désolé. Réessaie dans un instant."
            )
        except Exception:
            pass


# ----------------------------------------------------------------------------
# Lancement de l'application
# ----------------------------------------------------------------------------
async def _run_bot_async() -> None:
    """Pilote manuellement le cycle de vie de l'Application Telegram.

    On évite Application.run_polling() ici : cette méthode est conçue pour
    être appelée depuis le thread principal (elle gère ses propres signaux
    et sa propre boucle asyncio en interne via asyncio.run()). Dans un
    thread secondaire comme celui utilisé sur HF Spaces, cela provoque
    l'erreur "There is no current event loop in thread ...". On reproduit
    donc manuellement ce que fait run_polling() en interne, mais à
    l'intérieur d'une coroutine que NOUS lançons via asyncio.run(), ce qui
    garantit une boucle d'événements propre et valide pour ce thread.
    """
    builder = Application.builder().token(TELEGRAM_BOT_TOKEN)

    # Timeouts généreux : le chemin réseau passe par un proxy Cloudflare
    # Worker (cold start variable) avant d'atteindre Telegram, ce qui ajoute
    # de la latence et de la variabilité par rapport à une connexion directe.
    # Les valeurs par défaut de la librairie (5s) sont trop courtes pour ce
    # chemin en cascade et provoquent des ConnectTimeout/TimedOut sporadiques.
    builder = builder.connect_timeout(20.0)
    builder = builder.read_timeout(20.0)
    builder = builder.write_timeout(20.0)
    builder = builder.pool_timeout(20.0)
    # get_updates utilise le long-polling : Telegram garde la connexion
    # ouverte jusqu'à 'timeout' secondes (par défaut côté updater) en
    # attendant un nouveau message. Il faut donc un read_timeout dédié plus
    # long que celui des autres appels, sinon chaque cycle de polling
    # timeout avant que Telegram n'ait eu la chance de répondre.
    builder = builder.get_updates_connect_timeout(20.0)
    builder = builder.get_updates_read_timeout(15.0)

    if TELEGRAM_API_BASE_URL:
        # HF Spaces (tier gratuit) bloque les connexions sortantes directes
        # vers api.telegram.org. On route donc tout le trafic via un proxy
        # (ex: Cloudflare Worker) qui relaie les requêtes à notre place.
        base = TELEGRAM_API_BASE_URL.strip().rstrip("/")
        # Tolère que le secret ait été saisi sans schéma (ex:
        # "mon-proxy.workers.dev" au lieu de "https://mon-proxy.workers.dev").
        # httpx exige un schéma explicite, sinon UnsupportedProtocol.
        if not base.startswith("http://") and not base.startswith("https://"):
            logger.warning(
                "TELEGRAM_API_BASE_URL ne contient pas de schéma (http/https) : "
                "'%s'. Ajout automatique de 'https://'.", base
            )
            base = "https://" + base
        # Format attendu par base_url : se termine par "/bot" (sans le
        # token, qui est ajouté automatiquement par la librairie après).
        builder = builder.base_url(f"{base}/bot")
        logger.info("Utilisation du proxy Telegram : %s", base)

    app = builder.build()

    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("about", cmd_about))
    app.add_handler(CommandHandler("reset", cmd_reset))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    app.add_error_handler(error_handler)

    logger.info("Hilal Ai démarre (polling)...")

    # Le Worker Cloudflare a une latence de cold start variable : on retente
    # quelques fois avant d'abandonner, au lieu de planter au premier échec.
    max_retries = 5
    for attempt in range(1, max_retries + 1):
        try:
            await app.initialize()
            logger.info("Application initialisée (tentative %d/%d).", attempt, max_retries)
            break
        except Exception:
            logger.warning(
                "Échec de app.initialize() (tentative %d/%d), nouvelle tentative dans 3s...",
                attempt, max_retries, exc_info=True,
            )
            if attempt == max_retries:
                raise
            await asyncio.sleep(3)

    await app.start()
    logger.info("Application démarrée (app.start() terminé).")
    await app.updater.start_polling(
        allowed_updates=Update.ALL_TYPES,
        drop_pending_updates=True,
        # timeout court (au lieu du défaut de 10s) : réduit la durée pendant
        # laquelle la connexion HTTP de long-polling reste ouverte via le
        # proxy Cloudflare Worker, ce qui limite le risque qu'une coupure
        # réseau intermédiaire interrompe la requête avant sa fin. Quitte à
        # faire des cycles plus fréquents, c'est plus fiable sur ce chemin
        # réseau en cascade (HF -> Cloudflare -> Telegram).
        timeout=3,
        poll_interval=1.0,
    )
    logger.info(
        "start_polling() terminé. updater.running=%s",
        app.updater.running,
    )

    # Empêche la coroutine de se terminer : start_polling() retourne
    # immédiatement après avoir lancé le polling en arrière-plan, il faut
    # donc bloquer ici indéfiniment pour garder le bot actif.
    stop_event = asyncio.Event()
    try:
        # Boucle de surveillance : log périodique pour confirmer que le
        # polling tourne toujours et n'a pas crashé silencieusement.
        while True:
            await asyncio.sleep(60)
            logger.info(
                "Heartbeat : bot toujours actif. updater.running=%s",
                app.updater.running,
            )
    finally:
        await app.updater.stop()
        await app.stop()
        await app.shutdown()


def main() -> None:
    # asyncio.run() crée une boucle d'événements neuve et valide pour CE
    # thread (qu'il soit principal ou secondaire), puis la ferme proprement
    # à la fin. C'est la façon recommandée de lancer du code async depuis
    # un thread secondaire en Python, contrairement à run_polling() qui
    # suppose implicitement le thread principal.
    asyncio.run(_run_bot_async())


def run_bot_blocking() -> None:
    """Point d'entrée appelé depuis un thread secondaire par app.py
    (le serveur web de ping reste dans le thread principal sur HF Spaces)."""
    main()


if __name__ == "__main__":
    main()