From 61400eb783719f84c3ef292e0e0625d776f80361 Mon Sep 17 00:00:00 2001 From: Kristjan Kruus Date: Tue, 24 Mar 2026 13:57:50 +0200 Subject: [PATCH] My custom setup for picoclaw --- docker/Dockerfile.heavy | 15 +- docker/bsky.py | 25 ++- docker/docker-compose.yml | 3 + docker/instagram.py | 449 +++++++++++++++++++++++++++++++++++++ docker/spotify-playlist.py | 236 +++++++++++++++++++ docker/threads.py | 385 +++++++++++++++++++++++++++++++ 6 files changed, 1110 insertions(+), 3 deletions(-) create mode 100644 docker/instagram.py create mode 100644 docker/spotify-playlist.py create mode 100644 docker/threads.py diff --git a/docker/Dockerfile.heavy b/docker/Dockerfile.heavy index a34940580..37bd27daf 100644 --- a/docker/Dockerfile.heavy +++ b/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 diff --git a/docker/bsky.py b/docker/bsky.py index 592eb0c49..c0cc4571c 100644 --- a/docker/bsky.py +++ b/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 diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 232fa0dfe..0b62d95ae 100644 --- a/docker/docker-compose.yml +++ b/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" diff --git a/docker/instagram.py b/docker/instagram.py new file mode 100644 index 000000000..ad997f61b --- /dev/null +++ b/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() diff --git a/docker/spotify-playlist.py b/docker/spotify-playlist.py new file mode 100644 index 000000000..347d1f687 --- /dev/null +++ b/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() diff --git a/docker/threads.py b/docker/threads.py new file mode 100644 index 000000000..3a324787a --- /dev/null +++ b/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()