Privacy Record Linkage

"""
Privacy-preserving record linkage using Bloom filter encoding.
Implements the approach described by Schnell, Bachteler, and Reiher (2009).
"""

import hashlib
import hmac
import math
from typing import Optional
import numpy as np


class BloomFilterEncoder:
    """
    Encode string attributes into Bloom filters for privacy-preserving
    record linkage using cryptographic keyed hashing.
    """

    def __init__(
        self,
        filter_size: int = 1024,
        num_hash_functions: int = 30,
        ngram_size: int = 2,
        secret_key: bytes = b""
    ):
        """
        Args:
            filter_size: Number of bits in the Bloom filter
            num_hash_functions: Number of hash functions (k)
            ngram_size: Size of character n-grams (bigrams = 2)
            secret_key: Shared secret key for HMAC hashing
        """
        self.filter_size = filter_size
        self.num_hash_functions = num_hash_functions
        self.ngram_size = ngram_size
        self.secret_key = secret_key

    def _generate_ngrams(self, value: str) -> list[str]:
        """Generate character n-grams from a string value."""
        # Pad the string to handle edge characters
        padded = f"_{value}_"
        return [
            padded[i:i + self.ngram_size]
            for i in range(len(padded) - self.ngram_size + 1)
        ]

    def _hash_ngram(self, ngram: str, hash_index: int) -> int:
        """
        Hash an n-gram using HMAC with the shared key and hash index.
        Returns a bit position in the Bloom filter.
        """
        message = f"{hash_index}:{ngram}".encode("utf-8")
        digest = hmac.new(self.secret_key, message, hashlib.sha256).digest()
        # Convert first 4 bytes to integer and modulo by filter size
        position = int.from_bytes(digest[:4], byteorder="big") % self.filter_size
        return position

    def encode_value(self, value: str) -> np.ndarray:
        """
        Encode a single attribute value into a Bloom filter.

        Args:
            value: The string value to encode (e.g., a name)

        Returns:
            Numpy array of bits (0/1) representing the Bloom filter
        """
        bloom_filter = np.zeros(self.filter_size, dtype=np.uint8)

        # Normalize the input
        normalized = value.strip().lower()

        # Generate n-grams
        ngrams = self._generate_ngrams(normalized)

        # Hash each n-gram with each hash function
        for ngram in ngrams:
            for h in range(self.num_hash_functions):
                position = self._hash_ngram(ngram, h)
                bloom_filter[position] = 1

        return bloom_filter

    def encode_record(self, attributes: dict[str, str]) -> np.ndarray:
        """
        Encode multiple attributes into a single composite Bloom filter
        using Cryptographic Longterm Key (CLK) approach.

        Args:
            attributes: Dict mapping attribute names to values
                       e.g., {"first_name": "John", "last_name": "Smith", "dob": "1990-01-15"}

        Returns:
            Composite Bloom filter as numpy array
        """
        composite = np.zeros(self.filter_size, dtype=np.uint8)

        for attr_name, attr_value in attributes.items():
            if attr_value:
                # Use attribute name as additional salt
                salted_key = self.secret_key + attr_name.encode("utf-8")
                encoder = BloomFilterEncoder(
                    filter_size=self.filter_size,
                    num_hash_functions=self.num_hash_functions,
                    ngram_size=self.ngram_size,
                    secret_key=salted_key
                )
                attr_bf = encoder.encode_value(attr_value)
                composite = np.bitwise_or(composite, attr_bf)

        return composite


class BloomFilterMatcher:
    """
    Compare Bloom filter-encoded records to find matching pairs.
    """

    @staticmethod
    def dice_coefficient(bf1: np.ndarray, bf2: np.ndarray) -> float:
        """
        Calculate the Dice coefficient between two Bloom filters.

        Dice = 2 * |bf1 AND bf2| / (|bf1| + |bf2|)

        Returns a similarity score between 0 and 1.
        """
        intersection = np.sum(np.bitwise_and(bf1, bf2))
        cardinality_sum = np.sum(bf1) + np.sum(bf2)

        if cardinality_sum == 0:
            return 0.0

        return 2.0 * intersection / cardinality_sum

    @staticmethod
    def jaccard_similarity(bf1: np.ndarray, bf2: np.ndarray) -> float:
        """
        Calculate the Jaccard similarity between two Bloom filters.

        Jaccard = |bf1 AND bf2| / |bf1 OR bf2|
        """
        intersection = np.sum(np.bitwise_and(bf1, bf2))
        union = np.sum(np.bitwise_or(bf1, bf2))

        if union == 0:
            return 0.0

        return intersection / union

    def find_matches(
        self,
        encodings_a: list[tuple[str, np.ndarray]],
        encodings_b: list[tuple[str, np.ndarray]],
        threshold: float = 0.8,
        similarity_metric: str = "dice"
    ) -> list[tuple[str, str, float]]:
        """
        Find matching record pairs between two encoded datasets.

        Args:
            encodings_a: List of (record_id, bloom_filter) from organization A
            encodings_b: List of (record_id, bloom_filter) from organization B
            threshold: Minimum similarity score for a match
            similarity_metric: "dice" or "jaccard"

        Returns:
            List of (id_a, id_b, similarity_score) for matching pairs
        """
        metric_fn = (
            self.dice_coefficient if similarity_metric == "dice"
            else self.jaccard_similarity
        )

        matches = []

        for id_a, bf_a in encodings_a:
            best_score = 0.0
            best_id_b = None

            for id_b, bf_b in encodings_b:
                score = metric_fn(bf_a, bf_b)
                if score > best_score:
                    best_score = score
                    best_id_b = id_b

            if best_score >= threshold and best_id_b is not None:
                matches.append((id_a, best_id_b, best_score))

        return matches

"""
Secure hash-based record linkage for exact matching.
Uses keyed HMAC to prevent rainbow table attacks.
"""

import hashlib
import hmac


class SecureHashLinker:
    """
    Link records across organizations using keyed hash matching.
    Suitable for exact match on standardized identifiers.
    """

    def __init__(self, shared_key: bytes):
        self.shared_key = shared_key

    def hash_identifier(self, *fields: str) -> str:
        """
        Create a keyed hash of concatenated identifier fields.

        Args:
            fields: Identifier fields in standardized order
                    e.g., ("john", "smith", "19900115")

        Returns:
            Hex-encoded HMAC-SHA256 hash
        """
        # Normalize and concatenate fields
        normalized = "|".join(f.strip().lower() for f in fields)

        # Generate keyed hash
        digest = hmac.new(
            self.shared_key,
            normalized.encode("utf-8"),
            hashlib.sha256
        ).hexdigest()

        return digest

    def hash_dataset(
        self,
        records: list[dict],
        id_field: str,
        linkage_fields: list[str]
    ) -> dict[str, str]:
        """
        Hash all records in a dataset for linkage.

        Returns mapping of hash -> record_id.
        """
        hash_map = {}

        for record in records:
            fields = [str(record.get(f, "")) for f in linkage_fields]
            record_hash = self.hash_identifier(*fields)
            hash_map[record_hash] = record[id_field]

        return hash_map

    @staticmethod
    def find_exact_matches(
        hashes_a: dict[str, str],
        hashes_b: dict[str, str]
    ) -> list[tuple[str, str]]:
        """
        Find exact matches between two hash maps.

        Returns list of (id_a, id_b) pairs.
        """
        common_hashes = set(hashes_a.keys()) & set(hashes_b.keys())
        return [(hashes_a[h], hashes_b[h]) for h in common_hashes]