fix: extract zip documents without external command

owl/utils/document_toolkit.py

@@ -26,12 +26,12 @@
 from typing import List, Optional, Tuple, Literal
 from urllib.parse import urlparse
 import os
-import subprocess
 import xmltodict
 import nest_asyncio
 import traceback
 import asyncio
 from crawl4ai import AsyncWebCrawler
+from owl.utils.zip_utils import extract_zip
 
 nest_asyncio.apply()
 
@@ -299,17 +299,7 @@ def _unzip_file(self, zip_path: str) -> List[str]:
         extract_path = os.path.join(self.cache_dir, zip_name)
         os.makedirs(extract_path, exist_ok=True)
 
-        try:
-            subprocess.run(["unzip", "-o", zip_path, "-d", extract_path], check=True)
-        except subprocess.CalledProcessError as e:
-            raise RuntimeError(f"Failed to unzip file: {e}")
-
-        extracted_files = []
-        for root, _, files in os.walk(extract_path):
-            for file in files:
-                extracted_files.append(os.path.join(root, file))
-
-        return extracted_files
+        return extract_zip(zip_path, extract_path)
 
     def get_tools(self) -> List[FunctionTool]:
         r"""Returns a list of FunctionTool objects representing the functions in the toolkit.

owl/utils/zip_utils.py

@@ -0,0 +1,17 @@
+import os
+import zipfile
+
+
+def extract_zip(zip_path: str, extract_path: str) -> list[str]:
+    """Extract a zip archive with Python's cross-platform standard library."""
+    try:
+        with zipfile.ZipFile(zip_path, "r") as archive:
+            archive.extractall(extract_path)
+    except zipfile.BadZipFile as e:
+        raise RuntimeError(f"Failed to unzip file: {e}") from e
+
+    extracted_files = []
+    for root, _, files in os.walk(extract_path):
+        for file in files:
+            extracted_files.append(os.path.join(root, file))
+    return extracted_files

tests/test_document_toolkit.py

@@ -0,0 +1,25 @@
+import importlib.util
+import zipfile
+from pathlib import Path
+
+
+def _load_extract_zip():
+    module_path = Path(__file__).parents[1] / "owl" / "utils" / "zip_utils.py"
+    spec = importlib.util.spec_from_file_location("zip_utils", module_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.extract_zip
+
+
+def test_extract_zip_uses_python_zip_support(tmp_path):
+    archive_path = tmp_path / "documents.zip"
+    with zipfile.ZipFile(archive_path, "w") as archive:
+        archive.writestr("nested/notes.txt", "hello")
+
+    extract_path = tmp_path / "cache" / "documents"
+    extracted = _load_extract_zip()(str(archive_path), str(extract_path))
+
+    notes_path = extract_path / "nested" / "notes.txt"
+    assert extracted == [str(notes_path)]
+    assert notes_path.read_text() == "hello"