API Reference

This page is automatically generated from source code using mkdocstrings.

It reflects the current public surface of pdf-anonymizer-core. The hand-written usage guide is in SDK & API Usage. For practical examples see Recipes & Common Workflows.

Info

The CLI (pdf-anonymizer-cli) is a thin wrapper built with Typer. Most logic lives in the core package.

---

Core Functions

pdf_anonymizer_core.core.anonymize_file(file_path: str, characters_to_anonymize: int, prompt_template: str, model_name: str, anonymized_entities: Optional[List[str]] = None, chunk_overlap: int = DEFAULT_CHUNK_OVERLAP, regex_patterns: Optional[Dict[str, str]] = None, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0, operators: Optional[Dict[str, str]] = None, fake_secret: Optional[str] = None, encrypt_secret: Optional[str] = None, seed_mapping: Optional[Dict[str, str]] = None, keep_list: Optional[List[str]] = None, deny_list: Optional[List[str]] = None, use_llm: bool = True, use_ner: bool = False, ocr: bool = False, min_confidence: float = 0.0) -> Tuple[Optional[str], Optional[Dict[str, str]]]

Anonymize a file by processing its text content.

Performs a two-stage entity detection (fast regex first pass followed by LLM-based semantic detection), deduplicates, consolidates base forms for coreference (e.g. "Dr. Smith" / "Smith"), generates typed placeholders (PERSON_1, ORGANIZATION_3.v_1, ...), and replaces non-overlapping spans (longest-first, written from the end) to produce reversible anonymized output.

The function streams large inputs via chunking (Markdown-aware for PDF/MD) so that very large files (hundreds of MB) can be processed without exhausting context windows or memory.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the file to anonymize (.pdf, .md, .txt, .csv, .xlsx, or .docx). .xlsx needs the [excel] extra. .docx needs the [docx] extra. .xls / .xlsm / .ods / .xlsb and .doc / .docm / .dot / .dotm / .dotx are rejected.	required
characters_to_anonymize	int	Target character size of each chunk sent to the LLM.	required
prompt_template	str	The full prompt template string (use one from pdf_anonymizer_core.prompts or supply your own).	required
model_name	str	Model identifier or "provider/model" string (e.g. "gemini-2.5-flash", "ollama/phi4-mini", "google/gemini-2.0-flash-exp").	required
anonymized_entities	Optional[List[str]]	Optional whitelist of entity types (e.g. ["PERSON", "ORGANIZATION"]). When provided, only matching entities are replaced.	None
chunk_overlap	int	Number of characters of overlap between consecutive chunks.	DEFAULT_CHUNK_OVERLAP
regex_patterns	Optional[Dict[str, str]]	Custom first-stage regex map. Defaults to the large built-in collection in DEFAULT_REGEX_PATTERNS (see conf.py). The collection covers universal PII (email, URLs, credit cards, crypto wallets, IBANs, VIN, MAC, IPv4/IPv6, dates) plus country-partitioned national IDs, tax IDs, driver licenses, VAT/business numbers, passports etc. for 30+ countries (mandatory: US, CA, GB, ES, IT, FR, IN, CN plus DE, JP, BR, AU, NL, ...). Keys become entity TYPEs (IPV4_ADDRESS, SSN_US, IBAN, CRYPTO_ETH, ...). All patterns are RE2 (google-re2) safe.	None
max_retries	int	Maximum LLM call attempts per chunk (with exponential backoff).	3
base_retry_delay	float	Base delay in seconds for retry backoff.	1.0
max_retry_delay	float	Maximum delay cap for retry backoff.	10.0
operators	Optional[Dict[str, str]]	Optional map of entity type → operator (replace, mask, hash, generalize, shift, fake, encrypt). Types not listed keep replace. CREDIT_CARD_LIKE follows CREDIT_CARD.	None
fake_secret	Optional[str]	Optional seed material for the fake operator. Same person + type + secret always yields the same fake.	None
encrypt_secret	Optional[str]	Secret for the encrypt operator. Same text always yields the same token. Required when any type uses encrypt.	None
seed_mapping	Optional[Dict[str, str]]	Optional original → written map from a previous file so the same person keeps PERSON_1 (or the same fake) across documents.	None
keep_list	Optional[List[str]]	Phrases that must stay visible even if detected.	None
deny_list	Optional[List[str]]	Phrases that must be replaced even if detection missed them.	None
use_llm	bool	When False, skip identify_entities_with_llm. Regex, checksums, operators, gazetteers, and span replacement still run. Names and identity clues will be missed. Default True.	True
use_ner	bool	When True, run local span NER (GLiNER extra) for names and organizations. Default False so the SDK never downloads a checkpoint unless the caller asks.	False
min_confidence	float	Drop entities whose score is below this value (0–1). Default 0 keeps today's accept-all behaviour. Scores are recognizer hints, not calibrated probabilities.	0.0
ocr	bool	When True and a PDF has no text layer, run Tesseract via PyMuPDF and stash word boxes for a later native-PDF write. A scan with OCR off (or OCR that returns nothing) raises.	False

Returns:

Type	Description
Tuple[Optional[str], Optional[Dict[str, str]]]	A tuple (anonymized_text, mapping) where: - anonymized_text is the masked document (or None on failure) - mapping is a dict of original_value -> placeholder (or None on failure)

Note

The returned mapping is in original -> placeholder direction. The CLI later converts it to placeholder -> original for deanonymization.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/core.py

def anonymize_file(
    file_path: str,
    characters_to_anonymize: int,
    prompt_template: str,
    model_name: str,
    anonymized_entities: Optional[List[str]] = None,
    chunk_overlap: int = DEFAULT_CHUNK_OVERLAP,
    regex_patterns: Optional[Dict[str, str]] = None,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
    operators: Optional[Dict[str, str]] = None,
    fake_secret: Optional[str] = None,
    encrypt_secret: Optional[str] = None,
    seed_mapping: Optional[Dict[str, str]] = None,
    keep_list: Optional[List[str]] = None,
    deny_list: Optional[List[str]] = None,
    use_llm: bool = True,
    use_ner: bool = False,
    ocr: bool = False,
    min_confidence: float = 0.0,
) -> Tuple[Optional[str], Optional[Dict[str, str]]]:
    """Anonymize a file by processing its text content.

    Performs a two-stage entity detection (fast regex first pass followed by
    LLM-based semantic detection), deduplicates, consolidates base forms for
    coreference (e.g. "Dr. Smith" / "Smith"), generates typed placeholders
    (PERSON_1, ORGANIZATION_3.v_1, ...), and replaces non-overlapping spans
    (longest-first, written from the end) to produce reversible anonymized output.

    The function streams large inputs via chunking (Markdown-aware for PDF/MD)
    so that very large files (hundreds of MB) can be processed without
    exhausting context windows or memory.

    Args:
        file_path: Path to the file to anonymize (.pdf, .md, .txt, .csv,
            .xlsx, or .docx). ``.xlsx`` needs the ``[excel]`` extra.
            ``.docx`` needs the ``[docx]`` extra. ``.xls`` / ``.xlsm`` /
            ``.ods`` / ``.xlsb`` and ``.doc`` / ``.docm`` / ``.dot`` /
            ``.dotm`` / ``.dotx`` are rejected.
        characters_to_anonymize: Target character size of each chunk sent to the LLM.
        prompt_template: The full prompt template string (use one from
            pdf_anonymizer_core.prompts or supply your own).
        model_name: Model identifier or "provider/model" string
            (e.g. "gemini-2.5-flash", "ollama/phi4-mini", "google/gemini-2.0-flash-exp").
        anonymized_entities: Optional whitelist of entity *types* (e.g. ["PERSON", "ORGANIZATION"]).
            When provided, only matching entities are replaced.
        chunk_overlap: Number of characters of overlap between consecutive chunks.
        regex_patterns: Custom first-stage regex map. Defaults to the large built-in
            collection in DEFAULT_REGEX_PATTERNS (see conf.py). The collection covers
            universal PII (email, URLs, credit cards, crypto wallets, IBANs, VIN, MAC,
            IPv4/IPv6, dates) plus country-partitioned national IDs, tax IDs, driver
            licenses, VAT/business numbers, passports etc. for 30+ countries
            (mandatory: US, CA, GB, ES, IT, FR, IN, CN plus DE, JP, BR, AU, NL, ...).
            Keys become entity TYPEs (IPV4_ADDRESS, SSN_US, IBAN, CRYPTO_ETH, ...).
            All patterns are RE2 (google-re2) safe.
        max_retries: Maximum LLM call attempts per chunk (with exponential backoff).
        base_retry_delay: Base delay in seconds for retry backoff.
        max_retry_delay: Maximum delay cap for retry backoff.
        operators: Optional map of entity type → operator (replace, mask, hash,
            generalize, shift, fake, encrypt). Types not listed keep ``replace``.
            ``CREDIT_CARD_LIKE`` follows ``CREDIT_CARD``.
        fake_secret: Optional seed material for the ``fake`` operator. Same
            person + type + secret always yields the same fake.
        encrypt_secret: Secret for the ``encrypt`` operator. Same text always
            yields the same token. Required when any type uses ``encrypt``.
        seed_mapping: Optional original → written map from a previous file so
            the same person keeps PERSON_1 (or the same fake) across documents.
        keep_list: Phrases that must stay visible even if detected.
        deny_list: Phrases that must be replaced even if detection missed them.
        use_llm: When False, skip ``identify_entities_with_llm``. Regex,
            checksums, operators, gazetteers, and span replacement still run.
            Names and identity clues will be missed. Default True.
        use_ner: When True, run local span NER (GLiNER extra) for names and
            organizations. Default False so the SDK never downloads a
            checkpoint unless the caller asks.
        min_confidence: Drop entities whose ``score`` is below this value
            (0–1). Default 0 keeps today's accept-all behaviour. Scores are
            recognizer hints, not calibrated probabilities.
        ocr: When True and a PDF has no text layer, run Tesseract via
            PyMuPDF and stash word boxes for a later native-PDF write.
            A scan with OCR off (or OCR that returns nothing) raises.

    Returns:
        A tuple (anonymized_text, mapping) where:
            - anonymized_text is the masked document (or None on failure)
            - mapping is a dict of original_value -> placeholder (or None on failure)

    Note:
        The returned mapping is in original -> placeholder direction.
        The CLI later converts it to placeholder -> original for deanonymization.
    """
    if regex_patterns is None:
        regex_patterns = DEFAULT_REGEX_PATTERNS

    if is_rejected_word(file_path):
        raise rejected_word_error(file_path)
    if is_word_path(file_path):
        review, mapping, _entity_texts = anonymize_docx_file(
            file_path,
            characters_to_anonymize,
            prompt_template,
            model_name,
            anonymized_entities=anonymized_entities,
            chunk_overlap=chunk_overlap,
            regex_patterns=regex_patterns,
            max_retries=max_retries,
            base_retry_delay=base_retry_delay,
            max_retry_delay=max_retry_delay,
            operators=operators,
            fake_secret=fake_secret,
            encrypt_secret=encrypt_secret,
            seed_mapping=seed_mapping,
            keep_list=keep_list,
            deny_list=deny_list,
            use_llm=use_llm,
            use_ner=use_ner,
            min_confidence=min_confidence,
        )
        return review, mapping

    if is_rejected_spreadsheet(file_path):
        raise rejected_spreadsheet_error(file_path)
    if is_tabular_path(file_path):
        review, mapping, _entity_texts = anonymize_tabular_file(
            file_path,
            characters_to_anonymize,
            prompt_template,
            model_name,
            anonymized_entities=anonymized_entities,
            chunk_overlap=chunk_overlap,
            regex_patterns=regex_patterns,
            max_retries=max_retries,
            base_retry_delay=base_retry_delay,
            max_retry_delay=max_retry_delay,
            operators=operators,
            fake_secret=fake_secret,
            encrypt_secret=encrypt_secret,
            seed_mapping=seed_mapping,
            keep_list=keep_list,
            deny_list=deny_list,
            use_llm=use_llm,
            use_ner=use_ner,
            min_confidence=min_confidence,
        )
        return review, mapping

    file_size = os.path.getsize(file_path)
    full_text, text_pages = load_and_extract_text_from_file(
        file_path, characters_to_anonymize, chunk_overlap, ocr=ocr
    )

    if not text_pages:
        logging.warning("No text could be extracted from the file.")
        return None, None

    logging.info(f"Extracted text pages: {text_pages[0][:50]} ...")
    extracted_text_size = len(full_text)

    logging.info(f"  - File size: {file_size / 1024:.2f} KB")
    logging.info(f"  - Extracted text size: {extracted_text_size / 1024:.2f} KB")

    return anonymize_text_content(
        full_text,
        text_pages,
        prompt_template=prompt_template,
        model_name=model_name,
        anonymized_entities=anonymized_entities,
        regex_patterns=regex_patterns,
        max_retries=max_retries,
        base_retry_delay=base_retry_delay,
        max_retry_delay=max_retry_delay,
        operators=operators,
        fake_secret=fake_secret,
        encrypt_secret=encrypt_secret,
        seed_mapping=seed_mapping,
        keep_list=keep_list,
        deny_list=deny_list,
        use_llm=use_llm,
        use_ner=use_ner,
        min_confidence=min_confidence,
    )

pdf_anonymizer_core.core.anonymize_tabular_file(file_path: str, characters_to_anonymize: int, prompt_template: str, model_name: str, anonymized_entities: Optional[List[str]] = None, chunk_overlap: int = DEFAULT_CHUNK_OVERLAP, regex_patterns: Optional[Dict[str, str]] = None, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0, operators: Optional[Dict[str, str]] = None, fake_secret: Optional[str] = None, encrypt_secret: Optional[str] = None, seed_mapping: Optional[Dict[str, str]] = None, keep_list: Optional[List[str]] = None, deny_list: Optional[List[str]] = None, use_llm: bool = True, use_ner: bool = False, min_confidence: float = 0.0) -> Tuple[str, Dict[str, str], Tuple[str, ...]]

Anonymize a CSV/Excel file cell by cell.

Returns (review_flatten, orig→written, entity_texts). entity_texts is the same entity["text"] list the text engine passes to replace_entities.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/core.py

def anonymize_tabular_file(
    file_path: str,
    characters_to_anonymize: int,
    prompt_template: str,
    model_name: str,
    anonymized_entities: Optional[List[str]] = None,
    chunk_overlap: int = DEFAULT_CHUNK_OVERLAP,
    regex_patterns: Optional[Dict[str, str]] = None,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
    operators: Optional[Dict[str, str]] = None,
    fake_secret: Optional[str] = None,
    encrypt_secret: Optional[str] = None,
    seed_mapping: Optional[Dict[str, str]] = None,
    keep_list: Optional[List[str]] = None,
    deny_list: Optional[List[str]] = None,
    use_llm: bool = True,
    use_ner: bool = False,
    min_confidence: float = 0.0,
) -> Tuple[str, Dict[str, str], Tuple[str, ...]]:
    """Anonymize a CSV/Excel file cell by cell.

    Returns ``(review_flatten, orig→written, entity_texts)``. ``entity_texts``
    is the same ``entity["text"]`` list the text engine passes to
    ``replace_entities``.
    """
    del chunk_overlap  # row boundaries replace chunk overlap
    if regex_patterns is None:
        regex_patterns = DEFAULT_REGEX_PATTERNS

    if is_rejected_spreadsheet(file_path):
        raise rejected_spreadsheet_error(file_path)

    doc = load_table(file_path)
    cells = list(iter_cells(doc))
    collected_entities: List[dict] = []

    for cell in cells:
        if not cell.search_text or cell.kind not in REGEX_CELL_KINDS:
            continue
        collected_entities.extend(
            extract_entities_via_regex(cell.search_text, regex_patterns)
        )
        if use_ner:
            collected_entities.extend(extract_entities_via_ner(cell.search_text))

    if use_llm:
        batches = build_llm_batches(doc, characters_to_anonymize)
        for i, batch in enumerate(batches):
            logging.info(
                f"Identifying entities in table batch {i + 1}/{len(batches)}..."
            )
            llm_entities = identify_entities_with_llm(
                batch,
                prompt_template,
                model_name,
                max_retries=max_retries,
                base_retry_delay=base_retry_delay,
                max_retry_delay=max_retry_delay,
            )
            for entity in llm_entities:
                text = entity.get("text") or ""
                if _llm_entity_in_cells(text, cells):
                    collected_entities.append(entity)
    else:
        logging.info(
            "Regex-only / offline mode: skipping the language model. "
            "Names and identity clues will be missed."
        )

    if deny_list:
        for cell in cells:
            if not cell.search_text:
                continue
            collected_entities.extend(apply_deny_list(cell.search_text, [], deny_list))

    entities_to_process = finalize_entities(
        collected_entities,
        "",
        anonymized_entities=anonymized_entities,
        keep_list=keep_list,
        deny_list=deny_list,
        apply_deny=False,
        seed_mapping=seed_mapping,
        min_confidence=min_confidence,
    )
    final_mapping = build_mapping(
        entities_to_process,
        seed_mapping=seed_mapping,
        operators=operators,
        fake_secret=fake_secret,
        encrypt_secret=encrypt_secret,
    )
    entity_texts = tuple(
        entity["text"] for entity in entities_to_process if entity.get("text")
    )
    apply_mapping_to_table(doc, final_mapping, entity_texts)
    review = flatten_table_for_review(doc, anonymized=True)
    return review, final_mapping, entity_texts

