Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion openkb/converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,24 @@ def _registry_path(path: Path, kb_dir: Path) -> str:

_SAFE_STEM_RE = re.compile(r"[^\w\-]+")
_SUFFIX_LEN = 8
_MAX_STEM_LEN = 40


def _sanitize_stem(stem: str) -> str:
"""Sanitize and cap ``stem``, appending a hash suffix if truncated.

``doc_name`` (this function's return value) can appear twice in the
per-add staging path (staging dir name + images subdir), so an
unbounded stem risks Windows' ~260-char path limit. The suffix is a
hash of the FULL cleaned stem (not just the truncated prefix) so two
different overlong stems sharing the same prefix don't collide.
"""
normalized = unicodedata.normalize("NFKC", stem)
return _SAFE_STEM_RE.sub("-", normalized).strip("-") or "document"
cleaned = _SAFE_STEM_RE.sub("-", normalized).strip("-") or "document"
if len(cleaned) > _MAX_STEM_LEN:
digest = hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:_SUFFIX_LEN]
cleaned = f"{cleaned[:_MAX_STEM_LEN].rstrip('-')}-{digest}"
return cleaned


def _name_taken(candidate: str, registry: HashRegistry) -> bool:
Expand Down
34 changes: 34 additions & 0 deletions tests/test_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,40 @@ def test_outside_kb_is_absolute_posix(self, kb_dir, tmp_path_factory):
assert result.startswith("/")


# ---------------------------------------------------------------------------
# _sanitize_stem
# ---------------------------------------------------------------------------


class TestSanitizeStem:
def test_short_stem_is_unchanged(self):
from openkb.converter import _sanitize_stem

assert _sanitize_stem("report") == "report"

def test_long_stem_is_capped_with_hash_suffix(self):
from openkb.converter import _MAX_STEM_LEN, _sanitize_stem

stem = "bulkQuery_result_" + "a" * 60
result = _sanitize_stem(stem)
assert len(result) == _MAX_STEM_LEN + 1 + 8 # prefix + "-" + 8-hex digest
assert result.startswith(stem[:_MAX_STEM_LEN])

def test_long_stem_truncation_is_deterministic(self):
from openkb.converter import _sanitize_stem

stem = "x" * 100
assert _sanitize_stem(stem) == _sanitize_stem(stem)

def test_different_long_stems_with_same_prefix_do_not_collide(self):
from openkb.converter import _MAX_STEM_LEN, _sanitize_stem

prefix = "a" * _MAX_STEM_LEN
first = _sanitize_stem(prefix + "-one")
second = _sanitize_stem(prefix + "-two")
assert first != second


# ---------------------------------------------------------------------------
# resolve_doc_name
# ---------------------------------------------------------------------------
Expand Down