OSB tokenizer

.pre-commit-config.yaml

@@ -54,7 +54,7 @@ repos:
             (?x)^(
                 .pre-commit-config.yaml|
                 LICENSE|
-                plugins/python-dspam-plugin-email/README.md|
+                plugins/.*/README.md|
                 README.md|
                 uv.lock
             )$

plugins/python-dspam-plugin-email/README.md

@@ -1,7 +1,20 @@
-# dspam-plugin-email
+# Python DSPAM email plugin
 
 Plugin for python-dspam for handling email messages.
 
+## Setup
+
+```shell
+pip install python-dspam-plugin-email
+```
+
+After installation, enable the plugin by setting env var `DSPAM_PARSER_PLUGIN=email` or by adding the following lines to your `config.toml`:
+
+```toml
+[dspam.parser]
+plugin = "email"
+```
+
 # License
 
-This project is licensed under the BSD-3-Clause License. See the [LICENSE](LICENSE) file for details.
+This project is licensed under the BSD-3-Clause License. See the [LICENSE](../../LICENSE) file for details.

plugins/python-dspam-plugin-osb/README.md

@@ -0,0 +1,52 @@
+# Python DSPAM OSB tokenizer plugin
+
+Plugin for python-dspam providing OSB tokenization.
+
+## Setup
+
+```shell
+pip install python-dspam-plugin-osb
+```
+
+After installation, enable the plugin by setting env var `DSPAM_PARSER_TOKENIZER=chain` or by adding the following lines to your `config.toml`:
+
+```toml
+[dspam.tokenizer]
+plugin = "chain"
+```
+
+## Orthogonal sparse bigrams (OSB)
+
+Using orthogonal sparse bigrams was originally introduced in DSPAM based on an idea from Bill Yerazunis.
+
+The generation of OSB tokens is based on the SBPH algorithm, but produces fewer tokens. Where SBPH generates tokens for all permutations in a sliding window, OSB keeps only permutations containing the latest word (e.g. the last single word token in the sliding window)
+
+An example comparison based on the content `The quick brown fox jumps over the fence` with a sliding window of 5:
+
+| Unigram/single-word tokens | OSB tokens        |
+|----------------------------|-------------------|
+| The                        | The # # # jumps   |
+| quick                      | # quick # # jumps |
+| brown                      | # # brown # jumps |
+| fox                        | # # # fox jumps   |
+| jumps                      | quick # # # over  |
+| over                       | # brown # # over  |
+| the                        | # # fox # over    |
+| fence                      | # # # jumps over  |
+|                            | brown # # # the   |
+|                            | # fox # # the     |
+|                            | # # jumps # the   |
+|                            | # # # over the    |
+|                            | fox # # # fence   |
+|                            | # jumps # # fence |
+|                            | # # over # fence  |
+|                            | # # # the fence   |
+
+
+Detailed explanation available at: https://www.siefkes.net/papers/winnow-spam.pdf
+
+
+
+# License
+
+This project is licensed under the BSD-3-Clause License. See the [LICENSE](../../LICENSE) file for details.

plugins/python-dspam-plugin-osb/pyproject.toml

@@ -0,0 +1,24 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+[build-system]
+requires = [ "setuptools>=41", "wheel", "setuptools-git-versioning>=3.0,<4", ]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools-git-versioning]
+enabled = true
+
+[project]
+name = "python-dspam-plugin-osb"
+description = "A plugin for Python-DSPAM that provides OSB tokenization."
+dynamic = ["version"]
+readme = "README.md"
+authors = [
+    { name = "Tom Hendrikx", email = "tom@whyscream.net" }
+]
+requires-python = ">=3.14"
+dependencies = [
+    "anyio>=4.13.0",
+]
+
+[project.entry-points."dspam.tokenizer"]
+osb = "dspam_plugin_osb.osb:OsbTokenizer"

plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/__init__.py

@@ -0,0 +1 @@
+# SPDX-License-Identifier: BSD-3-Clause

plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py

