How to Integrate Replit with Pinterest Ads

1import os
2import time
3import requests
4from base64 import b64encode
5from typing import Optional
6
7CLIENT_ID = os.environ["PINTEREST_CLIENT_ID"]
8CLIENT_SECRET = os.environ["PINTEREST_CLIENT_SECRET"]
9REFRESH_TOKEN = os.environ["PINTEREST_REFRESH_TOKEN"]
10BASE_URL = "https://api.pinterest.com/v5"
11TOKEN_URL = f"{BASE_URL}/oauth/token"
12
13# In-memory token cache
14_token_cache = {"access_token": None, "expires_at": 0}
15
16def get_access_token() -> str:
17    """Get a valid access token, refreshing if expired or missing."""
18    if time.time() < _token_cache["expires_at"] - 60:
19        return _token_cache["access_token"]
20
21    credentials = b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
22    headers = {
23        "Authorization": f"Basic {credentials}",
24        "Content-Type": "application/x-www-form-urlencoded"
25    }
26    data = {
27        "grant_type": "refresh_token",
28        "refresh_token": REFRESH_TOKEN,
29        "scope": "pins:read pins:write boards:read boards:write ads:read ads:write"
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 get_user_boards() -> list:
46    """Get all boards belonging to the authenticated user."""
47    response = requests.get(f"{BASE_URL}/boards", headers=api_headers())
48    response.raise_for_status()
49    return response.json().get("items", [])
50
51def create_pin(board_id: str, title: str, description: str,
52              image_url: str, link: str = "") -> dict:
53    """Create a new pin on the specified board."""
54    payload = {
55        "board_id": board_id,
56        "title": title,
57        "description": description,
58        "media_source": {
59            "source_type": "image_url",
60            "url": image_url
61        }
62    }
63    if link:
64        payload["link"] = link
65
66    response = requests.post(f"{BASE_URL}/pins", json=payload, headers=api_headers())
67    response.raise_for_status()
68    return response.json()
69
70def get_ad_accounts() -> list:
71    """List all ad accounts accessible to the authenticated user."""
72    response = requests.get(f"{BASE_URL}/ad_accounts", headers=api_headers())
73    response.raise_for_status()
74    return response.json().get("items", [])
75
76def get_campaign_analytics(ad_account_id: str, campaign_ids: list,
77                           start_date: str, end_date: str) -> dict:
78    """Get analytics for specific campaigns."""
79    params = {
80        "start_date": start_date,   # YYYY-MM-DD
81        "end_date": end_date,
82        "campaign_ids": campaign_ids,
83        "columns": "SPEND_IN_DOLLAR,IMPRESSION_1,CLICK_1,SAVE",
84        "granularity": "DAY"
85    }
86    response = requests.get(
87        f"{BASE_URL}/ad_accounts/{ad_account_id}/campaigns/analytics",
88        params=params,
89        headers=api_headers()
90    )
91    response.raise_for_status()
92    return response.json()
93
94# Example usage
95if __name__ == "__main__":
96    boards = get_user_boards()
97    print("Your boards:")
98    for board in boards:
99        print(f"  {board['id']}: {board['name']}")
100
101    ad_accounts = get_ad_accounts()
102    if ad_accounts:
103        print(f"\nAd account: {ad_accounts[0]['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.PINTEREST_CLIENT_ID;
9const CLIENT_SECRET = process.env.PINTEREST_CLIENT_SECRET;
10const REFRESH_TOKEN = process.env.PINTEREST_REFRESH_TOKEN;
11const BASE_URL = 'https://api.pinterest.com/v5';
12
13let tokenCache = { accessToken: null, expiresAt: 0 };
14
15async function getAccessToken() {
16  if (Date.now() / 1000 < tokenCache.expiresAt - 60) {
17    return tokenCache.accessToken;
18  }
19  const credentials = Buffer.from(`${CLIENT_ID}:${CLIENT_SECRET}`).toString('base64');
20  const response = await axios.post(
21    `${BASE_URL}/oauth/token`,
22    new URLSearchParams({
23      grant_type: 'refresh_token',
24      refresh_token: REFRESH_TOKEN,
25      scope: 'pins:read pins:write boards:read ads:read ads:write'
26    }),
27    {
28      headers: {
29        'Authorization': `Basic ${credentials}`,
30        'Content-Type': 'application/x-www-form-urlencoded'
31      }
32    }
33  );
34  tokenCache.accessToken = response.data.access_token;
35  tokenCache.expiresAt = Date.now() / 1000 + (response.data.expires_in || 3600);
36  return tokenCache.accessToken;
37}
38
39async function pinterest() {
40  const token = await getAccessToken();
41  return axios.create({
42    baseURL: BASE_URL,
43    headers: { 'Authorization': `Bearer ${token}`, 'Content-Type': 'application/json' }
44  });
45}
46
47// Create a pin on a board
48app.post('/pins', async (req, res) => {
49  const { boardId, title, description, imageUrl, link } = req.body;
50  if (!boardId || !imageUrl) {
51    return res.status(400).json({ error: 'boardId and imageUrl are required' });
52  }
53  try {
54    const api = await pinterest();
55    const payload = {
56      board_id: boardId,
57      title: title || '',
58      description: description || '',
59      media_source: { source_type: 'image_url', url: imageUrl }
60    };
61    if (link) payload.link = link;
62    const { data } = await api.post('/pins', payload);
63    res.json({ success: true, pinId: data.id, url: `https://pinterest.com/pin/${data.id}` });
64  } catch (err) {
65    console.error('Pin error:', err.response?.data || err.message);
66    res.status(500).json({ error: err.message });
67  }
68});
69
70// Get campaign analytics
71app.get('/analytics/:adAccountId', async (req, res) => {
72  const { startDate, endDate, campaignIds } = req.query;
73  if (!startDate || !endDate) {
74    return res.status(400).json({ error: 'startDate and endDate are required (YYYY-MM-DD)' });
75  }
76  try {
77    const api = await pinterest();
78    const params = {
79      start_date: startDate,
80      end_date: endDate,
81      columns: 'SPEND_IN_DOLLAR,IMPRESSION_1,CLICK_1,SAVE,TOTAL_CONVERSIONS',
82      granularity: 'TOTAL'
83    };
84    if (campaignIds) params.campaign_ids = campaignIds.split(',');
85
86    const { data } = await api.get(
87      `/ad_accounts/${req.params.adAccountId}/campaigns/analytics`,
88      { params }
89    );
90    res.json(data);
91  } catch (err) {
92    console.error('Analytics error:', err.response?.data || err.message);
93    res.status(500).json({ error: err.message });
94  }
95});
96
97// List boards
98app.get('/boards', async (req, res) => {
99  try {
100    const api = await pinterest();
101    const { data } = await api.get('/boards');
102    res.json(data.items || []);
103  } catch (err) {
104    res.status(500).json({ error: err.message });
105  }
106});
107
108app.get('/health', (req, res) => res.json({ status: 'ok' }));
109
110app.listen(3000, '0.0.0.0', () => {
111  console.log('Pinterest integration server running on port 3000');
112});

1import hashlib
2import requests
3from datetime import datetime
4import os
5
6AD_ACCOUNT_ID = os.environ["PINTEREST_AD_ACCOUNT_ID"]
7
8def sha256_hash(value: str) -> str:
9    """Hash PII data before sending to Pinterest API."""
10    return hashlib.sha256(value.lower().strip().encode()).hexdigest()
11
12def send_conversion_event(event_name: str, email: str,
13                          order_value: float, order_id: str,
14                          ip_address: str = "", user_agent: str = "") -> dict:
15    """
16    Send a server-side conversion event to Pinterest.
17    event_name: checkout, add_to_cart, page_visit, signup, lead
18    """
19    payload = {
20        "data": [
21            {
22                "event_name": event_name,
23                "action_source": "web",
24                "event_time": int(datetime.now().timestamp()),
25                "event_id": order_id,
26                "user_data": {
27                    "em": [sha256_hash(email)],  # Hashed email
28                    "client_ip_address": ip_address,
29                    "client_user_agent": user_agent
30                },
31                "custom_data": {
32                    "currency": "USD",
33                    "value": str(order_value),
34                    "order_id": order_id
35                }
36            }
37        ]
38    }
39
40    response = requests.post(
41        f"https://api.pinterest.com/v5/ad_accounts/{AD_ACCOUNT_ID}/events",
42        json=payload,
43        headers=api_headers()  # Using the token from Step 2
44    )
45    response.raise_for_status()
46    return response.json()
47
48# Example: send a checkout conversion
49if __name__ == "__main__":
50    result = send_conversion_event(
51        event_name="checkout",
52        email="customer@example.com",
53        order_value=129.99,
54        order_id="ORD-2024-001"
55    )
56    print(f"Conversion event sent: {result}")