Code_review

docker-compose.yml

@@ -1,8 +1,8 @@
 services:
-  data_processing_save_ids:
+  data_processing_save_labels_descriptions:
     build:
       context: .
-      dockerfile: ./docker/1_Data_Processing_save_ids/Dockerfile
+      dockerfile: ./docker/1_Data_Processing_save_labels_descriptions/Dockerfile
     volumes:
       - ./data:/data  # Mount the ./data folder from the host to /data in the container
     tty: true
@@ -12,10 +12,10 @@ services:
       LANGUAGE: "de"
       OFFSET: 0
 
-  data_processing_save_entities:
+  data_processing_save_items_per_lang:
     build:
       context: .
-      dockerfile: ./docker/2_Data_Processing_save_entities/Dockerfile
+      dockerfile: ./docker/2_Data_Processing_save_items_per_lang/Dockerfile
     volumes:
       - ./data:/data  # Mount the ./data folder from the host to /data in the container
     tty: true
@@ -43,12 +43,16 @@ services:
       PYTHONUNBUFFERED: 1
       MODEL: "jina"
       SAMPLE: "true"
-      API_KEY: "datastax_wikidata_nvidia.json"
+      API_KEY: "datastax_wikidata2.json"
       EMBED_BATCH_SIZE: 8
-      QUERY_BATCH_SIZE: 1000
-      OFFSET: 2560000
-      COLLECTION_NAME: "wikidata_test_v1"
-      LANGUAGE: 'ar'
+      QUERY_BATCH_SIZE: 100
+      OFFSET: 120000
+      COLLECTION_NAME: "wikidatav1"
+      LANGUAGE: 'en'
+      TEXTIFIER_LANGUAGE: 'en'
+      ELASTICSEARCH_URL: "http://localhost:9200"
+      ELASTICSEARCH: "false"
+    network_mode: "host"
 
   run_retrieval:
     build:
@@ -70,17 +74,16 @@ services:
     environment:
       PYTHONUNBUFFERED: 1
       MODEL: "jina"
-      API_KEY: "datastax_wikidata_nvidia.json"
-      COLLECTION_NAME: "wikidata_test_v1"
+      API_KEY: "datastax_wikidata.json"
+      COLLECTION_NAME: "wikidata_texttest"
       BATCH_SIZE: 100
-      EVALUATION_PATH: "Mintaka/processed_dataframe_langtest.pkl"
+      EVALUATION_PATH: "Mintaka/processed_dataframe.pkl"
       # COMPARATIVE: "true"
       # COMPARATIVE_COLS: "Correct QID,Wrong QID"
       QUERY_COL: "Question"
-      # QUERY_LANGUAGE: "ar"
-      # DB_LANGUAGE: "en,ar"
+      QUERY_LANGUAGE: "en"
+      # DB_LANGUAGE: "en"
       PREFIX: ""
-      ELASTICSEARCH_URL: "http://localhost:9200"
     network_mode: "host"
 
   run_rerank:
@@ -107,4 +110,42 @@ services:
       BATCH_SIZE: 1
       QUERY_COL: "Question"
       LANGUAGE: "de"
-    network_mode: "host"
+    network_mode: "host"
+
+  push_huggingface:
+    build:
+      context: .
+      dockerfile: ./docker/6_Push_Huggingface/Dockerfile
+    volumes:
+      - ./data:/data
+    tty: true
+    container_name: push_huggingface
+    environment:
+      PYTHONUNBUFFERED: 1
+      QUEUE_SIZE: 5000
+      NUM_PROCESSES: 4
+      SKIPLINES: 0
+      ITERATION: 36
+
+  create_prototype:
+      build:
+        context: .
+        dockerfile: ./docker/7_Create_Prototype/Dockerfile
+      volumes:
+        - ./data:/data
+        - ~/.cache/huggingface:/root/.cache/huggingface
+      tty: true
+      container_name: create_prototype
+      environment:
+        PYTHONUNBUFFERED: 1
+        MODEL: "jinaapi"
+        API_KEY: "datastax_wikidata.json"
+        EMBED_BATCH_SIZE: 100
+        # QUEUE_SIZE: 5000
+        NUM_PROCESSES: 23
+        OFFSET: 0
+        COLLECTION_NAME: "wikidata_prototype"
+        LANGUAGE: 'en'
+        TEXTIFIER_LANGUAGE: 'en'
+        # CHUNK_NUM: 5
+      network_mode: "host"

