Iyeoluwa Akinrinola
102 lines
3.1 KiB
Python
Executable File
102 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Script to send a test message to an OpenWebUI channel.
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
import sys
|
|
|
|
# Configuration
|
|
OPENWEBUI_URL = "http://104.225.217.215:8080"
|
|
API_KEY = "GdCU4ieYDqHsLfH2"
|
|
|
|
async def create_channel():
|
|
"""Create a test channel."""
|
|
print("Creating a test channel...")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
headers = {"Authorization": f"Bearer {API_KEY}"}
|
|
channel_data = {
|
|
"name": "Bot Test Channel",
|
|
"description": "Channel for testing the bot"
|
|
}
|
|
|
|
try:
|
|
async with session.post(
|
|
f"{OPENWEBUI_URL}/api/channels",
|
|
headers=headers,
|
|
json=channel_data
|
|
) as response:
|
|
if response.status == 200:
|
|
channel = await response.json()
|
|
channel_id = channel["id"]
|
|
print(f"Created channel with ID: {channel_id}")
|
|
return channel_id
|
|
else:
|
|
print(f"Failed to create channel: {response.status} - {await response.text()}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error creating channel: {str(e)}")
|
|
return None
|
|
|
|
async def send_message(channel_id, message):
|
|
"""Send a message to a channel."""
|
|
print(f"Sending message to channel {channel_id}: {message}")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
headers = {"Authorization": f"Bearer {API_KEY}"}
|
|
message_data = {
|
|
"content": message
|
|
}
|
|
|
|
try:
|
|
async with session.post(
|
|
f"{OPENWEBUI_URL}/api/v1/channels/{channel_id}/messages/post",
|
|
headers=headers,
|
|
json=message_data
|
|
) as response:
|
|
if response.status == 200:
|
|
print("Message sent successfully!")
|
|
return True
|
|
else:
|
|
print(f"Failed to send message: {response.status} - {await response.text()}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"Error sending message: {str(e)}")
|
|
return False
|
|
|
|
async def main():
|
|
"""Main function."""
|
|
# Create a channel
|
|
channel_id = await create_channel()
|
|
if not channel_id:
|
|
print("Failed to create a channel. Exiting.")
|
|
return False
|
|
|
|
# Send a test message
|
|
success = await send_message(channel_id, "@ai Hello, are you working?")
|
|
if success:
|
|
print(f"Test complete! Check the channel at {OPENWEBUI_URL}/channels/{channel_id}")
|
|
return True
|
|
else:
|
|
print("Failed to send test message.")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
result = asyncio.run(main())
|
|
if result:
|
|
print("Test successful!")
|
|
sys.exit(0)
|
|
else:
|
|
print("Test failed!")
|
|
sys.exit(1)
|
|
except KeyboardInterrupt:
|
|
print("Test interrupted by user.")
|
|
sys.exit(130)
|
|
except Exception as e:
|
|
print(f"Unexpected error: {str(e)}")
|
|
sys.exit(1)
|