File size: 6,251 Bytes
f577535 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """
MyAnimeList API v2 — OAuth2 (PKCE) helpers and anime list update utilities.
"""
import os
import hashlib
import base64
import secrets
import time
import logging
import requests
from ..core.config import Config
logger = logging.getLogger(__name__)
MAL_AUTH_URL = "https://myanimelist.net/v1/oauth2/authorize"
MAL_TOKEN_URL = "https://myanimelist.net/v1/oauth2/token"
MAL_API_BASE = "https://api.myanimelist.net/v2"
# ── PKCE helpers ────────────────────────────────────────────────
def _generate_code_verifier(length: int = 128) -> str:
"""Generate a high-entropy code_verifier (43–128 chars, URL-safe)."""
return secrets.token_urlsafe(length)[:128]
def _generate_code_challenge(verifier: str) -> str:
"""MAL uses 'plain' code_challenge_method — verifier == challenge."""
return verifier
# ── OAuth2 flow ─────────────────────────────────────────────────
def get_mal_auth_url(state: str, code_verifier: str) -> str:
"""Build the MAL authorization URL (with PKCE)."""
params = {
"response_type": "code",
"client_id": Config.MAL_CLIENT_ID,
"redirect_uri": Config.MAL_REDIRECT_URI,
"state": state,
"code_challenge": _generate_code_challenge(code_verifier),
"code_challenge_method": "plain",
}
from urllib.parse import urlencode
return f"{MAL_AUTH_URL}?{urlencode(params)}"
def exchange_mal_code(code: str, code_verifier: str) -> dict | None:
"""Exchange authorization code for access + refresh tokens."""
try:
resp = requests.post(MAL_TOKEN_URL, data={
"client_id": Config.MAL_CLIENT_ID,
"client_secret": Config.MAL_CLIENT_SECRET,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": Config.MAL_REDIRECT_URI,
"code_verifier": code_verifier,
}, timeout=15)
if resp.status_code != 200:
logger.error(f"MAL token exchange failed ({resp.status_code}): {resp.text[:300]}")
return None
data = resp.json()
return {
"access_token": data["access_token"],
"refresh_token": data.get("refresh_token"),
"expires_in": data.get("expires_in", 3600),
}
except Exception as e:
logger.error(f"MAL token exchange error: {e}")
return None
def refresh_mal_token(refresh_token: str) -> dict | None:
"""Refresh an expired MAL access token."""
try:
resp = requests.post(MAL_TOKEN_URL, data={
"client_id": Config.MAL_CLIENT_ID,
"client_secret": Config.MAL_CLIENT_SECRET,
"grant_type": "refresh_token",
"refresh_token": refresh_token,
}, timeout=15)
if resp.status_code != 200:
logger.error(f"MAL token refresh failed ({resp.status_code}): {resp.text[:300]}")
return None
data = resp.json()
return {
"access_token": data["access_token"],
"refresh_token": data.get("refresh_token", refresh_token),
"expires_in": data.get("expires_in", 3600),
}
except Exception as e:
logger.error(f"MAL token refresh error: {e}")
return None
# ── Authenticated API calls ─────────────────────────────────────
def _mal_headers(access_token: str) -> dict:
return {"Authorization": f"Bearer {access_token}"}
def get_mal_user_info(access_token: str) -> dict | None:
"""GET /v2/users/@me — returns basic profile info."""
try:
resp = requests.get(
f"{MAL_API_BASE}/users/@me",
headers=_mal_headers(access_token),
params={"fields": "anime_statistics"},
timeout=10,
)
if resp.status_code != 200:
logger.error(f"MAL user info failed ({resp.status_code}): {resp.text[:300]}")
return None
return resp.json()
except Exception as e:
logger.error(f"MAL user info error: {e}")
return None
def update_mal_anime_status(
access_token: str,
mal_id: int,
*,
status: str | None = None,
num_watched_episodes: int | None = None,
score: int | None = None,
) -> bool:
"""
PATCH /v2/anime/{mal_id}/my_list_status
Updates the user's list entry on MAL.
status: watching | completed | on_hold | dropped | plan_to_watch
"""
url = f"{MAL_API_BASE}/anime/{mal_id}/my_list_status"
data = {}
if status:
data["status"] = status
if num_watched_episodes is not None:
data["num_watched_episodes"] = num_watched_episodes
if score is not None:
data["score"] = score
if not data:
return True # nothing to update
try:
resp = requests.patch(
url,
headers=_mal_headers(access_token),
data=data,
timeout=10,
)
if resp.status_code == 200:
logger.info(f"MAL update OK: mal_id={mal_id} {data}")
return True
logger.error(f"MAL update failed ({resp.status_code}): {resp.text[:300]}")
return False
except Exception as e:
logger.error(f"MAL update error for mal_id={mal_id}: {e}")
return False
def get_mal_anime_status(access_token: str, mal_id: int) -> dict | None:
"""GET /v2/anime/{mal_id} with list_status field — check current progress."""
try:
resp = requests.get(
f"{MAL_API_BASE}/anime/{mal_id}",
headers=_mal_headers(access_token),
params={"fields": "my_list_status,num_episodes"},
timeout=10,
)
if resp.status_code != 200:
return None
return resp.json()
except Exception as e:
logger.error(f"MAL get status error for mal_id={mal_id}: {e}")
return None
|