classify_posts.py

"""Bucket a list of TikTok posts by TikTok's own content categories.

`post` returns `categories`, TikTok's own classification of a video (not a guess
from the caption), so you can group a competitor's output by theme without
writing a classifier. This reads one post URL per line, fetches each through the
API, writes a CSV, and prints a category tally.

    export CHOCODATA_API_KEY="your_key"
    python tiktok_post_scraper_api_codes/classify_posts.py my_urls.txt

With no file it falls back to urls.txt beside the script, then to a built-in
sample list. A 502 is retryable and uncharged, so each URL is retried once.
"""
import csv
import os
import sys
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor

import requests

from _common import ENDPOINT, api_key, rounded

SAMPLE = [
    "https://www.tiktok.com/@duolingo/video/7459895174467177774",
    "https://www.tiktok.com/@tiktok/video/7106594312292453675",
]
OUT_CSV = "tiktok_posts_classified.csv"
MAX_WORKERS = 4  # keep inside the Free plan's concurrency of 10


def load_urls(argv):
    if len(argv) > 1 and os.path.exists(argv[1]):
        path = argv[1]
    else:
        beside = os.path.join(os.path.dirname(__file__), "urls.txt")
        path = beside if os.path.exists(beside) else None
    if not path:
        print("No URL file given; using the built-in sample list.")
        return list(SAMPLE)
    with open(path, encoding="utf-8") as f:
        return [ln.strip() for ln in f if ln.strip() and not ln.startswith("#")]


def fetch(url):
    key = api_key()
    for attempt in (1, 2):
        try:
            r = requests.get(ENDPOINT, params={"api_key": key, "url": url}, timeout=90)
        except requests.RequestException:
            time.sleep(8)
            continue
        if r.status_code == 200:
            return url, r.json()
        if r.status_code == 502 and attempt == 1:
            time.sleep(8)
            continue
        return url, None
    return url, None


def main():
    urls = load_urls(sys.argv)
    print(f"Classifying {len(urls)} posts through {ENDPOINT}\n")

    rows, tally = [], Counter()
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        for url, p in pool.map(fetch, urls):
            if not p:
                print(f"  skipped (502/again later): {url}")
                continue
            cats = p.get("categories") or []
            for c in cats:
                tally[c] += 1
            s = p["stats"]
            rows.append({
                "author": p["author"]["uniqueId"],
                "post_id": p["id"],
                "created_at": p["created_at"][:10],
                "primary_category": cats[0] if cats else "",
                "all_categories": " | ".join(cats),
                "location": p["location"],
                "plays": s["plays"],
                "plays_precision": "rounded" if rounded(s["plays"]) else "exact",
                "likes": s["likes"],
                "saves": s["saves"],
                "hashtags": " ".join(p["hashtags"]),
                "url": p["url"],
            })

    rows.sort(key=lambda r: (r["primary_category"], -r["plays"]))
    if rows:
        with open(OUT_CSV, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
            w.writeheader()
            w.writerows(rows)
        print(f"\nWrote {len(rows)} rows to {OUT_CSV}\n")

    print("Category tally (TikTok's own labels):")
    for cat, n in tally.most_common():
        print(f"  {n:>3}  {cat}")
    return 0


if __name__ == "__main__":
    sys.exit(main())