diff --git a/README.md b/README.md index 3d4fb70..e3250a0 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,9 @@ with cortext.Cortext(":memory:") as memory: print(len(emb)) ``` +For binary attachments, pass `Media(data, mimetype)` or raw bytes with +`media_mimetype` to `process_text_with_media`. + Python 3.10+. Binding **1.2.4** tracks engine **v1.2.4**. ## Why not pypi.org? diff --git a/cortext/__init__.py b/cortext/__init__.py index 58c7025..f704975 100644 --- a/cortext/__init__.py +++ b/cortext/__init__.py @@ -411,6 +411,30 @@ def _configure_library(lib: ctypes.CDLL) -> ctypes.CDLL: ] lib.cortext_process_audio_json_with_options.restype = ctypes.c_void_p + text_media_json = getattr(lib, "cortext_process_text_with_media_json", None) + text_media_json_with_options = getattr( + lib, "cortext_process_text_with_media_json_with_options", None + ) + lib._cortext_text_media_json_available = text_media_json_with_options is not None + if text_media_json is not None: + text_media_json.argtypes = [ + ctypes.c_void_p, + ctypes.c_char_p, + ctypes.c_char_p, + ctypes.POINTER(_NativeMedia), + ] + text_media_json.restype = ctypes.c_void_p + + if text_media_json_with_options is not None: + text_media_json_with_options.argtypes = [ + ctypes.c_void_p, + ctypes.c_char_p, + ctypes.c_char_p, + ctypes.POINTER(_NativeMedia), + ctypes.POINTER(_NativeProcessJsonOptions), + ] + text_media_json_with_options.restype = ctypes.c_void_p + lib.cortext_process_audio_with_media_json.argtypes = [ ctypes.c_void_p, ctypes.POINTER(ctypes.c_float), @@ -1134,6 +1158,52 @@ def process_text( ) ) + def process_text_with_media_json( + self, + text: str, + source_id: str, + media: Media | bytes | bytearray | memoryview | None = None, + media_mimetype: str | None = None, + include_embedding: bool = True, + retention: Retention | int | None = None, + ) -> str: + if not getattr(self._lib, "_cortext_text_media_json_available", False): + raise RuntimeError( + "loaded native library does not support text media processing" + ) + native_media, media_buffer, mimetype_buffer = _native_media( + media, media_mimetype + ) + options = _native_process_json_options(include_embedding, retention) + _ = (media_buffer, mimetype_buffer) + return self._call_json( + self._lib.cortext_process_text_with_media_json_with_options, + text.encode("utf-8"), + source_id.encode("utf-8"), + ctypes.byref(native_media), + ctypes.byref(options), + ) + + def process_text_with_media( + self, + text: str, + source_id: str, + media: Media | bytes | bytearray | memoryview | None = None, + media_mimetype: str | None = None, + include_embedding: bool = True, + retention: Retention | int | None = None, + ) -> dict[str, Any]: + return json.loads( + self.process_text_with_media_json( + text, + source_id, + media, + media_mimetype, + include_embedding, + retention, + ) + ) + def embed_text_json(self, text: str) -> str: return self._call_json( self._lib.cortext_embed_text_json, diff --git a/tests/test_cortext.py b/tests/test_cortext.py index 20b55ba..3f0e616 100644 --- a/tests/test_cortext.py +++ b/tests/test_cortext.py @@ -128,6 +128,69 @@ def test_process_text_and_embed_text(tmp_path): clear_materialize_cache() +def test_text_media_capability_only_requires_options_symbol(): + import cortext + + class FakeFunction: + pass + + class FakeLibrary: + def __init__(self): + self.functions = {} + + def __getattr__(self, name): + if name == "cortext_process_text_with_media_json": + raise AttributeError(name) + function = self.functions.setdefault(name, FakeFunction()) + return function + + lib = cortext._configure_library(FakeLibrary()) + + assert lib._cortext_text_media_json_available is True + assert "cortext_process_text_with_media_json_with_options" in lib.functions + + +def test_process_text_with_media_forwards_binary_bytes_and_mimetype(): + import ctypes + + import cortext + + native_fn = object() + fake_lib = type("FakeLib", (), { + "cortext_process_text_with_media_json_with_options": native_fn, + "_cortext_text_media_json_available": True, + })() + engine = object.__new__(cortext.Cortext) + engine._lib = fake_lib + engine._handle = None + captured = {} + + def call_json(fn, *args): + captured["fn"] = fn + captured["text"] = args[0] + captured["source_id"] = args[1] + media = ctypes.cast( + args[2], ctypes.POINTER(cortext._NativeMedia) + ).contents + captured["media"] = ctypes.string_at(media.data, media.size) + captured["mimetype"] = ctypes.string_at(media.mimetype) + return '{"ok": true}' + + engine._call_json = call_json + payload = b"\x00\xff\x01" + + assert engine.process_text_with_media( + "text", "source", payload, "application/octet-stream" + ) == {"ok": True} + assert captured == { + "fn": native_fn, + "text": b"text", + "source_id": b"source", + "media": payload, + "mimetype": b"application/octet-stream", + } + + def test_embedding_length_stable(tmp_path): import cortext from cortext.model import clear_materialize_cache