gitea/conjurer
1
0
Fork 0
mirror of https://github.com/migatu/conjurer.git synced 2026-07-14 21:38:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fixes

Browse Source
This commit is contained in:
Migatu
2024-04-12 23:07:02 +02:00
parent 714fcc9e1d
commit 63f340f03a
5 changed files with 139 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot.py
+102 -138
View File
@@ -46,7 +46,12 @@ from spotify_dl import youtube as youtube_download
from communication_subroutine import comm_subroutine, IN_COMM_Q, OUT_COMM_Q
# *=========================== Classes definition
# The above code is defining a TypedDict called `Music_Config` in Python. This TypedDict has three
# keys: "ctx" which is an optional string, "queue" which is a list of strings, and "requester" which
# is a list of strings. This TypedDict is used to define the structure of a dictionary that should
# have these specific keys and value types.
Music_Config = TypedDict(
"Music_Config",
{
@@ -58,6 +63,10 @@ Music_Config = TypedDict(
class QueryControl:
"""
This class `QueryControl` is used to manage queries with information about the author, UUID,
content, logger, context, and replies.
"""
def __init__(self, query_author, query_uuid, query_content, uplogger, ctx) -> None:
self.author = query_author
self.uuid = query_uuid
@@ -70,6 +79,95 @@ class QueryControl:
)
self.replies = []
class MusicFileList(object):
"""
The `MusicFileList` class manages a list of music files, with the ability to refresh the list from a
file service or local directory and update the list with new items.
"""
def __init__(self, uplogger) -> None:
self.music_file_list = []
self.file_service_active = False
self.logger = uplogger
self.logger.info("Created Playlist organizer class")
def refresh_file_list(self):
"""
The `refresh_file_list` function attempts to connect to a file service to retrieve a list of music
files, and if unsuccessful, it populates the list by scanning a local directory for .mp3 files.
"""
try:
self.logger.info("Attempt to connect to file service")
response = requests.get(f"{FILE_SERVICE_ADDRESS}{GET_MP3}", timeout=360)
self.music_file_list = response.json()["music_file_list"]
self.file_service_active = True
except requests.exceptions.RequestException as e:
for mp3_item in Path.glob(Path(MUSIC_FOLDER), "**/*.mp3"):
temp_music_file = mp3_item.as_posix()
if platform == "win32":
temp_music_file = temp_music_file.replace("/", "\\")
self.music_file_list.append(temp_music_file)
self.file_service_active = False
self.logger.error(e.strerror)
self.logger.error("Service Unavailable")
def get_file_list(self):
"""
The `get_file_list` function returns the list of music files stored in the object.
:return: The `music_file_list` attribute is being returned.
"""
return self.music_file_list
def update_file_list(self, item):
"""
The `update_file_list` function appends an item to a music file list and sends a POST request with
the item data to a file service address.
:param item: The `item` parameter in the `update_file_list` method is the file that you want to add
to the `music_file_list` attribute of the class instance. It is then sent as a JSON payload in a
POST request to a file service address along with the endpoint `SEND_MP3`
"""
self.music_file_list.append(item)
post_data = {"item": str(item)}
requests.post(f"{FILE_SERVICE_ADDRESS}{SEND_MP3}", json=post_data, timeout=360)
class DoSearchView(discord.ui.View):
def __init__(self, _uuid, result_list, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.uuid = _uuid
self.result_list = result_list
@discord.ui.button(
label="Nastepna strona", style=discord.ButtonStyle.primary, row=0
)
async def next_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
@discord.ui.button(label="Wyszukaj wiecej", style=discord.ButtonStyle.danger, row=0)
async def reset_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
@discord.ui.button(
label="Poprzednia Strona", style=discord.ButtonStyle.primary, row=0
)
async def perv_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
# *=========================================== Predefines
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
@@ -193,60 +291,6 @@ logger.info("Playlist generation")
# *=========================================== Load Data files
class MusicFileList(object):
"""
The `MusicFileList` class manages a list of music files, with the ability to refresh the list from a
file service or local directory and update the list with new items.
"""
def __init__(self, uplogger) -> None:
self.music_file_list = []
self.file_service_active = False
self.logger = uplogger
self.logger.info("Created Playlist organizer class")
def refresh_file_list(self):
"""
The `refresh_file_list` function attempts to connect to a file service to retrieve a list of music
files, and if unsuccessful, it populates the list by scanning a local directory for .mp3 files.
"""
try:
self.logger.info("Attempt to connect to file service")
response = requests.get(f"{FILE_SERVICE_ADDRESS}{GET_MP3}", timeout=360)
self.music_file_list = response.json()["music_file_list"]
self.file_service_active = True
except requests.exceptions.RequestException as e:
for mp3_item in Path.glob(Path(MUSIC_FOLDER), "**/*.mp3"):
temp_music_file = mp3_item.as_posix()
if platform == "win32":
temp_music_file = temp_music_file.replace("/", "\\")
self.music_file_list.append(temp_music_file)
self.file_service_active = False
self.logger.error(e.strerror)
self.logger.error("Service Unavailable")
def get_file_list(self):
"""
The `get_file_list` function returns the list of music files stored in the object.
:return: The `music_file_list` attribute is being returned.
"""
return self.music_file_list
def update_file_list(self, item):
"""
The `update_file_list` function appends an item to a music file list and sends a POST request with
the item data to a file service address.
:param item: The `item` parameter in the `update_file_list` method is the file that you want to add
to the `music_file_list` attribute of the class instance. It is then sent as a JSON payload in a
POST request to a file service address along with the endpoint `SEND_MP3`
"""
self.music_file_list.append(item)
post_data = {"item": str(item)}
requests.post(f"{FILE_SERVICE_ADDRESS}{SEND_MP3}", json=post_data, timeout=360)
music_file_list = []
music_file_list = MusicFileList(logger)
music_file_list.refresh_file_list()
@@ -283,6 +327,8 @@ for key in word_reactions:
word_reactions[key][2] = datetime.now()
logger.info("Done")
# *=========================================== Create Client
logger.info("Creating discord bot")
client = commands.Bot(intents=intents, command_prefix="$")
@@ -343,9 +389,8 @@ async def on_message(message):
channel = client.get_channel(1064888712565100614)
await channel.send("Słyszałem ja żem że: " + message.content)
return
else:
channel = message.channel
await client.process_commands(message)
channel = message.channel
await client.process_commands(message)
message.content = message.content.lower()
@@ -2098,44 +2143,6 @@ async def parametry_muzyki_mego_ludu(
logger.info(exce)
class DoSearchView(discord.ui.View):
def __init__(self, uuid, result_list, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.uuid = uuid
self.result_list = result_list
@discord.ui.button(
label="Nastepna strona", style=discord.ButtonStyle.primary, row=0
)
async def next_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
@discord.ui.button(label="Wyszukaj wiecej", style=discord.ButtonStyle.danger, row=0)
async def reset_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
@discord.ui.button(
label="Poprzednia Strona", style=discord.ButtonStyle.primary, row=0
)
async def perv_callback(self, interaction, button):
# emoji = await ctx.guild.fetch_emoji(id)
# button.emoji=1102174413887111189
logger.info("Pressed")
await interaction.response.edit_message(
content="You clicked the button!"
) # Send a message when the button is clicked
@client.hybrid_command(
name="wyszukaj_linki_do_dokumentow",
description="Szuka linkow doi w bazie crossref i podaje linki do scihuba",
@@ -2146,9 +2153,6 @@ async def wyszukaj_linki_do_dokumentow(ctx):
query_uuid = uuid.uuid4()
#TODO: TESTING ONLY!!
#query_uuid = uuid.UUID('{12345678-1234-5678-1234-567812345678}')
# query_uuid = "test"
# 3 wyslij Librariana
json_query = {"UUID": str(query_uuid), "query": str(query)[30:], "page": 1}
coroutine = asyncio.to_thread(
requests.post,
@@ -2181,46 +2185,6 @@ async def wyszukaj_linki_do_dokumentow(ctx):
await ctx.send(f"No dobra poszło. Zapamiętaj proszę {query_uuid} - to identyfikator twojego zapytania. Jesteś {queue_size} w kolejce."+
" Zapytania obsługuje algorytm zasilany czterema chomikami zapierdalającymi w kołowrotku - więc wyniki najwcześniej za kilka godzi - ale mogą być też dni.")
'''
async def zaciagnij_czesciowe_wyniki(ctx):
# 4 zwroc 4 wyniki od librariana
# 5 przyciski przy każdym wierszu pozwalają sprawdzić czy plik istnieje na scihubie - jeśli tak podaje link do sciagniecia
# 6 ostatni wiersz to dalej, wróć, skasuj, zaciagnij kolejne wyniki (jeśli jest ich więcej niż 1000)
result_view = DoSearchView(
uuid=str(query_uuid), result_list=query_response.json()["data"]
)
resulttab = []
resultlink = []
hit, fetched, total, _ = query_response.json()["data"][0]
answer = query_response.json()["data"][1]
for item in answer:
resulttab.append(str(item["title"]))
item_link = "https://sci-hub.se/" + str(item["DOI"])
resultlink.append(item_link)
logger.info(len(resulttab))
for i in range(0, min(4, len(resulttab))):
logger.info(i)
logger.info(resulttab[i])
logger.info(resultlink[i])
url = resultlink[i]
logger.info(url)
temp_label = str(resulttab[i])
temp_label = temp_label.strip("[']")
result_view.add_item(
discord.ui.Button(label=str(temp_label[0:79]), row=i + 1, url=url)
)
labelka = (
"Wyniki wyszukiwania dla: "
+ str(query)[30:]
+ f"\nWszystkich {total} sprawdzonych: {fetched} znalezionych: {hit} "
) # ile znaleziono artykułów w ogóle, ile zostało po refine
await ctx.send(
labelka, view=result_view
) # Send a message with our View class that contains the button
'''
@client.hybrid_command(
name="krecimy_pornola",
bot_classes.py
View File
bot_commands.py
View File
bot_functions.py
View File
communication_subroutine.py
+37
View File
@@ -16,6 +16,12 @@ app = Flask(__name__)
@app.route("/conjurer", methods=["POST"])
def answer_external_command():
"""
The function `answer_external_command` logs the request data, loads the data as JSON, logs the
record, and then puts the record into an incoming queue before returning a success message.
:return: The function `answer_external_command()` is returning a JSON response with the message
"SUCCESS".
"""
app.logger.info(request)
record = json.loads(request.data)
app.logger.info(record)
@@ -25,6 +31,13 @@ def answer_external_command():
@app.route("/conjurer", methods=["GET"])
def check_alive():
"""
The function "check_alive" returns a JSON response with the message "ALIVE" when the "/conjurer"
route is accessed via a GET request.
:return: The code snippet is a Flask route that listens for GET requests to the "/conjurer"
endpoint. When a GET request is made to this endpoint, the function check_alive() is called, which
returns a JSON response with the message "ALIVE".
"""
return jsonify("ALIVE")
def flask_debug(_logger):
@@ -46,12 +59,29 @@ def waitress_run(_logger):
serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS)
def scan_queue(_logger):
"""
The function `scan_queue` reads data from a queue, logs it, and appends it to another queue.
:param _logger: The `_logger` parameter is typically an instance of a logging object that is used to
record and store log messages. It is commonly used to track the flow of the program, record errors,
and provide information for debugging purposes. In this code snippet, the `_logger` object is used
to log the
"""
while True:
data = OUT_COMM_Q.get()
_logger.info(data)
awaiting_q.append(data)
def scan_incoming(_logger):
"""
The `scan_incoming` function continuously checks for incoming data, processes it, and logs when data
is found.
:param _logger: The `_logger` parameter in the `scan_incoming` function is a logger object that is
used to log messages or information during the execution of the function. It is typically used for
debugging, monitoring, or tracking the flow of the program. In this case, the `_logger` is being
used
"""
while True:
try:
answer = incoming_q.get(block=False)
@@ -66,6 +96,13 @@ def scan_incoming(_logger):
def comm_subroutine(logger):
"""
The `comm_subroutine` function starts multiple threads to run different tasks concurrently.
:param logger: The `logger` parameter in the `comm_subroutine` function is an instance of a logger
object that is used for logging messages at various levels (e.g., debug, info, warning, error). In
the provided code snippet, the logger is used to log messages at the "info" level
"""
#logger.setLevel(logging.DEBUG)
logger.info("Started")
threads = []