244 lines
7.8 KiB
Python
244 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Download a few raw Onshape feature-list responses for format inspection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import getpass
|
|
import json
|
|
import os
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
API_BASE = "https://cad.onshape.com"
|
|
DEFAULT_OUTPUT = Path("json_to_cdsl/input/onshape_api_samples")
|
|
DEFAULT_SAMPLES = (
|
|
(
|
|
"00000352",
|
|
"https://cad.onshape.com/documents/4185972a944744d8a7a0f2b4/"
|
|
"w/d82d7eef8edf4342b7e49732/e/b6d6b562e8b64e7ea50d8325",
|
|
),
|
|
(
|
|
"00000001",
|
|
"https://cad.onshape.com/documents/1ffb81a71e5b402e966b9341/"
|
|
"w/6e295017d1b34be684565c40/e/8df255ee6705423d8e85234e",
|
|
),
|
|
(
|
|
"00000002",
|
|
"https://cad.onshape.com/documents/1ffb81a71e5b402e966b9341/"
|
|
"w/6e295017d1b34be684565c40/e/bb398e4615fe4025b34ea8f0",
|
|
),
|
|
)
|
|
ONSHAPE_URL_RE = re.compile(
|
|
r"^https://(?P<stack>[^/]+)/documents/(?P<did>[^/]+)/"
|
|
r"(?P<wvm>w|v|m)/(?P<wvmid>[^/]+)/e/(?P<eid>[^/?#]+)"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PartStudioRef:
|
|
sample_id: str
|
|
source_url: str
|
|
stack: str
|
|
did: str
|
|
wvm: str
|
|
wvmid: str
|
|
eid: str
|
|
|
|
|
|
def parse_part_studio_url(sample_id: str, url: str) -> PartStudioRef:
|
|
match = ONSHAPE_URL_RE.match(url.strip())
|
|
if match is None:
|
|
raise ValueError(f"not an Onshape Part Studio URL: {url}")
|
|
return PartStudioRef(sample_id=sample_id, source_url=url.strip(), **match.groupdict())
|
|
|
|
|
|
def read_url_file(path: Path) -> list[tuple[str, str]]:
|
|
"""Read either ABC objects YAML lines or plain '<id> <url>' lines."""
|
|
records: list[tuple[str, str]] = []
|
|
url_pattern = re.compile(r"https://cad\.onshape\.com/documents/[^'\"\s]+")
|
|
id_pattern = re.compile(r"^\s*['\"]?(\d+)['\"]?\s*[:\s]")
|
|
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
url_match = url_pattern.search(line)
|
|
if url_match is None:
|
|
continue
|
|
id_match = id_pattern.search(line)
|
|
if id_match is None:
|
|
raise ValueError(f"{path}:{line_number}: URL has no leading sample ID")
|
|
records.append((id_match.group(1).zfill(8), url_match.group(0)))
|
|
if not records:
|
|
raise ValueError(f"no Onshape Part Studio URLs found in {path}")
|
|
return records
|
|
|
|
|
|
def credentials() -> tuple[str, str]:
|
|
access_key = os.environ.get("ONSHAPE_ACCESS_KEY")
|
|
secret_key = os.environ.get("ONSHAPE_SECRET_KEY")
|
|
if not access_key:
|
|
access_key = getpass.getpass("Onshape access key: ")
|
|
if not secret_key:
|
|
secret_key = getpass.getpass("Onshape secret key: ")
|
|
if not access_key or not secret_key:
|
|
raise ValueError("both Onshape access and secret keys are required")
|
|
return access_key, secret_key
|
|
|
|
|
|
def api_url(ref: PartStudioRef) -> str:
|
|
query = urllib.parse.urlencode(
|
|
{
|
|
"rollbackBarIndex": -1,
|
|
"includeGeometryIds": "true",
|
|
"noSketchGeometry": "false",
|
|
}
|
|
)
|
|
return (
|
|
f"https://{ref.stack}/api/v9/partstudios/d/{ref.did}/"
|
|
f"{ref.wvm}/{ref.wvmid}/e/{ref.eid}/features?{query}"
|
|
)
|
|
|
|
|
|
def fetch_json(url: str, authorization: str, timeout: float) -> Any:
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json;charset=UTF-8; qs=0.09",
|
|
"Authorization": authorization,
|
|
"User-Agent": "cdsl-cad-onshape-sample-downloader/1.0",
|
|
},
|
|
)
|
|
context = ssl.create_default_context(cafile=_ca_bundle())
|
|
with urllib.request.urlopen(request, timeout=timeout, context=context) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def _ca_bundle() -> str | None:
|
|
try:
|
|
import certifi
|
|
except ImportError:
|
|
return None
|
|
return certifi.where()
|
|
|
|
|
|
def write_json(path: Path, value: Any) -> None:
|
|
path.write_text(json.dumps(value, ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def download_one(
|
|
ref: PartStudioRef,
|
|
authorization: str,
|
|
output_dir: Path,
|
|
timeout: float,
|
|
) -> dict[str, Any]:
|
|
sample_dir = output_dir / ref.sample_id
|
|
sample_dir.mkdir(parents=True, exist_ok=True)
|
|
request_url = api_url(ref)
|
|
metadata = {
|
|
**asdict(ref),
|
|
"api_version": "v9",
|
|
"features_url": request_url,
|
|
}
|
|
write_json(sample_dir / "request.json", metadata)
|
|
try:
|
|
payload = fetch_json(request_url, authorization, timeout)
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
return {
|
|
"sample_id": ref.sample_id,
|
|
"status": "http_error",
|
|
"http_status": exc.code,
|
|
"message": body[:1000],
|
|
}
|
|
except (urllib.error.URLError, TimeoutError) as exc:
|
|
return {
|
|
"sample_id": ref.sample_id,
|
|
"status": "network_error",
|
|
"message": str(exc),
|
|
}
|
|
|
|
output_path = sample_dir / "features.json"
|
|
write_json(output_path, payload)
|
|
features = payload.get("features", []) if isinstance(payload, dict) else []
|
|
feature_types: dict[str, int] = {}
|
|
for feature in features:
|
|
if not isinstance(feature, dict):
|
|
continue
|
|
feature_type = str(feature.get("featureType", "unknown"))
|
|
feature_types[feature_type] = feature_types.get(feature_type, 0) + 1
|
|
return {
|
|
"sample_id": ref.sample_id,
|
|
"status": "downloaded",
|
|
"output": str(output_path),
|
|
"feature_count": len(features),
|
|
"feature_types": feature_types,
|
|
}
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--url",
|
|
action="append",
|
|
default=[],
|
|
metavar="ID=URL",
|
|
help="Onshape Part Studio URL with a stable local sample ID; repeatable",
|
|
)
|
|
parser.add_argument(
|
|
"--url-file",
|
|
type=Path,
|
|
help="ABC objects YAML or plain text file containing '<id> <url>' records",
|
|
)
|
|
parser.add_argument("--count", type=int, default=3, help="maximum samples to download")
|
|
parser.add_argument("--out", type=Path, default=DEFAULT_OUTPUT, help="output directory")
|
|
parser.add_argument("--timeout", type=float, default=60.0, help="request timeout in seconds")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
if args.count < 1:
|
|
raise ValueError("--count must be at least 1")
|
|
|
|
records: list[tuple[str, str]] = []
|
|
for value in args.url:
|
|
if "=" not in value:
|
|
raise ValueError("--url must use ID=URL syntax")
|
|
records.append(tuple(value.split("=", 1)))
|
|
if args.url_file:
|
|
records.extend(read_url_file(args.url_file))
|
|
if not records:
|
|
records.extend(DEFAULT_SAMPLES)
|
|
|
|
refs = [parse_part_studio_url(sample_id, url) for sample_id, url in records[: args.count]]
|
|
access_key, secret_key = credentials()
|
|
token = base64.b64encode(f"{access_key}:{secret_key}".encode("utf-8")).decode("ascii")
|
|
authorization = f"Basic {token}"
|
|
|
|
args.out.mkdir(parents=True, exist_ok=True)
|
|
results = [download_one(ref, authorization, args.out, args.timeout) for ref in refs]
|
|
manifest = {
|
|
"schema": "onshape.api.samples.v1",
|
|
"api_base": API_BASE,
|
|
"sample_count": len(results),
|
|
"results": results,
|
|
}
|
|
write_json(args.out / "manifest.json", manifest)
|
|
print(json.dumps(manifest, ensure_ascii=True, indent=2))
|
|
return 0 if all(item["status"] == "downloaded" for item in results) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (OSError, ValueError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
raise SystemExit(2)
|