feat: add German regex locale support

Author: sidmohan0

README.md

@@ -76,6 +76,25 @@ print(datafog.sanitize("Card: 4111-1111-1111-1111", engine="regex"))
 # Card: [CREDIT_CARD_1]
 ```
 
+## German Structured PII
+
+German VAT IDs and German IBANs are detected by the default regex path because
+their country-code structure is specific enough for default-on screening.
+Broader German identifiers are available with explicit locale selection or
+explicit entity-type filtering.
+
+```python
+import datafog
+
+text = "Steuer-ID 12345678901 liegt vor."
+
+print(datafog.scan(text, engine="regex").entities)
+# []
+
+print(datafog.scan(text, engine="regex", locales=["de"]).entities)
+# [Entity(type='DE_TAX_ID', text='12345678901', ...)]
+```
+
 ### Guardrails
 
 ```python
@@ -98,7 +117,8 @@ Use the engine that matches your accuracy and dependency constraints:
 
 - `regex`:
   - Fastest and always available.
-  - Best for structured entities: `EMAIL`, `PHONE`, `SSN`, `CREDIT_CARD`, `IP_ADDRESS`, `DATE`, `ZIP_CODE`.
+  - Best for structured entities: `EMAIL`, `PHONE`, `SSN`, `CREDIT_CARD`, `IP_ADDRESS`, `DE_VAT_ID`, `DE_IBAN`, `DATE`, `ZIP_CODE`.
+  - Use `locales=["de"]` for broader German structured IDs such as `DE_TAX_ID`, `DE_POSTAL_CODE`, and passport or residence permit numbers.
 - `spacy`:
   - Requires `pip install datafog[nlp]`.
   - Useful for unstructured entities like person and organization names.
@@ -169,6 +189,9 @@ datafog replace-text "john@example.com"
 
 # Hash detected entities
 datafog hash-text "john@example.com"
