How to Integrate Replit with TikTok Ads

1import os
2import requests
3from datetime import datetime, timedelta
4from typing import Optional
5
6ACCESS_TOKEN = os.environ["TIKTOK_ACCESS_TOKEN"]
7ADVERTISER_ID = os.environ["TIKTOK_ADVERTISER_ID"]
8BASE_URL = "https://business-api.tiktok.com/open_api/v1.3"
9
10# TikTok uses Access-Token header, not Authorization: Bearer
11HEADERS = {
12    "Access-Token": ACCESS_TOKEN,
13    "Content-Type": "application/json"
14}
15
16def api_get(endpoint: str, params: dict = None) -> dict:
17    """Make a GET request to the TikTok Marketing API."""
18    if params is None:
19        params = {}
20    params['advertiser_id'] = ADVERTISER_ID
21    response = requests.get(f"{BASE_URL}/{endpoint}", params=params, headers=HEADERS)
22    response.raise_for_status()
23    data = response.json()
24    if data.get('code') != 0:
25        raise Exception(f"TikTok API error: {data.get('message')} (code: {data.get('code')})")
26    return data.get('data', {})
27
28def api_post(endpoint: str, payload: dict) -> dict:
29    """Make a POST request to the TikTok Marketing API."""
30    payload['advertiser_id'] = ADVERTISER_ID
31    response = requests.post(f"{BASE_URL}/{endpoint}", json=payload, headers=HEADERS)
32    response.raise_for_status()
33    data = response.json()
34    if data.get('code') != 0:
35        raise Exception(f"TikTok API error: {data.get('message')} (code: {data.get('code')})")
36    return data.get('data', {})
37
38def create_campaign(
39    name: str,
40    objective: str = "TRAFFIC",  # AWARENESS, TRAFFIC, APP_PROMOTION, LEAD_GENERATION, CONVERSIONS
41    budget_mode: str = "BUDGET_MODE_INFINITE",
42    budget: float = 0
43) -> dict:
44    """
45    Create a TikTok ad campaign.
46    Returns campaign data including campaign_id.
47    """
48    payload = {
49        "campaign_name": name,
50        "objective_type": objective,
51        "budget_mode": budget_mode,
52        "operation_status": "DISABLE"  # Start paused, enable manually after review
53    }
54    if budget > 0:
55        payload["budget"] = budget
56        payload["budget_mode"] = "BUDGET_MODE_DAY"
57
58    return api_post("campaign/create/", payload)
59
60def get_campaigns(status_filter: str = "ALL") -> list:
61    """List all campaigns for the advertiser."""
62    data = api_get("campaign/get/", {"primary_status": status_filter})
63    return data.get('list', [])
64
65def get_campaign_report(
66    campaign_ids: list,
67    start_date: str,  # YYYY-MM-DD
68    end_date: str,
69    metrics: list = None
70) -> list:
71    """
72    Retrieve performance metrics for specified campaigns.
73    Default metrics: impressions, clicks, spend, conversions, ROAS.
74    """
75    if metrics is None:
76        metrics = [
77            "campaign_name", "spend", "impressions", "clicks",
78            "ctr", "cpc", "conversions", "cost_per_conversion",
79            "total_complete_payment_roas"
80        ]
81    payload = {
82        "report_type": "BASIC",
83        "dimensions": ["campaign_id"],
84        "metrics": metrics,
85        "data_level": "AUCTION_CAMPAIGN",
86        "start_date": start_date,
87        "end_date": end_date,
88        "filtering": [
89            {"field_name": "campaign_ids", "filter_type": "IN",
90             "filter_value": str(campaign_ids)}
91        ],
92        "page_size": 100
93    }
94    return api_post("report/integrated/get/", payload).get('list', [])
95
96def update_ad_group_status(ad_group_id: str, status: str) -> dict:
97    """Enable or pause an ad group. status: 'ENABLE' or 'DISABLE'"""
98    return api_post("adgroup/status/update/", {
99        "adgroup_ids": [ad_group_id],
100        "operation_status": status
101    })
102
103# Example usage
104if __name__ == "__main__":
105    # List all campaigns
106    campaigns = get_campaigns()
107    print(f"Total campaigns: {len(campaigns)}")
108    for c in campaigns[:5]:
109        print(f"  {c['campaign_id']}: {c['campaign_name']} [{c['primary_status']}]")
110
111    # Get last 7 days performance report
112    today = datetime.now()
113    start = (today - timedelta(days=7)).strftime("%Y-%m-%d")
114    end = today.strftime("%Y-%m-%d")
115
116    if campaigns:
117        campaign_ids = [c['campaign_id'] for c in campaigns[:5]]
118        report = get_campaign_report(campaign_ids, start, end)
119        for row in report:
120            dims = row.get('dimensions', {})
121            metrics = row.get('metrics', {})
122            print(f"Campaign {dims.get('campaign_id')}: "
123                  f"Spend=${metrics.get('spend', 0)} | "
124                  f"Impressions={metrics.get('impressions', 0)} | "
125                  f"ROAS={metrics.get('total_complete_payment_roas', 'N/A')}")

