"""
TikTok Data Sync Example
This example demonstrates syncing TikTok marketing data to Lark Base.
The data will automatically update any dashboards that use this table.
"""
import os
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Any
from dotenv import load_dotenv
from lark_client import LarkBaseClient, LarkConfig
from data_manager import DataManager
def fetch_tiktok_campaign_data() -> List[Dict[str, Any]]:
"""
Fetch campaign data from TikTok Ads API.
In a real implementation, this would call the TikTok Ads API.
For this example, we return mock data.
Returns:
List of campaign data dictionaries
"""
# Mock TikTok campaign data
campaigns = [
{
"campaign_id": "TT001",
"campaign_name": "Summer Sale 2025",
"status": "Active",
"budget": 5000.00,
"spend": 3250.50,
"impressions": 125000,
"clicks": 4500,
"conversions": 180,
"start_date": datetime(2025, 6, 1, tzinfo=timezone.utc),
"updated_at": datetime.now(timezone.utc) - timedelta(hours=2)
},
{
"campaign_id": "TT002",
"campaign_name": "Brand Awareness Q2",
"status": "Active",
"budget": 8000.00,
"spend": 7200.00,
"impressions": 350000,
"clicks": 12000,
"conversions": 450,
"start_date": datetime(2025, 4, 1, tzinfo=timezone.utc),
"updated_at": datetime.now(timezone.utc) - timedelta(hours=1)
},
{
"campaign_id": "TT003",
"campaign_name": "Product Launch",
"status": "Active",
"budget": 10000.00,
"spend": 5500.00,
"impressions": 200000,
"clicks": 8000,
"conversions": 320,
"start_date": datetime(2025, 5, 15, tzinfo=timezone.utc),
"updated_at": datetime.now(timezone.utc) - timedelta(minutes=30)
},
{
"campaign_id": "TT004",
"campaign_name": "Holiday Special",
"status": "Paused",
"budget": 3000.00,
"spend": 2800.00,
"impressions": 95000,
"clicks": 3200,
"conversions": 125,
"start_date": datetime(2024, 12, 1, tzinfo=timezone.utc),
"updated_at": datetime.now(timezone.utc) - timedelta(days=1)
}
]
return campaigns
def transform_tiktok_to_lark(record: Dict[str, Any]) -> Dict[str, Any]:
"""
Transform TikTok campaign data to Lark Base format.
Args:
record: TikTok campaign record
Returns:
Transformed record for Lark Base
"""
# Calculate metrics
ctr = (record['clicks'] / record['impressions'] * 100) if record['impressions'] > 0 else 0
cvr = (record['conversions'] / record['clicks'] * 100) if record['clicks'] > 0 else 0
cpc = (record['spend'] / record['clicks']) if record['clicks'] > 0 else 0
cpa = (record['spend'] / record['conversions']) if record['conversions'] > 0 else 0
budget_usage = (record['spend'] / record['budget'] * 100) if record['budget'] > 0 else 0
# Convert to Lark Base format
return {
"Campaign ID": record['campaign_id'],
"Campaign Name": record['campaign_name'],
"Status": record['status'],
"Budget": record['budget'],
"Spend": record['spend'],
"Budget Usage (%)": round(budget_usage, 2),
"Impressions": record['impressions'],
"Clicks": record['clicks'],
"Conversions": record['conversions'],
"CTR (%)": round(ctr, 2),
"CVR (%)": round(cvr, 2),
"CPC": round(cpc, 2),
"CPA": round(cpa, 2),
"Start Date": DataManager.convert_to_lark_datetime(record['start_date']),
"Last Updated": DataManager.convert_to_lark_datetime(record['updated_at'])
}
def setup_tiktok_table(client: LarkBaseClient, app_token: str) -> str:
"""
Set up the TikTok campaigns table with all necessary fields.
Args:
client: LarkBaseClient instance
app_token: Lark Base app token
Returns:
Table ID of the created table
"""
print("Setting up TikTok Campaigns table...")
# Create table
table_id = client.create_table(
app_token=app_token,
table_name="TikTok Campaigns",
default_view_name="All Campaigns"
)
print(f"Created table: {table_id}")
# Define fields
fields = [
("Campaign ID", 1, None), # Text
("Campaign Name", 1, None), # Text
("Status", 3, {"options": [{"name": "Active"}, {"name": "Paused"}, {"name": "Completed"}]}), # Single Select
("Budget", 2, None), # Number
("Spend", 2, None), # Number
("Budget Usage (%)", 2, None), # Number
("Impressions", 2, None), # Number
("Clicks", 2, None), # Number
("Conversions", 2, None), # Number
("CTR (%)", 2, None), # Number
("CVR (%)", 2, None), # Number
("CPC", 2, None), # Number
("CPA", 2, None), # Number
("Start Date", 5, None), # DateTime
("Last Updated", 5, None), # DateTime
]
# Create fields
for field_name, field_type, property_dict in fields:
client.create_field(
app_token=app_token,
table_id=table_id,
field_name=field_name,
field_type=field_type,
property_dict=property_dict
)
print(f"Created field: {field_name}")
return table_id
def main():
"""Run TikTok data sync example."""
# Load environment variables
load_dotenv()
# Initialize client and data manager
config = LarkConfig(
app_id=os.getenv('LARK_APP_ID'),
app_secret=os.getenv('LARK_APP_SECRET'),
log_level='INFO'
)
client = LarkBaseClient(config, rate_limit=10)
data_manager = DataManager(client)
app_token = os.getenv('LARK_APP_TOKEN')
print("=== TikTok Data Sync Example ===\n")
# Set up table (only needed first time)
print("Step 1: Set up TikTok campaigns table")
table_id = setup_tiktok_table(client, app_token)
print(f"Table ID: {table_id}\n")
# Fetch TikTok data
print("Step 2: Fetch TikTok campaign data")
tiktok_data = fetch_tiktok_campaign_data()
print(f"Fetched {len(tiktok_data)} campaigns from TikTok\n")
# Transform and load data
print("Step 3: Transform and sync data to Lark Base")
result = data_manager.transform_and_load(
app_token=app_token,
table_id=table_id,
source_records=tiktok_data,
transformer=transform_tiktok_to_lark,
key_field="Campaign ID",
parallel=True,
max_workers=4
)
print(f"\nSync Results:")
print(f" - Created: {result['created']} campaigns")
print(f" - Updated: {result['updated']} campaigns")
print(f" - Total: {result['total']} campaigns\n")
# Verify data
print("Step 4: Verify synced data")
records = client.get_all_records(app_token, table_id)
print(f"Total records in table: {len(records)}\n")
# Display summary
print("Campaign Summary:")
print("-" * 80)
for record in records:
fields = record['fields']
print(f" {fields['Campaign Name']}")
print(f" Status: {fields['Status']}")
print(f" Budget: ${fields['Budget']:,.2f} | Spend: ${fields['Spend']:,.2f} "
f"({fields['Budget Usage (%)']}%)")
print(f" Impressions: {fields['Impressions']:,} | Clicks: {fields['Clicks']:,} "
f"| Conversions: {fields['Conversions']:,}")
print(f" CTR: {fields['CTR (%)']}% | CVR: {fields['CVR (%)']}% "
f"| CPC: ${fields['CPC']:.2f} | CPA: ${fields['CPA']:.2f}")
print()
print("=== Sync completed successfully! ===")
print("\nNext steps:")
print("1. Create a dashboard in Lark")
print("2. Add charts visualizing this data (spend, CTR, conversions, etc.)")
print("3. The dashboard will automatically update when you re-run this sync")
print(f"\nTable ID for dashboard creation: {table_id}")
# Example: Incremental sync (for subsequent runs)
print("\n--- Incremental Sync Example ---")
print("For subsequent syncs, use incremental sync to only update changed data:")
last_sync = datetime.now(timezone.utc) - timedelta(hours=24)
incremental_result = data_manager.sync_data_incremental(
app_token=app_token,
table_id=table_id,
data_source=fetch_tiktok_campaign_data,
key_field="Campaign ID",
timestamp_field="updated_at",
last_sync_time=last_sync
)
print(f"Incremental sync updated {incremental_result['updated']} records")
if __name__ == '__main__':
main()
