update update

twitch_bot/twitch_helix.py

@@ -0,0 +1,125 @@
+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}"}
+