"""Original bounded CSV structural preflight. Local only; no network or edits.""" import csv import io import json import sys def check(input_value): if not isinstance(input_value, dict) or set(input_value) != {'csv_text'}: raise ValueError('Supply exactly {csv_text: public or redacted CSV string}.') text = input_value['csv_text'] if not isinstance(text, str) or not text or len(text) > 8000: raise ValueError('CSV must contain 1-8000 characters.') if len(text.encode('utf-8')) > 16000: raise ValueError('CSV text exceeds16000UTF8bytes.') try: records = list(csv.reader(io.StringIO(text, newline=''), strict=True)) except csv.Error as exc: return {'status': 'parse_error', 'parser_error': str(exc), 'complete_record_counts_verified': False} if not records: raise ValueError('CSV has no logical header record.') header, rows = records[0], records[1:] observed = {'data_records': len(rows), 'header_fields': len(header), 'maximum_record_fields': max(map(len, records))} limits = {'csv_characters': 8000, 'csv_utf8_bytes': 16000, 'data_records': 200, 'record_fields': 10} if len(rows) > 200 or observed['maximum_record_fields'] > 10: return {'status': 'out_of_scope', 'observed': observed, 'limits': limits, 'complete_record_counts_verified': True, 'findings_generated': False} positions = {} for n, value in enumerate(header, 1): positions.setdefault(value, []).append(n) duplicates = [{'name': name, 'positions': places} for name, places in positions.items() if len(places) > 1] ragged = [{'logical_record': n, 'expected_fields': len(header), 'actual_fields': len(row)} for n, row in enumerate(rows, 2) if len(row) != len(header)] groups = {} for n, row in enumerate(rows, 2): groups.setdefault(tuple(row), []).append(n) return {'status': 'checked', 'observed': observed, 'duplicate_headers': duplicates, 'ragged_records': ragged, 'duplicate_data_records': [n for n in groups.values() if len(n) > 1], 'empty_header_positions': [n for n, value in enumerate(header, 1) if value == ''], 'record_numbering': 'Header is logical record1. Quoted embedded newlines do not create records; blank records are counted.', 'limits': limits, 'values_changed': False, 'native_dot_verified': False, 'scope': 'Python standard-library csv strict parsing. Structural findings only; no business-rule, import or safety approval.'} if __name__ == '__main__': try: print(json.dumps(check(json.load(sys.stdin)), ensure_ascii=False, indent=2)) except (ValueError, TypeError, json.JSONDecodeError) as exc: print(json.dumps({'status': 'invalid_input', 'error': str(exc)})) raise SystemExit(2)