Spaces:
Running on Zero
Running on Zero
| """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 | |
| class CurationRule: | |
| label: str | |
| include: tuple[str, ...] | |
| exclude: tuple[str, ...] = () | |
| 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") | |
| 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)) | |
| 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') | |
| def taxonomy(self): | |
| return self._taxonomy | |
| 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 | |