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) -> 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 performs length-descending safe replacement 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, or .txt).	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

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,
) -> 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 performs length-descending safe
    replacement 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, or .txt).
        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.

    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

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

    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")

    # Accumulate all entities from all chunks
    collected_entities: List[dict] = []

    for i, text_page in enumerate(text_pages):
        logging.info(f"Identifying entities in part {i + 1}/{len(text_pages)}...")
        start_time = time.time()

        # 1st Stage: Regex NER
        regex_entities = extract_entities_via_regex(text_page, regex_patterns)

        # 2nd Stage: LLM NER
        llm_entities = identify_entities_with_llm(
            text_page,
            prompt_template,
            model_name,
            max_retries=max_retries,
            base_retry_delay=base_retry_delay,
            max_retry_delay=max_retry_delay,
        )

        end_time = time.time()
        duration = end_time - start_time
        minutes = int(duration // 60)
        seconds = int(duration % 60)
        logging.info(f"   NER stage duration (Regex + LLM): {minutes}:{seconds:02d}")
        logging.info(
            f"   Found {len(regex_entities)} via Regex, {len(llm_entities)} via LLM."
        )

        collected_entities.extend(regex_entities)
        collected_entities.extend(llm_entities)

    # Deduplicate entities by text, prioritizing more specific types if matched multiple times.
    # Higher numbers win. Financial, government ID, crypto, and network identifiers
    # receive elevated priority because they are high-risk and unambiguous.
    type_priority = {
        # Highest sensitivity / unambiguous structured PII (regex-first wins)
        "CREDIT_CARD": 15,
        "IBAN": 14,
        "CRYPTO_BTC": 13,
        "CRYPTO_ETH": 13,
        "EMAIL": 12,
        "SSN": 11,
        "SSN_US": 11,
        "SIN_CA": 11,
        "NINO_GB": 11,
        "INSEE_FR": 11,
        "AADHAAR_IN": 11,
        "RESIDENT_ID_CN": 11,
        "EIN_US": 10,
        "VAT_GB": 10,
        "VAT_FR": 10,
        "VAT_ES": 10,
        "VAT_IT": 10,
        "VAT_DE": 10,
        "PAN_IN": 10,
        "GSTIN_IN": 10,
        "UNIFIED_SOCIAL_CREDIT_CODE_CN": 10,
        "IPV4_ADDRESS": 9,
        "IP_ADDRESS": 9,  # legacy
        "IPV6_ADDRESS": 9,
        "MAC_ADDRESS": 8,
        "VIN": 8,
        "MEDICAL_NPI_US": 8,
        "PASSPORT": 7,
        "US_PASSPORT": 7,
        "GB_PASSPORT": 7,
        "DRIVERS_LICENSE_US": 6,
        "DRIVERS_LICENSE_GB": 6,
        "DRIVERS_LICENSE_FR": 6,
        "DRIVERS_LICENSE_CA": 6,
        "DATE_ISO": 5,
        "CURRENCY_AMOUNT": 5,
        "BIC_SWIFT": 5,
        "PHONE": 4,
        "URL": 4,
        # LLM-detected semantic types (lower than strong regex matches)
        "PERSON": 3,
        "ORGANIZATION": 2,
        "LOCATION": 1,
        "ADDRESS": 1,
    }

    best_entities: Dict[str, dict] = {}
    for ent in collected_entities:
        text = ent["text"]
        ent_type = ent["type"].upper()
        if text not in best_entities:
            best_entities[text] = ent
        else:
            existing_type = best_entities[text]["type"].upper()
            if type_priority.get(ent_type, 0) > type_priority.get(existing_type, 0):
                best_entities[text] = ent

    deduped_entities = list(best_entities.values())

    entities_to_process = deduped_entities
    if anonymized_entities:
        anonymized_entities_upper = [e.upper() for e in anonymized_entities]
        entities_to_process = [
            e
            for e in deduped_entities
            if e["type"].upper() in anonymized_entities_upper
        ]

    logging.info(
        f"Collected {len(collected_entities)} total entities. "
        f"Deduplicated to {len(deduped_entities)}. "
        f"Processing {len(entities_to_process)} filtered entities."
    )

    # Consolidate base forms to handle variations like "John" vs "John Doe"
    base_forms = {e.get("base_form") for e in entities_to_process if e.get("base_form")}
    sorted_base_forms = sorted(list(base_forms), key=len, reverse=True)
    for entity in entities_to_process:
        base_form = entity.get("base_form")
        if not base_form:
            continue
        for potential_full_form in sorted_base_forms:
            if base_form != potential_full_form and base_form in potential_full_form:
                entity["base_form"] = potential_full_form
                break

    # Generate placeholders
    final_mapping: Dict[str, str] = {}
    placeholder_counts: Dict[str, int] = {}
    base_entity_placeholders: Dict[str, str] = {}
    variation_counters: Dict[str, int] = {}

    for entity in entities_to_process:
        entity_text = entity["text"]
        entity_type = entity["type"].upper()
        base_form = entity.get("base_form") or entity_text

        if entity_text in final_mapping:
            continue

        if base_form not in base_entity_placeholders:
            # New base entity, create main placeholder
            current_count = placeholder_counts.get(entity_type, 0) + 1
            placeholder_counts[entity_type] = current_count
            main_placeholder = f"{entity_type}_{current_count}"
            base_entity_placeholders[base_form] = main_placeholder
            if base_form not in final_mapping:
                final_mapping[base_form] = main_placeholder

        main_placeholder = base_entity_placeholders[base_form]

        if entity_text != base_form:
            # It's a variation, create variation placeholder
            current_variation_count = variation_counters.get(main_placeholder, 0) + 1
            variation_counters[main_placeholder] = current_variation_count
            variation_placeholder = f"{main_placeholder}.v_{current_variation_count}"
            final_mapping[entity_text] = variation_placeholder
        else:
            final_mapping[entity_text] = main_placeholder

    anonymized_text = full_text
    if entities_to_process:
        # Sort entities by length descending to match longer strings first
        entities_to_process.sort(key=lambda e: len(e["text"]), reverse=True)

        # Build selective boundary checks to prevent partial matching (e.g. "John" inside "Johnson")
        def make_boundary_pattern(text: str) -> str:
            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}"

        pattern = re.compile(
            "|".join(make_boundary_pattern(e["text"]) for e in entities_to_process)
        )
        anonymized_text = pattern.sub(
            lambda m: final_mapping[m.group(0)], anonymized_text
        )

    return anonymized_text, final_mapping