pdf_anonymizer_core.core.anonymize_docx_file(file_path: str, characters_to_anonymize: int, prompt_template: str, model_name: str, anonymized_entities: Optional[List[str]] = None, chunk_overlap: int = DEFAULT_CHUNK_OVERLAP, regex_patterns: Optional[Dict[str, str]] = None, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0, operators: Optional[Dict[str, str]] = None, fake_secret: Optional[str] = None, encrypt_secret: Optional[str] = None, seed_mapping: Optional[Dict[str, str]] = None, keep_list: Optional[List[str]] = None, deny_list: Optional[List[str]] = None, use_llm: bool = True, use_ner: bool = False, min_confidence: float = 0.0) -> Tuple[str, Dict[str, str], Tuple[str, ...]]

Anonymize a Word .docx file paragraph by paragraph.

Returns (review_flatten, orig→written, entity_texts). entity_texts is the same entity["text"] list the text engine passes to replace_entities.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/core.py

def anonymize_docx_file(
    file_path: str,
    characters_to_anonymize: int,
    prompt_template: str,
    model_name: str,
    anonymized_entities: Optional[List[str]] = None,
    chunk_overlap: int = DEFAULT_CHUNK_OVERLAP,
    regex_patterns: Optional[Dict[str, str]] = None,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
    operators: Optional[Dict[str, str]] = None,
    fake_secret: Optional[str] = None,
    encrypt_secret: Optional[str] = None,
    seed_mapping: Optional[Dict[str, str]] = None,
    keep_list: Optional[List[str]] = None,
    deny_list: Optional[List[str]] = None,
    use_llm: bool = True,
    use_ner: bool = False,
    min_confidence: float = 0.0,
) -> Tuple[str, Dict[str, str], Tuple[str, ...]]:
    """Anonymize a Word ``.docx`` file paragraph by paragraph.

    Returns ``(review_flatten, orig→written, entity_texts)``. ``entity_texts``
    is the same ``entity["text"]`` list the text engine passes to
    ``replace_entities``.
    """
    if regex_patterns is None:
        regex_patterns = DEFAULT_REGEX_PATTERNS

    if is_rejected_word(file_path):
        raise rejected_word_error(file_path)

    doc = load_docx(file_path)
    blocks = list(doc.blocks)
    collected_entities: List[dict] = []

    for block in blocks:
        if not block.search_text:
            continue
        collected_entities.extend(
            extract_entities_via_regex(block.search_text, regex_patterns)
        )
        if use_ner:
            collected_entities.extend(extract_entities_via_ner(block.search_text))

    full_text = flatten_docx_for_review(doc, anonymized=False)
    if use_llm:
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=max(characters_to_anonymize, 1),
            chunk_overlap=max(chunk_overlap, 0),
        )
        chunks = splitter.split_text(full_text) if full_text.strip() else []
        for i, chunk in enumerate(chunks):
            logging.info(f"Identifying entities in Word chunk {i + 1}/{len(chunks)}...")
            llm_entities = identify_entities_with_llm(
                chunk,
                prompt_template,
                model_name,
                max_retries=max_retries,
                base_retry_delay=base_retry_delay,
                max_retry_delay=max_retry_delay,
            )
            for entity in llm_entities:
                text = entity.get("text") or ""
                if _llm_entity_in_blocks(text, blocks):
                    collected_entities.append(entity)
    else:
        logging.info(
            "Regex-only / offline mode: skipping the language model. "
            "Names and identity clues will be missed."
        )

    if deny_list:
        for block in blocks:
            if not block.search_text:
                continue
            collected_entities.extend(apply_deny_list(block.search_text, [], deny_list))

    entities_to_process = finalize_entities(
        collected_entities,
        full_text,
        anonymized_entities=anonymized_entities,
        keep_list=keep_list,
        deny_list=deny_list,
        apply_deny=False,
        seed_mapping=seed_mapping,
        min_confidence=min_confidence,
    )
    final_mapping = build_mapping(
        entities_to_process,
        seed_mapping=seed_mapping,
        operators=operators,
        fake_secret=fake_secret,
        encrypt_secret=encrypt_secret,
    )
    entity_texts = tuple(
        entity["text"] for entity in entities_to_process if entity.get("text")
    )
    apply_mapping_to_docx(doc, final_mapping, entity_texts)
    review = flatten_docx_for_review(doc, anonymized=True)
    return review, final_mapping, entity_texts

pdf_anonymizer_core.utils.deanonymize_file(anonymized_file_path: str, mapping_file_path: str, mapping_passphrase: str | None = None, *, expected_source_sha256: str | None = None, encrypt_secret: str | None = None) -> tuple[str, str]

Deanonymize a file using a mapping file.

Restores original PII values from placeholders. Supports both current (placeholder -> original) and legacy (original -> placeholder) mapping directions via auto-detection.

The implementation correctly handles variation placeholders (PERSON_1.v_2 etc.) by falling back to the base placeholder's original value.

After processing it also produces an audit statistics JSON file.

Parameters:

Name	Type	Description	Default
anonymized_file_path	str	Path to the previously anonymized document.	required
mapping_file_path	str	Path to the JSON mapping file (plaintext or .enc).	required
mapping_passphrase	str | None	Required when the mapping file is encrypted.	None
expected_source_sha256	str | None	Optional SHA-256 of the original source document. When set, an encrypted mapping locked to a different file is rejected (AAD mismatch).	None
encrypt_secret	str | None	Secret used by the encrypt operator. Decrypts ENC1_ tokens after placeholder restore.	None

Returns:

Type	Description
str	A tuple (deanonymized_file_path, stats_file_path).
str	The stats file contains: - unused_mappings: placeholders in the map that did not appear in text - not_found_mappings: placeholders seen in text but missing from the map

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/utils.py

def deanonymize_file(
    anonymized_file_path: str,
    mapping_file_path: str,
    mapping_passphrase: str | None = None,
    *,
    expected_source_sha256: str | None = None,
    encrypt_secret: str | None = None,
) -> tuple[str, str]:
    """Deanonymize a file using a mapping file.

    Restores original PII values from placeholders. Supports both current
    (placeholder -> original) and legacy (original -> placeholder) mapping
    directions via auto-detection.

    The implementation correctly handles variation placeholders (PERSON_1.v_2
    etc.) by falling back to the base placeholder's original value.

    After processing it also produces an audit statistics JSON file.

    Args:
        anonymized_file_path: Path to the previously anonymized document.
        mapping_file_path: Path to the JSON mapping file (plaintext or ``.enc``).
        mapping_passphrase: Required when the mapping file is encrypted.
        expected_source_sha256: Optional SHA-256 of the original source
            document. When set, an encrypted mapping locked to a different
            file is rejected (AAD mismatch).
        encrypt_secret: Secret used by the ``encrypt`` operator. Decrypts
            ``ENC1_`` tokens after placeholder restore.

    Returns:
        A tuple (deanonymized_file_path, stats_file_path).
        The stats file contains:
            - unused_mappings: placeholders in the map that did not appear in text
            - not_found_mappings: placeholders seen in text but missing from the map
    """
    with open(mapping_file_path, "r", encoding="utf-8") as f:
        raw_mapping = json.load(f)

    raw_mapping = load_mapping_payload(
        raw_mapping,
        mapping_passphrase,
        source_sha256=expected_source_sha256,
    )

    placeholder_to_original = mapping_to_placeholder_original(raw_mapping)

    sorted_placeholders = sorted(placeholder_to_original.keys(), key=len, reverse=True)
    tabular = is_tabular_path(anonymized_file_path)
    word = is_word_path(anonymized_file_path)
    native_pdf = Path(anonymized_file_path).suffix.lower() == ".pdf"
    deanonymized_text = ""

    if tabular:
        doc = load_table(anonymized_file_path)
        anonymized_text = flatten_table_for_review(doc, anonymized=True)
        used_placeholders: set[str] = set()
        for cell in iter_cells(doc):
            if not cell.search_text:
                continue
            restored, cell_used = restore_placeholders_in_text(
                cell.search_text, placeholder_to_original
            )
            if encrypt_secret:
                restored = restore_encrypt_tokens(restored, encrypt_secret)
            used_placeholders |= cell_used
            if doc.kind == "csv":
                restored = unneutralize_csv_equals(restored)
            if restored != cell.search_text:
                cell.search_text = restored
    elif word:
        word_doc = load_docx(anonymized_file_path)
        anonymized_text = flatten_docx_for_review(word_doc, anonymized=True)
        used_placeholders = set()
        for block in word_doc.blocks:
            if not block.search_text:
                continue
            restored, block_used = restore_placeholders_in_text(
                block.search_text, placeholder_to_original
            )
            if encrypt_secret:
                restored = restore_encrypt_tokens(restored, encrypt_secret)
            used_placeholders |= block_used
            if restored != block.search_text:
                write_block_text(block, restored)
    elif native_pdf:
        try:
            import pymupdf
        except ImportError as exc:
            raise ValueError("Deanonymizing a PDF requires pymupdf.") from exc
        opened = pymupdf.open(anonymized_file_path)
        try:
            anonymized_text = "\n".join(page.get_text() or "" for page in opened)
        finally:
            opened.close()
        deanonymized_text, used_placeholders = restore_placeholders_in_text(
            anonymized_text, placeholder_to_original
        )
        if encrypt_secret:
            deanonymized_text = restore_encrypt_tokens(
                deanonymized_text, encrypt_secret
            )
            extra: Dict[str, str] = {}
            for token in find_encrypt_tokens(anonymized_text):
                try:
                    extra[token] = decrypt_value(token, encrypt_secret)
                except ValueError:
                    continue
            placeholder_to_original.update(extra)
    else:
        with open(anonymized_file_path, "r", encoding="utf-8") as f:
            anonymized_text = f.read()
        deanonymized_text, used_placeholders = restore_placeholders_in_text(
            anonymized_text, placeholder_to_original
        )
        if encrypt_secret:
            deanonymized_text = restore_encrypt_tokens(
                deanonymized_text, encrypt_secret
            )

    # Gather stats
    all_placeholders_in_text = set(
        re.findall(r"[A-Z_]+_[0-9]+(?:\.v_[0-9]+)?", anonymized_text)
    )

    not_found_mappings = sorted(list(all_placeholders_in_text - used_placeholders))

    # Unused mappings: base placeholders that never occurred (neither base nor any variation)
    used_bases = {p.split(".v_")[0] for p in used_placeholders}
    unused_mappings = sorted([p for p in sorted_placeholders if p not in used_bases])

    anonymized_path = Path(anonymized_file_path)
    file_stem = anonymized_path.name.replace(f".anonymized{anonymized_path.suffix}", "")
    output_extension = anonymized_path.suffix

    deanonymized_dir = DEFAULT_DEANONYMIZED_DIR
    stats_dir = DEFAULT_STATS_DIR
    os.makedirs(deanonymized_dir, exist_ok=True)
    os.makedirs(stats_dir, exist_ok=True)

    deanonymized_file = f"{deanonymized_dir}/{file_stem}.deanonymized{output_extension}"
    if tabular:
        save_table(doc, deanonymized_file)
    elif word:
        save_docx(word_doc, deanonymized_file)
    elif native_pdf:
        write_deanonymized_pdf(
            anonymized_file_path, deanonymized_file, placeholder_to_original
        )
    else:
        with open(deanonymized_file, "w", encoding="utf-8") as f:
            f.write(deanonymized_text)

    stats_file = f"{stats_dir}/{file_stem}.deanonymization_stat.json"
    stats = {
        "anonymized_file": anonymized_file_path,
        "mapping_file": mapping_file_path,
        "deanonymized_file": deanonymized_file,
        "unused_mappings": unused_mappings,
        "not_found_mappings": not_found_mappings,
    }
    with open(stats_file, "w", encoding="utf-8") as f:
        json.dump(stats, f, indent=4)

    # Drop in-process references to recovered PII. Python strings cannot be
    # reliably overwritten; clearing the dicts is the explicit protocol.
    raw_mapping.clear()
    placeholder_to_original.clear()

    return deanonymized_file, stats_file

pdf_anonymizer_core.utils.consolidate_mapping(anonymized_text: str, mapping: Dict[str, str]) -> Tuple[str, Dict[str, str]]

Consolidates the mapping to ensure one-to-one correspondence and updates the text.

Parameters:

Name	Type	Description	Default
anonymized_text	str	The text with anonymized placeholders.	required
mapping	Dict[str, str]	The dictionary mapping placeholders to original PII.	required

Returns:

Type	Description
Tuple[str, Dict[str, str]]	A tuple containing the updated anonymized text and the consolidated mapping.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/utils.py

def consolidate_mapping(
    anonymized_text: str, mapping: Dict[str, str]
) -> Tuple[str, Dict[str, str]]:
    """
    Consolidates the mapping to ensure one-to-one correspondence and updates the text.

    Args:
        anonymized_text: The text with anonymized placeholders.
        mapping: The dictionary mapping placeholders to original PII.

    Returns:
        A tuple containing the updated anonymized text and the consolidated mapping.
    """
    # Invert the mapping to find duplicates
    value_to_keys: Dict[str, list] = {}
    for key, value in mapping.items():
        if value not in value_to_keys:
            value_to_keys[value] = []
        value_to_keys[value].append(key)

    consolidation_map = {}
    consolidated_mapping = mapping.copy()

    for value, keys in value_to_keys.items():
        if len(keys) > 1:
            canonical_key = keys[0]
            for key_to_replace in keys[1:]:
                consolidation_map[key_to_replace] = canonical_key
                if key_to_replace in consolidated_mapping:
                    del consolidated_mapping[key_to_replace]

    # Update the anonymized text in a single pass
    if consolidation_map:
        # Use word boundaries to avoid replacing parts of other words
        # Replace by longest keys first to ensure correct matching
        sorted_keys = sorted(consolidation_map.keys(), key=len, reverse=True)
        pattern = re.compile(
            r"\b(" + "|".join(re.escape(key) for key in sorted_keys) + r")\b"
        )
        anonymized_text = pattern.sub(
            lambda m: consolidation_map[m.group(1)], anonymized_text
        )

    return anonymized_text, consolidated_mapping

pdf_anonymizer_core.utils.save_results(full_anonymized_text: str, final_mapping: dict[str, str], file_path: str, mapping_passphrase: str | None = None, *, ephemeral_mapping: bool = False, entity_texts: Optional[Iterable[str]] = None, orig_to_written: Optional[Dict[str, str]] = None, output_pdf: bool = False, redact: bool = False) -> tuple[str, str]

Save the anonymized text and the mapping to files.

Parameters:

Name	Type	Description	Default
full_anonymized_text	str	The anonymized text.	required
final_mapping	dict[str, str]	Mapping written to the mapping file (CLI invert: placeholder → original).	required
file_path	str	The path to the original file.	required
mapping_passphrase	str | None	If set, write *.mapping.json.enc (AES-256-GCM + Argon2id) instead of plaintext JSON.	None
ephemeral_mapping	bool	If true, never write a mapping file. The second return value is an empty string. The caller already holds final_mapping in memory.	False
entity_texts	Optional[Iterable[str]]	Detected entity["text"] values used to apply the mapping. Required for table and Word paths; ignored for text.	None
orig_to_written	Optional[Dict[str, str]]	Engine original → written map used to re-apply on tables, Word files, and native PDF writes. Required for colliding mask/generalize/fake forms; when omitted, mapping_to_original_to_written(final_mapping) is used.	None
output_pdf	bool	Also write a sanitized .anonymized.pdf. Markdown is still written. Only valid for .pdf inputs.	False
redact	bool	Irreversible native PDF: black boxes, no stand-in text. Implies output_pdf.	False

Returns:

Type	Description
str	tuple[str, str]: The paths to the anonymized text file and the mapping
str	file. The mapping path is "" when ephemeral_mapping is true.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/utils.py

