# Jeremylongshore Claude Code Plugins Plus Skills Klingai Usage Analytics > Kling AI Usage Analytics - Skill: `tomevault-io/jeremylongshore-claude-code-plugins-plus-skills-klingai-usag` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add tomevault-io/jeremylongshore-claude-code-plugins-plus-skills-klingai-usag` - Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/jeremylongshore-claude-code-plugins-plus-skills-klingai-usag/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Data & Analytics - Author: tomevault-io (https://skillmd.com/u/tomevault-io) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/tomevault-io/jeremylongshore-claude-code-plugins-plus-skills-klingai-usag --- # 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}") ``` ## Resources - [Developer Portal](https://app.klingai.com/global/dev) - [Pricing Reference](https://app.klingai.com/global/dev/document-api/productBilling/prePaidResourcePackage) --- > Converted and distributed by [TomeVault](https://tomevault.io/claim/jeremylongshore) — claim your Tome and manage your conversions.