"""Liseuse PDF — SDK Python 1.1.0 · https://liseuse-pdf.com/docs/sdk-serveur · MIT Python 3.8+, bibliothèque standard uniquement. Côté serveur : la clé API ne doit jamais atteindre le navigateur. from liseuse_pdf import LiseusePDF liseuse = LiseusePDF(os.environ["LISEUSE_CLE_API"]) lien = liseuse.creer_lien("liseuse-pdf.com:inklura-escales-n12", jours=1) # webhook (corps brut, en octets) : evenement = LiseusePDF.verifier_webhook(request.body, request.headers["Liseuse-Signature"], secret) """ import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request __version__ = "1.1.0" class LiseuseError(Exception): def __init__(self, status, code, message): super().__init__(message) self.status = status self.code = code class LiseusePDF: def __init__(self, cle_api, base="https://liseuse-pdf.com/api/gestion/v1", delai=10): if not (isinstance(cle_api, str) and cle_api.startswith("lsp_") and 24 <= len(cle_api) <= 68): raise ValueError("Clé API Liseuse PDF invalide (lsp_…)") self.cle_api = cle_api self.base = base.rstrip("/") self.delai = delai def _appel(self, methode, chemin, corps=None, idempotence=None): data = json.dumps(corps).encode() if corps is not None else None entetes = {"Idempotency-Key": idempotence} if idempotence else {} req = urllib.request.Request(self.base + chemin, data=data, method=methode, headers={**entetes, "Authorization": "Bearer " + self.cle_api, "Content-Type": "application/json", # un nom d'outil explicite : le pare-feu du service refuse l'agent par défaut de urllib "User-Agent": "liseuse-pdf-python/" + __version__, }) try: with urllib.request.urlopen(req, timeout=self.delai) as res: return json.loads(res.read().decode()) except urllib.error.HTTPError as e: try: err = json.loads(e.read().decode()).get("error", {}) except ValueError: err = {} raise LiseuseError(e.code, err.get("code", "http_%d" % e.code), err.get("message", "Erreur %d" % e.code)) from None except urllib.error.URLError as e: raise LiseuseError(0, "reseau", "Liseuse PDF injoignable : %s" % e.reason) from None def creer_lien(self, id, jours=None, page=None, reference=None): """Lien de lecture signé pour un abonné : {"url", "expire_le", "iframe"}. reference : votre identifiant d'abonné, pour révoquer ses liens plus tard.""" corps = {"id": id} for k, v in (("jours", jours), ("page", page), ("reference", reference)): if v is not None: corps[k] = v return self._appel("POST", "/liens", corps) def revoquer_liens(self, reference): """Désactive tous les liens créés avec cette référence d'abonné.""" return self._appel("POST", "/liens/revoquer", {"reference": reference}) def numero(self, id): """Un numéro par son identifiant.""" return self._appel("GET", "/numeros?id=" + urllib.parse.quote(id))["numero"] def publier(self, fichier, titre, couverture=None, **champs): """Publie un PDF (chemin) : envoi du fichier puis création du numéro. Champs : publication, nouvelle_publication, parution, acces, lien_achat, telechargement, impression, domaine, adresse, remplace. Permission « publication ».""" with open(fichier, "rb") as f: pdf = f.read() domaine = champs.get("domaine") sha = self._televerser(pdf, "pdf", domaine) if couverture: with open(couverture, "rb") as f: jpg = f.read() champs["couverture_sha256"] = self._televerser(jpg, "couverture", domaine) champs["couverture_taille"] = len(jpg) corps = {**champs, "titre": titre, "sha256": sha, "taille": len(pdf), "nom_fichier": fichier.rsplit("/", 1)[-1]} return self._appel("POST", "/numeros", corps, idempotence="publier-%s-%s" % (sha, champs.get("remplace") or titre)) def _televerser(self, octets, type_, domaine): sha = hashlib.sha256(octets).hexdigest() corps = {"type": type_, "sha256": sha, "taille": len(octets)} if domaine: corps["domaine"] = domaine t = self._appel("POST", "/televersements", corps) if not t.get("deja_present"): req = urllib.request.Request(t["url"], data=octets, method="PUT", headers=t["en_tetes"]) try: urllib.request.urlopen(req, timeout=600).close() except urllib.error.HTTPError as e: raise LiseuseError(e.code, "televersement", "Envoi du fichier refusé par le stockage (%d)" % e.code) from None return sha def modifier_numero(self, id, **champs): """Modifie un numéro : titre, acces, lien_achat, en_ligne, telechargement, impression, publication.""" return self._appel("PATCH", "/numeros?id=" + urllib.parse.quote(id), champs)["numero"] def publications(self, publication=None, limite=None): """Vos numéros en ligne, du plus récent au plus ancien.""" q = {k: v for k, v in {"publication": publication, "limite": limite}.items() if v} return self._appel("GET", "/publications" + ("?" + urllib.parse.urlencode(q) if q else ""))["publications"] def statistiques(self, id, depuis=None): """Lectures d'un numéro (30 derniers jours par défaut).""" q = {"id": id} if depuis: q["depuis"] = depuis return self._appel("GET", "/statistiques?" + urllib.parse.urlencode(q)) def compte(self): """Formule, statut, limites et utilisation du compte.""" return self._appel("GET", "/compte")["compte"] @staticmethod def verifier_webhook(corps_brut, entete, secret, tolerance=300): """Vérifie l'en-tête Liseuse-Signature et renvoie l'événement ; lève LiseuseError sinon.""" if isinstance(corps_brut, str): corps_brut = corps_brut.encode() t, v1 = "", [] for partie in (entete or "").split(","): k, _, v = partie.strip().partition("=") if k == "t": t = v elif k == "v1" and v: v1.append(v) if not t or not v1: raise LiseuseError(400, "signature_absente", "En-tête Liseuse-Signature absent ou incomplet") if abs(time.time() - int(t)) > tolerance: raise LiseuseError(400, "signature_expiree", "Signature trop ancienne") attendu = hmac.new(secret.encode(), t.encode() + b"." + corps_brut, hashlib.sha256).hexdigest() if not any(hmac.compare_digest(attendu, s) for s in v1): raise LiseuseError(400, "signature_invalide", "Signature invalide") return json.loads(corps_brut.decode())