def save_results(
    full_anonymized_text: str,
    final_mapping: dict[str, str],
    file_path: str,
    mapping_passphrase: str | None = None,
    *,
    ephemeral_mapping: bool = False,
    entity_texts: Optional[Iterable[str]] = None,
    orig_to_written: Optional[Dict[str, str]] = None,
    output_pdf: bool = False,
    redact: bool = False,
) -> tuple[str, str]:
    """
    Save the anonymized text and the mapping to files.

    Args:
        full_anonymized_text (str): The anonymized text.
        final_mapping (dict[str, str]): Mapping written to the mapping file
            (CLI invert: placeholder → original).
        file_path (str): The path to the original file.
        mapping_passphrase: If set, write ``*.mapping.json.enc`` (AES-256-GCM
            + Argon2id) instead of plaintext JSON.
        ephemeral_mapping: If true, never write a mapping file. The second
            return value is an empty string. The caller already holds
            ``final_mapping`` in memory.
        entity_texts: Detected ``entity["text"]`` values used to apply the
            mapping. Required for table and Word paths; ignored for text.
        orig_to_written: Engine original → written map used to re-apply on
            tables, Word files, and native PDF writes. Required for colliding
            mask/generalize/fake forms; when omitted,
            ``mapping_to_original_to_written(final_mapping)`` is used.
        output_pdf: Also write a sanitized ``.anonymized.pdf``. Markdown is
            still written. Only valid for ``.pdf`` inputs.
        redact: Irreversible native PDF: black boxes, no stand-in text.
            Implies ``output_pdf``.

    Returns:
        tuple[str, str]: The paths to the anonymized text file and the mapping
        file. The mapping path is ``""`` when ``ephemeral_mapping`` is true.
    """
    original_path = Path(file_path)
    file_stem = original_path.stem
    file_extension = original_path.suffix.lower()

    anonymized_dir = DEFAULT_ANONYMIZED_DIR
    mappings_dir = DEFAULT_MAPPINGS_DIR
    os.makedirs(anonymized_dir, exist_ok=True)

    if file_extension == ".pdf":
        output_extension = ".md"
    else:
        output_extension = file_extension

    anonymized_output_file = (
        f"{anonymized_dir}/{file_stem}.anonymized{output_extension}"
    )
    structured = is_tabular_path(file_path) or is_word_path(file_path)
    if structured:
        if entity_texts is None:
            raise ValueError(
                "save_results() on a table or Word document requires "
                "entity_texts= (the same entity['text'] list the engine applied)."
            )
        apply_map = (
            orig_to_written
            if orig_to_written is not None
            else mapping_to_original_to_written(final_mapping)
        )
        texts = [text for text in entity_texts if text]
        if texts and not any(text in apply_map for text in texts):
            raise ValueError(
                "save_results() entity_texts are not keys of orig_to_written. "
                "Pass the engine original→written map as orig_to_written=."
            )
        if is_word_path(file_path):
            write_anonymized_docx(file_path, anonymized_output_file, apply_map, texts)
        else:
            write_anonymized_table(file_path, anonymized_output_file, apply_map, texts)
    else:
        with open(anonymized_output_file, "w", encoding="utf-8") as f:
            f.write(full_anonymized_text)

    layout_source, layout_words = take_pdf_layout()
    if layout_words and layout_source:
        try:
            same_source = Path(layout_source).resolve() == original_path.resolve()
        except OSError:
            same_source = layout_source == file_path
        if same_source:
            write_layout_sidecar(anonymized_output_file, layout_source, layout_words)

    primary_output = anonymized_output_file
    if output_pdf or redact:
        if file_extension != ".pdf":
            raise ValueError(OUTPUT_PDF_NOT_PDF_MESSAGE)
        apply_map = (
            orig_to_written
            if orig_to_written is not None
            else mapping_to_original_to_written(final_mapping)
        )
        texts = (
            [text for text in entity_texts if text]
            if entity_texts is not None
            else list(apply_map.keys())
        )
        pdf_output = f"{anonymized_dir}/{file_stem}.anonymized.pdf"
        write_anonymized_pdf(file_path, pdf_output, apply_map, texts, redact=redact)
        primary_output = pdf_output

    if ephemeral_mapping:
        return primary_output, ""

    source_digest = ""
    if original_path.is_file():
        source_digest = sha256_file(original_path)

    persist_mapping = mapping_without_encrypt_plaintexts(final_mapping)
    if mapping_passphrase:
        mapping_file = f"{mappings_dir}/{file_stem}.mapping.json.enc"
        payload = encrypt_mapping(
            persist_mapping,
            mapping_passphrase,
            source_sha256=source_digest or None,
        )
        write_private_json(mapping_file, payload)
    else:
        mapping_file = f"{mappings_dir}/{file_stem}.mapping.json"
        # Persist mapping as placeholder -> original for correct deanonymization.
        # Encrypt tokens are omitted so a leaked map does not hold those originals.
        write_private_json(mapping_file, persist_mapping)

    return primary_output, mapping_file

pdf_anonymizer_api.app.create_app()

Build the FastAPI app.

Source code in packages/pdf-anonymizer-api/src/pdf_anonymizer_api/app.py

