My custom setup for picoclaw

docker/Dockerfile.heavy

@@ -73,12 +73,25 @@ RUN curl -LsSf https://astral.sh/uv/install.sh | sh && \
   uv --version
 
 # Install Python packages (system-wide via uv to avoid PEP 668 issues)
-RUN uv pip install --system --break-system-packages atproto
+# atproto: Bluesky AT Protocol SDK
+# moviepy: Programmatic video editing (timed text overlays, compositing)
+# librosa: Audio analysis (beat detection, loudness, onset detection)
+# pydub: Simple audio slicing/mixing
+# srt: SRT subtitle generation for ffmpeg lyric burn-in
+RUN uv pip install --system --break-system-packages atproto moviepy librosa pydub srt
 
 # Install bsky CLI (Python wrapper around atproto SDK)
 COPY docker/bsky.py /usr/local/bin/bsky
 RUN sed -i 's/\r$//' /usr/local/bin/bsky && chmod +x /usr/local/bin/bsky
 
+# Install threads CLI (Threads API wrapper)
+COPY docker/threads.py /usr/local/bin/threads
+RUN sed -i 's/\r$//' /usr/local/bin/threads && chmod +x /usr/local/bin/threads
+
+# Install instagram CLI (Instagram Graph API wrapper)
+COPY docker/instagram.py /usr/local/bin/instagram
+RUN sed -i 's/\r$//' /usr/local/bin/instagram && chmod +x /usr/local/bin/instagram
+
 # Health check
 HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
   CMD wget -q --spider http://localhost:18790/health || exit 1

docker/bsky.py

