|
|
from fastapi import FastAPI, Request |
|
|
from fastapi.responses import JSONResponse |
|
|
import requests |
|
|
import os |
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
|
|
|
CLIENT_KEY = os.getenv("TIKTOK_CLIENT_KEY", "YOUR_APP_ID") |
|
|
CLIENT_SECRET = os.getenv("TIKTOK_CLIENT_SECRET", "YOUR_APP_SECRET") |
|
|
|
|
|
@app.get("/auth/callback") |
|
|
async def tiktok_auth_callback(request: Request): |
|
|
code = request.query_params.get("code") |
|
|
state = request.query_params.get("state") |
|
|
|
|
|
if not code: |
|
|
return JSONResponse( |
|
|
status_code=400, |
|
|
content={"message": "Missing 'code' parameter from TikTok OAuth."} |
|
|
) |
|
|
|
|
|
token_url = "https://business-api.tiktok.com/open_api/v1.3/oauth2/access_token/" |
|
|
payload = { |
|
|
"app_id": CLIENT_KEY, |
|
|
"secret": CLIENT_SECRET, |
|
|
"auth_code": code |
|
|
} |
|
|
|
|
|
response = requests.post(token_url, json=payload) |
|
|
data = response.json() |
|
|
|
|
|
if data.get("code") != 0: |
|
|
return JSONResponse( |
|
|
status_code=400, |
|
|
content={"message": "Token exchange failed", "details": data} |
|
|
) |
|
|
|
|
|
access_token = data["data"].get("access_token") |
|
|
refresh_token = data["data"].get("refresh_token", None) |
|
|
advertiser_ids = data["data"].get("advertiser_ids", []) |
|
|
|
|
|
print("✅ Access Token:", access_token) |
|
|
print("🔁 Refresh Token:", refresh_token) |
|
|
print("📢 Advertiser IDs:", advertiser_ids) |
|
|
|
|
|
return JSONResponse( |
|
|
content={ |
|
|
"message": "Token exchange successful", |
|
|
"access_token": access_token, |
|
|
"refresh_token": refresh_token, |
|
|
"advertiser_ids": advertiser_ids |
|
|
} |
|
|
) |
