监控 TikTok 商店评分和评论数量变化

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

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


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


def fetch_product(product_id, region):
    body = json.dumps({
        "actor": "scraper.tiktok.shop.page",
        "input": {"product_id": str(product_id), "region": region},
    }).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 parse_rating(value):
    try:
        rating = Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return None
    return rating if Decimal("0") <= rating <= Decimal("5") else None


def parse_review_count(value):
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return None
    return parsed if parsed >= 0 else None


database = sqlite3.connect("tiktok-shop-ratings.sqlite3")
database.executescript("""
CREATE TABLE IF NOT EXISTS collection_runs (
    run_id TEXT PRIMARY KEY,
    collected_at TEXT NOT NULL,
    requested_product_id TEXT NOT NULL,
    requested_region TEXT NOT NULL,
    status TEXT NOT NULL,
    detail TEXT
);
CREATE TABLE IF NOT EXISTS rating_snapshots (
    run_id TEXT PRIMARY KEY,
    collected_at TEXT NOT NULL,
    product_id TEXT NOT NULL,
    region TEXT NOT NULL,
    rating TEXT NOT NULL,
    review_count INTEGER NOT NULL,
    raw_json TEXT NOT NULL
);
""")

run_id = str(uuid.uuid4())
collected_at = utc_now()
product_id = os.environ["TIKTOK_SHOP_PRODUCT_ID"]
region = os.environ["TIKTOK_SHOP_REGION"]

try:
    raw = fetch_product(product_id, region)
    rating = parse_rating(raw.get("rating"))
    review_count = parse_review_count(raw.get("review_count"))
    returned_product_id = str(raw.get("product_id") or product_id)
    returned_region = str(raw.get("region") or region).casefold()

    if rating is None or review_count is None:
        raise ValueError("rating or review_count is missing or invalid")

    database.execute(
        "INSERT INTO rating_snapshots VALUES (?, ?, ?, ?, ?, ?, ?)",
        (
            run_id, collected_at, returned_product_id, returned_region,
            str(rating), review_count, json.dumps(raw, ensure_ascii=False),
        ),
    )
    status, detail = "success", None
except Exception as error:
    status, detail = "failed", f"{type(error).__name__}: {error}"

database.execute(
    "INSERT INTO collection_runs VALUES (?, ?, ?, ?, ?, ?)",
    (run_id, collected_at, product_id, region, status, detail),
)
database.commit()
database.close()

-- Illustrative query over the schema created above.
WITH ordered AS (
  SELECT
    collected_at,
    product_id,
    region,
    CAST(rating AS REAL) AS rating,
    review_count,
    LAG(CAST(rating AS REAL)) OVER (
      PARTITION BY product_id, region ORDER BY collected_at
    ) AS prior_rating,
    LAG(review_count) OVER (
      PARTITION BY product_id, region ORDER BY collected_at
    ) AS prior_review_count
  FROM rating_snapshots
)
SELECT
  *,
  rating - prior_rating AS rating_delta,
  review_count - prior_review_count AS review_count_delta
FROM ordered
WHERE prior_rating IS NOT NULL;