import http.server import os import socketserver import sys import urllib.parse import webbrowser from dotenv import load_dotenv # Load environment variables first load_dotenv() # Add backend to sys.path sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.oauth_handler import NOTION_OAUTH_CONFIG, OAuthHandler PORT = 8080 CODE = None class OAuthCallbackHandler(http.server.SimpleHTTPRequestHandler): def do_GET(self): global CODE query = urllib.parse.urlparse(self.path).query params = urllib.parse.parse_qs(query) if "code" in params: CODE = params["code"][0] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() html = """ Success | Atom Authentication
Connected

Notion authenticated

Your workspace is now successfully linked to Atom. You can close this tab and return to the terminal.

""" self.wfile.write(html.encode()) else: self.send_response(400) self.end_headers() self.wfile.write(b"

Authentication Failed!

No code found in redirect.

") def run_reauth(): if not NOTION_OAUTH_CONFIG.client_id or not NOTION_OAUTH_CONFIG.client_secret: print("❌ Error: NOTION_CLIENT_ID or NOTION_CLIENT_SECRET not found in .env") return handler = OAuthHandler(NOTION_OAUTH_CONFIG) # Generate auth URL # Notion doesn't use scopes in the same way, but OAuthConfig handles it auth_url = handler.get_authorization_url() print(f"\n1. Opening browser for Notion Authorization...") print(f"URL: {auth_url}\n") print(f"⚠️ IMPORTANT: Ensure your 'Redirect URI' in Notion dashboard is set to: {NOTION_OAUTH_CONFIG.redirect_uri}") webbrowser.open(auth_url) print(f"2. Waiting for callback on {NOTION_OAUTH_CONFIG.redirect_uri} ...") # Determine port from redirect_uri parsed_uri = urllib.parse.urlparse(NOTION_OAUTH_CONFIG.redirect_uri) port = parsed_uri.port or 80 try: # Bind to 127.0.0.1 only (localhost) to prevent external access - security fix with socketserver.TCPServer(("127.0.0.1", port), OAuthCallbackHandler) as httpd: httpd.handle_request() except Exception as e: print(f"❌ Error starting local server: {e}") return if CODE: print(f"3. Exchanging code for tokens...") import asyncio try: tokens = asyncio.run(handler.exchange_code_for_tokens(CODE)) access_token = tokens.get("access_token") workspace_name = tokens.get("workspace_name") print(f"\n✅ SUCCESS!") print(f"Workspace: {workspace_name}") print(f"Access Token: {access_token[:10]}...") print(f"\nUpdating .env file...") with open(".env", "r") as f: lines = f.readlines() with open(".env", "w") as f: found = False for line in lines: if line.startswith("NOTION_TOKEN="): f.write(f"NOTION_TOKEN={access_token}\n") found = True else: f.write(line) if not found: f.write(f"NOTION_TOKEN={access_token}\n") print("Done! Token saved to NOTION_TOKEN in .env") except Exception as e: print(f"❌ Token exchange failed: {e}") else: print("\n❌ Failed to get authorization code.") if __name__ == "__main__": run_reauth()