Text filtering service + profanity config
checkpoint / checkpoint (push) Successful in 2m56s

TextFilter protocol + ListBasedTextFilter: normalized (case-fold +
leetspeak/lookalike substitution, non-alnum stripped) substring matching
against a configurable JSON term list, with an index map so matched spans
in normalized text translate back to the original text for redaction.
Supports both redact and reject modes. No external service dependency;
the protocol itself is the extension point for a future third-party
provider. Standalone service module, not yet wired into any endpoint
(sibling tasks own that).

Independently re-verified: 193/193 backend tests, no regressions.
This commit is contained in:
2026-09-15 01:02:00 +00:00
3 changed files with 518 additions and 0 deletions
+304
View File
@@ -0,0 +1,304 @@
"""Pluggable server-side text filtering, `docs/planning/social.md` §4/§5 (US-4).
social.md §4 calls for a "Pluggable `TextFilter` interface with a
deterministic default (profanity list + normalization) so it runs without
external services; third-party hook later". This module is that interface
plus the default implementation:
- `TextFilter` is a `Protocol` -- the extension point. A future third-party
moderation provider (a paid API, an ML classifier, etc.) just needs to
implement `check(text) -> FilterResult`; nothing else in this module
needs to change, and callers depend on the protocol, not a concrete
class.
- `ListBasedTextFilter` is the deterministic default: a configurable
term list (JSON file, see `text_filter_terms.json` next to this module)
matched against a normalized form of the input text so that trivial
evasion (case changes, leetspeak substitutions, and punctuation/spacing
inserted between letters) doesn't slip through.
Nothing here talks to a database, an HTTP client, or the FastAPI app -- it
is a pure, side-effect-free (aside from reading the term list file once, at
construction) service so it can be unit tested in isolation. It is NOT
wired into any router/endpoint by this task; that's for the chat feature
tasks that depend on it (social.md §6) to do.
Normalization approach
-----------------------
Matching naively against raw text misses "sh1t", "@ss", or "s.h.i.t". To
catch these without a real tokenizer, every character of the input is
independently normalized to at most one lowercase ASCII alphanumeric
character (case-folded, then run through a small leetspeak substitution
table); characters that don't reduce to `[a-z0-9]` (punctuation, spacing,
symbols, non-ASCII) are dropped entirely rather than kept as separators.
This collapses "s.h.i.t", "s h i t", and "SHIT" to the same normalized
string ("shit") that a plain lowercase term list entry matches, while a
side table (`index_map`) remembers which original character produced each
surviving normalized character -- so a match found in the normalized
string can still be translated back into a span of the *original* text for
redaction.
This is a deliberate precision/recall tradeoff common to list-based
filters: stripping separators and folding leetspeak increases recall
against evasion at the cost of occasionally matching a term inside an
unrelated longer word (e.g. a 3-letter term inside a longer benign word).
That tradeoff is exactly why this is pluggable -- a deployment that needs
tighter precision can swap in a smarter `TextFilter` implementation behind
the same protocol.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Iterable, Protocol, runtime_checkable
# ---------------------------------------------------------------------------
# Public result / protocol
# ---------------------------------------------------------------------------
class FilterAction(str, Enum):
"""What the caller should do with a checked message."""
ALLOW = "allow"
REDACT = "redact"
REJECT = "reject"
@dataclass(frozen=True)
class FilterResult:
"""Outcome of running a piece of text through a `TextFilter`.
- `action`: ALLOW (no matches), REDACT (matches replaced in
`filtered_text`), or REJECT (matches found; the message should be
blocked -- callers should not persist/deliver it even though
`filtered_text` is populated for logging/audit purposes).
- `filtered_text`: the text to actually store/deliver. Equal to the
input for ALLOW and REJECT; has matched spans replaced for REDACT.
- `matched_terms`: the distinct matched terms (original, un-normalized
list entries), in first-occurrence order. Empty for ALLOW.
"""
action: FilterAction
filtered_text: str
matched_terms: tuple[str, ...] = ()
@runtime_checkable
class TextFilter(Protocol):
"""Extension point for text filtering/moderation providers.
Any implementation -- the deterministic default below, or a future
third-party service call -- just needs this one method.
"""
def check(self, text: str) -> FilterResult:
"""Check `text` and return the action to take plus filtered text."""
...
class FilterMode(str, Enum):
"""Configures what `ListBasedTextFilter` does when it finds a match."""
REDACT = "redact"
REJECT = "reject"
# ---------------------------------------------------------------------------
# Normalization
# ---------------------------------------------------------------------------
# Common leetspeak / lookalike substitutions, applied after case-folding.
# Keys are already-lowercased single characters.
_LEET_MAP: dict[str, str] = {
"0": "o",
"1": "i",
"!": "i",
"3": "e",
"4": "a",
"@": "a",
"5": "s",
"$": "s",
"7": "t",
"+": "t",
"8": "b",
"|": "l",
}
_ALNUM_RE = re.compile(r"[a-z0-9]")
def _normalize(text: str) -> tuple[str, list[int]]:
"""Fold `text` to a lowercase-alnum-only string plus an index map.
Returns `(normalized, index_map)` where `index_map[i]` is the index
into the *original* `text` that produced `normalized[i]`. Characters
that don't reduce to a single `[a-z0-9]` character (punctuation,
whitespace, symbols, most non-ASCII) are dropped from both outputs.
"""
normalized_chars: list[str] = []
index_map: list[int] = []
for i, ch in enumerate(text):
folded = ch.casefold()
# case-fold can expand to multiple characters for some unicode
# (e.g. German "ß" -> "ss"); such characters are not part of the
# leetspeak table and are not plain ascii alnum, so just skip them
# like any other non-matching character rather than guessing.
if len(folded) != 1:
continue
folded = _LEET_MAP.get(folded, folded)
if _ALNUM_RE.fullmatch(folded):
normalized_chars.append(folded)
index_map.append(i)
return "".join(normalized_chars), index_map
def _normalize_term(term: str) -> str:
normalized, _ = _normalize(term)
return normalized
# ---------------------------------------------------------------------------
# Default term list loading
# ---------------------------------------------------------------------------
DEFAULT_TERMS_PATH = Path(__file__).with_name("text_filter_terms.json")
def load_terms(path: Path | str | None = None) -> list[str]:
"""Load a term list from a JSON file shaped `{"terms": [...]}`.
Any other top-level keys (e.g. `_readme`) are ignored. Defaults to the
bundled `text_filter_terms.json` next to this module.
"""
resolved = Path(path) if path is not None else DEFAULT_TERMS_PATH
with resolved.open("r", encoding="utf-8") as f:
data = json.load(f)
terms = data.get("terms", [])
if not isinstance(terms, list):
raise ValueError(f"{resolved}: 'terms' must be a list")
return [str(t) for t in terms]
# ---------------------------------------------------------------------------
# Span merging / redaction helpers
# ---------------------------------------------------------------------------
def _merge_spans(spans: list[tuple[int, int, str]]) -> list[tuple[int, int]]:
"""Merge overlapping/adjacent (start, end) spans, sorted by start."""
if not spans:
return []
ordered = sorted((s, e) for s, e, _ in spans)
merged: list[tuple[int, int]] = [ordered[0]]
for start, end in ordered[1:]:
last_start, last_end = merged[-1]
if start <= last_end:
merged[-1] = (last_start, max(last_end, end))
else:
merged.append((start, end))
return merged
def _apply_redactions(text: str, merged_spans: list[tuple[int, int]], redaction_char: str) -> str:
if not merged_spans:
return text
pieces: list[str] = []
cursor = 0
for start, end in merged_spans:
pieces.append(text[cursor:start])
pieces.append(redaction_char * (end - start))
cursor = end
pieces.append(text[cursor:])
return "".join(pieces)
# ---------------------------------------------------------------------------
# Default implementation
# ---------------------------------------------------------------------------
class ListBasedTextFilter:
"""Deterministic `TextFilter`: normalized substring match against a term list.
`mode` controls what happens on a match:
- `FilterMode.REDACT` (default): matched spans in the *original* text
are replaced with `redaction_char` repeated to the matched span's
original length; `action` is REDACT if anything matched, else ALLOW.
- `FilterMode.REJECT`: `filtered_text` is returned unchanged but
`action` is REJECT if anything matched (callers should treat this as
"do not store/deliver"), else ALLOW.
An empty or whitespace-only string always returns ALLOW with no
matches -- there is nothing to filter.
"""
def __init__(
self,
terms: Iterable[str] | None = None,
*,
mode: FilterMode = FilterMode.REDACT,
terms_path: Path | str | None = None,
redaction_char: str = "*",
) -> None:
if terms is None:
terms = load_terms(terms_path)
# De-dupe while preserving order; drop blanks.
seen: dict[str, None] = {}
for t in terms:
t = t.strip()
if t:
seen.setdefault(t, None)
self._terms: tuple[str, ...] = tuple(seen)
# Pre-normalize once at construction; skip terms that normalize to
# nothing (e.g. a term that's pure punctuation).
self._normalized_terms: list[tuple[str, str]] = [
(normalized, original)
for original in self._terms
if (normalized := _normalize_term(original))
]
self.mode = mode
self.redaction_char = redaction_char
def check(self, text: str) -> FilterResult:
if not text or not text.strip():
return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=())
normalized, index_map = _normalize(text)
if not normalized:
return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=())
spans: list[tuple[int, int, str]] = []
for normalized_term, original_term in self._normalized_terms:
start = 0
term_len = len(normalized_term)
while True:
idx = normalized.find(normalized_term, start)
if idx == -1:
break
end = idx + term_len
orig_start = index_map[idx]
orig_end = index_map[end - 1] + 1
spans.append((orig_start, orig_end, original_term))
start = end # non-overlapping: resume after this match
if not spans:
return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=())
spans.sort(key=lambda s: s[0])
matched_terms = tuple(dict.fromkeys(term for _, _, term in spans))
if self.mode is FilterMode.REJECT:
return FilterResult(action=FilterAction.REJECT, filtered_text=text, matched_terms=matched_terms)
merged = _merge_spans(spans)
filtered_text = _apply_redactions(text, merged, self.redaction_char)
return FilterResult(action=FilterAction.REDACT, filtered_text=filtered_text, matched_terms=matched_terms)
def default_text_filter(mode: FilterMode = FilterMode.REDACT) -> ListBasedTextFilter:
"""Convenience factory for the bundled default term list."""
return ListBasedTextFilter(mode=mode)
@@ -0,0 +1,20 @@
{
"_readme": "Default deterministic term list for app.services.text_filter.ListBasedTextFilter. This is a minimal, illustrative starter list (common mild-to-strong English profanity, no slurs) meant to be replaced/extended per-deployment -- swap this file, or pass an explicit `terms`/`terms_path` to the filter, rather than editing this list in place for production moderation policy.",
"terms": [
"fuck",
"shit",
"ass",
"asshole",
"bitch",
"bastard",
"damn",
"crap",
"dick",
"piss",
"slut",
"whore",
"cunt",
"douche",
"bullshit"
]
}
+194
View File
@@ -0,0 +1,194 @@
"""Unit tests for app.services.text_filter (docs/planning/social.md §4/§5 US-4).
Pure, side-effect-free service -- no DB/Redis fixtures needed.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from app.services.text_filter import (
DEFAULT_TERMS_PATH,
FilterAction,
FilterMode,
FilterResult,
ListBasedTextFilter,
TextFilter,
default_text_filter,
load_terms,
)
@pytest.fixture
def redact_filter() -> ListBasedTextFilter:
return ListBasedTextFilter(terms=["badword", "shoot"], mode=FilterMode.REDACT)
@pytest.fixture
def reject_filter() -> ListBasedTextFilter:
return ListBasedTextFilter(terms=["badword", "shoot"], mode=FilterMode.REJECT)
class TestAllowPath:
def test_clean_text_is_allowed_unchanged(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("hello there, how are you today?")
assert result.action is FilterAction.ALLOW
assert result.filtered_text == "hello there, how are you today?"
assert result.matched_terms == ()
def test_message_with_no_matches(self, reject_filter: ListBasedTextFilter) -> None:
# Distinct from the clean-text case above: exercises the reject-mode
# filter's no-match path specifically (mode should not matter when
# nothing matches).
text = "let's meet at the plaza around noon"
result = reject_filter.check(text)
assert result.action is FilterAction.ALLOW
assert result.filtered_text == text
assert result.matched_terms == ()
class TestRedactPath:
def test_exact_match_is_redacted(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("don't be a badword about it")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("badword",)
assert "badword" not in result.filtered_text
assert result.filtered_text == "don't be a ******* about it"
def test_multiple_distinct_matches(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("badword, don't shoot!")
assert result.action is FilterAction.REDACT
assert set(result.matched_terms) == {"badword", "shoot"}
assert "badword" not in result.filtered_text
assert "shoot" not in result.filtered_text
# Length is preserved (asterisks replace matched spans 1:1).
assert len(result.filtered_text) == len("badword, don't shoot!")
def test_redaction_char_is_configurable(self) -> None:
f = ListBasedTextFilter(terms=["badword"], mode=FilterMode.REDACT, redaction_char="#")
result = f.check("that's a badword thing to say")
assert result.filtered_text == "that's a ####### thing to say"
class TestRejectPath:
def test_match_is_rejected_and_text_left_intact(self, reject_filter: ListBasedTextFilter) -> None:
text = "stop, badword!"
result = reject_filter.check(text)
assert result.action is FilterAction.REJECT
assert result.matched_terms == ("badword",)
# Reject mode does not mutate the text -- callers are expected to
# not store/deliver it at all, based on `action`.
assert result.filtered_text == text
class TestNormalization:
def test_leetspeak_variant_is_caught(self, redact_filter: ListBasedTextFilter) -> None:
# "b4dw0rd" -> normalizes to "badword" via the leet map (4->a, 0->o).
result = redact_filter.check("you're being a b4dw0rd right now")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("badword",)
def test_separators_between_letters_are_stripped(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("that is b.a.d.w.o.r.d behavior")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("badword",)
def test_spacing_between_letters_is_stripped(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("b a d w o r d, seriously")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("badword",)
def test_case_folding(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("BADWORD in all caps")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("badword",)
def test_symbol_lookalikes(self) -> None:
# @ -> a, $ -> s: "@ss" should normalize to "ass".
f = ListBasedTextFilter(terms=["ass"], mode=FilterMode.REDACT)
result = f.check("what an @$$ move")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("ass",)
class TestEmptyAndWhitespace:
def test_empty_string(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("")
assert result.action is FilterAction.ALLOW
assert result.filtered_text == ""
assert result.matched_terms == ()
def test_whitespace_only_string(self, redact_filter: ListBasedTextFilter) -> None:
text = " \t\n "
result = redact_filter.check(text)
assert result.action is FilterAction.ALLOW
assert result.filtered_text == text
assert result.matched_terms == ()
def test_punctuation_only_string_has_nothing_to_normalize(self, redact_filter: ListBasedTextFilter) -> None:
# No alnum characters survive normalization at all -- must not
# crash on an empty normalized string / empty index_map.
result = redact_filter.check("...!!!???")
assert result.action is FilterAction.ALLOW
assert result.matched_terms == ()
class TestConfigurability:
def test_custom_term_list_is_used_instead_of_default(self) -> None:
f = ListBasedTextFilter(terms=["zzzcustomzzz"], mode=FilterMode.REDACT)
result = f.check("this contains zzzcustomzzz right here")
assert result.action is FilterAction.REDACT
assert result.matched_terms == ("zzzcustomzzz",)
# Default list terms should NOT match since we overrode the list.
result2 = f.check("this is fucking unrelated")
assert result2.action is FilterAction.ALLOW
def test_duplicate_and_blank_terms_are_deduped(self) -> None:
f = ListBasedTextFilter(terms=["badword", "badword", " ", ""], mode=FilterMode.REDACT)
assert f._terms == ("badword",)
def test_default_terms_file_loads_and_is_nonempty(self) -> None:
terms = load_terms()
assert isinstance(terms, list)
assert len(terms) > 0
assert all(isinstance(t, str) and t for t in terms)
def test_default_terms_path_exists(self) -> None:
assert DEFAULT_TERMS_PATH.exists()
assert DEFAULT_TERMS_PATH.suffix == ".json"
def test_load_terms_from_explicit_path(self, tmp_path: Path) -> None:
custom = tmp_path / "custom_terms.json"
custom.write_text('{"terms": ["foo", "bar"]}', encoding="utf-8")
assert load_terms(custom) == ["foo", "bar"]
def test_default_text_filter_factory_matches_default_list(self) -> None:
f = default_text_filter(mode=FilterMode.REJECT)
result = f.check("that's fucking ridiculous")
assert result.action is FilterAction.REJECT
assert "fuck" in result.matched_terms
class TestProtocol:
def test_list_based_filter_satisfies_text_filter_protocol(self, redact_filter: ListBasedTextFilter) -> None:
assert isinstance(redact_filter, TextFilter)
def test_check_returns_filter_result(self, redact_filter: ListBasedTextFilter) -> None:
result = redact_filter.check("hi")
assert isinstance(result, FilterResult)
def test_a_third_party_style_implementation_satisfies_the_protocol(self) -> None:
# Demonstrates the extension point: any object with a matching
# `check` method satisfies TextFilter without inheriting from
# anything -- this is how a future third-party provider would plug
# in without touching this module.
class AlwaysAllow:
def check(self, text: str) -> FilterResult:
return FilterResult(action=FilterAction.ALLOW, filtered_text=text, matched_terms=())
provider: TextFilter = AlwaysAllow()
assert isinstance(provider, TextFilter)
result = provider.check("anything at all, even fuck")
assert result.action is FilterAction.ALLOW