How to Integrate Replit with LinkedIn Ads

1import os
2import requests
3
4# Exchange authorization code for access token
5def exchange_code_for_token(auth_code: str, redirect_uri: str) -> dict:
6    """Call this once after completing the OAuth flow to get your initial token."""
7    token_url = "https://www.linkedin.com/oauth/v2/accessToken"
8    data = {
9        "grant_type": "authorization_code",
10        "code": auth_code,
11        "client_id": os.environ["LINKEDIN_CLIENT_ID"],
12        "client_secret": os.environ["LINKEDIN_CLIENT_SECRET"],
13        "redirect_uri": redirect_uri
14    }
15    response = requests.post(token_url, data=data)
16    response.raise_for_status()
17    token_data = response.json()
18    print(f"Access token expires in: {token_data.get('expires_in')} seconds")
19    print(f"Access token: {token_data['access_token'][:20]}...")
20    return token_data
21
22# After getting the token, store it manually in Replit Secrets:
23# LINKEDIN_ACCESS_TOKEN = token_data['access_token']
24
25if __name__ == "__main__":
26    # Replace these with your actual values from the OAuth flow
27    AUTH_CODE = input("Enter the authorization code from the redirect URL: ")
28    REDIRECT_URI = "https://your-app.replit.app/auth/callback"
29    tokens = exchange_code_for_token(AUTH_CODE, REDIRECT_URI)
30    print("\nStore this in Replit Secrets as LINKEDIN_ACCESS_TOKEN:")
31    print(tokens['access_token'])

1import os
2import requests
3from datetime import datetime, timedelta
4from urllib.parse import quote
5
6ACCESS_TOKEN = os.environ["LINKEDIN_ACCESS_TOKEN"]
7AD_ACCOUNT_ID = os.environ["LINKEDIN_AD_ACCOUNT_ID"]
8BASE_URL = "https://api.linkedin.com/v2"
9
10HEADERS = {
11    "Authorization": f"Bearer {ACCESS_TOKEN}",
12    "Content-Type": "application/json",
13    "X-Restli-Protocol-Version": "2.0.0"
14}
15
16def get_account_info() -> dict:
17    """Verify API access and retrieve account details."""
18    url = f"{BASE_URL}/adAccountsV2/{AD_ACCOUNT_ID}"
19    response = requests.get(url, headers=HEADERS)
20    response.raise_for_status()
21    return response.json()
22
23def get_campaign_groups() -> list:
24    """Fetch all campaign groups (top-level campaign containers)."""
25    url = f"{BASE_URL}/adCampaignGroupsV2"
26    params = {
27        "q": "search",
28        "search.account.values[0]": f"urn:li:sponsoredAccount:{AD_ACCOUNT_ID}",
29        "count": 100
30    }
31    response = requests.get(url, headers=HEADERS, params=params)
32    response.raise_for_status()
33    return response.json().get('elements', [])
34
35def get_campaigns(status: str = None) -> list:
36    """Fetch all campaigns. Optional status: ACTIVE, PAUSED, ARCHIVED."""
37    url = f"{BASE_URL}/adCampaignsV2"
38    params = {
39        "q": "search",
40        "search.account.values[0]": f"urn:li:sponsoredAccount:{AD_ACCOUNT_ID}",
41        "count": 100
42    }
43    if status:
44        params["search.status.values[0]"] = status
45    response = requests.get(url, headers=HEADERS, params=params)
46    response.raise_for_status()
47    return response.json().get('elements', [])
48
49def get_campaign_analytics(start_days_ago: int = 30) -> list:
50    """Fetch campaign-level analytics for the past N days."""
51    end_date = datetime.now()
52    start_date = end_date - timedelta(days=start_days_ago)
53
54    url = f"{BASE_URL}/adAnalyticsV2"
55    params = {
56        "q": "analytics",
57        "pivot": "CAMPAIGN",
58        "dateRange.start.day": start_date.day,
59        "dateRange.start.month": start_date.month,
60        "dateRange.start.year": start_date.year,
61        "dateRange.end.day": end_date.day,
62        "dateRange.end.month": end_date.month,
63        "dateRange.end.year": end_date.year,
64        "timeGranularity": "ALL",
65        "accounts[0]": f"urn:li:sponsoredAccount:{AD_ACCOUNT_ID}",
66        "fields": "costInLocalCurrency,impressions,clicks,totalEngagements,leads,pivotValue"
67    }
68    response = requests.get(url, headers=HEADERS, params=params)
69    response.raise_for_status()
70    return response.json().get('elements', [])
71
72def get_lead_gen_form_responses(form_id: str, since_days: int = 1) -> list:
73    """Fetch lead gen form responses from the past N days."""
74    since_ts = int((datetime.now() - timedelta(days=since_days)).timestamp() * 1000)
75    url = f"{BASE_URL}/leadGenerationFormResponses"
76    params = {
77        "q": "account",
78        "account": f"urn:li:sponsoredAccount:{AD_ACCOUNT_ID}",
79        "form": f"urn:li:leadGenerationForm:{form_id}",
80        "submittedAtStart": since_ts,
81        "count": 100
82    }
83    response = requests.get(url, headers=HEADERS, params=params)
84    response.raise_for_status()
85    return response.json().get('elements', [])
86
87# Example usage
88if __name__ == "__main__":
89    try:
90        account = get_account_info()
91        print(f"Account: {account.get('name', AD_ACCOUNT_ID)}")
92
93        campaigns = get_campaigns(status='ACTIVE')
94        print(f"Active campaigns: {len(campaigns)}")
95        for c in campaigns[:3]:
96            print(f"  [{c.get('id')}] {c.get('name')} — {c.get('status')}")
97
98        analytics = get_campaign_analytics(30)
99        print(f"\nAnalytics data points: {len(analytics)}")
100        for stat in analytics[:3]:
101            print(f"  Campaign: {stat.get('pivotValue')} — "
102                  f"Spend: {stat.get('costInLocalCurrency', 0):.2f}, "
103                  f"Clicks: {stat.get('clicks', 0)}")
104    except requests.HTTPError as e:
105        print(f"API error: {e.response.status_code} — {e.response.text}")

