Intermediate commits (oldest → newest):
- Hit list queue for radio
- fix
- Documentation of conjurer_librarian
- Reformatting of conjurer_librarian
- Documentation of conjurer_musician
- Formatting of conjurer_musician
- Documentation of radio_conjurer.liq
- Merge branch 'main' of https://github.com/migatu/conjurer
- Removal from tracking
- Fixing librarian result (#6)
- Potential bugfix for some lazycoding in spotipy... Another one. fuck it
- Music tunneling (#7)
- Even more logging in spotify download
- test requests
- Fine tuning
- Optimization bug fix #1
- Fix crossfade
- Crossfadeing ctnd
- Fix
- Starting work on librarian headless mode
This commit is contained in:
2025-10-30 16:59:00 +01:00
parent 806cada1c1
commit a8797df45c
20 changed files with 1161 additions and 509 deletions
+345 -186
View File
@@ -1,243 +1,378 @@
from logging import handlers
import re
"""
This module contains the implementation of the Librarian class and
related functions for searching and refining queries.
Classes:
- Librarian: Represents a librarian object that performs search and
refinement operations on queries
Functions:
- flask_debug: Starts a Flask application in debug mode without using the reloader.
- waitress_run: Serves the Flask application using the Waitress WSGI server.
- BackgroundTaskSearch: Represents a background task for running the
Librarian object asynchronously
"""
import asyncio
import json
import logging
import netrc
import threading
import time
from flask import Flask, jsonify, request
import netrc
import json
import asyncio
from json.decoder import JSONDecodeError
from urllib.request import urlopen
from logging import handlers
from queue import Queue
import requests
import scrape_bot
import search_bot
from flask import Flask, jsonify, request
from habanero import Crossref
from waitress import serve
from queue import Queue
import search_bot
import scrape_bot
import requests
NETRC_FILE = r"C:\Users\Activcom.pl\.netrc"
# Constants
NETRC_FILE = r"C:\Users\Activcom.pl\.netrc"
HOST_ADDRESS = "192.168.1.192"
PORT_ADDRESS = 5001
MAIN_BOT_ADDRESS = "http://192.168.1.191:5000"
SEND_RESULTS = "/conjurer"
BDSM_UUID_TEST = '96b7f85a-1142-4908-8986-62a2ea25a147'
MAX_CR_RESULTS = 150000
BDSM_UUID_TEST = "96b7f85a-1142-4908-8986-62a2ea25a147"
MAX_CR_RESULTS = 7000
ENCODING = "utf-8"
app = Flask(__name__)
librarian_queue = Queue()
librarian_list = []
class Librarian(object):
def __init__(self, app, query, uuid) -> None:
# trunk-ignore(pylint/R0902)
class Librarian(object):
"""
Represents a librarian object that performs search and refinement operations on queries.
"""
def __init__(self, _app, query, uuid, _deep_search) -> None:
"""
Initializes a Librarian object.
Args:
- _app: The Flask application object.
- query: The query to be searched.
- uuid: The unique identifier for the search.
Attributes:
- cr: The Crossref object for performing the search.
- query: The query to be searched.
- uuid: The unique identifier for the search.
- limit: The maximum number of search results to fetch.
- fetched: The number of search results fetched so far.
- hit: The number of search results that match the refinement criteria.
- total: The total number of search results.
- app: The Flask application object.
- live_results: A list to store live search results.
- final_result: A list to store the final refined search results.
- not_in_db: A list to store search results that are not in the local database.
- search_result_from_cr: A dictionary to store the search results from Crossref.
- done: A flag indicating if the search is done.
"""
netrc_mod = netrc.netrc(NETRC_FILE)
authTokens = netrc_mod.authenticators("crossref")
self.cr = Crossref(mailto=authTokens[0])
auth_tokens = netrc_mod.authenticators("crossref")
self.cr = Crossref(
mailto=auth_tokens[0],
ua_string=f"Conjurer project. mailto:{auth_tokens[0]}"
)
self.query = query
self.uuid = str(uuid)
self.limit = MAX_CR_RESULTS
self.fetched = 0
self.hit = 0
self.total = 0
self.app = app
self.app = _app
self.live_results = []
self.final_result = []
self.not_in_db = []
self.final_result = {}
self.not_in_db = {}
self.search_result_from_cr = {}
self.done = False
self.deep_search = _deep_search
async def search_crossref(self, query, deep_search=False):
"""
Performs a search on Crossref for the given query.
Args:
- query: The query to be searched.
Returns:
- result: The search result from Crossref.
Raises:
- None.
"""
async def search_crossref(self, query):
self.app.logger.info("STARTED SEARCH")
result = self.cr.works(query = query, limit = 1000)
self.search_result_from_cr.update(result)
self.total = result['message']['total-results']
self.fetched += len(result['message']['items'])
self.app.logger.info (self.total)
self.app.logger.info(self.fetched)
while self.total > self.fetched and self.limit > self.fetched:
result = self.cr.works(query = query, limit = 1000)
self.search_result_from_cr.update(result)
self.total = result['message']['total-results']
self.fetched += len(result['message']['items'])
self.app.logger.info (self.total)
self.app.logger.info(self.fetched)
time.sleep(0.1)
self.app.logger.info("CROSSREF DONE")
#TODO: add to sql with UUID
#sql row - title, doi, author, uuid, last retrieved, is_available
if not deep_search:
cr_result = self.cr.works(query=query, limit=1000)
self.search_result_from_cr.update(cr_result)
self.total = cr_result["message"]["total-results"]
self.fetched += len(cr_result["message"]["items"])
self.app.logger.info(self.total)
self.app.logger.info(self.fetched)
while self.total > self.fetched and self.limit > self.fetched:
tmp_result = self.cr.works(query=query, limit=1000, offset=self.fetched)
cr_result["message"]["items"].extend(tmp_result["message"]["items"])
self.total = tmp_result["message"]["total-results"]
self.fetched = len(cr_result["message"]["items"])
self.app.logger.info(self.total)
self.app.logger.info(self.fetched)
time.sleep(0.1)
else:
cr_result = self.cr.works(query=query, cursor_max=10000, cursor='*', progress_bar = True)
result = cr_result[0]
for item in cr_result[1:]:
result["message"]["items"].extend(item["message"]["items"])
self.total = item["message"]["total-results"]
self.fetched = len(result["message"]["items"])
self.app.logger.info(self.total)
self.app.logger.info(self.fetched)
cr_result = result
self.app.logger.info("Total, fetched:")
self.app.logger.info(self.total)
self.app.logger.info(self.fetched)
self.search_result_from_cr.update(cr_result)
self.total = cr_result["message"]["total-results"]
self.app.logger.info("CROSSREF DONE")
self.app.logger.info("CROSSREF DONE")
with open("cr_results.json", "r+", encoding="utf-8") as data_file:
# First we load existing data into a dict.
try:
file_data = json.load(data_file)
except JSONDecodeError:
file_data = {}
data_file.truncate(0)
data_file.seek(0)
tmp = {self.uuid : self.search_result_from_cr}
if file_data:
file_data.update(tmp)
else:
file_data = tmp
json.dump(file_data, data_file, indent=4)
return cr_result
async def refine_search(self, unrefined_result):
"""
Refines the search query based on the unrefined search result.
Args:
- unrefined_result: The unrefined search result.
Returns:
- refined_result: The refined search result.
Raises:
- None.
"""
summarized_results = []
self.app.logger.info("REFINE: Removing all derived works from the list")
for item in unrefined_result["message"]["items"]:
summarized_results.append(
{
"DOI": item["DOI"],
"title": item["title"] if "title" in item else None,
"type": item["type"] if "type" in item else None,
}
)
partial_result = {
self.uuid: {
"total_results": self.search_result_from_cr["message"][
"total-results"
],
"on_page": 1,
"summary": summarized_results,
"results": self.search_result_from_cr["message"]["items"],
}
}
self.app.logger.info("REFINE: Dumping to file")
temp = []
refined_result = {}
for key in partial_result:
self.app.logger.info("KEY:")
self.app.logger.info(key)
for item in partial_result[self.uuid]["summary"]:
if item["title"]:
temp.append(item)
for item in temp:
refined_result[item["DOI"]]= item
with open("rr_results.json", "r+", encoding="utf-8") as data_file:
# First we load existing data into a dict.
try:
file_data = json.load(data_file)
except JSONDecodeError:
file_data = {}
summarized_results = []
self.app.logger.info("REFINE: Removing all derived works from the list")
for item in self.search_result_from_cr['message']['items']:
container = "NO"
if "container-title" in item:
container = "YES"
else:
summarized_results.append({"DOI" : item['DOI'], "title": item['title'] if 'title' in item else None,"type": item['type'] if 'type' in item else None, "container": container})
result = {self.uuid : {"total_results": self.search_result_from_cr['message']['total-results'], "on_page" : 1, "summary":summarized_results, "results":self.search_result_from_cr['message']['items']}}
if file_data:
file_data.update(result)
else:
file_data = result
self.app.logger.info("REFINE: Dumping to file")
data_file.truncate(0)
data_file.seek(0)
tmp = {self.uuid: refined_result}
if file_data:
file_data.update(tmp)
else:
file_data = tmp
json.dump(file_data, data_file, indent=4)
return result
return refined_result
async def check_dois_in_local_db(self, doi):
self.app.logger.info("REFINE: Running search in the backend app")
coro = asyncio.to_thread(search_bot.search_for_doi, doi, self.live_results, self.app.logger)
result = await coro
return result
async def check_if_exists(self, refined_result):
"""
Checks if the given DOI exists.
async def check_if_exists_brute_force(self, page_url):
self.app.logger.error("REFINE: Brute force search!")
if not page_url.startswith(("http:", "https:")):
raise ValueError("URL must start with 'http:' or 'https:'")
# trunk-ignore(bandit/B310)
with urlopen(page_url) as response:
data = response.read()
text = data.decode("utf-8")
for line in text.splitlines():
if re.match(r".*Unfortunately, Sci-Hub doesn't have the requested document.*", line):
return False
if m := re.match(r".*<embed type=\"application.pdf\"\s*src=\"(.*\.pdf)",line):
direct_download_link = "https:" + str(m.group(1))
self.app.logger.info (direct_download_link)
return True
return False
Args:
- doi: The DOI to be checked.
- brute_force: A flag indicating if brute force method should be used.
async def check_if_exists(self, doi, brute_force = False):
Returns:
- result: The search result.
Raises:
- None.
"""
result = {}
result = await self.check_dois_in_local_db(doi)
if brute_force and not result:
for item in doi:
item_link = "https://sci-hub.se/" + item[0]
tmp = {"DOI" : item["DOI"], "exists" : await self.check_if_exists_brute_force(item_link), "data": item}
result.update(tmp)
return result
async def refine_query(self, unrefined_reqult, brute_force = False):
temp = []
refined_result = []
for key in unrefined_reqult:
self.app.logger.info("KEY:")
self.app.logger.info(key)
for item in unrefined_reqult[self.uuid]["summary"]:
if item["container"] == "NO" and item["title"]:
refined_result.append(item)
for item in refined_result:
temp.append([item["DOI"], item])
refined_result = await self.check_if_exists(temp)
for item in refined_result:
self.app.logger.info("REFINE: Running search in the backend app")
dois = []
for item, value in refined_result.items():
dois.append([item, value])
coro = asyncio.to_thread(
search_bot.search_for_doi, dois, self.live_results, self.app.logger
)
result = await coro
result_list = []
result_no_db = []
for item in result:
if item["exists"]:
self.final_result.append(item["data"])
result_list.append(item)
else:
self.not_in_db.append(item["data"])
self.hit = len(self.final_result)
return self.final_result
#TODO: DOłożyć sprawdzenie czy w rafinowanym pliku już nie mamy częściowego wyniku
async def answer_query(self):
#TODO: DOłożyć sprawdzenie czy w rafinowanym pliku już nie mamy częściowego wyniku tutaj bo zanim w ogóle otworzymy nierafinowany
#Search sql for UUID
#SQL SELECT * FROM RESULTS WHERE UUID = uuid
self.app.logger.info(f"Search started {self.uuid}")
database = None
query = self.query
with open("cr_results.json", "r+", encoding="utf-8") as cr_file:
#tu logika komunikacji z SQL i cała logika związana z wyszukaniem
#ale tymczasowo plik
try:
database = json.load(cr_file)
except JSONDecodeError:
pass
#TODO: Zwracać i final_result i not_in_db. Not in db_do pliku - i do ręcznego przeglądania potem.
if database:
for item in database.keys():
if item in self.uuid:
self.result = await self.refine_query(database)
temp = {}
for item in self.result:
temp[item['DOI']] = {"Title" : item['title'], "type": item['type']}
self.result = temp
temp2 = {}
for item in self.not_in_db:
temp2[item['DOI']] = {"Title" : item['title'], "type": item['type']}
self.not_in_db = temp2
return self.result
answer = await self.search_crossref(query=query)
database.update(answer)
cr_file.truncate(0)
cr_file.seek(0)
json.dump(database,cr_file)
self.result = await self.refine_query(answer)
else:
answer = await self.search_crossref(query=query)
database = answer
cr_file.truncate(0)
cr_file.seek(0)
json.dump(database,cr_file)
self.result = await self.refine_query(answer)
self.app.logger.info("Refined")
self.done = True
temp = {}
for item in self.result:
temp[item['DOI']] = {"Title" : item['title'], "type": item['type']}
self.result = temp
temp2 = {}
for item in self.not_in_db:
temp2[item['DOI']] = {"Title" : item['title'], "type": item['type']}
self.not_in_db = temp2
return self.result
result_no_db.append(item)
self.hit = len(result)
return result_list, result_no_db
async def answer_query(self, deep_search=False):
"""
Answers the search query.
Args:
- None.
Returns:
- result: The search result.
Raises:
- None.
"""
self.app.logger.info(f"Search started {self.uuid}")
cr_result = await self.search_crossref(query=self.query, deep_search=deep_search)
refined_result = await self.refine_search(cr_result)
answer, negative_answer = await self.check_if_exists(refined_result)
self.app.logger.info("Returning result")
self.app.logger.info(answer)
self.app.logger.info(negative_answer)
for item in answer:
self.final_result[item["DOI"]] = {"Title": item["data"]["title"], "type": item["data"]["type"]}
for item in negative_answer:
self.not_in_db[item["DOI"]] = {"Title": item["data"]["title"], "type": item["data"]["type"]}
self.app.logger.info("Returning result case2")
self.app.logger.info(self.final_result)
return self.final_result
# ============================= FLASK INTERNALS===============================
#============================= FLASK INTERNALS===============================
def flask_debug():
"""
The `flask_debug` function starts a Flask application in debug mode without using the reloader.
Do not use for production for fucks sake.
Starts a Flask application in debug mode without using the reloader.
Args:
- None.
Returns:
- None.
Raises:
- None.
"""
# trunk-ignore(bandit/B201)
app.run(debug=True, use_reloader=False, host=HOST_ADDRESS, port=PORT_ADDRESS)
def waitress_run():
"""
The `waitress_run` function serves the `app` on host "0.0.0.0"
and port 5000 using the Waitress WSGI server.
Serves the Flask application using the Waitress WSGI server.
Args:
- None.
Returns:
- None.
Raises:
- None.
"""
serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS)
class BackgroundTaskSearch(threading.Thread):
"""
A background task for searching and saving results to files.
This class extends the `threading.Thread` class and is responsible for running
the search task in the background. It retrieves queries from a queue, performs
the search, and saves the results to files.
Attributes:
app (App): The application instance.
"""
def run(self):
loop = asyncio.new_event_loop() # loop = asyncio.get_event_loop()
"""
Run the background task.
This method is called when the thread is started. It creates a new event loop,
runs the `_run` method, and closes the event loop.
"""
loop = asyncio.new_event_loop()
loop.run_until_complete(self._run())
loop.close()
async def _run(self):
"""
Perform the search task.
This method is an asynchronous coroutine that runs in a loop. It retrieves a
librarian from the queue, answers the query, and saves the results to files.
It also sends the results to a remote server.
The search task continues running indefinitely until the thread is stopped.
"""
while True:
database = None
ndb_database = None
librarian = librarian_queue.get()
self.app.logger.info("STARTED")
result = await librarian.answer_query()
result = await librarian.answer_query(librarian.deep_search)
result = {librarian.uuid: result}
self.app.logger.info("Saving to file")
# Save results to "not_in_db.json" file
with open("not_in_db.json", "r+", encoding="utf-8") as ndb_file:
try:
ndb_database = json.load(ndb_file)
ndb_database = json.load(ndb_file)
except JSONDecodeError:
pass
if ndb_database:
@@ -248,9 +383,10 @@ class BackgroundTaskSearch(threading.Thread):
ndb_file.seek(0)
json.dump(ndb_database, ndb_file)
# Save results to "s_results.json" file
with open("s_results.json", "r+", encoding="utf-8") as s_file:
try:
database = json.load(s_file)
database = json.load(s_file)
except JSONDecodeError:
pass
if database:
@@ -267,26 +403,41 @@ class BackgroundTaskSearch(threading.Thread):
self.app.logger.info(result)
coroutine = asyncio.to_thread(
requests.post,
f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}",
json=result,
timeout=360,
)
requests.post,
f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}",
json=result,
timeout=360,
)
self.app.logger.info("SENT")
result = await coroutine
self.app.logger.info(result.status_code)
self.app.logger.info("SEND CONFIRMED")
time.sleep(1)
#==================================SERVER ROUTES==========================================
# ==================================SERVER ROUTES==========================================
@app.route("/query", methods=["POST"])
async def query_database():
"""
Endpoint for querying the database.
This function receives a POST request containing a JSON payload with a query and a UUID.
It creates a Librarian object with the query and UUID,
and adds it to the librarian_queue and librarian_list.
Finally, it returns a JSON response indicating the success
of the operation, along with the query, UUID,
and the current size of the librarian_queue.
Returns:
tuple: A tuple containing a JSON response and a status code.
"""
record = json.loads(request.data)
app.logger.info(record)
app.logger.info(record["query"])
app.logger.info(record["UUID"])
uuid = record["UUID"]
cl = Librarian(app, record["query"], uuid)
deep_search = record["deep_search"]
cl = Librarian(app, record["query"], uuid, deep_search)
librarian_queue.put(cl)
librarian_list.append(cl)
answer_data = (record["query"], record["UUID"], librarian_queue.qsize())
@@ -296,8 +447,15 @@ async def query_database():
)
return return_data
@app.route("/get_partial_result", methods=["POST"])
async def get_partial():
"""
Retrieves the partial result for a given UUID.
Returns:
A JSON response containing the partial result.
"""
record = json.loads(request.data)
app.logger.info(record)
app.logger.info(record["UUID"])
@@ -311,30 +469,31 @@ async def get_partial():
)
return return_data
#=======================================MAIN===================================================
# =======================================MAIN===================================================
if __name__ == "__main__":
app.logger.setLevel(logging.DEBUG)
h1 = handlers.RotatingFileHandler(
filename="E:\\logs\\librarian.log",
encoding="utf-8",
mode="a",
maxBytes=6 * 1024 * 1024,
backupCount=6,
filename="D:\\logs\\librarian.log",
encoding="utf-8",
mode="a",
maxBytes=6 * 1024 * 1024,
backupCount=6,
)
app.logger.addHandler(h1)
threads = []
threads.append(threading.Thread(target=waitress_run))
#threads.append(threading.Thread(target=flask_debug))
# threads.append(threading.Thread(target=flask_debug))
bgtask = BackgroundTaskSearch()
bgtask.app = app
threads.append(bgtask)
threads.append(threading.Thread(target = scrape_bot.scraper, args=(app.logger,)))
threads.append(threading.Thread(target=scrape_bot.scraper, args=(app.logger,)))
i = 0
for worker in threads:
try:
app.logger.info(f"App number: {i}")
i+=1
app.logger.info("App number: %s", i)
i += 1
worker.start()
except RuntimeError as e:
app.logger.error("Exploded")