Xây dựng một Pipeline Dữ liệu TikTok cho Phân tích Có thể Lặp lại

TL;DR:

• Một pipeline dữ liệu TikTok nên bảo tồn các phản hồi thô trước khi chuẩn hóa. Các payload nguồn giúp thay đổi trình phân tích và xem xét sự trôi dạt trường.
• Các lần thu thập cần có bảng trạng thái riêng. Một yêu cầu thất bại là một lỗ hổng trong phạm vi, không phải là một hồ sơ trống, danh sách bài đăng, hoặc sản phẩm.
• Các định danh TikTok nên nằm trong các cột văn bản. Lưu trữ chuỗi ngăn chặn các ID dài bị thay đổi bởi chuyển đổi số.
• Các bảng snapshot mô tả sự quan sát theo thời gian. Chúng không nên ghi đè lên các giá trị trước đó của người sáng tạo, bài đăng, hoặc Cửa hàng.
• Phạm vi lịch sử phụ thuộc vào các trang đã thu thập và dữ liệu tiếp tục đã xác minh. Một phản hồi trang đầu tiên không phải là một lịch sử tài khoản đầy đủ.
• Miễn phí để bắt đầu. Các tài khoản Scrapeless mới bao gồm tín dụng miễn phí thông qua Bảng điều khiển Scrapeless.

Giới thiệu: phân tích bắt đầu với bằng chứng thu thập

Một bảng điều khiển chỉ có thể giải thích các bản ghi đã đến cơ sở dữ liệu của nó. Nếu không có các payload thô, trạng thái chạy, và timestamp thu thập, một biểu đồ trống không thể phân biệt không hoạt động với một lần thu thập thất bại hoặc một sự thay đổi trình phân tích.

Hướng dẫn này xây dựng một pipeline dữ liệu TikTok nhỏ gọn từ các diễn viên TikTok của Scrapeless đến SQLite. Thiết kế bảo tồn JSON thô, chuẩn hóa các snapshot của người sáng tạo, bài đăng, và Cửa hàng, thực hiện kiểm tra chất lượng dữ liệu, và cung cấp các truy vấn SQL nhỏ cho phân tích. Lịch trình, hoạt động cơ sở dữ liệu sản xuất, và cung cấp thông tin kinh doanh vẫn là trách nhiệm của ứng dụng.

Hướng dẫn diễn viên TikTok tài liệu bản đồ diễn viên được sử dụng bởi bộ thu thập.

Tổng Quan về Pipeline

Pipeline ghi lại những gì mỗi yêu cầu trả về. Nó không tuyên bố lịch sử TikTok đầy đủ trừ khi mọi trang cần thiết và giá trị tiếp tục đã được thu thập và xác minh.

Các yêu cầu trước

• Một tài khoản Scrapeless và khóa API từ Bảng điều khiển Scrapeless
• Python 3 với thư viện tiêu chuẩn và SQLite
• Một tên người dùng TikTok công khai để thu thập hồ sơ và bài đăng
• ID sản phẩm và khu vực Cửa hàng TikTok tùy chọn cho các snapshot sản phẩm
• Một địa điểm lưu trữ, chính sách giữ lại, và lịch thu thập
• SCRAPELESS_API_KEY và TIKTOK_USERNAME cho bộ thu thập; các biến môi trường Cửa hàng là tùy chọn

Mã end-to-end là một khoảng cách yêu cầu trước vì một thông tin đăng nhập Scrapeless sống và các định danh mục tiêu đến từ người đọc. Tên diễn viên, trường đầu vào, và các cột đã chuẩn hóa theo tài liệu giao diện đã cung cấp mà không trình bày đầu ra tự tạo.

Giai đoạn 1: Cung cấp Danh tính cho Mỗi Nỗ lực Thu thập

Tạo một run_id cho một lần thu thập logic và một hàng cho mỗi yêu cầu diễn viên. Lưu trữ tên diễn viên, đầu vào yêu cầu, thời gian thu thập, trạng thái, và bất kỳ chi tiết lỗi nào. Bảng payload thô nên tham chiếu đến cùng một lần chạy và ghi lại phiên bản trình phân tích.

Xử lý phản hồi HTTP nên phân biệt các phản hồi ứng dụng thành công với các thất bại. tiêu chuẩn ngữ nghĩa HTTP định nghĩa khung mã trạng thái được sử dụng bởi các client và server.

