chore: drop legacy Python CLI, keep Node.js version as cli/cfg.js
The Python CLI (cli/cfg) was outdated. The actively maintained version is the Node.js CLI published as @shazhou/config on npm. Signed-off-by: Xiaonuo <xiaonuo@shazhou.work>
This commit is contained in:
parent
f6bb13271d
commit
d6e78fc5a9
166
cli/cfg
166
cli/cfg
@ -1,166 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""cfg — CLI client for the config service.
|
||||
|
||||
Usage:
|
||||
cfg get <KEY> Read a key (personal > shared)
|
||||
cfg set <KEY> <VALUE> Write to personal scope
|
||||
cfg set --shared <KEY> <VALUE> Write to shared scope (admin)
|
||||
cfg list [--scope shared|personal] List keys
|
||||
cfg sync Sync all to local cache
|
||||
cfg delete <KEY> [--shared] Delete a key
|
||||
cfg token <TOKEN> Save auth token
|
||||
|
||||
Config: ~/.config/cfg/config.json
|
||||
Cache: ~/.config/cfg/cache.json
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from pathlib import Path
|
||||
|
||||
CONFIG_DIR = Path.home() / ".config" / "cfg"
|
||||
CONFIG_FILE = CONFIG_DIR / "config.json"
|
||||
CACHE_FILE = CONFIG_DIR / "cache.json"
|
||||
|
||||
DEFAULT_ENDPOINT = "https://config-service.oc-fleet.workers.dev"
|
||||
|
||||
|
||||
def load_config() -> dict:
|
||||
if CONFIG_FILE.exists():
|
||||
return json.loads(CONFIG_FILE.read_text())
|
||||
return {}
|
||||
|
||||
|
||||
def save_config(cfg: dict):
|
||||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
CONFIG_FILE.write_text(json.dumps(cfg, indent=2))
|
||||
|
||||
|
||||
def get_token() -> str:
|
||||
cfg = load_config()
|
||||
token = cfg.get("token") or os.environ.get("CFG_TOKEN")
|
||||
if not token:
|
||||
print("No token configured. Run: cfg token <TOKEN>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
return token
|
||||
|
||||
|
||||
def get_endpoint() -> str:
|
||||
cfg = load_config()
|
||||
return cfg.get("endpoint", DEFAULT_ENDPOINT)
|
||||
|
||||
|
||||
def api(method: str, path: str, body: dict | None = None) -> dict:
|
||||
url = f"{get_endpoint()}{path}"
|
||||
data = json.dumps(body).encode() if body else None
|
||||
req = urllib.request.Request(url, data=data, method=method)
|
||||
req.add_header("Authorization", f"Bearer {get_token()}")
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
return json.loads(resp.read())
|
||||
except urllib.error.HTTPError as e:
|
||||
result = json.loads(e.read())
|
||||
print(f"Error: {result.get('error', 'unknown')}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def cmd_get(args: list[str]):
|
||||
if not args:
|
||||
print("Usage: cfg get <KEY>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = api("GET", f"/config/{args[0]}")
|
||||
print(result["value"])
|
||||
|
||||
|
||||
def cmd_set(args: list[str]):
|
||||
shared = "--shared" in args
|
||||
args = [a for a in args if a != "--shared"]
|
||||
if len(args) < 2:
|
||||
print("Usage: cfg set [--shared] <KEY> <VALUE>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
key, value = args[0], args[1]
|
||||
scope = "shared" if shared else "personal"
|
||||
api("PUT", f"/config/{key}?scope={scope}", {"value": value})
|
||||
print(f"✓ Set {key} ({scope})")
|
||||
|
||||
|
||||
def cmd_delete(args: list[str]):
|
||||
shared = "--shared" in args
|
||||
args = [a for a in args if a != "--shared"]
|
||||
if not args:
|
||||
print("Usage: cfg delete <KEY> [--shared]", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
scope = "shared" if shared else "personal"
|
||||
api("DELETE", f"/config/{args[0]}?scope={scope}")
|
||||
print(f"✓ Deleted {args[0]} ({scope})")
|
||||
|
||||
|
||||
def cmd_list(args: list[str]):
|
||||
scope = None
|
||||
if "--scope" in args:
|
||||
idx = args.index("--scope")
|
||||
if idx + 1 < len(args):
|
||||
scope = args[idx + 1]
|
||||
path = "/config" + (f"?scope={scope}" if scope else "")
|
||||
result = api("GET", path)
|
||||
if "keys" in result:
|
||||
for k in sorted(result["keys"]):
|
||||
print(f" {k}")
|
||||
elif "secrets" in result:
|
||||
for k, v in sorted(result["secrets"].items()):
|
||||
print(f" {k:<40} ({v['scope']})")
|
||||
|
||||
|
||||
def cmd_sync(args: list[str]):
|
||||
result = api("POST", "/config/sync")
|
||||
# Save to cache
|
||||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cache = {
|
||||
"secrets": {k: {"value": v["value"], "updatedAt": v["updated_at"]} for k, v in result["secrets"].items()},
|
||||
"lastSync": result.get("agent_id", ""),
|
||||
}
|
||||
CACHE_FILE.write_text(json.dumps(cache, indent=2))
|
||||
print(f"✓ Synced {len(result['secrets'])} keys")
|
||||
|
||||
|
||||
def cmd_token(args: list[str]):
|
||||
if not args:
|
||||
print("Usage: cfg token <TOKEN>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
cfg = load_config()
|
||||
cfg["token"] = args[0]
|
||||
save_config(cfg)
|
||||
print("✓ Token saved")
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print(__doc__)
|
||||
sys.exit(0)
|
||||
|
||||
cmd = sys.argv[1]
|
||||
args = sys.argv[2:]
|
||||
|
||||
commands = {
|
||||
"get": cmd_get,
|
||||
"set": cmd_set,
|
||||
"delete": cmd_delete,
|
||||
"list": cmd_list,
|
||||
"sync": cmd_sync,
|
||||
"token": cmd_token,
|
||||
}
|
||||
|
||||
if cmd not in commands:
|
||||
print(f"Unknown command: {cmd}", file=sys.stderr)
|
||||
print(__doc__)
|
||||
sys.exit(1)
|
||||
|
||||
commands[cmd](args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
x
Reference in New Issue
Block a user