import json from typing import Any, Dict, List, Optional import requests def _read_json(path: str, default): try: with open(path, "r", encoding="utf-8") as f: v = json.load(f) return v if v is not None else default except Exception: return default def twitch_client_id(config_path: str = "config/config.json") -> str: cfg = _read_json(config_path, default={}) if isinstance(cfg, dict): v = cfg.get("twitch_client_id") or cfg.get("twitch_clientId") or cfg.get("twitchClientId") if isinstance(v, str): return v.strip() return "" def _normalize_bearer(raw: str) -> str: raw = (raw or "").strip() if raw.lower().startswith("oauth:"): raw = raw.split(":", 1)[1].strip() return raw def bearer_token_for_pseudo(pseudo: str, users_path: str = "config/user.json") -> str: users = _read_json(users_path, default=[]) if not isinstance(users, list): users = [] wanted = (pseudo or "").strip().lstrip("@").lower() for u in users: if not isinstance(u, dict): continue p = (u.get("tw_acc_pseudo") or "").strip().lower() if p == wanted: tok = _normalize_bearer(u.get("tw_acc_token") or "") if not tok: raise ValueError("Token OAuth manquant") return tok raise ValueError("Utilisateur (pseudo) non trouvé") def _helix_headers(*, client_id: str, bearer: str) -> Dict[str, str]: if not client_id: raise ValueError("Client-ID Twitch manquant (config/config.json:twitch_client_id)") if not bearer: raise ValueError("Token OAuth manquant") return {"Client-ID": client_id, "Authorization": f"Bearer {bearer}"} def get_user_id_by_login(*, client_id: str, bearer: str, login: str, timeout_s: int = 10) -> str: login = (login or "").strip().lstrip("@") if not login: raise ValueError("Nom de chaîne requis") r = requests.get( "https://api.twitch.tv/helix/users", headers=_helix_headers(client_id=client_id, bearer=bearer), params={"login": login}, timeout=timeout_s, ) payload: Optional[Dict[str, Any]] try: payload = r.json() except Exception: payload = None if r.status_code >= 400: msg = "" if isinstance(payload, dict): msg = payload.get("message") or payload.get("error") or "" raise RuntimeError(f"Erreur Twitch /helix/users ({r.status_code}) {msg}".strip()) data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, list) or not data: raise ValueError("Chaîne Twitch introuvable (login)") uid = (data[0] or {}).get("id") if not uid: raise RuntimeError("Réponse Twitch invalide: id manquant") return str(uid) def create_clip( *, client_id: str, bearer: str, broadcaster_id: str, has_delay: bool = False, timeout_s: int = 15, ) -> Dict[str, str]: r = requests.post( "https://api.twitch.tv/helix/clips", headers=_helix_headers(client_id=client_id, bearer=bearer), params={"broadcaster_id": str(broadcaster_id), "has_delay": "true" if has_delay else "false"}, timeout=timeout_s, ) payload: Optional[Dict[str, Any]] try: payload = r.json() except Exception: payload = None if r.status_code >= 400: msg = "" if isinstance(payload, dict): msg = payload.get("message") or payload.get("error") or "" hint = "" if r.status_code in (401, 403): hint = " (scope requis: clips:edit)" raise RuntimeError(f"Erreur Twitch /helix/clips ({r.status_code}) {msg}{hint}".strip()) data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, list) or not data: raise RuntimeError("Réponse Twitch invalide: data manquant") clip = data[0] or {} clip_id = clip.get("id") edit_url = clip.get("edit_url") if not clip_id: raise RuntimeError("Réponse Twitch invalide: clip id manquant") return {"id": str(clip_id), "edit_url": str(edit_url or ""), "url": f"https://clips.twitch.tv/{clip_id}"}