Add Copilot image scraping, update and fix telegram async, improve shared API

This commit is contained in:
2024-06-19 01:40:33 +02:00
parent 09cf925850
commit 4c403e516b
21 changed files with 386 additions and 98 deletions

3
.gitignore vendored
View File

@ -1,5 +1,8 @@
/Config.py
/Database.sqlite
/Dump.txt
/Log.txt
/Selenium-WinDog/
/downloaded_files
/session.txt
*.pyc

View File

@ -2,33 +2,31 @@
# WinDog multi-purpose chatbot #
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start # """
# If you have modified the bot's code, you should set this
ModifiedSourceUrl = ""
# Only for the platforms you want to use, uncomment the below credentials and fill with your own:
LogToConsole = True
LogToFile = True
# MastodonUrl = "https://mastodon.example.com"
# MastodonToken = ""
# MatrixUrl = "https://matrix.example.com"
# MatrixUsername = "username"
# MatrixPassword = "hunter2"
# TelegramToken = "1234567890:abcdefghijklmnopqrstuvwxyz123456789"
DumpToConsole = False
DumpToFile = False
AdminIds = [ "123456789@telegram", "634314973@telegram", "admin@activitypub@mastodon.example.com", ]
DefaultLang = "en"
Debug = False
Dumper = False
CmdPrefixes = ".!/"
# False: ASCII output; True: ANSI Output (must be escaped)
ExecAllowed = {"date": False, "fortune": False, "neofetch": True, "uptime": False}
WebUserAgent = "WinDog v.Staging"
ModuleGroups = (ModuleGroups | {
#ModuleGroups = (ModuleGroups | {
ModuleGroups = {
"Basic": "",
"Geek": "",
})
}
# Only for the platforms you want to use, uncomment the below credentials and fill with your own:
""" # end windog config # """

0
LibWinDog/Database.py Normal file → Executable file
View File

11
LibWinDog/Platforms/Mastodon/Mastodon.py Normal file → Executable file
View File

@ -3,6 +3,13 @@
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start #
# MastodonUrl = "https://mastodon.example.com"
# MastodonToken = ""
# end windog config # """
MastodonUrl, MastodonToken = None, None
import mastodon
@ -15,7 +22,7 @@ def MastodonMain() -> bool:
class MastodonListener(mastodon.StreamListener):
def on_notification(self, event):
if event['type'] == 'mention':
OnMessageReceived()
#OnMessageParsed()
message = BeautifulSoup(event['status']['content'], 'html.parser').get_text(' ').strip().replace('\t', ' ')
if not message.split('@')[0]:
message = ' '.join('@'.join(message.split('@')[1:]).strip().split(' ')[1:]).strip()
@ -24,7 +31,7 @@ def MastodonMain() -> bool:
if command:
command.messageId = event['status']['id']
if command.Name in Endpoints:
Endpoints[command.Name]({"Event": event, "Manager": Mastodon}, command)
Endpoints[command.Name]["handler"]({"Event": event, "Manager": Mastodon}, command)
Mastodon.stream_user(MastodonListener(), run_async=True)
return True

0
LibWinDog/Platforms/Mastodon/requirements.txt Normal file → Executable file
View File

14
LibWinDog/Platforms/Matrix/Matrix.py Normal file → Executable file
View File

@ -3,6 +3,14 @@
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start #
# MatrixUrl = "https://matrix.example.com"
# MatrixUsername = "username"
# MatrixPassword = "hunter2"
# end windog config # """
MatrixUrl, MatrixUsername, MatrixPassword = None, None, None
import nio
@ -19,12 +27,10 @@ def MatrixMain() -> bool:
pass
#print(message)
#match = MatrixBotLib.MessageMatch(room, message, MatrixBot)
#OnMessageReceived()
#OnMessageParsed()
#if match.is_not_from_this_bot() and match.command("windogtest"):
# pass #await MatrixBot.api.send_text_message(room.room_id, " ".join(arg for arg in match.args()))
def runMatrixBot() -> None:
MatrixBot.run()
Thread(target=runMatrixBot).start()
Thread(target=lambda:MatrixBot.run()).start()
return True
def MatrixSender() -> None:

0
LibWinDog/Platforms/Matrix/requirements.txt Normal file → Executable file
View File

103
LibWinDog/Platforms/Telegram/Telegram.py Normal file → Executable file
View File

