Serverside work of search

conjurer_librarian/search_bot.py

@@ -0,0 +1,106 @@
+from queue import Queue, Empty
+from threading import Thread
+q = Queue()
+
+#TODO: TO w klasie jako parametry wybierane czy robimy pelen run czy testowy
+#MAXTHREADS = 3
+MAXTHREADS = 1
+
+#DATABASE_PATH = r'C:\\Database\\chunks_1\\'
+DATABASE_PATH = r'C:\\Database\\chunks\\'
+
+ENCODING = "utf-8"
+CHUNk = "_chunk.txt"
+_sentinel = object()
+result_list = []
+
+#TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
+
+def producer(out_q, control_q, filename):
+    with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file:
+        lines = operated_file.readlines()
+        for line in lines:
+            out_q.put(line)
+            try:
+                check = control_q.get(block=False)
+            except Empty:
+                check = False
+                pass
+            if check is _sentinel:
+                print("TERM signal received")
+                control_q.put(check)
+                break
+    print(f"Worker finished: {filename}")
+    out_q.put(_sentinel)
+ 
+#IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK
+#TODO: Change doi to tuple list and only put term when all doi are found => (doi, True)
+def consumer(in_q, control_q, doi, live_results, logger):
+    sentinels = 0
+    for item in doi:
+        result_list.append({"DOI" : item[0], "exists" : False, "data" : item[1]})
+    while True:
+        #if in_q.empty():
+        #    print("Queue empty!")
+        #if in_q.full():
+        #    print("Queue full!")
+        done_check = True
+        data = in_q.get()
+        if data is _sentinel:
+            logger.info("Worker finished")
+            sentinels += 1
+        else:
+            for item in result_list:
+                if item["DOI"] in data:
+                    logger.info(data)
+                    logger.info("HIT")
+                    item["exists"] = True
+                    live_results.append(item)
+                done_check = done_check and item["exists"]
+            if done_check:
+                control_q.put(_sentinel)
+        if sentinels >= MAXTHREADS:
+            logger.info("All workers finished")
+            break
+
+
+def search_for_doi(doi, live_results, logger):
+    threads = []
+    work_q = Queue(maxsize=200000)
+    control_q = Queue()
+    t_cons = Thread(target = consumer, args = (work_q, control_q, doi, live_results, logger))
+    logger.info("Consumer thread created")
+    threads.append(t_cons)
+    for i in range(0,MAXTHREADS):
+        #filename = "test" + str(i) + CHUNk
+        filename = "40" + CHUNk
+        #filename = str(i) + CHUNk
+        logger.info(f"Creating worker thread no: {i}")
+        threads.append(Thread(target = producer, args = (work_q, control_q, filename)))
+    for worker in threads:
+        worker.start()
+    for worker in threads:
+        worker.join()
+    return result_list
+
+
+
+if __name__ == "__main__":
+    import logging
+    logger = logging.getLogger()
+    logger.setLevel("DEBUG")
+    h1 = logging.StreamHandler()
+    logger.addHandler(h1)
+    logger.info("TEST RUN")
+    live_result = []
+    logger.info(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"),
+        ], live_result, logger))
+    logger.info(live_result)