#!/usr/bin/env python3 """Synal CLI: a dependency-free client for the Synal REST API (Python 3.9+).""" import argparse import getpass import json import os from pathlib import Path import sys import tempfile import urllib.error import urllib.parse import urllib.request BASE_URL = "https://app.synal.to/api/v1" VERSION = "1.0.0" class CLIError(Exception): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never forward a credential to another endpoint or replay a mutation. return None def config_path(): root = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) return root / "synal" / "config.json" def api_key(): key = os.environ.get("SYNAL_API_KEY", "").strip() if not key: try: config = json.loads(config_path().read_text()) key = config.get("api_key", "") if isinstance(config, dict) else "" except FileNotFoundError: pass except (ValueError, OSError) as error: raise CLIError("Cannot read config; run synal setup again.") from error if not isinstance(key, str) or not key.strip(): raise CLIError("Run synal setup or set SYNAL_API_KEY. An active yearly subscription is required.") if any(char.isspace() for char in key.strip()): raise CLIError("API key must not contain whitespace.") return key.strip() def request(method, path, key, params=None): url = BASE_URL + path data = None if method == "GET" and params: url += "?" + urllib.parse.urlencode(params) elif method == "POST": data = json.dumps(params or {}).encode() req = urllib.request.Request(url, data=data, method=method, headers={ "Authorization": "Bearer " + key, "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "synal-cli/" + VERSION, }) try: with urllib.request.build_opener(NoRedirect()).open(req, timeout=30) as response: return json.load(response) except urllib.error.HTTPError as error: messages = { 401: "Invalid API key. Run synal setup again or update SYNAL_API_KEY.", 403: "API access requires an active yearly subscription. Upgrade in the Synal app.", 404: "Not found. Check the supplied ID.", 422: "Request rejected. Check the discovery token or unlock channel.", 429: "Rate limit reached. Try again later.", } raise CLIError(messages.get(error.code, "API request failed.") + " (HTTP %s)" % error.code) from error except (urllib.error.URLError, TimeoutError, OSError) as error: raise CLIError("Could not reach the Synal API. Check your connection and try again.") from error except (ValueError, UnicodeError) as error: raise CLIError("The API returned invalid JSON.") from error def setup(): key = os.environ.get("SYNAL_API_KEY", "").strip() if not key: if not sys.stdin.isatty(): raise CLIError("Non-interactive setup requires SYNAL_API_KEY.") key = getpass.getpass("Synal API key (yearly subscription required): ").strip() if not key or any(char.isspace() for char in key): raise CLIError("Enter a valid API key without whitespace.") request("GET", "/funding_events", key, {"per_page": 1}) path = config_path() path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = None try: with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as file: temporary = file.name os.chmod(temporary, 0o600) json.dump({"api_key": key}, file) os.replace(temporary, path) finally: if temporary and os.path.exists(temporary): os.unlink(temporary) return {"configured": True, "config_path": str(path)} def positive_int(value): number = int(value) if number <= 0: raise argparse.ArgumentTypeError("must be a positive integer") return number def parser(): result = argparse.ArgumentParser(description="Synal REST API CLI for people and agents. API access requires a yearly subscription. JSON on stdout; errors on stderr.") result.add_argument("--version", action="version", version="synal " + VERSION) commands = result.add_subparsers(dest="command", required=True) commands.add_parser("setup", help="Validate and save an API key (hidden prompt or SYNAL_API_KEY)") events = commands.add_parser("events", help="List funding events, newest first") events.add_argument("--page", type=positive_int, default=1) events.add_argument("--per-page", type=positive_int, default=50, choices=range(1, 101), metavar="1-100") for name, help_text in [ ("event", "Get a funding event, company and contacts"), ("contacts", "List an event's contacts and discovery state"), ("discover", "Request contact discovery using its current token"), ("unlock", "Request a contact email or phone unlock"), ("unlock-status", "Get the current contact unlock result"), ]: command = commands.add_parser(name, help=help_text) command.add_argument("id", type=positive_int, help="Contact ID" if name.startswith("unlock") else "Funding event ID") if name == "discover": command.add_argument("--token", required=True, help="Current discovery token from the contacts response") if name == "unlock": command.add_argument("--channel", choices=["email", "phone"], default="email") return result def main(argv=None): args = parser().parse_args(argv) try: if args.command == "setup": output = setup() else: key = api_key() if args.command == "events": output = request("GET", "/funding_events", key, {"page": args.page, "per_page": args.per_page}) elif args.command == "event": output = request("GET", "/funding_events/%s" % args.id, key) elif args.command == "contacts": output = request("GET", "/funding_events/%s/contacts" % args.id, key) elif args.command == "discover": output = request("POST", "/funding_events/%s/contacts/discover" % args.id, key, {"token": args.token}) elif args.command == "unlock": output = request("POST", "/contacts/%s/unlock" % args.id, key, {"channel": args.channel}) else: output = request("GET", "/contacts/%s/unlock" % args.id, key) print(json.dumps(output, indent=2)) return 0 except (CLIError, OSError, EOFError) as error: message = str(error) if isinstance(error, CLIError) else "Could not read input or save configuration. Check file permissions and retry." print(json.dumps({"error": message}), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())