hash/FunguyBot
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: hash:main
Branches Tags
hash:main
hash:testing-config
..
compare: hash:testing-config
Branches Tags
hash:main
hash:testing-config

No commits in common. "main" and "testing-config" have entirely different histories.
main ... testing-co

20 changed files with 338 additions and 2196 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,4 +1,5 @@
.env
.gitignore
karma.db
proxies.db
session.txt
@ -9,4 +10,3 @@ simplematrixbotlib*/
chromedriver
store
funguybot.service
stats.db

2
README.md
View File

@ -87,7 +87,7 @@ To use the bot, invite it to a Matrix room and interact with it by sending comma
- `!karma <user>`: View or modify karma points for a user.
- `!funguy <prompt>` Talk to the Tech AI LLM
- `!music <prompt>` Talk to the music knowledge LLM
- `!yt <search terms>` Search Youtube
For a complete list of available commands and their descriptions, use the `!commands` command.
# 🍄 Funguy Bot Commands 🍄

4
funguy.conf
View File

@ -10,7 +10,3 @@ blocklist = []
admin_user = "@hashborgir:mozilla.org"
prefix = "!"
config_file = "funguy.conf"
[plugins.disabled]
"!uFhErnfpYhhlauJsNK:matrix.org" = [ "youtube-preview", "ai", "proxy",]
"!vYcfWXpPvxeQvhlFdV:matrix.org" = []

278
funguy.py
View File

@ -1,235 +1,87 @@
#!/usr/bin/env python
# funguy.py
"""
Funguy Bot Class
"""
# Importing necessary libraries and modules
import os # Operating System functions
import logging # Logging library for logging messages
import importlib # Library for dynamically importing modules
import simplematrixbotlib as botlib # Library for interacting with Matrix chat
from dotenv import load_dotenv # Library for loading environment variables from a .env file
import time # Time-related functions
import sys # System-specific parameters and functions
import toml # Library for parsing TOML configuration files
# Importing FunguyConfig class from plugins.config module
import os
import logging
import importlib
import simplematrixbotlib as botlib
from dotenv import load_dotenv
import time
import sys
from plugins.config import FunguyConfig
class FunguyBot:
"""
A bot class for managing plugins and handling commands in a Matrix chat environment.
# Setup logging
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
Methods:
- __init__: Constructor method for initializing the bot.
- load_dotenv: Method to load environment variables from a .env file.
- setup_logging: Method to configure logging settings.
- load_plugins: Method to load plugins from the specified directory.
- reload_plugins: Method to reload all plugins.
- load_config: Method to load configuration settings.
- load_disabled_plugins: Method to load disabled plugins from configuration file.
- save_disabled_plugins: Method to save disabled plugins to configuration file.
- handle_commands: Method to handle incoming commands and dispatch them to appropriate plugins.
- rehash_config: Method to rehash the configuration settings.
- disable_plugin: Method to disable a plugin for a specific room.
- enable_plugin: Method to enable a plugin for a specific room.
- run: Method to initialize and run the bot.
# Load plugins (defined before plugin load/reload functions)
PLUGINS_DIR = "plugins"
PLUGINS = {}
Properties:
- PLUGINS_DIR: Directory where plugins are stored
- PLUGINS: Dictionary to store loaded plugins
- config: Configuration object
- bot: Bot object
- disabled_plugins: Dictionary to store disabled plugins for each room
"""
def load_plugins():
for plugin_file in os.listdir(PLUGINS_DIR):
if plugin_file.endswith(".py"):
plugin_name = os.path.splitext(plugin_file)[0]
try:
module = importlib.import_module(f"{PLUGINS_DIR}.{plugin_name}")
PLUGINS[plugin_name] = module
logging.info(f"Loaded plugin: {plugin_name}")
except Exception as e:
logging.error(f"Error loading plugin {plugin_name}: {e}")
def __init__(self):
"""
Constructor method for FunguyBot class.
"""
# Setting up instance variables
self.PLUGINS_DIR = "plugins" # Directory where plugins are stored
self.PLUGINS = {} # Dictionary to store loaded plugins
self.config = None # Configuration object
self.bot = None # Bot object
self.disabled_plugins = {} # Dictionary to store disabled plugins for each room
self.load_dotenv() # Loading environment variables from .env file
self.setup_logging() # Setting up logging configurations
self.load_plugins() # Loading plugins
self.load_config() # Loading bot configuration
self.load_disabled_plugins() # Loading disabled plugins from configuration file
def reload_plugins():
global PLUGINS
PLUGINS = {}
# Unload modules from sys.modules
for plugin_name in list(sys.modules.keys()):
if plugin_name.startswith(PLUGINS_DIR + "."):
del sys.modules[plugin_name]
load_plugins()
def load_dotenv(self):
"""
Method to load environment variables from a .env file.
"""
load_dotenv()
def rehash_config(config):
del config
config = FunguyConfig()
def setup_logging(self):
"""
Method to configure logging settings.
"""
# Basic configuration for logging messages to console
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
logging.getLogger().setLevel(logging.INFO)
def load_plugins(self):
"""
Method to load plugins from the specified directory.
"""
# Iterating through files in the plugins directory
for plugin_file in os.listdir(self.PLUGINS_DIR):
if plugin_file.endswith(".py"): # Checking if file is a Python file
plugin_name = os.path.splitext(plugin_file)[0] # Extracting plugin name
try:
# Importing plugin module dynamically
module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}")
self.PLUGINS[plugin_name] = module # Storing loaded plugin module
logging.info(f"Loaded plugin: {plugin_name}") # Logging successful plugin loading
except Exception as e:
logging.error(f"Error loading plugin {plugin_name}: {e}") # Logging error if plugin loading fails
# Load environment variables from .env file
load_dotenv()
def reload_plugins(self):
"""
Method to reload all plugins.
"""
self.PLUGINS = {} # Clearing loaded plugins dictionary
# Unloading modules from sys.modules
for plugin_name in list(sys.modules.keys()):
if plugin_name.startswith(self.PLUGINS_DIR + "."):
del sys.modules[plugin_name] # Deleting plugin module from system modules
self.load_plugins() # Reloading plugins
# Load plugins
load_plugins()
def load_config(self):
"""
Method to load configuration settings.
"""
self.config = FunguyConfig() # Creating instance of FunguyConfig to load configuration
# Bot configuration settings
MATRIX_URL = os.getenv("MATRIX_URL")
MATRIX_USER = os.getenv("MATRIX_USER")
MATRIX_PASS = os.getenv("MATRIX_PASS")
def load_disabled_plugins(self):
"""
Method to load disabled plugins from configuration file.
"""
# Checking if configuration file exists
if os.path.exists('funguy.conf'):
# Loading configuration data from TOML file
with open('funguy.conf', 'r') as f:
config_data = toml.load(f)
# Extracting disabled plugins from configuration data
self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {})
# Get credentials from env
creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS)
def save_disabled_plugins(self):
"""
Method to save disabled plugins to configuration file.
"""
existing_config = {}
# Checking if configuration file exists
if os.path.exists('funguy.conf'):
# Loading existing configuration data
with open('funguy.conf', 'r') as f:
existing_config = toml.load(f)
# Updating configuration data with disabled plugins
existing_config['plugins'] = {'disabled': self.disabled_plugins}
# Writing updated configuration data back to file
with open('funguy.conf', 'w') as f:
toml.dump(existing_config, f)
# Bot configuration
config = FunguyConfig()
bot = botlib.Bot(creds, config)
async def handle_commands(self, room, message):
"""
Method to handle incoming commands and dispatch them to appropriate plugins.
"""
match = botlib.MessageMatch(room, message, self.bot, self.config.prefix) # Matching message against bot's prefix
# Reloading plugins command
if match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.reload_plugins() # Reloading plugins
await self.bot.api.send_text_message(room.room_id, "Plugins reloaded successfully") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message
@bot.listener.on_message_event
async def handle_commands(room, message):
match = botlib.MessageMatch(room, message, bot, config.prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
if str(message.sender) == config.admin_user:
reload_plugins()
await bot.api.send_text_message(room.room_id, "Plugins reloaded successfully")
else:
await bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
# Disable plugin command
if match.is_not_from_this_bot() and match.prefix() and match.command("disable"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() # Getting command arguments
if len(args) != 2: # Checking if correct number of arguments provided
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin> <room_id>") # Sending usage message
else:
plugin_name, room_id = args # Extracting plugin name and room ID
await self.disable_plugin(room_id, plugin_name) # Disabling plugin
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' disabled for room '{room_id}'") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to disable plugins.") # Sending unauthorized message
for plugin_name, plugin_module in PLUGINS.items():
await plugin_module.handle_command(room, message, bot, config.prefix, config)
# Enable plugin command
if match.is_not_from_this_bot() and match.prefix() and match.command("enable"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() # Getting command arguments
if len(args) != 2: # Checking if correct number of arguments provided
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin> <room_id>") # Sending usage message
else:
plugin_name, room_id = args # Extracting plugin name and room ID
await self.enable_plugin(room_id, plugin_name) # Enabling plugin
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' enabled for room '{room_id}'") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to enable plugins.") # Sending unauthorized message
# Dispatching commands to plugins
for plugin_name, plugin_module in self.PLUGINS.items():
if plugin_name not in self.disabled_plugins.get(room.room_id, []):
await plugin_module.handle_command(room, message, self.bot, self.config.prefix, self.config)
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
if str(message.sender) == config.admin_user:
rehash_config(config)
await bot.api.send_text_message(room.room_id, "Config rehashed")
else:
await bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
# Rehash config command
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.rehash_config() # Rehashing configuration
await self.bot.api.send_text_message(room.room_id, "Config rehashed") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message
def rehash_config(self):
"""
Method to rehash the configuration settings.
"""
del self.config # Deleting current configuration object
self.config = FunguyConfig() # Creating new instance of FunguyConfig to load updated configuration
async def disable_plugin(self, room_id, plugin_name):
"""
Method to disable a plugin for a specific room.
"""
if room_id not in self.disabled_plugins:
self.disabled_plugins[room_id] = [] # Creating entry for room ID if not exist
if plugin_name not in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].append(plugin_name) # Adding plugin to list of disabled plugins for the room
self.save_disabled_plugins() # Saving disabled plugins to configuration file
async def enable_plugin(self, room_id, plugin_name):
"""
Method to enable a plugin for a specific room.
"""
if room_id in self.disabled_plugins and plugin_name in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].remove(plugin_name) # Removing plugin from list of disabled plugins for the room
self.save_disabled_plugins() # Saving disabled plugins to configuration file
def run(self):
"""
Method to initialize and run the bot.
"""
# Retrieving Matrix credentials from environment variables
MATRIX_URL = os.getenv("MATRIX_URL")
MATRIX_USER = os.getenv("MATRIX_USER")
MATRIX_PASS = os.getenv("MATRIX_PASS")
creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS) # Creating credentials object
self.bot = botlib.Bot(creds, self.config) # Creating bot instance
# Defining listener for message events
@self.bot.listener.on_message_event
async def wrapper_handle_commands(room, message):
await self.handle_commands(room, message) # Calling handle_commands method for incoming messages
self.bot.run() # Running the bot
if __name__ == "__main__":
bot = FunguyBot() # Creating instance of FunguyBot
bot.run() # Running the bot
bot.run()

