"""Prueft die dokumentierte F11-Einzelschrittregel, nur Python-Standardbibliothek. Aufruf: python f11_validieren.py [f11_ausfuellbeispiel.json] [--self-test] Keine Netzaufrufe, keine Dateiaenderungen. JSON-Bericht nach stdout. Kein allgemeiner Archiv-, OASST1- oder Herkunftsvalidator. """ import argparse from collections import Counter, defaultdict from copy import deepcopy import json from pathlib import Path import re def words(record): text = record.get('original_text') return len(re.findall(r'\b\w+\b', text)) if isinstance(text, str) else 0 def validate(data): errors, pairs, decisions = [], [], [] records = data['records'] by_id = defaultdict(list) record_counts = Counter(r['record_id'] for r in records) for r in records: by_id[r['message_id']].append(r) def error(code, record_id, detail): errors.append({'code': code, 'record_id': record_id, 'detail': detail}) for key, rows in by_id.items(): if not isinstance(key, str) or not key: error('invalid_message_id', None, str(key)) if len(rows) > 1: error('duplicate_message_id', None, str(key)) for key, count in record_counts.items(): if not isinstance(key, str) or not key: error('invalid_record_id', None, str(key)) if count > 1: error('duplicate_record_id', key, str(count)) gap_ids = Counter(g['gap_id'] for g in data['gaps']) for key, count in gap_ids.items(): if count > 1: error('duplicate_gap_id', None, key) used_gaps = set() used_records = set() contexts = set() for r in records: if r['speaker_role'] not in ('user', 'assistant'): error('unsupported_role', r['record_id'], r['speaker_role']) if not isinstance(r['session_id'], str) or not r['session_id']: error('invalid_session_id', r['record_id'], str(r['session_id'])) if r['speaker_role'] != 'assistant': continue reasons = [] parent_rows = by_id.get(r['parent_id'], []) parent = parent_rows[0] if len(parent_rows) == 1 else None if len(by_id[r['message_id']]) != 1 or record_counts[r['record_id']] != 1: reasons.append('ambiguous_answer') if not parent_rows: reasons.append('missing_parent') matches = [g for g in data['gaps'] if g['gap_id'] == r['gap_reference'] and g['missing_reference'] == r['parent_id'] and g['referenced_from_record_id'] == r['record_id'] and g['missing_material_type'] == 'user_parent_message'] if len(matches) != 1: error('missing_or_ambiguous_gap', r['record_id'], str(r['parent_id'])) else: used_gaps.add(matches[0]['gap_id']) elif parent is None or record_counts[parent['record_id']] != 1: reasons.append('ambiguous_parent') else: if parent['session_id'] != r['session_id']: reasons.append('cross_session_parent') if parent['speaker_role'] != 'user': reasons.append('parent_not_user') if parent['parent_id'] is not None: reasons.append('context_not_single_turn_root') for candidate in (r, parent): if candidate is not None and (candidate['original_text_available'] is not True or not isinstance(candidate['original_text'], str) or not candidate['original_text'].strip()): reasons.append('text_unavailable') if parent_rows and r['gap_reference'] is not None: error('gap_for_existing_parent', r['record_id'], r['gap_reference']) for code in sorted(set(reasons) - {'missing_parent'}): error(code, r['record_id'], str(r['parent_id'])) included = not reasons expected = 'included_pair' if included else 'excluded_pair' if r['inclusion']['decision'] != expected: error('inclusion_mismatch', r['record_id'], expected) decision = {'answer_id': r['message_id'], 'parent_id': r['parent_id'], 'session_id': r['session_id'], 'included': included, 'reasons': sorted(set(reasons)), 'pair_words': None} if included: decision['pair_words'] = words(r) + words(parent) pairs.append(decision) used_records.update([r['record_id'], parent['record_id']]) contexts.add((parent['session_id'], parent['message_id'])) decisions.append(decision) for g in data['gaps']: if g['gap_id'] not in used_gaps: error('unmatched_gap', g['referenced_from_record_id'], g['gap_id']) if g['missing_reference'] in by_id: error('gap_message_is_present', g['referenced_from_record_id'], g['missing_reference']) for r in records: if r['speaker_role'] == 'user': expected = 'included_context' if r['record_id'] in used_records else 'not_used_context' if r['inclusion']['decision'] != expected: error('context_inclusion_mismatch', r['record_id'], expected) counts = { 'observed_messages': len(records), 'gap_entries': len(data['gaps']), 'session_labels_present': len({r['session_id'] for r in records}), 'candidate_answers': len(decisions), 'eligible_pairs': len(pairs), 'excluded_answers': len(decisions) - len(pairs), 'distinct_eligible_user_contexts': len(contexts), 'all_observed_message_words': sum(words(r) for r in records), 'eligible_unique_message_words': sum(words(r) for r in records if r['record_id'] in used_records), 'pair_words_with_shared_context_repeated': sum(p['pair_words'] for p in pairs), } return {'result': 'PASS' if not errors else 'FAIL', 'scope': 'Declared single-turn parent/session/duplicate/gap/text/inclusion checks; no factual or model-provenance verification.', 'counts': counts, 'answers': decisions, 'errors': errors} def self_test(data): cases = [] def check(name, changed, required): report = validate(changed) codes = {e['code'] for e in report['errors']} cases.append({'case': name, 'expected_error': required, 'detected': required in codes, 'result': report['result']}) m = deepcopy(data) m['records'][1]['parent_id'] = 'absent_test_parent' check('parent_reference_without_gap', m, 'missing_or_ambiguous_gap') m = deepcopy(data) m['records'][1]['session_id'] = 'S3' check('parent_in_different_session', m, 'cross_session_parent') m = deepcopy(data) m['records'].append(deepcopy(m['records'][0])) check('duplicate_parent_message_and_record', m, 'duplicate_message_id') m = deepcopy(data) m['records'][1]['record_id'] = m['records'][0]['record_id'] check('duplicate_record_id_only', m, 'duplicate_record_id') m = deepcopy(data) m['records'][4]['parent_id'] = 'a4' check('assistant_as_parent', m, 'parent_not_user') m = deepcopy(data) m['records'][0]['parent_id'] = 'u3' check('uncovered_earlier_context', m, 'context_not_single_turn_root') m = deepcopy(data) m['gaps'][0]['missing_reference'] = 'u1' check('gap_points_at_observed_message', m, 'gap_message_is_present') m = deepcopy(data) m['records'][3]['original_text_available'] = False check('parent_text_unavailable', m, 'text_unavailable') m = deepcopy(data) m['records'][5]['inclusion']['decision'] = 'excluded_pair' check('incorrect_manual_inclusion', m, 'inclusion_mismatch') # Same parent is allowed: it is not a duplicate message. Count the context once. baseline = validate(data) siblings = [a for a in baseline['answers'] if a['parent_id'] == 'u3' and a['included']] cases.append({'case': 'valid_sibling_answers_share_one_context', 'detected': len(siblings) == 2 and baseline['counts']['distinct_eligible_user_contexts'] == 2, 'result': baseline['result']}) return {'result': 'PASS' if all(c['detected'] for c in cases) else 'FAIL', 'note': 'Nine invalid local in-memory copies and one valid sibling check; source file unchanged.', 'cases': cases} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('path', nargs='?', type=Path, default=Path(__file__).with_name('f11_ausfuellbeispiel.json')) parser.add_argument('--self-test', action='store_true') args = parser.parse_args() try: data = json.loads(args.path.read_text(encoding='utf-8')) report = validate(data) if args.self_test: report['self_test'] = self_test(data) except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc: report = {'result': 'FAIL', 'input_error': str(exc)} print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if report['result'] == 'PASS' and report.get('self_test', {}).get('result', 'PASS') == 'PASS' else 1 if __name__ == '__main__': raise SystemExit(main())