Feature/pdf processed by tika

k8s/welearn-datastack/values.yaml

@@ -27,6 +27,7 @@ common:
       RCLONE_CONFIG_DEST_ENV_AUTH: true
       RCLONE_CONFIG_DEST_TYPE: azureblob
       TEAM_EMAIL: welearn@learningplanetinstitute.org
+      TIKA_ADDRESS: http://tika:9998
   embeddingModelFr: sentence-camembert-base
   embeddingModelEn: all-minilm-l6-v2
   modelsPathRoot: /models

pyproject.toml

@@ -29,9 +29,8 @@ requests = "^2.32.4"
 wikipedia-api = "^0.8.1"
 sentence-transformers = "^4.1.0"
 spacy = "^3.8.7"
-pypdf = "^5.6.0"
-refinedoc = "^0.0.3"
-qdrant-client = "^1.14.2"
+refinedoc = "^1.0.0"
+qdrant-client = "1.12.2"
 python-dotenv = "^1.1.0"
 beautifulsoup4 = "^4.13.4"
 pyphen = "^0.17.2"
@@ -46,6 +45,7 @@ psycopg2-binary = "^2.9.10"
 brotli = "^1.1.0"
 scikit-learn = "~=1.6.1"
 optimum = {extras = ["onnxruntime"], version = "^1.26.1"}
+azure-storage-blob = "^12.26.0"
 
 [tool.poetry.group.metrics.dependencies]
 alembic = "^1.16.1"

sql/89920abb7ff8_populate_corpus_category.sql

@@ -4,7 +4,7 @@ INSERT
 	INTO
 		corpus_related.category(title)
 	VALUES
-		('Academic scientific publications')
+		('academic scientific publications')
     RETURNING id AS catid
 )
 UPDATE

tests/document_collector_hub/plugins_test/test_hal.py

@@ -125,12 +125,15 @@ def test__convert_json_dict_to_welearndoc(self):
             doc0["abstract_s"][0].split(".")[0] + "...",
         )
 
+    @patch("welearn_datastack.modules.pdf_extractor._send_pdf_to_tika")
     @patch(
         "welearn_datastack.plugins.rest_requesters.hal.HAL_URL_BASE",
         "https://example.org/",
     )
     @patch("requests.Session.get")
-    def test__convert_json_dict_to_welearndoc_mode_pdf(self, mock_get):
+    def test__convert_json_dict_to_welearndoc_mode_pdf(
+        self, mock_get, mock_send_pdf_to_tika
+    ):
         class MockResponse:
             def __init__(self, status_code):
                 self.content = (
@@ -144,6 +147,9 @@ def __init__(self, status_code):
             def raise_for_status(self):
                 pass
 
+        mock_send_pdf_to_tika.return_value = {
+            "X-TIKA:content": "<div class='page'>For primary vpiRNAs that are produced from the abundant</div>"
+        }
         mock_get.side_effect = [MockResponse(200)]
         os.environ["PDF_SIZE_PAGE_LIMIT"] = "100000"
         doc0 = self.content_json["response"]["docs"][0]

tests/document_collector_hub/plugins_test/test_oapen.py

@@ -80,8 +80,9 @@ def test_get_oapen_url_from_handle_id(self):
             "https://library.oapen.org/handle/20.500.12657/12345",
         )
 
+    @patch("welearn_datastack.modules.pdf_extractor._send_pdf_to_tika")
     @patch("welearn_datastack.plugins.rest_requesters.oapen.get_new_https_session")
-    def test_get_pdf_content(self, mock_get_new_https_session):
+    def test_get_pdf_content(self, mock_get_new_https_session, mock_send_pdf_to_tika):
         class MockResponse:
             def __init__(self, status_code):
                 self.content = (
@@ -96,6 +97,9 @@ def raise_for_status(self):
                 pass
 
         os.environ["PDF_SIZE_PAGE_LIMIT"] = "100000"
+        mock_send_pdf_to_tika.return_value = {
+            "X-TIKA:content": "<div class='page'>For primary vpiRNAs that are produced from the abundant</div>"
+        }
 
         mock_response = MockResponse(200)
         mock_get_new_https_session.return_value.get.return_value = mock_response

tests/document_collector_hub/plugins_test/test_open_alex.py

@@ -72,16 +72,19 @@ def test__generate_api_query_params(self):
             tested_query,
         )
 
