Data Quality and Governance Questions

import time
import heapq
from typing import Iterable, Dict, Set

def monitor_freshness(stream_iterable: Iterable[dict], freshness_threshold_seconds: float):
    """
    Consumes events with {'entity_id', 'event_time'} (event_time: epoch seconds).
    Yields a tuple (now, set_of_stale_entity_ids) whenever called (example yields periodically).
    This implementation updates state on each event and maintains a min-heap for efficient stale checks.
    """
    last_seen: Dict[str, float] = {}            # entity_id -> last seen timestamp
    heap = []                                   # list of (last_seen, entity_id) for lazy expiry

    for event in stream_iterable:
        eid = event['entity_id']
        t = float(event['event_time'])
        # Update last_seen if event is newer
        prev = last_seen.get(eid)
        if prev is None or t > prev:
            last_seen[eid] = t
            heapq.heappush(heap, (t, eid))

        # Example: on each event, compute stale set (could be triggered less frequently)
        now = time.time()
        cutoff = now - freshness_threshold_seconds

        stale: Set[str] = set()
        # Pop heap entries older than cutoff; verify against last_seen for staleness
        while heap and heap[0][0] <= cutoff:
            ts, eid0 = heapq.heappop(heap)
            # If this heap record matches current last_seen, it's truly stale
            if last_seen.get(eid0) == ts:
                stale.add(eid0)

        yield now, stale

import random
import hashlib

# reproducible RNG for hash params
RNG = random.Random(42)

def _str_hash(s: str) -> int:
    # stable 64-bit hash of token
    return int(hashlib.sha1(s.encode('utf8')).hexdigest(), 16) & ((1 << 61) - 1)

def compute_minhash(tokens, num_hashes):
    """
    tokens: iterable of string tokens (shingles)
    num_hashes: length of signature vector
    returns: list[int] signature (non-negative ints)
    """
    if num_hashes <= 0:
        raise ValueError("num_hashes must be > 0")
    # choose a large prime > max token hash space
    prime = (1 << 61) - 1  # Mersenne-ish prime for 61-bit
    max_hash = prime
    # generate hash function parameters
    params = [(RNG.randrange(1, prime), RNG.randrange(0, prime)) for _ in range(num_hashes)]
    sig = [max_hash] * num_hashes
    # pre-hash tokens to ints
    for t in tokens:
        x = _str_hash(t)
        for i, (a, b) in enumerate(params):
            h = (a * x + b) % prime
            if h < sig[i]:
                sig[i] = h
    return sig

def band_signatures(signature, rows_per_band):
    """
    signature: list[int]
    rows_per_band: number of rows per band
    returns: list of tuples (band_idx, band_hash)
    """
    if rows_per_band <= 0:
        raise ValueError("rows_per_band must be > 0")
    n = len(signature)
    if n % rows_per_band != 0:
        raise ValueError("signature length must be divisible by rows_per_band")
    bands = []
    for band_idx in range(0, n // rows_per_band):
        start = band_idx * rows_per_band
        band = signature[start:start + rows_per_band]
        # combine band into single hash (stable)
        bstr = ",".join(str(x) for x in band)
        bh = int(hashlib.sha1(bstr.encode('utf8')).hexdigest(), 16)
        bands.append((band_idx, bh))
    return bands