Klingai Usage Analytics

---
name: klingai-usage-analytics
description: 'Build usage analytics and reporting for Kling AI video generation. Use
  when tracking patterns,

  analyzing costs, or building dashboards. Trigger with phrases like ''klingai analytics'',

  ''kling ai usage report'', ''klingai metrics'', ''video generation stats''.

  '
allowed-tools: Read, Write, Edit, Bash(npm:*), Grep
version: 1.18.0
license: MIT
author: Jeremy Longshore <jeremy@intentsolutions.io>
tags:
- saas
- kling-ai
- analytics
- reporting
compatibility: Designed for Claude Code
---
# Kling AI Usage Analytics

## Overview

Track video generation usage with structured logging, aggregate metrics, daily reports, and cost analysis. Built on JSONL event logs that can feed into any analytics platform.

## Event Logger

```python
import json
import time
from datetime import datetime
from pathlib import Path

class KlingEventLogger:
    """Append-only JSONL event log for Kling AI operations."""

    def __init__(self, log_dir: str = "logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)

    def _write(self, event: dict):
        date = datetime.utcnow().strftime("%Y-%m-%d")
        filepath = self.log_dir / f"kling-{date}.jsonl"
        event["timestamp"] = datetime.utcnow().isoformat()
        with open(filepath, "a") as f:
            f.write(json.dumps(event) + "\n")

    def log_submission(self, task_id, prompt, model, duration, mode):
        self._write({
            "event": "task_submitted",
            "task_id": task_id,
            "model": model,
            "duration": int(duration),
            "mode": mode,
            "prompt_len": len(prompt),
        })

    def log_completion(self, task_id, status, elapsed_sec, credits_used):
        self._write({
            "event": "task_completed",
            "task_id": task_id,
            "status": status,
            "elapsed_sec": elapsed_sec,
            "credits_used": credits_used,
        })

    def log_error(self, task_id, error_type, message):
        self._write({
            "event": "task_error",
            "task_id": task_id,
            "error_type": error_type,
            "message": message[:200],
        })
```

## Analytics Aggregator

```python
from collections import defaultdict

class UsageAnalytics:
    """Aggregate metrics from JSONL event logs."""

    def __init__(self, log_dir: str = "logs"):
        self.log_dir = Path(log_dir)

    def _read_events(self, date: str = None):
        pattern = f"kling-{date}.jsonl" if date else "kling-*.jsonl"
        events = []
        for filepath in sorted(self.log_dir.glob(pattern)):
            with open(filepath) as f:
                for line in f:
                    events.append(json.loads(line))
        return events

    def daily_summary(self, date: str = None) -> dict:
        date = date or datetime.utcnow().strftime("%Y-%m-%d")
        events = self._read_events(date)

        submitted = [e for e in events if e["event"] == "task_submitted"]
        completed = [e for e in events if e["event"] == "task_completed"]
        errors = [e for e in events if e["event"] == "task_error"]

        succeeded = [e for e in completed if e["status"] == "succeed"]
        failed = [e for e in completed if e["status"] == "failed"]

        total_credits = sum(e.get("credits_used", 0) for e in completed)
        avg_elapsed = (sum(e["elapsed_sec"] for e in succeeded) / len(succeeded)
                      if succeeded else 0)

        by_model = defaultdict(int)
        for e in submitted:
            by_model[e["model"]] += 1

        return {
            "date": date,
            "total_submitted": len(submitted),
            "succeeded": len(succeeded),
            "failed": len(failed),
            "errors": len(errors),
            "success_rate": f"{len(succeeded) / max(len(completed), 1) * 100:.1f}%",
            "total_credits": total_credits,
            "avg_generation_sec": round(avg_elapsed),
            "by_model": dict(by_model),
        }

    def print_report(self, date: str = None):
        s = self.daily_summary(date)
        print(f"\n=== Kling AI Usage Report: {s['date']} ===")
        print(f"Submitted:    {s['total_submitted']}")
        print(f"Succeeded:    {s['succeeded']}")
        print(f"Failed:       {s['failed']}")
        print(f"Success rate: {s['success_rate']}")
        print(f"Credits used: {s['total_credits']}")
        print(f"Avg time:     {s['avg_generation_sec']}s")
        print(f"By model:")
        for model, count in s["by_model"].items():
            print(f"  {model}: {count}")
```