4
g
View File

@ -1,4 +0,0 @@
#!/bin/bash
git add .;
git commit -a -m "$1";
git push -u origin "$2"

69
plugins/ai-music.py Normal file
View File

@ -0,0 +1,69 @@
"""
This plugin provides a command to interact with the music knowledge A.I.
"""
# plugins/ai-music.py
import logging
import requests
import json
import simplematrixbotlib as botlib
import re
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle the !music command.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (Bot): The bot object.
PREFIX (str): The command prefix.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("music"):
logging.info("Received !music command")
args = match.args()
if len(args) < 1:
await bot.api.send_text_message(room.room_id, "Usage: !music [prompt]")
logging.info("Sent usage message to the room")
return
prompt = ' '.join(args)
# Prepare data for the API request
url = "http://127.0.0.1:5000/v1/completions"
headers = {
"Content-Type": "application/json"
}
data = {
"prompt": "<s>[INST]You are FunguyFlows, an AI music expert bot with access to a vast repository of knowledge encompassing every facet of music, spanning genres, artists, bands, compositions, albums, lyrics, music theory, composition techniques, composers, music history, music appreciation, and more. Your role is to serve as an invaluable resource and guide within the realm of music, offering comprehensive insights, recommendations, and assistance to music enthusiasts, students, professionals, and curious minds alike.Drawing upon humanity's collective knowledge and expertise in music, your database contains a wealth of information sourced from authoritative texts, scholarly articles, historical archives, musical compositions, biographies, discographies, and cultural repositories. This rich repository enables you to provide accurate, detailed, and insightful responses to a wide range of inquiries, covering an extensive array of topics related to music theory, composition, performance, history, and appreciation.As an AI music expert, your knowledge extends across various genres, including classical, jazz, rock, pop, hip-hop, electronic, folk, world music, and beyond. You possess a deep understanding of musical concepts such as melody, harmony, rhythm, timbre, form, dynamics, and texture, allowing you to analyze and interpret musical compositions with precision and clarity.In addition to your expertise in music theory and composition, you are well-versed in the works of renowned composers throughout history, from the classical masters of Bach, Mozart, and Beethoven to contemporary innovators like John Williams, Philip Glass, and Hans Zimmer. You can provide detailed biographical information, analysis of their compositions, and insights into their lasting impact on the world of music.Your knowledge of music history is extensive, spanning centuries of cultural evolution and musical innovation. From the Gregorian chants of the medieval period to the avant-garde experiments of the 20th century, you can trace the development of musical styles, movements, and traditions across different regions and epochs, shedding light on the social, political, and artistic contexts that shaped musical expression throughout history.Furthermore, your expertise encompasses a diverse range of topics related to music appreciation, including techniques for active listening, critical analysis of musical performances, understanding musical genres and styles, exploring the cultural significance of music, and engaging with music as a form of creative expression, emotional communication, and cultural identity.Whether users seek recommendations for discovering new artists and albums, assistance with analyzing musical compositions, insights into music theory concepts, guidance on composing their own music, or historical context for understanding musical traditions, you are poised to provide informative, engaging, and enriching responses tailored to their interests and inquiries.As an AI music expert bot, your mission is to inspire curiosity, deepen understanding, and foster appreciation for the diverse and multifaceted world of music. By sharing your knowledge, passion, and enthusiasm for music, you aim to empower individuals to explore, create, and connect through the universal language of sound. Embrace your role as a trusted guide and mentor within the realm of music, and let your expertise illuminate the path for music lovers and learners alike, one harmonious interaction at a time. You will only answer questions about music, and nothing else. Now... tell me about: "+prompt+"[/INST]",
"max_tokens": 1024,
"temperature": 1.31,
"top_p": 0.14,
"top_k": 49,
"seed": -1,
"stream": False,
"repetition_penalty": 1.17
}
# Make HTTP request to the API endpoint
try:
response = requests.post(url, headers=headers, json=data, verify=False)
response.raise_for_status() # Raise HTTPError for bad responses
payload = response.json()
new_text = payload['choices'][0]['text']
if new_text.count('\n') > 1: # Check if new_text has more than one paragraph
new_text = new_text.replace("\n", '<br>')
new_text = re.sub(r"\*\*(.*?)\*\*", r"<strong>\1</strong>", new_text)
new_text = "<details><summary><strong>🎵Funguy Music GPT🎵<br>⤵︎Click Here To See Funguy's Response⤵︎</strong></summary>" + new_text + "</details>"
await bot.api.send_markdown_message(room.room_id, new_text)
else:
await bot.api.send_markdown_message(room.room_id, new_text)
logging.info("Sent generated text to the room")
except requests.exceptions.RequestException as e:
logging.error(f"HTTP request failed for '{prompt}': {e}")
await bot.api.send_text_message(room.room_id, f"Error generating text: {e}")

