"""Pinned public-data preparation, Unicode offsets, and leakage-resistant grouping."""
import ast
import hashlib
import json
import re
from collections import Counter

LABELS = ('first_name','last_name','email','phone_number','street_address','customer_id','account_number')
DATA_REVISION = 'b70ffaf5ff39e079776134c5bf4381f00a9fd1ed'
MODEL_REVISION = '6ea81172465e8b0ad3fddeed32b986cdcdcffcf0'

def digest(text):
    return hashlib.sha256(text.encode('utf-8')).hexdigest()

def validate(row):
    """Exclude invalid rows; safely parse observed literals without evaluating code."""
    try:
        spans = json.loads(row['spans'])
    except (ValueError, TypeError):
        try:
            spans = ast.literal_eval(row['spans'])
        except (ValueError, SyntaxError, TypeError):
            return None, 'parse'
    if not isinstance(row['text'], str) or not isinstance(spans, list):
        return None, 'schema'
    for s in spans:
        if not isinstance(s, dict) or not all(k in s for k in ('start','end','label','text')):
            return None, 'span_schema'
        if type(s['start']) is not int or type(s['end']) is not int or not isinstance(s['label'], str):
            return None, 'span_schema'
        if not 0 <= s['start'] < s['end'] <= len(row['text']):
            return None, 'bounds'
        if row['text'][s['start']:s['end']] != s['text']:
            return None, 'source_slice_mismatch'
    spans.sort(key=lambda s:(s['start'],s['end'],s['label']))
    if any(a['end'] > b['start'] for a,b in zip(spans,spans[1:])):
        return None, 'duplicate_or_overlap'
    return spans, None

def family_key(text, spans):
    # All annotated labels, not only target labels, identify reusable contexts.
    parts=[]; previous=0
    for s in spans:
        parts.extend((text[previous:s['start']], '<'+s['label']+'>')); previous=s['end']
    parts.append(text[previous:])
    return digest(re.sub(r'\s+', ' ', ''.join(parts)).strip().casefold())

def support(rows):
    docs=Counter(); entities=Counter(); negatives=0
    for row in rows:
        labels=[s['label'] for s in row['spans']]
        entities.update(labels); docs.update(set(labels)); negatives += not labels
    return {'documents':len(rows),'negative_documents':negatives,'positive_documents_by_label':dict(docs),'entities_by_label':dict(entities)}