def create_app():
    """Build the FastAPI app."""
    from fastapi import FastAPI, HTTPException

    application = FastAPI(
        title="PDF Anonymizer",
        description=(
            "Local HTTP wrapper around pdf-anonymizer-core. "
            "No authentication. Bind to localhost or a compose network."
        ),
        version="0.26.0",
    )

    @application.get("/health")
    def health() -> Dict[str, str]:
        return {"status": "ok", "version": "0.26.0"}

    @application.post("/anonymize")
    def anonymize(body: AnonymizeBody) -> Dict[str, Any]:
        try:
            return anonymize_text_request(
                body.text,
                use_llm=body.use_llm,
                use_ner=body.use_ner,
                min_confidence=body.min_confidence,
                keep_list=body.keep_list,
                deny_list=body.deny_list,
                operators=body.operators,
                seed_mapping=body.seed_mapping,
                fake_secret=body.fake_secret,
                encrypt_secret=body.encrypt_secret,
                model_name=body.model_name,
                prompt_name=body.prompt_name,
                anonymized_entities=body.anonymized_entities,
                countries=body.countries,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    @application.post("/deanonymize")
    def deanonymize(body: DeanonymizeBody) -> Dict[str, Any]:
        try:
            return deanonymize_text_request(
                body.text,
                body.mapping,
                encrypt_secret=body.encrypt_secret,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    @application.post("/verify")
    def verify(body: TextBody) -> Dict[str, Any]:
        try:
            return verify_anonymized_text(
                body.text,
                regex_patterns=filter_regex_patterns(body.countries),
                use_llm=body.use_llm,
                model_name=body.model_name,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    @application.post("/report")
    def report(body: TextBody) -> Dict[str, Any]:
        return assess_linkage_risk(body.text)

    return application

pdf_anonymizer_api.app.anonymize_text_request(text: str, *, use_llm: bool = False, use_ner: bool = False, min_confidence: float = 0.0, keep_list: Optional[List[str]] = None, deny_list: Optional[List[str]] = None, operators: Optional[Dict[str, str]] = None, seed_mapping: Optional[Dict[str, str]] = None, fake_secret: Optional[str] = None, encrypt_secret: Optional[str] = None, model_name: Optional[str] = None, prompt_name: str = 'simple', anonymized_entities: Optional[List[str]] = None, countries: Optional[List[str]] = None) -> Dict[str, Any]

Run the same engine as the CLI on an in-memory string.

Source code in packages/pdf-anonymizer-api/src/pdf_anonymizer_api/app.py

def anonymize_text_request(
    text: str,
    *,
    use_llm: bool = False,
    use_ner: bool = False,
    min_confidence: float = 0.0,
    keep_list: Optional[List[str]] = None,
    deny_list: Optional[List[str]] = None,
    operators: Optional[Dict[str, str]] = None,
    seed_mapping: Optional[Dict[str, str]] = None,
    fake_secret: Optional[str] = None,
    encrypt_secret: Optional[str] = None,
    model_name: Optional[str] = None,
    prompt_name: str = "simple",
    anonymized_entities: Optional[List[str]] = None,
    countries: Optional[List[str]] = None,
) -> Dict[str, Any]:
    """Run the same engine as the CLI on an in-memory string."""
    if len(text) > MAX_TEXT_CHARS:
        raise ValueError(
            f"text is {len(text):,} characters; the HTTP limit is {MAX_TEXT_CHARS:,}."
        )
    template = _PROMPTS.get(prompt_name)
    if template is None:
        raise ValueError(
            f"Unknown prompt_name {prompt_name!r}. Use simple, detailed, or hipaa."
        )
    if not 0.0 <= min_confidence <= 1.0:
        raise ValueError("min_confidence must be between 0 and 1.")

    collected = collect_entities_from_chunks(
        [text],
        prompt_template=template,
        model_name=model_name or "gemini-2.5-flash",
        regex_patterns=filter_regex_patterns(countries),
        max_retries=3,
        base_retry_delay=1.0,
        max_retry_delay=10.0,
        use_llm=use_llm,
        use_ner=use_ner,
    )
    entities = finalize_entities(
        collected,
        text,
        anonymized_entities=anonymized_entities,
        keep_list=keep_list,
        deny_list=deny_list,
        min_confidence=min_confidence,
        seed_mapping=seed_mapping,
    )
    mapping = build_mapping(
        entities,
        seed_mapping=seed_mapping,
        operators=operators,
        fake_secret=fake_secret,
        encrypt_secret=encrypt_secret,
    )
    anonymized = text
    if entities:
        anonymized = replace_entities(
            text, (entity["text"] for entity in entities), mapping
        )
    return {
        "anonymized_text": anonymized,
        "mapping": mapping,
        "entities": [_public_entity(entity) for entity in entities],
    }

---

Configuration & Models

pdf_anonymizer_core.conf

Central configuration, defaults, profiles, and model/provider enums.

This module defines: - Default constants and directories used by the anonymizer. - Built-in ConfigProfiles (best-quality, best-speed, best-cost, regex-only) that bundle model, prompt, chunk size, retries, and whether to call an LLM. - The AppConfig Pydantic model. - get_config_for_profile() helper (used heavily by the CLI). - Legacy enums (PromptEnum, ModelName, etc.) for compatibility and dynamic provider/model resolution.

EntityProfile

Bases: str, Enum

Named type + operator bundles. Separate from quality/speed ConfigProfile.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/conf.py

class EntityProfile(str, Enum):
    """Named type + operator bundles. Separate from quality/speed ConfigProfile."""

    HIPAA_SAFE_HARBOR = "hipaa-safe-harbor"

filter_regex_patterns(countries: Optional[Iterable[str]] = None, patterns: Optional[Dict[str, str]] = None) -> Dict[str, str]

Keep universal patterns plus national-ID patterns for countries.

Universal keys (EMAIL, IBAN, CREDIT_CARD, VIN, ...) always stay. Country keys (SSN_US, NINO_GB, PESEL_PL, legacy SSN, ...) stay only when their ISO-2 code is in countries.

countries is a list of ISO-2 codes such as ["US", "GB"]. None or an empty list returns a copy of all patterns. Unknown codes raise ValueError.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/conf.py

def filter_regex_patterns(
    countries: Optional[Iterable[str]] = None,
    patterns: Optional[Dict[str, str]] = None,
) -> Dict[str, str]:
    """Keep universal patterns plus national-ID patterns for ``countries``.

    Universal keys (EMAIL, IBAN, CREDIT_CARD, VIN, ...) always stay.
    Country keys (SSN_US, NINO_GB, PESEL_PL, legacy SSN, ...) stay only when
    their ISO-2 code is in ``countries``.

    ``countries`` is a list of ISO-2 codes such as ``["US", "GB"]``.
    ``None`` or an empty list returns a copy of all ``patterns``.
    Unknown codes raise ``ValueError``.
    """
    source = DEFAULT_REGEX_PATTERNS if patterns is None else patterns
    if not countries:
        return dict(source)

    wanted = {code.strip().upper() for code in countries if code and str(code).strip()}
    if not wanted:
        return dict(source)

    unknown = sorted(wanted - COUNTRY_PATTERN_SUFFIXES)
    if unknown:
        raise ValueError(
            "Unknown country code(s): "
            + ", ".join(unknown)
            + ". Use ISO-2 codes such as US, GB, FR."
        )

    filtered: Dict[str, str] = {}
    for key, pattern in source.items():
        country = pattern_country(key)
        if country is None or country in wanted:
            filtered[key] = pattern
    return filtered

get_config_for_profile(profile: ConfigProfile, model_name: Optional[str] = None, prompt_name: Optional[str] = None, chunk_size: Optional[int] = None, chunk_overlap: Optional[int] = None, countries: Optional[Iterable[str]] = None) -> AppConfig

Return an AppConfig populated from one of the built-in profiles.

Profiles provide convenient quality/speed/cost presets. Any of the scalar overrides (model_name, prompt_name, chunk_size, chunk_overlap) take precedence over the profile defaults.

The other fields (retries, delays, cache settings, directories) always come from the chosen profile.

Parameters:

Name	Type	Description	Default
profile	ConfigProfile	One of ConfigProfile.BEST_QUALITY, BEST_SPEED, BEST_COST, or REGEX_ONLY.	required
model_name	Optional[str]	Optional override for the model (string or provider/model).	None
prompt_name	Optional[str]	Optional override ("simple" or "detailed").	None
chunk_size	Optional[int]	Optional override for characters_to_anonymize / chunk_size.	None
chunk_overlap	Optional[int]	Optional override for chunk overlap.	None
countries	Optional[Iterable[str]]	Optional ISO-2 codes that limit national-ID regexes (universal patterns always stay).	None

Returns:

Type	Description
AppConfig	A fully populated AppConfig instance ready to drive anonymize_file
AppConfig	(or to be passed through configure_cache, etc.).

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/conf.py

def get_config_for_profile(
    profile: ConfigProfile,
    model_name: Optional[str] = None,
    prompt_name: Optional[str] = None,
    chunk_size: Optional[int] = None,
    chunk_overlap: Optional[int] = None,
    countries: Optional[Iterable[str]] = None,
) -> AppConfig:
    """Return an AppConfig populated from one of the built-in profiles.

    Profiles provide convenient quality/speed/cost presets. Any of the
    scalar overrides (model_name, prompt_name, chunk_size, chunk_overlap)
    take precedence over the profile defaults.

    The other fields (retries, delays, cache settings, directories) always
    come from the chosen profile.

    Args:
        profile: One of ConfigProfile.BEST_QUALITY, BEST_SPEED, BEST_COST,
            or REGEX_ONLY.
        model_name: Optional override for the model (string or provider/model).
        prompt_name: Optional override ("simple" or "detailed").
        chunk_size: Optional override for characters_to_anonymize / chunk_size.
        chunk_overlap: Optional override for chunk overlap.
        countries: Optional ISO-2 codes that limit national-ID regexes
            (universal patterns always stay).

    Returns:
        A fully populated AppConfig instance ready to drive anonymize_file
        (or to be passed through configure_cache, etc.).
    """
    profile_defaults = PROFILE_CONFIGS[profile]

    resolved_prompt_name = prompt_name or profile_defaults["prompt_name"]
    if isinstance(resolved_prompt_name, Enum):
        resolved_prompt_name = resolved_prompt_name.value

    return AppConfig(
        model_name=model_name or profile_defaults["model_name"],
        prompt_name=resolved_prompt_name,
        chunk_size=chunk_size
        if chunk_size is not None
        else profile_defaults["chunk_size"],
        chunk_overlap=chunk_overlap
        if chunk_overlap is not None
        else profile_defaults["chunk_overlap"],
        max_retries=profile_defaults["max_retries"],
        base_retry_delay=profile_defaults["base_retry_delay"],
        max_retry_delay=profile_defaults["max_retry_delay"],
        regex_patterns=filter_regex_patterns(countries),
        use_llm=profile_defaults.get("use_llm", True),
        enable_cache=profile_defaults.get("enable_cache", True),
    )

pattern_country(key: str) -> Optional[str]

Return the ISO-2 country for a pattern key, or None if it is universal.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/conf.py

def pattern_country(key: str) -> Optional[str]:
    """Return the ISO-2 country for a pattern key, or None if it is universal."""
    upper = key.upper()
    if upper in _PATTERN_COUNTRY_ALIASES:
        return _PATTERN_COUNTRY_ALIASES[upper]
    suffix = upper.rsplit("_", 1)[-1]
    if suffix in COUNTRY_PATTERN_SUFFIXES:
        return suffix
    return None

---

Prompts

The package ships two ready-to-use prompt templates.

pdf_anonymizer_core.prompts.detailed

Detailed (higher-quality) PII identification prompt.

This prompt instructs the LLM to: - Use deep contextual understanding - Return base_form for coreference handling - Identify a rich set of entity types including JOB_TITLE, ID, DATE (birthdates only), and INDIRECT (phrases that point to one person without saying their name) - Explicitly handle variations, possessives, and implied identity

Use this prompt (via pdf_anonymizer_core.prompts.detailed.prompt_template) when you want maximum accuracy (pairs well with BEST_QUALITY profile).

pdf_anonymizer_core.prompts.simple

Simple (faster/cheaper) PII identification prompt.

This prompt is a lightweight version that asks the model only for basic "text" + "type" entities. It is faster and cheaper but does not request base_form or the full entity type list.

Use this prompt (via pdf_anonymizer_core.prompts.simple.prompt_template) for speed/cost sensitive workloads (pairs well with BEST_SPEED or BEST_COST).

pdf_anonymizer_core.prompts.hipaa

HIPAA Safe Harbor aid prompt.

Asks for the identifier classes that apply to text (names, small-area geography, dates about a person, phones, emails, IDs, URLs, IPs, and so on).

This is a helper for coverage. It is not a legal determination that the output meets the HIPAA Safe Harbor standard.

---

Detection, operators, and reports

pdf_anonymizer_core.regex_ner

Fast first-stage Named Entity Recognition using regular expressions (RE2 engine).

The hybrid pipeline (core.anonymize_file) ALWAYS runs this regex NER pass before the (more expensive) LLM stage. Matches are later merged/deduped with LLM results via a priority system in core.py (EMAIL, CREDIT_CARD, IBAN, crypto, SSN_* etc. receive elevated priority).

ENGINE

This module uses the RE2 engine (package "google-re2") instead of Python's stdlib re. Benefits: * Linear-time matching (O(length) worst-case). Immune to catastrophic backtracking (ReDoS) even with complex patterns or adversarial input. * Predictable performance on large PDFs / logs / code dumps. * Same fundamental API surface: compile / finditer / match.group(0) / error.

RE2 LIMITATIONS (patterns here are written to be compliant)

• No look-ahead / look-behind assertions ((?=...), (?!...), (?<=...)).
• Limited support for some advanced Python re features (no recursive patterns, limited backreferences). All DEFAULT patterns and documented examples stay within the safe subset.
•  word boundaries are supported and used for many PII tokens.

COUNTRY PARTITIONING & EXTENSIVE PII COVERAGE

Patterns are defined in pdf_anonymizer_core.conf.DEFAULT_REGEX_PATTERNS and are partitioned by country using ISO-2 suffixes (SSN_US, NINO_GB, INSEE_FR, AADHAAR_IN, RESIDENT_ID_CN, DNI_ES, CODICE_FISCALE_IT, ...).

Mandatory countries covered: USA, Canada, UK, Spain, Italy, France, India, China. 30+ total countries supported via dedicated national/tax/driver/VAT/business ID patterns + universal patterns (IBAN covers most EU+ countries, credit cards, crypto, VIN, MAC, etc. are global).

Supported categories (non-exhaustive): EMAIL, PHONE, URL, IPV4/IPV6, MAC, CREDIT_CARD, CURRENCY_AMOUNT, CRYPTO_BTC / CRYPTO_ETH, IBAN, BIC_SWIFT, VIN, DATE_ISO, SSN / SSN_US / SIN_CA / NINO_GB / INSEE_FR / AADHAAR_IN / RESIDENT_ID_CN, EIN_US, VAT_, PASSPORT_, DRIVERS_LICENSE_, MEDICAL_ , business regs, CURP_MX, CPF_BR, TFN_AU, NRIC_SG, HKID_HK, PESEL_PL, etc.

CUSTOMISATION

You can supply a completely custom regex_patterns: Dict[str, str] (type -> pattern) when calling anonymize_file(). Keys become the emitted entity TYPE (upper-cased) and are used for placeholder generation (EMAIL_1, IBAN_7, DRIVERS_LICENSE_CA_2, ...). Only the keys you provide are used; there is no automatic merging with defaults unless you build your dict from DEFAULT_REGEX_PATTERNS.

The function is intentionally tiny. It only does structural scanning, then runs a cheap checksum when one exists (Luhn, IBAN mod-97, VIN check digit, and a few national-ID checks). Failures are kept and relabeled TYPE_LIKE (for example IBAN_LIKE) so a mistyped number is still hidden. Semantic disambiguation, name coreference, and hard-to-regex PII remain the responsibility of the LLM stage that always follows.

extract_entities_via_regex(text: str, patterns: Dict[str, str]) -> List[EntityDict]

Scans the text for PII using pre-configured regular expressions (RE2 engine). Matches that fail a registered checksum (see validators.py) are kept and labeled <TYPE>_LIKE (for example IBAN_LIKE).

Parameters:

Name	Type	Description	Default
text	str	Input text to analyze.	required
patterns	Dict[str, str]	Dictionary mapping entity type strings (e.g. "IBAN", "SSN_US", "CRYPTO_ETH") to RE2-compatible regex pattern strings.	required

Returns:

Type	Description
List[EntityDict]	A list of EntityDict representing identified PII. "type" is always
List[EntityDict]	upper-cased. "base_form" currently equals the matched text (core
List[EntityDict]	consolidation may later promote variations to a longer base form).
List[EntityDict]	Each hit also includes chunk-local start / end offsets.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/regex_ner.py

def extract_entities_via_regex(text: str, patterns: Dict[str, str]) -> List[EntityDict]:
    """
    Scans the text for PII using pre-configured regular expressions (RE2 engine).
    Matches that fail a registered checksum (see validators.py) are kept and
    labeled ``<TYPE>_LIKE`` (for example ``IBAN_LIKE``).

    Args:
        text: Input text to analyze.
        patterns: Dictionary mapping entity type strings (e.g. "IBAN", "SSN_US",
            "CRYPTO_ETH") to RE2-compatible regex pattern strings.

    Returns:
        A list of EntityDict representing identified PII. "type" is always
        upper-cased. "base_form" currently equals the matched text (core
        consolidation may later promote variations to a longer base form).
        Each hit also includes chunk-local ``start`` / ``end`` offsets.
    """
    entities: List[EntityDict] = []

    for entity_type, pattern_str in patterns.items():
        try:
            compiled_pattern = re.compile(pattern_str)
            for match in compiled_pattern.finditer(text):
                matched_text = match.group(0)
                # Filter out empty or whitespace-only matches
                if not matched_text.strip():
                    continue

                entity_type_upper = entity_type.upper()
                if has_checksum(entity_type_upper) and not passes_checksum(
                    entity_type_upper, matched_text
                ):
                    like = like_type(entity_type_upper)
                    logging.debug(
                        "Checksum failed for %s %r; labeling as %s",
                        entity_type_upper,
                        matched_text,
                        like,
                    )
                    entity_type_upper = like
                    score = 0.55
                elif has_checksum(entity_type_upper):
                    score = 0.95
                else:
                    score = 0.85

                entities.append(
                    {
                        "text": matched_text,
                        "type": entity_type_upper,
                        "base_form": matched_text,
                        "start": match.start(),
                        "end": match.end(),
                        "score": score,
                        "source": "regex",
                    }
                )
        except re.error as e:
            logging.error(f"Invalid regex pattern configured for {entity_type}: {e}")

    return entities

pdf_anonymizer_core.validators

Cheap, unambiguous checksums for structured regex hits.

The regex stage is structural on purpose (RE2 cannot do Luhn, IBAN mod-97, etc.). After a match, this module checks the extra digit when one exists.

• Check passes: keep the real type (IBAN).
• Check fails: keep the text, but relabel as IBAN_LIKE so a mistyped number is still hidden. Never drop the hit.
• No registered check: accept the type unchanged.

Only attach a check when it is cheap and unambiguous. Do not invent rules for identifiers that have none (most SSNs, many passports, most VAT numbers).

has_checksum(entity_type: str) -> bool

Return True if this type has a registered extra-digit check.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def has_checksum(entity_type: str) -> bool:
    """Return True if this type has a registered extra-digit check."""
    return entity_type.upper() in CHECKSUM_VALIDATORS

like_type(entity_type: str) -> str

IBAN -> IBAN_LIKE.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def like_type(entity_type: str) -> str:
    """``IBAN`` -> ``IBAN_LIKE``."""
    return f"{entity_type.upper()}{LIKE_SUFFIX}"

luhn_ok(digits: str) -> bool

Return True if digits (0-9 only) passes the Luhn check.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def luhn_ok(digits: str) -> bool:
    """Return True if ``digits`` (0-9 only) passes the Luhn check."""
    if not digits or not digits.isdigit():
        return False
    total = 0
    # Double every second digit from the right.
    reverse = digits[::-1]
    for i, ch in enumerate(reverse):
        n = ord(ch) - 48
        if i % 2 == 1:
            n *= 2
            if n > 9:
                n -= 9
        total += n
    return total % 10 == 0

parent_type(entity_type: str) -> str

IBAN_LIKE -> IBAN; IBAN -> IBAN.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def parent_type(entity_type: str) -> str:
    """``IBAN_LIKE`` -> ``IBAN``; ``IBAN`` -> ``IBAN``."""
    upper = entity_type.upper()
    if upper.endswith(LIKE_SUFFIX) and len(upper) > len(LIKE_SUFFIX):
        return upper[: -len(LIKE_SUFFIX)]
    return upper

passes_checksum(entity_type: str, text: str) -> bool

Return True if text has no check, or if its check succeeds.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def passes_checksum(entity_type: str, text: str) -> bool:
    """Return True if ``text`` has no check, or if its check succeeds."""
    validator = CHECKSUM_VALIDATORS.get(entity_type.upper())
    if validator is None:
        return True
    return validator(text)

type_matches_filter(entity_type: str, allowed: list[str] | set[str]) -> bool

True if the type is listed, is a _LIKE sibling, or matches a prefix.

A listed DRIVERS_LICENSE matches DRIVERS_LICENSE_US. A listed DATE matches DATE_ISO.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def type_matches_filter(entity_type: str, allowed: list[str] | set[str]) -> bool:
    """True if the type is listed, is a ``_LIKE`` sibling, or matches a prefix.

    A listed ``DRIVERS_LICENSE`` matches ``DRIVERS_LICENSE_US``.
    A listed ``DATE`` matches ``DATE_ISO``.
    """
    allowed_upper = {item.upper() for item in allowed}
    upper = entity_type.upper()
    parent = parent_type(upper)
    if upper in allowed_upper or parent in allowed_upper:
        return True
    for item in allowed_upper:
        if upper.startswith(item + "_") or parent.startswith(item + "_"):
            return True
        if upper.endswith("_" + item) or parent.endswith("_" + item):
            return True
    return False

verhoeff_ok(digits: str) -> bool

Return True if digits (0-9 only) passes the Verhoeff check.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/validators.py

def verhoeff_ok(digits: str) -> bool:
    """Return True if ``digits`` (0-9 only) passes the Verhoeff check."""
    if not digits or not digits.isdigit():
        return False
    checksum = 0
    for i, ch in enumerate(reversed(digits)):
        checksum = _VERHOEFF_D[checksum][_VERHOEFF_P[i % 8][ord(ch) - 48]]
    return checksum == 0

pdf_anonymizer_core.operators

Per-type operators for how a found value is written into the masked file.

Default is replace (typed stand-ins such as PERSON_1). Other operators change what the reader sees; the mapping still records original → written form so deanonymize can reverse when the written form is unique.

apply_operator(original: str, entity_type: str, placeholder: str, operator: str, base_form: Optional[str] = None, secret: str = '', encrypt_secret: str = '') -> str

Return the string to write in place of original.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def apply_operator(
    original: str,
    entity_type: str,
    placeholder: str,
    operator: str,
    base_form: Optional[str] = None,
    secret: str = "",
    encrypt_secret: str = "",
) -> str:
    """Return the string to write in place of ``original``."""
    if operator == OPERATOR_REPLACE:
        return placeholder
    if operator == OPERATOR_MASK:
        return mask_value(original, entity_type)
    if operator == OPERATOR_HASH:
        return hash_value(original)
    if operator == OPERATOR_GENERALIZE:
        return generalize_value(original, entity_type)
    if operator == OPERATOR_SHIFT:
        return shift_date_value(original, base_form or original)
    if operator == OPERATOR_FAKE:
        return fake_value(original, entity_type, base_form or original, secret)
    if operator == OPERATOR_ENCRYPT:
        return encrypt_value(original, encrypt_secret)
    return placeholder

decrypt_value(token: str, secret: str) -> str

Reverse encrypt_value. Raises ValueError on a bad token or secret.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def decrypt_value(token: str, secret: str) -> str:
    """Reverse ``encrypt_value``. Raises ValueError on a bad token or secret."""
    if not secret:
        raise ValueError(ENCRYPT_SECRET_MESSAGE)
    raw = token.strip()
    if not looks_like_encrypt_token(raw):
        raise ValueError("Not an encrypt token.")
    payload = raw[len(ENCRYPT_PREFIX) :]
    pad = "=" * ((4 - len(payload) % 4) % 4)
    try:
        blob = base64.urlsafe_b64decode(payload + pad)
    except (ValueError, OSError) as exc:
        raise ValueError("encrypt token is not valid base64.") from exc
    if len(blob) < 12 + 16:
        raise ValueError("encrypt token is truncated.")
    key = _encrypt_key(secret)
    try:
        plain = AESGCM(key).decrypt(blob[:12], blob[12:], ENCRYPT_AAD)
    except Exception as exc:
        raise ValueError("Could not decrypt token. Check --encrypt-secret.") from exc
    return plain.decode("utf-8")

encrypt_value(original: str, secret: str) -> str

AES-256-GCM token. Same secret + same text always yields the same token.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def encrypt_value(original: str, secret: str) -> str:
    """AES-256-GCM token. Same secret + same text always yields the same token."""
    if not secret:
        raise ValueError(ENCRYPT_SECRET_MESSAGE)
    key = _encrypt_key(secret)
    nonce = hmac.new(key, original.encode("utf-8"), hashlib.sha256).digest()[:12]
    cipher = AESGCM(key).encrypt(nonce, original.encode("utf-8"), ENCRYPT_AAD)
    blob = nonce + cipher
    token = base64.urlsafe_b64encode(blob).decode("ascii").rstrip("=")
    return f"{ENCRYPT_PREFIX}{token}"

fake_value(original: str, entity_type: str, base_form: str, secret: str = '') -> str

Stable fake in the same shape family. Same base_form always matches.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def fake_value(
    original: str,
    entity_type: str,
    base_form: str,
    secret: str = "",
) -> str:
    """Stable fake in the same shape family. Same base_form always matches."""
    from faker import Faker

    kind = parent_type(entity_type)
    fake = Faker("en_US")
    fake.seed_instance(_fake_seed(secret, kind, base_form))

    if kind == "PERSON":
        return fake.name()
    if kind == "EMAIL":
        return fake.safe_email()
    if kind == "PHONE" or kind == "FAX":
        return f"555-010{fake.random_int(0, 9)}"
    if kind == "LOCATION":
        return fake.city()
    if kind == "ADDRESS":
        return fake.street_address()
    if kind == "ORGANIZATION":
        return fake.company()
    if kind == "JOB_TITLE":
        return fake.job()
    if kind.startswith("DATE"):
        parsed = _parse_date(original)
        year = parsed.year if parsed else fake.random_int(1980, 2020)
        return fake.date_between(
            start_date=date(year, 1, 1), end_date=date(year, 12, 28)
        ).isoformat()
    if kind == "AGE":
        return str(fake.random_int(21, 80))
    if kind in {"CREDIT_CARD"} or kind.startswith("CREDIT_CARD"):
        body = f"{_fake_seed(secret, kind, base_form):016d}"[:12]
        return f"4111 {body[0:4]} {body[4:8]} {body[8:12]}"
    if kind.startswith("SSN") or kind == "SIN_CA":
        tail = f"{_fake_seed(secret, kind, base_form) % 10000:04d}"
        return f"000-00-{tail}"
    if kind == "IBAN" or kind.endswith("IBAN"):
        tail = f"{_fake_seed(secret, kind, base_form) % 10**10:010d}"
        return f"GB00FAKE{tail}"
    if kind in {"IPV4_ADDRESS", "IP_ADDRESS"}:
        return f"203.0.113.{_fake_seed(secret, kind, base_form) % 254 + 1}"
    if kind == "URL":
        return f"https://example.test/{fake.slug()}"
    if kind == "VIN":
        return f"1HGCM8263{fake.random_int(0, 9)}A{fake.random_int(100000, 999999)}"
    return fake.word().title()

generalize_value(original: str, entity_type: str) -> str

Coarser value: year, ZIP3, or age band. Falls back to the original shape.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def generalize_value(original: str, entity_type: str) -> str:
    """Coarser value: year, ZIP3, or age band. Falls back to the original shape."""
    kind = parent_type(entity_type)
    parsed = _parse_date(original)
    if parsed is not None and (
        kind.startswith("DATE") or kind == "DATE" or _ISO_DATE.match(original.strip())
    ):
        return str(parsed.year)

    zip_match = _US_ZIP.match(original.strip())
    if zip_match:
        return zip_match.group(1)[:3] + "**"

    if kind == "ADDRESS":
        return _ZIP_IN_TEXT.sub(lambda m: m.group(1)[:3] + "**", original)

    if kind == "AGE" or (
        kind in {"", "ID"}
        and original.strip().isdigit()
        and 1 <= int(original.strip()) <= 120
    ):
        return _age_band(int(original.strip()))

    if original.strip().isdigit() and kind.startswith("DATE"):
        # year-only already
        return original.strip()

    # Unknown shape: keep year if it looks like a date, else leave a coarse token
    if parsed is not None:
        return str(parsed.year)
    return original

mapping_without_encrypt_plaintexts(mapping: Dict[str, str]) -> Dict[str, str]

Drop encrypt tokens so a leaked map does not hold those originals.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def mapping_without_encrypt_plaintexts(mapping: Dict[str, str]) -> Dict[str, str]:
    """Drop encrypt tokens so a leaked map does not hold those originals."""
    return {
        key: value
        for key, value in mapping.items()
        if not looks_like_encrypt_token(key) and not looks_like_encrypt_token(value)
    }

mask_value(original: str, entity_type: str) -> str

Keep a little shape; hide the rest.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def mask_value(original: str, entity_type: str) -> str:
    """Keep a little shape; hide the rest."""
    kind = parent_type(entity_type)
    digits = _DIGITS.findall(original)
    if kind in {"CREDIT_CARD", "SSN", "SSN_US", "SIN_CA", "MEDICAL_NPI_US"} or (
        kind.startswith("SSN")
    ):
        return _mask_keep_last(original, 4)
    if kind == "IBAN" or kind.endswith("IBAN"):
        return _mask_keep_last(original, 4)
    if kind == "PHONE":
        return _mask_keep_last(original, 4)
    if kind == "EMAIL":
        return _mask_email(original)
    if digits and len(digits) >= 4:
        return _mask_keep_last(original, 4)
    return "".join("*" if ch.isalnum() else ch for ch in original) or "****"

operator_for_type(entity_type: str, operators: Optional[Dict[str, str]]) -> str

Resolve the operator for a type. CREDIT_CARD_LIKE follows CREDIT_CARD.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def operator_for_type(entity_type: str, operators: Optional[Dict[str, str]]) -> str:
    """Resolve the operator for a type. ``CREDIT_CARD_LIKE`` follows ``CREDIT_CARD``."""
    if not operators:
        return OPERATOR_REPLACE
    upper = entity_type.upper()
    if upper in operators:
        return operators[upper]
    parent = parent_type(upper)
    if parent in operators:
        return operators[parent]
    if upper.startswith("DATE") and "DATE" in operators:
        return operators["DATE"]
    return OPERATOR_REPLACE

parse_operator_specs(specs: Optional[Iterable[str]]) -> Dict[str, str]

Parse TYPE=operator strings. Unknown operators raise ValueError.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def parse_operator_specs(specs: Optional[Iterable[str]]) -> Dict[str, str]:
    """Parse ``TYPE=operator`` strings. Unknown operators raise ValueError."""
    result: Dict[str, str] = {}
    if not specs:
        return result
    for raw in specs:
        if not raw or "=" not in raw:
            raise ValueError(
                f"Invalid --operator {raw!r}. Use TYPE=operator, e.g. CREDIT_CARD=mask."
            )
        type_name, operator = raw.split("=", 1)
        type_name = type_name.strip().upper()
        operator = operator.strip().lower()
        if not type_name:
            raise ValueError(f"Invalid --operator {raw!r}. Missing type name.")
        if operator not in OPERATORS:
            raise ValueError(
                f"Unknown operator {operator!r} for {type_name}. "
                f"Use one of: {', '.join(sorted(OPERATORS))}."
            )
        result[type_name] = operator
    return result

restore_encrypt_tokens(text: str, secret: str) -> str

Replace every ENC1_ token. Unknown or wrong-key tokens stay as-is.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def restore_encrypt_tokens(text: str, secret: str) -> str:
    """Replace every ``ENC1_`` token. Unknown or wrong-key tokens stay as-is."""
    if not secret or ENCRYPT_PREFIX not in text:
        return text

    def _replace(match: re.Match[str]) -> str:
        try:
            return decrypt_value(match.group(0), secret)
        except ValueError:
            return match.group(0)

    return _ENCRYPT_TOKEN.sub(_replace, text)

shift_date_value(original: str, base_form: str) -> str

Shift a date by a stable offset derived from base_form.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/operators.py

def shift_date_value(original: str, base_form: str) -> str:
    """Shift a date by a stable offset derived from ``base_form``."""
    parsed = _parse_date(original)
    if parsed is None:
        return original
    digest = hashlib.sha256(f"pdf-anonymizer-date-shift:{base_form}".encode()).digest()
    offset = int.from_bytes(digest[:2], "big") % 365 - 182
    shifted = parsed + timedelta(days=offset)
    # Preserve a trailing time suffix if the original had ISO time.
    rest = original.strip()[10:] if len(original.strip()) > 10 else ""
    return shifted.isoformat() + rest

pdf_anonymizer_core.spans

Locate and apply non-overlapping replacement intervals.

Entity texts are found in the full document with the same word-boundary rules as before. Longer intervals win when two hits overlap, so John Doe is replaced and the inner John is left alone.

apply_spans(full_text: str, spans: Sequence[Span], mapping: Dict[str, str]) -> str

Write replacements from the end of the string so earlier offsets stay valid.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/spans.py

def apply_spans(full_text: str, spans: Sequence[Span], mapping: Dict[str, str]) -> str:
    """Write replacements from the end of the string so earlier offsets stay valid."""
    pieces = list(full_text)
    # Walk right-to-left so later slices do not shift earlier indexes.
    for start, end, text in sorted(spans, key=lambda item: item[0], reverse=True):
        replacement = mapping.get(text)
        if replacement is None:
            continue
        pieces[start:end] = list(replacement)
    return "".join(pieces)

locate_spans(full_text: str, entity_texts: Iterable[str]) -> List[Span]

Find every bounded occurrence of each entity text in full_text.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/spans.py

def locate_spans(full_text: str, entity_texts: Iterable[str]) -> List[Span]:
    """Find every bounded occurrence of each entity text in ``full_text``."""
    spans: List[Span] = []
    seen = set()
    for text in entity_texts:
        if not text or text in seen:
            continue
        seen.add(text)
        pattern = re.compile(make_boundary_pattern(text))
        for match in pattern.finditer(full_text):
            spans.append((match.start(), match.end(), text))
    return spans

make_boundary_pattern(text: str) -> str

Word-boundary wrap when the first/last character is alphanumeric.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/spans.py

def make_boundary_pattern(text: str) -> str:
    """Word-boundary wrap when the first/last character is alphanumeric."""
    prefix = r"\b" if text[0].isalnum() or text[0] == "_" else ""
    suffix = r"\b" if text[-1].isalnum() or text[-1] == "_" else ""
    return f"{prefix}{re.escape(text)}{suffix}"

pick_non_overlapping(spans: Sequence[Span]) -> List[Span]

Keep longest spans first; drop any that overlap an accepted span.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/spans.py

def pick_non_overlapping(spans: Sequence[Span]) -> List[Span]:
    """Keep longest spans first; drop any that overlap an accepted span."""
    ordered = sorted(spans, key=lambda item: (item[1] - item[0], -item[0]), reverse=True)
    accepted: List[Span] = []
    taken: List[Tuple[int, int]] = []
    for start, end, text in ordered:
        if _overlaps(start, end, taken):
            continue
        accepted.append((start, end, text))
        taken.append((start, end))
    return accepted

replace_entities(full_text: str, entity_texts: Iterable[str], mapping: Dict[str, str]) -> str

Locate, resolve overlaps, and replace. Empty entity list is a no-op.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/spans.py

def replace_entities(full_text: str, entity_texts: Iterable[str], mapping: Dict[str, str]) -> str:
    """Locate, resolve overlaps, and replace. Empty entity list is a no-op."""
    texts = [text for text in entity_texts if text]
    if not texts:
        return full_text
    spans = pick_non_overlapping(locate_spans(full_text, texts))
    return apply_spans(full_text, spans, mapping)

pdf_anonymizer_core.gazetteers

Keep-lists and deny-lists (gazetteers).

Keep-list: never replace this phrase, even if regex or the model found it. Deny-list: always replace this phrase, even if both stages missed it.

If a phrase is on both lists, keep wins (it stays visible).

apply_deny_list(text: str, entities: Sequence[Dict[str, str]], phrases: Iterable[str]) -> List[Dict[str, str]]

Add a CUSTOM entity for each deny-list phrase that appears in text.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/gazetteers.py

def apply_deny_list(
    text: str, entities: Sequence[Dict[str, str]], phrases: Iterable[str]
) -> List[Dict[str, str]]:
    """Add a CUSTOM entity for each deny-list phrase that appears in ``text``."""
    extra: List[Dict[str, str]] = []
    already = {(entity.get("text") or "").lower() for entity in entities}
    for phrase in phrases:
        if not phrase or phrase.lower() in already:
            continue
        match = re.search(re.escape(phrase), text, flags=re.IGNORECASE)
        if not match:
            continue
        found = match.group(0)
        extra.append(
            {
                "text": found,
                "type": "CUSTOM",
                "base_form": found,
                "score": 1.0,
                "source": "deny-list",
            }
        )
        already.add(found.lower())
    return list(entities) + extra

apply_keep_list(entities: Sequence[Dict[str, str]], phrases: Iterable[str]) -> List[Dict[str, str]]

Drop entities whose text or base form is on the keep-list (case-insensitive).

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/gazetteers.py

def apply_keep_list(
    entities: Sequence[Dict[str, str]], phrases: Iterable[str]
) -> List[Dict[str, str]]:
    """Drop entities whose text or base form is on the keep-list (case-insensitive)."""
    skip = {phrase.lower() for phrase in phrases if phrase}
    if not skip:
        return list(entities)
    kept: List[Dict[str, str]] = []
    for entity in entities:
        text = (entity.get("text") or "").lower()
        base = (entity.get("base_form") or "").lower()
        if text in skip or base in skip:
            continue
        kept.append(entity)
    return kept

load_phrase_list(path: str) -> List[str]

Load one phrase per line. Blank lines and # comments are ignored.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/gazetteers.py

def load_phrase_list(path: str) -> List[str]:
    """Load one phrase per line. Blank lines and # comments are ignored."""
    phrases: List[str] = []
    for raw in Path(path).read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        phrases.append(line)
    return phrases

pdf_anonymizer_core.verify

Residual-PII scan for already-anonymized text.

Re-runs the cheap regex pass on the masked document and optionally asks a language model to look again. Hits that are only stand-in labels (PERSON_1, IBAN_LIKE_2) are ignored. The scan reports; it does not rewrite the file.

residual_report_path(anonymized_file_path: str) -> str

data/stats/<stem>.residual_pii.json next to other stats files.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/verify.py

def residual_report_path(anonymized_file_path: str) -> str:
    """``data/stats/<stem>.residual_pii.json`` next to other stats files."""
    anonymized_path = Path(anonymized_file_path)
    file_stem = anonymized_path.name.replace(
        f".anonymized{anonymized_path.suffix}", ""
    )
    if file_stem == anonymized_path.name:
        file_stem = anonymized_path.stem
    os.makedirs(DEFAULT_STATS_DIR, exist_ok=True)
    return f"{DEFAULT_STATS_DIR}/{file_stem}.residual_pii.json"

scan_residual_llm(text: str, model_name: str, prompt_template: str = RESIDUAL_LLM_PROMPT, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0) -> List[Dict[str, str]]

Ask a language model for leftover personal details. May return [].

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/verify.py

def scan_residual_llm(
    text: str,
    model_name: str,
    prompt_template: str = RESIDUAL_LLM_PROMPT,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
) -> List[Dict[str, str]]:
    """Ask a language model for leftover personal details. May return []."""
    raw_entities = identify_entities_with_llm(
        text,
        prompt_template,
        model_name,
        max_retries=max_retries,
        base_retry_delay=base_retry_delay,
        max_retry_delay=max_retry_delay,
    )
    hits: List[Dict[str, str]] = []
    seen = set()
    for entity in raw_entities:
        raw = (entity.get("text") or "").strip()
        if not raw or looks_like_placeholder(raw):
            continue
        key = (raw, str(entity.get("type", "")).upper())
        if key in seen:
            continue
        seen.add(key)
        hits.append(_public_hit(entity))
    return hits

scan_residual_regex(text: str, regex_patterns: Optional[Dict[str, str]] = None) -> List[Dict[str, str]]

Return regex hits that are not stand-in labels.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/verify.py

def scan_residual_regex(
    text: str, regex_patterns: Optional[Dict[str, str]] = None
) -> List[Dict[str, str]]:
    """Return regex hits that are not stand-in labels."""
    patterns = DEFAULT_REGEX_PATTERNS if regex_patterns is None else regex_patterns
    hits: List[Dict[str, str]] = []
    seen = set()
    for entity in extract_entities_via_regex(text, patterns):
        raw = entity["text"].strip()
        if not raw or looks_like_placeholder(raw):
            continue
        key = (raw, entity["type"])
        if key in seen:
            continue
        seen.add(key)
        hits.append(_public_hit(entity))
    return hits

verify_anonymized_text(text: str, *, anonymized_file: Optional[str] = None, regex_patterns: Optional[Dict[str, str]] = None, use_llm: bool = False, model_name: Optional[str] = None, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0) -> Dict[str, Any]

Scan masked text. Never rewrites it.

Returns a report dict with regex_hits, optional llm_hits, and counts.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/verify.py

def verify_anonymized_text(
    text: str,
    *,
    anonymized_file: Optional[str] = None,
    regex_patterns: Optional[Dict[str, str]] = None,
    use_llm: bool = False,
    model_name: Optional[str] = None,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
) -> Dict[str, Any]:
    """Scan masked text. Never rewrites it.

    Returns a report dict with regex_hits, optional llm_hits, and counts.
    """
    regex_hits = scan_residual_regex(text, regex_patterns)
    llm_hits: List[Dict[str, str]] = []
    if use_llm:
        if not model_name:
            raise ValueError("model_name is required when use_llm is True")
        llm_hits = scan_residual_llm(
            text,
            model_name,
            max_retries=max_retries,
            base_retry_delay=base_retry_delay,
            max_retry_delay=max_retry_delay,
        )

    report: Dict[str, Any] = {
        "anonymized_file": anonymized_file,
        "regex_hits": regex_hits,
        "llm_hits": llm_hits,
        "residual_count": len(regex_hits) + len(llm_hits),
        "rewritten": False,
    }
    return report

write_residual_report(report: Dict[str, Any], anonymized_file_path: str) -> str

Write the report JSON and return its path.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/verify.py

def write_residual_report(report: Dict[str, Any], anonymized_file_path: str) -> str:
    """Write the report JSON and return its path."""
    path = residual_report_path(anonymized_file_path)
    report = dict(report)
    report["anonymized_file"] = anonymized_file_path
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(report, handle, indent=4)
    return path

pdf_anonymizer_core.risk

Linkage-risk report for already-masked text.

A name can be gone and the page can still point to one person: job + company + city in the same paragraph. This module only scores those clumps. It does not change the file.

assess_entity_list(entities: Iterable[Dict[str, Any]]) -> Dict[str, Any]

Score a flat entity list as one window (no positions). Does not rewrite.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/risk.py

def assess_entity_list(entities: Iterable[Dict[str, Any]]) -> Dict[str, Any]:
    """Score a flat entity list as one window (no positions). Does not rewrite."""
    types = sorted(
        {
            str(ent.get("type", "")).upper()
            for ent in entities
            if ent.get("type")
        }
    )
    if not types:
        return {
            "overall": "low",
            "window_count": 0,
            "high_count": 0,
            "medium_count": 0,
            "low_count": 0,
            "windows": [],
            "rewritten": False,
        }
    level, reason = _score_types(types)
    texts = [str(ent.get("text", "")) for ent in entities if ent.get("text")]
    finding = {
        "level": level,
        "types": [t for t in types if t in QUASI_TYPES],
        "placeholders": [],
        "excerpt": "; ".join(texts)[:280],
        "reason": reason,
    }
    return {
        "overall": level,
        "window_count": 1,
        "high_count": int(level == "high"),
        "medium_count": int(level == "medium"),
        "low_count": int(level == "low"),
        "windows": [finding],
        "rewritten": False,
    }

assess_linkage_risk(text: str) -> Dict[str, Any]

Score identity-clue clumps in masked text. Does not rewrite it.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/risk.py

def assess_linkage_risk(text: str) -> Dict[str, Any]:
    """Score identity-clue clumps in masked text. Does not rewrite it."""
    findings: List[Dict[str, Any]] = []
    overall = "low"
    for excerpt in _windows(text):
        tokens = _PLACEHOLDER.findall(excerpt)
        if not tokens:
            continue
        types = sorted({placeholder_type(tok) for tok in tokens})
        quasi_here = [t for t in types if t in QUASI_TYPES]
        if not quasi_here:
            continue
        level, reason = _score_types(types)
        if LEVEL_ORDER[level] > LEVEL_ORDER[overall]:
            overall = level
        findings.append(
            {
                "level": level,
                "types": quasi_here,
                "placeholders": sorted(set(tokens)),
                "excerpt": excerpt if len(excerpt) <= 280 else excerpt[:277] + "...",
                "reason": reason,
            }
        )

    # Singleton combos (a set of placeholders that appear in only one window)
    # stay at least medium — that is the "rare combination" signal.
    combo_counts: Dict[Tuple[str, ...], int] = {}
    for finding in findings:
        key = tuple(sorted(finding["placeholders"]))
        combo_counts[key] = combo_counts.get(key, 0) + 1
    for finding in findings:
        key = tuple(sorted(finding["placeholders"]))
        if combo_counts.get(key, 0) == 1 and len(finding["types"]) >= 2:
            if LEVEL_ORDER[finding["level"]] < LEVEL_ORDER["medium"]:
                finding["level"] = "medium"
                finding["reason"] = (
                    "This mix of clues appears only once, so it is easier to pin on one person."
                )
            if LEVEL_ORDER[finding["level"]] > LEVEL_ORDER[overall]:
                overall = finding["level"]

    return {
        "overall": overall,
        "window_count": len(findings),
        "high_count": sum(1 for f in findings if f["level"] == "high"),
        "medium_count": sum(1 for f in findings if f["level"] == "medium"),
        "low_count": sum(1 for f in findings if f["level"] == "low"),
        "windows": findings,
        "rewritten": False,
    }

placeholder_type(token: str) -> str

PERSON_1.v_2 → PERSON; IBAN_LIKE_1 → IBAN_LIKE.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/risk.py

def placeholder_type(token: str) -> str:
    """PERSON_1.v_2 → PERSON; IBAN_LIKE_1 → IBAN_LIKE."""
    base = token.split(".v_")[0]
    if "_" not in base:
        return base
    name, _, maybe_num = base.rpartition("_")
    if maybe_num.isdigit():
        return name
    return base

pdf_anonymizer_core.mapping_crypto

Encrypt and decrypt mapping files (AES-256-GCM + Argon2id).

The mapping file is the key to the original names. With a passphrase the JSON is locked so a leaked file is not an instant deanonymization.

Envelope v2 (current):

• AES-256-GCM authenticated encryption
• Argon2id key derivation (OWASP interactive parameters by default)
• Document SHA-256 and mapping schema version bound as AEAD AAD
• Envelope fields validated before any KDF or decrypt
• Authentication checks use constant-time compares
• Derived keys and plaintext PII live in wipeable buffers

Envelope v1 (legacy, decrypt only): scrypt, no AAD. Still accepted so existing *.mapping.json.enc files keep working.

Envelope is still JSON so you can tell an encrypted map from a plaintext one without a special file type. We write *.mapping.json.enc by default.

EnvelopeMeta dataclass

Validated envelope fields. Built before any KDF or AEAD call.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

@dataclass(frozen=True)
class EnvelopeMeta:
    """Validated envelope fields. Built *before* any KDF or AEAD call."""

    version: int
    kdf: str
    cipher: str
    salt: bytes
    nonce: bytes
    ciphertext: bytes
    source_sha256: str
    schema_version: int
    kdf_params: Dict[str, int]
    aad: Optional[bytes]

decrypt_mapping(payload: Dict[str, Any], passphrase: str, *, source_sha256: str | None = None, schema_version: int | None = None) -> Dict[str, str]

Decrypt an envelope back to placeholder -> original.

Envelope metadata is validated first. If the caller supplies an expected source hash or schema version those values are compared in constant time against the authenticated AAD before the KDF runs. GCM then re-authenticates the same AAD so a swapped or replayed mapping fails the tag check.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def decrypt_mapping(
    payload: Dict[str, Any],
    passphrase: str,
    *,
    source_sha256: str | None = None,
    schema_version: int | None = None,
) -> Dict[str, str]:
    """Decrypt an envelope back to placeholder -> original.

    Envelope metadata is validated first. If the caller supplies an expected
    source hash or schema version those values are compared in constant time
    against the authenticated AAD *before* the KDF runs. GCM then
    re-authenticates the same AAD so a swapped or replayed mapping fails
    the tag check.
    """
    if not passphrase:
        raise ValueError("A passphrase is required to open an encrypted mapping.")

    meta = validate_envelope(payload)
    expected_hash = (
        _normalize_sha256(source_sha256) if source_sha256 is not None else None
    )
    if expected_hash is not None:
        if not constant_time_equals(meta.source_sha256, expected_hash):
            raise ValueError(
                "Could not decrypt the mapping. Check the passphrase and the file."
            )
    if schema_version is not None:
        if not constant_time_int_equals(meta.schema_version, int(schema_version)):
            raise ValueError(
                "Could not decrypt the mapping. Check the passphrase and the file."
            )

    try:
        with SecureBytes(passphrase.encode("utf-8")) as password:
            with SecureBytes(_KEY_LEN) as key:
                if meta.version == LEGACY_VERSION:
                    _derive_key_scrypt(password, meta.salt, key)
                else:
                    _derive_key_argon2id(password, meta.salt, meta.kdf_params, key)
                raw = AESGCM(bytes(key.view())).decrypt(
                    meta.nonce, meta.ciphertext, meta.aad
                )
        plaintext = bytearray(raw)
        try:
            data = json.loads(bytes(plaintext).decode("utf-8"))
        finally:
            wipe_mutable(plaintext)
    except (ValueError, InvalidTag, json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise ValueError(
            "Could not decrypt the mapping. Check the passphrase and the file."
        ) from exc

    if not isinstance(data, dict):
        raise ValueError("Decrypted mapping is not a JSON object.")
    return {str(k): str(v) for k, v in data.items()}

encode_aad(source_sha256: str, schema_version: int, envelope_version: int = VERSION) -> bytes

Canonical AAD bytes bound into AES-GCM.

Newline-delimited, not JSON: callers cannot change the binding by adding spaces or reordering keys.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def encode_aad(
    source_sha256: str,
    schema_version: int,
    envelope_version: int = VERSION,
) -> bytes:
    """Canonical AAD bytes bound into AES-GCM.

    Newline-delimited, not JSON: callers cannot change the binding by
    adding spaces or reordering keys.
    """
    return (
        f"{FORMAT}\n"
        f"v={envelope_version}\n"
        f"schema={schema_version}\n"
        f"sha256={source_sha256}\n"
    ).encode("utf-8")

encrypt_mapping(mapping: Dict[str, str], passphrase: str, *, source_sha256: str | None = None, schema_version: int | None = None, kdf_params: Mapping[str, int] | None = None) -> Dict[str, Any]

Return a JSON-serializable encrypted envelope.

Extra keyword arguments are optional so existing callers encrypt_mapping(mapping, passphrase) keep working. When source_sha256 is omitted the AAD still binds an empty hash so the field cannot be filled in later without breaking the tag.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def encrypt_mapping(
    mapping: Dict[str, str],
    passphrase: str,
    *,
    source_sha256: str | None = None,
    schema_version: int | None = None,
    kdf_params: Mapping[str, int] | None = None,
) -> Dict[str, Any]:
    """Return a JSON-serializable encrypted envelope.

    Extra keyword arguments are optional so existing callers
    ``encrypt_mapping(mapping, passphrase)`` keep working. When
    ``source_sha256`` is omitted the AAD still binds an empty hash so the
    field cannot be filled in later without breaking the tag.
    """
    if not passphrase:
        raise ValueError("A non-empty passphrase is required to encrypt a mapping.")
    if not isinstance(mapping, dict):
        raise ValueError("Mapping must be a JSON object.")

    digest = _normalize_sha256(source_sha256)
    schema = MAPPING_SCHEMA_VERSION if schema_version is None else int(schema_version)
    if schema < 1 or schema > 16:
        raise ValueError("schema_version is out of range.")

    params = {
        "iterations": DEFAULT_ARGON2_ITERATIONS,
        "lanes": DEFAULT_ARGON2_LANES,
        "memory_cost": DEFAULT_ARGON2_MEMORY,
    }
    if kdf_params:
        params.update({k: int(v) for k, v in kdf_params.items()})
    params = _validate_kdf_params(VERSION, params)

    salt = secrets.token_bytes(_SALT_LEN)
    nonce = secrets.token_bytes(_NONCE_LEN)
    aad = encode_aad(digest, schema, VERSION)

    plaintext = bytearray(
        json.dumps(mapping, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    )
    try:
        with SecureBytes(passphrase.encode("utf-8")) as password:
            with SecureBytes(_KEY_LEN) as key:
                _derive_key_argon2id(password, salt, params, key)
                ciphertext = AESGCM(bytes(key.view())).encrypt(
                    nonce, bytes(plaintext), aad
                )
    finally:
        wipe_mutable(plaintext)

    return {
        "format": FORMAT,
        "v": VERSION,
        "kdf": KDF_ARGON2ID,
        "cipher": CIPHER_NAME,
        "kdf_params": params,
        "salt": _b64(salt),
        "nonce": _b64(nonce),
        "ciphertext": _b64(ciphertext),
        "aad": {
            "source_sha256": digest,
            "schema_version": schema,
        },
    }

is_encrypted_mapping(payload: Any) -> bool

True if payload looks like our encrypted envelope, not a name map.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def is_encrypted_mapping(payload: Any) -> bool:
    """True if ``payload`` looks like our encrypted envelope, not a name map."""
    if not isinstance(payload, dict):
        return False
    if not constant_time_equals(str(payload.get("format", "")), FORMAT):
        return False
    return "ciphertext" in payload

load_mapping_payload(raw_mapping: Any, passphrase: str | None, *, source_sha256: str | None = None, schema_version: int | None = None) -> Dict[str, str]

Load plaintext or encrypted mapping JSON already parsed from disk.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def load_mapping_payload(
    raw_mapping: Any,
    passphrase: str | None,
    *,
    source_sha256: str | None = None,
    schema_version: int | None = None,
) -> Dict[str, str]:
    """Load plaintext or encrypted mapping JSON already parsed from disk."""
    if is_encrypted_mapping(raw_mapping):
        if not passphrase:
            raise ValueError(
                "This mapping is encrypted. Pass --mapping-passphrase or set "
                "ANONYMIZER_MAPPING_KEY."
            )
        return decrypt_mapping(
            raw_mapping,
            passphrase,
            source_sha256=source_sha256,
            schema_version=schema_version,
        )
    if not isinstance(raw_mapping, dict):
        raise ValueError("Mapping file must be a JSON object.")
    return {str(k): str(v) for k, v in raw_mapping.items()}

resolve_mapping_passphrase(explicit: str | None = None) -> str | None

CLI flag wins; otherwise ANONYMIZER_MAPPING_KEY.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def resolve_mapping_passphrase(explicit: str | None = None) -> str | None:
    """CLI flag wins; otherwise ``ANONYMIZER_MAPPING_KEY``."""
    if explicit:
        return explicit
    value = os.getenv(MAPPING_KEY_ENV)
    return value or None

sha256_file(path: str | os.PathLike[str]) -> str

Return the lowercase hex SHA-256 of a file, streamed in 1 MiB chunks.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def sha256_file(path: str | os.PathLike[str]) -> str:
    """Return the lowercase hex SHA-256 of a file, streamed in 1 MiB chunks."""
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        while True:
            chunk = handle.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()

validate_envelope(payload: Any) -> EnvelopeMeta

Strict metadata checks. Raises ValueError before any crypto work.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/mapping_crypto.py

def validate_envelope(payload: Any) -> EnvelopeMeta:
    """Strict metadata checks. Raises ``ValueError`` before any crypto work."""
    data = _require_dict(payload)
    if not constant_time_equals(str(data.get("format", "")), FORMAT):
        raise ValueError("This file is not an encrypted mapping.")

    version_raw = data.get("v")
    if not isinstance(version_raw, int) or isinstance(version_raw, bool):
        raise ValueError("Envelope version is missing or not an integer.")
    is_v1 = constant_time_int_equals(version_raw, LEGACY_VERSION)
    is_v2 = constant_time_int_equals(version_raw, VERSION)
    if not (is_v1 or is_v2):
        raise ValueError("Unsupported mapping envelope version.")
    version = LEGACY_VERSION if is_v1 else VERSION

    kdf = str(data.get("kdf", ""))
    cipher = str(data.get("cipher", ""))
    expected_kdf = KDF_SCRYPT if version == LEGACY_VERSION else KDF_ARGON2ID
    if not constant_time_equals(kdf, expected_kdf):
        raise ValueError("Envelope key-derivation algorithm is not allowed.")
    if not constant_time_equals(cipher, CIPHER_NAME):
        raise ValueError("Envelope cipher is not allowed.")

    salt = _unb64(data.get("salt", ""))
    nonce = _unb64(data.get("nonce", ""))
    ciphertext = _unb64(data.get("ciphertext", ""))
    if len(salt) != _SALT_LEN:
        raise ValueError("Envelope salt has an unexpected length.")
    if len(nonce) != _NONCE_LEN:
        raise ValueError("Envelope nonce has an unexpected length.")
    if not ciphertext:
        raise ValueError("Envelope ciphertext is empty.")
    if len(ciphertext) > _MAX_CIPHERTEXT_LEN:
        raise ValueError("Envelope ciphertext is larger than the allowed maximum.")

    kdf_params = _validate_kdf_params(version, data.get("kdf_params"))
    source_sha256 = ""
    schema_version = MAPPING_SCHEMA_VERSION
    aad: Optional[bytes] = None
    if version == VERSION:
        aad_block = data.get("aad")
        if not isinstance(aad_block, dict):
            raise ValueError("Envelope AAD metadata is missing.")
        source_sha256 = _normalize_sha256(str(aad_block.get("source_sha256", "")))
        schema_raw = aad_block.get("schema_version", MAPPING_SCHEMA_VERSION)
        if not isinstance(schema_raw, int) or isinstance(schema_raw, bool):
            raise ValueError("Envelope schema_version is not an integer.")
        if schema_raw < 1 or schema_raw > 16:
            raise ValueError("Envelope schema_version is out of range.")
        schema_version = schema_raw
        aad = encode_aad(source_sha256, schema_version, version)

    return EnvelopeMeta(
        version=version,
        kdf=kdf,
        cipher=cipher,
        salt=salt,
        nonce=nonce,
        ciphertext=ciphertext,
        source_sha256=source_sha256,
        schema_version=schema_version,
        kdf_params=kdf_params,
        aad=aad,
    )

---

Low-Level Components (for advanced use / extension)

pdf_anonymizer_core.llm_provider

LLM provider adapters + response caching layer.

This module contains: - LocalLLMCache: thread-safe on-disk caching of LLM responses (keyed by model + prompt hash) to avoid repeated calls during development or re-processing. - LLMProvider abstract base + concrete implementations for Google, Ollama, Hugging Face, OpenRouter, OpenAI, and Anthropic. - configure_cache() and get_provider() factory.

All providers implement a uniform .call(prompt, model_name) that goes through the cache when enabled.

LLMProvider

Bases: ABC

Abstract base class for LLM providers.

Subclasses must implement _call_raw. The public .call() method (inherited) adds transparent caching when the global cache is enabled.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/llm_provider.py

class LLMProvider(ABC):
    """Abstract base class for LLM providers.

    Subclasses must implement _call_raw. The public .call() method (inherited)
    adds transparent caching when the global cache is enabled.
    """

    @abstractmethod
    def _call_raw(
        self, prompt: str, model_name: str, max_output_tokens: Optional[int] = None
    ) -> str:
        """Raw provider call. Must be overridden by concrete providers."""
        pass

    def call(
        self, prompt: str, model_name: str, max_output_tokens: Optional[int] = None
    ) -> str:
        """Public entry point used by the anonymizer.

        Checks the cache (if enabled) before delegating to the concrete provider.
        """
        global _cache_instance, _cache_enabled

        # Initialize default cache if enabled and not yet initialized
        if _cache_enabled and _cache_instance is None:
            configure_cache(True, DEFAULT_CACHE_DIR, DEFAULT_CACHE_FILE)

        if _cache_enabled and _cache_instance is not None:
            cached_val = _cache_instance.get(model_name, prompt)
            if cached_val is not None:
                logging.info(f"Cache hit for model '{model_name}'")
                return cached_val

        response = self._call_raw(prompt, model_name, max_output_tokens)

        if _cache_enabled and _cache_instance is not None and response:
            _cache_instance.set(model_name, prompt, response)

        return response

call(prompt: str, model_name: str, max_output_tokens: Optional[int] = None) -> str

Public entry point used by the anonymizer.

Checks the cache (if enabled) before delegating to the concrete provider.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/llm_provider.py

def call(
    self, prompt: str, model_name: str, max_output_tokens: Optional[int] = None
) -> str:
    """Public entry point used by the anonymizer.

    Checks the cache (if enabled) before delegating to the concrete provider.
    """
    global _cache_instance, _cache_enabled

    # Initialize default cache if enabled and not yet initialized
    if _cache_enabled and _cache_instance is None:
        configure_cache(True, DEFAULT_CACHE_DIR, DEFAULT_CACHE_FILE)

    if _cache_enabled and _cache_instance is not None:
        cached_val = _cache_instance.get(model_name, prompt)
        if cached_val is not None:
            logging.info(f"Cache hit for model '{model_name}'")
            return cached_val

    response = self._call_raw(prompt, model_name, max_output_tokens)

    if _cache_enabled and _cache_instance is not None and response:
        _cache_instance.set(model_name, prompt, response)

    return response

configure_cache(enabled: bool, cache_dir: str = 'data/cache', cache_file: str = 'llm_responses.json')

Enable or disable (and optionally relocate) the global LLM response cache.

Called automatically by the CLI according to the active AppConfig. You can also call it directly when using the core SDK.

Parameters:

Name	Type	Description	Default
enabled	bool	Whether caching should be active.	required
cache_dir	str	Directory in which llm_responses.json will be stored.	'data/cache'
cache_file	str	Filename for the JSON cache.	'llm_responses.json'

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/llm_provider.py

def configure_cache(
    enabled: bool, cache_dir: str = "data/cache", cache_file: str = "llm_responses.json"
):
    """Enable or disable (and optionally relocate) the global LLM response cache.

    Called automatically by the CLI according to the active AppConfig.
    You can also call it directly when using the core SDK.

    Args:
        enabled: Whether caching should be active.
        cache_dir: Directory in which llm_responses.json will be stored.
        cache_file: Filename for the JSON cache.
    """
    global _cache_instance, _cache_enabled
    _cache_enabled = enabled
    if enabled:
        _cache_instance = LocalLLMCache(cache_dir, cache_file)
    else:
        _cache_instance = None

get_provider(provider_name: str) -> LLMProvider

Factory function to get a provider instance.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/llm_provider.py

def get_provider(provider_name: str) -> LLMProvider:
    """Factory function to get a provider instance."""
    provider_map = {
        "google": GoogleProvider,
        "ollama": OllamaProvider,
        "huggingface": HuggingFaceProvider,
        "openrouter": OpenRouterProvider,
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
    }
    provider_class = provider_map.get(provider_name)
    if provider_class:
        return provider_class()
    raise ValueError(f"Unknown provider: {provider_name}")

pdf_anonymizer_core.call_llm

LLM calling logic with structured parsing and resilient retries.

Contains: - Pydantic models used to validate LLM JSON responses (EntityModel, IdentificationResult). - classify_error(): decides which exceptions are worth retrying. - identify_entities_with_llm(): the main retrying wrapper that talks to providers.

classify_error(exception: Exception) -> tuple[bool, str]

Classify an exception to determine if it is retryable and get a descriptive label.

Parameters:

Name	Type	Description	Default
exception	Exception	The exception object.	required

Returns:

Type	Description
tuple[bool, str]	A tuple of (is_retryable, error_category).

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/call_llm.py

def classify_error(exception: Exception) -> tuple[bool, str]:
    """
    Classify an exception to determine if it is retryable and get a descriptive label.

    Args:
        exception: The exception object.

    Returns:
        A tuple of (is_retryable, error_category).
    """
    exc_name = type(exception).__name__
    exc_msg = str(exception).lower()

    # RateLimit errors (429)
    if "ratelimit" in exc_name.lower() or "429" in exc_msg or "rate limit" in exc_msg:
        return True, "RATE_LIMIT_ERROR"

    # Server / Temporary errors (5xx)
    if (
        "500" in exc_msg
        or "502" in exc_msg
        or "503" in exc_msg
        or "504" in exc_msg
        or "server error" in exc_msg
    ):
        return True, "SERVER_ERROR"

    # Connection / Timeout errors
    if (
        "connection" in exc_msg
        or "timeout" in exc_msg
        or "api-connection" in exc_msg
        or "apiconnection" in exc_name.lower()
    ):
        return True, "CONNECTION_ERROR"

    # Authentication / Permissions (401, 403)
    if (
        "authentication" in exc_name.lower()
        or "401" in exc_msg
        or "403" in exc_msg
        or "apikey" in exc_msg
        or "api key" in exc_msg
        or "unauthorized" in exc_msg
    ):
        return False, "AUTHENTICATION_ERROR"

    # JSON decoding / Pydantic validation errors (retryable, LLM might fix it next attempt)
    if "jsondecodeerror" in exc_name.lower() or "validationerror" in exc_name.lower():
        return True, "PARSING_ERROR"

    # Standard fallback
    return True, "GENERIC_ERROR"

identify_entities_with_llm(text: str, prompt_template: str, model_name: str, max_retries: int = 3, base_retry_delay: float = 1.0, max_retry_delay: float = 10.0) -> List[dict]

Call an LLM (via the configured provider) to extract PII entities from one chunk.

The call is wrapped with retry logic: - Transient errors (rate limits, 5xx, connection, parsing) are retried. - Auth errors are not retried. - Uses exponential backoff + jitter.

The response is cleaned of markdown fences and validated with Pydantic before returning a list of plain dicts.

Parameters:

Name	Type	Description	Default
text	str	The chunk of text to analyze.	required
prompt_template	str	Prompt containing a {text} placeholder.	required
model_name	str	Model string (passed through get_provider_and_model_name).	required
max_retries	int	Maximum number of attempts.	3
base_retry_delay	float	Starting backoff delay (seconds).	1.0
max_retry_delay	float	Upper bound on backoff delay (seconds).	10.0

Returns:

Type	Description
List[dict]	List of entity dicts (each with "text", "type", "base_form",
List[dict]	"score" 0.70, and "source" "llm"). Returns [] on unrecoverable
List[dict]	failure after retries.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/call_llm.py

def identify_entities_with_llm(
    text: str,
    prompt_template: str,
    model_name: str,
    max_retries: int = 3,
    base_retry_delay: float = 1.0,
    max_retry_delay: float = 10.0,
) -> List[dict]:
    """Call an LLM (via the configured provider) to extract PII entities from one chunk.

    The call is wrapped with retry logic:
    - Transient errors (rate limits, 5xx, connection, parsing) are retried.
    - Auth errors are not retried.
    - Uses exponential backoff + jitter.

    The response is cleaned of markdown fences and validated with Pydantic
    before returning a list of plain dicts.

    Args:
        text: The chunk of text to analyze.
        prompt_template: Prompt containing a {text} placeholder.
        model_name: Model string (passed through get_provider_and_model_name).
        max_retries: Maximum number of attempts.
        base_retry_delay: Starting backoff delay (seconds).
        max_retry_delay: Upper bound on backoff delay (seconds).

    Returns:
        List of entity dicts (each with "text", "type", "base_form",
        "score" 0.70, and "source" "llm"). Returns [] on unrecoverable
        failure after retries.
    """
    prompt = prompt_template.format(text=text)

    for attempt in range(max_retries):
        try:
            logging.info(
                f"Calling '{model_name}': text: {len(text):,}, attempt {attempt + 1}"
            )
            provider_name, actual_model_name = get_provider_and_model_name(model_name)
            provider = get_provider(provider_name)
            raw_text = provider.call(prompt, actual_model_name)

            cleaned_response = (
                raw_text.strip().replace("```json", "").replace("```", "").strip()
            )

            # Validate and parse response using Pydantic
            result = IdentificationResult.model_validate_json(cleaned_response)

            dumped: List[dict] = []
            for entity in result.entities:
                item = entity.model_dump()
                item.setdefault("score", 0.70)
                item.setdefault("source", "llm")
                if not item.get("base_form"):
                    item["base_form"] = item["text"]
                dumped.append(item)
            return dumped

        except Exception as e:
            is_retryable, category = classify_error(e)
            logging.error(
                f"Attempt {attempt + 1} failed with error category '{category}': {e}"
            )

            if not is_retryable or attempt + 1 == max_retries:
                if attempt + 1 == max_retries:
                    logging.error("Max retries reached. Returning empty list.")
                else:
                    logging.error(
                        f"Fatal error category '{category}'. Stopping retries."
                    )
                return []

            # Exponential backoff with jitter
            backoff = min(base_retry_delay * (2**attempt), max_retry_delay)
            jitter = random.uniform(0, 0.1 * backoff)
            sleep_time = backoff + jitter

            logging.info(f"Retrying in {sleep_time:.2f} seconds...")
            time.sleep(sleep_time)

    return []

pdf_anonymizer_core.load_and_extract

Text extraction and semantic chunking for PDF, Markdown, and plain text.

CSV, Excel, and Word files do not use this loader.

Uses pymupdf4llm for high-quality PDF → Markdown conversion (preserves structure useful for LLMs) and langchain text splitters: - MarkdownTextSplitter for .pdf and .md (respects headers/code blocks) - RecursiveCharacterTextSplitter for .txt / fallback

Chunk size and overlap are the primary controls for memory usage and LLM context consumption.

load_and_extract_text_from_file(file_path: str, characters_to_anonymize: int = 100000, chunk_overlap: int = 0, ocr: bool = False) -> Tuple[str, List[str]]

Loads a file and extracts text, returning the full text and chunked text.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the file.	required
characters_to_anonymize	int	Number of characters to process in each chunk.	100000
chunk_overlap	int	Overlap size between chunks.	0

Returns:

Type	Description
Tuple[str, List[str]]	Tuple[str, List[str]]: The full text as a string, and a list of chunk strings.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/load_and_extract.py

def load_and_extract_text_from_file(
    file_path: str,
    characters_to_anonymize: int = 100000,
    chunk_overlap: int = 0,
    ocr: bool = False,
) -> Tuple[str, List[str]]:
    """
    Loads a file and extracts text, returning the full text and chunked text.

    Args:
        file_path (str): The path to the file.
        characters_to_anonymize: Number of characters to process in each chunk.
        chunk_overlap: Overlap size between chunks.

    Returns:
        Tuple[str, List[str]]: The full text as a string, and a list of chunk strings.
    """
    path = Path(file_path)
    file_extension = path.suffix.lower()

    if is_rejected_word(file_path):
        raise rejected_word_error(file_path)
    if is_word_path(file_path):
        raise ValueError(
            ".docx files must be loaded as Word documents, not as plain text."
        )
    if is_rejected_spreadsheet(file_path):
        raise rejected_spreadsheet_error(file_path)
    if is_tabular_path(file_path):
        if file_extension == ".xlsx":
            raise ValueError(EXCEL_EXTRA_MESSAGE)
        raise ValueError(
            f"{file_extension} files must be loaded as tables, not as plain text."
        )

    try:
        if file_extension == ".pdf":
            return load_and_extract_text_from_pdf(
                file_path, characters_to_anonymize, chunk_overlap, ocr=ocr
            )
        elif file_extension == ".md":
            with open(file_path, "r", encoding="utf-8") as f:
                text = f.read()
            splitter = MarkdownTextSplitter(
                chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
            )
            docs = splitter.create_documents([text])
            return text, [doc.page_content for doc in docs]
        elif file_extension == ".txt":
            with open(file_path, "r", encoding="utf-8") as f:
                text = f.read()
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
            )
            docs = splitter.create_documents([text])
            return text, [doc.page_content for doc in docs]
        else:
            logging.warning(
                f"Unsupported file type: {file_extension}. Treating as plain text."
            )
            with open(file_path, "r", encoding="utf-8") as f:
                text = f.read()
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
            )
            docs = splitter.create_documents([text])
            return text, [doc.page_content for doc in docs]
    except FileNotFoundError as e:
        logging.error(f"Error: The file at {file_path} was not found.")
        raise e
    except Exception as e:
        logging.error(f"An error occurred while reading the file: {e}")
        raise e

load_and_extract_text_from_pdf(file_path: str, characters_to_anonymize: int = 100000, chunk_overlap: int = 0, ocr: bool = False) -> Tuple[str, List[str]]

Loads a PDF file and extracts text from each page, returning the full text and chunked text.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the PDF file.	required
characters_to_anonymize	int	Number of characters to anonymize in one go (chunk size).	100000
chunk_overlap	int	Overlap size between chunks.	0

Returns:

Type	Description
Tuple[str, List[str]]	Tuple[str, List[str]]: The full text as a string, and a list of chunk strings.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/load_and_extract.py

def load_and_extract_text_from_pdf(
    file_path: str,
    characters_to_anonymize: int = 100000,
    chunk_overlap: int = 0,
    ocr: bool = False,
) -> Tuple[str, List[str]]:
    """
    Loads a PDF file and extracts text from each page, returning the full text and chunked text.

    Args:
        file_path (str): The path to the PDF file.
        characters_to_anonymize: Number of characters to anonymize in one go (chunk size).
        chunk_overlap: Overlap size between chunks.

    Returns:
        Tuple[str, List[str]]: The full text as a string, and a list of chunk strings.
    """
    try:
        md_text = pymupdf4llm.to_markdown(file_path, show_progress=False)
        if (md_text or "").strip():
            splitter = MarkdownTextSplitter(
                chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
            )
            docs = splitter.create_documents([md_text])
            return md_text, [doc.page_content for doc in docs]

        if ocr:
            ocr_text, words = ocr_pdf(file_path)
            if (ocr_text or "").strip():
                store_pdf_layout(file_path, words)
                splitter = RecursiveCharacterTextSplitter(
                    chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
                )
                docs = splitter.create_documents([ocr_text])
                return ocr_text, [doc.page_content for doc in docs]
            _reject_empty_pdf_extract(file_path, ocr_attempted=True)

        _reject_empty_pdf_extract(file_path, ocr_attempted=False)
        splitter = MarkdownTextSplitter(
            chunk_size=characters_to_anonymize, chunk_overlap=chunk_overlap
        )
        docs = splitter.create_documents([md_text or ""])
        return md_text or "", [doc.page_content for doc in docs]
    except FileNotFoundError as e:
        logging.error(f"Error: The file at {file_path} was not found.")
        raise e
    except Exception as e:
        logging.error(f"An error occurred while reading the PDF: {e}")
        raise e

pdf_anonymizer_core.span_ner.extract_entities_via_ner(text: str, *, model_name: str = DEFAULT_NER_MODEL, labels: Optional[Iterable[str]] = None) -> List[EntityDict]

Return PERSON / ORGANIZATION / LOCATION / ADDRESS / DATE spans.

Does not emit identity clues (INDIRECT). That stays on the LLM path.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/span_ner.py

def extract_entities_via_ner(
    text: str,
    *,
    model_name: str = DEFAULT_NER_MODEL,
    labels: Optional[Iterable[str]] = None,
) -> List[EntityDict]:
    """Return PERSON / ORGANIZATION / LOCATION / ADDRESS / DATE spans.

    Does not emit identity clues (``INDIRECT``). That stays on the LLM path.
    """
    if not (text or "").strip():
        return []
    if not ner_available():
        raise ValueError(NER_EXTRA_MESSAGE)

    label_list = list(labels) if labels is not None else list(DEFAULT_NER_LABELS)
    model = _load_model(model_name)
    entities: List[EntityDict] = []
    seen: set[tuple[str, str]] = set()
    for _offset, window in _windows(text):
        try:
            hits = model.predict_entities(window, label_list)
        except Exception as exc:
            logging.warning("Local span NER failed on a window: %s", exc)
            continue
        for hit in hits or []:
            raw = (hit.get("text") or "").strip()
            label = str(hit.get("label") or "").strip().lower()
            if not raw or label not in _LABEL_TO_TYPE:
                continue
            ent_type = _LABEL_TO_TYPE[label]
            key = (raw, ent_type)
            if key in seen:
                continue
            seen.add(key)
            raw_score = hit.get("score")
            if isinstance(raw_score, (int, float)):
                score = max(0.0, min(1.0, float(raw_score)))
            else:
                score = 0.80
            entities.append(
                {
                    "text": raw,
                    "type": ent_type,
                    "base_form": raw,
                    "score": score,
                    "source": "ner",
                }
            )
    return entities

pdf_anonymizer_core.span_ner.resolve_semantic_stages(*, use_llm: bool, use_ner: Optional[bool], replace_llm_when_ner: bool) -> tuple[bool, bool]

Return (run_ner, run_llm).

use_ner is True / False / None (auto). Auto turns NER on only when the extra is installed and the language model would have run. Speed and cost profiles set replace_llm_when_ner so NER replaces the LLM.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/span_ner.py

def resolve_semantic_stages(
    *,
    use_llm: bool,
    use_ner: Optional[bool],
    replace_llm_when_ner: bool,
) -> tuple[bool, bool]:
    """Return ``(run_ner, run_llm)``.

    ``use_ner`` is True / False / None (auto). Auto turns NER on only when
    the extra is installed and the language model would have run. Speed and
    cost profiles set ``replace_llm_when_ner`` so NER replaces the LLM.
    """
    if use_ner is True:
        if not ner_available():
            raise ValueError(NER_EXTRA_MESSAGE)
        run_ner = True
    elif use_ner is False:
        run_ner = False
    else:
        run_ner = bool(use_llm) and ner_available()

    run_llm = bool(use_llm)
    if run_ner and replace_llm_when_ner:
        run_llm = False
    return run_ner, run_llm

pdf_anonymizer_core.pdf_ocr.ocr_pdf(path: str, *, language: str = 'eng', dpi: int = 200) -> tuple[str, list[PdfWord]]

OCR every page. Returns (plain_text, words_with_boxes).

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/pdf_ocr.py

def ocr_pdf(
    path: str,
    *,
    language: str = "eng",
    dpi: int = 200,
) -> tuple[str, list[PdfWord]]:
    """OCR every page. Returns ``(plain_text, words_with_boxes)``."""
    if not tesseract_available():
        raise ValueError(TESSERACT_MISSING_MESSAGE)

    try:
        import pymupdf
    except ImportError as exc:
        raise ValueError("OCR requires pymupdf.") from exc

    words: list[PdfWord] = []
    page_texts: list[str] = []
    try:
        document = pymupdf.open(path)
    except Exception as exc:
        raise ValueError(f"Cannot open PDF for OCR: {exc}") from exc

    try:
        for index, page in enumerate(document):
            try:
                textpage = page.get_textpage_ocr(language=language, dpi=dpi, full=True)
            except Exception as exc:
                raise ValueError(f"OCR failed on page {index + 1}: {exc}") from exc
            raw_words = page.get_text("words", textpage=textpage) or []
            tokens: list[str] = []
            for item in raw_words:
                if len(item) < 5:
                    continue
                x0, y0, x1, y1, token = item[:5]
                text = str(token)
                if not text.strip():
                    continue
                words.append(
                    PdfWord(
                        page=index,
                        text=text,
                        x0=float(x0),
                        y0=float(y0),
                        x1=float(x1),
                        y1=float(y1),
                    )
                )
                tokens.append(text)
            if tokens:
                page_texts.append(" ".join(tokens))
    finally:
        document.close()

    return "\n\n".join(page_texts), words

pdf_anonymizer_core.pdf_ocr.write_layout_sidecar(dest_path: str, source: str, words: Iterable[PdfWord]) -> str

Write <anonymized-stem>.layout.json next to the anonymized file.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/pdf_ocr.py

def write_layout_sidecar(dest_path: str, source: str, words: Iterable[PdfWord]) -> str:
    """Write ``<anonymized-stem>.layout.json`` next to the anonymized file."""
    dest = Path(dest_path)
    # letter.anonymized.md → letter.anonymized.layout.json
    layout_path = dest.with_suffix(".layout.json")
    layout_path.parent.mkdir(parents=True, exist_ok=True)
    layout_path.write_text(
        json.dumps(layout_to_json(source, words), indent=2),
        encoding="utf-8",
    )
    return str(layout_path)

pdf_anonymizer_core.pdf_output.write_anonymized_pdf(source_path: str, dest_path: str, orig_to_written: Dict[str, str], entity_texts: Iterable[str], *, redact: bool = False) -> None

Rewrite source_path as a sanitized .pdf.

When redact is false, each hit is replaced with its written stand-in (reversible with the mapping). When redact is true, the hit becomes a black box and cannot be restored from the page.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/pdf_output.py

def write_anonymized_pdf(
    source_path: str,
    dest_path: str,
    orig_to_written: Dict[str, str],
    entity_texts: Iterable[str],
    *,
    redact: bool = False,
) -> None:
    """Rewrite ``source_path`` as a sanitized ``.pdf``.

    When ``redact`` is false, each hit is replaced with its written stand-in
    (reversible with the mapping). When ``redact`` is true, the hit becomes a
    black box and cannot be restored from the page.
    """
    if Path(source_path).suffix.lower() != ".pdf":
        raise ValueError(OUTPUT_PDF_NOT_PDF_MESSAGE)

    pymupdf = _require_pymupdf()
    texts = [text for text in entity_texts if text]
    try:
        document = pymupdf.open(source_path)
    except Exception as exc:
        raise ValueError(f"Cannot open PDF {source_path}") from exc

    dest = Path(dest_path)
    dest.parent.mkdir(parents=True, exist_ok=True)
    try:
        for page in document:
            hits = _search_hits(page, texts, orig_to_written)
            for rect, _original, written in hits:
                box = pymupdf.Rect(*rect)
                if redact:
                    page.add_redact_annot(box, fill=(0, 0, 0), cross_out=False)
                else:
                    fontsize = max(4.0, min(11.0, box.height * 0.8 or 8.0))
                    page.add_redact_annot(
                        box,
                        text=written,
                        fontsize=fontsize,
                        fill=(1, 1, 1),
                        text_color=(0, 0, 0),
                        cross_out=False,
                    )
            image_mode = (
                pymupdf.PDF_REDACT_IMAGE_PIXELS
                if redact
                else pymupdf.PDF_REDACT_IMAGE_NONE
            )
            page.apply_redactions(images=image_mode)
        sanitize_pdf(document)
        document.save(
            str(dest),
            garbage=4,
            deflate=True,
            incremental=False,
            encryption=pymupdf.PDF_ENCRYPT_NONE,
        )
    except Exception as exc:
        raise ValueError(f"Cannot write native PDF {dest_path}") from exc
    finally:
        document.close()

pdf_anonymizer_core.pdf_output.sanitize_pdf(document) -> None

Drop identity-bearing extras on every native-PDF write.

Clears /Info, XMP, embedded files, leftover annotations, optional content groups, and outline titles are left alone only when they have no leftover PII we can see. Incremental /Prev history is avoided by a full rewrite at save time (garbage=4, not incremental).

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/pdf_output.py

def sanitize_pdf(document) -> None:
    """Drop identity-bearing extras on every native-PDF write.

    Clears ``/Info``, XMP, embedded files, leftover annotations, optional
    content groups, and outline titles are left alone only when they have
    no leftover PII we can see. Incremental ``/Prev`` history is avoided by
    a full rewrite at save time (``garbage=4``, not incremental).
    """
    document.set_metadata({})
    try:
        document.del_xml_metadata()
    except Exception as exc:
        logging.info("Could not delete XMP metadata: %s", exc)

    try:
        count = document.embfile_count()
        for index in range(count - 1, -1, -1):
            document.embfile_del(index)
    except Exception as exc:
        logging.info("Could not delete embedded files: %s", exc)

    try:
        catalog = document.pdf_catalog()
        kind, _value = document.xref_get_key(catalog, "OCProperties")
        if kind != "null":
            document.xref_set_key(catalog, "OCProperties", "null")
    except Exception as exc:
        logging.info("Could not drop optional-content groups: %s", exc)

    try:
        catalog = document.pdf_catalog()
        kind, _value = document.xref_get_key(catalog, "Thumb")
        if kind != "null":
            document.xref_set_key(catalog, "Thumb", "null")
    except Exception:
        pass

    for page in document:
        annots = list(page.annots() or [])
        for annot in annots:
            try:
                page.delete_annot(annot)
            except Exception:
                continue

---

Tables (CSV / Excel)

pdf_anonymizer_core.tables.load_table(path: str) -> TableDocument

Load a .csv or .xlsx file as a TableDocument.

Raises ValueError for rejected spreadsheet suffixes, a missing [excel] extra, or a file over the size / cell cap.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/tables.py

def load_table(path: str) -> TableDocument:
    """Load a ``.csv`` or ``.xlsx`` file as a ``TableDocument``.

    Raises ``ValueError`` for rejected spreadsheet suffixes, a missing
    ``[excel]`` extra, or a file over the size / cell cap.
    """
    suffix = Path(path).suffix.lower()
    if suffix in REJECT_SPREADSHEET_SUFFIXES:
        raise rejected_spreadsheet_error(path)
    if suffix == ".csv":
        return load_csv(path)
    if suffix == ".xlsx":
        return load_xlsx(path)
    raise ValueError(f"Not a supported table file: {path}")

pdf_anonymizer_core.tables.save_table(doc: TableDocument, path: str) -> None

Write doc as .csv or .xlsx. Excel formulas are not persisted.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/tables.py

def save_table(doc: TableDocument, path: str) -> None:
    """Write ``doc`` as ``.csv`` or ``.xlsx``. Excel formulas are not persisted."""
    suffix = Path(path).suffix.lower()
    if suffix == ".xlsx" or doc.kind == "xlsx":
        save_xlsx(doc, path)
        return
    save_csv(doc, path)

pdf_anonymizer_core.tables.apply_mapping_to_table(doc: TableDocument, orig_to_written: Dict[str, str], entity_texts: Iterable[str]) -> TableDocument

Replace detected entity texts in each cell. Does not use mapping keys.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/tables.py

def apply_mapping_to_table(
    doc: TableDocument,
    orig_to_written: Dict[str, str],
    entity_texts: Iterable[str],
) -> TableDocument:
    """Replace detected entity texts in each cell. Does not use mapping keys."""
    texts = [text for text in entity_texts if text]
    if not texts:
        return doc
    for cell in iter_cells(doc):
        if not cell.search_text:
            continue
        new = replace_entities(cell.search_text, texts, orig_to_written)
        if new != cell.search_text:
            cell.search_text = new
    return doc

pdf_anonymizer_core.tables.flatten_table_for_review(doc: TableDocument, *, anonymized: bool = True) -> str

Row-wise flatten for verify / risk / consolidate.

Blank line after the sheet header and after every row, including the last, so risk windows do not glue a header or the next sheet onto a data row.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/tables.py

def flatten_table_for_review(doc: TableDocument, *, anonymized: bool = True) -> str:
    """Row-wise flatten for verify / risk / consolidate.

    Blank line after the sheet header and after every row, including the last,
    so risk windows do not glue a header or the next sheet onto a data row.
    """
    parts: list[str] = []
    for sheet in doc.sheets:
        parts.append(f"# Sheet: {sheet.name}")
        parts.append("")
        lookup = _cells_by_address(sheet)
        for row in range(1, sheet.max_row + 1):
            values = [
                _flatten_cell_value(lookup.get((row, col)), anonymized=anonymized)
                for col in range(1, sheet.max_column + 1)
            ]
            parts.append(" | ".join(values))
            parts.append("")
    if not parts:
        return ""
    return "\n".join(parts) + "\n"

pdf_anonymizer_core.tables.load_review_text(path: str) -> str

Load text for verify / report. Tables and Word files are flattened.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/tables.py

def load_review_text(path: str) -> str:
    """Load text for ``verify`` / ``report``. Tables and Word files are flattened."""
    if is_rejected_spreadsheet(path):
        raise rejected_spreadsheet_error(path)
    if is_tabular_path(path):
        return flatten_table_for_review(load_table(path), anonymized=True)
    from pdf_anonymizer_core.word import (
        flatten_docx_for_review,
        is_rejected_word,
        is_word_path,
        load_docx,
        rejected_word_error,
    )

    if is_rejected_word(path):
        raise rejected_word_error(path)
    if is_word_path(path):
        return flatten_docx_for_review(load_docx(path), anonymized=True)
    return Path(path).read_text(encoding="utf-8")

---

Word (DOCX)

pdf_anonymizer_core.word.load_docx(path: str) -> WordDocument

Load a .docx file as a WordDocument.

Raises ValueError for rejected Word suffixes, a missing [docx] extra, a file over the size / block cap, or an unreadable package.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/word.py

def load_docx(path: str) -> WordDocument:
    """Load a ``.docx`` file as a ``WordDocument``.

    Raises ``ValueError`` for rejected Word suffixes, a missing ``[docx]``
    extra, a file over the size / block cap, or an unreadable package.
    """
    suffix = Path(path).suffix.lower()
    if suffix in REJECT_WORD_SUFFIXES:
        raise rejected_word_error(path)
    if suffix != ".docx":
        raise ValueError(f"Not a supported Word file: {path}")

    _require_docx()
    from docx import Document

    file_size = os.path.getsize(path)
    if file_size > conf.MAX_DOCX_BYTES:
        raise ValueError(
            f"Word file exceeds the size limit of {conf.MAX_DOCX_BYTES} bytes."
        )

    try:
        document = Document(path)
    except ValueError:
        raise
    except Exception as exc:
        raise ValueError(f"Cannot open Word document {path}") from exc

    for part in _iter_package_parts(document):
        if _is_macro_part(part):
            raise ValueError(
                "Macro-enabled Word documents are not supported "
                "(macros can re-derive PII). Re-save as .docx."
            )

    blocks: list[WordBlock] = []
    nonempty = 0
    for part in _iter_story_parts(document):
        element = getattr(part, "element", None)
        if element is None:
            continue
        part_name = _part_name(part)
        index = 0
        for p_elem in element.iter(W_P):
            text = paragraph_visible_text(p_elem)
            if not text:
                continue
            nonempty += 1
            if nonempty > conf.MAX_DOCX_BLOCKS:
                raise ValueError(
                    f"Word file exceeds the limit of {conf.MAX_DOCX_BLOCKS} "
                    "non-empty paragraphs."
                )
            index += 1
            blocks.append(
                WordBlock(
                    part_name=part_name,
                    index=index,
                    search_text=text,
                    kind="paragraph",
                    _p=p_elem,
                )
            )
        field_index = 0
        for instr in element.iter(W_INSTR):
            raw = instr.text or ""
            if not raw:
                continue
            nonempty += 1
            if nonempty > conf.MAX_DOCX_BLOCKS:
                raise ValueError(
                    f"Word file exceeds the limit of {conf.MAX_DOCX_BLOCKS} "
                    "non-empty paragraphs."
                )
            field_index += 1
            blocks.append(
                WordBlock(
                    part_name=f"{part_name}#fields",
                    index=field_index,
                    search_text=raw,
                    kind="field",
                    _instr=instr,
                )
            )

    link_index = 0
    for rel in _iter_hyperlink_rels(document):
        target = rel.target_ref
        nonempty += 1
        if nonempty > conf.MAX_DOCX_BLOCKS:
            raise ValueError(
                f"Word file exceeds the limit of {conf.MAX_DOCX_BLOCKS} "
                "non-empty paragraphs."
            )
        link_index += 1
        blocks.append(
            WordBlock(
                part_name="hyperlinks",
                index=link_index,
                search_text=target,
                kind="hyperlink",
                _rel=rel,
            )
        )

    return WordDocument(path=path, blocks=blocks, _document=document)

pdf_anonymizer_core.word.save_docx(doc: WordDocument, path: str) -> None

Write the live python-docx package to path.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/word.py

def save_docx(doc: WordDocument, path: str) -> None:
    """Write the live ``python-docx`` package to ``path``."""
    if doc._document is None:
        raise ValueError("Word document has no in-memory package to save.")
    dest = Path(path)
    dest.parent.mkdir(parents=True, exist_ok=True)
    try:
        doc._document.save(path)
    except Exception as exc:
        raise ValueError(f"Cannot write Word document {path}") from exc

pdf_anonymizer_core.word.apply_mapping_to_docx(doc: WordDocument, orig_to_written: Dict[str, str], entity_texts: Iterable[str]) -> WordDocument

Replace detected entity texts in each block. Does not use mapping keys.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/word.py

def apply_mapping_to_docx(
    doc: WordDocument,
    orig_to_written: Dict[str, str],
    entity_texts: Iterable[str],
) -> WordDocument:
    """Replace detected entity texts in each block. Does not use mapping keys."""
    texts = [text for text in entity_texts if text]
    if not texts:
        return doc
    for block in doc.blocks:
        if not block.search_text:
            continue
        new = replace_entities(block.search_text, texts, orig_to_written)
        if new != block.search_text:
            write_block_text(block, new)
    return doc

pdf_anonymizer_core.word.flatten_docx_for_review(doc: WordDocument, *, anonymized: bool = True) -> str

Part-wise flatten for verify / risk / consolidate.

Blank line after the part header and after every block, including the last, so risk windows do not glue a header onto the next paragraph.

Source code in packages/pdf-anonymizer-core/src/pdf_anonymizer_core/word.py

def flatten_docx_for_review(doc: WordDocument, *, anonymized: bool = True) -> str:
    """Part-wise flatten for verify / risk / consolidate.

    Blank line after the part header and after every block, including the last,
    so risk windows do not glue a header onto the next paragraph.
    """
    del anonymized  # blocks already hold the current (maybe replaced) text
    parts: list[str] = []
    current: Optional[str] = None
    for block in doc.blocks:
        if block.part_name != current:
            current = block.part_name
            parts.append(f"# Part: {current}")
            parts.append("")
        parts.append(block.search_text)
        parts.append("")
    if not parts:
        return ""
    return "\n".join(parts) + "\n"