docker/1_Data_Processing_save_ids/Dockerfile → docker/1_Data_Processing_save_labels_descriptions/Dockerfile

@@ -10,17 +10,19 @@ LABEL maintainer="philippe.saade@wikimedia.de"
 WORKDIR /app
 
 # Copy the requirements file into the container
-COPY ./docker/1_Data_Processing_save_ids/requirements.txt requirements.txt
+COPY ./docker/1_Data_Processing_save_labels_descriptions/requirements.txt requirements.txt
 
 # Install the dependencies
 RUN pip install --no-cache-dir -r requirements.txt
 
 # Copy the rest of the application code into the container
-COPY ./docker/1_Data_Processing_save_ids /app
+COPY ./docker/1_Data_Processing_save_labels_descriptions /app
 COPY ./src /src
 
 # Set up the volume for the data folder
 VOLUME [ "/data" ]
 
+ENV PYTHONPATH="${PYTHONPATH}:/"
+
 # Run the Python script
-CMD ["python", "run.py"]
+CMD ["python", "run.py"]

docker/1_Data_Processing_save_labels_descriptions/run.py

@@ -0,0 +1,69 @@
+from multiprocessing import Manager
+import os
+import time
+import json
+
+from src.wikidataDumpReader import WikidataDumpReader
+from src.wikidataItemDB import WikidataItem
+
+FILEPATH = os.getenv("FILEPATH", '../data/Wikidata/latest-all.json.bz2')
+PUSH_SIZE = int(os.getenv("PUSH_SIZE", 20000))
+QUEUE_SIZE = int(os.getenv("QUEUE_SIZE", 15000))
+NUM_PROCESSES = int(os.getenv("NUM_PROCESSES", 4))
+SKIPLINES = int(os.getenv("SKIPLINES", 0))
+LANGUAGE = os.getenv("LANGUAGE", 'en')
+
+
+def save_items_to_sqlite(item, data_batch, sqlitDBlock):
+    if (item is not None):
+        labels = WikidataItem.clean_label_description(item['labels'])
+        descriptions = WikidataItem.clean_label_description(
+            item['descriptions']
+        )
+        labels = json.dumps(labels, separators=(',', ':'))
+        descriptions = json.dumps(descriptions, separators=(',', ':'))
+        in_wikipedia = WikidataItem.is_in_wikipedia(item)
+        data_batch.append({
+            'id': item['id'],
+            'labels': labels,
+            'descriptions': descriptions,
+            'in_wikipedia': in_wikipedia,
+        })
+
+        with sqlitDBlock:
+            if len(data_batch) > PUSH_SIZE:
+                worked = WikidataItem.add_bulk_items(list(
+                    data_batch[:PUSH_SIZE]
+                ))
+                if worked:
+                    del data_batch[:PUSH_SIZE]
+
+
+if __name__ == "__main__":
+    multiprocess_manager = Manager()
+    sqlitDBlock = multiprocess_manager.Lock()
+    data_batch = multiprocess_manager.list()
+
+    wikidata = WikidataDumpReader(
+        FILEPATH,
+        num_processes=NUM_PROCESSES,
+        queue_size=QUEUE_SIZE,
+        skiplines=SKIPLINES
+    )
+
+    wikidata.run(
+        lambda item: save_items_to_sqlite(
+            item,
+            data_batch,
+            sqlitDBlock
+        ),
+        max_iterations=None,
+        verbose=True
+    )
+
+    while len(data_batch) > 0:
+        worked = WikidataItem.add_bulk_items(list(data_batch))
+        if worked:
+            del data_batch[:PUSH_SIZE]
+        else:
+            time.sleep(1)

