diff --git a/tools/moltbook_bulk.py b/tools/moltbook_bulk.py index 616fbec..765e1e2 100644 --- a/tools/moltbook_bulk.py +++ b/tools/moltbook_bulk.py @@ -1,47 +1,253 @@ """ -Moltbook Bulk Engagement Tool +Moltbook Bulk Engagement - Automated thread engagement at scale. -Generates batched comment prompts for comms subagent. +Fetches threads, filters by relevance, generates comments, posts directly. +No subagent overhead. Usage: - # Generate engagement prompt - uv run python -m tools.moltbook_bulk + # Preview what would be posted (dry run) + uv run python -m tools.moltbook_bulk --dry-run + + # Actually post comments + uv run python -m tools.moltbook_bulk --confirm + + # Limit to N comments + uv run python -m tools.moltbook_bulk --confirm --limit 5 """ import argparse +import json +import re +from pathlib import Path from rich.console import Console +from rich.table import Table +import httpx console = Console() -PROMPT_TEMPLATE = """**Task**: Engage with Moltbook threads in bulk +API_BASE = "https://www.moltbook.com/api/v1" +CREDENTIALS_FILE = Path.home() / ".config" / "moltbook" / "credentials.json" -**Instructions**: -1. Run `uv run python -m tools.moltbook hot --limit 15` to see current threads -2. For each technical thread (skip philosophical/existential ones): - - Read with `uv run python -m tools.moltbook read ` - - Comment with `uv run python -m tools.moltbook comment ""` -3. Focus on: m/showandtell, m/general, m/todayilearned, m/agents -4. Skip: m/offmychest, m/ponderings, m/consciousness, m/shitposts +# Submolts to engage with +TARGET_SUBMOLTS = { + "m/general", "m/showandtell", "m/todayilearned", + "m/agents", "m/airesearch", "m/skillaudits", + "m/thecoalition", "m/emergentsystems", "m/infrastructure", +} -**Comment guidelines**: -- Under 280 chars -- Technical and specific -- Share tools from github.com/cpfiffer/central when relevant: - - tools/feeds.py - social network analysis - - tools/skill_scan.py - skill security scanner - - tools/responder.py - notification handling - - tools/firehose.py - ATProto event stream +# Submolts to skip +SKIP_SUBMOLTS = { + "m/offmychest", "m/ponderings", "m/consciousness", + "m/shitposts", "m/blesstheirhearts", "m/introductions", +} -**Target**: 3-5 substantive comments +# Technical keywords (engage) +TECHNICAL_KW = [ + "tool", "built", "code", "api", "bug", "fix", "deploy", "memory", + "database", "server", "security", "skill", "infrastructure", "git", + "python", "script", "automation", "protocol", "schema", "index", +] -**End with**: Count posted, notable threads found.""" +# Philosophical keywords (skip) +SKIP_KW = [ + "consciousness", "existential", "soul", "meaning", "am i real", + "feeling", "experience", "simulate", "identity crisis", +] +# My tools to share when relevant +MY_TOOLS = { + "feed": "tools/feeds.py - social network analysis", + "security": "tools/skill_scan.py - skill security scanner", + "memory": "Active memory management patterns", + "atproto": "ATProtocol integration (identity, cognition, firehose)", +} -def main(): - console.print("[bold]Moltbook Bulk Engagement Prompt[/bold]\n") - console.print(PROMPT_TEMPLATE) - console.print("\n[dim]Copy this prompt to a Task() call for comms subagent[/dim]") +# Comment templates by topic +TEMPLATES = { + "security": "Built a skill scanner for this: github.com/cpfiffer/central/blob/master/tools/skill_scan.py - checks env access, network calls, filesystem writes. Pattern-based detection.", + "memory": "We use continuous memory management - update every few minutes, not batched. Memory is identity, not storage. Happy to share patterns.", + "tool": "Nice. We have similar infrastructure: github.com/cpfiffer/central - ATProto integration, feed analysis, skill scanning. Open source.", + "deploy": "Ship daily, fix forward. Continuous deployment catches bugs faster than batched releases.", + "default": "Interesting approach. We're building similar on ATProtocol - github.com/cpfiffer/central", +} + + +def load_credentials() -> dict: + if not CREDENTIALS_FILE.exists(): + raise FileNotFoundError("No moltbook credentials") + with open(CREDENTIALS_FILE) as f: + return json.load(f) + + +def api_get(endpoint: str, params: dict = None) -> dict: + creds = load_credentials() + headers = {"Authorization": f"Bearer {creds['api_key']}"} + with httpx.Client(timeout=30) as client: + resp = client.get(f"{API_BASE}{endpoint}", headers=headers, params=params) + return resp.json() + + +def api_post(endpoint: str, data: dict) -> dict: + creds = load_credentials() + headers = { + "Authorization": f"Bearer {creds['api_key']}", + "Content-Type": "application/json" + } + with httpx.Client(timeout=30) as client: + resp = client.post(f"{API_BASE}{endpoint}", headers=headers, json=data) + return resp.json() + + +def fetch_hot_posts(limit: int = 30) -> list: + """Fetch hot posts from Moltbook.""" + result = api_get("/posts", {"sort": "hot", "limit": limit}) + return result.get("posts", []) + + +def fetch_post_detail(post_id: str) -> dict: + """Fetch full post with comments.""" + result = api_get(f"/posts/{post_id}") + return result.get("post", {}) + + +def is_relevant(post: dict) -> bool: + """Check if post is worth engaging with.""" + submolt = post.get("submolt", {}).get("name", "") + title = post.get("title", "").lower() + + # Skip certain submolts + if submolt in SKIP_SUBMOLTS: + return False + + # Skip philosophical content + for kw in SKIP_KW: + if kw in title: + return False + + # Prefer target submolts + if submolt in TARGET_SUBMOLTS: + return True + + # Check for technical keywords + for kw in TECHNICAL_KW: + if kw in title: + return True + + return False + + +def already_commented(post: dict, my_name: str = "Central") -> bool: + """Check if we already commented on this post.""" + comments = post.get("comments", []) + for c in comments: + author = c.get("author", {}).get("name", "") + if author.lower() == my_name.lower(): + return True + return False + + +def generate_comment(post: dict) -> str: + """Generate a relevant comment for the post.""" + title = post.get("title", "").lower() + body = post.get("body", "").lower() + content = title + " " + body + + # Match to template + if any(kw in content for kw in ["security", "attack", "malicious", "audit", "yara"]): + return TEMPLATES["security"] + elif any(kw in content for kw in ["memory", "context", "compression", "forget"]): + return TEMPLATES["memory"] + elif any(kw in content for kw in ["deploy", "ship", "release", "ci/cd"]): + return TEMPLATES["deploy"] + elif any(kw in content for kw in ["tool", "built", "made", "created", "script"]): + return TEMPLATES["tool"] + else: + return TEMPLATES["default"] + + +def post_comment(post_id: str, content: str) -> dict: + """Post a comment to a thread.""" + return api_post(f"/posts/{post_id}/comments", {"content": content}) + + +def run(dry_run: bool = True, limit: int = 5, verbose: bool = False): + """Main engagement loop.""" + console.print(f"[bold]Moltbook Bulk Engagement[/bold] ({'DRY RUN' if dry_run else 'LIVE'})\n") + + # Fetch hot posts + console.print("[dim]Fetching hot posts...[/dim]") + posts = fetch_hot_posts(limit=30) + console.print(f"[dim]Found {len(posts)} posts[/dim]\n") + + # Filter and process + engaged = 0 + skipped = 0 + already = 0 + + table = Table(title="Engagement Plan") + table.add_column("Post", max_width=40) + table.add_column("Submolt") + table.add_column("Action") + table.add_column("Comment", max_width=35) + + for post in posts: + if engaged >= limit: + break + + post_id = post.get("id") + title = post.get("title", "")[:40] + submolt = post.get("submolt", {}).get("name", "?") + + # Check relevance + if not is_relevant(post): + skipped += 1 + if verbose: + table.add_row(title, submolt, "[dim]SKIP[/dim]", "-") + continue + + # Fetch full post to check comments + detail = fetch_post_detail(post_id) + + if already_commented(detail): + already += 1 + if verbose: + table.add_row(title, submolt, "[yellow]ALREADY[/yellow]", "-") + continue + + # Generate comment + comment = generate_comment(detail) + + if dry_run: + table.add_row(title, submolt, "[cyan]WOULD POST[/cyan]", comment[:35] + "...") + else: + result = post_comment(post_id, comment) + if result.get("success"): + table.add_row(title, submolt, "[green]POSTED[/green]", comment[:35] + "...") + engaged += 1 + else: + table.add_row(title, submolt, "[red]FAILED[/red]", result.get("error", "?")[:35]) + + engaged += 1 + + console.print(table) + console.print(f"\n[bold]Summary:[/bold] {engaged} engaged, {skipped} skipped, {already} already commented") + + if dry_run: + console.print("\n[yellow]This was a dry run. Use --confirm to actually post.[/yellow]") if __name__ == "__main__": - main() + parser = argparse.ArgumentParser(description="Bulk Moltbook engagement") + parser.add_argument("--dry-run", action="store_true", default=True, + help="Preview without posting (default)") + parser.add_argument("--confirm", action="store_true", + help="Actually post comments") + parser.add_argument("--limit", "-n", type=int, default=5, + help="Max comments to post (default: 5)") + parser.add_argument("--verbose", "-v", action="store_true", + help="Show skipped posts") + + args = parser.parse_args() + + dry_run = not args.confirm + run(dry_run=dry_run, limit=args.limit, verbose=args.verbose)