#!/usr/bin/env python3
"""
SWAPA Email Analytics — One-time Constant Contact OAuth Authorization
Run this once to grant read-only campaign reporting access for the SWAPA
Constant Contact account. Opens a browser window for you to sign in and
approve access, then prints a refresh token to paste into fetch_email_stats.py.
Usage:
python3 get_email_oauth_token.py
"""
import json
import secrets
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from base64 import b64encode
from http.server import BaseHTTPRequestHandler, HTTPServer
CLIENT_ID = "65b87674-4697-486a-8d58-b3e2e6f01d4f"
CLIENT_SECRET = "jv-07NGjA_91dprCebmKZw"
SCOPE = "campaign_data offline_access"
PORT = 8080
REDIRECT_URI = f"http://localhost:{PORT}/"
AUTH_URL = "https://authz.constantcontact.com/oauth2/default/v1/authorize"
TOKEN_URL = "https://authz.constantcontact.com/oauth2/default/v1/token"
received = {}
class CallbackHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass # quiet
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
params = urllib.parse.parse_qs(parsed.query)
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
if "code" in params:
received["code"] = params["code"][0]
received["state"] = params.get("state", [""])[0]
self.wfile.write(b"Authorized. You can close this tab and return to the terminal.
")
else:
error = params.get("error_description", params.get("error", ["unknown error"]))[0]
self.wfile.write(f"Authorization failed: {error}
".encode())
def get_authorization_code():
state = secrets.token_urlsafe(16)
params = urllib.parse.urlencode({
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"scope": SCOPE,
"state": state,
})
auth_url = f"{AUTH_URL}?{params}"
print(f"Opening browser for authorization:\n {auth_url}\n")
webbrowser.open(auth_url)
server = HTTPServer(("localhost", PORT), CallbackHandler)
print(f"Waiting for you to sign in and approve access (listening on {REDIRECT_URI})...")
while "code" not in received:
server.handle_request()
server.server_close()
if received.get("state") != state:
raise RuntimeError("State mismatch — possible CSRF, aborting.")
return received["code"]
def exchange_code_for_tokens(code):
basic_auth = b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
data = urllib.parse.urlencode({
"code": code,
"redirect_uri": REDIRECT_URI,
"grant_type": "authorization_code",
}).encode()
req = urllib.request.Request(TOKEN_URL, data=data)
req.add_header("Content-Type", "application/x-www-form-urlencoded")
req.add_header("Authorization", f"Basic {basic_auth}")
req.add_header("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req, timeout=30) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
body = e.read().decode(errors="replace")
raise RuntimeError(f"Token exchange failed: HTTP {e.code}\n{body}") from None
def main():
code = get_authorization_code()
print("\nAuthorization code received. Exchanging for tokens...")
tokens = exchange_code_for_tokens(code)
if "refresh_token" not in tokens:
print("\n⚠ No refresh_token in response:", tokens)
print("Make sure the app's scope request included 'offline_access'.")
return
print("\n✓ Success. Here is your refresh token:\n")
print(f" {tokens['refresh_token']}\n")
print("Send this back so it can be added to fetch_email_stats.py.")
print("(Kept private, like the other API keys in this project. Long-lived —")
print(" only needs replacing if revoked or the client secret is regenerated.)")
if __name__ == "__main__":
main()