batch_socialblade_profiles.py

"""Run a bounded batch of Social Blade profiles through the hosted Actor."""

from __future__ import annotations

import argparse
import json
import os
from typing import Any

from apify_client import ApifyClient


ACTOR_ID = "datascraperes/socialblade-scraper"


def _dataset_id(run: Any) -> str:
    value = getattr(run, "default_dataset_id", None)
    if value:
        return value
    value = getattr(run, "defaultDatasetId", None)
    if value:
        return value
    return run["defaultDatasetId"]


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--profile",
        action="append",
        required=True,
        help="Social Blade URL or bare handle; repeat for a batch.",
    )
    args = parser.parse_args()

    token = os.environ.get("APIFY_API_TOKEN")
    if not token:
        raise SystemExit("Set APIFY_API_TOKEN before running this example.")

    client = ApifyClient(token)
    run = client.actor(ACTOR_ID).call(run_input={"profiles": args.profile})
    for item in client.dataset(_dataset_id(run)).iterate_items():
        print(json.dumps(item, ensure_ascii=False))


if __name__ == "__main__":
    main()