|
|
|
|
|
import streamlit as st |
|
|
import requests |
|
|
import webbrowser |
|
|
import secrets |
|
|
import threading |
|
|
import time |
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler |
|
|
from urllib.parse import urlparse, parse_qs |
|
|
from dotenv import load_dotenv |
|
|
import os |
|
|
|
|
|
load_dotenv() |
|
|
|
|
|
|
|
|
CLIENT_ID = os.getenv("MAL_CLIENT_ID") |
|
|
CLIENT_SECRET = os.getenv("MAL_CLIENT_SECRET") |
|
|
REDIRECT_URI = os.getenv("MAL_REDIRECT_URI", "http://localhost:8000/callback") |
|
|
PORT = int(os.getenv("MAL_PORT", 8000)) |
|
|
|
|
|
class MALAuth: |
|
|
def __init__(self): |
|
|
self.code_verifier = secrets.token_urlsafe(64) |
|
|
self.code_challenge = self.code_verifier |
|
|
self.auth_code = None |
|
|
self.error = None |
|
|
|
|
|
class OAuthHandler(BaseHTTPRequestHandler): |
|
|
auth_code = None |
|
|
error = None |
|
|
|
|
|
def do_GET(self): |
|
|
parsed = urlparse(self.path) |
|
|
params = parse_qs(parsed.query) |
|
|
if "code" in params: |
|
|
MALAuth.OAuthHandler.auth_code = params["code"][0] |
|
|
self.send_response(200) |
|
|
self.end_headers() |
|
|
self.wfile.write(b"Authorization successful! Return to the app.") |
|
|
elif "error" in params: |
|
|
MALAuth.OAuthHandler.error = params["error"][0] |
|
|
self.send_response(400) |
|
|
self.end_headers() |
|
|
self.wfile.write(b"Authorization failed. Check your settings.") |
|
|
else: |
|
|
self.send_response(400) |
|
|
self.end_headers() |
|
|
self.wfile.write(b"Invalid request") |
|
|
|
|
|
def run_server(self): |
|
|
server = HTTPServer(('localhost', PORT), self.OAuthHandler) |
|
|
server.timeout = 120 |
|
|
server.handle_request() |
|
|
|
|
|
def start_oauth_flow(self): |
|
|
server_thread = threading.Thread(target=self.run_server) |
|
|
server_thread.daemon = True |
|
|
server_thread.start() |
|
|
|
|
|
auth_url = ( |
|
|
"https://myanimelist.net/v1/oauth2/authorize?" |
|
|
f"response_type=code&" |
|
|
f"client_id={CLIENT_ID}&" |
|
|
f"code_challenge={self.code_challenge}&" |
|
|
f"redirect_uri={REDIRECT_URI}" |
|
|
) |
|
|
webbrowser.open(auth_url) |
|
|
|
|
|
start_time = time.time() |
|
|
while not self.OAuthHandler.auth_code and not self.OAuthHandler.error: |
|
|
if time.time() - start_time > 120: |
|
|
return None, "Authorization timed out" |
|
|
time.sleep(0.5) |
|
|
|
|
|
return self.OAuthHandler.auth_code, self.OAuthHandler.error |
|
|
|
|
|
def get_access_token(self, auth_code): |
|
|
token_url = "https://myanimelist.net/v1/oauth2/token" |
|
|
data = { |
|
|
"client_id": CLIENT_ID, |
|
|
"client_secret": CLIENT_SECRET, |
|
|
"code": auth_code, |
|
|
"code_verifier": self.code_verifier, |
|
|
"grant_type": "authorization_code", |
|
|
"redirect_uri": REDIRECT_URI |
|
|
} |
|
|
try: |
|
|
response = requests.post(token_url, data=data) |
|
|
response.raise_for_status() |
|
|
return response.json()["access_token"], None |
|
|
except requests.exceptions.HTTPError as e: |
|
|
return None, f"Token exchange failed: {e.response.status_code} {e.response.text}" |
|
|
|
|
|
def login_button(): |
|
|
"""Streamlit component for handling MAL login""" |
|
|
if st.button("Login with MyAnimeList"): |
|
|
with st.spinner("Authenticating..."): |
|
|
auth = MALAuth() |
|
|
auth_code, error = auth.start_oauth_flow() |
|
|
|
|
|
if error: |
|
|
st.error(error) |
|
|
return False |
|
|
|
|
|
access_token, error = auth.get_access_token(auth_code) |
|
|
if error: |
|
|
st.error(error) |
|
|
return False |
|
|
|
|
|
st.session_state.access_token = access_token |
|
|
st.rerun() |
|
|
return True |
|
|
|
