TikTok 竞争对手分析：建立每周内容报告

import csv
import json
import os
from datetime import datetime, timezone
from statistics import median
from urllib.request import Request, urlopen

ENDPOINT = "https://api.scrapeless.com/api/v1/scraper/request"
TOKEN = os.environ["SCRAPELESS_API_KEY"]
ACCOUNTS = [name.strip().lstrip("@") for name in os.environ[
    "TIKTOK_COMPETITOR_USERNAMES"
].split(",") if name.strip()]
REQUESTED_COUNT = 20

THEMES = {
    "product_education": {"how to", "tips", "explained", "tutorial", "howto"},
    "launch": {"new", "introducing", "available now", "launch", "newproduct"},
    "promotion": {"sale", "discount", "limited offer", "deal"},
    "community": {"duet", "challenge", "your turn", "community"},
}


def run_actor(actor, actor_input):
    data = json.dumps({"actor": actor, "input": actor_input}).encode()
    request = Request(
        ENDPOINT,
        data=data,
        headers={"x-api-token": TOKEN, "content-type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=60) as response:
        return json.load(response)


def labels_for(item):
    hashtag_values = []
    for tag in item.get("hashtags") or []:
        hashtag_values.append(
            str(tag.get("name", "")) if isinstance(tag, dict) else str(tag)
        )
    searchable = " ".join([
        str(item.get("description") or ""),
        " ".join(hashtag_values),
    ]).lower()
    labels = [
        theme for theme, terms in THEMES.items()
        if any(term in searchable for term in terms)
    ]
    return labels or ["uncategorized"]


def metric(item, name):
    value = item.get(name)
    return value if isinstance(value, (int, float)) else None


collected_at = datetime.now(timezone.utc).isoformat()
rows = []

for username in ACCOUNTS:
    profile = run_actor("scraper.tiktok.user.detail", {"unique_id": username})
    response = run_actor(
        "scraper.tiktok.user.work",
        {"sec_uid": profile["sec_uid"], "cursor": "0", "count": REQUESTED_COUNT},
    )
    with open(f"{username}-posts-raw.json", "w", encoding="utf-8") as raw_file:
        json.dump(response, raw_file, ensure_ascii=False, indent=2)
    for item in response.get("items") or []:
        rows.append({
            "account": username,
            "collected_at": collected_at,
            "requested_count": REQUESTED_COUNT,
            "post_id": str(item.get("id") or item.get("post_id") or ""),
            "post_url": item.get("url") or item.get("post_url") or "",
            "created_at": item.get("create_time") or item.get("date") or "",
            "themes": "|".join(labels_for(item)),
            "play_count": metric(item, "play_count"),
            "like_count": metric(item, "like_count"),
            "comment_count": metric(item, "comment_count"),
            "share_count": metric(item, "share_count"),
            "is_pinned": item.get("is_pinned"),
        })

with open("competitor-posts.csv", "w", newline="", encoding="utf-8") as output:
    writer = csv.DictWriter(output, fieldnames=rows[0].keys() if rows else ["account"])
    writer.writeheader()
    writer.writerows(rows)

with open("competitor-weekly-summary.csv", "w", newline="", encoding="utf-8") as output:
    fields = ["account", "observed_posts", "median_plays", "median_likes"]
    writer = csv.DictWriter(output, fieldnames=fields)
    writer.writeheader()
    for username in ACCOUNTS:
        account_rows = [row for row in rows if row["account"] == username]
        plays = [row["play_count"] for row in account_rows if row["play_count"] is not None]
        likes = [row["like_count"] for row in account_rows if row["like_count"] is not None]
        writer.writerow({
            "account": username,
            "observed_posts": len(account_rows),
            "median_plays": median(plays) if plays else "",
            "median_likes": median(likes) if likes else "",
        })