Một bảng phạm vi sau đó có thể trả lời ba câu hỏi cơ bản:

• Các thực thể nào đã được yêu cầu?
• Các yêu cầu nào tạo ra các payload có thể sử dụng?
• Các bảng đã chuẩn hóa nào nhận hàng từ mỗi payload?

Đừng biểu diễn một yêu cầu bài đăng thất bại như một mảng items trống. Những trạng thái đó có ý nghĩa phân tích khác nhau.

Giai đoạn 2: Bảo tồn JSON Thô Trước Khi Phân Tích

Các phản hồi thô là lớp kiểm toán cho một quy trình làm việc API TikTok đến cơ sở dữ liệu. Lưu trữ tên diễn viên, đầu vào yêu cầu, thời gian thu thập, JSON phản hồi, và phiên bản trình phân tích cùng nhau. Tài liệu JSON của Python định nghĩa giao diện tuần tự được sử dụng trong ví dụ.

Lưu trữ thô phục vụ ba mục đích thực tiễn:

1. Một trình phân tích có thể chạy lại sau một thay đổi lược đồ.
2. Một giá trị chuẩn hóa nghi vấn có thể được truy vết đến trường nguồn của nó.
3. Các trường mới có thể được đầy từ các payload đã giữ mà không cần lặp lại thu thập.

Xóa bỏ bí mật trước khi ghi metadata yêu cầu. Khóa API thuộc về cấu hình quy trình và không bao giờ nên vào bảng payload thô hoặc chạy.

Bắt Đầu Lập Trình với Scrapeless

Tăng cường quy trình thu thập dữ liệu và tự động hóa của bạn với Scrapeless!
Đăng ký hôm nay và nhận $5 tín dụng miễn phí — không cần thẻ tín dụng.
Nhận tín dụng miễn phí của bạn ngay bây giờ tại Scrapeless Dashboard.

Giai đoạn 3: Chuẩn hóa Hồ sơ Người sáng tạo, Bài đăng và Sản phẩm

Giữ nguyên các định danh tự nhiên dưới dạng văn bản và bao gồm collected_at trong mỗi khóa snapshot. Hồ sơ người sáng tạo có thể thay đổi, số lượng bài đăng có thể tăng và giá của sản phẩm trong Shop có thể thay đổi về giá, hàng tồn kho, đánh giá hoặc số lượng nhận xét.

Lớp chuẩn hóa có thể bắt đầu với các bảng sau:

Tài liệu CREATE TABLE của SQLite mô tả khóa chính và các ràng buộc loại phía sau mô hình này.

Lưu ý: Mã bên dưới yêu cầu sống SCRAPELESS_API_KEY và TIKTOK_USERNAME giá trị. TIKTOK_SHOP_PRODUCT_ID và TIKTOK_SHOP_REGION là tùy chọn và cho phép nhánh Shop.

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()

Ví dụ thu thập chỉ trang bài đăng yêu cầu đầu tiên. Nó không hứa hẹn lịch sử tài khoản hoàn chỉnh vì không giả định trường tiếp tục nào ngoài phản hồi đã xác minh.

Giai đoạn 4: Thêm Kiểm tra Chất lượng Dữ liệu Trước Phân Tích

Kiểm tra chất lượng nên được thực hiện ở cả hai lớp thu thập và chuẩn hóa. Tối thiểu, đánh dấu:

• Yêu cầu diễn viên thất bại trong collection_runs
• Payload thô thành công không sản xuất được hàng thực thể mong đợi
• ID tài khoản, bài đăng hoặc sản phẩm trống
• Bộ đếm âm
• Thời gian video bằng không hoặc thiếu trong phân tích thời gian
• Dòng sản phẩm thiếu ngữ cảnh khu vực hoặc tiền tệ
• Khóa thực thể trùng lặp cho một dấu thời gian thu thập

Giữ các kiểm tra như là kết quả truy vấn thay vì chỉ thông báo trên bảng điều khiển. Một bảng điều khiển có thể hiển thị khoảng trống bao phủ bên cạnh chỉ số mà nó ảnh hưởng.

Giai đoạn 5: Truy vấn Snapshots Mà Không Xóa Thời Gian

Các hàm cửa sổ so sánh mỗi thực thể với quan sát trước đó của nó. Tài liệu hàm cửa sổ của SQLite định nghĩa LAG() cho trường hợp sử dụng này.

-- 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;

