export_socialblade_csv.py

"""Flatten saved Social Blade Dataset JSON into a useful CSV export."""

from __future__ import annotations

import csv
import json
import sys
from pathlib import Path
from typing import Any


FIELDS = [
    "recordType",
    "status",
    "platform",
    "creatorId",
    "username",
    "displayName",
    "primaryAudience",
    "primaryAudienceMetric",
    "mediaCount",
    "engagementRate",
    "averageLikes",
    "averageComments",
    "globalRank",
    "audienceRank",
    "viewsRank",
    "countryRank",
    "categoryRank",
    "earningsStatus",
    "responseTimeMs",
    "httpRequests",
    "scrapedAt",
]


def _read_items(path: Path) -> list[dict[str, Any]]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if isinstance(value, dict):
        return [value]
    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
        raise ValueError("Input JSON must contain one object or an array of objects.")
    return value


def _row(item: dict[str, Any]) -> dict[str, Any]:
    row = {field: item.get(field) for field in FIELDS}
    row["audience"] = json.dumps(item.get("audience"), ensure_ascii=False)
    row["content"] = json.dumps(item.get("content"), ensure_ascii=False)
    row["dataQuality"] = json.dumps(item.get("dataQuality"), ensure_ascii=False)
    row["error"] = json.dumps(item.get("error"), ensure_ascii=False)
    return row


def main() -> None:
    if len(sys.argv) != 3:
        raise SystemExit("Usage: python export_socialblade_csv.py INPUT.json OUTPUT.csv")

    input_path, output_path = map(Path, sys.argv[1:])
    rows = [_row(item) for item in _read_items(input_path)]
    fieldnames = FIELDS + ["audience", "content", "dataQuality", "error"]
    with output_path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


if __name__ == "__main__":
    main()