Cantoche/fetch_tickets.py

#!/usr/bin/env python3
"""
Fetches new tickets (receipts) from the Toqla/Sodexo platform and sends them
to a Telegram chat.

The list of receipts is retrieved from:
  GET /api/2.0/users/badges/<badge_uuid>/receipts?start-date=...&end-date=...
Each receipt's detail (a printable HTML ticket) is retrieved from:
  GET /api/2.0/users/badges/<badge_uuid>/receipts/<id>

The <badge_uuid> is specific to the user and is discovered automatically from:
  GET /api/2.0/users/badges

Only tickets newer than the last one processed are sent (tracked in
last_ticket.json, next to this script). If no reference exists yet (first
run), only the single most recent ticket is processed.

Authentication is shared with fetch_menus.py via toqla_auth.py (see that
module's docstring for how to provide Auth0 session cookies).

Usage:
  python3 fetch_tickets.py
  python3 fetch_tickets.py --cookies "did=...; auth0=..."
  python3 fetch_tickets.py --cookies-file cookies.txt
  python3 fetch_tickets.py --dry-run       # print instead of sending to Telegram
"""

import argparse
import html
import json
import logging
import os
import re
import sys
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import requests

import toqla_auth

# ---------------------------------------------------------------------------
# Config (edit as needed)
# ---------------------------------------------------------------------------
LOOKBACK_DAYS      = 30    # how many days back to look for tickets on each run
LOCAL_UTC_OFFSET   = timezone(timedelta(hours=2))  # Europe/Paris (adjust for DST if needed)
LAST_TICKET_FILE   = "last_ticket.json"

TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "5994009260:AAGPZC_fGpcQ0SjfTnyxgfiBkhbgN6xZRD4")
TELEGRAM_CHAT_ID   = os.environ.get("TELEGRAM_CHAT_ID", "1058786961")

TELEGRAM_API_URL   = "https://api.telegram.org/bot{token}/sendMessage"
TELEGRAM_MAX_LEN   = 4096

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Toqla API calls
# ---------------------------------------------------------------------------
def _list_from(data: object, *keys: str) -> list:
    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        for k in keys:
            if isinstance(data.get(k), list):
                return data[k]
    return []


def fetch_badge_uuid(session: requests.Session) -> tuple[str, str]:
    """Discover the badge UUID to use for the receipts endpoints.

    Picks the first badge with a non-empty `available_sites` list (the one
    actually tied to a real site), falling back to the first badge returned.
    Returns (badge_uuid, site_label) for logging.
    """
    r = session.get(f"{toqla_auth.BASE_URL}/api/2.0/users/badges", timeout=30)
    r.raise_for_status()
    badges = _list_from(r.json(), "items", "badges", "data", "results")
    if not badges:
        log.error("No badges returned — check your token.")
        sys.exit(1)

    chosen = next((b for b in badges if b.get("available_sites")), badges[0])
    checkout_system = chosen.get("checkout_system") or {}
    badge_uuid = checkout_system.get("id")
    if not badge_uuid:
        log.error("Could not find a badge UUID in the badges response.")
        sys.exit(1)

    sites = [s.get("name") for s in (chosen.get("available_sites") or [])]
    label = ", ".join(filter(None, sites)) or checkout_system.get("display_name") or badge_uuid
    return badge_uuid, label


def fetch_receipts(session: requests.Session, badge_uuid: str, days: int) -> list[dict]:
    """Return the receipts list for the last `days` days, sorted ascending by date."""
    end = datetime.now(LOCAL_UTC_OFFSET)
    start = end - timedelta(days=days)
    params = {
        "start-date": _iso_with_colon_offset(start),
        "end-date": _iso_with_colon_offset(end),
    }
    r = session.get(
        f"{toqla_auth.BASE_URL}/api/2.0/users/badges/{badge_uuid}/receipts",
        params=params, timeout=30,
    )
    r.raise_for_status()
    receipts = _list_from(r.json(), "items", "receipts", "data", "results")
    receipts.sort(key=lambda t: t.get("date") or "")
    return receipts


def fetch_receipt_detail(session: requests.Session, badge_uuid: str, ticket_id: str) -> str:
    """Fetch a single receipt and return its content as clean plain text."""
    r = session.get(
        f"{toqla_auth.BASE_URL}/api/2.0/users/badges/{badge_uuid}/receipts/{ticket_id}",
        timeout=30,
    )
    r.raise_for_status()
    body = r.json() or {}
    return _html_to_text(body.get("html") or "")


def _iso_with_colon_offset(dt: datetime) -> str:
    """Format a datetime as e.g. 2026-08-01T00:00:00+02:00 (colon in the offset)."""
    s = dt.strftime("%Y-%m-%dT%H:%M:%S%z")
    return s[:-2] + ":" + s[-2:]


# ---------------------------------------------------------------------------
# HTML → plain text
# ---------------------------------------------------------------------------
_TAG_RE = re.compile(r"<[^>]+>")


def _html_to_text(raw_html: str) -> str:
    """Turn the receipt's inline-styled HTML into readable plain text."""
    if not raw_html:
        return ""
    text = re.sub(r"(?i)<br\s*/?>", "\n", raw_html)
    text = re.sub(r"(?i)</pre\s*>", "\n", text)
    text = _TAG_RE.sub("", text)
    text = html.unescape(text)
    lines = [line.rstrip() for line in text.splitlines()]
    while lines and not lines[0].strip():
        lines.pop(0)
    while lines and not lines[-1].strip():
        lines.pop()
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Last-processed reference
# ---------------------------------------------------------------------------
def load_last_ticket(path: Path) -> dict | None:
    if not path.exists():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        if isinstance(data, dict) and data.get("last_date"):
            return data
    except (json.JSONDecodeError, OSError):
        pass
    return None