@ -3,10 +3,18 @@
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start #
# TelegramToken = "1234567890:abcdefghijklmnopqrstuvwxyz123456789"
# end windog config # """
TelegramToken = None
import telegram, telegram.ext
from telegram import ForceReply, Bot
from telegram import ForceReply, Bot #, Update
#from telegram.helpers import escape_markdown
#from telegram.ext import Application, filters, CommandHandler, MessageHandler, CallbackContext
from telegram.utils.helpers import escape_markdown
from telegram.ext import CommandHandler, MessageHandler, Filters, CallbackContext
@ -15,20 +23,38 @@ def TelegramMain() -> bool:
return False
updater = telegram.ext.Updater(TelegramToken)
dispatcher = updater.dispatcher
dispatcher.add_handler(MessageHandler(Filters.text | Filters.command, TelegramHandler))
dispatcher.add_handler(MessageHandler(Filters.text | Filters.command, TelegramHandlerWrapper))
updater.start_polling()
#app = Application.builder().token(TelegramToken).build()
#app.add_handler(MessageHandler(filters.TEXT | filters.COMMAND, TelegramHandler))
#app.run_polling(allowed_updates=Update.ALL_TYPES)
return True
def TelegramHandler(update:telegram.Update, context:CallbackContext=None) -> None:
if not (update and update.message):
def TelegramHandlerWrapper(update:telegram.Update, context:CallbackContext=None) -> None:
Thread(target=lambda:TelegramHandlerCore(update, context)).start()
def TelegramHandlerCore(update:telegram.Update, context:CallbackContext=None) -> None:
if not update.message:
return
OnMessageReceived()
data = SimpleNamespace()
data.room_id = f"{update.message.chat.id}@telegram"
data.message_id = f"{update.message.message_id}@telegram"
data.text_plain = update.message.text
data.text_markdown = update.message.text_markdown_v2
data.text_auto = GetWeightedText(data.text_markdown, data.text_plain)
data.command = ParseCommand(data.text_plain)
data.user = SimpleNamespace()
data.user.name = update.message.from_user.first_name
data.user.tag = update.message.from_user.username
data.user.id = f"{update.message.from_user.id}@telegram"
OnMessageParsed(data)
cmd = ParseCmd(update.message.text)
if cmd:
cmd.command = data.command
cmd.messageId = update.message.message_id
cmd.TextPlain = cmd.Body
cmd.TextMarkdown = update.message.text_markdown_v2
cmd.Text = GetWeightedText((cmd.TextMarkdown, cmd.TextPlain))
cmd.Text = GetWeightedText(cmd.TextMarkdown, cmd.TextPlain)
if cmd.Tokens[0][0] in CmdPrefixes and cmd.Name in Endpoints:
cmd.User = SimpleNamespace(**{
"Name": update.message.from_user.first_name,
@ -41,36 +67,57 @@ def TelegramHandler(update:telegram.Update, context:CallbackContext=None) -> Non
"Body": update.message.reply_to_message.text,
"TextPlain": update.message.reply_to_message.text,
"TextMarkdown": update.message.reply_to_message.text_markdown_v2,
"Text": GetWeightedText((update.message.reply_to_message.text_markdown_v2, update.message.reply_to_message.text)),
"Text": GetWeightedText(update.message.reply_to_message.text_markdown_v2, update.message.reply_to_message.text),
"User": SimpleNamespace(**{
"Name": update.message.reply_to_message.from_user.first_name,
"Tag": update.message.reply_to_message.from_user.username,
"Id": f'{update.message.reply_to_message.from_user.id}@telegram',
}),
})
Endpoints[cmd.Name]({"Event": update, "Manager": context}, cmd)
if Debug and Dumper:
Text = update.message.text
Text = (Text.replace('\n', '\\n') if Text else '')
with open('Dump.txt', 'a') as File:
File.write(f'[{time.ctime()}] [{int(time.time())}] [{update.message.chat.id}] [{update.message.message_id}] [{update.message.from_user.id}] {Text}\n')
Endpoints[cmd.Name]["handler"]({"Event": update, "Manager": context}, cmd)
def TelegramSender(event, manager, Data, Destination, TextPlain, TextMarkdown) -> None:
if Destination:
manager.bot.send_message(Destination, text=TextPlain)
def TelegramSender(event, manager, data, destination, textPlain, textMarkdown) -> None:
if destination:
manager.bot.send_message(destination, text=textPlain)
else:
replyToId = (Data["ReplyTo"] if ("ReplyTo" in Data and Data["ReplyTo"]) else event.message.message_id)
if InDict(Data, 'Media'):
event.message.reply_photo(
Data['Media'],
caption=(TextMarkdown if TextMarkdown else TextPlain if TextPlain else None),
parse_mode=('MarkdownV2' if TextMarkdown else None),
reply_to_message_id=replyToId,
)
elif TextMarkdown:
event.message.reply_markdown_v2(TextMarkdown, reply_to_message_id=replyToId)
elif TextPlain:
event.message.reply_text(TextPlain, reply_to_message_id=replyToId)
replyToId = (data["ReplyTo"] if ("ReplyTo" in data and data["ReplyTo"]) else event.message.message_id)
if InDict(data, "Media") and not InDict(data, "media"):
data["media"] = {"bytes": data["Media"]}
if InDict(data, "media"):
#data["media"] = SureArray(data["media"])
#media = (data["media"][0]["bytes"] if "bytes" in data["media"][0] else data["media"][0]["url"])
#if len(data["media"]) > 1:
# media_list = []
# media_list.append(telegram.InputMediaPhoto(
# media[0],
# caption=(textMarkdown if textMarkdown else textPlain if textPlain else None),
# parse_mode=("MarkdownV2" if textMarkdown else None)))
# for medium in media[1:]:
# media_list.append(telegram.InputMediaPhoto(medium))
# event.message.reply_media_group(media_list, reply_to_message_id=replyToId)
#else:
# event.message.reply_photo(
# media,
# caption=(textMarkdown if textMarkdown else textPlain if textPlain else None),
# parse_mode=("MarkdownV2" if textMarkdown else None),
# reply_to_message_id=replyToId)
#event.message.reply_photo(
# (DictGet(media[0], "bytes") or DictGet(media[0], "url")),
# caption=(textMarkdown if textMarkdown else textPlain if textPlain else None),
# parse_mode=("MarkdownV2" if textMarkdown else None),
# reply_to_message_id=replyToId)
#for medium in media[1:]:
# event.message.reply_photo((DictGet(medium, "bytes") or DictGet(medium, "url")), reply_to_message_id=replyToId)
for medium in SureArray(data["media"]):
event.message.reply_photo(
(DictGet(medium, "bytes") or DictGet(medium, "url")),
caption=(textMarkdown if textMarkdown else textPlain if textPlain else None),
parse_mode=("MarkdownV2" if textMarkdown else None),
reply_to_message_id=replyToId)
elif textMarkdown:
event.message.reply_markdown_v2(textMarkdown, reply_to_message_id=replyToId)
elif textPlain:
event.message.reply_text(textPlain, reply_to_message_id=replyToId)
RegisterPlatform(name="Telegram", main=TelegramMain, sender=TelegramSender, eventClass=telegram.Update)

