1#!/usr/bin/env python3
2"""Miro Board API helper for the marketplace skill bundle."""
3 
4import argparse
5import json
6import re
7import sys
8import urllib.request
9import urllib.parse
10import urllib.error
11from datetime import datetime, timedelta, timezone
12from http.server import HTTPServer, BaseHTTPRequestHandler
13from pathlib import Path
14 
15TOKEN_PATH = Path.home() / ".miro-token"
16API_BASE = "https://api.miro.com/v2"
17TOKEN_REFRESH_MARGIN_SECONDS = 300
18 
19 
20def load_token_state() -> dict | None:
21    if not TOKEN_PATH.exists():
22        return None
23 
24    raw = TOKEN_PATH.read_text().strip()
25    if not raw:
26        return None
27 
28    try:
29        data = json.loads(raw)
30    except json.JSONDecodeError:
31        return {"access_token": raw}
32 
33    if isinstance(data, dict):
34        return data
35 
36    return {"access_token": raw}
37 
38 
39def save_token_state(token_state: dict):
40    payload = dict(token_state)
41    payload.setdefault("saved_at", datetime.now(timezone.utc).isoformat())
42    TOKEN_PATH.write_text(json.dumps(payload, indent=2) + "\n")
43    TOKEN_PATH.chmod(0o600)
44    print(f"Token saved to {TOKEN_PATH}")
45 
46 
47def api_get(path: str, token: str) -> dict:
48    url = f"{API_BASE}{path}"
49    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
50    try:
51        with urllib.request.urlopen(req) as resp:
52            return json.loads(resp.read())
53    except urllib.error.HTTPError as e:
54        body = e.read().decode(errors="replace")
55        raise RuntimeError(f"Miro API request failed ({e.code}): {body}") from e
56    except urllib.error.URLError as e:
57        raise RuntimeError(f"Could not reach Miro API: {e.reason}") from e
58 
59 
60def exchange_oauth_token(payload: dict, action: str) -> dict:
61    data = urllib.parse.urlencode(payload).encode()
62    req = urllib.request.Request(
63        "https://api.miro.com/v1/oauth/token",
64        data=data,
65        headers={"Content-Type": "application/x-www-form-urlencoded"},
66    )
67 
68    try:
69        with urllib.request.urlopen(req) as resp:
70            return json.loads(resp.read())
71    except urllib.error.HTTPError as e:
72        body = e.read().decode(errors="replace")
73        raise RuntimeError(f"{action} failed ({e.code}): {body}") from e
74    except urllib.error.URLError as e:
75        raise RuntimeError(f"{action} failed: {e.reason}") from e
76 
77 
78def token_needs_refresh(token_state: dict) -> bool:
79    expires_in = token_state.get("expires_in")
80    saved_at = token_state.get("saved_at")
81    if not expires_in or not saved_at:
82        return False
83 
84    try:
85        saved_at_dt = datetime.fromisoformat(saved_at)
86        expires_in_seconds = int(expires_in)
87    except (TypeError, ValueError):
88        return False
89 
90    if saved_at_dt.tzinfo is None:
91        saved_at_dt = saved_at_dt.replace(tzinfo=timezone.utc)
92 
93    refresh_at = saved_at_dt + timedelta(
94        seconds=max(0, expires_in_seconds - TOKEN_REFRESH_MARGIN_SECONDS)
95    )
96    return datetime.now(timezone.utc) >= refresh_at
97 
98 
99def refresh_access_token(token_state: dict) -> dict:
100    refresh_token = token_state.get("refresh_token")
101    client_id = token_state.get("client_id")
102    client_secret = token_state.get("client_secret")
103    if not refresh_token or not client_id or not client_secret:
104        raise RuntimeError("Stored token expired and cannot be refreshed. Run 'oauth' again.")
105 
106    refreshed = exchange_oauth_token(
107        {
108            "grant_type": "refresh_token",
109            "client_id": client_id,
110            "client_secret": client_secret,
111            "refresh_token": refresh_token,
112        },
113        "Token refresh",
114    )
115    refreshed["client_id"] = client_id
116    refreshed["client_secret"] = client_secret
117    refreshed.setdefault("refresh_token", refresh_token)
118    refreshed["saved_at"] = datetime.now(timezone.utc).isoformat()
119    save_token_state(refreshed)
120    return refreshed
121 
122 
123def get_token_state() -> dict | None:
124    token_state = load_token_state()
125    if not token_state or not token_state.get("access_token"):
126        return None
127 
128    if token_needs_refresh(token_state):
129        print("Access token expired or expiring soon. Refreshing...")
130        token_state = refresh_access_token(token_state)
131 
132    return token_state
133 
134 
135def download_url(url: str, headers: dict | None = None) -> tuple[bytes, str, str]:
136    req = urllib.request.Request(url, headers=headers or {})
137    try:
138        with urllib.request.urlopen(req) as resp:
139            return resp.read(), resp.geturl(), resp.headers.get_content_type()
140    except urllib.error.HTTPError as e:
141        body = e.read().decode(errors="replace")
142        raise RuntimeError(f"Download failed ({e.code}): {body}") from e
143    except urllib.error.URLError as e:
144        raise RuntimeError(f"Download failed: {e.reason}") from e
145 
146 
147def detect_extension(content_type: str, url: str) -> str:
148    content_type = (content_type or "").split(";", 1)[0].strip().lower()
149    if content_type == "image/jpeg":
150        return ".jpg"
151    if content_type == "image/png":
152        return ".png"
153    if content_type == "image/gif":
154        return ".gif"
155    if content_type == "image/webp":
156        return ".webp"
157    if content_type == "image/svg+xml":
158        return ".svg"
159 
160    suffix = Path(urllib.parse.urlparse(url).path).suffix.lower()
161    if suffix in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"}:
162        return suffix
163 
164    return ".png"
165 
166 
167def extract_board_id(board_url: str) -> str:
168    """Extract board ID from Miro board URL."""
169    # https://miro.com/app/board/uXjVKiV7tuU=/
170    m = re.search(r"/board/([^/]+)/?", board_url)
171    if m:
172        return m.group(1)
173    return board_url  # assume it's already a board ID
174 
175 
176def api_post(path: str, token: str, payload: dict) -> dict:
177    url = f"{API_BASE}{path}"
178    body = json.dumps(payload).encode()
179    req = urllib.request.Request(
180        url, data=body,
181        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
182    )
183    try:
184        with urllib.request.urlopen(req) as resp:
185            return json.loads(resp.read())
186    except urllib.error.HTTPError as e:
187        body_text = e.read().decode(errors="replace")
188        raise RuntimeError(f"Miro API POST failed ({e.code}): {body_text}") from e
189    except urllib.error.URLError as e:
190        raise RuntimeError(f"Could not reach Miro API: {e.reason}") from e
191 
192 
193def strip_html(html: str) -> str:
194    return re.sub(r"<[^>]+>", "", html).strip()
195 
196 
197# --- Commands ---
198 
199 
200def cmd_test_token(args):
201    token_state = get_token_state()
202    if not token_state:
203        print("ERROR: No token found. Run 'oauth' first.")
204        sys.exit(1)
205 
206    token = token_state["access_token"]
207    try:
208        # Try listing boards to validate
209        req = urllib.request.Request(
210            f"{API_BASE}/boards?limit=1",
211            headers={"Authorization": f"Bearer {token}"},
212        )
213        with urllib.request.urlopen(req) as resp:
214            data = json.loads(resp.read())
215        print(f"OK: Token valid. You have access to {data.get('total', '?')} boards.")
216    except urllib.error.HTTPError as e:
217        print(f"ERROR: Token invalid ({e.code}). Run 'oauth' to refresh.")
218        sys.exit(1)
219    except urllib.error.URLError as e:
220        print(f"ERROR: Could not reach Miro API: {e.reason}")
221        sys.exit(1)
222 
223 
224def cmd_oauth(args):
225    client_id = args.client_id
226    client_secret = args.client_secret
227    redirect_uri = "http://localhost:9876/callback"
228    captured_code = {}
229 
230    class Handler(BaseHTTPRequestHandler):
231        def do_GET(self):
232            from urllib.parse import urlparse, parse_qs
233 
234            parsed = urlparse(self.path)
235            if parsed.path == "/callback":
236                params = parse_qs(parsed.query)
237                code = params.get("code", [None])[0]
238                if code:
239                    captured_code["code"] = code
240                    self.send_response(200)
241                    self.send_header("Content-Type", "text/html")
242                    self.end_headers()
243                    self.wfile.write(
244                        b"<h1>Done! You can close this tab.</h1>"
245                        b"<p>Return to Claude Code.</p>"
246                    )
247                else:
248                    self.send_response(400)
249                    self.end_headers()
250                    self.wfile.write(b"No authorization code received.")
251            else:
252                self.send_response(404)
253                self.end_headers()
254 
255        def log_message(self, format, *args):
256            pass
257 
258    # Start server
259    server = HTTPServer(("localhost", 9876), Handler)
260 
261    auth_url = (
262        f"https://miro.com/oauth/authorize"
263        f"?response_type=code"
264        f"&client_id={client_id}"
265        f"&redirect_uri={urllib.parse.quote(redirect_uri)}"
266    )
267 
268    print(f"\nOpen this URL in your browser to authorize:\n\n{auth_url}\n")
269    print("Waiting for authorization...")
270 
271    # Handle one request (the callback)
272    server.handle_request()
273 
274    if "code" not in captured_code:
275        print("ERROR: No authorization code received.")
276        sys.exit(1)
277 
278    code = captured_code["code"]
279    print(f"Authorization code received. Exchanging for token...")
280 
281    token_data = exchange_oauth_token(
282        {
283            "grant_type": "authorization_code",
284            "client_id": client_id,
285            "client_secret": client_secret,
286            "code": code,
287            "redirect_uri": redirect_uri,
288        },
289        "Token exchange",
290    )
291 
292    access_token = token_data.get("access_token")
293    if not access_token:
294        print(f"ERROR: No access_token in response: {json.dumps(token_data)}")
295        sys.exit(1)
296 
297    token_data["client_id"] = client_id
298    token_data["client_secret"] = client_secret
299    token_data["saved_at"] = datetime.now(timezone.utc).isoformat()
300    save_token_state(token_data)
301    print(f"Authenticated as user {token_data.get('user_id', '?')}")
302    print(f"Scope: {token_data.get('scope', '?')}")
303    if token_data.get("refresh_token"):
304        print("Refresh token saved. Future runs will auto-refresh the access token.")
305 
306 
307def cmd_board_info(args):
308    token_state = get_token_state()
309    if not token_state:
310        print("ERROR: No token. Run 'oauth' first.")
311        sys.exit(1)
312 
313    token = token_state["access_token"]
314    board_id = extract_board_id(args.board_url)
315    data = api_get(f"/boards/{board_id}", token)
316 
317    print(f"Board: {data.get('name', 'N/A')}")
318    print(f"Description: {data.get('description', 'N/A')}")
319    print(f"Created: {data.get('createdAt', 'N/A')}")
320    print(f"Modified: {data.get('modifiedAt', 'N/A')}")
321 
322 
323def cmd_list_items(args):
324    token_state = get_token_state()
325    if not token_state:
326        print("ERROR: No token. Run 'oauth' first.")
327        sys.exit(1)
328 
329    token = token_state["access_token"]
330    board_id = extract_board_id(args.board_url)
331 
332    all_items = []
333    cursor = None
334    while True:
335        path = f"/boards/{board_id}/items?limit=50"
336        if cursor:
337            path += f"&cursor={cursor}"
338        data = api_get(path, token)
339        all_items.extend(data.get("data", []))
340        cursor = data.get("cursor")
341        if not cursor:
342            break
343 
344    print(f"Total items: {len(all_items)}\n")
345 
346    # Group by type
347    types = {}
348    for item in all_items:
349        t = item.get("type", "?")
350        types[t] = types.get(t, 0) + 1
351    print("Item types:")
352    for t, count in sorted(types.items()):
353        print(f"  {t}: {count}")
354    print()
355 
356    # Print text items (section headers)
357    texts = [i for i in all_items if i["type"] == "text"]
358    if texts:
359        print("TEXT ITEMS (section headers):")
360        for t in sorted(texts, key=lambda x: (x["position"]["x"], x["position"]["y"])):
361            content = strip_html(t.get("data", {}).get("content", ""))
362            pos = t["position"]
363            print(f"  pos=({pos['x']:.0f},{pos['y']:.0f}) \"{content[:120]}\"")
364        print()
365 
366    # Print shapes with content
367    if args.verbose:
368        shapes = [i for i in all_items if i["type"] == "shape"]
369        if shapes:
370            print("SHAPES:")
371            for s in sorted(
372                shapes, key=lambda x: (x["position"]["x"], x["position"]["y"])
373            ):
374                content = strip_html(s.get("data", {}).get("content", ""))
375                if content:
376                    pos = s["position"]
377                    geo = s.get("geometry", {})
378                    print(
379                        f"  pos=({pos['x']:.0f},{pos['y']:.0f}) "
380                        f"size=({geo.get('width', 0):.0f}x{geo.get('height', 0):.0f}) "
381                        f"\"{content[:120]}\""
382                    )
383            print()
384 
385    # Print image positions
386    images = [i for i in all_items if i["type"] == "image"]
387    if images:
388        print(f"IMAGES ({len(images)}):")
389        for img in sorted(
390            images, key=lambda x: (x["position"]["x"], x["position"]["y"])
391        ):
392            pos = img["position"]
393            geo = img.get("geometry", {})
394            print(
395                f"  id={img['id']} "
396                f"pos=({pos['x']:.0f},{pos['y']:.0f}) "
397                f"size=({geo.get('width', 0):.0f}x{geo.get('height', 0):.0f})"
398            )
399 
400 
401def cmd_download_images(args):
402    token_state = get_token_state()
403    if not token_state:
404        print("ERROR: No token. Run 'oauth' first.")
405        sys.exit(1)
406 
407    token = token_state["access_token"]
408    board_id = extract_board_id(args.board_url)
409    output_dir = Path(args.output_dir)
410    output_dir.mkdir(parents=True, exist_ok=True)
411    fmt = args.format  # "original" or "preview"
412 
413    # Parse region if provided
414    region = None
415    if args.region:
416        parts = [float(x) for x in args.region.split(",")]
417        if len(parts) == 4:
418            region = {
419                "x_min": parts[0],
420                "y_min": parts[1],
421                "x_max": parts[2],
422                "y_max": parts[3],
423            }
424 
425    # Fetch all items
426    all_items = []
427    cursor = None
428    while True:
429        path = f"/boards/{board_id}/items?limit=50"
430        if cursor:
431            path += f"&cursor={cursor}"
432        data = api_get(path, token)
433        all_items.extend(data.get("data", []))
434        cursor = data.get("cursor")
435        if not cursor:
436            break
437 
438    images = [i for i in all_items if i["type"] == "image"]
439 
440    # Filter by region
441    if region:
442        images = [
443            i
444            for i in images
445            if region["x_min"] <= i["position"]["x"] <= region["x_max"]
446            and region["y_min"] <= i["position"]["y"] <= region["y_max"]
447        ]
448 
449    print(f"Found {len(images)} images to download")
450 
451    for idx, img in enumerate(
452        sorted(images, key=lambda x: (x["position"]["x"], x["position"]["y"]))
453    ):
454        pos = img["position"]
455        geo = img.get("geometry", {})
456        img_id = img["id"]
457 
458        # Get item detail for image URL
459        item_data = api_get(f"/boards/{board_id}/items/{img_id}", token)
460        image_url = item_data.get("data", {}).get("imageUrl", "")
461        if not image_url:
462            print(f"  [{idx}] No imageUrl — skipping")
463            continue
464 
465        # Request original or preview format
466        if fmt == "original":
467            image_url = image_url.replace("format=preview", "format=original")
468 
469        # Get signed URL
470        content, final_url, content_type = download_url(
471            image_url, headers={"Authorization": f"Bearer {token}"}
472        )
473 
474        # Some signed image endpoints return JSON with a temporary redirect URL.
475        if content_type == "application/json":
476            redirect_data = json.loads(content)
477            actual_url = redirect_data.get("url", "")
478            if actual_url:
479                content, final_url, content_type = download_url(actual_url)
480 
481        extension = detect_extension(content_type, final_url)
482        fname = output_dir / f"img_{idx:03d}_{pos['x']:.0f}_{pos['y']:.0f}{extension}"
483        fname.write_bytes(content)
484        fsize = len(content)
485        print(
486