How to Integrate Replit with Constant Contact

1import os
2import time
3import requests
4from base64 import b64encode
5from typing import Optional
6
7CLIENT_ID = os.environ["CONSTANT_CONTACT_CLIENT_ID"]
8CLIENT_SECRET = os.environ["CONSTANT_CONTACT_CLIENT_SECRET"]
9REFRESH_TOKEN = os.environ["CONSTANT_CONTACT_REFRESH_TOKEN"]
10BASE_URL = "https://api.cc.email/v3"
11TOKEN_URL = "https://authz.constantcontact.com/oauth2/default/v1/token"
12
13# In-memory token cache (use a database for multi-process deployments)
14_token_cache = {"access_token": None, "expires_at": 0}
15
16def get_access_token() -> str:
17    """Get a valid access token, refreshing if expired."""
18    if time.time() < _token_cache["expires_at"] - 60:
19        return _token_cache["access_token"]
20
21    # Encode credentials for Basic Auth
22    credentials = b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
23    headers = {
24        "Authorization": f"Basic {credentials}",
25        "Content-Type": "application/x-www-form-urlencoded"
26    }
27    data = {
28        "grant_type": "refresh_token",
29        "refresh_token": REFRESH_TOKEN
30    }
31    response = requests.post(TOKEN_URL, headers=headers, data=data)
32    response.raise_for_status()
33    token_data = response.json()
34
35    _token_cache["access_token"] = token_data["access_token"]
36    _token_cache["expires_at"] = time.time() + token_data.get("expires_in", 3600)
37    return _token_cache["access_token"]
38
39def api_headers() -> dict:
40    return {
41        "Authorization": f"Bearer {get_access_token()}",
42        "Content-Type": "application/json"
43    }
44
45def add_contact(email: str, first_name: str = "", last_name: str = "",
46                list_ids: list = None) -> dict:
47    """Add or update a contact in Constant Contact."""
48    contact_data = {
49        "email_address": {"address": email},
50        "first_name": first_name,
51        "last_name": last_name,
52        "create_source": "Account"
53    }
54    if list_ids:
55        contact_data["list_memberships"] = list_ids
56
57    response = requests.post(
58        f"{BASE_URL}/contacts",
59        json=contact_data,
60        headers=api_headers()
61    )
62    response.raise_for_status()
63    return response.json()
64
65def get_contact_lists() -> list:
66    """Retrieve all contact lists in the account."""
67    response = requests.get(f"{BASE_URL}/contact_lists", headers=api_headers())
68    response.raise_for_status()
69    return response.json().get("lists", [])
70
71def add_contact_to_list(contact_id: str, list_id: str) -> None:
72    """Add an existing contact to a specific list."""
73    response = requests.put(
74        f"{BASE_URL}/contacts/{contact_id}",
75        json={"list_memberships": [list_id]},
76        headers=api_headers()
77    )
78    response.raise_for_status()
79
80# Example usage
81if __name__ == "__main__":
82    lists = get_contact_lists()
83    print("Available lists:")
84    for lst in lists:
85        print(f"  {lst['list_id']}: {lst['name']}")
86
87    if lists:
88        result = add_contact(
89            email="test@example.com",
90            first_name="Test",
91            last_name="User",
92            list_ids=[lists[0]["list_id"]]
93        )
94        print(f"Contact added: {result.get('contact_id')}")

1const express = require('express');
2const axios = require('axios');
3const { Buffer } = require('buffer');
4
5const app = express();
6app.use(express.json());
7
8const CLIENT_ID = process.env.CONSTANT_CONTACT_CLIENT_ID;
9const CLIENT_SECRET = process.env.CONSTANT_CONTACT_CLIENT_SECRET;
10const REFRESH_TOKEN = process.env.CONSTANT_CONTACT_REFRESH_TOKEN;
11const BASE_URL = 'https://api.cc.email/v3';
12const TOKEN_URL = 'https://authz.constantcontact.com/oauth2/default/v1/token';
13
14// In-memory token cache
15let tokenCache = { accessToken: null, expiresAt: 0 };
16
17async function getAccessToken() {
18  if (Date.now() / 1000 < tokenCache.expiresAt - 60) {
19    return tokenCache.accessToken;
20  }
21  const credentials = Buffer.from(`${CLIENT_ID}:${CLIENT_SECRET}`).toString('base64');
22  const response = await axios.post(
23    TOKEN_URL,
24    new URLSearchParams({
25      grant_type: 'refresh_token',
26      refresh_token: REFRESH_TOKEN
27    }),
28    {
29      headers: {
30        'Authorization': `Basic ${credentials}`,
31        'Content-Type': 'application/x-www-form-urlencoded'
32      }
33    }
34  );
35  tokenCache.accessToken = response.data.access_token;
36  tokenCache.expiresAt = Date.now() / 1000 + (response.data.expires_in || 3600);
37  return tokenCache.accessToken;
38}
39
40async function apiHeaders() {
41  const token = await getAccessToken();
42  return {
43    'Authorization': `Bearer ${token}`,
44    'Content-Type': 'application/json'
45  };
46}
47
48// Get all contact lists
49app.get('/lists', async (req, res) => {
50  try {
51    const headers = await apiHeaders();
52    const { data } = await axios.get(`${BASE_URL}/contact_lists`, { headers });
53    res.json(data.lists || []);
54  } catch (err) {
55    console.error('Lists error:', err.response?.data || err.message);
56    res.status(500).json({ error: err.message });
57  }
58});
59
60// Subscribe a contact to a list
61app.post('/subscribe', async (req, res) => {
62  const { email, firstName, lastName, listId } = req.body;
63  if (!email || !listId) {
64    return res.status(400).json({ error: 'email and listId are required' });
65  }
66  try {
67    const headers = await apiHeaders();
68    const payload = {
69      email_address: { address: email },
70      first_name: firstName || '',
71      last_name: lastName || '',
72      create_source: 'Account',
73      list_memberships: [listId]
74    };
75    const { data } = await axios.post(`${BASE_URL}/contacts`, payload, { headers });
76    res.json({ success: true, contactId: data.contact_id });
77  } catch (err) {
78    if (err.response?.status === 409) {
79      res.status(409).json({ error: 'Contact already exists', detail: err.response.data });
80    } else {
81      console.error('Subscribe error:', err.response?.data || err.message);
82      res.status(500).json({ error: err.message });
83    }
84  }
85});
86
87// Health check
88app.get('/health', (req, res) => res.json({ status: 'ok' }));
89
90app.listen(3000, '0.0.0.0', () => {
91  console.log('Constant Contact integration server running on port 3000');
92});