#!/usr/bin/env python3
"""Offline, synthetic split audit. No model calls or semantic deduplication.

Run without arguments for the demonstration and tests. Pass a JSON manifest
path to audit your own records; exit 0 means this narrow audit passed, 1 means
it failed. Fields: case_id, family_id, split (dev|test), question, answer.
Family IDs must be assigned upstream from verified source lineage.
"""
import hashlib
import json
import sys
import unittest


def fingerprint(row):
    # Whitespace-only normalisation; preserve case, punctuation and meaning.
    pair = [' '.join(row[k].split()) for k in ('question', 'answer')]
    return hashlib.sha256(json.dumps(pair, ensure_ascii=False).encode()).hexdigest()


def audit(rows):
    if not isinstance(rows, list) or not rows:
        return ['manifest must be a nonempty list']
    errors, ids, families, pairs, splits = [], set(), {}, {}, set()
    for index, row in enumerate(rows):
        required = ('case_id', 'family_id', 'split', 'question', 'answer')
        if not isinstance(row, dict) or any(
            not isinstance(row.get(k), str) or not row[k].strip() for k in required
        ):
            errors.append(f'row {index}: missing or invalid required text')
            continue
        cid, family, split = row['case_id'], row['family_id'], row['split']
        if any(row[k] != row[k].strip() for k in ('case_id', 'family_id', 'split')):
            errors.append(f'row {index}: padded identifier')
        if split not in ('dev', 'test'):
            errors.append(f'row {index}: unknown split')
            continue
        if cid in ids:
            errors.append(f'row {index}: duplicate case_id')
        ids.add(cid)
        splits.add(split)
        families.setdefault(family, set()).add(split)
        pairs.setdefault(fingerprint(row), set()).add(split)
    if splits != {'dev', 'test'}:
        errors.append('both dev and test must be present')
    if any(len(s) > 1 for s in families.values()):
        errors.append('source family crosses dev/test boundary')
    if any(len(s) > 1 for s in pairs.values()):
        errors.append('normalised question/answer pair crosses dev/test boundary')
    return sorted(set(errors))


def case(cid, family, split, question='What is the limit?', answer='Example limit'):
    return dict(case_id=cid, family_id=family, split=split, question=question, answer=answer)


def good():
    return [case('a', 'refund-policy', 'dev', answer='Refund within 30 days'),
            case('b', 'travel-policy', 'test', answer='Travel limit is 100')]


class Tests(unittest.TestCase):
    def test_disjoint(self):
        self.assertEqual(audit(good()), [])

    def test_family_across_versions(self):
        rows = good()
        rows[1]['family_id'] = rows[0]['family_id']
        self.assertIn('source family crosses dev/test boundary', audit(rows))

    def test_pair_despite_false_family(self):
        rows = good()
        rows[1]['answer'] = ' Refund  within 30 days '
        self.assertIn('normalised question/answer pair crosses dev/test boundary', audit(rows))

    def test_duplicate_id(self):
        rows = good()
        rows[1]['case_id'] = 'a'
        self.assertIn('row 1: duplicate case_id', audit(rows))

    def test_invalid_records(self):
        for rows in ([], {}, [None], [dict(good()[0], family_id='')],
                     [dict(good()[0], question=4)]):
            with self.subTest(rows=rows):
                self.assertTrue(audit(rows))

    def test_missing_split(self):
        self.assertIn('both dev and test must be present', audit(good()[:1]))

    def test_unknown_split(self):
        rows = good()
        rows[1]['split'] = 'holdout'
        self.assertIn('row 1: unknown split', audit(rows))

    def test_padded_identifier(self):
        rows = good()
        rows[1]['family_id'] = ' travel-policy'
        self.assertIn('row 1: padded identifier', audit(rows))

    def test_semantic_similarity_is_not_detected(self):
        rows = good()
        rows[1].update(question='When can I get a refund?', answer='Within thirty days')
        # Wrong lineage plus paraphrasing escapes this narrow audit.
        self.assertEqual(audit(rows), [])


if __name__ == '__main__':
    if len(sys.argv) > 1:
        try:
            with open(sys.argv[1], encoding='utf-8') as f:
                errors = audit(json.load(f))
        except (OSError, ValueError) as exc:
            print(f'BLOCK: cannot read manifest ({type(exc).__name__})')
            sys.exit(1)
        print('BLOCK: ' + '; '.join(errors) if errors else 'PASS: narrow split audit')
        sys.exit(1 if errors else 0)
    contaminated = good()
    contaminated[1]['family_id'] = 'refund-policy'
    print('shared family:', 'BLOCK' if audit(contaminated) else 'PASS', flush=True)
    print('disjoint families:', 'BLOCK' if audit(good()) else 'PASS', flush=True)
    unittest.main(argv=[sys.argv[0]], verbosity=2)