Kéo báo cáo bằng Python

Script hoàn chỉnh: lấy báo cáo hiệu quả theo ngày cho mọi advertiser của một kết nối, ghép tên campaign, xuất ra CSV.

Cài đặt

pip install httpx
export LOGIDATA_API_KEY="ld_live_..."
export LOGIDATA_ACCOUNT_ID="ldc_9fK2xQ..."

Client tối thiểu

import json, os, time
import httpx

BASE = "https://api.logidata.vn/v1/tiktok-ads"

class LogidataError(RuntimeError):
    pass

class Client:
    def __init__(self, api_key=None, account_id=None):
        self._c = httpx.Client(
            base_url=BASE, timeout=60,
            headers={"X-API-Key": api_key or os.environ["LOGIDATA_API_KEY"],
                     "X-Account-Id": account_id or os.environ["LOGIDATA_ACCOUNT_ID"]},
        )

    def get(self, path, **params):
        # dimensions/metrics/filtering phải là CHUỖI JSON, không phải list
        params = {k: (json.dumps(v) if isinstance(v, (list, dict)) else v)
                  for k, v in params.items() if v is not None}

        for attempt in range(4):
            r = self._c.get(path, params=params)

            if r.status_code in (429, 502):
                if attempt == 3:
                    raise LogidataError(f"vẫn lỗi sau 4 lần: {r.status_code} {r.text}")
                # Circuit breaker có thể yêu cầu chờ tới 300s — cứ tôn trọng
                time.sleep(int(r.headers.get("Retry-After", 2 ** attempt)))
                continue

            if r.status_code >= 400:
                e = r.json()["error"]
                raise LogidataError(f"{e['code']}: {e['message']} "
                                    f"(request_id={e['request_id']})")

            body = r.json()
            if body.get("code") != 0:      # lỗi nghiệp vụ TikTok đi kèm HTTP 200
                raise LogidataError(f"TikTok {body['code']}: {body.get('message')}")
            return body["data"]

    def paginate(self, path, **params):
        """Lặp hết các trang. page_size=1000 để tốn ít hạn mức nhất."""
        page, params = 1, {**params, "page_size": 1000}
        while True:
            data = self.get(path, page=page, **params)
            yield from data.get("list", [])
            info = data.get("page_info") or {}
            if page >= info.get("total_page", 1):
                return
            page += 1

Kéo báo cáo và ghép tên campaign

import csv
from datetime import date, timedelta

def pull(client, start: date, end: date, out="tiktok_ads.csv"):
    advertisers = client.get("/v1.3/oauth2/advertiser/get/")["list"]
    print(f"{len(advertisers)} advertiser")

    rows = []
    for adv in advertisers:
        aid = adv["advertiser_id"]

        # Báo cáo chỉ trả ID, nên lấy tên campaign trước rồi ghép
        names = {c["campaign_id"]: c["campaign_name"]
                 for c in client.paginate("/v1.3/campaign/get/", advertiser_id=aid)}

        for item in client.paginate(
                "/v1.3/report/integrated/get/",
                advertiser_id=aid,
                report_type="BASIC",
                data_level="AUCTION_CAMPAIGN",
                dimensions=["campaign_id", "stat_time_day"],
                metrics=["spend", "impressions", "clicks", "conversion"],
                start_date=start.isoformat(), end_date=end.isoformat()):
            d, m = item["dimensions"], item["metrics"]
            rows.append({
                "advertiser_id": aid,
                "advertiser_name": adv["advertiser_name"],
                "campaign_id": d["campaign_id"],
                "campaign_name": names.get(d["campaign_id"], ""),
                "date": d["stat_time_day"][:10],
                "spend": float(m["spend"]),
                "impressions": int(m["impressions"]),
                "clicks": int(m["clicks"]),
                "conversion": int(m["conversion"]),
            })

    with open(out, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]))
        w.writeheader()
        w.writerows(rows)
    print(f"{len(rows)} dòng -> {out}")


if __name__ == "__main__":
    today = date.today()
    # Kéo lại 3 ngày gần nhất mỗi lần chạy: TikTok còn điều chỉnh số liệu tới 48h
    pull(Client(), today - timedelta(days=3), today - timedelta(days=1))

encoding="utf-8-sig" để Excel bản tiếng Việt mở CSV không bị vỡ dấu.

Ghi thẳng vào database

import pandas as pd
from sqlalchemy import create_engine

df = pd.DataFrame(rows)
engine = create_engine(os.environ["DATABASE_URL"])

with engine.begin() as conn:
    # Idempotent: xoá đúng khoảng ngày rồi ghi lại, chạy lại bao nhiêu lần cũng đúng
    conn.exec_driver_sql(
        "DELETE FROM tiktok_ads_daily WHERE date BETWEEN %s AND %s",
        (df["date"].min(), df["date"].max()))
    df.to_sql("tiktok_ads_daily", conn, if_exists="append", index=False)

Kiểm tra hạn mức còn lại

r = httpx.get("https://api.logidata.vn/v1/me",
              headers={"X-API-Key": os.environ["LOGIDATA_API_KEY"]})
me = r.json()
print(f"{me['quota_used']}/{me['monthly_quota']} request tháng này")

Endpoint /v1/me không tính vào hạn mức.