1const express = require('express');
2const axios = require('axios');
3
4const app = express();
5app.use(express.json());
6
7const ACCESS_TOKEN = process.env.TIKTOK_ACCESS_TOKEN;
8const ADVERTISER_ID = process.env.TIKTOK_ADVERTISER_ID;
9const BASE_URL = 'https://business-api.tiktok.com/open_api/v1.3';
10
11const ttHeaders = {
12  'Access-Token': ACCESS_TOKEN,
13  'Content-Type': 'application/json'
14};
15
16async function tiktokGet(endpoint, params = {}) {
17  const response = await axios.get(`${BASE_URL}/${endpoint}`, {
18    headers: ttHeaders,
19    params: { ...params, advertiser_id: ADVERTISER_ID }
20  });
21  const { code, message, data } = response.data;
22  if (code !== 0) throw new Error(`TikTok API error: ${message} (code: ${code})`);
23  return data;
24}
25
26async function tiktokPost(endpoint, payload) {
27  const response = await axios.post(`${BASE_URL}/${endpoint}`, {
28    ...payload,
29    advertiser_id: ADVERTISER_ID
30  }, { headers: ttHeaders });
31  const { code, message, data } = response.data;
32  if (code !== 0) throw new Error(`TikTok API error: ${message} (code: ${code})`);
33  return data;
34}
35
36// GET /campaigns — list all campaigns
37app.get('/campaigns', async (req, res) => {
38  try {
39    const data = await tiktokGet('campaign/get/');
40    res.json({
41      total: data.page_info?.total_number || 0,
42      campaigns: data.list || []
43    });
44  } catch (error) {
45    console.error('TikTok error:', error.message);
46    res.status(500).json({ error: error.message });
47  }
48});
49
50// POST /campaigns — create a new campaign
51app.post('/campaigns', async (req, res) => {
52  const { name, objective, dailyBudget } = req.body;
53  if (!name || !objective) {
54    return res.status(400).json({ error: 'name and objective are required' });
55  }
56  try {
57    const payload = {
58      campaign_name: name,
59      objective_type: objective,
60      operation_status: 'DISABLE'  // Start paused for review
61    };
62    if (dailyBudget) {
63      payload.budget = dailyBudget;
64      payload.budget_mode = 'BUDGET_MODE_DAY';
65    } else {
66      payload.budget_mode = 'BUDGET_MODE_INFINITE';
67    }
68    const data = await tiktokPost('campaign/create/', payload);
69    res.json({ success: true, campaignId: data.campaign_id });
70  } catch (error) {
71    console.error('TikTok error:', error.message);
72    res.status(500).json({ error: error.message });
73  }
74});
75
76// GET /report?startDate=YYYY-MM-DD&endDate=YYYY-MM-DD — performance report
77app.get('/report', async (req, res) => {
78  const { startDate, endDate } = req.query;
79  if (!startDate || !endDate) {
80    return res.status(400).json({ error: 'startDate and endDate query params required (YYYY-MM-DD)' });
81  }
82  try {
83    const data = await tiktokPost('report/integrated/get/', {
84      report_type: 'BASIC',
85      dimensions: ['campaign_id'],
86      metrics: ['campaign_name', 'spend', 'impressions', 'clicks', 'ctr', 'cpc', 'conversions'],
87      data_level: 'AUCTION_CAMPAIGN',
88      start_date: startDate,
89      end_date: endDate,
90      page_size: 50
91    });
92    res.json({ rows: data.list || [] });
93  } catch (error) {
94    console.error('TikTok error:', error.message);
95    res.status(500).json({ error: error.message });
96  }
97});
98
99app.listen(3000, '0.0.0.0', () => {
100  console.log('TikTok Ads integration server running on port 3000');
101});