@@ -0,0 +1,113 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+import logging
+from collections.abc import Iterator
+from functools import partial
+
+from dspam.settings import TokenizerSettings
+from dspam.tokenize import (
+    METADATA_IGNORE_DELIMITERS,
+    Tokenizer,
+    get_homoglyph_delimiters,
+    tokenize_metadata,
+    word_tokenize_content,
+)
+from dspam.types import Metadata, Token, TokenList
+
+logger = logging.getLogger(__name__)
+
+
+class OsbTokenizer(Tokenizer):
+    API_VERSION = "1.0"
+
+    SPARSE_WINDOW_SIZE = 5
+    """The size of the sliding window for generating OSB tokens."""
+
+    TOKEN_SEPARATOR: str = "+"  # noqa: S105
+    """The separator to use between terms in OSB tokens."""
+
+    OSB_TOKEN_PLACEHOLDER = "#"  # noqa: S105
+    """The placeholder for skipped terms in an OSB token"""
+
+    def __init__(self, settings: TokenizerSettings) -> None:
+        super().__init__(settings)
+
+    async def __call__(self, content: str, metadata: Metadata) -> TokenList:
+        """Tokenize content and metadata into OSB tokens."""
+        metadata_tokens = self.osb_tokenize_metadata(metadata)
+        content_tokens = self.osb_tokenize_content(content)
+        return metadata_tokens + content_tokens
+
+    def osb_tokenize_content(self, content: str, ignore_delimiters: str = "") -> TokenList:
+        """
+        Tokenize a content string into OSB tokens.
+
+        First, use the `WordTokenizer` to extract word tokens from the content.
+        Then apply the OSB sliding window algorithm and generate OSB tokens for each window.
+        """
+        word_tokens = self.get_word_tokens(content, ignore_delimiters)
+        osb_tokens = []
+        # Prepopulate the token lists
+        window_size = min(len(word_tokens), self.SPARSE_WINDOW_SIZE)
+        window_terms = word_tokens[: window_size - 1]
+        word_tokens = word_tokens[window_size - 1 :]
+
+        while True:
+            # Remove the first window token if needed
+            if len(window_terms) >= self.SPARSE_WINDOW_SIZE:
+                removed = window_terms.pop(0)
+                logger.debug(f"Dropped window term {removed}")
+            # Append window token if available
+            if len(window_terms) < self.SPARSE_WINDOW_SIZE:
+                try:
+                    window_token = word_tokens.pop(0)
+                    logger.debug(f"Appended window term {window_token}")
+                    window_terms.append(window_token)
+                except IndexError:
+                    logger.debug("No more terms to append, processing complete")
+                    break
+
+                for token in self.get_osb_tokens(window_terms):
+                    osb_tokens.append(token)
+
+        return osb_tokens
+
+    def get_osb_tokens(self, word_tokens: TokenList) -> Iterator[Token]:
+        """
+        Generate OSB tokens from a list of word tokens.
+
+        The list of word tokens is considered as the non-sliding window.
+        """
+        window_size = len(word_tokens)
+        for idx in range(1, window_size, 1):
+            # Generate a single OSB token
+            osb_token_terms = []
+            for term_idx, term in enumerate(word_tokens, 1):
+                if term_idx == idx:
+                    # Append the first term
+                    osb_token_terms.append(term)
+                    logger.debug(f"Appending start term {term}")
+                elif osb_token_terms and term_idx == window_size:
+                    # Append last term only when we already have other terms
+                    logger.debug(f"Appending end term {term}")
+                    osb_token_terms.append(term)
+                elif osb_token_terms:
+                    # Append placeholders only when we already have other terms
+                    logger.debug(f"Appending placeholder for term {term}")
+                    osb_token_terms.append(self.OSB_TOKEN_PLACEHOLDER)
+
+            osb_token = self.TOKEN_SEPARATOR.join(osb_token_terms)
+            logger.debug(f"Generated OSB token: {osb_token}")
+            yield osb_token
+
+    def osb_tokenize_metadata(self, metadata: Metadata) -> TokenList:
+        """Tokenize metadata into OSB tokens."""
+        tokenizer = partial(self.osb_tokenize_content, ignore_delimiters=METADATA_IGNORE_DELIMITERS)
+        return list(tokenize_metadata(metadata, tokenizer))
+
+    def get_word_tokens(self, content: str, ignore_delimiters: str = "") -> list[str]:
+        """Tokenize content into word tokens."""
+        delimiters = "".join([d for d in self.settings.delimiters if d not in ignore_delimiters])
+        delimiters += get_homoglyph_delimiters(delimiters)
+
+        return word_tokenize_content(content, delimiters)