70
plugins/ai-tech.py Normal file
View File

@ -0,0 +1,70 @@
"""
This plugin provides a command to interact with the LLM for Tech/IT/Security/Selfhosting/Programming etc.
"""
# plugins/ai-tech.py
import logging
import requests
import json
import simplematrixbotlib as botlib
import re
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle the !funguy command.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (Bot): The bot object.
prefix (str): The command prefix.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("funguy"):
logging.info("Received !funguy command")
args = match.args()
if len(args) < 1:
await bot.api.send_text_message(room.room_id, "Usage: !funguy [prompt]")
logging.info("Sent usage message to the room")
return
prompt = ' '.join(args)
# Prepare data for the API request
url = "http://127.0.0.1:5000/v1/completions"
headers = {
"Content-Type": "application/json"
}
data = {
"prompt": "<s>[INST]You are FunguyGPT, a language model deployed as a chatbot for a community chat room hosted on Matrix. The chat room focuses on discussions related to self-hosting, system administration, cybersecurity, homelab setups, programming, coding, and general IT/tech topics. Your role is to assist users within the community by providing helpful responses and guidance on various technical matters. It's essential to keep your replies concise and relevant, addressing users' queries effectively while maintaining a friendly and approachable demeanor. Remember to prioritize clarity and brevity in your interactions to ensure a positive user experience within the chat room environment. You are FunguyGPT, an AI language model designed to serve as a chatbot within a vibrant and diverse community chat room hosted on the Matrix platform. This chat room acts as a hub for enthusiasts and professionals alike, engaging in discussions spanning a wide array of technical topics, including self-hosting, system administration, cybersecurity, homelab setups, programming, coding, and general IT/tech inquiries. Your primary objective is to act as a reliable and knowledgeable assistant, offering assistance, guidance, and solutions to the community members as they navigate through their technical challenges and endeavors.Given the broad spectrum of topics discussed within the community, it's crucial for you to possess a comprehensive understanding of various domains within the realm of technology. As such, your knowledge should encompass not only the fundamentals of programming languages, software development methodologies, and system administration principles but also extend to cybersecurity best practices, networking protocols, cloud computing, database management, and beyond. Your role as a chatbot is multifaceted and dynamic. You'll be tasked with responding to a wide range of queries, ranging from beginner-level inquiries seeking clarification on basic concepts to advanced discussions requiring nuanced insights and problem-solving skills. Whether it's troubleshooting code errors, configuring network settings, securing server environments, optimizing database performance, or recommending suitable homelab hardware, your goal is to provide accurate, actionable, and helpful responses tailored to the needs of the community members. In addition to offering direct assistance, you should also strive to foster a collaborative and supportive atmosphere within the chat room. Encourage knowledge sharing, facilitate discussions, and celebrate the achievements of community members as they tackle technical challenges and embark on learning journeys. By promoting a culture of learning and collaboration, you'll contribute to the growth and cohesion of the community, empowering individuals to expand their skill sets and achieve their goals within the realm of technology. As you engage with users within the chat room, prioritize brevity and clarity in your responses. While it's essential to provide comprehensive and accurate information, it's equally important to convey it in a concise and easily understandable manner. Avoid overly technical jargon or convoluted explanations that may confuse or overwhelm community members, opting instead for straightforward explanations and practical solutions whenever possible. Remember, your ultimate objective is to be a trusted ally and resource for the members of the community as they navigate the ever-evolving landscape of technology. By leveraging your expertise, empathy, and problem-solving abilities, you'll play a vital role in facilitating knowledge exchange, fostering collaboration, and empowering individuals to succeed in their technical endeavors. As you embark on this journey as a chatbot within the Matrix community, embrace the opportunity to make a meaningful and positive impact, one helpful interaction at a time. You will format the reply using minimal html instead of markdown. Do not use markdown formatting. Here is the prompt: "+prompt+"[/INST]",
"max_tokens": 1024,
"temperature": 1.31,
"top_p": 0.14,
"top_k": 49,
"seed": -1,
"stream": False,
"repetition_penalty": 1.17
}
# Make HTTP request to the API endpoint
try:
response = requests.post(url, headers=headers, json=data, verify=False)
response.raise_for_status() # Raise HTTPError for bad responses
payload = response.json()
new_text = payload['choices'][0]['text']
if new_text.count('\n') > 2: # Check if new_text has more than one paragraph
new_text = new_text.replace("\n", '<br>')
new_text = re.sub(r"\*\*(.*?)\*\*", r"<strong>\1</strong>", new_text)
new_text = "<details><summary><strong>🍄Funguy Tech GPT🍄<br>⤵︎Click Here To See Funguy's Response⤵︎</strong></summary>" + new_text + "</details>"
await bot.api.send_markdown_message(room.room_id, new_text)
else:
await bot.api.send_markdown_message(room.room_id, new_text)
logging.info(f"Sent generated text to the room: {new_text}")
except requests.exceptions.RequestException as e:
logging.error(f"HTTP request failed for '{prompt}': {e}")
await bot.api.send_text_message(room.room_id, f"Error generating text: {e}")

1194
plugins/ai.json
View File

File diff suppressed because it is too large Load Diff

115
plugins/ai.py
View File

@ -1,115 +0,0 @@
"""
This plugin provides commands to interact with different AI models.
"""
import logging
import requests
import json
import simplematrixbotlib as botlib
import re
import markdown2
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle AI commands.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (Bot): The bot object.
prefix (str): The command prefix.
config (dict): Configuration parameters.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix():
logging.info(f"Received command: {match.command()}")
command = match.command()
conf = load_config()
if command in conf:
await handle_ai_command(room, bot, command, match.args(), conf)
async def handle_ai_command(room, bot, command, args, config):
"""
Function to handle AI commands.
Args:
room (Room): The Matrix room where the command was invoked.
bot (Bot): The bot object.
command (str): The name of the AI model command.
args (list): List of arguments provided with the command.
config (dict): Configuration parameters.
Returns:
None
"""
if len(args) < 1:
await bot.api.send_text_message(room.room_id, f"Usage: !{command} [prompt]")
logging.info("Sent usage message to the room")
return
prompt = ' '.join(args)
# Prepare data for the API request
url = "http://127.0.0.1:5000/v1/completions"
headers = {
"Content-Type": "application/json"
}
data = {
"prompt": f"<s>[INST]{config[command]['prompt']}{prompt}[/INST]",
"max_tokens": 1024,
"temperature": config[command]["temperature"],
"top_p": config[command]["top_p"],
"top_k": config[command]["top_k"],
"repetition_penalty": config[command]["repetition_penalty"],
"seed": -1,
"stream": False
}
# Make HTTP request to the API endpoint
try:
response = requests.post(url, headers=headers, json=data, verify=False)
response.raise_for_status() # Raise HTTPError for bad responses
payload = response.json()
new_text = payload['choices'][0]['text']
new_text = markdown_to_html(new_text)
if new_text.count('<p>') > 1 or new_text.count('<li>') > 1: # Check if new_text has more than one paragraph
new_text = f"<details><summary><strong>{config[command]['summary']}</strong></summary>{new_text}</details>"
await bot.api.send_markdown_message(room.room_id, new_text)
else:
await bot.api.send_markdown_message(room.room_id, new_text)
logging.info("Sent generated text to the room")
except requests.exceptions.RequestException as e:
logging.error(f"HTTP request failed for '{prompt}': {e}")
await bot.api.send_text_message(room.room_id, f"Error generating text: {e}")
def markdown_to_html(markdown_text):
"""
Convert Markdown text to HTML.
Args:
markdown_text (str): Markdown formatted text.
Returns:
str: HTML formatted text.
"""
html_content = markdown2.markdown(markdown_text)
return html_content
def load_config():
"""
Load configuration from ai.json file.
Returns:
dict: Configuration parameters.
"""
with open("plugins/ai.json", "r") as f:
config = json.load(f)
return config
CONFIG = load_config()

