-
Notifications
You must be signed in to change notification settings - Fork 218
Expand file tree
/
Copy pathcloudsave.py
More file actions
159 lines (135 loc) · 5.25 KB
/
cloudsave.py
File metadata and controls
159 lines (135 loc) · 5.25 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""
Cloud sync for cheetahclaws sessions via GitHub Gist.
Supported provider: GitHub Gist
- No extra cloud account needed beyond a GitHub Personal Access Token
- Sessions stored as private Gists (JSON), browsable in GitHub UI
- Zero extra dependencies (uses urllib from stdlib)
Config keys (stored in ~/.cheetahclaws/config.json):
gist_token — GitHub Personal Access Token (needs 'gist' scope)
cloudsave_auto — bool: auto-upload on /exit
cloudsave_last_gist_id — last uploaded gist ID (for in-place update)
"""
from __future__ import annotations
import json
import urllib.request
import urllib.error
from datetime import datetime
GIST_TAG = "[cheetahclaws]"
_API = "https://api.github.com"
# ── Low-level Gist API ────────────────────────────────────────────────────────
def _request(method: str, path: str, token: str, body: dict | None = None) -> dict:
url = f"{_API}{path}"
data = json.dumps(body).encode() if body else None
req = urllib.request.Request(
url,
data=data,
method=method,
headers={
"Authorization": f"token {token}",
"Accept": "application/vnd.github+json",
"Content-Type": "application/json",
"X-GitHub-Api-Version": "2022-11-28",
},
)
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read())
def _request_safe(method: str, path: str, token: str, body: dict | None = None):
"""Like _request but returns (result, error_str)."""
try:
return _request(method, path, token, body), None
except urllib.error.HTTPError as e:
msg = e.read().decode(errors="replace")
try:
msg = json.loads(msg).get("message", msg)
except Exception:
pass
return None, f"GitHub API {e.code}: {msg}"
except Exception as e:
return None, str(e)
# ── Public API ────────────────────────────────────────────────────────────────
def validate_token(token: str) -> tuple[bool, str]:
"""Check token is valid and has gist scope. Returns (ok, message)."""
result, err = _request_safe("GET", "/user", token)
if err:
return False, f"Token validation failed: {err}"
scopes_needed = {"gist"}
# GitHub returns X-OAuth-Scopes header but urllib doesn't easily expose it;
# a successful /user call is sufficient for basic validation.
login = result.get("login", "unknown")
return True, login
def upload_session(
session_data: dict,
token: str,
description: str = "",
gist_id: str | None = None,
) -> tuple[str | None, str | None]:
"""
Create or update a Gist with the session JSON.
Returns (gist_id, error). On success gist_id is the Gist ID.
"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M")
desc = f"{GIST_TAG} {description or ts}"
filename = f"cheetahclaws_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
content = json.dumps(session_data, indent=2, default=str)
body = {
"description": desc,
"public": False,
"files": {filename: {"content": content}},
}
if gist_id:
result, err = _request_safe("PATCH", f"/gists/{gist_id}", token, body)
else:
result, err = _request_safe("POST", "/gists", token, body)
if err:
return None, err
return result["id"], None
def list_sessions(token: str, max_results: int = 20) -> tuple[list[dict], str | None]:
"""
List Gists tagged as cheetahclaws sessions.
Returns (list of {id, description, updated_at, url}), error).
"""
result, err = _request_safe("GET", "/gists?per_page=100", token)
if err:
return [], err
sessions = [
{
"id": g["id"],
"description": g["description"],
"updated_at": g["updated_at"],
"url": g["html_url"],
"files": list(g["files"].keys()),
}
for g in result
if g.get("description", "").startswith(GIST_TAG)
]
return sessions[:max_results], None
def download_session(token: str, gist_id: str) -> tuple[dict | None, str | None]:
"""
Fetch a Gist and return the parsed session JSON.
Returns (session_data, error).
"""
result, err = _request_safe("GET", f"/gists/{gist_id}", token)
if err:
return None, err
files = result.get("files", {})
if not files:
return None, "Gist has no files"
# Take the first (and usually only) file
file_info = next(iter(files.values()))
raw_content = file_info.get("content")
if not raw_content:
# Truncated — fetch raw URL
raw_url = file_info.get("raw_url")
if not raw_url:
return None, "Could not retrieve file content"
req = urllib.request.Request(
raw_url,
headers={"Authorization": f"token {token}"},
)
with urllib.request.urlopen(req) as resp:
raw_content = resp.read().decode()
try:
data = json.loads(raw_content)
except json.JSONDecodeError as e:
return None, f"Invalid JSON in Gist: {e}"
return data, None