import datetime
import hashlib
import html
import json
import math
import re
def validate_schema(schema):
if not isinstance(schema, list) or not 1 <= len(schema) <= 30:
raise ValueError('Schema needs 1 to 30 fields')
names = set()
aliases = set()
for field in schema:
if not isinstance(field, dict) or not isinstance(field.get('name'), str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,40}', field.get('name', '')):
raise ValueError('Field needs a simple lowercase name')
if field['name'] in names or field.get('type') not in {'text', 'integer', 'number', 'date'} or not isinstance(field.get('required'), bool):
raise ValueError('Duplicate name or invalid type/required flag')
labels = field.get('labels')
if not isinstance(labels, list) or not labels or any(not isinstance(label, str) or not label.strip() or ':' in label or '\n' in label for label in labels):
raise ValueError('Each field needs one or more line labels')
for label in labels:
key = label.strip().casefold()
if key in aliases:
raise ValueError('Labels must be unique across the schema')
aliases.add(key)
names.add(field['name'])
return schema
def extract_candidates(text, schema):
validate_schema(schema)
if not isinstance(text, str) or len(text) > 100000:
raise ValueError('Document must contain at most 100000 characters')
labels = {label.strip().casefold(): field['name'] for field in schema for label in field['labels']}
candidates = []
for match in re.finditer(r'(?m)^([^:\r\n]+):[ \t]*([^\r\n]*)', text):
name = labels.get(match.group(1).strip().casefold())
quote = match.group(2).strip()
if name and quote:
start = match.start(2) + len(match.group(2)) - len(match.group(2).lstrip())
candidates.append({'name': name, 'quote': quote, 'start': start, 'end': start + len(quote)})
return candidates
def typed_value(quote, kind):
if kind == 'text':
return quote
if kind == 'integer':
if not re.fullmatch(r'[+-]?\d+', quote):
raise ValueError('Expected a whole number')
return int(quote)
if kind == 'number':
value = float(quote)
if not math.isfinite(value):
raise ValueError('Expected a finite number')
return value
if kind == 'date':
if not re.fullmatch(r'\d{4}-\d{2}-\d{2}', quote):
raise ValueError('Use an ISO date: YYYY-MM-DD')
return datetime.date.fromisoformat(quote).isoformat()
raise ValueError('Unknown field type')
def review_document(text, schema, candidates, decisions=None):
validate_schema(schema)
if not isinstance(text, str) or len(text) > 100000 or not isinstance(candidates, list) or len(candidates) > 300:
raise ValueError('Invalid document or candidate limits')
digest = hashlib.sha256(text.encode()).hexdigest()
choices = {}
if decisions is not None:
if not isinstance(decisions, dict) or decisions.get('sourceSha256') != digest or not isinstance(decisions.get('choices'), dict):
raise ValueError('Approval belongs to another document version')
choices = decisions['choices']
fields = {field['name']: field for field in schema}
if set(choices) - set(fields):
raise ValueError('Approval names an unknown field')
grouped = {name: [] for name in fields}
seen = set()
for candidate in candidates:
if not isinstance(candidate, dict) or candidate.get('name') not in fields:
raise ValueError('Proposal names an unknown field')
start, end, quote = candidate.get('start'), candidate.get('end'), candidate.get('quote')
if type(start) is not int or type(end) is not int or not 0 <= start < end <= len(text) or text[start:end] != quote:
raise ValueError('Proposal does not match its exact source span')
key = (candidate['name'], start, end)
if key in seen:
continue
seen.add(key)
try:
value = typed_value(quote, fields[candidate['name']]['type'])
issue = None
except ValueError as error:
value, issue = None, str(error)
grouped[candidate['name']].append({**candidate, 'value': value, 'issue': issue})
result, approved = [], {}
for name, field in fields.items():
options = grouped[name]
chosen = choices.get(name)
matches = [c for c in options if chosen == {'start': c['start'], 'end': c['end']} and c['issue'] is None]
if chosen is not None and len(matches) != 1:
raise ValueError('Approval must select one valid source candidate')
if matches:
state = 'approved'
elif not options:
state = 'missing'
elif len(options) > 1:
state = 'ambiguous'
elif options[0]['issue']:
state = 'invalid'
else:
state = 'proposed'
if matches:
approved[name] = matches[0]['value']
result.append({'name': name, 'type': field['type'], 'required': field['required'], 'state': state, 'selected': {'start': matches[0]['start'], 'end': matches[0]['end']} if matches else None, 'candidates': options})
ready = all(not field['required'] or field['state'] == 'approved' for field in result)
return {'schemaVersion': 1, 'sourceSha256': digest, 'status': 'approved' if ready else 'needs_review',
'fields': result, 'approvedValues': approved, 'text': text,
'scope': 'Offsets prove provenance. A reviewer must still confirm that a value belongs to the named field.'}
def render_review(report):
esc = lambda value: html.escape(str(value), quote=True)
fields = []
for field in report['fields']:
options = []
for c in field['candidates']:
disabled = ' disabled' if c['issue'] else ''
checked = ' checked' if field.get('selected') == {'start': c['start'], 'end': c['end']} else ''
options.append('')
fields.append('