From 08829268dc18983580b14c7cb174a366046ba5d4 Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Fri, 3 Jul 2026 02:35:05 +0200 Subject: [PATCH 1/6] Refactor the version detection, no more workarounds needed --- src/dspam/__init__.py | 5 ++--- src/dspam/plugins.py | 24 +++++++++++++++++++----- 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/src/dspam/__init__.py b/src/dspam/__init__.py index 72c8ff8..f217d80 100644 --- a/src/dspam/__init__.py +++ b/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" diff --git a/src/dspam/plugins.py b/src/dspam/plugins.py index 86abed2..67451b3 100644 --- a/src/dspam/plugins.py +++ b/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( From bbebb7d2d9e44f6317d99a9ac2e87b508fa7636a Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Fri, 3 Jul 2026 02:46:02 +0200 Subject: [PATCH 2/6] OSB tokenizer, content tokens only for now --- .pre-commit-config.yaml | 2 +- plugins/python-dspam-plugin-email/README.md | 17 +++- plugins/python-dspam-plugin-osb/README.md | 52 +++++++++++ .../python-dspam-plugin-osb/pyproject.toml | 24 +++++ .../src/dspam_plugin_osb/__init__.py | 1 + .../src/dspam_plugin_osb/osb.py | 90 +++++++++++++++++++ .../python-dspam-plugin-osb/tests/test_osb.py | 41 +++++++++ uv.lock | 11 +++ 8 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 plugins/python-dspam-plugin-osb/README.md create mode 100644 plugins/python-dspam-plugin-osb/pyproject.toml create mode 100644 plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/__init__.py create mode 100644 plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py create mode 100644 plugins/python-dspam-plugin-osb/tests/test_osb.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index fe8579d..520448d 100644 --- a/.pre-commit-config.yaml +++ b/.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 )$ diff --git a/plugins/python-dspam-plugin-email/README.md b/plugins/python-dspam-plugin-email/README.md index 712eb3f..8bbd9b8 100644 --- a/plugins/python-dspam-plugin-email/README.md +++ b/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. diff --git a/plugins/python-dspam-plugin-osb/README.md b/plugins/python-dspam-plugin-osb/README.md new file mode 100644 index 0000000..aec57f8 --- /dev/null +++ b/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. diff --git a/plugins/python-dspam-plugin-osb/pyproject.toml b/plugins/python-dspam-plugin-osb/pyproject.toml new file mode 100644 index 0000000..c3a17aa --- /dev/null +++ b/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" diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/__init__.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/__init__.py new file mode 100644 index 0000000..dde388e --- /dev/null +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/__init__.py @@ -0,0 +1 @@ +# SPDX-License-Identifier: BSD-3-Clause diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py new file mode 100644 index 0000000..f37706d --- /dev/null +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py @@ -0,0 +1,90 @@ +# SPDX-License-Identifier: BSD-3-Clause + +import logging + +from dspam.settings import TokenizerSettings +from dspam.tokenize import Tokenizer, WordTokenizer +from dspam.types import Metadata, 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""" + + word_tokenizer: WordTokenizer + + def __init__(self, settings: TokenizerSettings) -> None: + super().__init__(settings) + + self.word_tokenizer = WordTokenizer(self.settings) + + async def __call__(self, content: str, metadata: Metadata) -> TokenList: + """Tokenize content and metadata into OSB tokens.""" + content_tokens = self.osb_tokenize_content(content) + return content_tokens + + def osb_tokenize_content(self, content: 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) + 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: + window_token = word_tokens.pop(0) if word_tokens else None + if window_token: + logger.debug(f"Appended window term {window_token}") + window_terms.append(window_token) + else: + logger.debug("No more terms to append, processing complete") + break + + for idx in range(1, window_size, 1): + # Generate a single OSB token + osb_token_terms = [] + for term_idx, term in enumerate(window_terms, 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}") + osb_tokens.append(osb_token) + + return osb_tokens + + def get_word_tokens(self, content: str) -> list[str]: + """Tokenize content into word tokens.""" + return self.word_tokenizer.tokenize_content(content) diff --git a/plugins/python-dspam-plugin-osb/tests/test_osb.py b/plugins/python-dspam-plugin-osb/tests/test_osb.py new file mode 100644 index 0000000..bfde24f --- /dev/null +++ b/plugins/python-dspam-plugin-osb/tests/test_osb.py @@ -0,0 +1,41 @@ +# 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}" diff --git a/uv.lock b/uv.lock index 080a4df..d910607 100644 --- a/uv.lock +++ b/uv.lock @@ -10,6 +10,7 @@ resolution-markers = [ members = [ "python-dspam", "python-dspam-plugin-email", + "python-dspam-plugin-osb", ] [[package]] @@ -761,6 +762,16 @@ requires-dist = [ { name = "pydantic-settings", git = "https://github.com/pydantic/pydantic-settings.git?rev=9941cae4c633089b7bdb682515c423d0aa80b46c" }, ] +[[package]] +name = "python-dspam-plugin-osb" +source = { editable = "plugins/python-dspam-plugin-osb" } +dependencies = [ + { name = "anyio" }, +] + +[package.metadata] +requires-dist = [{ name = "anyio", specifier = ">=4.13.0" }] + [[package]] name = "pyyaml" version = "6.0.3" From 69a24463d7eb8e9648fa592594f940a1da36906d Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Mon, 6 Jul 2026 23:03:30 +0200 Subject: [PATCH 3/6] OSB tokenization of metadata --- .../src/dspam_plugin_osb/osb.py | 27 +++++++++++++++---- .../python-dspam-plugin-osb/tests/test_osb.py | 6 +++++ src/dspam/tokenize.py | 21 ++++++++------- 3 files changed, 39 insertions(+), 15 deletions(-) diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py index f37706d..dffe647 100644 --- a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py @@ -30,17 +30,18 @@ def __init__(self, settings: TokenizerSettings) -> None: 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 content_tokens + return metadata_tokens + content_tokens - def osb_tokenize_content(self, content: str) -> TokenList: + 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) + 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) @@ -85,6 +86,22 @@ def osb_tokenize_content(self, content: str) -> TokenList: return osb_tokens - def get_word_tokens(self, content: str) -> list[str]: + def osb_tokenize_metadata(self, metadata: Metadata) -> TokenList: + """Tokenize metadata into OSB tokens.""" + metadata_tokens = [] + for key, value in metadata.items(): + if isinstance(value, str): + value = [value] + + for item in value: + value_tokens = self.osb_tokenize_content( + item, ignore_delimiters=self.word_tokenizer.METADATA_IGNORE_DELIMITERS + ) + for value_token in value_tokens: + metadata_tokens.append(f"{key}{self.word_tokenizer.METADATA_TOKEN_SEPARATOR}{value_token}") + + return metadata_tokens + + def get_word_tokens(self, content: str, ignore_delimiters: str = "") -> list[str]: """Tokenize content into word tokens.""" - return self.word_tokenizer.tokenize_content(content) + return self.word_tokenizer.tokenize_content(content, ignore_delimiters) diff --git a/plugins/python-dspam-plugin-osb/tests/test_osb.py b/plugins/python-dspam-plugin-osb/tests/test_osb.py index bfde24f..264c1f4 100644 --- a/plugins/python-dspam-plugin-osb/tests/test_osb.py +++ b/plugins/python-dspam-plugin-osb/tests/test_osb.py @@ -39,3 +39,9 @@ def test_osb_tokenize_content_token_count(content, content_length, tokenizer, nu 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"] diff --git a/src/dspam/tokenize.py b/src/dspam/tokenize.py index 4f9a78c..b679a68 100644 --- a/src/dspam/tokenize.py +++ b/src/dspam/tokenize.py @@ -64,28 +64,29 @@ def tokenize_content(self, content: str, ignore_delimiters: str = "") -> TokenLi def tokenize_metadata(self, metadata: Metadata) -> TokenList: """ - Generate a list of word tokens from the metadata dictionary. + Generate a list of tokens from the metadata dictionary. The metadata key is combined with word tokens from the metadata values to add key-value context in the result. Word tokenization for metadata ignores a few delimiters to improve handling of email headers, so domain names and ip addresses are preserved. + + Args: + metadata (Metadata): the metadata to tokenize. + Returns: + TokenList: a list of tokens generated from the metadata dictionary. """ metadata_tokens = [] for key, value in metadata.items(): if isinstance(value, str): - value_tokens = self.tokenize_content(value, self.METADATA_IGNORE_DELIMITERS) - metadata_tokens.extend(self.make_metadata_token(key, value_tokens)) + value = [value] - elif isinstance(value, list): - for item in value: - value_tokens = self.tokenize_content(item, self.METADATA_IGNORE_DELIMITERS) - metadata_tokens.extend(self.make_metadata_token(key, value_tokens)) + for item in value: + value_tokens = self.tokenize_content(item, ignore_delimiters=self.METADATA_IGNORE_DELIMITERS) + for value_token in value_tokens: + metadata_tokens.append(f"{key}{self.METADATA_TOKEN_SEPARATOR}{value_token}") return metadata_tokens - def make_metadata_token(self, key: str, value_tokens: TokenList) -> TokenList: - return [f"{key}{self.METADATA_TOKEN_SEPARATOR}{token}" for token in value_tokens] - @lru_cache(maxsize=128) def get_homoglyph_delimiters(self, ascii_delimiters: str) -> str: """ From b44dd1fd1ba0d84b06dba8e8561a590dd210095b Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Mon, 6 Jul 2026 23:22:11 +0200 Subject: [PATCH 4/6] Extract metadata unrolling into standalone method --- .../src/dspam_plugin_osb/osb.py | 18 ++---- src/dspam/tokenize.py | 56 ++++++++++++------- 2 files changed, 39 insertions(+), 35 deletions(-) diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py index dffe647..d07d84f 100644 --- a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py @@ -1,9 +1,10 @@ # SPDX-License-Identifier: BSD-3-Clause import logging +from functools import partial from dspam.settings import TokenizerSettings -from dspam.tokenize import Tokenizer, WordTokenizer +from dspam.tokenize import METADATA_IGNORE_DELIMITERS, Tokenizer, WordTokenizer, tokenize_metadata from dspam.types import Metadata, TokenList logger = logging.getLogger(__name__) @@ -88,19 +89,8 @@ def osb_tokenize_content(self, content: str, ignore_delimiters: str = "") -> Tok def osb_tokenize_metadata(self, metadata: Metadata) -> TokenList: """Tokenize metadata into OSB tokens.""" - metadata_tokens = [] - for key, value in metadata.items(): - if isinstance(value, str): - value = [value] - - for item in value: - value_tokens = self.osb_tokenize_content( - item, ignore_delimiters=self.word_tokenizer.METADATA_IGNORE_DELIMITERS - ) - for value_token in value_tokens: - metadata_tokens.append(f"{key}{self.word_tokenizer.METADATA_TOKEN_SEPARATOR}{value_token}") - - return metadata_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.""" diff --git a/src/dspam/tokenize.py b/src/dspam/tokenize.py index b679a68..f1c7734 100644 --- a/src/dspam/tokenize.py +++ b/src/dspam/tokenize.py @@ -1,6 +1,8 @@ # SPDX-License-Identifier: BSD-3-Clause from abc import ABC, abstractmethod +from collections.abc import Callable, Generator +from functools import partial from typing import ClassVar import homoglyphs_fork as hg # type: ignore[import-untyped] @@ -9,6 +11,12 @@ from dspam.settings import TokenizerSettings from dspam.types import Metadata, TokenList +METADATA_TOKEN_SEPARATOR: str = "*" # noqa: S105 +"""The separator that is used between the metadata key and the tokenized values.""" + +METADATA_IGNORE_DELIMITERS: str = ".:" +"""The delimiters to filter from the default set when tokenizing metadata values.""" + class Tokenizer(ABC): API_VERSION: ClassVar[str] @@ -24,6 +32,31 @@ def __str__(self) -> str: return f"{self.__class__.__name__}(API_VERSION={self.API_VERSION})" # pragma: no cover +def tokenize_metadata(metadata: Metadata, value_tokenizer: Callable[[str], TokenList]) -> Generator[str]: + """ + Walk a metadata dictionary and tokenize it into a list of tokens. + + For each metadata key and value, the value is tokenized using the provided `value_tokenizer` function. + Each token is then combined with the metadata key using the `METADATA_TOKEN_SEPARATOR` to create a metadata token. + + Args: + metadata (Metadata): the metadata to tokenize. + value_tokenizer (Callable[[str], TokenList]): the callable used to tokenize the metadata values. + Returns: + Generator[str]: a list of tokens. + """ + + for key, values in metadata.items(): + if isinstance(values, str): + values = [values] + + for value in values: + value_tokens = value_tokenizer(value) + for value_token in value_tokens: + metadata_token = f"{key}{METADATA_TOKEN_SEPARATOR}{value_token}" + yield metadata_token + + class WordTokenizer(Tokenizer): """ A simple tokenizer that splits the input text into words based on whitespace and punctuation. @@ -33,12 +66,6 @@ class WordTokenizer(Tokenizer): API_VERSION = "1.0" - METADATA_IGNORE_DELIMITERS: str = ".:" - """The delimiters to filter from the default set when tokenizing metadata values.""" - - METADATA_TOKEN_SEPARATOR: str = "*" # noqa: S105 - """The separator that is used when composing key-value tokens from metadata.""" - HOMOGLYPH_IGNORE_DELIMITERS: str = "@" """ The ascii delimiters to filter from the configured delimiters when generating homoglyphs. @@ -66,26 +93,13 @@ def tokenize_metadata(self, metadata: Metadata) -> TokenList: """ Generate a list of tokens from the metadata dictionary. - The metadata key is combined with word tokens from the metadata values to add key-value context in the result. - Word tokenization for metadata ignores a few delimiters to improve handling of email headers, so domain names - and ip addresses are preserved. - Args: metadata (Metadata): the metadata to tokenize. Returns: TokenList: a list of tokens generated from the metadata dictionary. """ - metadata_tokens = [] - for key, value in metadata.items(): - if isinstance(value, str): - value = [value] - - for item in value: - value_tokens = self.tokenize_content(item, ignore_delimiters=self.METADATA_IGNORE_DELIMITERS) - for value_token in value_tokens: - metadata_tokens.append(f"{key}{self.METADATA_TOKEN_SEPARATOR}{value_token}") - - return metadata_tokens + content_tokenizer = partial(self.tokenize_content, ignore_delimiters=METADATA_IGNORE_DELIMITERS) + return list(tokenize_metadata(metadata, content_tokenizer)) @lru_cache(maxsize=128) def get_homoglyph_delimiters(self, ascii_delimiters: str) -> str: From 61d1fe8baa1d34d00f924d6cd6084ade8321377e Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Mon, 6 Jul 2026 23:48:23 +0200 Subject: [PATCH 5/6] Extract word tokenization and delimiter setup into standalone methods --- .../src/dspam_plugin_osb/osb.py | 23 +++-- src/dspam/tokenize.py | 89 +++++++++++-------- tests/test_tokenize_word.py | 6 +- 3 files changed, 70 insertions(+), 48 deletions(-) diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py index d07d84f..fecea62 100644 --- a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py @@ -4,7 +4,13 @@ from functools import partial from dspam.settings import TokenizerSettings -from dspam.tokenize import METADATA_IGNORE_DELIMITERS, Tokenizer, WordTokenizer, tokenize_metadata +from dspam.tokenize import ( + METADATA_IGNORE_DELIMITERS, + Tokenizer, + get_homoglyph_delimiters, + tokenize_metadata, + word_tokenize_content, +) from dspam.types import Metadata, TokenList logger = logging.getLogger(__name__) @@ -22,13 +28,9 @@ class OsbTokenizer(Tokenizer): OSB_TOKEN_PLACEHOLDER = "#" # noqa: S105 """The placeholder for skipped terms in an OSB token""" - word_tokenizer: WordTokenizer - def __init__(self, settings: TokenizerSettings) -> None: super().__init__(settings) - self.word_tokenizer = WordTokenizer(self.settings) - async def __call__(self, content: str, metadata: Metadata) -> TokenList: """Tokenize content and metadata into OSB tokens.""" metadata_tokens = self.osb_tokenize_metadata(metadata) @@ -56,11 +58,11 @@ def osb_tokenize_content(self, content: str, ignore_delimiters: str = "") -> Tok logger.debug(f"Dropped window term {removed}") # Append window token if available if len(window_terms) < self.SPARSE_WINDOW_SIZE: - window_token = word_tokens.pop(0) if word_tokens else None - if window_token: + try: + window_token = word_tokens.pop(0) logger.debug(f"Appended window term {window_token}") window_terms.append(window_token) - else: + except IndexError: logger.debug("No more terms to append, processing complete") break @@ -94,4 +96,7 @@ def osb_tokenize_metadata(self, metadata: Metadata) -> TokenList: def get_word_tokens(self, content: str, ignore_delimiters: str = "") -> list[str]: """Tokenize content into word tokens.""" - return self.word_tokenizer.tokenize_content(content, ignore_delimiters) + 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) diff --git a/src/dspam/tokenize.py b/src/dspam/tokenize.py index f1c7734..c93a0aa 100644 --- a/src/dspam/tokenize.py +++ b/src/dspam/tokenize.py @@ -17,6 +17,14 @@ METADATA_IGNORE_DELIMITERS: str = ".:" """The delimiters to filter from the default set when tokenizing metadata values.""" +HOMOGLYPH_IGNORE_DELIMITERS: str = "@" +""" +The ascii delimiters to filter from the configured delimiters when generating homoglyphs. + +Some delimiters generate homoglyphs that are unwanted because they're too common. +E.g. the delimiter '@' triggers the regular 'a' character as a homoglyph. +""" + class Tokenizer(ABC): API_VERSION: ClassVar[str] @@ -32,6 +40,49 @@ def __str__(self) -> str: return f"{self.__class__.__name__}(API_VERSION={self.API_VERSION})" # pragma: no cover +def word_tokenize_content(content: str, delimiters: str) -> TokenList: + """ + Split a content into a list of word token. + + As word tokenization is the base for most other tokenizers, this is provided as a utility + function for many tokenizers to use. + + Args: + content (str): the content to tokenize. + delimiters (str): the delimiters to use for tokenization. + Returns: + TokenList: a list of word tokens. + """ + for char in delimiters: + content = content.replace(char, " ") + return content.split() + + +@lru_cache(maxsize=128) +def get_homoglyph_delimiters(ascii_delimiters: str) -> str: + """ + Generate a set of Unicode delimiter variants based on their ascii variants. + """ + + def decode_escaped(char_: str) -> str: + if char.startswith("\\u"): + return chr(int(char_, 16)) + return char_ + + # Filter out delimiters that we don't want homoglyphs for + ascii_delimiters = "".join([d for d in ascii_delimiters if d not in HOMOGLYPH_IGNORE_DELIMITERS]) + + homoglyphs = hg.Homoglyphs() + + unicode_delimiters = [] + for char in ascii_delimiters: + for homoglyph in homoglyphs.get_combinations(char): + homoglyph = decode_escaped(homoglyph) + unicode_delimiters.append(homoglyph) + + return "".join(set(unicode_delimiters)) + + def tokenize_metadata(metadata: Metadata, value_tokenizer: Callable[[str], TokenList]) -> Generator[str]: """ Walk a metadata dictionary and tokenize it into a list of tokens. @@ -66,14 +117,6 @@ class WordTokenizer(Tokenizer): API_VERSION = "1.0" - HOMOGLYPH_IGNORE_DELIMITERS: str = "@" - """ - The ascii delimiters to filter from the configured delimiters when generating homoglyphs. - - Some delimiters generate homoglyphs that are unwanted because they're too common. - E.g. the delimiter '@' triggers the regular 'a' character as a homoglyph. - """ - async def __call__(self, content: str, metadata: Metadata) -> TokenList: """Tokenize content and metadata into words based on whitespace and punctuation.""" metadata_tokens = self.tokenize_metadata(metadata) @@ -83,11 +126,9 @@ async def __call__(self, content: str, metadata: Metadata) -> TokenList: def tokenize_content(self, content: str, ignore_delimiters: str = "") -> TokenList: """Generate a list of word tokens from the content string.""" delimiters = "".join([d for d in self.settings.delimiters if d not in ignore_delimiters]) - delimiters += self.get_homoglyph_delimiters(delimiters) + delimiters += get_homoglyph_delimiters(delimiters) - for char in delimiters: - content = content.replace(char, " ") - return content.split() + return word_tokenize_content(content, delimiters) def tokenize_metadata(self, metadata: Metadata) -> TokenList: """ @@ -100,27 +141,3 @@ def tokenize_metadata(self, metadata: Metadata) -> TokenList: """ content_tokenizer = partial(self.tokenize_content, ignore_delimiters=METADATA_IGNORE_DELIMITERS) return list(tokenize_metadata(metadata, content_tokenizer)) - - @lru_cache(maxsize=128) - def get_homoglyph_delimiters(self, ascii_delimiters: str) -> str: - """ - Generate a set of Unicode delimiter variants based on their ascii variants. - """ - - def decode_escaped(char: str) -> str: - if char.startswith("\\u"): - return chr(int(char, 16)) - return char - - # Filter out delimiters that we don't want homoglyphs for - ascii_delimiters = "".join([d for d in ascii_delimiters if d not in self.HOMOGLYPH_IGNORE_DELIMITERS]) - - homoglyphs = hg.Homoglyphs() - - unicode_delimiters = [] - for char in ascii_delimiters: - for homoglyph in homoglyphs.get_combinations(char): - homoglyph = decode_escaped(homoglyph) - unicode_delimiters.append(homoglyph) - - return "".join(set(unicode_delimiters)) diff --git a/tests/test_tokenize_word.py b/tests/test_tokenize_word.py index 73cbe6a..46a8f49 100644 --- a/tests/test_tokenize_word.py +++ b/tests/test_tokenize_word.py @@ -3,7 +3,7 @@ import pytest from dspam.settings import TokenizerSettings -from dspam.tokenize import WordTokenizer +from dspam.tokenize import HOMOGLYPH_IGNORE_DELIMITERS, WordTokenizer, get_homoglyph_delimiters @pytest.fixture @@ -141,9 +141,9 @@ async def test_tokenize_unicode_content(): async def test_unicode_delimiters_default(): settings = TokenizerSettings() - homoglyph_delimiters = WordTokenizer(settings).get_homoglyph_delimiters(settings.delimiters) + homoglyph_delimiters = get_homoglyph_delimiters(settings.delimiters) for char in settings.delimiters: - if char in WordTokenizer.HOMOGLYPH_IGNORE_DELIMITERS: + if char in HOMOGLYPH_IGNORE_DELIMITERS: continue assert char in homoglyph_delimiters From e3fff2ac9bd5d64cc09ca48ce80cfa2be4323722 Mon Sep 17 00:00:00 2001 From: Tom Hendrikx Date: Tue, 7 Jul 2026 00:12:19 +0200 Subject: [PATCH 6/6] Simplify OSB content tokenization code --- .../src/dspam_plugin_osb/osb.py | 53 +++++++++++-------- src/dspam/tokenize.py | 6 +-- src/dspam/types.py | 7 ++- 3 files changed, 40 insertions(+), 26 deletions(-) diff --git a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py index fecea62..20adae4 100644 --- a/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py +++ b/plugins/python-dspam-plugin-osb/src/dspam_plugin_osb/osb.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: BSD-3-Clause import logging +from collections.abc import Iterator from functools import partial from dspam.settings import TokenizerSettings @@ -11,7 +12,7 @@ tokenize_metadata, word_tokenize_content, ) -from dspam.types import Metadata, TokenList +from dspam.types import Metadata, Token, TokenList logger = logging.getLogger(__name__) @@ -66,29 +67,39 @@ def osb_tokenize_content(self, content: str, ignore_delimiters: str = "") -> Tok logger.debug("No more terms to append, processing complete") break - for idx in range(1, window_size, 1): - # Generate a single OSB token - osb_token_terms = [] - for term_idx, term in enumerate(window_terms, 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}") - osb_tokens.append(osb_token) + 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) diff --git a/src/dspam/tokenize.py b/src/dspam/tokenize.py index c93a0aa..84eabe5 100644 --- a/src/dspam/tokenize.py +++ b/src/dspam/tokenize.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: BSD-3-Clause from abc import ABC, abstractmethod -from collections.abc import Callable, Generator +from collections.abc import Callable, Iterator from functools import partial from typing import ClassVar @@ -9,7 +9,7 @@ from anyio.functools import lru_cache from dspam.settings import TokenizerSettings -from dspam.types import Metadata, TokenList +from dspam.types import Metadata, Token, TokenList METADATA_TOKEN_SEPARATOR: str = "*" # noqa: S105 """The separator that is used between the metadata key and the tokenized values.""" @@ -83,7 +83,7 @@ def decode_escaped(char_: str) -> str: return "".join(set(unicode_delimiters)) -def tokenize_metadata(metadata: Metadata, value_tokenizer: Callable[[str], TokenList]) -> Generator[str]: +def tokenize_metadata(metadata: Metadata, value_tokenizer: Callable[[str], TokenList]) -> Iterator[Token]: """ Walk a metadata dictionary and tokenize it into a list of tokens. diff --git a/src/dspam/types.py b/src/dspam/types.py index a3a90b4..0e7e0e6 100644 --- a/src/dspam/types.py +++ b/src/dspam/types.py @@ -12,8 +12,11 @@ type Metadata = Mapping[str, str | Sequence[str]] """Metadata as produced by a parser.""" -type TokenList = list[str] -"""TokenList as produced by a tokenizer.""" +type Token = str +"""A single token as produced by a tokenizer.""" + +type TokenList = list[Token] +"""A list of Tokens as produced by a tokenizer.""" Verdict = Literal["innocent", "spam"] """Outcome of a classification."""