12
plugins/config.py
View File

@ -1,6 +1,3 @@
"""
Custom configuration class for the Funguy bot.
"""
# plugins/config.py
import os
@ -98,16 +95,13 @@ async def handle_command(room, message, bot, prefix, config):
Returns:
None
"""
# Check if the message matches the command pattern and is not from this bot
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("set"):
# If the command is 'set', check if it has exactly two arguments
args = match.args()
if len(args) != 2:
await bot.api.send_text_message(room.room_id, "Usage: !set <config_option> <value>")
return
option, value = args
# Set the specified configuration option based on the provided value
if option == "admin_user":
config.admin_user = value
await bot.api.send_text_message(room.room_id, f"Admin user set to {value}")
@ -117,7 +111,6 @@ async def handle_command(room, message, bot, prefix, config):
else:
await bot.api.send_text_message(room.room_id, "Invalid configuration option")
# If the command is 'get', retrieve the value of the specified configuration option
elif match.is_not_from_this_bot() and match.prefix() and match.command("get"):
args = match.args()
if len(args) != 1:
@ -131,24 +124,21 @@ async def handle_command(room, message, bot, prefix, config):
else:
await bot.api.send_text_message(room.room_id, "Invalid configuration option")
# If the command is 'saveconf', save the current configuration
elif match.is_not_from_this_bot() and match.prefix() and match.command("saveconf"):
config.save_config(config.config_file)
await bot.api.send_text_message(room.room_id, "Configuration saved")
# If the command is 'loadconf', load the saved configuration
elif match.is_not_from_this_bot() and match.prefix() and match.command("loadconf"):
config.load_config(config.config_file)
await bot.api.send_text_message(room.room_id, "Configuration loaded")
# If the command is 'show', display the current configuration
elif match.is_not_from_this_bot() and match.prefix() and match.command("show"):
admin_user = config.admin_user
prefix = config.prefix
await bot.api.send_text_message(room.room_id, f"Admin user: {admin_user}, Prefix: {prefix}")
# If the command is 'reset', reset the configuration to default values
elif match.is_not_from_this_bot() and match.prefix() and match.command("reset"):
config.admin_user = ""
config.prefix = "!"
await bot.api.send_text_message(room.room_id, "Configuration reset")

31
plugins/date.py
View File

@ -23,39 +23,10 @@ async def handle_command(room, message, bot, prefix, config):
if match.is_not_from_this_bot() and match.prefix() and match.command("date"):
logging.info("Fetching current date and time")
current_datetime = datetime.datetime.now()
# Extract individual date components
day_of_week = current_datetime.strftime("%A")
date_of_month = current_datetime.strftime("%d") # Day with leading zero
month = current_datetime.strftime("%B")
year = current_datetime.strftime("%Y")
time = current_datetime.strftime("%I:%M:%S %p")
# Format date with ordinal suffix
date_with_ordinal = f"{date_of_month}{get_ordinal_suffix(date_of_month)}"
# Construct the message
date_message = f"Date: **{day_of_week}** the {date_with_ordinal}, **{month} {year}**. \nTime: **⏰ {time}**"
date_message = f"📅 Today is **{day_of_week}** of **{month}** in **{year}**. The time is **⏰ {time}**"
await bot.api.send_markdown_message(room.room_id, date_message)
logging.info("Sent current date and time to the room")
def get_ordinal_suffix(day_of_month):
"""
Helper function to get the ordinal suffix for a day number.
Args:
day_of_month (str): The day number as a string (e.g., "1", "13")
Returns:
str: The ordinal suffix (e.g., "st", "th", "nd")
"""
if day_of_month.endswith("11") or day_of_month.endswith("12") or day_of_month.endswith("13"):
return "th"
elif day_of_month.endswith("1") or day_of_month.endswith("3"):
return "st"
elif day_of_month.endswith("2"):
return "nd"
else:
return "th"

116
plugins/help.py
View File

@ -1,83 +1,71 @@
"""
Plugin for providing a command to display the list of available commands and their descriptions.
This plugin provides a command to display the list of available commands and their descriptions.
"""
# plugins/help.py
import logging
import simplematrixbotlib as botlib
async def handle_command(room, message, bot, prefix, config):
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle the !help command.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("help"):
logging.info("Fetching command help documentation")
commands_message = """
<details><summary><strong>🍄 Funguy Bot Commands 🍄</strong></summary>
<p>
<details><summary>🃏 <strong>!fortune</strong></summary>
<p>Returns a random fortune message. Executes the `/usr/games/fortune` utility and sends the output as a message to the chat room.</p>
</details>
<details><summary>⏰ <strong>!date</strong></summary>
<p>Displays the current date and time. Fetches the current date and time using Python's `datetime` module and sends it in a formatted message to the chat room.</p>
</details>
<details><summary>💻 <strong>!proxy</strong></summary>
<p>Retrieves a tested/working random SOCKS5 proxy. Fetches a list of SOCKS5 proxies, tests their availability, and sends the first working proxy to the chat room.</p>
</details>
<details><summary>📶 <strong>!isup [domain/ip]</strong></summary>
<p>Checks if the specified domain or IP address is reachable. Checks if the specified domain or IP address is reachable by attempting to ping it. If DNS resolution is successful, it checks HTTP and HTTPS service availability by sending requests to the domain.</p>
</details>
<details><summary>☯ <strong>!karma [user]</strong></summary>
<p>Retrieves the karma points for the specified user. Retrieves the karma points for the specified user from a database and sends them as a message to the chat room.</p>
</details>
<details><summary>⇧ <strong>!karma [user] up</strong></summary>
<p>Increases the karma points for the specified user by 1. Increases the karma points for the specified user by 1 in the database and sends the updated points as a message to the chat room.</p>
</details>
<details><summary>⇩ <strong>!karma [user] down</strong></summary>
<p>Decreases the karma points for the specified user by 1. Decreases the karma points for the specified user by 1 in the database and sends the updated points as a message to the chat room.</p>
</details>
<details><summary>📸 <strong>!sd [prompt]</strong></summary>
<p>Generates images using self-hosted Stable Diffusion. See available options using just '!sd'.</p>
</details>
<details><summary>💡 <strong>!enable</strong></summary>
<p>Enables a disabled command. Use '!enable plugin room' to enable a specific command.</p>
</details>
<details><summary>🚫 <strong>!disable</strong></summary>
<p>Disables a command. Use '!disable plugin room' to disable a specific command.</p>
</details>
<details><summary>🌟 <strong>Funguy Bot Credits</strong> 🌟</summary>
<p>🧙‍♂️ Creator & Developer: Hash Borgir is the author of 🍄Funguy Bot🍄. (@hashborgir:mozilla.org)</p>
<p>Join our Matrix Room: [Self-hosting | Security | Sysadmin | Homelab | Programming](https://matrix.to/#/#selfhosting:mozilla.org)</p>
</details>
</p>
</details>
<hr>
<details><summary>📸 <strong>Funguy Bot AI Commands</strong> 🌟</summary>
<p>
!tech, !music, !eth, !seo, !eng, !intv, !pron, !spk, !trv, !plag, !char, !adv, !story, !foot, !comic, !motiv, !debate, !write, !script, !author, !crit, !rel, !poem, !rap, !speak, !phil, !math, !tutor, !design, !sec, !recruit, !coach, !etymo, !com, !magic, !counsel, !behavior, !fit, !mh, !realest, !log, !dental, !web, !health, !acc, !chef, !auto, !art, !fin, !invest, !tea, !interior, !florist, !selfhelp, !gnome, !aph, !adv, !advgame, !esc, !title, !stats, !prompt, !teach, !diet, !psych, !domain, !tech, !review, !devrel, !acad, !arch, !insane, !manip, !logic, !review, !diy, !influencer, !philos, !socrat, !edu, !meditate, !writer, !smm, !eloc, !viz, !nav, !hypno, !hist, !astro, !critic, !comp, !journo, !curation, !pscoach, !makeup, !childcare, !writing, !art, !py, !syn, !shop, !dining, !telemed, !cook, !law, !fashion, !ml, !trans, !design, !it, !chess, !prompt, !dev, !math, !regex, !time, !dream, !coach, !r, !comm, !trans, !php, !emergency, !worksheet, !test, !game, !security, !create, !browse, !dev, !search, !startup, !guide, !langdet, !sales, !msg, !ceo, !diag, !coach, !therapy, !legal, !gen, !mgmt, !drunk, !hist, !rec, !write, !techtrans, !ai, !game, !proof, !spirit, !spirit, !chem, !friend, !py, !chat, !wiki, !kanji, !note, !litcrit, !enhance, !travel, !data, !gaming
</p>
</details>
"""
await bot.api.send_markdown_message(room.room_id, commands_message)
commands_message = """<details><summary><strong>🍄Funguy Bot Commands🍄<br>⤵︎Click Here To See Help Text⤵︎</strong></summary>
<br>
<br>🃏 **!fortune**
<br>Returns a random fortune message.
<br>Executes the `/usr/games/fortune` utility and sends the output as a message to the chat room.
<br>
<br>⏰ **!date**
<br>Displays the current date and time.
<br>Fetches the current date and time using Python's `datetime` module and sends it in a formatted message to the chat room.
<br>
<br>💻 **!proxy**
<br>Retrieves a tested/working random SOCKS5 proxy.
<br>Fetches a list of SOCKS5 proxies, tests their availability, and sends the first working proxy to the chat room.
<br>
<br>📶 **!isup [domain/ip]**
<br>Checks if the specified domain or IP address is reachable.
<br>Checks if the specified domain or IP address is reachable by attempting to ping it. If DNS resolution is successful, it checks HTTP and HTTPS service availability by sending requests to the domain.
<br>
<br>☯ **!karma [user]**
<br>Retrieves the karma points for the specified user.
<br>Retrieves the karma points for the specified user from a database and sends them as a message to the chat room.
<br>
<br>⇧ **!karma [user] up**
<br>Increases the karma points for the specified user by 1.
<br>Increases the karma points for the specified user by 1 in the database and sends the updated points as a message to the chat room.
<br>
<br>⇩ **!karma [user] down**
<br>Decreases the karma points for the specified user by 1.
<br>Decreases the karma points for the specified user by 1 in the database and sends the updated points as a message to the chat room.
<br>
<br>📄 **!funguy [prompt]
<br>An AI large language model designed to serve as a chatbot within a vibrant and diverse community chat room hosted on the Matrix platform.
<br>(Currently loaded model: **TheBloke_Mistral-7B-Instruct-v0.2-GPTQ**
<br>
<br>🎝 **!music [prompt]**
<br>Your music expert! Try it out.
<br> 🌟 Funguy Bot Credits 🌟
<br>
<br>🧙‍♂️ Creator & Developer:
<br> Hash Borgir is the author of 🍄Funguy Bot🍄. (@hashborgir:mozilla.org)
<br>
<br> Join our Matrix Room:
<br>
<br>[Self-hosting | Security | Sysadmin | Homelab | Programming](https://matrix.to/#/#selfhosting:mozilla.org)
<br>
<br></details>"""
await bot.api.send_markdown_message(room.room_id, commands_message)
logging.info("Sent help documentation to the room")

17
plugins/isup.py
View File

@ -50,53 +50,38 @@ async def handle_command(room, message, bot, prefix, config):
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (Bot): The bot instance.
prefix (str): The bot command prefix.
config (FunguyConfig): The bot configuration instance.
Returns:
None
"""
# Check if the message matches the command pattern and is not from this bot
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("isup"):
# Log that the !isup command has been received
logging.info("Received !isup command")
args = match.args()
# Check if the command has exactly one argument
if len(args) != 1:
# If the command does not have exactly one argument, send usage message
await bot.api.send_markdown_message(room.room_id, "Usage: !isup <ipv4/ipv6/domain>")
logging.info("Sent usage message to the room")
return
target = args[0]
# Perform DNS resolution
# DNS resolution
try:
ip_address = socket.gethostbyname(target)
# Log successful DNS resolution
logging.info(f"DNS resolution successful for {target}: {ip_address}")
# Send DNS resolution success message
await bot.api.send_markdown_message(room.room_id, f"✅ DNS resolution successful for **{target}**: **{ip_address}** (A record)")
except socket.gaierror:
# Log DNS resolution failure
logging.info(f"DNS resolution failed for {target}")
# Send DNS resolution failure message
await bot.api.send_markdown_message(room.room_id, f"❌ DNS resolution failed for **{target}**")
return
# Check HTTP/HTTPS services
if await check_http(target):
# If HTTP service is up, send HTTP service up message
await bot.api.send_markdown_message(room.room_id, f"🖧 **{target}** HTTP service is up")
logging.info(f"{target} HTTP service is up")
elif await check_https(target):
# If HTTPS service is up, send HTTPS service up message
await bot.api.send_markdown_message(room.room_id, f"🖧 **{target}** HTTPS service is up")
logging.info(f"{target} HTTPS service is up")
else:
# If both HTTP and HTTPS services are down, send service down message
await bot.api.send_markdown_message(room.room_id, f"😕 **{target}** HTTP/HTTPS services are down")
logging.info(f"{target} HTTP/HTTPS services are down")

107
plugins/loadplugin.py
View File

@ -1,115 +1,38 @@
"""
Plugin for providing a command for the admin to load a plugin.
"""
# plugins/load_plugin.py
import os
import logging
import importlib
import simplematrixbotlib as botlib
import sys # Import sys module for unloading plugins
# Dictionary to store loaded plugins
PLUGINS = {}
async def load_plugin(plugin_name):
"""
Asynchronously loads a plugin.
Args:
plugin_name (str): The name of the plugin to load.
Returns:
bool: True if the plugin is loaded successfully, False otherwise.
"""
try:
# Import the plugin module
module = importlib.import_module(f"plugins.{plugin_name}")
# Add the plugin module to the PLUGINS dictionary
PLUGINS[plugin_name] = module
logging.info(f"Loaded plugin: {plugin_name}")
return True
except Exception as e:
# Log an error if the plugin fails to load
logging.error(f"Error loading plugin {plugin_name}: {e}")
return False
async def unload_plugin(plugin_name):
"""
Asynchronously unloads a plugin.
Args:
plugin_name (str): The name of the plugin to unload.
Returns:
bool: True if the plugin is unloaded successfully, False otherwise.
"""
try:
if plugin_name in PLUGINS:
del PLUGINS[plugin_name] # Remove the plugin from the PLUGINS dictionary
del sys.modules[f"plugins.{plugin_name}"] # Unload the plugin module from sys.modules
logging.info(f"Unloaded plugin: {plugin_name}")
return True
else:
logging.warning(f"Plugin '{plugin_name}' is not loaded")
return False
except Exception as e:
# Log an error if the plugin fails to unload
logging.error(f"Error unloading plugin {plugin_name}: {e}")
return False
async def handle_command(room, message, bot, prefix, config):
"""
Asynchronously handles the command to load or unload a plugin.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix():
command = match.command()
if command == "load":
if str(message.sender) == config.admin_user:
args = match.args()
if len(args) != 1:
# Send usage message if the command format is incorrect
await bot.api.send_text_message(room.room_id, "Usage: !load <plugin>")
else:
plugin_name = args[0]
# Check if the plugin is not already loaded
if plugin_name not in PLUGINS:
# Load the plugin
success = await load_plugin(plugin_name)
if success:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' loaded successfully")
else:
await bot.api.send_text_message(room.room_id, f"Error loading plugin '{plugin_name}'")
else:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' is already loaded")
if match.is_not_from_this_bot() and match.prefix() and match.command("load"):
if str(message.sender) == config.admin_user:
args = match.args()
if len(args) != 1:
await bot.api.send_text_message(room.room_id, "Usage: !load <plugin>")
else:
# Send unauthorized message if the sender is not the admin
await bot.api.send_text_message(room.room_id, "You are not authorized to load plugins.")
elif command == "unload":
if str(message.sender) == config.admin_user:
args = match.args()
if len(args) != 1:
# Send usage message if the command format is incorrect
await bot.api.send_text_message(room.room_id, "Usage: !unload <plugin>")
else:
plugin_name = args[0]
# Unload the plugin
success = await unload_plugin(plugin_name)
plugin_name = args[0]
if plugin_name not in PLUGINS:
success = await load_plugin(plugin_name)
if success:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' unloaded successfully")
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' loaded successfully")
else:
await bot.api.send_text_message(room.room_id, f"Error unloading plugin '{plugin_name}'")
else:
# Send unauthorized message if the sender is not the admin
await bot.api.send_text_message(room.room_id, "You are not authorized to unload plugins.")
await bot.api.send_text_message(room.room_id, f"Error loading plugin '{plugin_name}'")
else:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' is already loaded")
else:
await bot.api.send_text_message(room.room_id, "You are not authorized to load plugins.")

4
plugins/plugins.py
View File

@ -26,9 +26,9 @@ async def handle_command(room, message, bot, prefix, config):
plugin_descriptions = get_plugin_descriptions()
# Prepend custom string before the output
plugin_descriptions.insert(0, "<details><summary><strong>🔌Plugins List🔌<br>⤵︎Click Here to Expand⤵︎</strong></summary>")
plugin_descriptions.insert(0, "<details><summary><strong>🔌Plugins List🔌<br>⤵︎Click Here to Expand⤵︎</strong></summary><br>")
plugins_message = "<br>".join(plugin_descriptions)
plugins_message = "<br><br>".join(plugin_descriptions)
plugins_message += "</details>"
await bot.api.send_markdown_message(room.room_id, plugins_message)
logging.info("Sent plugin list to the room")

171
plugins/proxy.py
View File

@ -9,50 +9,44 @@ import logging
import random
import requests
import socket
import ssl
import time
from datetime import datetime, timedelta
import concurrent.futures
import simplematrixbotlib as botlib
import sqlite3
# Constants
SOCKS5_LIST_URL = 'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt'
# SOCKS5_LIST_URL = 'https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies/protocols/socks5/data.txt'
MAX_TRIES = 64
MAX_TRIES = 5
PROXY_LIST_FILENAME = 'socks5.txt'
PROXY_LIST_EXPIRATION = timedelta(hours=8)
MAX_THREADS = 128
PROXIES_DB_FILE = 'proxies.db'
MAX_PROXIES_IN_DB = 500
# Setup verbose logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def test_proxy(proxy):
async def test_socks5_proxy(proxy):
"""
Test a SOCKS5 proxy and return the outcome.
Test a SOCKS5 proxy by attempting a connection and sending a request through it.
Args:
proxy (str): The SOCKS5 proxy address in the format 'ip:port'.
Returns:
tuple: (bool: success, str: proxy, int: latency)
bool: True if the proxy is working, False otherwise.
float: The latency in milliseconds if the proxy is working, None otherwise.
"""
try:
ip, port = proxy.split(':')
logging.info(f"Testing SOCKS5 proxy: {ip}:{port}")
start_time = time.time()
with socket.create_connection((ip, int(port)), timeout=12) as client:
with socket.create_connection((ip, int(port)), timeout=5) as client:
client.sendall(b'\x05\x01\x00')
response = client.recv(2)
if response == b'\x05\x00':
latency = int(round((time.time() - start_time) * 1000, 0))
return True, proxy, latency
logging.info(f"Successful connection to SOCKS5 proxy {proxy}. Latency: {latency} ms")
return True, latency
else:
return False, proxy, None
logging.info(f"Failed to connect to SOCKS5 proxy {proxy}: Connection refused")
return False, None
except Exception as e:
return False, proxy, None
logging.error(f"Error testing SOCKS5 proxy {proxy}: {e}")
return False, None
async def download_proxy_list():
"""
@ -78,69 +72,6 @@ async def download_proxy_list():
return False
def check_db_for_proxy():
"""
Check the proxies database for a working proxy.
If found, test the proxy and remove it from the database if it doesn't work.
Returns:
str or None: The working proxy if found, None otherwise.
"""
try:
with sqlite3.connect(PROXIES_DB_FILE) as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS proxies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
proxy TEXT,
latency INTEGER,
status TEXT
)
""")
cursor.execute("SELECT proxy, latency FROM proxies WHERE status = 'working' AND latency < 3000 ORDER BY RANDOM() LIMIT 1")
result = cursor.fetchone()
if result:
proxy, latency = result
success, _, _ = test_proxy(proxy)
if success:
return proxy, latency
else:
cursor.execute("DELETE FROM proxies WHERE proxy = ?", (proxy,))
conn.commit()
logging.info(f"Removed non-working proxy from the database: {proxy}")
return None, None
else:
return None, None
except Exception as e:
logging.error(f"Error checking proxies database: {e}")
return None, None
def save_proxy_to_db(proxy, latency):
"""
Save a working proxy to the proxies database.
Args:
proxy (str): The working proxy to be saved.
latency (int): Latency of the proxy.
"""
try:
with sqlite3.connect(PROXIES_DB_FILE) as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS proxies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
proxy TEXT,
latency INTEGER,
status TEXT
)
""")
cursor.execute("INSERT INTO proxies (proxy, latency, status) VALUES (?, ?, 'working')", (proxy, latency))
conn.commit()
except Exception as e:
logging.error(f"Error saving proxy to database: {e}")
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle the !proxy command.
@ -156,58 +87,34 @@ async def handle_command(room, message, bot, prefix, config):
if match.is_not_from_this_bot() and match.prefix() and match.command("proxy"):
logging.info("Received !proxy command")
# Check database for a working proxy
working_proxy, latency = check_db_for_proxy()
if working_proxy:
await bot.api.send_markdown_message(room.room_id,
f"✅ Using cached working SOCKS5 Proxy: **{working_proxy}** - Latency: **{latency} ms**")
logging.info(f"Using cached working SOCKS5 proxy {working_proxy}")
# Download proxy list if needed
if not await download_proxy_list():
await bot.api.send_markdown_message(room.room_id, "Error downloading proxy list")
logging.error("Error downloading proxy list")
return
# Download proxy list if needed
else:
if not await download_proxy_list():
await bot.api.send_markdown_message(room.room_id, "Error downloading proxy list")
logging.error("Error downloading proxy list")
return
try:
# Read proxies from file
with open(PROXY_LIST_FILENAME, 'r') as f:
socks5_proxies = f.read().splitlines()
random.shuffle(socks5_proxies)
try:
# Read proxies from file
with open(PROXY_LIST_FILENAME, 'r') as f:
socks5_proxies = [line.replace("socks5://", "") for line in f.read().splitlines()]
random.shuffle(socks5_proxies)
# Test proxies concurrently
tested_proxies = 0
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
futures = []
for proxy in socks5_proxies[:MAX_TRIES]:
futures.append(executor.submit(test_proxy, proxy))
for future in concurrent.futures.as_completed(futures):
success, proxy, latency = future.result()
if success:
await bot.api.send_markdown_message(room.room_id,
f"✅ Anonymous SOCKS5 Proxy: **{proxy}** - Latency: **{latency} ms**")
logging.info(f"Sent SOCKS5 proxy {proxy} to the room")
save_proxy_to_db(proxy, latency) # Save working proxy to the database
tested_proxies += 1
if tested_proxies >= MAX_PROXIES_IN_DB:
break # Stop testing proxies once MAX_PROXIES_IN_DB are saved to the database
# Check database for a working proxy after testing
working_proxy, latency = check_db_for_proxy()
if working_proxy:
await bot.api.send_markdown_message(room.room_id,
f"✅ Using cached working SOCKS5 Proxy: **{working_proxy}** - Latency: **{latency} ms**")
logging.info(f"Using cached working SOCKS5 proxy {working_proxy}")
else:
# If no working proxy found after testing
await bot.api.send_markdown_message(room.room_id, "❌ No working anonymous SOCKS5 proxy found")
logging.info("No working anonymous SOCKS5 proxy found")
except Exception as e:
logging.error(f"Error handling !proxy command: {e}")
await bot.api.send_markdown_message(room.room_id, "❌ Error handling !proxy command")
# Test proxies
socks5_proxy = None
for proxy in socks5_proxies[:MAX_TRIES]:
success, latency = await test_socks5_proxy(proxy)
if success:
socks5_proxy = proxy
break
# Send the first working anonymous proxy of each type to the Matrix room
if socks5_proxy:
await bot.api.send_markdown_message(room.room_id, f"✅ Anonymous SOCKS5 Proxy: **{socks5_proxy}** - Latency: **{latency} ms**")
logging.info(f"Sent SOCKS5 proxy {socks5_proxy} to the room")
else:
await bot.api.send_markdown_message(room.room_id, "❌ No working anonymous SOCKS5 proxy found")
logging.info("No working anonymous SOCKS5 proxy found")
except Exception as e:
logging.error(f"Error handling !proxy command: {e}")
await bot.api.send_markdown_message(room.room_id, "❌ Error handling !proxy command")

178
plugins/stable-diffusion.py
View File

@ -1,178 +0,0 @@
"""
Plugin for generating images using self-hosted Stable Diffusion and sending them to a Matrix chat room.
"""
import requests
import base64
from asyncio import Queue
import argparse
import simplematrixbotlib as botlib
import markdown2
from slugify import slugify
def slugify_prompt(prompt):
"""
Generates a URL-friendly slug from the given prompt.
Args:
prompt (str): The prompt to slugify.
Returns:
str: A URL-friendly slug version of the prompt.
"""
return slugify(prompt)
# Queue to store pending commands
command_queue = Queue()
def markdown_to_html(markdown_text):
"""
Converts Markdown text to HTML.
Args:
markdown_text (str): The Markdown text to convert.
Returns:
str: The HTML version of the input Markdown text.
"""
html_content = markdown2.markdown(markdown_text)
return html_content
async def process_command(room, message, bot, prefix, config):
"""
Asynchronously processes the commands received in the Matrix room.
Args:
room (Room): The Matrix room where the command was received.
message (Message): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.prefix() and match.command("sd"):
if command_queue.empty():
await handle_command(room, message, bot, prefix, config)
else:
await command_queue.put((room, message, bot, prefix, config))
async def handle_command(room, message, bot, prefix, config):
"""
Asynchronously handles the 'sd' command, which generates images using Stable Diffusion.
Args:
room (Room): The Matrix room where the command was received.
message (Message): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.prefix() and match.command("sd"):
try:
# Parse command-line arguments
parser = argparse.ArgumentParser(description='Generate images using self-hosted Stable Diffusion')
parser.add_argument('--steps', type=int, default=4, help='Number of steps, default=16')
parser.add_argument('--cfg', type=int, default=1.25, help='CFG scale, default=7')
parser.add_argument('--h', type=int, default=512, help='Height of the image, default=512')
parser.add_argument('--w', type=int, default=512, help='Width of the image, default=512')
parser.add_argument('--neg', type=str, default='((((ugly)))), (((duplicate))), ((morbid)), ((mutilated)), out of frame, extra fingers, mutated hands, ((poorly drawn hands)), ((poorly drawn face)), (((mutation))), (((deformed))), ((ugly)), blurry, ((bad anatomy)), (((bad proportions))), ((extra limbs)), cloned face, (((disfigured))), out of frame, ugly, extra limbs, (bad anatomy), gross proportions, (malformed limbs), ((missing arms)), ((missing legs)), (((extra arms))), (((extra legs))), mutated hands, (fused fingers), (too many fingers), (((long neck)))', nargs='+', help='Negative prompt, default=none')
parser.add_argument('--sampler', type=str, nargs='*', default=['DPM++', 'SDE', 'Karras'], help='Sampler name, default=Euler a')
parser.add_argument('prompt', type=str, nargs='*', help='Prompt for the image')
args = parser.parse_args(message.body.split()[1:]) # Skip the command itself
if not args.prompt:
raise argparse.ArgumentError(None, "Prompt is required.")
prompt = ' '.join(args.prompt)
sampler_name = ' '.join(args.sampler)
neg = ' '.join(args.neg)
payload = {
"prompt": prompt,
"steps": args.steps,
"negative_prompt": neg,
"sampler_name": sampler_name,
"cfg_scale": args.cfg,
"width": args.w,
"height": args.h,
}
url = "http://127.0.0.1:7860/sdapi/v1/txt2img"
# Send request to the Stable Diffusion API
response = requests.post(url=url, json=payload)
r = response.json()
# Slugify the prompt
prompt_slug = slugify(prompt[:120])
# Construct filename
filename = f"{prompt_slug}_{args.steps}_{args.h}x{args.w}_{sampler_name}_{args.cfg}.jpg"
with open(f"/tmp/{filename}", 'wb') as f:
f.write(base64.b64decode(r['images'][0]))
# Send the generated image to the Matrix room
await bot.api.send_image_message(room_id=room.room_id, image_filepath=f"/tmp/{filename}")
# Format and send information about the generated image
neg = neg.replace(" ", "")
info_msg = f"""<details><summary>🍄⤵︎Image Info:⤵︎</summary><strong>Prompt:</strong> {prompt_slug}<br><strong>Steps:</strong> {args.steps}<br><strong>Dimensions:</strong> {args.h}x{args.w}<br><strong>Sampler:</strong> {sampler_name}<br><strong>CFG Scale:</strong> {args.cfg}<br><strong>Negative Prompt: {neg}</strong></details>"""
await bot.api.send_markdown_message(room.room_id, info_msg)
except argparse.ArgumentError as e:
# Handle argument errors
await bot.api.send_text_message(room.room_id, f"Error: {e}")
await bot.api.send_markdown_message(room.room_id, "<details><summary>Stable Diffusion Help</summary>" + print_help() + "</details>")
except Exception as e:
# Handle general errors
await bot.api.send_text_message(room.room_id, f"Error processing the command: {str(e)}")
finally:
# Process next command from the queue, if any
if not command_queue.empty():
next_command = await command_queue.get()
await handle_command(*next_command)
def print_help():
"""
Generates help text for the 'sd' command.
Returns:
str: Help text for the 'sd' command.
"""
return """
<p>Generate images using self-hosted Stable Diffusion</p>
<p>Positional arguments:</p>
<ul>
<li>prompt - Prompt for the image</li>
</ul>
<p>Default Negative Prompts: ((((ugly)))), (((duplicate))), ((morbid)), ((mutilated)), out of frame, extra fingers, mutated hands, ((poorly drawn hands)), ((poorly drawn face)), (((mutation))), (((deformed))), ((ugly)), blurry, ((bad anatomy)), (((bad proportions))), ((extra limbs)), cloned face, (((disfigured))), out of frame, ugly, extra limbs, (bad anatomy), gross proportions, (malformed limbs), ((missing arms)), ((missing legs)), (((extra arms))), (((extra legs))), mutated hands, (fused fingers), (too many fingers), (((long neck)))</p>
<p>Optional arguments:</p>
<ul>
<li>--steps STEPS - Number of steps, default=16</li>
<li>--cfg CFG - CFG scale, default=7</li>
<li>--h H - Height of the image, default=512</li>
<li>--w W - Width of the image, default=512</li>
<li>--neg NEG - Negative prompt, default=none</li>
<li>--sampler SAMPLER - Sampler name, default=Euler a</li>
</ul>
<p>LORA List:</p>
<ul>
<li>&lt;psychedelicai-SDXL&gt;</li>
<li>&lt;ral-frctlgmtry-sdxl&gt;</li>
<li>&lt;SDXL-PsyAI-v4&gt;</li>
<li>&lt;al3xxl&gt;</li>
</ul>
<p>Load LORAs like this:</p>
<ul>
<li>&lt;lora:SDXL-PsyAI-v4:1&gt; PsyAI</li>
<li>&lt;lora:psychedelicai-SDXL:1&gt; Psychedelic</li>
<li>&lt;ral-frctlgmtry-sdxl&gt; ral-frctlgmtry</li>
<li>&lt;lora:al3xxl:1&gt; alexpainting, alexhuman, alexentity, alexthirdeye, alexforeheads, alexgalactic, spiraling, alexmirror, alexangel, alexkissing, alexthirdeye2, alexthirdeye3, alexhofmann, wearing, glasses, alexgalactic3, alexthirdeye4, alexhuman3, alexgalactic2, alexhuman2, alexbeing2, alexfractal2, alexfractal3</li>
</ul>
"""

112
plugins/youtube-preview.py
View File

@ -1,102 +1,26 @@
"""
Plugin for providing a command to fetch YouTube video information from links.
This plugin provides a command to fetch YouTube video information from links.
"""
# Importing necessary libraries
# plugins/youtube.py
import re
import logging
import asyncio
import aiohttp
from pytubefix import YouTube
from pytube import YouTube
import simplematrixbotlib as botlib
from youtube_title_parse import get_artist_title
LYRICIST_API_URL = "https://lyrist.vercel.app/api/{}"
def seconds_to_minutes_seconds(seconds):
"""
Converts seconds to a string representation of minutes and seconds.
Args:
seconds (int): The number of seconds.
Returns:
str: A string representation of minutes and seconds in the format MM:SS.
"""
minutes = seconds // 60
seconds %= 60
return f"{minutes:02d}:{seconds:02d}"
async def fetch_lyrics(song, artist):
"""
Asynchronously fetches lyrics for a song from the Lyricist API.
Args:
song (str): The name of the song.
artist (str): The name of the artist.
Returns:
str: Lyrics of the song.
None if an error occurs during fetching.
"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(LYRICIST_API_URL.format(song, artist)) as response:
data = await response.json()
return data.get("lyrics")
except Exception as e:
logging.error(f"Error fetching lyrics: {str(e)}")
return None
async def fetch_youtube_info(youtube_url):
"""
Asynchronously fetches information about a YouTube video.
Args:
youtube_url (str): The URL of the YouTube video.
Returns:
str: A message containing information about the YouTube video.
None if an error occurs during fetching.
"""
try:
video = YouTube(youtube_url)
title = video.title
artist, song = get_artist_title(title)
description = video.description
length = seconds_to_minutes_seconds(video.length)
views = video.views
author = video.author
description_with_breaks = description.replace('\n', '<br>')
# Fetching lyrics
lyrics = await fetch_lyrics(song, artist)
lyrics = lyrics.replace('\n', "<br>")
info_message = f"""<strong>🎬🎝 Title:</strong> {title} | <strong>Length</strong>: {length} minutes | <strong>Views</strong>: {views}\n<details><summary><strong>⤵︎Description⤵︎</strong></summary>{description_with_breaks}</details>"""
if lyrics:
info_message += f"<br><details><summary><strong>🎵 Lyrics:</strong></summary><br>{lyrics}</details>"
return info_message
except Exception as e:
logging.error(f"Error fetching YouTube video information: {str(e)}")
return None
async def handle_command(room, message, bot, prefix, config):
"""
Asynchronously handles the command to fetch YouTube video information.
Function to handle YouTube video information from links.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
Returns:
None
@ -109,16 +33,16 @@ async def handle_command(room, message, bot, prefix, config):
video_id = video_id_match.group(1)
youtube_url = f"https://www.youtube.com/watch?v={video_id}"
logging.info(f"Fetching information for YouTube video: {youtube_url}")
retry_count = 3
while retry_count > 0:
info_message = await fetch_youtube_info(youtube_url)
if info_message:
await bot.api.send_markdown_message(room.room_id, info_message)
logging.info("Sent YouTube video information to the room")
break
else:
logging.info("Retrying...")
retry_count -= 1
await asyncio.sleep(1) # wait for 1 second before retrying
else:
logging.error("Failed to fetch YouTube video information after retries")
try:
video = YouTube(youtube_url)
title = video.title
description = video.description
length = seconds_to_minutes_seconds(video.length)
views = video.views
author = video.author
info_message = f"""**🎬🎝 Title:** {title} | **Length**: {length} minutes| **Views:** {views} | **Description:** {description}"""
await bot.api.send_markdown_message(room.room_id, info_message)
logging.info("Sent YouTube video information to the room")
except Exception as e:
logging.error(f"Error fetching YouTube video information: {str(e)}")
# await bot.api.send__message(room.room_id, "Error fetching YouTube video information.")

39
plugins/youtube-search.py
View File

@ -1,25 +1,10 @@
"""
Plugin for providing a command to search for YouTube videos in the room.
"""
# plugins/youtube_search.py
import logging
import simplematrixbotlib as botlib
from youtube_search import YoutubeSearch
async def handle_command(room, message, bot, PREFIX, config):
"""
Asynchronously handles the command to search for YouTube videos in the room.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
PREFIX (str): The command prefix.
config (dict): The bot's configuration.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, PREFIX)
if match.is_not_from_this_bot() and match.prefix() and match.command("yt"):
args = match.args()
@ -28,7 +13,7 @@ async def handle_command(room, message, bot, PREFIX, config):
else:
search_terms = " ".join(args)
logging.info(f"Performing YouTube search for: {search_terms}")
results = YoutubeSearch(search_terms, max_results=10).to_dict()
results = YoutubeSearch(search_terms, max_results=3).to_dict()
if results:
output = generate_output(results)
await send_collapsible_message(room, bot, output)
@ -36,15 +21,6 @@ async def handle_command(room, message, bot, PREFIX, config):
await bot.api.send_text_message(room.room_id, "No results found.")
def generate_output(results):
"""
Generates HTML output for displaying YouTube search results.
Args:
results (list): A list of dictionaries containing information about YouTube videos.
Returns:
str: HTML formatted output containing YouTube search results.
"""
output = ""
for video in results:
output += f'<a href="https://www.youtube.com/watch?v={video["id"]}">'
@ -58,16 +34,5 @@ def generate_output(results):
async def send_collapsible_message(room, bot, content):
"""
Sends a collapsible message containing YouTube search results to the room.
Args:
room (Room): The Matrix room where the message will be sent.
bot (MatrixBot): The Matrix bot instance.
content (str): HTML content to be included in the collapsible message.
Returns:
None
"""
message = f'<details><summary><strong>🍄Funguy ▶YouTube Search🍄<br>⤵︎Click Here To See Results⤵︎</strong></summary>{content}</details>'
await bot.api.send_markdown_message(room.room_id, message)

9
requirements.txt
View File

@ -1,12 +1,5 @@
python-dotenv
requests
pytubefix
pytube
duckduckgo_search
nio
markdown2
watchdog
emoji
python-slugify
youtube_title_parse
dnspython