## Cost Analysis

```python
def cost_analysis(analytics: UsageAnalytics, days: int = 7):
    """Analyze cost trends over recent days."""
    from datetime import timedelta

    daily_costs = []
    for i in range(days):
        date = (datetime.utcnow() - timedelta(days=i)).strftime("%Y-%m-%d")
        summary = analytics.daily_summary(date)
        daily_costs.append({
            "date": date,
            "credits": summary["total_credits"],
            "videos": summary["total_submitted"],
            "estimated_usd": summary["total_credits"] * 0.14,
        })

    total_credits = sum(d["credits"] for d in daily_costs)
    total_videos = sum(d["videos"] for d in daily_costs)
    total_cost = sum(d["estimated_usd"] for d in daily_costs)

    print(f"\n=== {days}-Day Cost Summary ===")
    print(f"Total credits: {total_credits}")
    print(f"Total videos:  {total_videos}")
    print(f"Est. cost:     ${total_cost:.2f}")
    print(f"Avg/day:       ${total_cost / days:.2f}")

    for d in daily_costs:
        print(f"  {d['date']}: {d['credits']} credits, {d['videos']} videos, ${d['estimated_usd']:.2f}")
```

## Export to CSV

```python
import csv

def export_usage_csv(analytics: UsageAnalytics, output: str = "kling_usage.csv"):
    events = analytics._read_events()
    with open(output, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["timestamp", "event", "task_id",
                                                "model", "status", "credits_used",
                                                "elapsed_sec"])
        writer.writeheader()
        for e in events:
            writer.writerow({k: e.get(k, "") for k in writer.fieldnames})
    print(f"Exported {len(events)} events to {output}")
```

## Prerequisites

- An append-only event sink with restricted access, an explicit retention/deletion schedule, a timezone, and an approved pricing source for aggregate cost calculations.
- Define an opaque run/task identifier and a schema that records operational facts only. Prompts, source media, faces/voices, signed URLs, personal data, credentials, and free-form provider messages are prohibited in logs.
- Seed the parser and dashboard with synthetic events, and define anomaly thresholds, export destinations, and an owner who can pause generation when spend or policy signals drift.

## Instructions

1. Validate each event against the allowlisted schema before writing it. Normalize timestamps to the documented timezone and reject records with raw prompts, media, identities, or secrets.
2. Use opaque IDs to correlate submission, completion, error, and cleanup events. Deduplicate retries and preserve the original event time so totals cannot be inflated by a replay.
3. Aggregate by date, model, mode, status, and credit bucket. Apply the current approved rate only to totals, and keep dashboards and CSV exports at aggregate level.
4. Alert on failure-rate, latency, credit, retention, policy, and unexpected-destination thresholds. Pause new work for a confirmed anomaly, run a synthetic canary after remediation, and require owner approval before resuming.
5. Enforce retention deletion, verify the deletion receipt, and retain only a redacted aggregate report and incident/rollback reference.

## Output

Produce an aggregate report containing date range, submitted/succeeded/failed counts, success rate, latency summary, credits, estimated cost range, model distribution, anomaly state, retention/deletion status, and owner approval or rollback state. It must not expose prompts, media, likenesses, audio, signed URLs, contact details, billing identifiers, or credentials.

## Error Handling

- Quarantine malformed, duplicate, out-of-order, or schema-breaking events without counting them twice; report the aggregate gap and repair from an approved source.
- If a log contains prohibited content, stop exports, restrict access, remove the offending record under the retention policy, and record only the redacted cleanup receipt.
- If pricing is missing or stale, show credits without currency conversion and mark cost as unknown. If storage, deletion, or anomaly checks fail, pause dependent reporting and generation until an owner approves recovery.

## Examples

A safe fixture can contain `run_id=synthetic-run-01`, `event=task_completed`, `model=kling-v2-5-turbo`, `status=succeed`, `credits_used=10`, `elapsed_sec=42`, and no prompt or URL. A seven-day report may publish counts and cost ranges to an internal dashboard only after `schema=pass`, `pii_scan=pass`, `retention=verified`, and `export=aggregate-only`.

## Resources

- [Developer Portal](https://app.klingai.com/global/dev)
- [Pricing Reference](https://app.klingai.com/global/dev/document-api/productBilling/prePaidResourcePackage)