TikTok商店产品研究：比较产品观察清单

import csv
import json
import os
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"]


def fetch_product(product_id, region):
    data = json.dumps({
        "actor": "scraper.tiktok.shop.page",
        "input": {"product_id": str(product_id), "region": region},
    }).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("product-watchlist.csv", newline="", encoding="utf-8") as source:
    watchlist = list(csv.DictReader(source))

product_rows = []
sku_rows = []

for candidate in watchlist:
    raw = fetch_product(candidate["product_id"], candidate["request_region"])
    collected_at = datetime.now(timezone.utc).isoformat()
    product_id = str(raw.get("product_id") or candidate["product_id"])
    region = str(raw.get("region") or "")
    price = raw.get("price") or {}
    stock = raw.get("stock") or {}

    with open(f"{candidate['candidate_id']}-raw.json", "w", encoding="utf-8") as output:
        json.dump(raw, output, ensure_ascii=False, indent=2)

    product_rows.append({
        "candidate_id": candidate["candidate_id"],
        "category_group": candidate["category_group"],
        "collected_at": collected_at,
        "product_id": product_id,
        "region": region,
        "seller_id": str(raw.get("seller_id") or ""),
        "name": raw.get("name"),
        "currency": price.get("currency"),
        "sale_price": price.get("sale_price"),
        "original_price": price.get("original_price"),
        "available_quantity": stock.get("available_quantity"),
        "in_stock": stock.get("in_stock"),
        "rating": raw.get("rating"),
        "review_count": raw.get("review_count"),
        "sold_count": raw.get("sold_count"),
    })

    for sku in raw.get("skus") or []:
        sku_price = sku.get("price") or {}
        sku_rows.append({
            "candidate_id": candidate["candidate_id"],
            "product_id": product_id,
            "region": region,
            "sku_id": str(sku.get("sku_id") or ""),
            "options": " | ".join(
                f"{v.get('name', '')}={v.get('value', '')}"
                for v in (sku.get("options") or [])
            ),
            "currency": sku_price.get("currency"),
            "sale_price": sku_price.get("sale_price"),
            "available_quantity": sku.get("available_quantity"),
            "in_stock": sku.get("in_stock"),
            "collected_at": collected_at,
        })

for filename, rows in (("products.csv", product_rows), ("skus.csv", sku_rows)):
    if not rows:
        continue
    with open(filename, "w", newline="", encoding="utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=rows[0].keys())
        writer.writeheader()
        writer.writerows(rows)