New deasrc

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)