Files

187 lines
6.0 KiB
Bash
Executable File

#!/usr/bin/env bash
# Backend per il plasmoide KDE Plasma 6: Antigravity Usage Dashboard.
# Esegue il client Python incorporato per recuperare le quote da Google CloudCode PA.
exec /usr/bin/python3 - "$@" << 'EOF'
import os
import sys
import json
import time
import urllib.request
import urllib.error
from datetime import datetime, timezone
TOKEN_FILE = os.path.expanduser("~/.gemini/antigravity-cli/antigravity-oauth-token")
OAUTH_URL = "https://oauth2.googleapis.com/token"
BASE_URL = "https://daily-cloudcode-pa.googleapis.com"
CLIENT_ID = "1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"
CLIENT_SECRET = "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"
UA = "antigravity/cli/1.1.17"
API_CLIENT = "google-cloud-sdk vscode_cloudshelleditor/0.1"
METADATA = json.dumps({"ideType": "ANTIGRAVITY", "platform": "LINUX", "pluginType": "GEMINI"})
def get_token_data():
if not os.path.exists(TOKEN_FILE):
return None
try:
with open(TOKEN_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return None
def save_token_data(data):
try:
tmp = TOKEN_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
os.replace(tmp, TOKEN_FILE)
except Exception:
pass
def refresh_token(refresh_tok):
req_body = json.dumps({
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"refresh_token": refresh_tok,
"grant_type": "refresh_token"
}).encode("utf-8")
req = urllib.request.Request(OAUTH_URL, data=req_body, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=15) as resp:
res = json.loads(resp.read().decode("utf-8"))
return res.get("access_token"), int(res.get("expires_in", 3600))
def call_api(path, token, payload):
url = f"{BASE_URL}{path}"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": UA,
"X-Goog-Api-Client": API_CLIENT,
"Client-Metadata": METADATA
}
body = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=body, headers=headers)
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
def parse_iso_epoch(ts):
if not ts:
return 0
try:
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return int(dt.timestamp())
except Exception:
return 0
def main():
data = get_token_data()
if not data or "token" not in data:
print(json.dumps({
"ok": False,
"error": "no_token",
"detail": f"Nessun token OAuth in {TOKEN_FILE}"
}))
sys.exit(0)
tok = data["token"]
access_tok = tok.get("access_token")
refresh_tok = tok.get("refresh_token")
project_id = "healthy-topic-dwr0v"
try:
quota_data = call_api("/v1internal:retrieveUserQuota", access_tok, {"project": project_id})
except urllib.error.HTTPError as e:
if e.code in (401, 403) and refresh_tok:
try:
new_acc, exp_in = refresh_token(refresh_tok)
tok["access_token"] = new_acc
save_token_data(data)
access_tok = new_acc
quota_data = call_api("/v1internal:retrieveUserQuota", access_tok, {"project": project_id})
except Exception as e2:
print(json.dumps({
"ok": False,
"error": "auth_error",
"detail": f"Refresh fallito: {e2}"
}))
sys.exit(0)
else:
print(json.dumps({
"ok": False,
"error": f"http_{e.code}",
"detail": f"Errore server Google HTTP {e.code}"
}))
sys.exit(0)
except Exception as e:
print(json.dumps({
"ok": False,
"error": "network_error",
"detail": str(e)
}))
sys.exit(0)
buckets = quota_data.get("buckets", [])
bucket_map = {b.get("modelId"): b for b in buckets}
# 1. Gemini Family
gem_b = (bucket_map.get("gemini-3.1-pro-high") or
bucket_map.get("gemini-3.7-flash-high") or
bucket_map.get("gemini-3.5-flash-low") or {})
gem_rem = float(gem_b.get("remainingFraction", 1.0))
gem_reset_str = gem_b.get("resetTime", "")
gemini_info = {
"id": "gemini",
"tag": "GEMINI",
"name": "Gemini 3.x",
"models": "Pro / Flash",
"remaining": round(gem_rem, 4),
"used_percent": round((1.0 - gem_rem) * 100, 1),
"reset_time": gem_reset_str,
"reset_at": parse_iso_epoch(gem_reset_str)
}
# 2. Claude Family
cld_b = (bucket_map.get("claude-sonnet-4-6") or
bucket_map.get("claude-opus-4-6-thinking") or {})
cld_rem = float(cld_b.get("remainingFraction", 1.0))
cld_reset_str = cld_b.get("resetTime", "")
claude_info = {
"id": "claude",
"tag": "CLAUDE",
"name": "Claude 4.6",
"models": "Sonnet / Opus",
"remaining": round(cld_rem, 4),
"used_percent": round((1.0 - cld_rem) * 100, 1),
"reset_time": cld_reset_str,
"reset_at": parse_iso_epoch(cld_reset_str)
}
# 3. GPT-OSS Family
gpt_b = bucket_map.get("gpt-oss-120b-medium") or {}
gpt_rem = float(gpt_b.get("remainingFraction", 1.0))
gpt_reset_str = gpt_b.get("resetTime", "")
gpt_info = {
"id": "gpt-oss",
"tag": "GPT",
"name": "GPT-OSS",
"models": "120B Medium",
"remaining": round(gpt_rem, 4),
"used_percent": round((1.0 - gpt_rem) * 100, 1),
"reset_time": gpt_reset_str,
"reset_at": parse_iso_epoch(gpt_reset_str)
}
out = {
"ok": True,
"project": project_id,
"gemini": gemini_info,
"claude": claude_info,
"gpt": gpt_info,
"groups": [gemini_info, claude_info, gpt_info]
}
print(json.dumps(out, indent=2))
if __name__ == "__main__":
main()
EOF