__version__ = "1.0.2 DOcker-SHELL update" + ". https://git.swee.codes/swee/MeowNex"
import socket
import subprocess
from time import sleep, time, ctime
from os import system as ossystem, path, environ, listdir, getcwd
import sqlite3
cwd = getcwd()
import re
from random import choice, randint as random, randrange
import traceback
import threading
from pathlib import Path
from urllib.parse import urlparse, parse_qs
from requests import get, exceptions as rex
from bs4 import BeautifulSoup
from googleapiclient.discovery import build
import uuid
import ssl
import requests
import docker
dock = docker.from_env()
global lastquery
AIurl = environ["AIurl"]
AIsession = requests.Session()
AIsession.auth = (environ["AIuser"], environ["AIpass"])
AImodel = environ["AImodel"] if "AImodel" in environ else "llama2"
run = 0
block = 0
parsed = 0
lastquery = "None yet."
# Dashboard thread
from flask import Flask, send_file, abort as abrt, redirect as redir
app = Flask(__name__)
responses_ai = {} # {uuid: response}
@app.route('/')
def index_html():
return send_file("dashboard.html", mimetype='text/html')
@app.route('/run.txt')
def runs():
return str(run)
@app.route('/block.txt')
def blocks():
return str(block)
@app.route('/link.txt')
def parses():
return str(parsed)
@app.route('/script.js')
def script_js():
return send_file("script.js", mimetype='application/javascript')
@app.route('/lastquery.txt')
def lastquery_txt():
return str(lastquery)
@app.route('/docs/config')
def config_docs():
return redir("https://git.swee.codes/swee/MeowNex/src/branch/main/docs/config.md")
@app.route('/response/<uuidd>')
def ai_response_truncate(uuidd):
if uuidd in responses_ai:
res = responses_ai[uuidd]
return f'<title>AI response {uuidd}</title>\n<meta name="viewport" content="width=device-width, initial-scale=1.0">\n<link rel="stylesheet" href="https://swee.codes/style.css" type="text/css">\n<body><center><h1>AI response {uuidd}</h1><br><br><pre class=code width="100%" style="word-wrap: break-word;white-space: pre-wrap;text-align: left;">' + res
else:
abrt(404)
threading.Thread(target=app.run, daemon=True, kwargs={"port": 2005}).start()
cntr = None
# YouTube API
DEVELOPER_KEY = environ["ytapi"]
headers = {
'User-Agent': 'MeowNexUS IRC ' + __version__
}
def get_yt_id(url):
query = urlparse(url)
# youtu.be already contains the ID in the path
if query.hostname == 'youtu.be': return query.path[1:]
if query.hostname in {'www.youtube.com', 'youtube.com', 'music.youtube.com'}:
# URLs that have the ID in the path instead of the query.
integrated_in_url = ["watch", "embed", "v", "shorts"]
try:
# The regular /watch path, which stores the ID in the query.
if query.path == '/watch': return parse_qs(query.query)['v'][0]
# Alternatively, it will get the ID in the path if the path was in the list above.
elif query.path.split('/')[1] in integrated_in_url: return query.path.split('/')[2]
except:
return None
class config:
def __init__(self):
self.conn = sqlite3.connect(environ["SBconfig"])
self.database = self.conn.cursor()
def perms(self, mask: str):
try:
self.database.execute(f"SELECT * FROM users;")
output = self.database.fetchall()
for i in output:
if re.match(i[0].replace("*", ".+"), mask):
return i[1]
return ''
except:
print(traceback.format_exc())
return ''
def chansettings(self, chan: str):
try:
self.database.execute("SELECT * FROM chans WHERE chan = ?;", [chan])
output = self.database.fetchall()
print(output)
temp = output[0][1].split(",")
return temp if temp != [''] else []
except:
print(traceback.format_exc())
return []
def setchanconfig(self, chan: str, flags: list):
try:
print(self.chansettings(chan))
if self.chansettings(chan) == []:
print("[!!!] Channel doesn't exist in config")
self.database.execute("INSERT INTO chans (chan, flags) values(?, ?);", [chan, "+config," + ",".join(flags)])
else:
self.database.execute("UPDATE chans SET FLAGS = ? WHERE chan = ?", [",".join(flags),chan])
self.conn.commit()
except:
print(traceback.format_exc())
return False
def cflagexist(self, chan:str, flag:str):
return flag in self.chansettings(chan)
# Code snippet from my buddy Irish
# Define a dictionary to map terminal color codes to IRC color codes
color_map = {
'\033[30m': '\x0301', # Black
'\033[31m': '\x0305', # Red
'\033[32m': '\x0303', # Green
'\033[33m': '\x0307', # Yellow
'\033[34m': '\x0302', # Blue
'\033[35m': '\x0306', # Magenta
'\033[36m': '\x0310', # Cyan
'\033[37m': '\x0315', # White
'\033[90m': '\x0314', # Bright Black (Gray)
'\033[91m': '\x0304', # Bright Red
'\033[92m': '\x0309', # Bright Green
'\033[93m': '\x0308', # Bright Yellow
'\033[94m': '\x0312', # Bright Blue
'\033[95m': '\x0313', # Bright Magenta
'\033[96m': '\x0311', # Bright Cyan
'\033[97m': '\x0316', # Bright White
'\033[0m': '\x03', # Reset
}
pattern = re.compile(r'\033\[\d+(;\d+)*m|\033\[\?25[lh]|\033\[\?7[lh]|\033\[\d+C|\033\[\d+A|\033\[\d+D|\033\[\d+B')
def humanbytes(B):
"""Return the given bytes as a human friendly KB, MB, GB, or TB string."""
B = float(B)
KB = float(1024)
MB = float(KB ** 2) # 1,048,576
GB = float(KB ** 3) # 1,073,741,824
TB = float(KB ** 4) # 1,099,511,627,776
if B < KB:
return '{0}{1}'.format(B,'Bytes' if 0 == B > 1 else 'Byte')
elif KB <= B < MB:
return '{0:.2f}KB'.format(B / KB)
elif MB <= B < GB:
return '{0:.2f}MB'.format(B / MB)
elif GB <= B < TB:
return '{0:.2f}GB'.format(B / GB)
elif TB <= B:
return '{0:.2f}TB'.format(B / TB)
def rplce(text, old, new, n):
parts = text.split(old)
if len(parts) <= n:
return text # Not enough occurrences to replace
return old.join(parts[:n]) + new + old.join(parts[n:])
def replace_color_codes(text):
def replacer(match):
code = match.group(0)
if code in color_map:
return color_map[code]
elif re.match(r'\033\[\d+C', code):
# Handle cursor move right (e.g., \033[30C)
spaces = int(re.findall(r'\d+', code)[0])
return ' ' * spaces
elif re.match(r'\033\[\d+A', code) or re.match(r'\033\[\d+B', code):
# Handle cursor move up (e.g., \033[17A) and down (e.g., \033[17B)
return ''
elif re.match(r'\033\[\d+D', code):
# Handle cursor move to the start of the line (e.g., \033[9999999D)
return ''
elif re.match(r'\033\[\?25[lh]', code):
# Handle cursor visibility (e.g., \033[?25l or \033[?25h)
return ''
elif re.match(r'\033\[\?7[lh]', code):
# Handle line wrapping (e.g., \033[?7l or \033[?7h)
return ''
else:
return ''
# Remove all control sequences
cleaned_text = pattern.sub(replacer, text)
# Split the text into lines and remove lines that contain control sequences or are empty
lines = cleaned_text.split('\n')
lines = [line for line in lines if line.strip() and not re.search(r'\033\[m', line)]
# Join the cleaned lines back into a single string
return '\n'.join(lines)
class bot_irc:
irc_socket = socket.socket()
def __init__(self):
self.irc_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def send_irc(self, channel, msg):
self.irc_socket.send(bytes("PRIVMSG " + channel + " :" + msg + "\n", "UTF-8"))
def connect_irc(self, server, port, channel, bot_nick, bot_pass, bot_nickpass):
print("Server connection: " + server)
self.irc_socket.connect((server, port))
self.irc_socket.send(
bytes(
"USER " + bot_nick + " " + bot_nick + " " + bot_nick + " :MeowNexUS, a very cool bot made by Swee :3\n",
"UTF-8",
)
)
self.irc_socket.send(bytes("NICK " + bot_nick + "\n", "UTF-8"))
self.irc_socket.send(
bytes("PASS " + bot_nickpass + ":" + bot_pass + "\n", "UTF-8")
)
for i in channel:
self.irc_socket.send(bytes("JOIN " + i + "\n", "UTF-8"))
def response_irc(self):
try:
r = self.irc_socket.recv(2040).decode()
if r.find("PING") != -1:
self.irc_socket.send(
bytes("PONG " + r.split()[1] + "\r\n", "UTF-8")
)
return r
except:
return ""
def ping(self):
try:
r = self.irc_socket.recv(2040).decode()
if r.find("PING") != -1:
self.irc_socket.send(
bytes("PONG " + r.split()[1] + "\r\n", "UTF-8")
)
except:
pass
server_irc = "127.0.0.1" # Use 127.0.0.1 for local ZNC
port_irc = 5000 # NO SSL FOR YOU
channel_irc = ["##sweezero"]
botnick_irc = environ.get('SBnick')
botnickpass_irc =environ.get('SBuser')
botpass_irc = environ.get('SBpass')
allowedparse = ["text/html", "application/x-httpd-php", "application/xhtml", "application/xhtml+xml"]
irc = bot_irc()
irc2 = bot_irc()
irc3 = bot_irc()
irc.connect_irc(
server_irc, port_irc, channel_irc, botnick_irc, botpass_irc, botnickpass_irc + "/A"
)
sbconfig = config()
threes = [":3", ":3c", "uwu", "owo", "/ᐠ｡ꞈ｡ᐟ\\", "(✿◠‿◠)"]
snacks = ["PepperMintHTMLs", "Dice App candies", "SweeCrypt codes", "Windows 11 Bloomberry flavored icecream"]
baps = ["T-T", "Ow!", "Beep!"]
pats = ["-w-", "Meep...", "Prr!"]
meows_happy = ['Meow!', 'Nyaa~', 'Mrow.', 'Prr! :3', "Mrrp?", "Mreow.", "!woeM", "3: !rrP", "~aayN", "Mew!", "Moew!"]
meows_upset = ['Hiss!', "!ssiH", "Grrr..."]
my_self = ["MeowNexUS", "MeowNexU5", "MeowN3xUS"]
logs = {} # {channel: [line, line]} max 10 lines
happiness = 5
global times
times = 0
ai_storage = {}
ai_quotes = ["Oki lemme think about this one...", "Oki gimme a sec...", "Processing...", "Stand by...", "Meow..."]
emoticons = {";3": "I think you forgot to press shift", ":#": "Wait no, don't hold shift THAT long!", ";#": "You are joking with me at this point."}
def irci2():
irc2.connect_irc(
server_irc, port_irc, channel_irc, "sweeB0t", botpass_irc, botnickpass_irc + "/B"
)
while True:
irc2.ping()
def irci3():
irc3.connect_irc(
server_irc, port_irc, channel_irc, "swe3Bot", botpass_irc, botnickpass_irc + "/C"
)
while True:
irc3.ping()
def multiline(text, channel):
global times
if text.__class__ == bytes:
text = text.decode()
for t in text.split("\n"):
if t != "" and t != " ":
if times > 8:
if times == 9:
sleep(1)
irc3.send_irc(channel, replace_color_codes(t))
times +=1
if times == 13:
times = 0
elif times > 4:
if times == 5:
sleep(1)
irc2.send_irc(channel, replace_color_codes(t))
times +=1
else:
if times==0:
sleep(1)
irc.send_irc(channel, replace_color_codes(t))
times += 1
from sys import exit
def restart():
if cntr != None:
cntr.kill()
cntr.remove()
irc.irc_socket.close()
irc2.irc_socket.close()
irc3.irc_socket.close()
exit(0)
def system(cmd, chan, rstrip=False):
try:
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
outpp = ""
for line in iter(p.stdout.readline, b''):
if text != "" and text != " ":
if not rstrip:
multiline(line.rstrip(), chan)
else:
outpp += line.decode().rstrip() + " "
if rstrip:
multiline(outpp[:-1], chan)
p.stdout.close()
p.wait()
except FileNotFoundError:
multiline(cmd[0] + " not found", chan)
except:
multiline(traceback.format_exc(), chan)
irl2 = threading.Thread(target=irci2, daemon=True)
irl2.start()
irl3 = threading.Thread(target=irci3, daemon=True)
irl3.start()
ticker=0
while True:
teext = irc.response_irc().split("\r\n")
for text in teext:
if text != "":
try:
print(text)
except:
print("")
try:
command = text.split("PRIVMSG")[1].split(":")[1:]
command = ":".join(command).split(" ")
channel = text.split("PRIVMSG")[1].split(" ")[1]
nick = text.split(" ")[0][1:].split("!")[0]
username = text.split(" ")[0][1:].split("@")[1]
mask = text.split(" ")[0][1:]
try:
perms = sbconfig.perms(mask)
except:
perms = ""
print(command)
print(channel)
print(nick)
print(username)
print(perms)
if not nick in ["MeowNexUS", "MeowNexU5", "MeowN3xUS"]:
cont = " ".join(command)
if not channel in logs:
logs[channel] = [{"nick": nick, "content": cont}]
else:
logs[channel].append({"nick": nick, "content": cont})
if len(logs[channel]) > 128: logs[channel] = logs[channel][::-1][:128][::-1]
#open("log-text-"+channel, "a").write(" ".join(command) + "\n")
#open("log-name-"+channel, "a").write(nick + "\n")
except:
print(traceback.format_exc())
def replased(expression):
if channel in logs:
try:
if expression.split("/")[0].lower() != "s": raise IndexError()
find = expression.split("/")[1]
replace = expression.split("/")[2]
user = None
repeat = False
index = 1
try:
extraflags = expression.split("/")[3:]
for i in extraflags:
if i.strip() == "":
pass
elif i.lower() == "g":
repeat = True
elif i.strip().isdigit():
print(f"index = {int(i)}")
index = int(i)
else:
print(f"user = {i.lower()}")
user = i.lower()
except: pass
cache = None
for i in logs[channel][::-1]:
if find in i["content"] and (user == None or i["nick"].lower() == user) and not "ignore" in i:
cache = i
break
if cache == None:
raise Exception("The specified text couldn't be found")
else:
if repeat:
cache["content"] = cache["content"].replace(find, replace)
else:
cache["content"] = rplce(cache["content"], find, replace, index)
cachenick = cache["nick"]
cachecontent = cache["content"]
multiline(f"[RPL] <{cachenick}> {cachecontent}", channel)
except IndexError:
irc.send_irc(channel, nick + ": [ERR] Sed expression might be incorrectly written!")
except Exception as ex:
irc.send_irc(channel, nick + ": [ERR] " + str(ex))
else:
irc.send_irc(channel, nick + ": [!!!] No logs saved in " + channel)
try:
#if True:
if "PRIVMSG" in text and (not "ignore" in sbconfig.perms(mask)) and not nick in my_self:
if "PRIVMSG" in text and command[0][0] == "$":
logs[channel] = logs[channel][:-1]
run+=1
if command[0] == "$ping":
if random(1,2) == 1:
irc.send_irc(channel, nick + ": Pnog")
else:
irc.send_irc(channel, nick + ": Pong")
elif command[0] == "$help":
if len(command) > 1:
if "." in " ".join(command[1:]):
irc.send_irc(channel, nick + ": Nice try")
block+=1
run-=1
else:
try:
if path.isfile(cwd + "/helps/" + " ".join(command[1:])):
output = open(cwd + "/helps/" + " ".join(command[1:])).read()
multiline(output, nick if sbconfig.cflagexist(channel, "-multiline") else channel)
else:
irc.send_irc(channel, "Either the command isn't documented yet or it doesn't exist lol")
except:
multiline(output, channel)
else:
irc.send_irc(channel, "tip, ping, whoami, perms, version, figlet, tdfiglet, cowsay, uptime, cc, joke, botsnack. (restart, join, part, shell, config, pull.) Use '$help (command)' for explanation.")
elif command[0] == "$tip":
system(["/usr/games/fortune", "debian-hints"], channel, True)
elif command[0] == "$whoami":
if perms != "":
irc.send_irc(channel, nick + ": " + mask + " (Your hostmask has permissions, use the 'perms' command to see)")
else:
irc.send_irc(channel, nick + ": " + mask)
elif command[0] == "$welcome":
name = nick if not len(command) > 1 else " ".join(command[1:])
if random(1,2) == 1:
irc.send_irc(channel, f"Heyo! Welcome to the channel, {name}")
else:
irc.send_irc(channel, f"Welcome, {name}")
elif command[0] == "$perms":
if len(command) == 2:
if sbconfig.perms(command[1]) != "":
irc.send_irc(channel, nick + ": permissions of hostmask " + command[1] + ": " + sbconfig.perms(command[1]))
else:
irc.send_irc(channel, nick + ": The hostmask wildcard " + command[1] + " Doesn't have permissions.")
else:
if perms != "":
irc.send_irc(channel, nick + ": " + perms)
else:
irc.send_irc(channel, nick + ": none")
elif command[0] == "$restart":
if perms == "full" or "reboot" in perms.split(",") or "restart" in perms.split(","):
irc.send_irc(channel, nick + ": Restarting bot...")
sleep(1)
restart()
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$pull":
if perms == "full" or "git" in perms.split(","):
ossystem("git fetch")
ossystem("git pull")
irc.send_irc(channel, nick + ": Pulled from Git, restarting bot...")
restart()
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$join":
if perms == "full" or "join" in perms.split(","):
if len(command) == 2:
irc.send_irc(channel, nick + ": Joining " + command[1])
irc.irc_socket.send(bytes("JOIN " + command[1] + "\n", "UTF-8"))
irc2.irc_socket.send(bytes("JOIN " + command[1] + "\n", "UTF-8"))
irc3.irc_socket.send(bytes("JOIN " + command[1] + "\n", "UTF-8"))
multiline(channel=command[1], text="A wild bot appears! ($join initiated by " + nick + ")")
else:
if len(command) == 1:
irc.send_irc(channel, nick + ": I NEED A CHANNEL NAME TO JOIN!!!")
else:
irc.send_irc(channel, nick + ": Y'know this only uses one argument? JUST THE CHANNEL NAME!!!")
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$part":
if perms == "full" or "part" in perms.split(","):
if len(command) == 2:
irc.send_irc(channel, nick + ": Parting " + command[1])
irc.irc_socket.send(bytes("PART " + command[1] + "\n", "UTF-8"))
irc2.irc_socket.send(bytes("PART " + command[1] + "\n", "UTF-8"))
irc3.irc_socket.send(bytes("PART " + command[1] + "\n", "UTF-8"))
else:
if len(command) == 1:
irc.send_irc(channel, "Bye bye!")
sleep(1)
irc.irc_socket.send(bytes("PART " + channel + "\n", "UTF-8"))
irc2.irc_socket.send(bytes("PART " + channel + "\n", "UTF-8"))
irc3.irc_socket.send(bytes("PART " + channel + "\n", "UTF-8"))
else:
irc.send_irc(channel, nick + ": Y'know this only uses one or zero arguments?")
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$ai-chat":
try:
multiline(choice(ai_quotes), channel)
prettylogs = ""
for i in logs[channel][-15:]:
prettylogs += i["nick"] + ": " + i["content"] + "\n"
prettylogs = prettylogs.strip()
chatquery = f"--Details--\nYour happiness level is {happiness}/10, your emotion level will impact the conversation's tone.\nYour name is MeowNexUS, you're a bot who acts like an anthropomorphic cat/furry\nYour beloved creator is named Swee.\nDo not start your reply with your name, people already know who you are.\nYou should not use ANY emojis at all, instead, use the much cuter ASCII emoticons and kaomojis such as :3\nIf the user's message contains ^ that means the person wants more details from the last message in the provided history.\nYou will be provided with the last 15 lines of this channel's chat logs to improve the conversation's context, Don't mind if your own messages have mutliple lines, and do not repeat any of the messages listed, it will be considered annoying.\nThe bottom of the chat history in the prompt will be the main question\nThis chat is on IRC, you should answer the question in the simplest way possible\n--Chat on {channel}--\n{prettylogs}\n--main question--\n{nick}: " + " ".join(command)
postAttempt = AIsession.post(AIurl, json={"model": AImodel, "prompt": chatquery, "stream": False})
print(postAttempt.content)
if not postAttempt.ok:
raise Exception("API Returned non-okay status code: " + str(postAttempt.status_code))
aimsg = postAttempt.json()["response"]
randomname = str(uuid.uuid4())
responses_ai[randomname] = aimsg
mtline = ""
lastquery = f'<p>Channel: {channel}</p><p>Request:</p><br><br><pre class=code width="100%" style="word-wrap: break-word;white-space: pre-wrap;text-align: left;">{chatquery}</pre><hr><p>Response</p><br><br><pre class=code width="100%" style="word-wrap: break-word;white-space: pre-wrap;text-align: left;">{aimsg}</pre>'
broke = False
logs[channel].append({"nick": nick, "content": " ".join(command)})
logs[channel].append({"nick": "MeowNexUS", "content": aimsg, "ignore": True})
for num, i in enumerate(aimsg.split("\n")):
if num == 5:
broke = True
break
elif len(i) > 256:
mtline += "\n" + i[:256] + "..."
broke = True
else:
mtline += "\n" + i
if broke:
mtline += f"\nView full response: https://irc-bot.swee.codes/response/{randomname}"
multiline(mtline, channel)
except Exception as ex:
multiline(f"Something went wrong while connecting to the AI server: {ex}", channel)
print(traceback.format_exc())
#elif command[0] == "$ai-clear":
# if channel in ai_storage:
# del ai_storage[channel]
# multiline("Done.", channel)
# else:
# multiline(f"There is no existing AI conversation in {channel}", channel)
elif command[0] == "$socket":
if perms == "full":
if len(command) > 1:
irc.send_irc(channel, nick + ": Sending to socket...")
irc.irc_socket.send(bytes(" ".join(command[1:]) + "\n", "UTF-8"))
irc2.irc_socket.send(bytes(" ".join(command[1:]) + "\n", "UTF-8"))
irc3.irc_socket.send(bytes(" ".join(command[1:]) + "\n", "UTF-8"))
else:
irc.send_irc(channel, nick + ": what to send to the server?")
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$patpat" or command[0] == "$headpat":
if len(command) == 2:
if int(command[1]) < 50:
for i in range(0,int(command[1])):
happiness += (1 if happiness != 10 else 0)
else:
happiness = 10
elif len(command) == 1:
happiness += (1 if happiness != 10 else 0)
if len(command) != 2 or int(command[1]) < 50:
multiline(choice(pats) + " (Emotion: Upset " +
("#" * int(happiness)) +
("-" * int(10-happiness)) + " Happy)",
channel)
else:
multiline("Okay that's enough!!" + " (Emotion: Upset " +
("#" * int(happiness)) +
("-" * int(10-happiness)) + " Happy)",
channel)
elif command[0] == "$emotion":
multiline("Emotion: Upset " +
("#" * int(happiness)) +
("-" * int(10-happiness)) + " Happy",
channel)
elif command[0] == "$bap":
if len(command) == 2:
if int(command[1]) < 50:
for i in range(0,int(command[1])):
happiness -= (1 if happiness != 0 else 0)
else:
happiness = 0
elif len(command) == 1:
happiness -= (1 if happiness != 0 else 0)
if len(command) != 2 or int(command[1]) < 50:
multiline(choice(baps) + " (Emotion: Upset " +
("#" * int(happiness)) +
("-" * int(10-happiness)) + " Happy)",
channel)
else:
multiline("Okay that's enough!!" + " (Emotion: Upset " +
("#" * int(happiness)) +
("-" * int(10-happiness)) + " Happy)",
channel)
elif command[0] == "$meow":
if random(1,100) > (happiness*10):
multiline(choice(meows_upset), channel)
else:
multiline(choice(meows_happy), channel)
elif command[0] == "$stats":
multiline(f"{nick}: {run} command(s) run, {block} refused, and {parsed} link(s) parsed, check out https://irc-bot.swee.codes", channel)
elif command[0] == "$shell":
if perms == "full":
#or "shell" in perms.split(",")
if len(command) > 1:
try:
system(command[1:], channel)
except Exception as ex:
multiline(traceback.format_exc(), channel)
else:
irc.send_irc(channel, nick + ": What do you want me to F-ing run in the shell?")
else:
irc.send_irc(channel, nick + ": Permission denied.")
block+=1
run-=1
elif command[0] == "$dopull":
irc.send_irc(channel, nick + ": [...] Please wait, pulling docker image...")
image = dock.images.pull('alpine', tag="edge")
irc.send_irc(channel, nick + ": [OK] Pulled docker image, running command now.")
elif command[0] == "$doshell":
if len(command) > 1:
try:
print("Running container")
if cntr == None:
cntr = dock.containers.run('alpine:edge', "/bin/ash", detach=True, tty=True)
print("Executing...")
_, stream = cntr.exec_run(cmd=["timeout", "10", "ash", "-c", " ".join(command[1:])], tty=False)
multiline(replace_color_codes(stream.decode()), channel)
except Exception as ex:
irc.send_irc(channel, nick + ": [ERR] Traceback:")
multiline(traceback.format_exc(), channel)
else:
irc.send_irc(channel, nick + ": [ERR] No command specified, aborting")
elif command[0] == "$doreset":
if cntr == None:
irc.send_irc(channel, nick + ": [OK] No container is loaded yet.")
else:
cntr.kill()
cntr.remove()
cntr = None
irc.send_irc(channel, nick + ": [OK] Done.")
elif command[0] == "$version":
irc.send_irc(channel, "This is MeowNexUS IRC " + __version__)
elif command[0] == "$figlet":
if not sbconfig.cflagexist(channel, "-multiline"):
if len(command) > 1:
try:
system(['figlet'] + command[1:], channel)
except Exception as ex:
irc.send_irc(channel, nick + ": " + ex.__class__.__name__)
else:
irc.send_irc(channel, nick + ": What text should I enlarge?")
else:
multiline("Multiline commands are disabled in this channel.", channel)
elif command[0] == "$tdfiglet":
if not sbconfig.cflagexist(channel, "-multiline"):
if len(command) > 1:
try:
system(['tdfiglet', '-cm', "-r"] + command[1:], channel)
except Exception as ex:
irc.send_irc(channel, nick + ": " + ex.__class__.__name__)
else:
irc.send_irc(channel, nick + ": What text should I enlarge?")
else:
multiline("Multiline commands are disabled in this channel.", channel)
elif command[0] == "$cowsay":
if not sbconfig.cflagexist(channel, "-multiline"):
if len(command) > 1:
try:
system(['/usr/games/cowsay'] + command[1:], channel)
except Exception as ex:
irc.send_irc(channel, nick + ": " + ex.__class__.__name__)
else:
irc.send_irc(channel, nick + ": What should the cow say?")
else:
multiline("Multiline commands are disabled in this channel.", channel)
elif command[0] == "$uptime":
try:
system(['uptime', '-p'], channel)
except Exception as ex:
irc.send_irc(channel, nick + ": " + ex.__class__.__name__)
elif command[0] == "$joke":
try:
system(['pyjoke'] + command[1:], channel)
except Exception as ex:
multiline(traceback.format_exc(), channel)
elif command[0] == "$botsnack":
multiline(f"\x01ACTION eats some {choice(snacks)}\x01", channel)
elif command[0] == "$hug":
name = nick if not len(command) > 1 else " ".join(command[1:])
multiline(f"\x01ACTION gives a hug to {name}\x01", channel)
elif command[0] == "$sed":
if len(command) == 2:
replased(" ".join(command[1:]))
else:
irc.send_irc(channel, nick + ": [???] This command takes only one argument.")
elif command[0] == "$config":
try:
if len(command) > 1:
if perms == "full":
abort = False
flag = sbconfig.chansettings(channel)
for i in command[1:]:
if i == "clear":
flag = ["+config"]
elif i[0] == "+":
if "-" + i[1:] in flag:
flag[flag.index("-" + i[1:])] = i
else:
flag.append(i)
elif i[0] == "-":
if "+" + i[1:] in flag:
flag[flag.index("+" + i[1:])] = i
else:
flag.append(i)
else:
multiline(nick + ": flag name should begin with + or -", channel)
abort = True
break
if not abort:
print(flag)
sbconfig.setchanconfig(channel, flag)
multiline(nick + ": Successfuly applied configuration: " + " ".join(flag), channel)
else:
irc.send_irc(channel, nick + ": Permission denied")
block+=1
run-=1
else:
multiline("Configuration for " + channel + ": " + " ".join(sbconfig.chansettings(channel)), channel)
except ex:
multiline(str(ex), channel)
else:
if ".." in command[0][:3]:
irc.send_irc(channel, nick + ": Nice try.")
block+=1
run-=1
else:
if path.isfile("cc/" + command[0][1:]):
try:
if sbconfig.cflagexist(channel, "-multiline") and "#require-multiline" in open("cc/" + command[0][1:]).read():
irc.send_irc(channel, "Multiline commands are disabled in this channel.")
else:
com = ['python3']
comm = command
com.append("cc/" + comm[0][1:])
com.append(nick)
com.append(perms)
comm.pop(0)
for i in comm:
com.append(i)
system(com, channel)
except Exception as ex:
irc.send_irc(channel, nick + ": " + ex.__class__.__name__)
elif path.isdir("cc/" + command[0][1:]):
irc.send_irc(channel, nick + ": Command list under cc/debug: " + ", ".join(listdir("cc/" + command[0][1:])) + ".")
else:
print(nick + ": Unrecognised command")
elif " ".join(command[0:3]).lower() == "i am meownexus":
irc.send_irc(channel, "No I am!")
elif " ".join(command[0:3]).lower() == "no i am!":
if ticker >= 5:
system(['figlet', 'NO I AM!'], channel)
ticker = 0
else:
irc.send_irc(channel, "No I am!")
ticker+=1
elif command[0] == ":3":
if not sbconfig.cflagexist(channel, "-colonthree"):
irc.send_irc(channel, ":3")
elif command[0] in emoticons:
if not sbconfig.cflagexist(channel, "-colonthree"):
irc.send_irc(channel, emoticons[command[0]])
elif ":3c" in command:
if sbconfig.cflagexist(channel, "+:3c") or not sbconfig.cflagexist(channel, "-colonthree"):
multiline(choice(threes), channel)
elif command[0][:2].lower() == "s/":
logs[channel] = logs[channel][:-1]
if sbconfig.cflagexist(channel, "+sed"):
replased(" ".join(command))
# try to parse and find a link
if sbconfig.cflagexist(channel, "+links"):
try:
for i in command:
parse = urlparse(i)
if parse.scheme in ["http", "https"]:
try:
if parse.hostname in ["youtube.com", "youtu.be", "www.youtube.com", "m.youtube.com", "youtube-nocookie.com"] and get_yt_id(i) != None:
try:
video_id = get_yt_id(i)
youtube = build('youtube', 'v3', developerKey=DEVELOPER_KEY)
request = youtube.videos().list(part='snippet,statistics', id=video_id)
details = request.execute()
title = details['items'][0]['snippet']['title']
channel_yt = details['items'][0]['snippet']['channelTitle']
views = details['items'][0]['statistics']['viewCount']
multiline(f"({nick}) [▶️ YouTube] {title} | Author: {channel_yt} | {views} Views", channel)
except Exception as ex:
multiline("(" + nick + ") [YouTube Error, is it a valid YouTube URL?]", channel)
print(traceback.format_exc())
else:
e = get(i, headers=headers, timeout=10)
header = e.headers
content_type = header.get('content-type').split(";")[0]
content_len = header.get('Content-length')
if content_type in allowedparse:
if e.ok:
soup = BeautifulSoup(e.text, 'html.parser')
multiline("(" + nick + ") " + (" ".join(soup.title.string.splitlines())[:256] if soup.title != None else "[No title provided]"), channel)
else:
multiline("(" + nick + ") [HTTP " + str(e.status_code) + "]", channel)
else:
multiline("(" + nick + ") [" + humanbytes(content_len) + " " + str(content_type) + "]", channel)
parsed += 1
except rex.SSLError as ex:
multiline("(" + nick + ") [SSL Error: " + str(ex.message) + "]", channel)
except Exception as ex:
multiline("(" + nick + ") [Request error: " + str(ex.message) + "]", channel)
except:
print(traceback.format_exc())
if sbconfig.cflagexist(channel, "+links") or sbconfig.cflagexist(channel, "+gemini"):
try:
for i in command:
parse = urlparse(i)
unparsed = i
while True:
if parse.scheme == "gemini":
contx = ssl.create_default_context()
contx.check_hostname = False
contx.verify_mode = ssl.CERT_NONE
gsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
gemsocket = contx.wrap_socket(gsocket, server_hostname=parse.hostname)
gemsocket.connect((parse.hostname, 1965))
gemsocket.send(bytes(unparsed + "\r\n", "UTF-8"))
received = ""
while True:
gemresponse = gemsocket.recv(2048)
if gemresponse:
received += gemresponse.decode()
else:
break
received = received.replace("\r", "")
firstline = True
title = False
redirected = False
for i in received.split("\n"):
if firstline:
if i.split(" ")[0][0] == "3":
redirected = True
parsedd = urlparse(i.split(" ")[1])
unparsed = i.split(" ")[1]
break
elif i.split(" ")[0][0] != "2":
title = True
multiline("(" + nick + f") [{i}]", channel)
break
else:
firstline = False
if i.split(" ")[1].split(";")[0] != "text/gemini":
typee = i.split(" ")[1]
multiline("(" + nick + f") [Non-Gemtext file: {typee}]", channel)
title = True
break
else:
try:
if i.strip()[0] == "#" and i.strip()[1] == " ":
print("Found title: " + i)
title = True
multiline("(" + nick + f") " + i.strip()[2:].strip(), channel)
break
except:
pass
if not title and not redirected:
multiline("(" + nick + ") [No title found]", channel)
if not redirected:
break
else:
break
except Exception as ex:
exc = str(ex)
print(traceback.format_exc())
multiline("(" + nick + f") [Request error: {exc}]", channel)
elif "JOIN" in text and "#nixsanctuary" in text:
nick = text.split(":")[1].split("!")[0]
if not "Meow" in nick:
irc.send_irc("##hiya", "hiya: " + nick + " has joined #nixsanctuary")
pass
except Exception as ex:
print(traceback.format_exc())