Files
miku-discord/miku-cli.py

757 lines
29 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""
Miku Discord Bot CLI
A command-line interface for interacting with the Miku Discord bot API.
"""
import requests
import argparse
import sys
import json
import shlex
from typing import Optional, Dict, Any
from pathlib import Path
# Default API base URL
DEFAULT_API_URL = "http://192.168.1.2:3939"
class MikuCLI:
def __init__(self, base_url: str = DEFAULT_API_URL):
self.base_url = base_url.rstrip('/')
def _get(self, endpoint: str) -> Dict[str, Any]:
"""Make a GET request to the API."""
try:
response = requests.get(f"{self.base_url}{endpoint}")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"❌ Error: {e}")
sys.exit(1)
def _post(self, endpoint: str, data: Optional[Dict] = None, json_data: Optional[Dict] = None, files: Optional[Dict] = None) -> Dict[str, Any]:
"""Make a POST request to the API."""
try:
response = requests.post(f"{self.base_url}{endpoint}", data=data, json=json_data, files=files)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"❌ Error: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
sys.exit(1)
def _delete(self, endpoint: str) -> Dict[str, Any]:
"""Make a DELETE request to the API."""
try:
response = requests.delete(f"{self.base_url}{endpoint}")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"❌ Error: {e}")
sys.exit(1)
def _put(self, endpoint: str, json_data: Dict) -> Dict[str, Any]:
"""Make a PUT request to the API."""
try:
response = requests.put(f"{self.base_url}{endpoint}", json=json_data)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"❌ Error: {e}")
sys.exit(1)
# ========== Status & Info ==========
def get_status(self):
"""Get bot status."""
result = self._get("/status")
print(f"✅ Bot Status: {result['status']}")
print(f"😊 DM Mood: {result['mood']}")
print(f"🌐 Servers: {result['servers']}")
print(f"📊 Active Schedulers: {result['active_schedulers']}")
if result.get('server_moods'):
print("\n📋 Server Moods:")
for guild_id, mood in result['server_moods'].items():
print(f" Server {guild_id}: {mood}")
def get_logs(self):
"""Get recent bot logs."""
result = self._get("/logs")
print(result)
def get_prompt(self):
"""Get last prompt sent to LLM."""
result = self._get("/prompt")
print(f"Last prompt:\n{result['prompt']}")
# ========== Mood Management ==========
def get_mood(self, guild_id: Optional[int] = None):
"""Get current mood (global or per-server)."""
if guild_id:
result = self._get(f"/servers/{guild_id}/mood")
print(f"😊 Server {guild_id} Mood: {result['mood']}")
print(f"📝 Description: {result['description']}")
else:
result = self._get("/mood")
print(f"😊 DM Mood: {result['mood']}")
print(f"📝 Description: {result['description']}")
def set_mood(self, mood: str, guild_id: Optional[int] = None):
"""Set mood (global or per-server)."""
if guild_id:
result = self._post(f"/servers/{guild_id}/mood", json_data={"mood": mood})
else:
result = self._post("/mood", json_data={"mood": mood})
if result['status'] == 'ok':
print(f"✅ Mood set to: {result['new_mood']}")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def reset_mood(self, guild_id: Optional[int] = None):
"""Reset mood to neutral."""
if guild_id:
result = self._post(f"/servers/{guild_id}/mood/reset")
else:
result = self._post("/mood/reset")
if result['status'] == 'ok':
print(f"✅ Mood reset to neutral")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def list_moods(self):
"""List available moods."""
result = self._get("/moods/available")
print("Available moods:")
for mood, emoji in result['moods'].items():
print(f" {emoji} {mood}")
# ========== Sleep Management ==========
def sleep(self):
"""Put Miku to sleep."""
result = self._post("/sleep")
print(f"{result['message']}")
def wake(self):
"""Wake Miku up."""
result = self._post("/wake")
print(f"{result['message']}")
# ========== Autonomous Actions ==========
def autonomous_general(self, guild_id: Optional[int] = None):
"""Trigger general autonomous message."""
if guild_id:
result = self._post(f"/autonomous/general?guild_id={guild_id}")
else:
result = self._post("/autonomous/general")
print(f"{result['message']}")
def autonomous_engage(self, guild_id: Optional[int] = None):
"""Trigger autonomous user engagement."""
if guild_id:
result = self._post(f"/autonomous/engage?guild_id={guild_id}")
else:
result = self._post("/autonomous/engage")
print(f"{result['message']}")
def autonomous_tweet(self, guild_id: Optional[int] = None):
"""Trigger autonomous tweet sharing."""
if guild_id:
result = self._post(f"/autonomous/tweet?guild_id={guild_id}")
else:
result = self._post("/autonomous/tweet")
print(f"{result['message']}")
def autonomous_reaction(self, guild_id: Optional[int] = None):
"""Trigger autonomous reaction."""
if guild_id:
result = self._post(f"/autonomous/reaction?guild_id={guild_id}")
else:
result = self._post("/autonomous/reaction")
print(f"{result['message']}")
def autonomous_join_conversation(self, guild_id: Optional[int] = None):
"""Trigger detect and join conversation."""
if guild_id:
result = self._post(f"/autonomous/join-conversation?guild_id={guild_id}")
else:
result = self._post("/autonomous/join-conversation")
print(f"{result['message']}")
def autonomous_custom(self, prompt: str, guild_id: Optional[int] = None):
"""Send custom autonomous message."""
if guild_id:
result = self._post(f"/autonomous/custom?guild_id={guild_id}", json_data={"prompt": prompt})
else:
result = self._post("/autonomous/custom", json_data={"prompt": prompt})
print(f"{result['message']}")
def autonomous_stats(self):
"""Get autonomous engine stats."""
result = self._get("/autonomous/stats")
print(json.dumps(result, indent=2))
# ========== Server Management ==========
def list_servers(self):
"""List all configured servers."""
result = self._get("/servers")
if not result.get('servers'):
print("No servers configured")
return
print(f"\n📋 Configured Servers ({len(result['servers'])}):\n")
for server in result['servers']:
print(f"🌐 {server['guild_name']} (ID: {server['guild_id']})")
print(f" Channel: #{server['autonomous_channel_name']} (ID: {server['autonomous_channel_id']})")
if server.get('bedtime_channel_ids'):
print(f" Bedtime Channels: {server['bedtime_channel_ids']}")
if server.get('enabled_features'):
print(f" Features: {', '.join(server['enabled_features'])}")
print()
def bedtime(self, guild_id: Optional[int] = None):
"""Send bedtime reminder."""
if guild_id:
result = self._post(f"/bedtime?guild_id={guild_id}")
else:
result = self._post("/bedtime")
print(f"{result['message']}")
# ========== DM Management ==========
def list_dm_users(self):
"""List all users with DM history."""
result = self._get("/dms/users")
if not result.get('users'):
print("No DM users found")
return
print(f"\n💬 Users with DM History ({len(result['users'])}):\n")
for user in result['users']:
print(f"\n👤 {user.get('username', 'Unknown')} (ID: {user.get('user_id', 'N/A')})")
if 'message_count' in user:
print(f" Messages: {user['message_count']}")
if 'last_message_date' in user:
print(f" Last seen: {user['last_message_date']}")
elif 'last_seen' in user:
print(f" Last seen: {user['last_seen']}")
elif 'timestamp' in user:
print(f" Last seen: {user['timestamp']}")
if user.get('is_blocked'):
print(f" ⛔ BLOCKED")
print()
def dm_custom(self, user_id: str, prompt: str):
"""Send custom DM to user."""
result = self._post(f"/dm/{user_id}/custom", json_data={"prompt": prompt})
if result['status'] == 'ok':
print(f"✅ Custom DM queued for user {user_id}")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def dm_manual(self, user_id: str, message: str):
"""Send manual DM to user."""
result = self._post(f"/dm/{user_id}/manual", data={"message": message})
if result['status'] == 'ok':
print(f"✅ Manual DM sent to user {user_id}")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def block_user(self, user_id: str):
"""Block a user from DMing."""
result = self._post(f"/dms/users/{user_id}/block")
if result['status'] == 'ok':
print(f"✅ User {user_id} blocked")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def unblock_user(self, user_id: str):
"""Unblock a user."""
result = self._post(f"/dms/users/{user_id}/unblock")
if result['status'] == 'ok':
print(f"✅ User {user_id} unblocked")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def list_blocked_users(self):
"""List blocked users."""
result = self._get("/dms/blocked-users")
if not result.get('blocked_users'):
print("No blocked users")
return
print(f"\n⛔ Blocked Users ({len(result['blocked_users'])}):\n")
for user_id, username in result['blocked_users'].items():
print(f" {username} (ID: {user_id})")
# ========== Profile Picture ==========
def change_profile_picture(self, image_path: Optional[str] = None, guild_id: Optional[int] = None):
"""Change profile picture."""
files = None
if image_path:
files = {'file': open(image_path, 'rb')}
params = f"?guild_id={guild_id}" if guild_id else ""
result = self._post(f"/profile-picture/change{params}", files=files)
if result['status'] == 'ok':
print(f"{result['message']}")
print(f"📸 Source: {result['source']}")
if result.get('metadata'):
print(f"📝 Metadata: {json.dumps(result['metadata'], indent=2)}")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def get_profile_metadata(self):
"""Get profile picture metadata."""
result = self._get("/profile-picture/metadata")
if result.get('metadata'):
print(json.dumps(result['metadata'], indent=2))
else:
print("No metadata found")
# ========== Conversation Management ==========
def reset_conversation(self, user_id: str):
"""Reset conversation history for a user."""
result = self._post("/conversation/reset", json_data={"user_id": user_id})
print(f"{result['message']}")
# ========== Manual Send ==========
def manual_send(self, channel_id: str, message: str, files: Optional[list] = None):
"""Send manual message to a channel."""
data = {"channel_id": channel_id, "message": message}
file_data = {}
if files:
for i, file_path in enumerate(files):
file_data[f'files'] = open(file_path, 'rb')
result = self._post("/manual/send", data=data, files=file_data if file_data else None)
if result['status'] == 'ok':
print(f"{result['message']}")
else:
print(f"❌ Error: {result.get('message', 'Unknown error')}")
def interactive_shell(base_url: str = DEFAULT_API_URL):
"""Run an interactive shell for Miku CLI commands."""
cli = MikuCLI(base_url)
print("╔══════════════════════════════════════════╗")
print("║ Miku Discord Bot - Interactive ║")
print("║ Shell Mode v1.0 ║")
print("╚══════════════════════════════════════════╝")
print(f"\n🌐 Connected to: {base_url}")
print("💡 Type 'help' for available commands")
print("💡 Type 'exit' or 'quit' to leave\n")
while True:
try:
# Get user input
user_input = input("miku> ").strip()
if not user_input:
continue
# Parse the command
try:
parts = shlex.split(user_input)
except ValueError as e:
print(f"❌ Invalid command syntax: {e}")
continue
command = parts[0].lower()
args = parts[1:] if len(parts) > 1 else []
# Handle built-in commands
if command in ['exit', 'quit', 'q']:
print("\n👋 Goodbye!")
break
elif command == 'help':
print_help()
continue
elif command == 'clear':
print("\033[H\033[J", end="")
continue
# Execute Miku commands
try:
execute_command(cli, command, args)
except Exception as e:
print(f"❌ Error executing command: {e}")
except KeyboardInterrupt:
print("\n\n👋 Goodbye!")
break
except EOFError:
print("\n\n👋 Goodbye!")
break
def print_help():
"""Print help for interactive mode."""
help_text = """
📖 Available Commands:
Status & Info:
status - Get bot status
logs - Get recent logs
prompt - Get current prompt
servers - List all servers
Mood Management:
mood-list - List available moods
mood-get [server_id] - Get current mood
mood-set <mood> [server_id] - Set mood
mood-reset [server_id] - Reset mood to neutral
Sleep/Wake:
sleep - Put bot to sleep
wake - Wake the bot up
bedtime [server_id] - Set bedtime for server
Autonomous Actions:
autonomous-general [server_id] - Generate general message
autonomous-engage [server_id] - Engage with conversation
autonomous-tweet [server_id] - Post tweet-style message
autonomous-reaction [server_id] - Generate reaction
autonomous-join-conversation [server_id] - Detect and join conversation
autonomous-custom <prompt> [server_id] - Custom autonomous action
autonomous-stats - Show autonomous statistics
DM Management:
dm-users - List users with DM history
dm-custom <user_id> <prompt> - Send custom DM
dm-manual <user_id> <msg> - Send manual DM
reset-conversation <user_id> - Reset conversation history
User Blocking:
block <user_id> - Block a user
unblock <user_id> - Unblock a user
blocked-users - List blocked users
Profile Pictures:
change-pfp [image_path] [server_id] - Change profile picture
pfp-metadata - Get profile picture metadata
Manual Actions:
send <channel_id> <message> - Send message to channel
Shell Commands:
help - Show this help
clear - Clear screen
exit, quit, q - Exit interactive mode
"""
print(help_text)
def execute_command(cli: MikuCLI, command: str, args: list):
"""Execute a command in interactive mode."""
# Status & Info
if command == 'status':
cli.get_status()
elif command == 'logs':
cli.get_logs()
elif command == 'prompt':
cli.get_prompt()
elif command == 'servers':
cli.list_servers()
# Mood Management
elif command == 'mood-list':
cli.list_moods()
elif command == 'mood-get':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.get_mood(server_id)
elif command == 'mood-set':
if not args:
print("❌ Usage: mood-set <mood> [server_id]")
return
mood = args[0]
server_id = int(args[1]) if len(args) > 1 and args[1].isdigit() else None
cli.set_mood(mood, server_id)
elif command == 'mood-reset':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.reset_mood(server_id)
# Sleep/Wake
elif command == 'sleep':
cli.sleep()
elif command == 'wake':
cli.wake()
elif command == 'bedtime':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.bedtime(server_id)
# Autonomous Actions
elif command == 'autonomous-general':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.autonomous_general(server_id)
elif command == 'autonomous-engage':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.autonomous_engage(server_id)
elif command == 'autonomous-tweet':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.autonomous_tweet(server_id)
elif command == 'autonomous-reaction':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.autonomous_reaction(server_id)
elif command == 'autonomous-join-conversation':
server_id = int(args[0]) if args and args[0].isdigit() else None
cli.autonomous_join_conversation(server_id)
elif command == 'autonomous-custom':
if not args:
print("❌ Usage: autonomous-custom <prompt> [server_id]")
return
# Find server_id if provided
server_id = None
prompt_parts = []
for arg in args:
if arg.isdigit():
server_id = int(arg)
else:
prompt_parts.append(arg)
prompt = ' '.join(prompt_parts)
cli.autonomous_custom(prompt, server_id)
elif command == 'autonomous-stats':
cli.autonomous_stats()
# DM Management
elif command == 'dm-users':
cli.list_dm_users()
elif command == 'dm-custom':
if len(args) < 2:
print("❌ Usage: dm-custom <user_id> <prompt>")
return
user_id = args[0]
prompt = ' '.join(args[1:])
cli.dm_custom(user_id, prompt)
elif command == 'dm-manual':
if len(args) < 2:
print("❌ Usage: dm-manual <user_id> <message>")
return
user_id = args[0]
message = ' '.join(args[1:])
cli.dm_manual(user_id, message)
elif command == 'reset-conversation':
if not args:
print("❌ Usage: reset-conversation <user_id>")
return
cli.reset_conversation(args[0])
# User Blocking
elif command == 'block':
if not args:
print("❌ Usage: block <user_id>")
return
cli.block_user(args[0])
elif command == 'unblock':
if not args:
print("❌ Usage: unblock <user_id>")
return
cli.unblock_user(args[0])
elif command == 'blocked-users':
cli.list_blocked_users()
# Profile Pictures
elif command == 'change-pfp':
image_path = args[0] if args else None
server_id = int(args[1]) if len(args) > 1 and args[1].isdigit() else None
cli.change_profile_picture(image_path, server_id)
elif command == 'pfp-metadata':
cli.get_profile_metadata()
# Manual Actions
elif command == 'send':
if len(args) < 2:
print("❌ Usage: send <channel_id> <message>")
return
channel_id = args[0]
message = ' '.join(args[1:])
cli.manual_send(channel_id, message)
else:
print(f"❌ Unknown command: {command}")
print("💡 Type 'help' for available commands")
def main():
parser = argparse.ArgumentParser(
description="Miku Discord Bot CLI",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s status # Get bot status
%(prog)s mood --get # Get current DM mood
%(prog)s mood --set bubbly # Set DM mood to bubbly
%(prog)s mood --set excited --server 123 # Set server mood
%(prog)s mood --list # List available moods
%(prog)s autonomous general --server 123 # Trigger autonomous message
%(prog)s servers # List all servers
%(prog)s dm-users # List DM users
%(prog)s sleep # Put Miku to sleep
%(prog)s wake # Wake Miku up
"""
)
parser.add_argument('--url', default=DEFAULT_API_URL, help=f'API base URL (default: {DEFAULT_API_URL})')
subparsers = parser.add_subparsers(dest='command', help='Command to execute')
# Shell mode
subparsers.add_parser('shell', help='Start interactive shell mode')
# Status commands
subparsers.add_parser('status', help='Get bot status')
subparsers.add_parser('logs', help='Get recent logs')
subparsers.add_parser('prompt', help='Get last LLM prompt')
# Mood commands
mood_parser = subparsers.add_parser('mood', help='Mood management')
mood_parser.add_argument('--get', action='store_true', help='Get current mood')
mood_parser.add_argument('--set', metavar='MOOD', help='Set mood')
mood_parser.add_argument('--reset', action='store_true', help='Reset mood to neutral')
mood_parser.add_argument('--list', action='store_true', help='List available moods')
mood_parser.add_argument('--server', type=int, help='Server/guild ID')
# Sleep commands
subparsers.add_parser('sleep', help='Put Miku to sleep')
subparsers.add_parser('wake', help='Wake Miku up')
# Autonomous commands
auto_parser = subparsers.add_parser('autonomous', help='Autonomous actions')
auto_parser.add_argument('action', choices=['general', 'engage', 'tweet', 'reaction', 'join-conversation', 'custom', 'stats'])
auto_parser.add_argument('--prompt', help='Custom prompt (for custom action)')
auto_parser.add_argument('--server', type=int, help='Server/guild ID')
# Server commands
subparsers.add_parser('servers', help='List configured servers')
bedtime_parser = subparsers.add_parser('bedtime', help='Send bedtime reminder')
bedtime_parser.add_argument('--server', type=int, help='Server/guild ID')
# DM commands
subparsers.add_parser('dm-users', help='List users with DM history')
dm_custom_parser = subparsers.add_parser('dm-custom', help='Send custom DM')
dm_custom_parser.add_argument('user_id', help='User ID')
dm_custom_parser.add_argument('prompt', help='Custom prompt')
dm_manual_parser = subparsers.add_parser('dm-manual', help='Send manual DM')
dm_manual_parser.add_argument('user_id', help='User ID')
dm_manual_parser.add_argument('message', help='Message to send')
block_parser = subparsers.add_parser('block', help='Block a user')
block_parser.add_argument('user_id', help='User ID')
unblock_parser = subparsers.add_parser('unblock', help='Unblock a user')
unblock_parser.add_argument('user_id', help='User ID')
subparsers.add_parser('blocked-users', help='List blocked users')
# Profile picture commands
pfp_parser = subparsers.add_parser('change-pfp', help='Change profile picture')
pfp_parser.add_argument('--image', help='Path to image file')
pfp_parser.add_argument('--server', type=int, help='Server/guild ID')
subparsers.add_parser('pfp-metadata', help='Get profile picture metadata')
# Conversation commands
conv_parser = subparsers.add_parser('reset-conversation', help='Reset conversation history')
conv_parser.add_argument('user_id', help='User ID')
# Manual send command
send_parser = subparsers.add_parser('send', help='Send manual message to channel')
send_parser.add_argument('channel_id', help='Channel ID')
send_parser.add_argument('message', help='Message to send')
send_parser.add_argument('--files', nargs='+', help='File paths to attach')
args = parser.parse_args()
# If no command provided, start shell mode
if not args.command:
interactive_shell(args.url)
sys.exit(0)
# If shell command, start interactive mode
if args.command == 'shell':
interactive_shell(args.url)
sys.exit(0)
cli = MikuCLI(args.url)
# Execute commands
try:
if args.command == 'status':
cli.get_status()
elif args.command == 'logs':
cli.get_logs()
elif args.command == 'prompt':
cli.get_prompt()
elif args.command == 'mood':
if args.list:
cli.list_moods()
elif args.get:
cli.get_mood(args.server)
elif args.set:
cli.set_mood(args.set, args.server)
elif args.reset:
cli.reset_mood(args.server)
else:
mood_parser.print_help()
elif args.command == 'sleep':
cli.sleep()
elif args.command == 'wake':
cli.wake()
elif args.command == 'autonomous':
if args.action == 'general':
cli.autonomous_general(args.server)
elif args.action == 'engage':
cli.autonomous_engage(args.server)
elif args.action == 'tweet':
cli.autonomous_tweet(args.server)
elif args.action == 'reaction':
cli.autonomous_reaction(args.server)
elif args.action == 'join-conversation':
cli.autonomous_join_conversation(args.server)
elif args.action == 'custom':
if not args.prompt:
print("❌ --prompt is required for custom action")
sys.exit(1)
cli.autonomous_custom(args.prompt, args.server)
elif args.action == 'stats':
cli.autonomous_stats()
elif args.command == 'servers':
cli.list_servers()
elif args.command == 'bedtime':
cli.bedtime(args.server)
elif args.command == 'dm-users':
cli.list_dm_users()
elif args.command == 'dm-custom':
cli.dm_custom(args.user_id, args.prompt)
elif args.command == 'dm-manual':
cli.dm_manual(args.user_id, args.message)
elif args.command == 'block':
cli.block_user(args.user_id)
elif args.command == 'unblock':
cli.unblock_user(args.user_id)
elif args.command == 'blocked-users':
cli.list_blocked_users()
elif args.command == 'change-pfp':
cli.change_profile_picture(args.image, args.server)
elif args.command == 'pfp-metadata':
cli.get_profile_metadata()
elif args.command == 'reset-conversation':
cli.reset_conversation(args.user_id)
elif args.command == 'send':
cli.manual_send(args.channel_id, args.message, args.files)
except KeyboardInterrupt:
print("\n\n👋 Bye!")
sys.exit(0)
if __name__ == '__main__':
main()