Authentification

"""Obtenir un jeton ``client_credentials`` et lire les factures, page par page.

Get a ``client_credentials`` token and read invoices page by page.

Bibliothèque standard seule (Python 3.9+). Variables d'environnement :
``TAURACO_CLIENT_ID`` (``tc_live_…``), ``TAURACO_CLIENT_SECRET`` (``tcs_…``, affiché une
seule fois à la création de l'application), et facultativement ``TAURACO_TENANT``
(obligatoire si plusieurs entreprises ont autorisé l'application).

Les adresses par défaut sont les surfaces prévues sous ``api.tauraco.io`` ; elles ne
répondent qu'une fois l'API publique ouverte (voir la page « Versions et changelog »).
"""
from __future__ import annotations

import base64
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Callable, Dict, Iterator, Optional, Tuple

API_BASE = os.environ.get("TAURACO_API_BASE", "https://api.tauraco.io/public/v1")
TOKEN_URL = os.environ.get("TAURACO_TOKEN_URL", "https://api.tauraco.io/oauth/token")

#: Transport : (méthode, url, en-têtes, corps) -> (statut, en-têtes, corps). Remplaçable en test.
Transport = Callable[[str, str, Dict[str, str], Optional[bytes]], Tuple[int, Dict[str, str], bytes]]


class TauracoError(RuntimeError):
    """Erreur de l'API : corps ``application/problem+json`` (RFC 9457) ou erreur OAuth."""

    def __init__(self, status: int, body: dict):
        self.status = status
        self.code = body.get("code") or body.get("error") or "unknown"
        self.request_id = body.get("request_id")
        super().__init__(f"{status} {self.code}: {body.get('detail') or body.get('error_description') or ''}")


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Aucune redirection suivie : urllib renverrait sinon l'en-tête ``Authorization`` (jeton)
    à l'hôte de destination. L'API Tauraco ne redirige jamais ; un 3xx est rendu tel quel."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


_OPENER = urllib.request.build_opener(_NoRedirect)


def http_transport(method: str, url: str, headers: Dict[str, str],
                   body: Optional[bytes]) -> Tuple[int, Dict[str, str], bytes]:
    request = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with _OPENER.open(request, timeout=30) as response:  # noqa: S310 (URL maîtrisée)
            return response.status, dict(response.headers.items()), response.read()
    except urllib.error.HTTPError as err:
        return err.code, dict(err.headers.items()), err.read()


def get_token(client_id: str, client_secret: str, scopes: str, *, tenant: Optional[str] = None,
              resource: str = API_BASE, transport: Transport = http_transport) -> dict:
    """``POST /oauth/token`` (``grant_type=client_credentials``), authentification HTTP Basic."""
    form = {"grant_type": "client_credentials", "scope": scopes, "resource": resource}
    if tenant:
        form["tenant"] = tenant
    basic = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode("ascii")
    status, _headers, raw = transport("POST", TOKEN_URL, {
        "Authorization": f"Basic {basic}",
        "Content-Type": "application/x-www-form-urlencoded",
        "Accept": "application/json",
    }, urllib.parse.urlencode(form).encode())
    payload = json.loads(raw or b"{}")
    if status != 200:
        raise TauracoError(status, payload)
    return payload  # {"access_token", "token_type": "Bearer", "expires_in": 600, "scope"}


def iterate(path: str, token: str, params: Optional[Dict[str, str]] = None, *,
            transport: Transport = http_transport, max_retries: int = 3) -> Iterator[dict]:
    """Toutes les lignes d'une liste paginée par curseur (``data``, ``next_cursor``, ``has_more``)."""
    query = dict(params or {})
    retries = 0
    while True:
        url = f"{API_BASE}{path}?{urllib.parse.urlencode(query)}" if query else f"{API_BASE}{path}"
        status, headers, raw = transport("GET", url, {"Authorization": f"Bearer {token}",
                                                      "Accept": "application/json"}, None)
        if status == 429 and retries < max_retries:
            retries += 1
            time.sleep(int(headers.get("Retry-After") or headers.get("retry-after") or 1))
            continue
        payload = json.loads(raw or b"{}")
        if status != 200:
            raise TauracoError(status, payload)
        retries = 0
        yield from payload.get("data", [])
        if not payload.get("has_more") or not payload.get("next_cursor"):
            return
        query["cursor"] = payload["next_cursor"]


def main() -> None:
    token = get_token(os.environ["TAURACO_CLIENT_ID"], os.environ["TAURACO_CLIENT_SECRET"],
                      "invoices:read", tenant=os.environ.get("TAURACO_TENANT"))["access_token"]
    for invoice in iterate("/invoices", token, {"status": "PAID", "limit": "100"}):
        # Montants : chaînes décimales en unités majeures ("1180000" XOF), jamais des flottants.
        print(invoice["number"], invoice.get("customer_name"), invoice["total"], invoice["currency"])


if __name__ == "__main__":
    main()