pdf_anonymizer_core.utils.deanonymize_file(anonymized_file_path: str, mapping_file_path: str) -> 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.	required

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
) -> 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.

    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(anonymized_file_path, "r", encoding="utf-8") as f:
        anonymized_text = f.read()

    with open(mapping_file_path, "r", encoding="utf-8") as f:
        raw_mapping = json.load(f)

    # Detect mapping direction and normalize to placeholder -> original
    # Heuristic: if most keys look like placeholders (e.g., PERSON_1), treat as placeholder->original
    placeholder_key_pattern = _PLACEHOLDER_PATTERN
    keys_look_like_placeholders = sum(
        1
        for k in raw_mapping.keys()
        if isinstance(k, str) and placeholder_key_pattern.match(k)
    )
    values_look_like_placeholders = sum(
        1
        for v in raw_mapping.values()
        if isinstance(v, str) and placeholder_key_pattern.match(v)
    )

    if keys_look_like_placeholders >= values_look_like_placeholders:
        placeholder_to_original = dict(raw_mapping)
    else:
        # Legacy: invert original -> placeholder to placeholder -> original
        placeholder_to_original = {}
        for original, placeholder in raw_mapping.items():
            if isinstance(placeholder, str):
                placeholder_to_original.setdefault(placeholder, original)

    deanonymized_text = anonymized_text
    used_placeholders = set()  # track actual placeholders (including variations) found

    # Replace placeholders by longest first to avoid partial overlaps
    sorted_placeholders = sorted(placeholder_to_original.keys(), key=len, reverse=True)

    if sorted_placeholders:
        # Match base and its variations: PERSON_1 and PERSON_1.v_1, PERSON_1.v_2, ...
        # We use a single regex pass for all placeholders to avoid ReDoS and performance bottlenecks.
        pattern = re.compile(
            r"\b("
            + "|".join(re.escape(p) for p in sorted_placeholders)
            + r")(?:\.v_\d+)?\b"
        )

        def replace_match(match):
            full_match = match.group(0)
            base_placeholder = match.group(1)
            used_placeholders.add(full_match)
            return placeholder_to_original[base_placeholder]

        deanonymized_text = pattern.sub(replace_match, deanonymized_text)

    # 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}"
    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)

    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) -> 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 of original text -> placeholder.	required
file_path	str	The path to the original file.	required

Returns:

Type	Description
tuple[str, str]	tuple[str, str]: The paths to the anonymized text file and the mapping file.

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
) -> 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 of original text -> placeholder.
        file_path (str): The path to the original file.

    Returns:
        tuple[str, str]: The paths to the anonymized text file and the mapping file.
    """
    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)
    os.makedirs(mappings_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}"
    )
    with open(anonymized_output_file, "w", encoding="utf-8") as f:
        f.write(full_anonymized_text)

    mapping_file = f"{mappings_dir}/{file_stem}.mapping.json"
    # Persist mapping as placeholder -> original for correct deanonymization
    with open(mapping_file, "w", encoding="utf-8") as f:
        json.dump(final_mapping, f, indent=4)

    return anonymized_output_file, mapping_file

---

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. - Three built-in ConfigProfiles (best-quality, best-speed, best-cost) that bundle sensible combinations of model, prompt, chunk size, retries, etc. - 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.

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) -> 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 or BEST_COST.	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

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,
) -> 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 or BEST_COST.
        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.

    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"],
    )

---

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, and DATE (birthdates only) - Explicitly handle variations and possessives

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).

---

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", and optionally "base_form").
List[dict]	Returns [] on unrecoverable 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", and optionally "base_form").
        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)

            return [entity.model_dump() for entity in result.entities]

        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.

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) -> 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
) -> 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()

    try:
        if file_extension == ".pdf":
            return load_and_extract_text_from_pdf(
                file_path, characters_to_anonymize, chunk_overlap
            )
        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) -> 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
) -> 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)
        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]
    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.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. 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).

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).

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).

    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).
    """
    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

                entities.append(
                    {
                        "text": matched_text,
                        "type": entity_type.upper(),
                        "base_form": matched_text,
                    }
                )
        except re.error as e:
            logging.error(f"Invalid regex pattern configured for {entity_type}: {e}")

    return entities