Tag: 1.3

Intermediate commits (oldest → newest):
- Almost deployment versio - for full deployment change parameteres in search_bot
- Preparation for deployment
- Main bot (#1)
- Search webservice (#2)
- Music service (#3)
- FIx

conjurer_librarian/search_bot.py

@@ -1,25 +1,30 @@
+#TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
+
 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
-
+#TODO: DATA FOR TEST ONLY
+#MAXTHREADS = 5
 #DATABASE_PATH = r'C:\\Database\\chunks_1\\'
+
+#TODO: DEPLOYMENT DATA
+MAXTHREADS = 41
 DATABASE_PATH = r'C:\\Database\\chunks\\'
 
 ENCODING = "utf-8"
 CHUNk = "_chunk.txt"
 _sentinel = object()
 result_list = []
+WORK_Q_SIZE = 15500000
 
-#TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
-
-def producer(out_q, control_q, filename):
+def producer(out_q, control_q, filename, logger):
     with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file:
-        lines = operated_file.readlines()
-        for line in lines:
+        logger.info(f"Worker {filename} ")
+        while True:
+            line = operated_file.readline()
+            if not line:
+                break
             out_q.put(line)
             try:
                 check = control_q.get(block=False)
@@ -27,11 +32,11 @@ def producer(out_q, control_q, filename):
                 check = False
                 pass
             if check is _sentinel:
-                print("TERM signal received")
+                logger.info("TERM signal received")
                 control_q.put(check)
                 break
-    print(f"Worker finished: {filename}")
-    out_q.put(_sentinel)
+        logger.info(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)
@@ -40,10 +45,6 @@ def consumer(in_q, control_q, doi, live_results, logger):
     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:
@@ -51,7 +52,7 @@ def consumer(in_q, control_q, doi, live_results, logger):
             sentinels += 1
         else:
             for item in result_list:
-                if item["DOI"] in data:
+                if item["DOI"] in data and not item["exists"]:
                     logger.info(data)
                     logger.info("HIT")
                     item["exists"] = True
@@ -66,25 +67,24 @@ def consumer(in_q, control_q, doi, live_results, logger):
 
 def search_for_doi(doi, live_results, logger):
     threads = []
-    work_q = Queue(maxsize=200000)
+    work_q = Queue(maxsize=WORK_Q_SIZE)
     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):
+        #TODO: TEST DATA
         #filename = "test" + str(i) + CHUNk
-        filename = "40" + CHUNk
-        #filename = str(i) + CHUNk
+        #TODO: DEPLOYMENT DATA
+        filename = str(i) + CHUNk
         logger.info(f"Creating worker thread no: {i}")
-        threads.append(Thread(target = producer, args = (work_q, control_q, filename)))
+        threads.append(Thread(target = producer, args = (work_q, control_q, filename,logger)))
     for worker in threads:
         worker.start()
     for worker in threads:
         worker.join()
     return result_list
 
-
-
 if __name__ == "__main__":
     import logging
     logger = logging.getLogger()