Merge pull request #240 from jwchmodx/fix/bugs-processor-parser-config

LarFii · web-flow · commit 050c59e4e30e · 2026-04-25T17:49:09.000+08:00
fix: prevent crashes from uninitialized LightRAG, env-var stripping, and parser cleanup
diff --git a/raganything/batch.py b/raganything/batch.py
@@ -70,7 +70,11 @@ async def process_folder_complete(
         if max_workers is None:
             max_workers = self.config.max_concurrent_files
 
-        await self._ensure_lightrag_initialized()
+        init_result = await self._ensure_lightrag_initialized()
+        if not init_result or not init_result.get("success"):
+            raise RuntimeError(
+                f"LightRAG initialization failed: {(init_result or {}).get('error', 'unknown error')}"
+            )
 
         # Get all files in the folder
         folder_path_obj = Path(folder_path)
@@ -105,10 +109,9 @@ async def process_folder_complete(
         async def process_single_file(file_path: Path):
             async with semaphore:
                 is_in_subdir = (
-                    lambda file_path, dir_path: len(
-                        file_path.relative_to(dir_path).parents
+                    lambda file_path, dir_path: (
+                        len(file_path.relative_to(dir_path).parents) > 1
                     )
-                    > 1
                 )(file_path, folder_path_obj)
 
                 try:
@@ -363,7 +366,11 @@ async def process_documents_with_rag_batch(
 
         # Step 2: Process with RAG
         # Initialize RAG system
-        await self._ensure_lightrag_initialized()
+        init_result = await self._ensure_lightrag_initialized()
+        if not init_result or not init_result.get("success"):
+            raise RuntimeError(
+                f"LightRAG initialization failed: {(init_result or {}).get('error', 'unknown error')}"
+            )
 
         # Then, process each successful file with RAG
         rag_results = {}
diff --git a/raganything/config.py b/raganything/config.py
@@ -59,11 +59,14 @@ class RAGAnythingConfig:
     """Maximum number of files to process concurrently."""
 
     supported_file_extensions: List[str] = field(
-        default_factory=lambda: get_env_value(
-            "SUPPORTED_FILE_EXTENSIONS",
-            ".pdf,.jpg,.jpeg,.png,.bmp,.tiff,.tif,.gif,.webp,.doc,.docx,.ppt,.pptx,.xls,.xlsx,.txt,.md",
-            str,
-        ).split(",")
+        default_factory=lambda: [
+            x.strip()
+            for x in get_env_value(
+                "SUPPORTED_FILE_EXTENSIONS",
+                ".pdf,.jpg,.jpeg,.png,.bmp,.tiff,.tif,.gif,.webp,.doc,.docx,.ppt,.pptx,.xls,.xlsx,.txt,.md",
+                str,
+            ).split(",")
+        ]
     )
     """List of supported file extensions for batch processing."""
 
@@ -94,9 +97,12 @@ class RAGAnythingConfig:
     """Whether to include image/table captions in context."""
 
     context_filter_content_types: List[str] = field(
-        default_factory=lambda: get_env_value(
-            "CONTEXT_FILTER_CONTENT_TYPES", "text", str
-        ).split(",")
+        default_factory=lambda: [
+            x.strip()
+            for x in get_env_value("CONTEXT_FILTER_CONTENT_TYPES", "text", str).split(
+                ","
+            )
+        ]
     )
     """Content types to include in context extraction (e.g., 'text', 'image', 'table')."""
 
diff --git a/raganything/parser.py b/raganything/parser.py
@@ -25,6 +25,7 @@
 
 
 import os
+import platform
 import hashlib
 import json
 import argparse
@@ -45,10 +46,9 @@
     Tuple,
     Any,
     Iterator,
-    TypeVar,
 )
 
-T = TypeVar("T")
+_IS_WINDOWS: bool = platform.system() == "Windows"
 
 
 class MineruExecutionError(Exception):
@@ -228,8 +228,6 @@ def convert_office_to_pdf(
                 )
 
                 # Prepare subprocess parameters to hide console window on Windows
-                import platform
-
                 # Try LibreOffice commands in order of preference
                 commands_to_try = ["libreoffice", "soffice"]
 
@@ -258,7 +256,7 @@ def convert_office_to_pdf(
                         }
 
                         # Hide console window on Windows
-                        if platform.system() == "Windows":
+                        if _IS_WINDOWS:
                             convert_subprocess_kwargs["creationflags"] = (
                                 subprocess.CREATE_NO_WINDOW
                             )
@@ -801,7 +799,6 @@ def _run_mineru_command(
 
         try:
             # Prepare subprocess parameters to hide console window on Windows
-            import platform
             import threading
             from queue import Queue, Empty
 
@@ -824,7 +821,7 @@ def _run_mineru_command(
             }
 
             # Hide console window on Windows
-            if platform.system() == "Windows":
+            if _IS_WINDOWS:
                 subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
 
             # Function to read output from subprocess and add to queue
@@ -1427,8 +1424,6 @@ def check_installation(self) -> bool:
         """
         try:
             # Prepare subprocess parameters to hide console window on Windows
-            import platform
-
             subprocess_kwargs = {
                 "capture_output": True,
                 "text": True,
@@ -1438,7 +1433,7 @@ def check_installation(self) -> bool:
             }
 
             # Hide console window on Windows
-            if platform.system() == "Windows":
+            if _IS_WINDOWS:
                 subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
 
             result = subprocess.run(["mineru", "--version"], **subprocess_kwargs)
@@ -1630,8 +1625,6 @@ def _run_docling_command(
 
         try:
             # Prepare subprocess parameters to hide console window on Windows
-            import platform
-
             env = None
             if custom_env:
                 env = os.environ.copy()
@@ -1647,7 +1640,7 @@ def _run_docling_command(
             }
 
             # Hide console window on Windows
-            if platform.system() == "Windows":
+            if _IS_WINDOWS:
                 docling_subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
 
             result = subprocess.run(cmd, **docling_subprocess_kwargs)
@@ -1737,9 +1730,20 @@ def read_from_block_recursive(
             for member in members:
                 cnt += 1
                 member_tag = member["$ref"]
-                member_type = member_tag.split("/")[1]
-                member_num = member_tag.split("/")[2]
-                member_block = docling_content[member_type][int(member_num)]
+                # JSON References follow the form "#/<type>/<index>" (e.g. "#/body/0")
+                ref_parts = member_tag.split("/")
+                if len(ref_parts) < 3:
+                    self.logger.warning(
+                        f"Unexpected $ref format (expected #/<type>/<index>): {member_tag!r}"
+                    )
+                    continue
+                member_type = ref_parts[1]
+                member_num = ref_parts[2]
+                try:
+                    member_block = docling_content[member_type][int(member_num)]
+                except (KeyError, ValueError, IndexError) as e:
+                    self.logger.warning(f"Could not resolve $ref {member_tag!r}: {e}")
+                    continue
                 content_list.extend(
                     self.read_from_block_recursive(
                         member_block,
@@ -1773,7 +1777,10 @@ def read_from_block(
         elif type == "pictures":
             try:
                 base64_uri = block["image"]["uri"]
-                base64_str = base64_uri.split(",")[1]
+                # base64 data URIs have the form "data:<mime>;base64,<data>"
+                # but some exporters may omit the prefix
+                parts = base64_uri.split(",", 1)
+                base64_str = parts[1] if len(parts) == 2 else parts[0]
                 # Create images directory within the docling subdirectory
                 image_dir = output_dir / "images"
                 image_dir.mkdir(parents=True, exist_ok=True)  # Ensure directory exists
@@ -1939,8 +1946,6 @@ def check_installation(self) -> bool:
         """
         try:
             # Prepare subprocess parameters to hide console window on Windows
-            import platform
-
             subprocess_kwargs = {
                 "capture_output": True,
                 "text": True,
@@ -1950,7 +1955,7 @@ def check_installation(self) -> bool:
             }
 
             # Hide console window on Windows
-            if platform.system() == "Windows":
+            if _IS_WINDOWS:
                 subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
 
             result = subprocess.run(["docling", "--version"], **subprocess_kwargs)
diff --git a/raganything/processor.py b/raganything/processor.py
@@ -531,6 +531,14 @@ async def _process_multimodal_content(
                 doc_id=doc_id,
             )
 
+        # Ensure LightRAG is initialized before accessing its storages
+        init_result = await self._ensure_lightrag_initialized()
+        if not init_result or not init_result.get("success"):
+            self.logger.error(
+                "LightRAG initialization failed; skipping multimodal processing"
+            )
+            return
+
         # Check multimodal processing status - handle LightRAG's early DocStatus.PROCESSED marking
         try:
             existing_doc_status = await self.lightrag.doc_status.get_by_id(doc_id)
@@ -573,9 +581,6 @@ async def _process_multimodal_content(
                 pipeline_status["history_messages"].append(log_message)
 
         try:
-            # Ensure LightRAG is initialized
-            await self._ensure_lightrag_initialized()
-
             await self._process_multimodal_content_batch_type_aware(
                 multimodal_items=multimodal_items, file_path=file_path, doc_id=doc_id
             )
@@ -1540,7 +1545,11 @@ async def process_document_complete(
 
         try:
             # Ensure LightRAG is initialized
-            await self._ensure_lightrag_initialized()
+            init_result = await self._ensure_lightrag_initialized()
+            if not init_result or not init_result.get("success"):
+                raise RuntimeError(
+                    f"LightRAG initialization failed: {(init_result or {}).get('error', 'unknown error')}"
+                )
 
             # Use config defaults if not provided
             if output_dir is None:
@@ -1675,25 +1684,61 @@ async def process_document_complete_lightrag_api(
         doc_pre_id = f"doc-pre-{file_name}"
         pipeline_status = None
         pipeline_status_lock = None
+        current_doc_status = {}  # initialised here so the except block can always unpack it
+
+        async def mark_initialization_failed(error_msg: str) -> None:
+            """Persist init failures when LightRAG doc_status is already available."""
+            lightrag = getattr(self, "lightrag", None)
+            doc_status = getattr(lightrag, "doc_status", None)
+            if doc_status is None:
+                self.logger.error(
+                    "LightRAG initialization failed before doc_status was available; "
+                    f"unable to persist failed status for {file_path}"
+                )
+                return
+
+            try:
+                existing_status = await doc_status.get_by_id(doc_pre_id)
+                failed_status = {
+                    "status": DocStatus.FAILED,
+                    "content": "",
+                    "error_msg": error_msg,
+                    "content_summary": "",
+                    "multimodal_content": [],
+                    "scheme_name": scheme_name,
+                    "content_length": 0,
+                    "created_at": "",
+                    "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
+                    "file_path": file_name,
+                }
+                if existing_status:
+                    failed_status = {
+                        **existing_status,
+                        "status": DocStatus.FAILED,
+                        "error_msg": error_msg,
+                        "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
+                    }
+                await doc_status.upsert({doc_pre_id: failed_status})
+                await doc_status.index_done_callback()
+            except Exception as status_error:
+                self.logger.error(
+                    f"Failed to persist initialization failure status for {file_path}: "
+                    f"{status_error}"
+                )
 
         if parser:
             self.config.parser = parser
 
-        current_doc_status = await self.lightrag.doc_status.get_by_id(doc_pre_id)
-
         try:
-            # Ensure LightRAG is initialized
+            # Ensure LightRAG is initialized before accessing its storages
             result = await self._ensure_lightrag_initialized()
-            if not result["success"]:
-                await self.lightrag.doc_status.upsert(
-                    {
-                        doc_pre_id: {
-                            **current_doc_status,
-                            "status": DocStatus.FAILED,
-                            "error_msg": result["error"],
-                        }
-                    }
+            if not result or not result.get("success"):
+                error_msg = (result or {}).get("error", "unknown error")
+                self.logger.error(
+                    f"LightRAG initialization failed: {error_msg}; "
+                    f"skipping document processing for {file_path}"
                 )
+                await mark_initialization_failed(str(error_msg))
                 return False
 
             # Use config defaults if not provided
@@ -1761,9 +1806,10 @@ async def process_document_complete_lightrag_api(
                     file_path, output_dir, parse_method, display_stats, **kwargs
                 )
             except MineruExecutionError as e:
-                error_message = e.error_msg
                 if isinstance(e.error_msg, list):
-                    error_message = "\n".join(e.error_msg)
+                    error_message = "\n".join(str(m) for m in e.error_msg)
+                else:
+                    error_message = str(e.error_msg)
                 await self.lightrag.doc_status.upsert(
                     {
                         doc_pre_id: {
@@ -1859,15 +1905,23 @@ async def process_document_complete_lightrag_api(
             return False
 
         finally:
-            async with pipeline_status_lock:
-                pipeline_status.update({"scan_disabled": False})
-                pipeline_status["latest_message"] = (
-                    f"RAGAnything processing completed for {file_name}"
-                )
-                pipeline_status["history_messages"].append(
-                    f"RAGAnything processing completed for {file_name}"
-                )
-                pipeline_status["history_messages"].append("Now is allowed to scan")
+            if pipeline_status_lock is not None and pipeline_status is not None:
+                try:
+                    async with pipeline_status_lock:
+                        pipeline_status.update({"scan_disabled": False})
+                        pipeline_status["latest_message"] = (
+                            f"RAGAnything processing completed for {file_name}"
+                        )
+                        pipeline_status["history_messages"].append(
+                            f"RAGAnything processing completed for {file_name}"
+                        )
+                        pipeline_status["history_messages"].append(
+                            "Now is allowed to scan"
+                        )
+                except Exception as _finally_err:
+                    self.logger.error(
+                        f"Failed to update pipeline status in finally block: {_finally_err}"
+                    )
 
     async def insert_content_list(
         self,
@@ -1907,7 +1961,11 @@ async def insert_content_list(
         doc_start_time = time.time()
 
         # Ensure LightRAG is initialized
-        await self._ensure_lightrag_initialized()
+        init_result = await self._ensure_lightrag_initialized()
+        if not init_result or not init_result.get("success"):
+            raise RuntimeError(
+                f"LightRAG initialization failed: {(init_result or {}).get('error', 'unknown error')}"
+            )
 
         # Use config defaults if not provided
         if display_stats is None:
diff --git a/raganything/query.py b/raganything/query.py
diff --git a/tests/test_processor_lightrag_api.py b/tests/test_processor_lightrag_api.py
