Un bot Telegram pour les séances de cinéma, en Python

Séances à Lyon, le 09/11

Moulin
Lumière Terreaux: 15:20 VFST, 17:55, 20:00
Pathe Vaise: 16:00, 18:55, 21:30
Comoedia: 16:25, 18:15 VFST, 20:35

Wild Horse Nine
Comoedia: 15:50 VOST, 18:20 VOST, 20:50 VOST
…

mkdir bot-seances && cd bot-seances
python -m venv .venv && source .venv/bin/activate
pip install "python-telegram-bot==22.8"

export TELEGRAM_TOKEN="123456789:AA…"
export API_CINEMA_KEY="ac_live_…"

"""Lire les séances du jour sur API Cinéma et les écrire en un message Telegram."""
import os
import time
from html import escape

import httpx

API = "https://api.api-cinema.com/v1"
CACHE_SECONDS = 15 * 60
_cache: dict[tuple, tuple[float, list[dict]]] = {}


async def fetch_showtimes(city: str | None = None, lat: float | None = None, lng: float | None = None,
                          version: str | None = None) -> list[dict]:
    """Les séances qui restent aujourd’hui dans une ville, ou à moins de 3 km d’un point (heure locale de chaque cinéma)."""
    params = {"include": "movie,cinema", "limit": "1000"}
    if city:
        params["city"] = city
    else:
        params["near"] = f"{lat:.3f},{lng:.3f}"  # trois décimales : environ 100 m, assez précis et plus discret
        params["radius_km"] = "3"
    if version:
        params["version"] = version

    key = tuple(sorted(params.items()))
    cached = _cache.get(key)
    if cached and time.monotonic() - cached[0] < CACHE_SECONDS:
        return cached[1]

    # sans date, l’API part de maintenant, triée par heure : lire les pages jusqu’à la fin du premier jour
    showtimes: list[dict] = []
    day = None
    async with httpx.AsyncClient(timeout=15, headers={"Authorization": f"Bearer {os.environ['API_CINEMA_KEY']}"}) as client:
        for _ in range(5):  # une grande ville a quelques milliers de séances par jour : cinq pages au plus
            res = await client.get(f"{API}/showtimes", params=params)
            res.raise_for_status()
            page = res.json()
            if not page["data"]:
                break
            day = day or page["data"][0]["startsAt"][:10]
            showtimes += [s for s in page["data"] if s["startsAt"].startswith(day)]
            offset = int(params.get("offset", 0)) + len(page["data"])
            if offset >= page["meta"]["total"] or not page["data"][-1]["startsAt"].startswith(day):
                break
            params["offset"] = str(offset)
    _cache[key] = (time.monotonic(), showtimes)
    return showtimes


def to_message(title: str, showtimes: list[dict], max_length: int = 4000) -> str:
    """Un bloc par film, le plus programmé d’abord, chaque cinéma sur une ligne avec ses horaires (HTML Telegram)."""
    films: dict[str, dict] = {}
    for s in showtimes:
        film = films.setdefault(s["movie"]["id"], {"title": s["movie"]["title"], "cinemas": {}, "count": 0})
        label = s["startsAt"][11:16] + ("" if s["version"] in ("VF", "VO") else f" {s['version']}")
        film["cinemas"].setdefault(s["cinema"]["name"], []).append(label)
        film["count"] += 1

    ordered = sorted(films.values(), key=lambda f: f["count"], reverse=True)
    text = f"<b>{escape(title)}</b>"
    for shown, film in enumerate(ordered):
        lines = [f"{escape(cinema)}: {', '.join(times)}" for cinema, times in film["cinemas"].items()]
        block = f"\n\n<b>{escape(film['title'])}</b>\n" + "\n".join(lines)
        # un message Telegram contient 4 096 caractères : s’arrêter avant, et dire combien de films manquent
        if len(text) + len(block) > max_length - 40:
            left = len(ordered) - shown
            text += f"\n\n… et {left} autre{'s' if left > 1 else ''} film{'s' if left > 1 else ''}"
            break
        text += block
    return text

"""Un bot Telegram qui répond avec les séances du jour : /seances <ville>, ou une position partagée."""
import logging
import os

import httpx
from telegram import KeyboardButton, ReplyKeyboardMarkup, Update
from telegram.constants import ChatAction, ParseMode
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters

from showtimes import fetch_showtimes, to_message

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logging.getLogger("httpx").setLevel(logging.WARNING)  # ses lignes INFO afficheraient le jeton du bot
VERSIONS = {"vost", "vo", "vf"}
KEYBOARD = ReplyKeyboardMarkup([[KeyboardButton("📍 Séances près de moi", request_location=True)]], resize_keyboard=True)


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        "Envoyez /seances suivi d’une ville (/seances Lyon, /seances Paris vost), "
        "ou partagez votre position pour les cinémas à moins de 3 km.",
        reply_markup=KEYBOARD,
    )


async def reply(update: Update, title: str, **query) -> None:
    await update.message.chat.send_action(ChatAction.TYPING)
    try:
        showtimes = await fetch_showtimes(**query)
    except httpx.HTTPError as error:  # un statut d’erreur (401 mauvaise clé, 429 quota épuisé) ou pas de réseau
        logging.warning("échec de la requête API Cinéma : %r", error)
        await update.message.reply_text("Les séances sont indisponibles pour le moment, réessayez plus tard.")
        return
    if not showtimes:
        await update.message.reply_text("Plus de séance aujourd’hui par ici.")
        return
    await update.message.reply_text(
        to_message(f"{title}, le {showtimes[0]['startsAt'][8:10]}/{showtimes[0]['startsAt'][5:7]}", showtimes),
        parse_mode=ParseMode.HTML,
    )


async def seances_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    words = list(context.args)
    version = words.pop().upper() if words and words[-1].lower() in VERSIONS else None
    city = " ".join(words)
    if not city:
        await update.message.reply_text("Quelle ville ? Par exemple : /seances Lyon")
        return
    await reply(update, f"Séances à {city}", city=city, version=version)


async def location(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    point = update.message.location
    await reply(update, "Séances à moins de 3 km", lat=point.latitude, lng=point.longitude)


def main() -> None:
    app = Application.builder().token(os.environ["TELEGRAM_TOKEN"]).build()
    app.add_handler(CommandHandler(["start", "help"], start))
    app.add_handler(CommandHandler("seances", seances_command))
    app.add_handler(MessageHandler(filters.LOCATION, location))
    app.run_polling()


if __name__ == "__main__":
    main()

python bot.py

seances - Les séances du jour dans une ville, ex. /seances Lyon
start - Comment utiliser le bot