Detect breakout posts

from __future__ import annotations

import os
from typing import Any

import httpx

BASE_URL = "https://api.creatoraudit.com/v2"


class CreatorAuditError(Exception):
    def __init__(self, *, status: int, code: str, detail: str, request_id: str | None):
        super().__init__(f"[{status} {code}] {detail}")
        self.status = status
        self.code = code
        self.detail = detail
        self.request_id = request_id


client = httpx.Client(
    base_url=BASE_URL,
    headers={"Authorization": f"Bearer {os.environ['CREATORAUDIT_API_KEY']}"},
    timeout=30.0,
)


def call(method: str, path: str, **kwargs: Any) -> dict[str, Any]:
    resp = client.request(method, path, **kwargs)
    if resp.is_error:
        try:
            problem = resp.json()
        except ValueError:
            problem = {}
        raise CreatorAuditError(
            status=problem.get("status", resp.status_code),
            code=problem.get("code", "INTERNAL_ERROR"),
            detail=problem.get("detail", resp.text),
            request_id=resp.headers.get("X-Request-ID"),
        )
    return resp.json()

def collect_video_ids(path: str) -> list[str]:
    ids: list[str] = []
    cursor: str | None = None
    while True:
        params: dict[str, Any] = {"limit": 200}
        if cursor is not None:
            params["cursor"] = cursor

        page = call("GET", path, params=params)
        ids.extend(item["id"] for item in page["data"])

        pagination = page["pagination"]
        if not pagination["has_next"]:
            break
        cursor = pagination["next_cursor"]
    return ids


video_ids = collect_video_ids("/account-videos")
print(f"{len(video_ids)} videos")

def chunked(items: list[str], size: int = 200):
    for i in range(0, len(items), size):
        yield items[i : i + size]


deltas: list[dict[str, Any]] = []
for batch in chunked(video_ids, 200):
    result = call(
        "POST",
        "/videos/deltas",
        json={"video_ids": batch, "windows": [7]},
    )
    deltas.extend(result["data"])

def view_delta(row: dict[str, Any]) -> float:
    # Read whichever change field the delta response carries for the window.
    return float(row.get("views_delta", row.get("views", 0)) or 0)


ranked = sorted(deltas, key=view_delta, reverse=True)
breakouts = ranked[:10]

for row in breakouts:
    print(row.get("id"), "Δviews(7d):", int(view_delta(row)))

client.close()