+    @patch("welearn_datastack.modules.pdf_extractor._send_pdf_to_tika")
     @patch("welearn_datastack.plugins.rest_requesters.open_alex.get_new_https_session")
-    def test__get_pdf_content(self, http_session_mock):
+    def test__get_pdf_content(self, http_session_mock, mock_send_pdf_to_tika):
 
         mock_session = Mock()
         http_session_mock.return_value = mock_session
 
         mock_session.get.return_value = MockResponse(
             status_code=200, content=self.pdf.read_bytes()
         )
-
+        mock_send_pdf_to_tika.return_value = {
+            "X-TIKA:content": "<div class='page'>2.2. Measurements of Fiber Parameters Small pieces were extracted from various positions on the strip. and mechanical properties to provide additional information regarding these areas of study and growth conditions.</div>"
+        }
         tested_result = self.openalexColector._get_pdf_content("https://example.org/1")
         self.assertTrue(
             tested_result.startswith(

tests/document_collector_hub/test_pdf_extractor.py

@@ -1,24 +1,16 @@
+import io
 import unittest
-from pathlib import Path
+from unittest.mock import MagicMock, patch
 
-from pypdf import PdfReader
-
-from welearn_datastack.data.enumerations import DeletePart
 from welearn_datastack.modules import pdf_extractor
+from welearn_datastack.modules.pdf_extractor import (
+    _parse_tika_content,
+    _send_pdf_to_tika,
+    extract_txt_from_pdf_with_tika,
+)
 
 
 class TestPDFExtractor(unittest.TestCase):
-    def setUp(self):
-        resources_fp = (
-            Path(__file__).parent / "resources" / "file_plugin_input" / "hal_pdf.pdf"
-        )
-
-        self.reader = PdfReader(resources_fp)
-
-    def test_extract_txt_from_pdf(self):
-        pdf_content = pdf_extractor.extract_txt_from_pdf(self.reader)
-        self.assertEqual(len(pdf_content), len(self.reader.pages))
-
     def test_replace_ligatures(self):
         text = "first flight"
         cleaned_text = pdf_extractor.replace_ligatures(text)
@@ -34,12 +26,67 @@ def test_remove_hyphens(self):
         cleaned_text = pdf_extractor.remove_hyphens(text)
         self.assertEqual(cleaned_text, "wellknown\n")
 
-    def test_check_page_size_positive(self):
-        ret = pdf_extractor.large_pages_size_flag(self.reader, 10000)
-        self.assertEqual(len(ret[0]), len(self.reader.pages))
-        self.assertTrue(ret[1])
+    @patch("welearn_datastack.modules.pdf_extractor.get_new_https_session")
+    def test_send_pdf_to_tika(self, mock_get_session):
+        # Mock de la session HTTP
+        mock_session = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "X-TIKA:content": "<html>Mock Content</html>"
+        }
+        mock_response.raise_for_status.return_value = None
+        mock_session.put.return_value = mock_response
+        mock_get_session.return_value.__enter__.return_value = mock_session
+
+        # Appel de la méthode
+        pdf_content = io.BytesIO(b"Mock PDF content")
+        tika_base_url = "http://mock-tika-url"
+        result = _send_pdf_to_tika(pdf_content, tika_base_url)
+
+        # Assertions
+        mock_session.put.assert_called_once_with(
+            url=f"{tika_base_url}/tika",
+            files={"file": pdf_content},
+            headers={
+                "Accept": "application/json",
+                "Content-type": "application/octet-stream",
+                "X-Tika-PDFOcrStrategy": "no_ocr",
+            },
+        )
+        self.assertEqual(result, {"X-TIKA:content": "<html>Mock Content</html>"})
 
-    def test_check_page_size_negative(self):
-        ret = pdf_extractor.large_pages_size_flag(self.reader, 100000)
-        self.assertEqual(len(ret[0]), len(self.reader.pages))
-        self.assertFalse(ret[1])
+    def test_parse_tika_content(self):
+        tika_content = {
+            "X-TIKA:content": """
+            <html>
+                <div class="page">Page 1 content</div>
+                <div class="page">Page 2 content</div>
+            </html>
+            """
+        }
+
+        result = _parse_tika_content(tika_content)
+
+        expected_result = [["Page 1 content"], ["Page 2 content"]]
+        self.assertEqual(result, expected_result)
+
+    @patch("welearn_datastack.modules.pdf_extractor._send_pdf_to_tika")
+    @patch("welearn_datastack.modules.pdf_extractor._parse_tika_content")
+    def test_extract_txt_from_pdf_with_tika(
+        self, mock_parse_tika_content, mock_send_pdf_to_tika
+    ):
+        pdf_content = io.BytesIO(b"%PDF-1.4 simulated content")
+        tika_base_url = "http://localhost:9998"
+
+        mock_send_pdf_to_tika.return_value = {
+            "X-TIKA:content": "<div class='page'>Page 1 content</div>"
+        }
+        mock_parse_tika_content.return_value = [["Page 1 content"]]
+
+        result = extract_txt_from_pdf_with_tika(pdf_content, tika_base_url)
+
+        self.assertEqual(result, [["Page 1 content"]])
+        mock_send_pdf_to_tika.assert_called_once_with(pdf_content, tika_base_url)
+        mock_parse_tika_content.assert_called_once_with(
+            mock_send_pdf_to_tika.return_value
+        )

welearn_datastack/modules/pdf_extractor.py

@@ -1,49 +1,69 @@
+import io
 import logging
-from typing import List, Tuple
+from typing import List
 
-from pypdf import PdfReader
+from bs4 import BeautifulSoup
 from refinedoc.refined_document import RefinedDocument
 
+from welearn_datastack.utils_.http_client_utils import get_new_https_session
+
 logger = logging.getLogger(__name__)
 
 
-def large_pages_size_flag(reader: PdfReader, limit: int) -> Tuple[List[int], bool]:
+def _send_pdf_to_tika(pdf_content: io.BytesIO, tika_base_url: str) -> dict:
+    """
+    Send a PDF document to Tika micro service and return the content as a dictionary
+    :param pdf_content: the PDF document content as a byte stream
+    :param tika_base_url: the base URL of the Tika micro service
+    :return: the content returned by Tika micro service as a dictionary (JSON)
+    """
+    if tika_base_url.endswith("/"):
+        tika_base_url = tika_base_url[:-1]
+    pdf_process_addr = f"{tika_base_url}/tika"
+    local_headers = {
+        "Accept": "application/json",
+        "Content-type": "application/octet-stream",
+        "X-Tika-PDFOcrStrategy": "no_ocr",
+    }
+
+    with get_new_https_session() as http_session:
+        resp = http_session.put(
+            url=pdf_process_addr,
+            files={"file": pdf_content},
+            headers=local_headers,
+        )
+        resp.raise_for_status()
+        tika_content = resp.json()
+    return tika_content
+
+
+def _parse_tika_content(tika_content: dict) -> list[list[str]]:
     """
-    Check the size of a PDF document page
-    :param limit: In byte, limit the pdf need to not exceed
-    :param reader: The PDF document already opened. Each page gonna be processed.
-    :return: List of sizes for each page (index in the list equal index in pdf) and if one of this exceed limit
+    Parse the content returned by Tika micro service
+    :param tika_content: the content returned by Tika micro service
+    :return: the parsed content as a list of list of strings (one list per page
     """
-    logger.info(f"Size limit for each page : {limit}")
-    ret = []
-    number_of_pages = len(reader.pages)
-    logger.info(f"Test size of {number_of_pages} pages")
-    flag = False
-    for i in range(number_of_pages):
-        page = reader.pages[i]
-        page_size = len(page.get_contents().get_data())  # type: ignore
-        ret.append(page_size)
-        if page_size > limit:
-            flag = True
+    htmlx = tika_content.get("X-TIKA:content")
+    soup = BeautifulSoup(htmlx, features="html.parser")
+    pages = soup.find_all("div", {"class": "page"})
+    res = [p.split("\n") for p in [page.get_text() for page in pages]]
 
-    return ret, flag
+    return res
 
 
-def extract_txt_from_pdf(
-    reader: PdfReader, remove_headers: bool = True, remove_footers: bool = True
+def extract_txt_from_pdf_with_tika(
+    pdf_content: io.BytesIO, tika_base_url: str
 ) -> List[List[str]]:
     """
     Extract the text from a PDF document and return it as a list of strings for each page of the document and a list of
-    strings for each page for a filtered document and the reference document (extracted with PyPDF)
+    strings for each page for a filtered document and the reference document (extracted with tika micro service)
 
-    :param reader: the PDF reader object
-    :return: a tuple containing the extracted & filtered text
+    :param pdf_content: the PDF document content as a byte stream
+    :param tika_base_url: the base URL of the Tika micro service
+    :return: Matrix of strings (list of list of strings) for each page of the document
     """
-    pdf_content: List[List[str]] = []
-    for page in reader.pages:
-        text = page.extract_text().split("\n")
-        page_content = [t.strip() for t in text if t.strip()]
-        pdf_content.append(page_content)
+    tika_content = _send_pdf_to_tika(pdf_content, tika_base_url)
+    pdf_content = _parse_tika_content(tika_content)
 
     refined_pdf_content = RefinedDocument(content=pdf_content)
 

welearn_datastack/nodes_workflow/DocumentPDFProcessor/document_pdf_processor.py

@@ -0,0 +1,64 @@
+import csv
+import logging
+import os
+from uuid import UUID
+
+from sqlalchemy.orm import Session
+
+from welearn_datastack.data.db_models import WeLearnDocument
+from welearn_datastack.utils_.database_utils import create_db_session
+from welearn_datastack.utils_.path_utils import setup_local_path
+
+log_level: int = logging.getLevelName(os.getenv("LOG_LEVEL", "INFO"))
+log_format: str = os.getenv(
+    "LOG_FORMAT", "[%(asctime)s][%(name)s][%(levelname)s] - %(message)s"
+)
+
+if not isinstance(log_level, int):
+    raise ValueError("Log level is not recognized : '%s'", log_level)
+
+logging.basicConfig(
+    level=logging.getLevelName(log_level),
+    format=log_format,
+)
+logger = logging.getLogger(__name__)
+
+
+def main() -> None:
+    logger.info("DocumentPDFProcessor starting...")
+    input_artifact_id_url = os.getenv("ARTIFACT_ID_URL_CSV_NAME", "batch_ids.csv")
+    logger.info("Input artifact url csv name: %s", input_artifact_id_url)
+    local_artifcat_input, local_artifcat_output = setup_local_path()
+
+    # retrieve url data from files
+    logger.info("Retrieve URLs from file")
+
+    # Input IDs
+    with (local_artifcat_input / input_artifact_id_url).open(
+        "r"
+    ) as artifact_file_input:
+        spamreader = csv.reader(artifact_file_input, delimiter=",", quotechar='"')
+        ids_urls: list[UUID] = [UUID(row[0]) for row in spamreader]
+        logger.info("'%s' IDs URLs were retrieved", len(ids_urls))
+
+    # Database management
+    logger.info("Create DB session")
+    db_session: Session = create_db_session()
+    logger.info("DB session created")
+
+    # Retrieve WeLearnDocument from database
+    logger.info("Retrieve WeLearnDocument from database")
+    welearn_documents: list = (
+        db_session.query(WeLearnDocument).filter(WeLearnDocument.id.in_(ids_urls)).all()
+    )
+    logger.info("'%s' WeLearnDocuments were retrieved", len(welearn_documents))
+
+    if len(ids_urls) != len(welearn_documents):
+        logger.warning(
+            "'%s' IDs URLs were not found in the database",
+            len(ids_urls) - len(welearn_documents),
+        )
+
+    if not welearn_documents:
+        logger.info("No WeLearnDocuments were retrieved")
+        return