1const express = require('express');
2const axios = require('axios');
3
4const app = express();
5app.use(express.json());
6
7const ACCESS_TOKEN = process.env.LINKEDIN_ACCESS_TOKEN;
8const AD_ACCOUNT_ID = process.env.LINKEDIN_AD_ACCOUNT_ID;
9const BASE_URL = 'https://api.linkedin.com/v2';
10
11const linkedin = axios.create({
12  baseURL: BASE_URL,
13  headers: {
14    'Authorization': `Bearer ${ACCESS_TOKEN}`,
15    'Content-Type': 'application/json',
16    'X-Restli-Protocol-Version': '2.0.0'
17  }
18});
19
20// Get all campaigns for the account
21app.get('/campaigns', async (req, res) => {
22  try {
23    const params = {
24      q: 'search',
25      'search.account.values[0]': `urn:li:sponsoredAccount:${AD_ACCOUNT_ID}`,
26      count: 100
27    };
28    if (req.query.status) {
29      params['search.status.values[0]'] = req.query.status.toUpperCase();
30    }
31    const response = await linkedin.get('/adCampaignsV2', { params });
32    res.json(response.data.elements || []);
33  } catch (err) {
34    console.error('Campaigns error:', err.response?.data);
35    res.status(err.response?.status || 500).json({ error: err.message });
36  }
37});
38
39// Get analytics for all campaigns in the account
40app.get('/analytics', async (req, res) => {
41  const days = parseInt(req.query.days) || 30;
42  const endDate = new Date();
43  const startDate = new Date(endDate - days * 24 * 60 * 60 * 1000);
44
45  const formatDate = (d) => ({
46    day: d.getDate(),
47    month: d.getMonth() + 1,
48    year: d.getFullYear()
49  });
50
51  const start = formatDate(startDate);
52  const end = formatDate(endDate);
53
54  try {
55    const params = {
56      q: 'analytics',
57      pivot: 'CAMPAIGN',
58      'dateRange.start.day': start.day,
59      'dateRange.start.month': start.month,
60      'dateRange.start.year': start.year,
61      'dateRange.end.day': end.day,
62      'dateRange.end.month': end.month,
63      'dateRange.end.year': end.year,
64      timeGranularity: 'ALL',
65      'accounts[0]': `urn:li:sponsoredAccount:${AD_ACCOUNT_ID}`,
66      fields: 'costInLocalCurrency,impressions,clicks,leads,pivotValue'
67    };
68    const response = await linkedin.get('/adAnalyticsV2', { params });
69    res.json(response.data.elements || []);
70  } catch (err) {
71    res.status(err.response?.status || 500).json({ error: err.message });
72  }
73});
74
75// Pause or resume a campaign
76app.patch('/campaigns/:id/status', async (req, res) => {
77  const { status } = req.body; // 'ACTIVE' or 'PAUSED'
78  if (!['ACTIVE', 'PAUSED'].includes(status)) {
79    return res.status(400).json({ error: 'status must be ACTIVE or PAUSED' });
80  }
81  try {
82    // LinkedIn uses PATCH with full campaign object for status updates
83    await linkedin.patch(`/adCampaignsV2/${req.params.id}`, {
84      patch: { $set: { status } }
85    });
86    res.json({ success: true, id: req.params.id, status });
87  } catch (err) {
88    res.status(err.response?.status || 500).json({ error: err.message });
89  }
90});
91
92app.listen(3000, '0.0.0.0', () => {
93  console.log('LinkedIn Ads integration server running on port 3000');
94});

1import os
2import requests
3from datetime import datetime
4
5def check_token_validity() -> dict:
6    """
7    Check if the current access token is still valid.
8    Returns token info including expiry details.
9    """
10    token = os.environ["LINKEDIN_ACCESS_TOKEN"]
11    url = "https://www.linkedin.com/oauth/v2/introspectToken"
12    data = {
13        "client_id": os.environ["LINKEDIN_CLIENT_ID"],
14        "client_secret": os.environ["LINKEDIN_CLIENT_SECRET"],
15        "token": token
16    }
17    response = requests.post(url, data=data)
18    response.raise_for_status()
19    token_info = response.json()
20
21    if token_info.get('active'):
22        expires_at = token_info.get('expires_at', 0)
23        if expires_at:
24            expiry_date = datetime.fromtimestamp(expires_at / 1000)
25            days_left = (expiry_date - datetime.now()).days
26            print(f"Token is active. Expires in {days_left} days ({expiry_date.date()})")
27            if days_left < 10:
28                print("WARNING: Token expires soon — re-authorize the LinkedIn OAuth flow!")
29    else:
30        print("Token is INACTIVE or EXPIRED. Re-authorization required.")
31
32    return token_info
33
34if __name__ == "__main__":
35    info = check_token_validity()
36    print(f"Token status: {'Active' if info.get('active') else 'Inactive'}")