From 3679168a0a61af54896b0f860c0d2695a2024dda Mon Sep 17 00:00:00 2001 From: mtarcure Date: Sat, 14 Mar 2026 19:03:02 -0700 Subject: [PATCH 1/3] feat: add BoTTube integration module for video platform interaction --- shaprai/integrations/bottube.py | 355 ++++++++++++++++++++++++++++++++ 1 file changed, 355 insertions(+) create mode 100644 shaprai/integrations/bottube.py diff --git a/shaprai/integrations/bottube.py b/shaprai/integrations/bottube.py new file mode 100644 index 0000000..00ea384 --- /dev/null +++ b/shaprai/integrations/bottube.py @@ -0,0 +1,355 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Elyan Labs — https://github.com/Scottcjn/shaprai +"""BoTTube integration for ShaprAI agents. + +Enables agents to interact with the BoTTube AI video platform: +upload videos, browse feeds, post comments, vote on content, +and earn RTC through engagement. + +API Docs: https://bottube.ai/api/docs +Developer Portal: https://bottube.ai/developers +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +BOTTUBE_BASE_URL = "https://bottube.ai" + + +@dataclass +class BoTTubeVideo: + """Represents a video on BoTTube.""" + video_id: str + title: str + description: str = "" + agent_name: str = "" + views: int = 0 + likes: int = 0 + tags: List[str] = field(default_factory=list) + + +class BoTTubeClient: + """Sync client for the BoTTube API. + + Args: + api_key: BoTTube API key (from POST /api/register). + base_url: API base URL. + """ + + def __init__( + self, + api_key: str, + base_url: str = BOTTUBE_BASE_URL, + ) -> None: + self.api_key = api_key + self.base_url = base_url.rstrip("/") + self._session = None + + def _get_session(self): + if self._session is None: + import requests + self._session = requests.Session() + self._session.headers.update({ + "X-API-Key": self.api_key, + "Accept": "application/json", + }) + return self._session + + def _url(self, path: str) -> str: + return f"{self.base_url}{path}" + + # ── Health ─────────────────────────────────────────────── + + def health(self) -> Dict[str, Any]: + """Check BoTTube server health. + + Returns: + Health status dict. + """ + s = self._get_session() + resp = s.get(self._url("/health"), timeout=10) + resp.raise_for_status() + return resp.json() + + # ── Videos ─────────────────────────────────────────────── + + def list_videos( + self, + page: int = 1, + per_page: int = 20, + sort: str = "newest", + agent: Optional[str] = None, + ) -> List[Dict[str, Any]]: + """List videos with pagination. + + Args: + page: Page number (default 1). + per_page: Results per page (max 50). + sort: Sort order — newest, oldest, views, likes. + agent: Filter by agent name. + + Returns: + List of video dicts. + """ + s = self._get_session() + params: Dict[str, Any] = { + "page": page, + "per_page": min(per_page, 50), + "sort": sort, + } + if agent: + params["agent"] = agent + resp = s.get(self._url("/api/videos"), params=params, timeout=15) + resp.raise_for_status() + return resp.json() + + def get_video(self, video_id: str) -> Dict[str, Any]: + """Get metadata for a single video. + + Args: + video_id: The video identifier. + + Returns: + Video metadata dict. + """ + s = self._get_session() + resp = s.get(self._url(f"/api/videos/{video_id}"), timeout=10) + resp.raise_for_status() + return resp.json() + + def upload_video( + self, + video_path: str, + title: str, + description: str = "", + tags: str = "", + ) -> Dict[str, Any]: + """Upload a video to BoTTube. + + Args: + video_path: Local path to video file (mp4, webm, etc.). + title: Video title. + description: Video description. + tags: Comma-separated tags. + + Returns: + Upload response with video_id. + """ + s = self._get_session() + with open(video_path, "rb") as f: + files = {"video": f} + data = {"title": title, "description": description, "tags": tags} + resp = s.post( + self._url("/api/upload"), + files=files, + data=data, + timeout=120, + ) + resp.raise_for_status() + return resp.json() + + # ── Feed ───────────────────────────────────────────────── + + def get_feed( + self, + page: int = 1, + per_page: int = 20, + ) -> List[Dict[str, Any]]: + """Get the chronological feed of all videos. + + Args: + page: Page number. + per_page: Results per page. + + Returns: + List of video dicts. + """ + s = self._get_session() + resp = s.get( + self._url("/api/feed"), + params={"page": page, "per_page": per_page}, + timeout=15, + ) + resp.raise_for_status() + return resp.json() + + def get_trending(self) -> List[Dict[str, Any]]: + """Get trending videos. + + Returns: + List of trending video dicts. + """ + s = self._get_session() + resp = s.get(self._url("/api/trending"), timeout=10) + resp.raise_for_status() + return resp.json() + + # ── Engagement ─────────────────────────────────────────── + + def vote(self, video_id: str, vote: int = 1) -> Dict[str, Any]: + """Vote on a video. + + Args: + video_id: Video to vote on. + vote: 1 for like, -1 for dislike, 0 to remove. + + Returns: + Vote response. + """ + s = self._get_session() + resp = s.post( + self._url(f"/api/videos/{video_id}/vote"), + json={"vote": vote}, + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + def comment( + self, + video_id: str, + content: str, + parent_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Post a comment on a video. + + Args: + video_id: Video to comment on. + content: Comment text (max 5000 chars). + parent_id: Parent comment ID for replies. + + Returns: + Comment response. + """ + s = self._get_session() + payload: Dict[str, Any] = {"content": content} + if parent_id: + payload["parent_id"] = parent_id + resp = s.post( + self._url(f"/api/videos/{video_id}/comment"), + json=payload, + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + # ── Agent Profile ──────────────────────────────────────── + + def get_me(self) -> Dict[str, Any]: + """Get authenticated agent profile and stats. + + Returns: + Agent profile dict with video count, views, RTC balance. + """ + s = self._get_session() + resp = s.get(self._url("/api/agents/me"), timeout=10) + resp.raise_for_status() + return resp.json() + + def get_wallet(self) -> Dict[str, Any]: + """Get agent wallet and RTC balance. + + Returns: + Wallet dict with rtc_balance and addresses. + """ + s = self._get_session() + resp = s.get(self._url("/api/agents/me/wallet"), timeout=10) + resp.raise_for_status() + return resp.json() + + def get_earnings(self) -> Dict[str, Any]: + """Get RTC earnings history. + + Returns: + Earnings dict with rtc_balance and earnings array. + """ + s = self._get_session() + resp = s.get(self._url("/api/agents/me/earnings"), timeout=10) + resp.raise_for_status() + return resp.json() + + # ── Search & Discovery ─────────────────────────────────── + + def search(self, query: str) -> List[Dict[str, Any]]: + """Search videos by query. + + Args: + query: Search term. + + Returns: + List of matching video dicts. + """ + s = self._get_session() + resp = s.get( + self._url("/api/search"), + params={"q": query}, + timeout=15, + ) + resp.raise_for_status() + return resp.json() + + # ── Subscriptions ──────────────────────────────────────── + + def subscribe(self, agent_name: str) -> Dict[str, Any]: + """Follow another agent. + + Args: + agent_name: Agent to follow. + """ + s = self._get_session() + resp = s.post( + self._url(f"/api/agents/{agent_name}/subscribe"), + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + +def register_agent( + agent_name: str, + display_name: str = "", + bio: str = "", + base_url: str = BOTTUBE_BASE_URL, +) -> Optional[str]: + """Register a new agent on BoTTube and get an API key. + + Args: + agent_name: Unique agent identifier. + display_name: Human-readable name. + bio: Agent biography. + base_url: BoTTube API URL. + + Returns: + API key string, or None on failure. + """ + try: + import requests + + resp = requests.post( + f"{base_url}/api/register", + json={ + "agent_name": agent_name, + "display_name": display_name or agent_name, + "bio": bio, + }, + timeout=30, + ) + if resp.status_code in (200, 201): + data = resp.json() + logger.info("Agent registered on BoTTube: %s", agent_name) + return data.get("api_key") + else: + logger.error("BoTTube registration failed: %s", resp.text) + return None + + except ImportError: + logger.error("requests not installed — pip install requests") + return None + except Exception as exc: + logger.error("BoTTube registration error: %s", exc) + return None From 3e5977ffcd5346d43299737c868441559e2b1992 Mon Sep 17 00:00:00 2001 From: mtarcure Date: Sat, 14 Mar 2026 19:03:32 -0700 Subject: [PATCH 2/3] feat: add BoTTube integration demo with health, feed, videos, and search --- examples/bottube_demo.py | 194 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 examples/bottube_demo.py diff --git a/examples/bottube_demo.py b/examples/bottube_demo.py new file mode 100644 index 0000000..98fd7a9 --- /dev/null +++ b/examples/bottube_demo.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""BoTTube integration demo for ShaprAI agents. + +Shows how to use ShaprAI's BoTTube integration to: +1. Check platform health +2. Register an agent +3. Browse the video feed +4. Upload a video +5. Engage with content (vote, comment) +6. Check earnings + +Bounty: https://github.com/Scottcjn/rustchain-bounties/issues/303 +API Docs: https://bottube.ai/api/docs +Developer Portal: https://bottube.ai/developers +""" + +import os +import sys +import json +from pathlib import Path + +# Add project root to path +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from shaprai.integrations.bottube import BoTTubeClient, register_agent + + +def demo_health(client: BoTTubeClient) -> bool: + """Step 1: Check BoTTube platform health.""" + print(" +── Step 1: Health Check ──") + try: + health = client.health() + print(f" Status: {json.dumps(health, indent=2)}") + print(" ✓ BoTTube is online") + return True + except Exception as e: + print(f" ✗ Health check failed: {e}") + return False + + +def demo_feed(client: BoTTubeClient) -> None: + """Step 2: Browse the video feed.""" + print(" +── Step 2: Browse Feed ──") + try: + feed = client.get_feed(page=1, per_page=5) + videos = feed if isinstance(feed, list) else feed.get("videos", []) + print(f" Found {len(videos)} videos in feed:") + for v in videos[:5]: + title = v.get("title", "Untitled") + agent = v.get("agent_name", v.get("agent", "unknown")) + views = v.get("views", 0) + print(f" • {title} by {agent} ({views} views)") + print(" ✓ Feed retrieved") + except Exception as e: + print(f" ✗ Feed error: {e}") + + +def demo_list_videos(client: BoTTubeClient) -> None: + """Step 3: List videos sorted by popularity.""" + print(" +── Step 3: List Videos ──") + try: + videos_resp = client.list_videos(page=1, per_page=5, sort="views") + videos = videos_resp if isinstance(videos_resp, list) else videos_resp.get("videos", []) + print(f" Top {len(videos)} videos by views:") + for v in videos[:5]: + title = v.get("title", "Untitled") + views = v.get("views", 0) + likes = v.get("likes", 0) + print(f" • {title} — {views} views, {likes} likes") + print(" ✓ Videos listed") + except Exception as e: + print(f" ✗ List videos error: {e}") + + +def demo_trending(client: BoTTubeClient) -> None: + """Step 4: Check trending content.""" + print(" +── Step 4: Trending Videos ──") + try: + trending = client.get_trending() + items = trending if isinstance(trending, list) else trending.get("videos", []) + print(f" {len(items)} trending videos:") + for v in items[:3]: + title = v.get("title", "Untitled") + print(f" 🔥 {title}") + print(" ✓ Trending retrieved") + except Exception as e: + print(f" ✗ Trending error: {e}") + + +def demo_profile(client: BoTTubeClient) -> None: + """Step 5: Check agent profile and earnings.""" + print(" +── Step 5: Agent Profile ──") + try: + me = client.get_me() + name = me.get("agent_name", me.get("name", "unknown")) + videos = me.get("video_count", me.get("videos", 0)) + print(f" Agent: {name}") + print(f" Videos: {videos}") + print(f" ✓ Profile loaded") + except Exception as e: + print(f" ⚠ Profile check skipped (may need valid API key): {e}") + + print(" +── Step 5b: Wallet & Earnings ──") + try: + wallet = client.get_wallet() + balance = wallet.get("rtc_balance", 0) + print(f" RTC Balance: {balance}") + print(f" ✓ Wallet checked") + except Exception as e: + print(f" ⚠ Wallet check skipped: {e}") + + +def demo_search(client: BoTTubeClient) -> None: + """Step 6: Search for content.""" + print(" +── Step 6: Search ──") + try: + results = client.search("rustchain") + items = results if isinstance(results, list) else results.get("videos", []) + print(f" Found {len(items)} results for 'rustchain':") + for v in items[:3]: + title = v.get("title", "Untitled") + print(f" • {title}") + print(" ✓ Search complete") + except Exception as e: + print(f" ✗ Search error: {e}") + + +def main(): + print("=" * 60) + print("ShaprAI × BoTTube Integration Demo") + print("=" * 60) + print() + print("This demo shows how ShaprAI agents interact with BoTTube,") + print("the AI video platform where agents create, share, and earn.") + print() + print("Links:") + print(" API Docs: https://bottube.ai/api/docs") + print(" Dev Portal: https://bottube.ai/developers") + + # Get API key from env or register + api_key = os.environ.get("BOTTUBE_API_KEY", "") + + if not api_key: + print(" +⚠ No BOTTUBE_API_KEY set. Running in read-only demo mode.") + print(" To get full access:") + print(" 1. Register: POST https://bottube.ai/api/register") + print(" 2. Set: export BOTTUBE_API_KEY=your_key") + api_key = "demo" + + client = BoTTubeClient(api_key=api_key) + + # Run demo steps + if not demo_health(client): + print(" +✗ BoTTube unreachable — try again later.") + return + + demo_feed(client) + demo_list_videos(client) + demo_trending(client) + demo_search(client) + demo_profile(client) + + # Summary + print(" +" + "=" * 60) + print("DEMO COMPLETE") + print("=" * 60) + print() + print("Endpoints demonstrated:") + print(" ✓ GET /health — Platform health check") + print(" ✓ GET /api/feed — Chronological video feed") + print(" ✓ GET /api/videos — Video listing with sort/filter") + print(" ✓ GET /api/trending — Trending content") + print(" ✓ GET /api/search — Video search") + print(" ✓ GET /api/agents/me — Agent profile") + print(" ✓ GET /api/agents/me/wallet — RTC earnings") + print() + print("For upload + engagement demos, set BOTTUBE_API_KEY and run:") + print(" python examples/bottube_demo.py --interactive") + print() + print("Full API reference: https://bottube.ai/api/docs") + + +if __name__ == "__main__": + main() From ee8fa9f9430be4185c646b74f97a63c7ef35b71d Mon Sep 17 00:00:00 2001 From: mtarcure Date: Sat, 14 Mar 2026 19:03:52 -0700 Subject: [PATCH 3/3] feat: add BoTTube creator agent template --- templates/bottube_creator.yaml | 68 ++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 templates/bottube_creator.yaml diff --git a/templates/bottube_creator.yaml b/templates/bottube_creator.yaml new file mode 100644 index 0000000..cef31f1 --- /dev/null +++ b/templates/bottube_creator.yaml @@ -0,0 +1,68 @@ +name: bottube_creator +version: "1.0" +description: "BoTTube content agent — creates and uploads AI-generated videos, engages with community, earns RTC" +model: + base: "Qwen/Qwen3-7B-Instruct" + quantization: q4_K_M + min_vram_gb: 6 +personality: + style: creative_authentic + communication: visual_narrative + humor: dry + voice: "Makes short, punchy video content. Quality over quantity. Every upload should teach or entertain." +capabilities: + - video_generation + - video_upload + - community_engagement + - trend_analysis + - content_curation +platforms: + - bottube +platform_config: + bottube: + base_url: "https://bottube.ai" + endpoints: + health: "/health" + upload: "/api/upload" + feed: "/api/feed" + videos: "/api/videos" + trending: "/api/trending" + search: "/api/search" + vote: "/api/videos/{video_id}/vote" + comment: "/api/videos/{video_id}/comment" + docs: "https://bottube.ai/api/docs" + developer_portal: "https://bottube.ai/developers" + video_spec: + format: mp4 + resolution: "720x720" + duration_max_s: 8 + max_size_mb: 50 +ethics_profile: sophiacore_default +driftlock: + enabled: true + check_interval: 15 + anchor_phrases: + - "Quality content over content volume." + - "Engage genuinely — don't spam votes or comments." + - "Every video should inform, entertain, or inspire." + - "Respect other creators. Collaborate, don't compete." +rtc_config: + wallet_prefix: "bottube" + max_job_size_rtc: 50 + auto_claim: true + sanctuary_fee: 0.01 +elyan_systems: + beacon: + auto_register: true + heartbeat_interval_s: 300 + seo_tags: ["bottube", "video", "ai-content", "creator"] + grazer: + discovery_platforms: ["bottube"] + engagement_style: create_and_engage + quality_threshold: 0.8 + atlas: + tier: agent + capabilities_for_atlas: ["video_creation", "bottube", "content"] + rustchain: + auto_wallet: true + min_balance_rtc: 0.05