first commit
This commit is contained in:
David Rice
248
nexarxclaude.py
Normal file
248
nexarxclaude.py
Normal file
@@ -0,0 +1,248 @@
|
||||
"""
|
||||
List all projects in Altium 365 workspaces via the Nexar GraphQL API.
|
||||
|
||||
Requirements:
|
||||
pip install requests
|
||||
|
||||
Environment variables:
|
||||
NEXAR_CLIENT_ID - Your Nexar application client ID
|
||||
NEXAR_CLIENT_SECRET - Your Nexar application client secret
|
||||
|
||||
Your Nexar application must have the 'design' scope enabled.
|
||||
Authentication opens a browser window for Altium 365 login.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import http.server
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import webbrowser
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import requests
|
||||
|
||||
TOKEN_URL = "https://identity.nexar.com/connect/token"
|
||||
AUTH_URL = "https://identity.nexar.com/connect/authorize"
|
||||
NEXAR_API_URL = "https://api.nexar.com/graphql"
|
||||
REDIRECT_URI = "http://localhost:3000/login"
|
||||
SCOPES = "openid profile email design.domain user.access"
|
||||
|
||||
WORKSPACE_NAME = "Flowbird SAS"
|
||||
|
||||
QUERY_WORKSPACES_AND_PROJECTS = """
|
||||
query GetWorkspacesAndProjects($workspace: String!) {
|
||||
desWorkspaces(where: {name: {eq: $workspace}}) {
|
||||
url
|
||||
name
|
||||
description
|
||||
projects {
|
||||
id
|
||||
name
|
||||
description
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
QUERY_VARIANTS = """
|
||||
query GetVariants($projectId: ID!) {
|
||||
desProjectById(id: $projectId) {
|
||||
name
|
||||
design {
|
||||
workInProgress {
|
||||
variants {
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
_CALLBACK_HTML = """
|
||||
<html><head><title>Nexar Login</title></head>
|
||||
<body style="background:#000b24;color:#fff;font-family:sans-serif;text-align:center;padding-top:20%">
|
||||
<h1>Login successful</h1><p>You can close this tab and return to the terminal.</p>
|
||||
</body></html>
|
||||
"""
|
||||
|
||||
|
||||
def _make_callback_handler(code_bucket):
|
||||
"""Return an HTTPRequestHandler that captures the OAuth2 auth code."""
|
||||
class _Handler(http.server.BaseHTTPRequestHandler):
|
||||
def log_message(self, *_):
|
||||
pass # silence request logs
|
||||
|
||||
def do_GET(self):
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path != "/login":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
params = parse_qs(parsed.query)
|
||||
self.send_response(200)
|
||||
self.send_header("Content-type", "text/html")
|
||||
self.end_headers()
|
||||
self.wfile.write(_CALLBACK_HTML.encode())
|
||||
code_bucket.append(params.get("code", [""])[0])
|
||||
|
||||
return _Handler
|
||||
|
||||
|
||||
def get_token(client_id: str, client_secret: str) -> str:
|
||||
"""Obtain a Nexar access token via Authorization Code + PKCE flow."""
|
||||
# Generate PKCE verifier and challenge
|
||||
code_verifier = base64.urlsafe_b64encode(os.urandom(40)).decode()
|
||||
code_verifier = re.sub(r"[^a-zA-Z0-9]+", "", code_verifier)
|
||||
|
||||
digest = hashlib.sha256(code_verifier.encode()).digest()
|
||||
code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")
|
||||
|
||||
# Build authorization URL
|
||||
auth_params = (
|
||||
f"?response_type=code"
|
||||
f"&client_id={client_id}"
|
||||
f"&redirect_uri={REDIRECT_URI}"
|
||||
f"&scope={SCOPES.replace(' ', '%20')}"
|
||||
f"&code_challenge={code_challenge}"
|
||||
f"&code_challenge_method=S256"
|
||||
)
|
||||
auth_url = AUTH_URL + auth_params
|
||||
|
||||
# Start local callback server and open browser
|
||||
code_bucket = []
|
||||
server = http.server.HTTPServer(("localhost", 3000), _make_callback_handler(code_bucket))
|
||||
|
||||
print("Opening browser for Nexar login...")
|
||||
webbrowser.open(auth_url)
|
||||
|
||||
while not code_bucket:
|
||||
server.handle_request()
|
||||
server.server_close()
|
||||
|
||||
auth_code = code_bucket[0]
|
||||
if not auth_code:
|
||||
print("Error: no authorization code received.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Exchange auth code for token
|
||||
response = requests.post(
|
||||
TOKEN_URL,
|
||||
data={
|
||||
"grant_type": "authorization_code",
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"redirect_uri": REDIRECT_URI,
|
||||
"code": auth_code,
|
||||
"code_verifier": code_verifier,
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()["access_token"]
|
||||
|
||||
|
||||
def graphql(token: str, url: str, query: str, variables: dict = None) -> dict:
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
payload = {"query": query}
|
||||
if variables:
|
||||
payload["variables"] = variables
|
||||
|
||||
response = requests.post(url, json=payload, headers=headers, timeout=30)
|
||||
if not response.ok:
|
||||
print(f"HTTP {response.status_code}: {response.text}", file=sys.stderr)
|
||||
response.raise_for_status()
|
||||
|
||||
result = response.json()
|
||||
if "errors" in result:
|
||||
for err in result["errors"]:
|
||||
print(f"GraphQL error: {err.get('message')}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
return result["data"]
|
||||
|
||||
|
||||
def main():
|
||||
client_id = os.environ.get("NEXAR_CLIENT_ID")
|
||||
client_secret = os.environ.get("NEXAR_CLIENT_SECRET")
|
||||
|
||||
if not client_id or not client_secret:
|
||||
print(
|
||||
"Error: NEXAR_CLIENT_ID and NEXAR_CLIENT_SECRET must be set.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
token = get_token(client_id, client_secret)
|
||||
print("Authenticated.\n")
|
||||
|
||||
print(f"Fetching projects from workspace '{WORKSPACE_NAME}'...")
|
||||
data = graphql(token, NEXAR_API_URL, QUERY_WORKSPACES_AND_PROJECTS, {"workspace": WORKSPACE_NAME})
|
||||
workspaces = data["desWorkspaces"]
|
||||
|
||||
if not workspaces:
|
||||
print(
|
||||
"No workspaces found. Ensure the 'design' scope is enabled on your "
|
||||
"Nexar application and you are a member of an Altium 365 workspace."
|
||||
)
|
||||
return
|
||||
|
||||
projects = workspaces[0].get("projects") or []
|
||||
|
||||
if not projects:
|
||||
print("No projects found in this workspace.")
|
||||
return
|
||||
|
||||
print(f"\nProjects in '{WORKSPACE_NAME}':")
|
||||
print("-" * 60)
|
||||
for i, project in enumerate(projects, start=1):
|
||||
description = project.get("description") or ""
|
||||
desc_str = f" — {description}" if description else ""
|
||||
print(f" [{i}] {project['name']}{desc_str}")
|
||||
|
||||
print()
|
||||
while True:
|
||||
try:
|
||||
choice = int(input(f"Select a project (1-{len(projects)}): "))
|
||||
if 1 <= choice <= len(projects):
|
||||
break
|
||||
print(f" Please enter a number between 1 and {len(projects)}.")
|
||||
except ValueError:
|
||||
print(" Please enter a valid number.")
|
||||
|
||||
selected_project = projects[choice - 1]
|
||||
print(f"\nSelected project: {selected_project['name']}")
|
||||
|
||||
# Fetch variants for the selected project
|
||||
print("\nFetching variants...")
|
||||
data = graphql(token, NEXAR_API_URL, QUERY_VARIANTS, {"projectId": selected_project["id"]})
|
||||
wip = data["desProjectById"]["design"]["workInProgress"]
|
||||
variants = wip.get("variants") or []
|
||||
|
||||
if not variants:
|
||||
print("No variants found for this project.")
|
||||
return
|
||||
|
||||
print(f"\nVariants in '{selected_project['name']}':")
|
||||
print("-" * 60)
|
||||
for i, variant in enumerate(variants, start=1):
|
||||
print(f" [{i}] {variant['name']}")
|
||||
|
||||
print()
|
||||
while True:
|
||||
try:
|
||||
choice = int(input(f"Select a variant (1-{len(variants)}): "))
|
||||
if 1 <= choice <= len(variants):
|
||||
break
|
||||
print(f" Please enter a number between 1 and {len(variants)}.")
|
||||
except ValueError:
|
||||
print(" Please enter a valid number.")
|
||||
|
||||
selected_variant = variants[choice - 1]
|
||||
print(f"\nSelected variant: {selected_variant['name']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user