2
LibWinDog/Platforms/Telegram/requirements.txt Normal file → Executable file
View File

@ -1 +1 @@
python-telegram-bot==13.4.1
python-telegram-bot==13.15

0
LibWinDog/Platforms/Web.py Normal file → Executable file
View File

0
ModWinDog/Codings.py Normal file → Executable file
View File

16
ModWinDog/Hashing.py Normal file → Executable file
View File

@ -6,17 +6,19 @@
import hashlib
def cHash(context, data) -> None:
if len(data.Tokens) >= 3 and data.Tokens[1] in hashlib.algorithms_available:
Alg = data.Tokens[1]
Hash = hashlib.new(Alg, Alg.join(data.Body.split(Alg)[1:]).strip().encode()).hexdigest()
algorithm = data.command.arguments["algorithm"]
if data.command.body and algorithm in hashlib.algorithms_available:
hashed = hashlib.new(algorithm, algorithm.join(data.Body.split(algorithm)[1:]).strip().encode()).hexdigest()
SendMsg(context, {
"TextPlain": Hash,
"TextMarkdown": MarkdownCode(Hash, True),
"TextPlain": hashed,
"TextMarkdown": MarkdownCode(hashed, True),
})
else:
SendMsg(context, {"Text": choice(Locale.__('hash.usage')).format(data.Tokens[0], hashlib.algorithms_available)})
SendMsg(context, {"Text": choice(Locale.__('hash.usage')).format(data.command.tokens[0], hashlib.algorithms_available)})
RegisterModule(name="Hashing", group="Geek", summary="Functions for hashing of textual content.", endpoints={
"Hash": CreateEndpoint(["hash"], summary="Responds with the hash-sum of a message received.", handler=cHash),
"Hash": CreateEndpoint(names=["hash"], summary="Responds with the hash-sum of a message received.", handler=cHash, arguments={
"algorithm": True,
}),
})

0
ModWinDog/Help.py Normal file → Executable file
View File

60
ModWinDog/Internet/Internet.py Normal file → Executable file
View File

