TikTok to YouTube Pipeline Part 1: Searching TikTok with the Apify Python Client

# config.py

HOTSPOT_SEARCHES = [
    {
        "keyword": "Ukraine war",
        "region": "UA",
        "sortType": 2,       # Most recent
        "publishTime": "WEEK",
        "limit": 50,
    },
    {
        "keyword": "earthquake Turkey",
        "region": "TR",
        "sortType": 2,
        "publishTime": "YESTERDAY",
        "limit": 50,
    },
    {
        "keyword": "protest France",
        "region": "FR",
        "sortType": 1,       # Most liked
        "publishTime": "WEEK",
        "limit": 30,
    },
]

# search_tiktok.py

import json
import os
from datetime import datetime
from pathlib import Path

from apify_client import ApifyClient
from dotenv import load_dotenv

from config import HOTSPOT_SEARCHES

load_dotenv()

ACTOR_ID = "novi/advanced-search-tiktok-api"
OUTPUT_DIR = Path("data/raw")


def run_search(client: ApifyClient, search_params: dict) -> list[dict]:
    """Run a single TikTok search and return the results."""
    print(f"🔍 Searching: '{search_params['keyword']}' "
          f"(region={search_params.get('region', 'ANY')})...")

    run = client.actor(ACTOR_ID).call(run_input=search_params)
    items = list(client.dataset(run["defaultDatasetId"]).iterate_items())

    print(f"   ✅ Found {len(items)} videos.")
    return items


def extract_essential_fields(video: dict) -> dict:
    """Extract only the fields we need for the pipeline."""
    stats = video.get("statistics", {})
    author = video.get("author", {})
    play_addr = video.get("video", {}).get("play_addr", {})
    download_addr = video.get("video", {}).get("download_addr", {})

    return {
        "aweme_id": video.get("aweme_id"),
        "description": video.get("desc", ""),
        "share_url": video.get("share_url", ""),
        "create_time": video.get("create_time"),
        "region": video.get("region"),
        # Author info
        "author_username": author.get("unique_id"),
        "author_nickname": author.get("nickname"),
        # Statistics
        "play_count": stats.get("play_count", 0),
        "like_count": stats.get("digg_count", 0),
        "comment_count": stats.get("comment_count", 0),
        "share_count": stats.get("share_count", 0),
        # Video URLs
        "video_url_no_watermark": (
            play_addr.get("url_list", [None])[0]
        ),
        "video_url_watermark": (
            download_addr.get("url_list", [None])[0]
        ),
        "video_duration_ms": video.get("video", {}).get("duration"),
        # Hashtags
        "hashtags": [
            t.get("hashtag_name", "")
            for t in video.get("text_extra", [])
            if t.get("type") == 1
        ],
    }


def main():
    token = os.environ.get("APIFY_TOKEN")
    if not token:
        raise SystemExit("❌ APIFY_TOKEN not set. Check your .env file.")

    client = ApifyClient(token)
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    all_results = []

    for search_params in HOTSPOT_SEARCHES:
        raw_videos = run_search(client, search_params)
        extracted = [extract_essential_fields(v) for v in raw_videos]

        # Tag each result with the original search keyword
        for item in extracted:
            item["_search_keyword"] = search_params["keyword"]

        all_results.extend(extracted)

    # Save to a single JSON file
    output_path = OUTPUT_DIR / f"tiktok_results_{timestamp}.json"
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(all_results, f, ensure_ascii=False, indent=2)

    print(f"\n📁 Saved {len(all_results)} videos to {output_path}")


if __name__ == "__main__":
    main()