Sử dụng các chế độ hiển thị trạng thái hiện tại cho bảng điều khiển trong khi giữ cho các bảng cơ sở dữ liệu chỉ có thể thêm mới. Mô hình tương tự hỗ trợ thay đổi số lượng bài đăng, báo cáo hashtag, so sánh khoảng thời gian, thay đổi giá sản phẩm, sự kiện hàng tồn kho và giám sát đánh giá.

Scrapeless cung cấp các diễn viên TikTok thông qua Scraping API. Hãy xem trang giá hiện tại trước khi chọn bao phủ thực thể và tần suất thu thập.

Xử lý Dữ liệu TikTok Một Cách Có Trách Nhiệm

Thu thập các trường công khai cần thiết cho mục đích phân tích được xác định, hạn chế quyền truy cập vào các payload thô và thiết lập giới hạn lưu giữ cho dữ liệu cấp người sáng tạo. Khuôn khổ Quyền riêng tư NIST cung cấp hướng dẫn chung về quản trị rủi ro quyền riêng tư và giảm thiểu dữ liệu.

Giữ cấu hình hoạt động bên ngoài các dòng cơ sở dữ liệu được chia sẻ với các nhà phân tích. Các khóa API, các tuyến cảnh báo nội bộ và thông tin xác thực truy cập nên vẫn nằm trong hệ thống bí mật được kiểm soát bởi môi trường ứng dụng.

Kết luận: làm cho việc bao phủ trở nên rõ ràng bên cạnh từng chỉ số

Một pipeline dữ liệu TikTok đáng tin cậy giữ cho JSON thô, trạng thái chạy, phiên bản trình phân tích, thời gian thu thập và các snapshots đã chuẩn hóa được kết nối bằng các ID ổn định. Cấu trúc đó cho phép các nhà phân tích phân tách hoạt động thực sự không có từ việc thu thập bị thiếu, chạy lại trình phân tích sau khi thay đổi trường và theo dõi từng chỉ số đến một quan sát. Lập lịch sản xuất, mở rộng lưu trữ và cung cấp BI có thể phát triển xung quanh cùng một mô hình bằng chứng.

Sẵn sàng Xây dựng Một Pipeline Phân Tích TikTok?

Tham gia Scrapeless Discord hoặc cộng đồng Telegram để thảo luận về các sơ đồ snapshot và kho. Tạo một tài khoản trong Scrapeless Dashboard khi danh sách thực thể đầu tiên đã sẵn sàng.

Câu Hỏi Thường Gặp

Q: API TikTok nên kết nối với cơ sở dữ liệu như thế nào?

API TikTok nên kết nối với cơ sở dữ liệu thông qua một bộ thu thập ghi lại trạng thái chạy, bảo tồn JSON thô, xác thực các trường và chèn các snapshots thực thể có dấu thời gian.

Q: Tại sao lại lưu trữ các phản hồi thô của TikTok?

Các phản hồi thô của TikTok cho phép một đội ngũ truy tìm các giá trị đã chuẩn hóa, xem xét các thay đổi sơ đồ và chạy lại các trình phân tích đối với dữ liệu nguồn đã được giữ lại.
Q: Liệu các ID TikTok có nên sử dụng các cột số nguyên không?

Các ID TikTok nên sử dụng các cột văn bản vì các định danh là chuỗi không trong suốt và không nên bị thay đổi bởi sự chuyển đổi số.

Q: Yêu cầu gửi bài đầu tiên có chứa lịch sử tài khoản hoàn chỉnh không?

Yêu cầu gửi bài đầu tiên không thiết lập lịch sử tài khoản hoàn chỉnh. Phạm vi phụ thuộc vào các trang đã thu thập và dữ liệu tiếp tục đã được xác minh.

Q: Scrapeless có quản lý lịch trình và kho dữ liệu không?

Scrapeless cung cấp các phản hồi diễn viên có cấu trúc cho việc thu thập. Người gọi quản lý lập lịch, các hoạt động cơ sở dữ liệu, các phép biến đổi, giám sát chất lượng và việc giao BI.

Q: Việc thu thập dữ liệu công khai trên TikTok có hợp pháp không?

Tính hợp pháp phụ thuộc vào quyền tài phán, mục đích, phương pháp truy cập, điều khoản áp dụng và các lĩnh vực đã thu thập. Sử dụng dữ liệu công khai cho một mục đích được phép và tìm kiếm tư vấn pháp lý cho đường ống dự kiến.