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
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
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[] == ]
errors = [e e events e[] == ]
succeeded = [e e completed e[] == ]
failed = [e e completed e[] == ]
total_credits = (e.get(, ) e completed)
avg_elapsed = ((e[] e succeeded) / (succeeded)
succeeded )
by_model = defaultdict()
e submitted:
by_model[e[]] +=
{
: date,
: (submitted),
: (succeeded),
: (failed),
: (errors),
: ,
: total_credits,
: (avg_elapsed),
: (by_model),
}
():
s = .daily_summary(date)
()
()
()
()
()
()
()
()
model, count s[].items():
()
Cost Analysis
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: $")
d daily_costs:
()
Export to CSV
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}")
Resources