def save_last_ticket(path: Path, ticket: dict) -> None:
    path.write_text(
        json.dumps({"last_date": ticket["date"], "last_id": ticket["id"]}, indent=2),
        encoding="utf-8",
    )


# ---------------------------------------------------------------------------
# Telegram
# ---------------------------------------------------------------------------
def build_message(ticket: dict, text: str) -> str:
    kind = "🛒 Achat" if ticket.get("type") == "Debit" else "💳 Rechargement"
    amount = ticket.get("amount")
    amount_str = f"{amount:.2f} €" if isinstance(amount, (int, float)) else str(amount)
    date_str = ticket.get("date", "")
    header = f"{kind} — {date_str} — {amount_str}"
    escaped = html.escape(text) if text else "(contenu indisponible)"
    return f"{html.escape(header)}\n<pre>{escaped}</pre>"


def _chunk_message(message: str, limit: int = TELEGRAM_MAX_LEN) -> list[str]:
    if len(message) <= limit:
        return [message]
    chunks = []
    while message:
        chunks.append(message[:limit])
        message = message[limit:]
    return chunks


def send_telegram_message(message: str) -> None:
    if "<PUT_TELEGRAM" in TELEGRAM_BOT_TOKEN or "<PUT_TELEGRAM" in TELEGRAM_CHAT_ID:
        log.error(
            "TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID are not configured. "
            "Set them at the top of fetch_tickets.py or via the "
            "TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID environment variables."
        )
        sys.exit(1)

    url = TELEGRAM_API_URL.format(token=TELEGRAM_BOT_TOKEN)
    for chunk in _chunk_message(message):
        for attempt in range(3):
            try:
                r = requests.post(
                    url,
                    data={
                        "chat_id": TELEGRAM_CHAT_ID,
                        "text": chunk,
                        "parse_mode": "HTML",
                    },
                    timeout=30,
                )
                r.raise_for_status()
                break
            except requests.RequestException as exc:
                log.warning("Telegram send failed (attempt %d/3): %s", attempt + 1, exc)
                if attempt < 2:
                    time.sleep(2 ** attempt)
                else:
                    raise


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Fetch new Toqla/Sodexo tickets and send them to Telegram.",
    )
    p.add_argument("--cookies", "-c", metavar="COOKIE_STR",
                   help='Auth0 session cookies, e.g. "did=...; auth0=..."')
    p.add_argument("--cookies-file", metavar="FILE",
                   help=f"File containing cookies (default: {toqla_auth.COOKIES_FILE})")
    p.add_argument("--dry-run", action="store_true",
                   help="Print the messages instead of sending them to Telegram, "
                        "and don't update the last-ticket reference.")
    return p.parse_args()


def run() -> None:
    args = _parse_args()
    last_ticket_path = Path(LAST_TICKET_FILE)

    if not args.dry_run and ("<PUT_TELEGRAM" in TELEGRAM_BOT_TOKEN or "<PUT_TELEGRAM" in TELEGRAM_CHAT_ID):
        log.error(
            "TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID are not configured. "
            "Set them at the top of fetch_tickets.py or via the "
            "TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID environment variables."
        )
        sys.exit(1)

    cookies_raw = args.cookies or os.environ.get("COOKIES", "")
    if args.cookies_file:
        try:
            cookies_raw = Path(args.cookies_file).read_text(encoding="utf-8").strip()
        except FileNotFoundError:
            log.error("Cookies file not found: %s", args.cookies_file)
            sys.exit(1)
    cookies = toqla_auth.parse_cookie_string(cookies_raw) if cookies_raw else {}

    token = toqla_auth.refresh_access_token(cookies if cookies else None)
    session = toqla_auth.build_session(token)

    badge_uuid, site_label = fetch_badge_uuid(session)
    log.info("Using badge %s (%s)", badge_uuid, site_label)

    receipts = fetch_receipts(session, badge_uuid, LOOKBACK_DAYS)
    log.info("Found %d receipt(s) in the last %d day(s).", len(receipts), LOOKBACK_DAYS)

    last_ticket = load_last_ticket(last_ticket_path)

    if last_ticket is None:
        log.info("No stored reference — processing only the most recent ticket.")
        candidates = [receipts[-1]] if receipts else []
    else:
        candidates = [r for r in receipts if (r.get("date") or "") > last_ticket["last_date"]]

    if not candidates:
        log.info("Nothing new to send.")
        return

    log.info("%d new ticket(s) to send.", len(candidates))

    for ticket in candidates:
        ticket_id = ticket.get("id")
        text = fetch_receipt_detail(session, badge_uuid, ticket_id)
        message = build_message(ticket, text)

        if args.dry_run:
            log.info("[dry-run] Would send ticket %s:\n%s", ticket_id, message)
        else:
            send_telegram_message(message)
            log.info("Sent ticket %s (%s, %s).", ticket_id, ticket.get("type"), ticket.get("date"))
            save_last_ticket(last_ticket_path, ticket)
        time.sleep(0.3)


if __name__ == "__main__":
    run()