+
+# Enable broader German regex identifiers
+datafog redact-text "Steuer-ID 12345678901" --locale de
 ```
 
 ## Telemetry

datafog/__init__.py

@@ -152,14 +152,15 @@ def scan(
     text: str,
     engine: str = "regex",
     entity_types: list[str] | None = None,
+    locales: list[str] | None = None,
 ) -> ScanResult:
     """
     v5-preview scan entrypoint.
 
     Defaults to the lightweight regex engine so the core install works without
     optional dependency fallback warnings.
     """
-    return _scan(text=text, engine=engine, entity_types=entity_types)
+    return _scan(text=text, engine=engine, entity_types=entity_types, locales=locales)
 
 
 def redact(
@@ -169,6 +170,7 @@ def redact(
     entity_types: list[str] | None = None,
     strategy: str = "token",
     preset: str | None = None,
+    locales: list[str] | None = None,
 ) -> RedactResult:
     """
     v5-preview redaction entrypoint.
@@ -191,6 +193,7 @@ def redact(
         engine=engine,
         entity_types=entity_types,
         strategy=strategy,
+        locales=locales,
     )
 
 
@@ -199,6 +202,7 @@ def protect(
     engine: str = "regex",
     strategy: str = "token",
     on_detect: str = "redact",
+    locales: list[str] | None = None,
 ):
     """
     v5-preview guardrail factory.
@@ -208,6 +212,7 @@ def protect(
         engine=engine,
         strategy=strategy,
         on_detect=on_detect,
+        locales=locales,
     )
 
 

datafog/agent.py

@@ -27,11 +27,7 @@ class GuardrailWatch:
 
     def scan(self, text: str) -> ScanResult:
         """Scan text and increment detection counters."""
-        result = scan(
-            text=text,
-            engine=self.guardrail.engine,
-            entity_types=self.guardrail.entity_types,
-        )
+        result = self.guardrail.scan(text)
         if result.entities:
             self.detections += len(result.entities)
         return result
@@ -51,6 +47,7 @@ class Guardrail:
     """Reusable text guardrail for wrapping LLM prompts and outputs."""
 
     entity_types: Optional[list[str]] = None
+    locales: Optional[list[str]] = None
     engine: str = "regex"
     strategy: str = "token"
     on_detect: str = "redact"
@@ -61,7 +58,12 @@ def __post_init__(self) -> None:
 
     def scan(self, text: str) -> ScanResult:
         """Scan a text value for entities."""
-        return scan(text=text, engine=self.engine, entity_types=self.entity_types)
+        return scan(
+            text=text,
+            engine=self.engine,
+            entity_types=self.entity_types,
+            locales=self.locales,
+        )
 
     def filter(self, text: str) -> RedactResult:
         """Scan then enforce configured behavior."""
@@ -70,6 +72,7 @@ def filter(self, text: str) -> RedactResult:
             engine=self.engine,
             entity_types=self.entity_types,
             strategy=self.strategy,
+            locales=self.locales,
         )
         if not result.entities:
             return result
@@ -137,6 +140,7 @@ def filter_output(output: str, engine: str = "regex", **kwargs: Any) -> RedactRe
 
 def create_guardrail(
     entity_types: Optional[list[str]] = None,
+    locales: Optional[list[str]] = None,
     engine: str = "regex",
     strategy: str = "token",
     on_detect: str = "redact",
@@ -146,6 +150,7 @@ def create_guardrail(
     """
     return Guardrail(
         entity_types=entity_types,
+        locales=locales,
         engine=engine,
         strategy=strategy,
         on_detect=on_detect,

datafog/client.py

@@ -6,7 +6,7 @@
 
 import asyncio
 import logging
-from typing import List
+from typing import List, Optional
 
 import typer
 
@@ -104,6 +104,11 @@ def scan_text(
         None, help="List of texts to extract text from"
     ),
     operations: str = typer.Option("scan", help="Operation to perform"),
+    locale: Optional[List[str]] = typer.Option(
+        None,
+        "--locale",
+        help="Regex locale to enable, for example de.",
+    ),
 ):
     """
     Scan texts for PII.
@@ -123,7 +128,7 @@ def scan_text(
     logging.basicConfig(level=logging.INFO)
     # Convert comma-separated string operations to a list of OperationType objects
     operation_list = [OperationType(op.strip()) for op in operations.split(",")]
-    text_client = DataFog(operations=operation_list)
+    text_client = DataFog(operations=operation_list, locales=locale)
     try:
         results = text_client.run_text_pipeline_sync(str_list=str_list)
         typer.echo(f"Text Pipeline Results: {results}")
@@ -316,7 +321,14 @@ def list_entities():
 
 
 @app.command()
-def redact_text(text: str = typer.Argument(None, help="Text to redact")):
+def redact_text(
+    text: str = typer.Argument(None, help="Text to redact"),
+    locale: Optional[List[str]] = typer.Option(
+        None,
+        "--locale",
+        help="Regex locale to enable, for example de.",
+    ),
+):
     """
     Redact PII in text.
 
@@ -329,7 +341,12 @@ def redact_text(text: str = typer.Argument(None, help="Text to redact")):
         typer.echo("No text provided to redact.")
         raise typer.Exit(code=1)
 
-    result = scan_and_redact(text=text, engine="smart", strategy="token")
+    result = scan_and_redact(
+        text=text,
+        engine="smart",
+        strategy="token",
+        locales=locale,
+    )
     typer.echo(result.redacted_text)
 
     try:
@@ -346,7 +363,14 @@ def redact_text(text: str = typer.Argument(None, help="Text to redact")):
 
 
 @app.command()
-def replace_text(text: str = typer.Argument(None, help="Text to replace PII")):
+def replace_text(
+    text: str = typer.Argument(None, help="Text to replace PII"),
+    locale: Optional[List[str]] = typer.Option(
+        None,
+        "--locale",
+        help="Regex locale to enable, for example de.",
+    ),
+):
     """
     Replace PII in text with anonymized values.
 
@@ -359,7 +383,12 @@ def replace_text(text: str = typer.Argument(None, help="Text to replace PII")):
         typer.echo("No text provided to replace PII.")
         raise typer.Exit(code=1)
 
-    result = scan_and_redact(text=text, engine="smart", strategy="pseudonymize")
+    result = scan_and_redact(
+        text=text,
+        engine="smart",
+        strategy="pseudonymize",
+        locales=locale,
+    )
     typer.echo(result.redacted_text)
 
     try:
@@ -379,6 +408,11 @@ def replace_text(text: str = typer.Argument(None, help="Text to replace PII")):
 def hash_text(
     text: str = typer.Argument(None, help="Text to hash PII"),
     hash_type: HashType = typer.Option(HashType.SHA256, help="Hash algorithm to use"),
+    locale: Optional[List[str]] = typer.Option(
+        None,
+        "--locale",
+        help="Regex locale to enable, for example de.",
+    ),
 ):
     """
     Choose from SHA256, MD5, or SHA3-256 algorithms to hash detected PII in text.
@@ -395,7 +429,12 @@ def hash_text(
 
     # HashType is retained for backward-compatible CLI signature.
     _ = hash_type
-    result = scan_and_redact(text=text, engine="smart", strategy="hash")
+    result = scan_and_redact(
+        text=text,
+        engine="smart",
+        strategy="hash",
+        locales=locale,
+    )
     typer.echo(result.redacted_text)
 
     try:

datafog/core.py

@@ -9,14 +9,18 @@
 
 from datafog.engine import scan, scan_and_redact
 from datafog.models.anonymizer import AnonymizerType
+from datafog.processing.text_processing.regex_annotator import RegexAnnotator
 
 # Engine types as constants
 REGEX_ENGINE = "regex"
 SPACY_ENGINE = "spacy"
 AUTO_ENGINE = "auto"
 
 
-def detect_pii(text: str) -> Dict[str, List[str]]:
+def detect_pii(
+    text: str,
+    locales: List[str] | None = None,
+) -> Dict[str, List[str]]:
     """
     Simple PII detection using lightweight regex engine.
 
@@ -37,7 +41,7 @@ def detect_pii(text: str) -> Dict[str, List[str]]:
 
     try:
         # Use engine boundary for canonical scan behavior.
-        scan_result = scan(text=text, engine=REGEX_ENGINE)
+        scan_result = scan(text=text, engine=REGEX_ENGINE, locales=locales)
         pii_dict: Dict[str, List[str]] = {}
         for entity in scan_result.entities:
             if not entity.text.strip():
@@ -81,7 +85,11 @@ def detect_pii(text: str) -> Dict[str, List[str]]:
         ) from e
 
 
-def anonymize_text(text: str, method: Union[str, AnonymizerType] = "redact") -> str:
+def anonymize_text(
+    text: str,
+    method: Union[str, AnonymizerType] = "redact",
+    locales: List[str] | None = None,
+) -> str:
     """
     Simple text anonymization using lightweight regex engine.
 
@@ -120,6 +128,7 @@ def anonymize_text(text: str, method: Union[str, AnonymizerType] = "redact") ->
             text=text,
             engine=REGEX_ENGINE,
             strategy=strategy_map[method],
+            locales=locales,
         )
 
         try:
@@ -155,7 +164,9 @@ def anonymize_text(text: str, method: Union[str, AnonymizerType] = "redact") ->
 
 
 def scan_text(
-    text: str, return_entities: bool = False
+    text: str,
+    return_entities: bool = False,
+    locales: List[str] | None = None,
 ) -> Union[bool, Dict[str, List[str]]]:
     """
     Quick scan to check if text contains any PII.
@@ -180,7 +191,7 @@ def scan_text(
 
     _start = _time.monotonic()
 
-    entities = detect_pii(text)
+    entities = detect_pii(text, locales=locales)
 
     result = entities if return_entities else len(entities) > 0
 
@@ -200,7 +211,7 @@ def scan_text(
     return result
 
 
-def get_supported_entities() -> List[str]:
+def get_supported_entities(locales: List[str] | None = None) -> List[str]:
     """
     Get list of PII entity types supported by the regex engine.
 
@@ -210,17 +221,15 @@ def get_supported_entities() -> List[str]:
     Example:
         >>> entities = get_supported_entities()
         >>> print(entities)
-        ['EMAIL', 'PHONE', 'SSN', 'CREDIT_CARD', 'IP_ADDRESS', 'DOB', 'ZIP']
+        ['EMAIL', 'PHONE', 'SSN', 'CREDIT_CARD', 'IP_ADDRESS', 'DE_VAT_ID', 'DE_IBAN', 'DATE', 'ZIP_CODE']
     """
-    result = [
-        "EMAIL",
-        "PHONE",
-        "SSN",
-        "CREDIT_CARD",
-        "IP_ADDRESS",
-        "DATE",
-        "ZIP_CODE",
-    ]
+    annotator = RegexAnnotator(locales=locales)
+    legacy_map = {"DOB": "DATE", "ZIP": "ZIP_CODE"}
+    result = []
+    for label in annotator.active_labels:
+        canonical = legacy_map.get(label, label)
+        if canonical not in result:
+            result.append(canonical)
 
     try:
         from datafog.telemetry import track_function_call