initial jeb

fetch_tickets.py

@@ -0,0 +1,313 @@
+#!/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", "<PUT_TELEGRAM_BOT_TOKEN_HERE>")
+TELEGRAM_CHAT_ID   = os.environ.get("TELEGRAM_CHAT_ID", "<PUT_TELEGRAM_CHAT_ID_HERE>")
+
+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()

toqla_auth.py

@@ -0,0 +1,311 @@
+#!/usr/bin/env python3
+"""
+Shared authentication helpers for the Toqla/Sodexo API.
+
+Handles OAuth2 PKCE (Proof Key for Code Exchange) with Auth0 session cookies
+to silently obtain an access token — no browser interaction needed.
+
+Provide Auth0 session cookies via:
+  --cookies "did=...; auth0=..."
+or COOKIES env var, or cookies.txt file.
+
+To get your cookies:
+  1. Open https://app.toqla.fr and log in.
+  2. DevTools (F12) > Network > any request to myid.sodexo.com
+  3. Copy the "Cookie" header value (did=...; auth0=...)
+
+The refresh_token obtained from Auth0 is saved to refresh_token.txt and reused
+on subsequent runs (no cookies needed) until it expires. This module is shared
+by fetch_menus.py and fetch_tickets.py so both scripts reuse the same cached
+refresh token / cookies file.
+"""
+
+import base64
+import hashlib
+import logging
+import os
+import re
+import secrets
+import sys
+from pathlib import Path
+from urllib.parse import urlparse, parse_qs, unquote
+
+import requests
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+BASE_URL           = "https://zeus.helium.sodexo.com"
+API_KEY            = "94770481-bd9c-4013-96c9-9b85d0759ee2"
+AUTH0_DOMAIN       = "https://myid.sodexo.com"
+AUTH0_CLIENT_ID    = "cnEsGucenOXKbw8XBJpKhitMB0JUiojz"
+AUTH0_REDIRECT_URI  = "https://app.toqla.fr/oidc/callback"
+AUTH0_AUDIENCE     = "https://api.pluxee.app/fr/consumer_prd/consumer-bff/api/"
+AUTH0_SCOPE        = (
+    "openid offline_access profile email phone "
+    "https://api.pluxee.app/fr/consumer_prd/consumer-bff/api/scopes/consumer:bff"
+)
+REFRESH_TOKEN_FILE  = "refresh_token.txt"
+COOKIES_FILE        = "cookies.txt"
+
+log = logging.getLogger(__name__)
+
+_COMMON_HEADERS = {
+    "accept": "*/*",
+    "content-type": "application/x-www-form-urlencoded",
+    "origin": "https://app.toqla.fr",
+    "referer": "https://app.toqla.fr/",
+    "auth0-client": "eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0=",
+    "user-agent": (
+        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36"
+    ),
+}
+
+
+# ---------------------------------------------------------------------------
+# Cookie handling
+# ---------------------------------------------------------------------------
+def parse_cookie_string(cookie_str: str) -> dict:
+    """Parse a cookie header string into a dict {name: value}."""
+    cookies = {}
+    for part in cookie_str.split(";"):
+        part = part.strip()
+        if "=" in part:
+            k, v = part.split("=", 1)
+            cookies[k.strip()] = v.strip()
+    return cookies
+
+
+def load_cookies(args_cookies: str | None, cookies_file: str = COOKIES_FILE) -> dict:
+    """Load cookies from args, env var, or a cookies file."""
+    # Priority: --cookies arg > COOKIES env > cookies.txt file
+    raw = args_cookies or os.environ.get("COOKIES")
+    if raw:
+        return parse_cookie_string(raw)
+
+    try:
+        raw = Path(cookies_file).read_text(encoding="utf-8").strip()
+        if raw:
+            log.info("Loaded cookies from %s.", cookies_file)
+            return parse_cookie_string(raw)
+    except FileNotFoundError:
+        pass
+
+    return {}
+
+
+# ---------------------------------------------------------------------------
+# PKCE
+# ---------------------------------------------------------------------------
+def generate_pkce_pair() -> tuple[str, str]:
+    """Generate a PKCE code_verifier and code_challenge (S256)."""
+    verifier = secrets.token_urlsafe(64)[:128]
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+    return verifier, challenge
+
+
+# ---------------------------------------------------------------------------
+# Token acquisition
+# ---------------------------------------------------------------------------
+def silent_auth_with_cookies(cookies: dict) -> str:
+    """Use Auth0 session cookies + PKCE to silently obtain an access token.
+
+    Flow (same as the browser does with prompt=none):
+      1. Generate PKCE code_verifier + code_challenge
+      2. GET /authorize with cookies + PKCE (response_mode=web_message)
+      3. Extract authorization code from the HTML response
+      4. POST /oauth/token with grant_type=authorization_code + code_verifier
+
+    Returns the access_token string.
+    Writes the new refresh token back to refresh_token.txt.
+    """
+    if not cookies:
+        log.error(
+            "No Auth0 session cookies available. Provide them via "
+            "--cookies, COOKIES env var, or cookies.txt file."
+        )
+        sys.exit(1)
+
+    verifier, challenge = generate_pkce_pair()
+    state = secrets.token_urlsafe(16)
+    nonce = secrets.token_urlsafe(16)
+
+    # Step 1: GET /authorize with cookies (silent, prompt=none)
+    auth_url = (
+        f"{AUTH0_DOMAIN}/authorize"
+        f"?client_id={AUTH0_CLIENT_ID}"
+        f"&scope={AUTH0_SCOPE.replace(' ', '+')}"
+        f"&audience={AUTH0_AUDIENCE}"
+        f"&redirect_uri={AUTH0_REDIRECT_URI}"
+        f"&response_type=code"
+        f"&prompt=none"
+        f"&response_mode=web_message"
+        f"&state={state}"
+        f"&nonce={nonce}"
+        f"&code_challenge={challenge}"
+        f"&code_challenge_method=S256"
+        f"&auth0Client=eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0%3D"
+    )
+
+    log.info("Silent authorization via Auth0 with session cookies…")
+    resp = requests.get(
+        auth_url,
+        headers={
+            "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "accept-language": "en-GB,en;q=0.9,en-US;q=0.8,fr;q=0.7",
+            "origin": "https://app.toqla.fr",
+            "referer": "https://app.toqla.fr/",
+            "auth0-client": "eyJuYW1lIjoiYXV0aDAtcmVhY3QiLCJ2ZXJzaW9uIjoiMi4xNi4yIn0=",
+            "user-agent": _COMMON_HEADERS["user-agent"],
+        },
+        cookies=cookies,
+        timeout=30,
+        allow_redirects=True,
+    )
+    resp.raise_for_status()
+
+    # Step 2: Extract the authorization code from the HTML response
+    # Auth0 returns an HTML page with the code in a <script> tag:
+    #   window.parent.postMessage({type: "authorization_response", response: {code: "...", ...}}, "*")
+    # or in an <input name="code" value="...">
+    html = resp.text
+    code = None
+
+    # Try to find code in postMessage JSON
+    m = re.search(r'"code"\s*:\s*"([^"]+)"', html)
+    if m:
+        code = m.group(1)
+
+    # Fallback: look for input name="code"
+    if not code:
+        m = re.search(r'<input[^>]*name="code"[^>]*value="([^"]+)"', html)
+        if m:
+            code = m.group(1)
+
+    # Fallback: look for code in URL fragment (response_mode=web_message uses hash)
+    if not code and "#" in resp.url:
+        fragment = urlparse(resp.url).fragment
+        params = parse_qs(fragment)
+        if "code" in params:
+            code = params["code"][0]
+
+    if not code:
+        # Maybe Auth0 returned an error
+        error_m = re.search(r'"error"\s*:\s*"([^"]+)"', html)
+        error_desc_m = re.search(r'"error_description"\s*:\s*"([^"]+)"', html)
+        if error_m:
+            error = error_m.group(1)
+            desc = error_desc_m.group(1) if error_desc_m else ""
+            log.error("Auth0 error: %s — %s", error, unquote(desc.replace("+", " ")))
+        log.error(
+            "Could not extract authorization code from Auth0 response. "
+            "Your cookies may have expired. Response length: %d chars.",
+            len(html),
+        )
+        sys.exit(1)
+
+    log.info("Authorization code obtained. Exchanging for tokens…")
+
+    # Step 3: POST /oauth/token with authorization_code + code_verifier
+    token_resp = requests.post(
+        f"{AUTH0_DOMAIN}/oauth/token",
+        headers=_COMMON_HEADERS,
+        data={
+            "client_id": AUTH0_CLIENT_ID,
+            "redirect_uri": AUTH0_REDIRECT_URI,
+            "response_type": "code",
+            "code_verifier": verifier,
+            "code": code,
+            "grant_type": "authorization_code",
+        },
+        timeout=30,
+    )
+    token_resp.raise_for_status()
+    body = token_resp.json()
+
+    access_token = body.get("access_token")
+    new_refresh  = body.get("refresh_token")
+    if not access_token:
+        log.error("No access_token in Auth0 response: %s", body)
+        sys.exit(1)
+
+    if new_refresh:
+        Path(REFRESH_TOKEN_FILE).write_text(new_refresh + "\n", encoding="utf-8")
+        log.info("New refresh token saved to %s.", REFRESH_TOKEN_FILE)
+
+    return access_token
+
+
+def refresh_access_token(cookies: dict | None = None) -> str:
+    """Get a fresh access token.
+
+    Strategy:
+      1. Try refresh_token.txt (no cookies needed).
+      2. If it fails or there's no refresh_token, use PKCE + Auth0 session cookies.
+      3. If no cookies either, fail with clear instructions.
+
+    Returns the access_token string on success.
+    Writes the new refresh token back to refresh_token.txt.
+    """
+    # Try refresh_token first
+    try:
+        rt = Path(REFRESH_TOKEN_FILE).read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        rt = ""
+
+    if rt:
+        log.info("Trying refresh token from %s…", REFRESH_TOKEN_FILE)
+        resp = requests.post(
+            f"{AUTH0_DOMAIN}/oauth/token",
+            headers=_COMMON_HEADERS,
+            data={
+                "client_id": AUTH0_CLIENT_ID,
+                "redirect_uri": AUTH0_REDIRECT_URI,
+                "response_type": "code",
+                "grant_type": "refresh_token",
+                "refresh_token": rt,
+            },
+            timeout=30,
+        )
+
+        if resp.status_code == 200:
+            body = resp.json()
+            access_token = body.get("access_token")
+            new_refresh = body.get("refresh_token")
+            if access_token:
+                if new_refresh:
+                    Path(REFRESH_TOKEN_FILE).write_text(new_refresh + "\n", encoding="utf-8")
+                    log.info("New refresh token saved to %s.", REFRESH_TOKEN_FILE)
+                return access_token
+
+        log.warning("Refresh token failed (HTTP %d) — trying cookie-based auth.", resp.status_code)
+
+    # Fallback: PKCE with cookies
+    if cookies:
+        return silent_auth_with_cookies(cookies)
+
+    log.error(
+        "No valid refresh token and no cookies available.\n"
+        "Provide Auth0 session cookies via --cookies, COOKIES env var, or cookies.txt.\n"
+        "To get cookies: open https://app.toqla.fr, DevTools > Network > copy Cookie header."
+    )
+    sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Session
+# ---------------------------------------------------------------------------
+def build_session(token: str) -> requests.Session:
+    """Build a requests.Session pre-configured with the Toqla API auth headers."""
+    session = requests.Session()
+    session.headers.update({
+        "Authorization": f"Bearer {token}",
+        "x-api-key": API_KEY,
+        "Accept": "application/json",
+        "Origin": "https://app.toqla.fr",
+        "Referer": "https://app.toqla.fr/",
+        "User-Agent": _COMMON_HEADERS["user-agent"],
+    })
+    return session