@@ -10,11 +10,15 @@ Config is stored at $XDG_CONFIG_HOME/bsky/config.json (defaults to ~/.config/bsk
 import argparse
 import json
 import os
+import re
 import sys
 import textwrap
 from pathlib import Path
 
-from atproto import Client, models
+from atproto import Client, client_utils, models
+
+# URL regex for detecting links in post text
+_URL_RE = re.compile(r'https?://[^\s)>"]+')
 
 # ---------------------------------------------------------------------------
 # Config helpers
@@ -190,10 +194,27 @@ def cmd_post(args) -> None:
     if images:
         embed = models.AppBskyEmbedImages.Main(images=images)
 
-    resp = client.send_post(text=text, embed=embed)
+    # Build rich text with clickable links
+    tb = _build_rich_text(text)
+    resp = client.send_post(text=tb, embed=embed)
     print(f"Posted: {resp.uri}")
 
 
+def _build_rich_text(text: str) -> client_utils.TextBuilder:
+    """Parse URLs in text and return a TextBuilder with link facets."""
+    tb = client_utils.TextBuilder()
+    last_end = 0
+    for m in _URL_RE.finditer(text):
+        if m.start() > last_end:
+            tb.text(text[last_end:m.start()])
+        url = m.group(0)
+        tb.link(url, url)
+        last_end = m.end()
+    if last_end < len(text):
+        tb.text(text[last_end:])
+    return tb
+
+
 def cmd_create_thread(args) -> None:
     client = _client()
     texts = args.texts

docker/docker-compose.yml

@@ -43,6 +43,9 @@ services:
       - DISPLAY=:99
       - BSKY_HANDLE=${BSKY_HANDLE}
       - BSKY_APP_PASSWORD=${BSKY_APP_PASSWORD}
+      - THREADS_ACCESS_TOKEN=${THREADS_ACCESS_TOKEN}
+      - INSTAGRAM_ACCESS_TOKEN=${INSTAGRAM_ACCESS_TOKEN}
+      - INSTAGRAM_USER_ID=${INSTAGRAM_USER_ID}
     ports:
       - "127.0.0.1:18790:18790"
       - "192.168.1.169:18790:18790"

docker/instagram.py

@@ -0,0 +1,449 @@
+#!/usr/bin/env python3
+"""instagram — Instagram CLI for picoclaw.
+
+Thin wrapper around the Instagram Graph API (graph.facebook.com).
+Installed to /usr/local/bin/instagram inside the Docker image
+so the agent can call `instagram post`, `instagram whoami`, etc.
+
+Auth: Set INSTAGRAM_ACCESS_TOKEN env var or pass --token.
+User ID: Set INSTAGRAM_USER_ID env var or pass --user-id.
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+import urllib.request
+import urllib.parse
+import urllib.error
+
+API_BASE = "https://graph.instagram.com/v21.0"
+
+
+# ---------------------------------------------------------------------------
+# HTTP helpers (stdlib only — no extra deps needed)
+# ---------------------------------------------------------------------------
+
+def _get_token(args) -> str:
+    token = getattr(args, "token", None) or os.environ.get("INSTAGRAM_ACCESS_TOKEN")
+    if not token:
+        print("No access token. Set INSTAGRAM_ACCESS_TOKEN or pass --token.", file=sys.stderr)
+        sys.exit(1)
+    return token
+
+
+def _get_user_id(args) -> str:
+    uid = getattr(args, "user_id", None) or os.environ.get("INSTAGRAM_USER_ID")
+    if not uid:
+        print("No user ID. Set INSTAGRAM_USER_ID or pass --user-id.", file=sys.stderr)
+        sys.exit(1)
+    return uid
+
+
+def _api_get(path: str, params: dict) -> dict:
+    qs = urllib.parse.urlencode(params)
+    url = f"{API_BASE}/{path}?{qs}"
+    req = urllib.request.Request(url, method="GET")
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode()
+        print(f"API error {e.code}: {body}", file=sys.stderr)
+        sys.exit(1)
+
+
+def _api_post(path: str, params: dict) -> dict:
+    url = f"{API_BASE}/{path}"
+    data = urllib.parse.urlencode(params).encode()
+    req = urllib.request.Request(url, data=data, method="POST")
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode()
+        print(f"API error {e.code}: {body}", file=sys.stderr)
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Commands
+# ---------------------------------------------------------------------------
+
+def cmd_whoami(args) -> None:
+    token = _get_token(args)
+    uid = _get_user_id(args)
+    data = _api_get(uid, {
+        "fields": "id,username,name,biography,followers_count,media_count",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"User ID:    {data.get('id', '?')}")
+        print(f"Username:   @{data.get('username', '?')}")
+        if data.get("name"):
+            print(f"Name:       {data['name']}")
+        if data.get("biography"):
+            print(f"Bio:        {data['biography']}")
+        if data.get("followers_count") is not None:
+            print(f"Followers:  {data['followers_count']}")
+        if data.get("media_count") is not None:
+            print(f"Posts:      {data['media_count']}")
+
+
+def cmd_post_image(args) -> None:
+    """Post a single image (Instagram requires an image — no text-only posts)."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post image:")
+        print(f"  Caption: {args.caption}")
+        print(f"  Image:   {args.image_url}")
+        return
+
+    # Step 1: Create media container
+    create_params = {
+        "image_url": args.image_url,
+        "caption": args.caption,
+        "access_token": token,
+    }
+    result = _api_post(f"{uid}/media", create_params)
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    # Step 2: Wait for container to be ready
+    _wait_for_container(token, container_id)
+
+    # Step 3: Publish
+    publish_result = _api_post(f"{uid}/media_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    post_id = publish_result.get("id")
+    print(f"Posted: {post_id}")
+
+
+def cmd_post_video(args) -> None:
+    """Post a reel (video). Instagram treats all video uploads as Reels."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post reel:")
+        print(f"  Caption: {args.caption}")
+        print(f"  Video:   {args.video_url}")
+        return
+
+    # Step 1: Create media container (media_type=REELS for video)
+    create_params = {
+        "media_type": "REELS",
+        "video_url": args.video_url,
+        "caption": args.caption,
+        "access_token": token,
+    }
+    result = _api_post(f"{uid}/media", create_params)
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    # Step 2: Wait for video processing (can take a while)
+    _wait_for_container(token, container_id, max_wait=120)
+
+    # Step 3: Publish
+    publish_result = _api_post(f"{uid}/media_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    post_id = publish_result.get("id")
+    print(f"Posted: {post_id}")
+
+
+def cmd_post_carousel(args) -> None:
+    """Post a carousel (multiple images/videos)."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post carousel:")
+        print(f"  Caption: {args.caption}")
+        for i, url in enumerate(args.urls):
+            print(f"  Item {i+1}: {url}")
+        return
+
+    # Step 1: Create individual item containers
+    children_ids = []
+    for url in args.urls:
+        # Detect if video (simple heuristic: extension)
+        is_video = any(url.lower().endswith(ext) for ext in (".mp4", ".mov", ".avi"))
+        if is_video:
+            params = {
+                "media_type": "VIDEO",
+                "video_url": url,
+                "is_carousel_item": "true",
+                "access_token": token,
+            }
+        else:
+            params = {
+                "image_url": url,
+                "is_carousel_item": "true",
+                "access_token": token,
+            }
+        result = _api_post(f"{uid}/media", params)
+        child_id = result.get("id")
+        if not child_id:
+            print(f"Failed to create carousel item: {result}", file=sys.stderr)
+            sys.exit(1)
+        _wait_for_container(token, child_id, max_wait=120 if is_video else 30)
+        children_ids.append(child_id)
+
+    # Step 2: Create carousel container
+    result = _api_post(f"{uid}/media", {
+        "media_type": "CAROUSEL",
+        "caption": args.caption,
+        "children": ",".join(children_ids),
+        "access_token": token,
+    })
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create carousel container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    _wait_for_container(token, container_id)
+
+    # Step 3: Publish
+    publish_result = _api_post(f"{uid}/media_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    print(f"Posted carousel: {publish_result.get('id')}")
+
+
+def cmd_post_story(args) -> None:
+    """Post a story (image or video)."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post story:")
+        print(f"  Media: {args.media_url}")
+        return
+
+    is_video = any(args.media_url.lower().endswith(ext) for ext in (".mp4", ".mov", ".avi"))
+    create_params = {
+        "media_type": "STORIES",
+        "access_token": token,
+    }
+    if is_video:
+        create_params["video_url"] = args.media_url
+    else:
+        create_params["image_url"] = args.media_url
+
+    result = _api_post(f"{uid}/media", create_params)
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create story container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    _wait_for_container(token, container_id, max_wait=120 if is_video else 30)
+
+    publish_result = _api_post(f"{uid}/media_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    print(f"Story posted: {publish_result.get('id')}")
+
+
+def cmd_reply(args) -> None:
+    """Reply to a comment on a post."""
+    token = _get_token(args)
+    result = _api_post(f"{args.comment_id}/replies", {
+        "message": args.text,
+        "access_token": token,
+    })
+    print(f"Replied: {result.get('id')}")
+
+
+def cmd_posts(args) -> None:
+    """List recent posts."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+    data = _api_get(f"{uid}/media", {
+        "fields": "id,caption,timestamp,media_type,permalink,like_count,comments_count",
+        "limit": args.n,
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        for post in data.get("data", []):
+            ts = post.get("timestamp", "")
+            caption = post.get("caption", "(no caption)")
+            pid = post.get("id", "?")
+            mtype = post.get("media_type", "?")
+            likes = post.get("like_count", 0)
+            comments = post.get("comments_count", 0)
+            permalink = post.get("permalink", "")
+            print(f"  [{ts}] {pid} ({mtype})")
+            for line in caption.splitlines()[:3]:
+                print(f"    {line}")
+            print(f"    Likes: {likes}  Comments: {comments}")
+            if permalink:
+                print(f"    {permalink}")
+            print()
+
+
+def cmd_insights(args) -> None:
+    """Get insights for a specific post."""
+    token = _get_token(args)
+    data = _api_get(f"{args.post_id}/insights", {
+        "metric": "impressions,reach,likes,comments,shares,saved",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"Insights for {args.post_id}:")
+        for item in data.get("data", []):
+            name = item.get("name", "?")
+            values = item.get("values", [{}])
+            val = values[0].get("value", "?") if values else "?"
+            print(f"  {name}: {val}")
+
+
+def cmd_profile(args) -> None:
+    """View profile details."""
+    token = _get_token(args)
+    uid = _get_user_id(args)
+    data = _api_get(uid, {
+        "fields": "id,username,name,biography,followers_count,follows_count,media_count,profile_picture_url,website",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"@{data.get('username', '?')}")
+        if data.get("name"):
+            print(f"  Name:       {data['name']}")
+        if data.get("biography"):
+            print(f"  Bio:        {data['biography']}")
+        print(f"  Followers:  {data.get('followers_count', '?')}")
+        print(f"  Following:  {data.get('follows_count', '?')}")
+        print(f"  Posts:      {data.get('media_count', '?')}")
+        if data.get("website"):
+            print(f"  Website:    {data['website']}")
+        print(f"  ID:         {data.get('id', '?')}")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _wait_for_container(token: str, container_id: str, max_wait: int = 30) -> None:
+    """Poll container status until FINISHED or timeout."""
+    s = None
+    for _ in range(max_wait):
+        status = _api_get(container_id, {
+            "fields": "status_code",
+            "access_token": token,
+        })
+        s = status.get("status_code")
+        if s == "FINISHED":
+            return
+        if s == "ERROR":
+            # Try to get error details
+            err_info = _api_get(container_id, {
+                "fields": "status_code,status",
+                "access_token": token,
+            })
+            print(f"Container error: {err_info}", file=sys.stderr)
+            sys.exit(1)
+        time.sleep(1)
+    print(f"Timeout waiting for container {container_id} (status: {s})", file=sys.stderr)
+    sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Argument parser
+# ---------------------------------------------------------------------------
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="instagram", description="Instagram CLI for picoclaw")
+    parser.add_argument("--json", dest="json_output", action="store_true", help="JSON output")
+    parser.add_argument("--token", help="Access token (overrides INSTAGRAM_ACCESS_TOKEN env var)")
+    parser.add_argument("--user-id", help="Instagram User ID (overrides INSTAGRAM_USER_ID env var)")
+    sub = parser.add_subparsers(dest="command")
+
+    # whoami
+    sub.add_parser("whoami", help="Show current user info")
+
+    # post-image
+    p = sub.add_parser("post-image", help="Post a single image")
+    p.add_argument("image_url", help="Public image URL (JPEG recommended)")
+    p.add_argument("--caption", default="", help="Post caption")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # post-video (reel)
+    p = sub.add_parser("post-video", help="Post a reel (video)")
+    p.add_argument("video_url", help="Public video URL (mp4)")
+    p.add_argument("--caption", default="", help="Reel caption")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # post-carousel
+    p = sub.add_parser("post-carousel", help="Post a carousel (multiple images/videos)")
+    p.add_argument("urls", nargs="+", help="Public media URLs (2-10 images/videos)")
+    p.add_argument("--caption", default="", help="Carousel caption")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # post-story
+    p = sub.add_parser("post-story", help="Post a story (image or video)")
+    p.add_argument("media_url", help="Public media URL")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # reply
+    p = sub.add_parser("reply", help="Reply to a comment")
+    p.add_argument("comment_id", help="Comment ID to reply to")
+    p.add_argument("text", help="Reply text")
+
+    # posts
+    p = sub.add_parser("posts", help="List recent posts")
+    p.add_argument("-n", type=int, default=10, help="Number of posts")
+
+    # insights
+    p = sub.add_parser("insights", help="Get insights for a post")
+    p.add_argument("post_id", help="Post ID")
+
+    # profile
+    sub.add_parser("profile", help="View your profile")
+
+    args = parser.parse_args()
+    if not args.command:
+        parser.print_help()
+        sys.exit(1)
+
+    handlers = {
+        "whoami": cmd_whoami,
+        "post-image": cmd_post_image,
+        "post-video": cmd_post_video,
+        "post-carousel": cmd_post_carousel,
+        "post-story": cmd_post_story,
+        "reply": cmd_reply,
+        "posts": cmd_posts,
+        "insights": cmd_insights,
+        "profile": cmd_profile,
+    }
+
+    try:
+        handlers[args.command](args)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

docker/spotify-playlist.py

@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+"""Fetch all tracks from a Spotify playlist and write them to JSON.
+
+Usage:
+  spotify-playlist.py [--playlist URL_OR_ID] [--output PATH]
+
+Requires SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET env vars.
+Get them at https://developer.spotify.com/dashboard (create a free app).
+Add http://127.0.0.1:8888/callback as a Redirect URI in your app settings.
+
+On first run, you'll be prompted to visit a URL and paste back the redirect.
+The token is cached at ~/.spotify-token.json for subsequent runs.
+
+Example:
+  export SPOTIFY_CLIENT_ID="abc123"
+  export SPOTIFY_CLIENT_SECRET="xyz789"
+  python3 spotify-playlist.py \
+    --playlist 530uTl6cSboMQWNRJSJrjz \
+    --output /root/.picoclaw/workspace/data/playlist.json
+"""
+
+import argparse
+import base64
+import json
+import os
+import re
+import secrets
+import sys
+import webbrowser
+from urllib.request import Request, urlopen
+from urllib.parse import urlencode, urlparse, parse_qs
+
+DEFAULT_PLAYLIST = "530uTl6cSboMQWNRJSJrjz"
+DEFAULT_OUTPUT = "playlist.json"
+TOKEN_URL = "https://accounts.spotify.com/api/token"
+AUTH_URL = "https://accounts.spotify.com/authorize"
+REDIRECT_URI = "http://127.0.0.1:8888/callback"
+SCOPE = "playlist-read-private playlist-read-collaborative"
+API_BASE = "https://api.spotify.com/v1"
+TOKEN_CACHE = os.path.join(os.path.expanduser("~"), ".spotify-token.json")
+
+
+def _load_cached_token() -> dict | None:
+    """Load cached token if it exists."""
+    if os.path.exists(TOKEN_CACHE):
+        with open(TOKEN_CACHE) as f:
+            return json.load(f)
+    return None
+
+
+def _save_token(token_data: dict) -> None:
+    """Cache the token to disk."""
+    with open(TOKEN_CACHE, "w") as f:
+        json.dump(token_data, f)
+
+
+def _refresh_token(client_id: str, client_secret: str, refresh_token: str) -> dict:
+    """Refresh an expired access token."""
+    creds = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
+    data = urlencode({
+        "grant_type": "refresh_token",
+        "refresh_token": refresh_token,
+    }).encode()
+    req = Request(TOKEN_URL, data=data, headers={
+        "Authorization": f"Basic {creds}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    })
+    with urlopen(req) as resp:
+        result = json.loads(resp.read())
+    # Spotify may or may not return a new refresh token
+    if "refresh_token" not in result:
+        result["refresh_token"] = refresh_token
+    _save_token(result)
+    return result
+
+
+def _authorize(client_id: str, client_secret: str) -> dict:
+    """Authorization Code flow via manual URL paste (no local server needed)."""
+    state = secrets.token_urlsafe(16)
+
+    params = urlencode({
+        "client_id": client_id,
+        "response_type": "code",
+        "redirect_uri": REDIRECT_URI,
+        "scope": SCOPE,
+        "state": state,
+    })
+    url = f"{AUTH_URL}?{params}"
+    print("1. Visit this URL and authorize the app:\n")
+    print(f"   {url}\n")
+    webbrowser.open(url)
+    print("2. After authorizing, you'll be redirected to a page that won't load.")
+    print("   Copy the FULL URL from your browser's address bar and paste it here.\n")
+
+    redirect_url = input("Paste redirect URL: ").strip()
+
+    qs = parse_qs(urlparse(redirect_url).query)
+    if qs.get("state", [None])[0] != state:
+        print("Error: State mismatch — possible CSRF. Try again.", file=sys.stderr)
+        sys.exit(1)
+    if "error" in qs:
+        print(f"Authorization denied: {qs['error'][0]}", file=sys.stderr)
+        sys.exit(1)
+
+    auth_code = qs.get("code", [None])[0]
+    if not auth_code:
+        print("Error: No authorization code found in URL.", file=sys.stderr)
+        sys.exit(1)
+
+    # Exchange code for token
+    creds = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
+    data = urlencode({
+        "grant_type": "authorization_code",
+        "code": auth_code,
+        "redirect_uri": REDIRECT_URI,
+    }).encode()
+    req = Request(TOKEN_URL, data=data, headers={
+        "Authorization": f"Basic {creds}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    })
+    with urlopen(req) as resp:
+        token_data = json.loads(resp.read())
+
+    _save_token(token_data)
+    return token_data
+
+
+def get_token(client_id: str, client_secret: str) -> str:
+    """Get a valid access token, refreshing or re-authorizing as needed."""
+    cached = _load_cached_token()
+    if cached:
+        # Try using the cached token
+        try:
+            req = Request(f"{API_BASE}/me", headers={
+                "Authorization": f"Bearer {cached['access_token']}"
+            })
+            urlopen(req)
+            return cached["access_token"]
+        except Exception:
+            pass
+        # Try refreshing
+        if cached.get("refresh_token"):
+            try:
+                refreshed = _refresh_token(client_id, client_secret, cached["refresh_token"])
+                return refreshed["access_token"]
+            except Exception:
+                pass
+    # Full re-authorization
+    token_data = _authorize(client_id, client_secret)
+    return token_data["access_token"]
+
+
+def _api_get(token: str, url: str) -> dict:
+    """GET a Spotify API endpoint."""
+    req = Request(url, headers={"Authorization": f"Bearer {token}"})
+    with urlopen(req) as resp:
+        return json.loads(resp.read())
+
+
+def _extract_playlist_id(url_or_id: str) -> str:
+    """Extract playlist ID from a URL or return as-is."""
+    m = re.search(r"playlist/([a-zA-Z0-9]+)", url_or_id)
+    return m.group(1) if m else url_or_id
+
+
+def fetch_playlist(token: str, playlist_id: str) -> dict:
+    """Fetch playlist metadata and all tracks (handles pagination)."""
+    info = _api_get(token, f"{API_BASE}/playlists/{playlist_id}?fields=name,description,external_urls")
+
+    tracks = []
+    url = f"{API_BASE}/playlists/{playlist_id}/items?limit=50"
+
+    while url:
+        page = _api_get(token, url)
+        for item in page.get("items", []):
+            t = item.get("item") or item.get("track")
+            if not t or not t.get("id") or t.get("type") == "episode":
+                continue  # skip local files / unavailable tracks
+
+            album = t.get("album") or {}
+            images = album.get("images") or []
+            art_url = images[0]["url"] if images else None
+
+            tracks.append({
+                "id": t["id"],
+                "name": t["name"],
+                "artists": [a["name"] for a in t.get("artists", [])],
+                "album": album.get("name", ""),
+                "release_date": album.get("release_date", ""),
+                "duration_ms": t.get("duration_ms", 0),
+                "spotify_url": (t.get("external_urls") or {}).get("spotify", ""),
+                "preview_url": t.get("preview_url"),
+                "album_art_url": art_url,
+            })
+        url = page.get("next")
+
+    return {
+        "playlist_name": info.get("name", ""),
+        "playlist_description": info.get("description", ""),
+        "playlist_url": (info.get("external_urls") or {}).get("spotify", ""),
+        "total_tracks": len(tracks),
+        "tracks": tracks,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Fetch Spotify playlist tracks to JSON")
+    parser.add_argument("--playlist", default=DEFAULT_PLAYLIST,
+                        help="Spotify playlist URL or ID (default: The Primer playlist)")
+    parser.add_argument("--output", "-o", default=DEFAULT_OUTPUT,
+                        help="Output JSON file path (default: playlist.json)")
+    args = parser.parse_args()
+
+    client_id = os.environ.get("SPOTIFY_CLIENT_ID", "")
+    client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET", "")
+    if not client_id or not client_secret:
+        print("Error: Set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET env vars.", file=sys.stderr)
+        print("Get them at https://developer.spotify.com/dashboard", file=sys.stderr)
+        sys.exit(1)
+
+    playlist_id = _extract_playlist_id(args.playlist)
+    print(f"Fetching playlist {playlist_id}...")
+
+    token = get_token(client_id, client_secret)
+    data = fetch_playlist(token, playlist_id)
+
+    os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
+    with open(args.output, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+
+    print(f"Wrote {data['total_tracks']} tracks to {args.output}")
+    print(f"Playlist: {data['playlist_name']}")
+
+
+if __name__ == "__main__":
+    main()

docker/threads.py

@@ -0,0 +1,385 @@
+#!/usr/bin/env python3
+"""threads — Threads CLI for picoclaw.
+
+Thin wrapper around the Threads API (graph.threads.net).
+Installed to /usr/local/bin/threads inside the Docker image
+so the agent can call `threads post`, `threads whoami`, etc.
+
+Auth: Set THREADS_ACCESS_TOKEN env var or pass --token.
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+import urllib.request
+import urllib.parse
+import urllib.error
+
+API_BASE = "https://graph.threads.net/v1.0"
+
+
+# ---------------------------------------------------------------------------
+# HTTP helpers (stdlib only — no extra deps needed)
+# ---------------------------------------------------------------------------
+
+def _get_token(args) -> str:
+    token = getattr(args, "token", None) or os.environ.get("THREADS_ACCESS_TOKEN")
+    if not token:
+        print("No access token. Set THREADS_ACCESS_TOKEN or pass --token.", file=sys.stderr)
+        sys.exit(1)
+    return token
+
+
+def _api_get(path: str, params: dict) -> dict:
+    qs = urllib.parse.urlencode(params)
+    url = f"{API_BASE}/{path}?{qs}"
+    req = urllib.request.Request(url, method="GET")
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode()
+        print(f"API error {e.code}: {body}", file=sys.stderr)
+        sys.exit(1)
+
+
+def _api_post(path: str, params: dict) -> dict:
+    url = f"{API_BASE}/{path}"
+    data = urllib.parse.urlencode(params).encode()
+    req = urllib.request.Request(url, data=data, method="POST")
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode()
+        print(f"API error {e.code}: {body}", file=sys.stderr)
+        sys.exit(1)
+
+
+def _get_user_id(token: str) -> str:
+    data = _api_get("me", {"fields": "id", "access_token": token})
+    return data["id"]
+
+
+# ---------------------------------------------------------------------------
+# Commands
+# ---------------------------------------------------------------------------
+
+def cmd_whoami(args) -> None:
+    token = _get_token(args)
+    data = _api_get("me", {
+        "fields": "id,username,threads_profile_picture_url,threads_biography",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"User ID:  {data.get('id', '?')}")
+        print(f"Username: @{data.get('username', '?')}")
+        if data.get("threads_biography"):
+            print(f"Bio:      {data['threads_biography']}")
+
+
+def cmd_post(args) -> None:
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+    text = args.text
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post ({len(text)} chars):")
+        print(text)
+        return
+
+    create_params = {
+        "media_type": "TEXT",
+        "text": text,
+        "access_token": token,
+    }
+
+    # Attach image if provided
+    if args.image:
+        create_params["media_type"] = "IMAGE"
+        create_params["image_url"] = args.image
+
+    # Step 1: Create media container
+    result = _api_post(f"{user_id}/threads", create_params)
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    # Step 2: Wait for container to be ready (poll status)
+    _wait_for_container(token, container_id)
+
+    # Step 3: Publish
+    publish_result = _api_post(f"{user_id}/threads_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    post_id = publish_result.get("id")
+    print(f"Posted: {post_id}")
+
+
+def cmd_post_image_url(args) -> None:
+    """Post with an image URL (Threads requires publicly accessible URLs)."""
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post image ({len(args.text)} chars):")
+        print(f"  Text: {args.text}")
+        print(f"  Image: {args.image_url}")
+        return
+
+    result = _api_post(f"{user_id}/threads", {
+        "media_type": "IMAGE",
+        "text": args.text,
+        "image_url": args.image_url,
+        "access_token": token,
+    })
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    _wait_for_container(token, container_id)
+
+    publish_result = _api_post(f"{user_id}/threads_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    print(f"Posted: {publish_result.get('id')}")
+
+
+def cmd_post_video_url(args) -> None:
+    """Post with a video URL (must be publicly accessible, mp4)."""
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+
+    if args.dry_run:
+        print(f"[DRY RUN] Would post video ({len(args.text)} chars):")
+        print(f"  Text: {args.text}")
+        print(f"  Video: {args.video_url}")
+        return
+
+    result = _api_post(f"{user_id}/threads", {
+        "media_type": "VIDEO",
+        "text": args.text,
+        "video_url": args.video_url,
+        "access_token": token,
+    })
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    # Videos take longer to process
+    _wait_for_container(token, container_id, max_wait=120)
+
+    publish_result = _api_post(f"{user_id}/threads_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    print(f"Posted: {publish_result.get('id')}")
+
+
+def cmd_reply(args) -> None:
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+
+    result = _api_post(f"{user_id}/threads", {
+        "media_type": "TEXT",
+        "text": args.text,
+        "reply_to_id": args.post_id,
+        "access_token": token,
+    })
+    container_id = result.get("id")
+    if not container_id:
+        print(f"Failed to create reply container: {result}", file=sys.stderr)
+        sys.exit(1)
+
+    _wait_for_container(token, container_id)
+
+    publish_result = _api_post(f"{user_id}/threads_publish", {
+        "creation_id": container_id,
+        "access_token": token,
+    })
+    print(f"Replied: {publish_result.get('id')}")
+
+
+def cmd_profile(args) -> None:
+    token = _get_token(args)
+    data = _api_get("me", {
+        "fields": "id,username,threads_profile_picture_url,threads_biography",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"@{data.get('username', '?')}")
+        if data.get("threads_biography"):
+            print(f"  Bio: {data['threads_biography']}")
+        print(f"  ID:  {data.get('id', '?')}")
+
+
+def cmd_insights(args) -> None:
+    """Get profile-level insights (follower count, views)."""
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+    data = _api_get(f"{user_id}/threads_insights", {
+        "metric": "views,likes,replies,reposts,quotes,followers_count",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        for item in data.get("data", []):
+            name = item.get("name", "?")
+            values = item.get("values", [{}])
+            val = values[0].get("value", "?") if values else "?"
+            print(f"  {name}: {val}")
+
+
+def cmd_posts(args) -> None:
+    """List recent posts."""
+    token = _get_token(args)
+    user_id = _get_user_id(token)
+    data = _api_get(f"{user_id}/threads", {
+        "fields": "id,text,timestamp,media_type,shortcode,permalink",
+        "limit": args.n,
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        for post in data.get("data", []):
+            ts = post.get("timestamp", "")
+            text = post.get("text", "")
+            pid = post.get("id", "?")
+            permalink = post.get("permalink", "")
+            print(f"  [{ts}] {pid}")
+            for line in text.splitlines():
+                print(f"    {line}")
+            if permalink:
+                print(f"    {permalink}")
+            print()
+
+
+def cmd_post_insights(args) -> None:
+    """Get insights for a specific post."""
+    token = _get_token(args)
+    data = _api_get(f"{args.post_id}/insights", {
+        "metric": "views,likes,replies,reposts,quotes",
+        "access_token": token,
+    })
+    if args.json_output:
+        print(json.dumps(data, indent=2))
+    else:
+        print(f"Insights for {args.post_id}:")
+        for item in data.get("data", []):
+            name = item.get("name", "?")
+            values = item.get("values", [{}])
+            val = values[0].get("value", "?") if values else "?"
+            print(f"  {name}: {val}")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _wait_for_container(token: str, container_id: str, max_wait: int = 30) -> None:
+    """Poll container status until FINISHED or timeout."""
+    for _ in range(max_wait):
+        status = _api_get(container_id, {
+            "fields": "status,error_message",
+            "access_token": token,
+        })
+        s = status.get("status")
+        if s == "FINISHED":
+            return
+        if s == "ERROR":
+            print(f"Container error: {status.get('error_message', 'unknown')}", file=sys.stderr)
+            sys.exit(1)
+        time.sleep(1)
+    print(f"Timeout waiting for container {container_id} (status: {s})", file=sys.stderr)
+    sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Argument parser
+# ---------------------------------------------------------------------------
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="threads", description="Threads CLI for picoclaw")
+    parser.add_argument("--json", dest="json_output", action="store_true", help="JSON output")
+    parser.add_argument("--token", help="Access token (overrides THREADS_ACCESS_TOKEN env var)")
+    sub = parser.add_subparsers(dest="command")
+
+    # whoami
+    sub.add_parser("whoami", help="Show current user info")
+
+    # post
+    p = sub.add_parser("post", help="Create a text post")
+    p.add_argument("text", help="Post text")
+    p.add_argument("--image", help="Public image URL to attach")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # post-image
+    p = sub.add_parser("post-image", help="Post with a public image URL")
+    p.add_argument("text", help="Post text")
+    p.add_argument("image_url", help="Public image URL")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # post-video
+    p = sub.add_parser("post-video", help="Post with a public video URL")
+    p.add_argument("text", help="Post text")
+    p.add_argument("video_url", help="Public video URL (mp4)")
+    p.add_argument("--dry-run", action="store_true", help="Preview without posting")
+
+    # reply
+    p = sub.add_parser("reply", help="Reply to a post")
+    p.add_argument("post_id", help="Post ID to reply to")
+    p.add_argument("text", help="Reply text")
+
+    # profile
+    sub.add_parser("profile", help="View your profile")
+
+    # insights
+    sub.add_parser("insights", help="View profile insights (followers, views)")
+
+    # posts
+    p = sub.add_parser("posts", help="List recent posts")
+    p.add_argument("-n", type=int, default=10, help="Number of posts")
+
+    # post-insights
+    p = sub.add_parser("post-insights", help="Get insights for a post")
+    p.add_argument("post_id", help="Post ID")
+
+    args = parser.parse_args()
+    if not args.command:
+        parser.print_help()
+        sys.exit(1)
+
+    handlers = {
+        "whoami": cmd_whoami,
+        "post": cmd_post,
+        "post-image": cmd_post_image_url,
+        "post-video": cmd_post_video_url,
+        "reply": cmd_reply,
+        "profile": cmd_profile,
+        "insights": cmd_insights,
+        "posts": cmd_posts,
+        "post-insights": cmd_post_insights,
+    }
+
+    try:
+        handlers[args.command](args)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()