diff --git a/conjurer_librarian/conjurer_librarian.py b/conjurer_librarian/conjurer_librarian.py index dc4c522..e280ed1 100644 --- a/conjurer_librarian/conjurer_librarian.py +++ b/conjurer_librarian/conjurer_librarian.py @@ -26,6 +26,7 @@ from queue import Queue import requests import scrape_bot import search_bot +#import search_bot2 as search_bot from flask import Flask, jsonify, request from habanero import Crossref from waitress import serve @@ -37,7 +38,10 @@ 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 = 7000 + +MAX_CR_RESULTS = 1000 +#TEST PURPOSES ONLY! +#MAX_CR_RESULTS = 5 ENCODING = "utf-8" @@ -114,14 +118,15 @@ class Librarian(object): self.app.logger.info("STARTED SEARCH") if not deep_search: - cr_result = self.cr.works(query=query, limit=1000) + query_limit = MAX_CR_RESULTS if MAX_CR_RESULTS < 1000 else 1000 + cr_result = self.cr.works(query=query, limit=query_limit) 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) + tmp_result = self.cr.works(query=query, limit=query_limit, 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"]) @@ -130,7 +135,7 @@ class Librarian(object): time.sleep(0.1) else: - cr_result = self.cr.works(query=query, cursor_max=10000, cursor='*', progress_bar = True) + cr_result = self.cr.works(query=query, cursor_max=15000, cursor='*', progress_bar = True) result = cr_result[0] for item in cr_result[1:]: result["message"]["items"].extend(item["message"]["items"]) diff --git a/conjurer_librarian/scrape_bot.py b/conjurer_librarian/scrape_bot.py index 91ecc2b..bc85a44 100644 --- a/conjurer_librarian/scrape_bot.py +++ b/conjurer_librarian/scrape_bot.py @@ -33,12 +33,14 @@ def load_ndb_to_q(logger): try: ndb_database = json.load(ndb_file) for _ in range (1,10): - key = next(iter(ndb_database)) - item = ndb_database.pop(key) - logger.info(key) - url = f"https://sci-hub.se/{key}" - WORK_Q.put([item, url, False]) - + try: + key = next(iter(ndb_database)) + _ = ndb_database.pop(key) + logger.info(key) + url = f"https://sci-hub.se/{key}" + WORK_Q.put([key, url, False]) + except StopIteration: + break ndb_file.truncate(0) ndb_file.seek(0) json.dump(ndb_database, ndb_file, indent=4) diff --git a/conjurer_librarian/search_bot.py b/conjurer_librarian/search_bot.py index e285979..c310c4b 100644 --- a/conjurer_librarian/search_bot.py +++ b/conjurer_librarian/search_bot.py @@ -21,7 +21,7 @@ Global Variables: # TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry from queue import Empty, Queue from threading import Thread - +import time q = Queue() # DATA FOR TEST ONLY @@ -49,9 +49,17 @@ def producer(out_q, control_q, filename, _logger): _logger: Logger object for logging. """ with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file: - _logger.info(f"Worker {filename} ") + print(f"Worker {filename} ") + line_no = 0 while True: line = operated_file.readline() + line_no += 1 + print(f"\t \t \t \t \t \t W{filename}{line_no}\r", end="") + + if not line: + print(f"EOF {filename}") + break + if not line: break out_q.put(line) @@ -59,11 +67,12 @@ def producer(out_q, control_q, filename, _logger): check = control_q.get(block=False) except Empty: check = False + if check is _sentinel: - _logger.info("TERM signal received") + print("TERM signal received") control_q.put(check) break - _logger.info(f"Worker finished: {filename}") + print(f"Worker finished: {filename}") out_q.put(_sentinel) @@ -79,19 +88,25 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, no, live_results (list): List to store the search results. _logger: Logger object for logging. """ - _logger.info("Consumer thread started:%s", no) - for item in doi: - result_list.append({"DOI": item[0], "exists": False, "data": item[1]}) + print(f"Consumer thread started: {no} no") + empty_counter = 0 + alive_no = 0 while True: done_check = True try: - data = in_q.get(block=False) + data = in_q.get(block=True, timeout = 1) if data is _sentinel: - _logger.info("Worker finished") control_dict["sentinels"] += 1 + print(f"Workers finished: {control_dict['sentinels']} reported by consumer {no}") + else: + empty_counter = 0 + alive_no += 1 + print(f"C{no}__{alive_no}\r", end="") + for item in result_list: if item["DOI"] in data and not item["exists"]: + print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item["exists"]}") _logger.info(data) _logger.info("HIT") item["exists"] = True @@ -100,9 +115,18 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, no, if done_check: control_q.put(_sentinel) except Empty: - pass + empty_counter += 1 + time.sleep(1) + print(f"Consumer {no} empty") + if empty_counter > 5: + print("Consumer %s empty lvl 2", no) + time.sleep(2) + elif empty_counter > 10: + print(f"Consumer thread finished {no}") + break + if control_dict["sentinels"] >= MAXTHREADS: - _logger.info("All workers finished") + _logger.info(f"All workers finished {no}") break @@ -121,6 +145,10 @@ def search_for_doi(doi, live_results, _logger): threads = [] work_q = Queue(maxsize=WORK_Q_SIZE) control_q = Queue() + + for item in doi: + result_list.append({"DOI": item[0], "exists": False, "data": item[1]}) + for i in range (0, (len(doi)//1000)+2): t_cons = Thread( target=consumer, args=(work_q, control_q, doi, live_results, result_list, control_dict, i, _logger) diff --git a/conjurer_librarian/search_bot2.py b/conjurer_librarian/search_bot2.py new file mode 100644 index 0000000..407b626 --- /dev/null +++ b/conjurer_librarian/search_bot2.py @@ -0,0 +1,206 @@ +""" +This module contains functions for searching for DOI (Digital Object Identifier) +in a list of live results. +It includes a producer-consumer pattern implementation using threads and queues. + +Functions: +- producer: Reads lines from a file and puts them into an output queue. +- consumer: Consumes items from an input queue and checks if DOI exists in the live results list. +- search_for_doi: Searches for DOI in live results using the producer-consumer pattern. + +Global Variables: +- MAXTHREADS: Maximum number of worker threads. +- DATABASE_PATH: Path to the database files. +- ENCODING: Encoding of the database files. +- CHUNK: File name pattern for the database files. +- _sentinel: Sentinel object used to signal termination. +- result_list: List to store the search results. +- WORK_Q_SIZE: Maximum size of the work queue. +""" + +# TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry +from multiprocessing import Manager, Process, Queue +from queue import Empty, Full +import time + +# DATA FOR TEST ONLY +#MAXTHREADS = 6 +#DATABASE_PATH = r'C:\\Database\\chunks_1\\' + +# DEPLOYMENT DATA +MAXTHREADS = 41 +DATABASE_PATH = r"C:\\Database\\chunks\\" + +ENCODING = "utf-8" +CHUNK = "_chunk.txt" +_sentinel = None +WORK_Q_SIZE = 355000 + + +def producer(out_q, control_q, filename, _logger): + """ + Produces items from the output queue and puts them into the control queue. + + Args: + out_q (Queue): Output queue. + control_q (Queue): Control queue. + filename (str): Name of the file. + _logger: Logger object for logging. + """ + with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file: + print(f"Worker {filename} ") + line_no = 0 + while True: + line = operated_file.readline() + line_no += 1 + print(f"\t \t \t W{filename}{line_no}\r", end="") + + if not line: + print(f"EOF {filename}") + break + try: + out_q.put([line, line_no, filename], block = True, timeout = 60) + except Full: + print("queue fulll") + try: + check = control_q.get(block=False) + except Empty: + check = False + if check is None: + print(f"TERM signal received {filename}") + control_q.put(check) + control_q.close() + break + print(f"Worker finished: {filename}") + out_q.put(_sentinel) + out_q.close() + + + +# IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK +def consumer(in_q, control_q, doi, live_results, result_list, control_dict, no, _logger): + """ + Consumes items from an input queue and checks if DOI exists in the live results list. + + Args: + in_q (Queue): Input queue. + control_q (Queue): Control queue. + doi (list): List of DOI to search for. + live_results (list): List to store the search results. + _logger: Logger object for logging. + """ + print(f"Consumer thread started: {no} no") + empty_counter = 0 + alive_no = 0 + while True: + done_check = True + try: + data = in_q.get(block=True, timeout = 1) + empty_counter = 0 + if data is None: + control_dict["sentinels"] += 1 + print(f"Workers finished: {control_dict['sentinels']} reported by consumer {no}") + else: + empty_counter = 0 + if alive_no > 9: + alive_no = 0 + alive_no += 1 + print(f"C{no}{alive_no}\r", end="") + for item in result_list: + if item["DOI"] in data[0] and not item["exists"]: + print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item["exists"]}") + item["exists"] = True + live_results.append(item) + done_check = done_check and item["exists"] + if done_check: + control_q.put(_sentinel) + except Empty: + empty_counter += 1 + time.sleep(1) + print(f"Consumer {no} empty") + if empty_counter > 5: + print("Consumer %s empty lvl 2", no) + time.sleep(2) + elif empty_counter > 10: + print(f"Consumer thread finished {no}") + break + if control_dict["sentinels"] >= MAXTHREADS: + print(f"All workers finished killing {no}") + break + print("Consumer thread finished %s",no ) + control_q.close() + + +def search_for_doi(doi, live_results, _logger): + """ + Search for DOI in live_results using _logger for logging. + + Args: + doi (list): List of DOI to search for. + live_results (list): List to store the search results. + _logger: Logger object for logging. + """ + with Manager() as manager: + control_dict = manager.dict() + control_dict["sentinels"] = 0 + result_list = [] + for item in doi: + result_list.append({"DOI": item[0], "exists": False, "data": item[1]}) + result_list_proxy = manager.list(result_list) + + threads = [] + work_q = Queue(maxsize=WORK_Q_SIZE) + control_q = Queue() + for i in range(0, MAXTHREADS): + # TEST DATA + #filename = "test" + str(i) + CHUNK + # DEPLOYMENT DATA + filename = str(i) + CHUNK + print(f"Creating worker thread no: {i}") + threads.append( + Process(target=producer, args=(work_q, control_q, filename, _logger)) + ) + for i in range (0, (len(doi)//1000)+5): + t_cons = Process( + target=consumer, args=(work_q, control_q, doi, live_results, result_list_proxy, control_dict, i, _logger) + ) + print("Consumer thread created") + threads.append(t_cons) + + for worker in threads: + worker.start() + for worker in threads: + worker.join() + result_list = list(result_list_proxy) + for worker in threads: + worker.close() + return result_list + + +if __name__ == "__main__": + import logging + + logger = logging.getLogger() + logger.setLevel("DEBUG") + h1 = logging.StreamHandler() + logger.addHandler(h1) + print("TEST RUN") + live_result = [] + print( + search_for_doi( + [ + ("10.1002/9781118786352.wbieg0998.pub2", "DATA"), + ("10.1002/j.2050-0416.2002.tb00563.x", "DATA"), + ("10.1111/j.1365-2958.1994.tb00448.x", "DATA"), + ("10.2165/00128415-200309690-00017", "DATA"), + ("10.5772/48313", "DATA"), + ("10.15803/ijnc.7.2_419", "DATA"), + ("10.1051/0004-6361/201321596e", "DATA"), + ("10.2307/40835941", "DATA"), + ("NOT_ON_LIST", "DATA"), + ], + live_result, + logger, + ) + ) + print(live_result)