src/range_curate/curation.py

"""Domain-neutral, explicit-taxonomy text curation; no timestamps required."""
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from itertools import islice
from typing import Iterable

import yaml


@dataclass(frozen=True)
class CurationRule:
    label: str
    include: tuple[str, ...]
    exclude: tuple[str, ...] = ()


@dataclass(frozen=True)
class CurationTaxonomy:
    name: str
    rules: tuple[CurationRule, ...]

    def __post_init__(self):
        if not isinstance(self.name, str) or not self.name.strip():
            raise ValueError("taxonomy name must be nonempty text")
        if not isinstance(self.rules, tuple) or not self.rules:
            raise ValueError("rules must be a nonempty tuple of CurationRule values")
        labels = set()
        for rule in self.rules:
            if not isinstance(rule, CurationRule) or not isinstance(rule.label, str) or not rule.label.strip() or rule.label in labels:
                raise ValueError("rule labels must be nonempty and unique")
            labels.add(rule.label)
            for values in (rule.include, rule.exclude):
                if not isinstance(values, tuple) or any(not isinstance(v, str) or not v.strip() for v in values):
                    raise ValueError("phrases must be tuples of nonempty strings")
            if not rule.include:
                raise ValueError("every rule requires an include phrase")

    @classmethod
    def from_yaml(cls, text: str) -> CurationTaxonomy:
        if not isinstance(text, str):
            raise TypeError("YAML must be text, not a file path or object")
        data = yaml.safe_load(text)
        if not isinstance(data, dict) or set(data) != {"version", "name", "rules"} or type(data['version']) is not int or data['version'] != 1:
            raise ValueError("expected version: 1, name, and rules only")
        if not isinstance(data['rules'], list):
            raise ValueError("rules must be a list")
        rules = []
        for rule in data['rules']:
            if not isinstance(rule, dict) or not {'label', 'include'} <= set(rule) or set(rule) - {'label', 'include', 'exclude'}:
                raise ValueError("each rule needs label/include and optional exclude")
            include, exclude = rule['include'], rule.get('exclude', [])
            if not isinstance(include, list) or not isinstance(exclude, list):
                raise ValueError("include/exclude must be lists")
            rules.append(CurationRule(rule['label'], tuple(include), tuple(exclude)))
        return cls(data['name'], tuple(rules))

    @property
    def fingerprint(self) -> str:
        payload = [1, self.name, [(r.label, r.include, r.exclude) for r in self.rules], 'range-frozen-unicode13-token-boundaries-v1']
        return hashlib.sha256(json.dumps(payload, ensure_ascii=True, separators=(',', ':')).encode()).hexdigest()


class CurationEngine:
    """Any include phrase selects a label; any exclude phrase vetoes that label.



    Exclusions apply across the whole document. Different labels are independent.

    Matching is normalized literal matching, not semantic or negation inference.

    """
    __slots__ = ('_taxonomy', '_workers', '_scanner')

    @property
    def taxonomy(self):
        return self._taxonomy

    @property
    def workers(self):
        return self._workers

    def __init__(self, taxonomy: CurationTaxonomy, *, workers: int = 1):
        if not isinstance(taxonomy, CurationTaxonomy):
            raise TypeError("supply an explicit CurationTaxonomy")
        if type(workers) is not int or not 1 <= workers <= 256:
            raise ValueError("workers must be between 1 and 256")
        from ._native import NativeCurationScanner
        self._taxonomy = taxonomy
        self._workers = workers
        self._scanner = NativeCurationScanner([(list(r.include), list(r.exclude)) for r in taxonomy.rules])

    def screen_batch(self, texts: list[str], *, evidence: bool = False, max_evidence: int = 1000) -> list[dict]:
        if not isinstance(texts, list) or any(not isinstance(t, str) for t in texts):
            raise TypeError("texts must be a list of strings; use curate_batches for a stream")
        if type(evidence) is not bool or type(max_evidence) is not int or not 1 <= max_evidence <= 10000:
            raise ValueError("evidence must be boolean and max_evidence between 1 and 10000")
        rows = self._scanner.scan(texts, self.workers, max_evidence if evidence else 0)
        fingerprint = self.taxonomy.fingerprint
        result = []
        for index, (labels, hits, truncated) in enumerate(rows):
            result.append({'row_index': index, 'keep': bool(labels),
                'labels': [self.taxonomy.rules[i].label for i in labels],
                'taxonomy_fingerprint': fingerprint, 'evidence_truncated': truncated,
                'evidence': [{'label': self.taxonomy.rules[i].label,
                    'kind': 'exclude' if negative else 'include', 'start': start, 'end': end,
                    'text': texts[index][start:end]} for i, negative, start, end in hits]})
        return result

    def curate(self, text: str, *, max_evidence: int = 1000) -> dict:
        return self.screen_batch([text], evidence=True, max_evidence=max_evidence)[0]

    def curate_batches(self, texts: Iterable[str], *, batch_size: int = 4096):
        """Yield bounded batches of decisions with stream-global row indices."""
        if type(batch_size) is not int or not 1 <= batch_size <= 100000:
            raise ValueError("batch_size must be between 1 and 100000")
        if isinstance(texts, (str, bytes)):
            raise TypeError("texts must be an iterable of documents")
        source = iter(texts); offset = 0
        while batch := list(islice(source, batch_size)):
            rows = self.screen_batch(batch)
            for row in rows: row['row_index'] += offset
            offset += len(batch)
            yield rows