@ -3,12 +3,18 @@
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start # """
MicrosoftBingSettings = {}
""" # end windog config # """
from urlextract import URLExtract
from urllib import parse as UrlParse
from urllib.request import urlopen, Request
def HttpGet(url:str):
return urlopen(Request(url, headers={"User-Agent": WebUserAgent}))
def HttpReq(url:str, method:str|None=None, *, body:bytes=None, headers:dict[str, str]={"User-Agent": WebUserAgent}):
return urlopen(Request(url, method=method, data=body, headers=headers))
def cEmbedded(context, data) -> None:
if len(data.Tokens) >= 2:
@ -49,7 +55,7 @@ def cWeb(context, data) -> None:
if data.Body:
try:
QueryUrl = UrlParse.quote(data.Body)
Req = HttpGet(f'https://html.duckduckgo.com/html?q={QueryUrl}')
Req = HttpReq(f'https://html.duckduckgo.com/html?q={QueryUrl}')
Caption = f'🦆🔎 "{data.Body}": https://duckduckgo.com/?q={QueryUrl}\n\n'
Index = 0
for Line in Req.read().decode().replace('\t', ' ').splitlines():
@ -80,14 +86,14 @@ def cTranslate(context, data) -> None:
try:
toLang = data.Tokens[1]
# TODO: Use many different public Lingva instances in rotation to avoid overloading a specific one
result = json.loads(HttpGet(f'https://lingva.ml/api/v1/auto/{toLang}/{UrlParse.quote(toLang.join(data.Body.split(toLang)[1:]))}').read())
result = json.loads(HttpReq(f'https://lingva.ml/api/v1/auto/{toLang}/{UrlParse.quote(toLang.join(data.Body.split(toLang)[1:]))}').read())
SendMsg(context, {"TextPlain": f"[{result['info']['detectedSource']} (auto) -> {toLang}]\n\n{result['translation']}"})
except Exception:
raise
def cUnsplash(context, data) -> None:
try:
Req = HttpGet(f'https://source.unsplash.com/random/?{UrlParse.quote(data.Body)}')
Req = HttpReq(f'https://source.unsplash.com/random/?{UrlParse.quote(data.Body)}')
ImgUrl = Req.geturl().split('?')[0]
SendMsg(context, {
"TextPlain": f'{{{ImgUrl}}}',
@ -102,18 +108,18 @@ def cSafebooru(context, data) -> None:
try:
if data.Body:
for i in range(7): # retry a bunch of times if we can't find a really random result
ImgUrls = HttpGet(f'{ApiUrl}md5:{RandHexStr(3)}%20{UrlParse.quote(data.Body)}').read().decode().split(' file_url="')[1:]
ImgUrls = HttpReq(f'{ApiUrl}md5:{RandHexStr(3)}%20{UrlParse.quote(data.Body)}').read().decode().split(' file_url="')[1:]
if ImgUrls:
break
if not ImgUrls: # literal search
ImgUrls = HttpGet(f'{ApiUrl}{UrlParse.quote(data.Body)}').read().decode().split(' file_url="')[1:]
ImgUrls = HttpReq(f'{ApiUrl}{UrlParse.quote(data.Body)}').read().decode().split(' file_url="')[1:]
if not ImgUrls:
return SendMsg(context, {"Text": "Error: Could not get any result from Safebooru."})
ImgXml = choice(ImgUrls)
ImgUrl = ImgXml.split('"')[0]
ImgId = ImgXml.split(' id="')[1].split('"')[0]
else:
HtmlReq = HttpGet(HttpGet('https://safebooru.org/index.php?page=post&s=random').geturl())
HtmlReq = HttpReq(HttpReq('https://safebooru.org/index.php?page=post&s=random').geturl())
for Line in HtmlReq.read().decode().replace('\t', ' ').splitlines():
if '<img ' in Line and ' id="image" ' in Line and ' src="':
ImgUrl = Line.split(' src="')[1].split('"')[0]
@ -123,18 +129,52 @@ def cSafebooru(context, data) -> None:
SendMsg(context, {
"TextPlain": f'[{ImgId}]\n{{{ImgUrl}}}',
"TextMarkdown": (f'\\[`{ImgId}`\\]\n' + MarkdownCode(ImgUrl, True)),
"Media": HttpGet(ImgUrl).read(),
"media": {"url": ImgUrl}, #, "bytes": HttpReq(ImgUrl).read()},
})
else:
pass
except Exception:
except Exception as error:
raise
def cDalle(context, data) -> None:
if not data.Body:
return SendMsg(context, {"Text": "Please tell me what to generate."})
image_filter = "&quot;https://th.bing.com/th/id/"
try:
retry_index = 3
result_list = ""
result_id = HttpReq(
f"https://www.bing.com/images/create?q={UrlParse.quote(data.Body)}&rt=3&FORM=GENCRE",#"4&FORM=GENCRE",
body=f"q={UrlParse.urlencode({'q': data.Body})}&qs=ds".encode(),
headers=MicrosoftBingSettings).read().decode()
print(result_id)
result_id = result_id.split('&amp;id=')[1].split('&amp;')[0]
results_url = f"https://www.bing.com/images/create/-/{result_id}?FORM=GENCRE"
SendMsg(context, {"Text": "Request sent, please wait..."})
while retry_index < 12 and image_filter not in result_list:
result_list = HttpReq(results_url, headers={"User-Agent": MicrosoftBingSettings["User-Agent"]}).read().decode()
time.sleep(1.25 * retry_index)
retry_index += 1
if image_filter in result_list:
SendMsg(context, {
"TextPlain": f"{{{results_url}}}",
"TextMarkdown": MarkdownCode(results_url, True),
"Media": HttpReq(
result_list.split(image_filter)[1].split('\\&quot;')[0],
headers={"User-Agent": MicrosoftBingSettings["User-Agent"]}).read(),
})
else:
raise Exception("Something went wrong.")
except Exception as error:
Log(error)
SendMsg(context, {"TextPlain": error})
RegisterModule(name="Internet", summary="Tools and toys related to the Internet.", endpoints={
"Embedded": CreateEndpoint(["embedded"], summary="Rewrites a link, trying to bypass embed view protection.", handler=cEmbedded),
"Web": CreateEndpoint(["web"], summary="Provides results of a DuckDuckGo search.", handler=cWeb),
"Translate": CreateEndpoint(["translate"], summary="Returns the received message after translating it in another language.", handler=cTranslate),
"Unsplash": CreateEndpoint(["unsplash"], summary="Sends a picture sourced from Unsplash.", handler=cUnsplash),
"Safebooru": CreateEndpoint(["safebooru"], summary="Sends a picture sourced from Safebooru.", handler=cSafebooru),
#"DALL-E": CreateEndpoint(["dalle"], summary="Sends an AI-generated picture from DALL-E 3 via Microsoft Bing.", handler=cDalle),
})

0
ModWinDog/Internet/requirements.txt Normal file → Executable file
View File

102
ModWinDog/Scrapers/Scrapers.py Executable file
View File

@ -0,0 +1,102 @@
# ================================== #
# WinDog multi-purpose chatbot #
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
""" # windog config start # """
SeleniumDriversLimit = 2
""" # end windog config # """
currentSeleniumDrivers = 0
#from selenium import webdriver
#from selenium.webdriver import Chrome
#from selenium.webdriver.common.by import By
from seleniumbase import Driver
def getSelenium() -> Driver:
global currentSeleniumDrivers
if currentSeleniumDrivers >= SeleniumDriversLimit:
return False
#options = webdriver.ChromeOptions()
#options.add_argument("headless=new")
#options.add_argument("user-data-dir=./Selenium-WinDog")
#seleniumDriver = Chrome(options=options)
currentSeleniumDrivers += 1
return Driver(uc=True, headless2=True, user_data_dir=f"./Selenium-WinDog/{currentSeleniumDrivers}")
def closeSelenium(driver:Driver) -> None:
global currentSeleniumDrivers
try:
driver.close()
driver.quit()
except:
Log(format_exc())
if currentSeleniumDrivers > 0:
currentSeleniumDrivers -= 1
def cDalleSelenium(context, data) -> None:
if not data.Body:
return SendMsg(context, {"Text": "Please tell me what to generate."})
#if not seleniumDriver:
# SendMsg(context, {"Text": "Initializing Selenium, please wait..."})
# loadSeleniumDriver()
try:
driver = getSelenium()
if not driver:
return SendMsg(context, {"Text": "Couldn't access a web scraping VM as they are all busy. Please try again later."})
driver.get("https://www.bing.com/images/create/")
driver.refresh()
#retry_index = 3
#while retry_index < 12:
# time.sleep(retry_index := retry_index + 1)
# try:
#seleniumDriver.find_element(By.CSS_SELECTOR, 'form input[name="q"]').send_keys(data.Body)
#seleniumDriver.find_element(By.CSS_SELECTOR, 'form a[role="button"]').submit()
driver.find_element('form input[name="q"]').send_keys(data.Body)
driver.find_element('form a[role="button"]').submit()
try:
driver.find_element('img[alt="Content warning"]')
SendMsg(context, {"Text": "This prompt has been blocked by Microsoft because it violates their content policy. Further attempts might lead to a ban on your profile."})
closeSelenium(driver)
return
except Exception: # warning element was not found, we should be good
pass
SendMsg(context, {"Text": "Request sent successfully, please wait..."})
# except Exception:
# pass
retry_index = 3
while retry_index < 12:
# note that sometimes generation fails and we will never get any image!
#try:
time.sleep(retry_index := retry_index + 1)
driver.refresh()
img_list = driver.find_elements(#By.CSS_SELECTOR,
'div.imgpt a img.mimg')
if not len(img_list):
continue
img_array = []
for img_url in img_list:
img_url = img_url.get_attribute("src").split('?')[0]
img_array.append({"url": img_url}) #, "bytes": HttpReq(img_url).read()})
page_url = driver.current_url.split('?')[0]
SendMsg(context, {
"TextPlain": f'"{data.Body}"\n{{{page_url}}}',
"TextMarkdown": (f'"_{CharEscape(data.Body, "MARKDOWN")}_"\n' + MarkdownCode(page_url, True)),
"media": img_array,
})
closeSelenium(driver)
break
#except Exception as ex:
# pass
except Exception as error:
Log(format_exc())
SendMsg(context, {"TextPlain": "An unexpected error occurred."})
closeSelenium(driver)
RegisterModule(name="Scrapers", endpoints={
"DALL-E": CreateEndpoint(["dalle"], summary="Sends an AI-generated picture from DALL-E 3 via Microsoft Bing.", handler=cDalleSelenium),
})

View File

@ -0,0 +1 @@
seleniumbase

30
ModWinDog/Scripting/Scripting.py Normal file → Executable file
View File

@ -3,13 +3,22 @@
# Licensed under AGPLv3 by OctoSpacc #
# ================================== #
luaCycleLimit = 10000
luaMemoryLimit = (512 * 1024) # 512 KB
luaCrashMessage = f"Script has been forcefully terminated due to having exceeded the max cycle count limit ({luaCycleLimit})."
""" # windog config start # """
# Use specific Lua version; always using the latest is risky due to possible new APIs and using JIT is vulnerable
LuaCycleLimit = 10000
LuaMemoryLimit = (512 * 1024) # 512 KB
LuaCrashMessage = f"Script has been forcefully terminated due to having exceeded the max cycle count limit ({LuaCycleLimit})."
# see <http://lua-users.org/wiki/SandBoxes> for a summary of certainly safe objects (outdated though)
LuaGlobalsWhitelist = ["_windog", "_VERSION", "print", "error", "assert", "tonumber", "tostring", "math", "string", "table"]
LuaTablesWhitelist = {"os": ["clock", "date", "difftime", "time"]}
""" # end windog config # """
# always specify a Lua version; using the default latest is risky due to possible new APIs and using JIT is vulnerable
from lupa.lua54 import LuaRuntime as NewLuaRuntime, LuaError, LuaSyntaxError
# I'm not sure this is actually needed, but better safe than sorry
def luaAttributeFilter(obj, attr_name, is_setting):
raise AttributeError("Access Denied.")
@ -18,15 +27,20 @@ def cLua(context, data=None) -> None:
scriptText = (data.Body or (data.Quoted and data.Quoted.Body))
if not scriptText:
return SendMsg(context, {"Text": "You must provide some Lua code to execute."})
luaRuntime = NewLuaRuntime(max_memory=luaMemoryLimit, register_eval=False, register_builtins=False, attribute_filter=luaAttributeFilter)
luaRuntime = NewLuaRuntime(max_memory=LuaMemoryLimit, register_eval=False, register_builtins=False, attribute_filter=luaAttributeFilter)
luaRuntime.eval(f"""(function()
_windog = {{ stdout = "" }}
function print (text, endl) _windog.stdout = _windog.stdout .. tostring(text) .. (endl ~= false and "\\n" or "") end
function luaCrashHandler () return error("{luaCrashMessage}") end
debug.sethook(luaCrashHandler, "", {luaCycleLimit})
function luaCrashHandler () return error("{LuaCrashMessage}") end
debug.sethook(luaCrashHandler, "", {LuaCycleLimit})
end)()""")
# delete unsafe objects
for key in luaRuntime.globals():
if key not in ["error", "assert", "math", "string", "tostring", "print", "_windog"]:
if key in LuaTablesWhitelist:
for tabKey in luaRuntime.globals()[key]:
if tabKey not in LuaTablesWhitelist[key]:
del luaRuntime.globals()[key][tabKey]
elif key not in LuaGlobalsWhitelist:
del luaRuntime.globals()[key]
try:
textOutput = ("[ʟᴜᴀ ꜱᴛᴅᴏᴜᴛ]\n\n" + luaRuntime.eval(f"""(function()

0
ModWinDog/Scripting/requirements.txt Normal file → Executable file
View File

4
README.md Normal file → Executable file
View File

@ -15,8 +15,8 @@ In case you want to run your own instance:
1. `git clone --depth 1 https://gitlab.com/octospacc/WinDog && cd ./WinDog` to get the code.
2. `find -type f -name requirements.txt -exec python3 -m pip install -U -r {} \;` to install the full package of dependencies.
3. `cp ./LibWinDog/Config.py ./` and, in the new file, edit essential fields (like user credentials), uncommenting them where needed, then delete the unmodified fields.
4. `sh ./StartWinDog.sh` to start the bot every time.
3. `sh ./StartWinDog.sh` to start the bot every time.
* The first time it runs, it will generate a `Config.py` file, where you should edit essential fields (like user credentials), uncommenting them where needed, then delete the unmodified fields. Restart the program to load the updated configuration.
All my source code mirrors for the bot:

116
WinDog.py
View File

@ -6,6 +6,7 @@
import json, time
from binascii import hexlify
from glob import glob
from magic import Magic
from os import listdir
from os.path import isfile, isdir
@ -15,6 +16,7 @@ from traceback import format_exc
from bs4 import BeautifulSoup
from html import unescape as HtmlUnescape
from markdown import markdown
from LibWinDog.Config import *
from LibWinDog.Database import *
# <https://daringfireball.net/projects/markdown/syntax#backslash>
@ -24,7 +26,11 @@ def Log(text:str, level:str="?", *, newline:bool|None=None, inline:bool=False) -
endline = '\n'
if newline == False or (inline and newline == None):
endline = ''
print((text if inline else f"[{level}] [{int(time.time())}] {text}"), end=endline)
text = (text if inline else f"[{level}] [{time.ctime()}] [{int(time.time())}] {text}")
if LogToConsole:
print(text, end=endline)
if LogToFile:
open("./Log.txt", 'a').write(text + endline)
def SetupLocales() -> None:
global Locale
@ -59,11 +65,14 @@ def SetupLocales() -> None:
Locale['Locale'] = Locale
Locale = SimpleNamespace(**Locale)
def InDict(Dict:dict, Key:str) -> any:
if Key in Dict:
return Dict[Key]
else:
return None
def SureArray(array:any) -> list|tuple:
return (array if type(array) in [list, tuple] else [array])
def InDict(dikt:dict, key:str, /) -> any:
return (dikt[key] if key in dikt else None)
def DictGet(dikt:dict, key:str, /) -> any:
return (dikt[key] if key in dikt else None)
def isinstanceSafe(clazz:any, instance:any) -> bool:
if instance != None:
@ -105,9 +114,12 @@ def HtmlEscapeFull(Raw:str) -> str:
def GetRawTokens(text:str) -> list:
return text.strip().replace('\t', ' ').replace(' ', ' ').replace(' ', ' ').split(' ')
def ParseCmd(msg) -> dict|None:
def ParseCmd(msg) -> SimpleNamespace|None:
#if not len(msg) or msg[1] not in CmdPrefixes:
# return
name = msg.replace('\n', ' ').replace('\t', ' ').replace(' ', ' ').replace(' ', ' ').split(' ')[0][1:].split('@')[0]
if not name: return
#if not name:
# return
return SimpleNamespace(**{
"Name": name.lower(),
"Body": name.join(msg.split(name)[1:]).strip(),
@ -116,7 +128,7 @@ def ParseCmd(msg) -> dict|None:
"Quoted": None,
})
def GetWeightedText(texts:tuple) -> str|None:
def GetWeightedText(*texts) -> str|None:
for text in texts:
if text:
return text
@ -131,10 +143,42 @@ def RandHexStr(length:int) -> str:
hexa += choice('0123456789abcdef')
return hexa
def OnMessageReceived() -> None:
pass
def ParseCommand(text:str) -> SimpleNamespace|None:
text = text.strip()
try: # ensure command is not empty
if not (text[0] in CmdPrefixes and text[1:].strip()):
return
except IndexError:
return
command = SimpleNamespace(**{})
command.tokens = text.replace("\r", " ").replace("\n", " ").replace("\t", " ").replace(" ", " ").replace(" ", " ").split(" ")
command.name = command.tokens[0][1:].lower()
command.body = text[len(command.tokens[0]):].strip()
if (endpoint_arguments := Endpoints[command.name]["arguments"]):
command.arguments = {}
# TODO differences between required (True) and optional (False) args
for index, key in enumerate(endpoint_arguments):
try:
value = command.tokens[index + 1]
command.body = command.body[len(value):].strip()
except IndexError:
value = None
command.arguments[key] = value
return command
def OnMessageParsed(data:SimpleNamespace) -> None:
if Debug and (DumpToFile or DumpToConsole):
text = (data.text_auto.replace('\n', '\\n') if data.text_auto else '')
text = f"[{int(time.time())}] [{time.ctime()}] [{data.room_id}] [{data.message_id}] [{data.user.id}] {text}"
if DumpToConsole:
print(text)
if DumpToFile:
open((DumpToFile if (DumpToFile and type(DumpToFile) == str) else "./Dump.txt"), 'a').write(text + '\n')
def SendMsg(context, data, destination=None) -> None:
return SendMessage(context, data, destination)
def SendMessage(context, data, destination=None) -> None:
if type(context) == dict:
event = context['Event'] if 'Event' in context else None
manager = context['Manager'] if 'Manager' in context else None
@ -154,6 +198,9 @@ def SendMsg(context, data, destination=None) -> None:
if isinstanceSafe(event, InDict(platform, "eventClass")) or isinstanceSafe(manager, InDict(platform, "managerClass")):
platform["sender"](event, manager, data, destination, textPlain, textMarkdown)
def SendReaction() -> None:
pass
def RegisterPlatform(name:str, main:callable, sender:callable, *, eventClass=None, managerClass=None) -> None:
Platforms[name] = {"main": main, "sender": sender, "eventClass": eventClass, "managerClass": managerClass}
Log(f"{name}, ", inline=True)
@ -164,9 +211,10 @@ def RegisterModule(name:str, endpoints:dict, *, group:str|None=None, summary:str
for endpoint in endpoints:
endpoint = endpoints[endpoint]
for name in endpoint["names"]:
Endpoints[name] = endpoint["handler"]
Endpoints[name] = endpoint
def CreateEndpoint(names:list[str], handler:callable, arguments:dict[str, dict]={}, *, summary:str|None=None) -> dict:
# TODO register endpoint with this instead of RegisterModule
def CreateEndpoint(names:list[str], handler:callable, arguments:dict[str, bool]|None=None, *, summary:str|None=None) -> dict:
return {"names": names, "summary": summary, "handler": handler, "arguments": arguments}
def Main() -> None:
@ -187,31 +235,51 @@ if __name__ == '__main__':
Locale = {"Fallback": {}}
Platforms, Modules, ModuleGroups, Endpoints = {}, {}, {}, {}
for dir in ("LibWinDog/Platforms", "ModWinDog"):
match dir:
for folder in ("LibWinDog/Platforms", "ModWinDog"):
match folder:
case "LibWinDog/Platforms":
Log("📩️ Loading Platforms... ", newline=False)
case "ModWinDog":
Log("🔩️ Loading Modules... ", newline=False)
for name in listdir(f"./{dir}"):
path = f"./{dir}/{name}"
for name in listdir(f"./{folder}"):
path = f"./{folder}/{name}"
if isfile(path):
exec(open(path, 'r').read())
elif isdir(path):
exec(open(f"{path}/{name}.py", 'r').read())
files = listdir(path)
if f"{name}.py" in files:
files.remove(f"{name}.py")
exec(open(f"{path}/{name}.py", 'r').read())
for file in files:
if file.endswith(".py"):
exec(open(f"{path}/{name}.py", 'r').read())
# TODO load locales
#for name in listdir(path):
# if name.lower().endswith('.json'):
#
Log("...Done. ✅️", inline=True, newline=True)
Log("💽️ Loading Configuration", newline=False)
exec(open("./LibWinDog/Config.py", 'r').read())
try:
Log("💽️ Loading Configuration... ", newline=False)
#exec(open("./LibWinDog/Config.py", 'r').read())
from Config import *
if isfile("./Config.py"):
from Config import *
except Exception:
Log(format_exc())
Log("...Done. ✅️", inline=True, newline=True)
else:
Log("💾️ No configuration found! Generating and writing to `./Config.py`... ", inline=True)
with open("./Config.py", 'w') as configFile:
opening = '# windog config start #'
closing = '# end windog config #'
for folder in ("LibWinDog", "ModWinDog"):
for file in glob(f"./{folder}/**/*.py", recursive=True):
try:
name = '/'.join(file.split('/')[1:-1])
heading = f"# ==={'=' * len(name)}=== #"
source = open(file, 'r').read().replace(f"''' {opening}", f'""" {opening}').replace(f"{closing} '''", f'{closing} """')
content = '\n'.join(content.split(f'""" {opening}')[1].split(f'{closing} """')[0].split('\n')[1:-1])
configFile.write(f"{heading}\n# 🔽️ {name} 🔽️ #\n{heading}\n{content}\n\n")
except IndexError:
pass
Log("Done. ✅️", inline=True, newline=True)
Main()
Log("🌚️ WinDog Stopping...")