|
import urllib.parse |
|
from utils.meldrx import MeldRxAPI |
|
import os |
|
|
|
def extract_code_from_url(url: str) -> str: |
|
"""Extracts the 'code' parameter from a given URL.""" |
|
parsed_url = urllib.parse.urlparse(url) |
|
query_params = urllib.parse.parse_qs(parsed_url.query) |
|
return query_params.get("code", [""])[0] |
|
|
|
class CallbackManager: |
|
def __init__(self, redirect_uri: str, client_secret: str = None): |
|
client_id = os.getenv("APPID") |
|
if not client_id: |
|
raise ValueError("APPID environment variable not set.") |
|
workspace_id = os.getenv("WORKSPACE_URL") |
|
if not workspace_id: |
|
raise ValueError("WORKSPACE_URL environment variable not set.") |
|
self.api = MeldRxAPI(client_id, client_secret, workspace_id, redirect_uri) |
|
self.auth_code = None |
|
self.access_token = None |
|
|
|
def handle_callback(self, callback_url: str) -> str: |
|
"""Handles the callback URL and extracts the code automatically.""" |
|
self.auth_code = extract_code_from_url(callback_url) |
|
if not self.auth_code: |
|
return "No authentication code found in URL." |
|
|
|
if self.api.authenticate_with_code(self.auth_code): |
|
self.access_token = self.api.access_token |
|
return f"Authentication successful! Access Token: {self.access_token[:10]}... (truncated)" |
|
return "Authentication failed. Please check the authorization code." |