TikTok活动报告与公共帖子指标

CREATE TABLE IF NOT EXISTS campaign_post_snapshots (
    campaign_id TEXT NOT NULL,
    post_id TEXT NOT NULL,
    collected_at TEXT NOT NULL,
    account TEXT NOT NULL,
    post_url TEXT,
    play_count INTEGER,
    like_count INTEGER,
    comment_count INTEGER,
    share_count INTEGER,
    collect_count INTEGER,
    repost_count INTEGER,
    PRIMARY KEY (campaign_id, post_id, collected_at)
);

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

ENDPOINT = "https://api.scrapeless.com/api/v1/scraper/request"
TOKEN = os.environ["SCRAPELESS_API_KEY"]
REQUESTED_COUNT = 35


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)


with open("campaign-roster.csv", newline="", encoding="utf-8") as roster_file:
    roster = list(csv.DictReader(roster_file))

for row in roster:
    row["username"] = row["username"].lstrip("@")
    row["post_id"] = str(row["post_id"])

collected_at = datetime.now(timezone.utc).isoformat()
items_by_account = {}

for username in sorted({row["username"] for row in roster}):
    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}-{collected_at[:10]}-raw.json", "w", encoding="utf-8") as raw_file:
        json.dump(response, raw_file, ensure_ascii=False, indent=2)
    items_by_account[username] = {
        str(item.get("id") or item.get("post_id") or ""): item
        for item in response.get("items") or []
    }

connection = sqlite3.connect("campaign-reporting.db")
connection.execute("""
CREATE TABLE IF NOT EXISTS campaign_post_snapshots (
    campaign_id TEXT NOT NULL,
    post_id TEXT NOT NULL,
    collected_at TEXT NOT NULL,
    account TEXT NOT NULL,
    post_url TEXT,
    play_count INTEGER,
    like_count INTEGER,
    comment_count INTEGER,
    share_count INTEGER,
    collect_count INTEGER,
    repost_count INTEGER,
    PRIMARY KEY (campaign_id, post_id, collected_at)
)
""")

coverage = []
for row in roster:
    item = items_by_account.get(row["username"], {}).get(row["post_id"])
    status = "matched" if item else "unmatched"
    coverage.append({**row, "status": status, "collected_at": collected_at})
    if not item:
        continue
    connection.execute(
        """INSERT OR REPLACE INTO campaign_post_snapshots VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            row["campaign_id"], row["post_id"], collected_at, row["username"],
            item.get("url") or item.get("post_url"), item.get("play_count"),
            item.get("like_count"), item.get("comment_count"), item.get("share_count"),
            item.get("collect_count"), item.get("repost_count"),
        ),
    )

connection.commit()
connection.close()

with open("campaign-coverage.csv", "w", newline="", encoding="utf-8") as output:
    writer = csv.DictWriter(output, fieldnames=coverage[0].keys())
    writer.writeheader()
    writer.writerows(coverage)