TikTokデータパイプラインを構築して、再利用可能な分析を行う

import json
import os
import sqlite3
import uuid
from datetime import datetime, timezone
from urllib.request import Request, urlopen

ENDPOINT = "https://api.scrapeless.com/api/v1/scraper/request"
PARSER_VERSION = "tiktok-v1"


def utc_now():
    return datetime.now(timezone.utc).isoformat()


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


def integer(value):
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


database = sqlite3.connect("tiktok-analytics.sqlite3")
database.executescript("""
CREATE TABLE IF NOT EXISTS collection_runs (
  run_id TEXT NOT NULL, actor TEXT NOT NULL, collected_at TEXT NOT NULL,
  request_json TEXT NOT NULL, status TEXT NOT NULL, detail TEXT,
  PRIMARY KEY (run_id, actor)
);
CREATE TABLE IF NOT EXISTS raw_payloads (
  run_id TEXT NOT NULL, actor TEXT NOT NULL, parser_version TEXT NOT NULL,
  payload_json TEXT NOT NULL, PRIMARY KEY (run_id, actor)
);
CREATE TABLE IF NOT EXISTS profile_snapshots (
  collected_at TEXT NOT NULL, account_id TEXT NOT NULL, unique_id TEXT,
  followers INTEGER, likes INTEGER, videos INTEGER,
  PRIMARY KEY (collected_at, account_id)
);
CREATE TABLE IF NOT EXISTS post_snapshots (
  collected_at TEXT NOT NULL, post_id TEXT NOT NULL, account_id TEXT NOT NULL,
  video_duration INTEGER, play_count INTEGER, like_count INTEGER,
  comment_count INTEGER, share_count INTEGER,
  PRIMARY KEY (collected_at, post_id)
);
CREATE TABLE IF NOT EXISTS post_hashtags (
  collected_at TEXT NOT NULL, post_id TEXT NOT NULL, hashtag TEXT NOT NULL,
  PRIMARY KEY (collected_at, post_id, hashtag)
);
CREATE TABLE IF NOT EXISTS product_snapshots (
  collected_at TEXT NOT NULL, product_id TEXT NOT NULL, region TEXT NOT NULL,
  name TEXT, sale_price TEXT, currency TEXT, available_quantity INTEGER,
  rating TEXT, review_count INTEGER,
  PRIMARY KEY (collected_at, product_id, region)
);
""")

run_id = str(uuid.uuid4())
collected_at = utc_now()


def collect(actor, actor_input):
    request_json = json.dumps(actor_input, sort_keys=True)
    try:
        payload = request_actor(actor, actor_input)
        database.execute(
            "INSERT INTO raw_payloads VALUES (?, ?, ?, ?)",
            (run_id, actor, PARSER_VERSION, json.dumps(payload, ensure_ascii=False)),
        )
        status, detail = "success", None
    except Exception as error:
        payload = None
        status, detail = "failed", f"{type(error).__name__}: {error}"
    database.execute(
        "INSERT INTO collection_runs VALUES (?, ?, ?, ?, ?, ?)",
        (run_id, actor, collected_at, request_json, status, detail),
    )
    return payload


profile = collect(
    "scraper.tiktok.user.detail",
    {"unique_id": os.environ["TIKTOK_USERNAME"]},
)

if profile:
    stats = profile.get("statistics") or {}
    account_id = str(profile.get("account_id") or "")
    database.execute(
        "INSERT INTO profile_snapshots VALUES (?, ?, ?, ?, ?, ?)",
        (
            collected_at, account_id, profile.get("unique_id"),
            integer(stats.get("followers")), integer(stats.get("likes")),
            integer(stats.get("videos")),
        ),
    )

    posts = collect(
        "scraper.tiktok.user.work",
        {"sec_uid": profile["sec_uid"], "cursor": "0", "count": 10},
    )
    if posts:
        for post in posts.get("items") or []:
            post_id = str(post.get("post_id") or post.get("video_id") or "")
            database.execute(
                "INSERT INTO post_snapshots VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    collected_at, post_id, account_id,
                    integer(post.get("video_duration")),
                    integer(post.get("play_count")), integer(post.get("like_count")),
                    integer(post.get("comment_count")), integer(post.get("share_count")),
                ),
            )
            for hashtag in set(post.get("hashtags") or []):
                normalized = str(hashtag).strip().removeprefix("#").casefold()
                if normalized:
                    database.execute(
                        "INSERT INTO post_hashtags VALUES (?, ?, ?)",
                        (collected_at, post_id, normalized),
                    )

product_id = os.getenv("TIKTOK_SHOP_PRODUCT_ID")
product_region = os.getenv("TIKTOK_SHOP_REGION")
if product_id and product_region:
    product = collect(
        "scraper.tiktok.shop.page",
        {"product_id": product_id, "region": product_region},
    )
    if product:
        price = product.get("price") or {}
        stock = product.get("stock") or {}
        database.execute(
            "INSERT INTO product_snapshots VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (
                collected_at, str(product.get("product_id") or product_id),
                str(product.get("region") or product_region).casefold(),
                product.get("name"), price.get("sale_price"),
                price.get("currency"), integer(stock.get("available_quantity")),
                str(product.get("rating")) if product.get("rating") is not None else None,
                integer(product.get("review_count")),
            ),
        )

database.commit()
database.close()

-- Illustrative follower-change query over the normalized schema.
SELECT
  account_id,
  collected_at,
  followers,
  followers - LAG(followers) OVER (
    PARTITION BY account_id ORDER BY collected_at
  ) AS follower_change
FROM profile_snapshots;

-- Illustrative run-coverage query.
SELECT actor, status, COUNT(*) AS run_count
FROM collection_runs
GROUP BY actor, status
ORDER BY actor, status;