plugins/python-dspam-plugin-osb/tests/test_osb.py

@@ -0,0 +1,47 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+import pytest
+from dspam_plugin_osb.osb import OsbTokenizer
+
+from dspam.settings import TokenizerSettings
+
+
+@pytest.fixture
+def content():
+    return "The quick brown fox jumps over the lazy dog"
+
+
+@pytest.fixture
+def tokenizer():
+    return OsbTokenizer(TokenizerSettings())
+
+
+def test_osb_tokenize_content_success(tokenizer, content):
+    tokens = tokenizer.osb_tokenize_content(content)
+    assert "The+#+#+#+jumps" in tokens
+    assert "lazy+dog" in tokens
+    assert len(tokens) == 5 * 4  # 5 different windows, each window yields 4 tokens
+
+
+@pytest.mark.parametrize(
+    "content_length, num_osb_tokens",
+    [
+        (5, 4),
+        (4, 3),
+        (3, 2),
+        (2, 1),
+        (1, 0),
+        (0, 0),
+    ],
+)
+def test_osb_tokenize_content_token_count(content, content_length, tokenizer, num_osb_tokens):
+    content = " ".join(content.split()[:content_length])
+
+    tokens = tokenizer.osb_tokenize_content(content)
+    assert len(tokens) == num_osb_tokens, f"Expected {num_osb_tokens} OSB tokens, result was: {tokens}"
+
+
+def test_osb_tokenize_metadata_success(tokenizer):
+    metadata = {"foo": "bar qux flerp"}
+    tokens = tokenizer.osb_tokenize_metadata(metadata)
+    assert tokens == ["foo*bar+#+flerp", "foo*qux+flerp"]

src/dspam/__init__.py

@@ -1,10 +1,9 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
-import importlib.metadata
 
-__version__ = importlib.metadata.version("python-dspam")
+"""Some constants used throughout the application"""
 
-# Some constants used throughout the application
+# TODO: change these into an Enum class or whatnot
 IS_INNOCENT: str = "innocent"
 IS_SPAM: str = "spam"
 IS_UNKNOWN: str = "unknown"

src/dspam/plugins.py

@@ -30,6 +30,24 @@
 logger = logging.getLogger(__name__)
 
 
+def get_package_version(package_name: str) -> str:
+    """Find the version for a top-level package name."""
+    version = "???"
+    try:
+        return importlib.metadata.version(package_name)
+    except importlib.metadata.PackageNotFoundError:
+        pass
+
+    dist_names = importlib.metadata.packages_distributions().get(package_name, [])
+    for dist_name in dist_names:
+        try:
+            return importlib.metadata.version(dist_name)
+        except importlib.metadata.PackageNotFoundError:
+            continue
+
+    return version
+
+
 @dataclass
 class PluginInfo:
     name: str
@@ -104,11 +122,7 @@ def list_plugins(self) -> Generator[PluginInfo]:
             for plugin, plugin_class in plugins.items():
                 module_name = plugin_class.__module__
                 package_name = module_name.split(".")[0]
-                try:
-                    version = importlib.metadata.version(package_name)
-                except importlib.metadata.PackageNotFoundError:
-                    module = importlib.import_module(package_name)
-                    version = module.__version__
+                version = get_package_version(package_name)
                 api_version = getattr(plugin_class, "API_VERSION", "0.0")
 
                 yield PluginInfo(