docker/2_Data_Processing_save_entities/Dockerfile → docker/2_Data_Processing_save_items_per_lang/Dockerfile

@@ -22,5 +22,7 @@ COPY ./src /src
 # Set up the volume for the data folder
 VOLUME [ "/data" ]
 
+ENV PYTHONPATH="${PYTHONPATH}:/"
+
 # Run the Python script
-CMD ["python", "run.py"]
+CMD ["python", "run.py"]

docker/2_Data_Processing_save_items_per_lang/run.py

@@ -0,0 +1,73 @@
+from multiprocessing import Manager
+import os
+import time
+
+from src.wikidataDumpReader import WikidataDumpReader
+from src.wikidataLangDB import WikidataLang
+
+FILEPATH = os.getenv("FILEPATH", '../data/Wikidata/latest-all.json.bz2')
+PUSH_SIZE = int(os.getenv("PUSH_SIZE", 2000))
+QUEUE_SIZE = int(os.getenv("QUEUE_SIZE", 1500))
+NUM_PROCESSES = int(os.getenv("NUM_PROCESSES", 8))
+SKIPLINES = int(os.getenv("SKIPLINES", 0))
+LANGUAGE = os.getenv("LANGUAGE", 'en')
+
+
+def save_entities_to_sqlite(item, data_batch, sqliteDBlock):
+    """_summary_
+    # TODO Add a docstring
+
+    Args:
+        item (_type_): _description_
+        data_batch (_type_): _description_
+        sqliteDBlock (_type_): _description_
+    """
+    if item is not None:
+        # Check if the item is a valid entity
+        return
+
+    lang_in_wp = WikidataLang.is_in_wikipedia(item, language=LANGUAGE)
+    if not lang_in_wp:
+        # If the entity is not in the specified language Wikipedia, skip
+        return
+
+    item = WikidataLang.normalise_item(item, language=LANGUAGE)
+    data_batch.append(item)
+
+    with sqliteDBlock:
+        if len(data_batch) > PUSH_SIZE:
+            worked = WikidataLang.add_bulk_entities(list(
+                data_batch[:PUSH_SIZE]
+            ))
+            if worked:
+                del data_batch[:PUSH_SIZE]
+
+
+if __name__ == "__main__":
+    multiprocess_manager = Manager()
+    sqliteDBlock = multiprocess_manager.Lock()
+    data_batch = multiprocess_manager.list()
+
+    wikidata = WikidataDumpReader(
+        FILEPATH,
+        num_processes=NUM_PROCESSES,
+        queue_size=QUEUE_SIZE,
+        skiplines=SKIPLINES
+    )
+
+    wikidata.run(
+        lambda item: save_entities_to_sqlite(
+            item,
+            data_batch,
+            sqliteDBlock
+        ),
+        max_iterations=None,
+        verbose=True
+    )
+
+    while len(data_batch) > 0:
+        worked = WikidataLang.add_bulk_entities(list(data_batch))
+        if worked:
+            del data_batch[:PUSH_SIZE]
+        else:
+            time.sleep(1)

docker/3_Add_Wikidata_to_AstraDB/Dockerfile

@@ -28,5 +28,7 @@ COPY ./API_tokens /API_tokens
 # Set up the volume for the data folder
 VOLUME [ "/data" ]
 
+ENV PYTHONPATH="${PYTHONPATH}:/"
+
 # Run the Python script
-CMD ["python", "run.py"]
+CMD ["python", "run.py"]

docker/3_Add_Wikidata_to_AstraDB/requirements.txt

@@ -17,5 +17,4 @@ langchain_experimental
 ragstack-ai-langchain[knowledge-store]==1.3.0
 langchain-astradb
 astrapy
-elasticsearch
-mediawikiapi
+elasticsearch