Setting the file. One moment.
Format Detector · AWS AI ML · aws/agent-toolkit-for-aws · Skills Docs
Issue No. 14 · AWS AI ML
↖ Back to the coverMessaging And Streaming Skills
Migration And Modernization Skills
Networking And Content Delivery Skills
Security And Identity Skills
Web And Mobile Development
120 chapters · 648 min
Back to the top of the page 10
Setup DevOps Agent
24.7
Code Output Guide · references
RDS Oracle
383
def _validate_rlvr_messages
— line 383
This file
Number 24.4
Position 4 of 92
Type Python
Size 27 KB
Lines 771 references/dataset-evaluation/scripts/ format_detector.py
Python · 771 lines · 27 KB
import
logging
17 from dataclasses import dataclass
18 from enum import Enum
19 from typing import Any, Callable, Optional
20
21 import boto3
22
23 logger = logging.getLogger( __name__ )
24
25 # Type aliases for schema validators
26 MessageValidator = Callable[[ list , int ], "list[ValidationError]" ]
27 RecordValidator = Callable[[ dict , int ], "list[ValidationError]" ]
28
29 __all__ = [
30 "FormatType" ,
31 "ConfidenceLevel" ,
32 "ValidationError" ,
33 "FormatDetectionResult" ,
34 "detect_format" ,
35 ]
36
37
38 class FormatType ( Enum ):
39 """Supported JSONL format types."""
40
41 NOVA_SFT = "nova_sft"
42 NOVA_DPO = "nova_dpo"
43 NOVA_RLVR = "nova_rlvr"
44 GPT_OSS_SFT = "gpt_oss_sft"
45 GPT_OSS_DPO = "gpt_oss_dpo"
46 OPEN_WEIGHTS_SFT = "open_weights_sft"
47 OPEN_WEIGHTS_SFT_CONV = "open_weights_sft_conv"
48 OPEN_WEIGHTS_DPO = "open_weights_dpo"
49 VERL = "verl"
50 VERL_LEGACY = "verl_legacy"
51 SAGEMAKER_EVAL = "sagemaker_eval"
52 UNKNOWN = "unknown"
53
54
55 class ConfidenceLevel ( Enum ):
56 """Confidence level for format detection results."""
57
58 HIGH = "high"
59 LOW = "low"
60 NONE = "none"
61
62
63 @dataclass
64 class ValidationError :
65 """Represents a validation error found during format detection."""
66
67 line_number: int
68 error_type: str
69 message: str
70
71
72 @dataclass
73 class FormatDetectionResult :
74 """Result of format detection operation."""
75
76 format_type: FormatType
77 is_valid: bool
78 lines_sampled: int
79 errors: list[ValidationError]
80 confidence: ConfidenceLevel
81
82
83 def _sample_local_file (file_path: str , sample_size: int ) -> list[ str ]:
84 """Sample lines from local JSONL file.
85
86 Args:
87 file_path: Path to local file
88 sample_size: Maximum bytes to read
89
90 Returns:
91 List of lines from file
92
93 Raises:
94 FileNotFoundError: If file doesn't exist
95 IOError: If file can't be read
96 """
97 logger.debug( "Sampling local file: %s " , file_path)
98 with open (file_path, "rb" ) as f:
99 data = f.read(sample_size)
100
101 if not data:
102 return []
103
104 text = data.decode( "utf-8" , errors = "replace" )
105
106 last_newline_idx = text.rfind( " \n " )
107 if last_newline_idx == - 1 :
108 return []
109
110 complete_text = text[: last_newline_idx + 1 ]
111 lines = [line for line in complete_text.split( " \n " ) if line]
112
113 return lines
114
115
116 def _sample_s3_file (s3_uri: str , sample_size_bytes: int , s3_client = None ) -> list[ str ]:
117 """Sample the first N bytes of an S3 file and return complete lines.
118
119 Reads the first sample_size_bytes from an S3 file using a Range request,
120 then truncates to the last complete newline to avoid partial lines.
121
122 Args:
123 s3_uri: S3 URI in format "s3://bucket/key"
124 sample_size_bytes: Number of bytes to sample (default 1MB)
125 s3_client: Optional boto3 S3 client to reuse
126
127 Returns:
128 List of complete JSONL lines (strings without trailing newlines)
129
130 Raises:
131 ValueError: If S3 URI is invalid (missing "s3://", bucket, or key)
132 botocore.exceptions.ClientError: If S3 access fails
133 """
134 logger.debug( "Sampling S3 file: %s ( %d bytes)" , s3_uri, sample_size_bytes)
135 # Parse S3 URI
136 if not s3_uri.startswith( "s3://" ):
137 raise ValueError ( f "Invalid S3 URI: must start with 's3://' (got: { s3_uri } )" )
138
139 uri_without_prefix = s3_uri[ 5 :] # Remove "s3://"
140 parts = uri_without_prefix.split( "/" , 1 )
141
142 if len (parts) != 2 or not parts[ 0 ] or not parts[ 1 ]:
143 raise ValueError ( f "Invalid S3 URI: must contain bucket and key (got: { s3_uri } )" )
144
145 bucket, key = parts
146
147 # Read first sample_size_bytes using Range header
148 client = s3_client or boto3.client( "s3" )
149 range_header = f "bytes=0- { sample_size_bytes - 1 } "
150
151 response = client.get_object( Bucket = bucket, Key = key, Range = range_header)
152 data = response[ "Body" ].read()
153
154 # Handle empty file
155 if not data:
156 return []
157
158 # Decode bytes to string
159 text = data.decode( "utf-8" , errors = "replace" )
160
161 # Find last complete newline to avoid truncated lines
162 last_newline_idx = text.rfind( " \n " )
163 if last_newline_idx == - 1 :
164 # No newlines found - return empty list if file is all one line
165 # (we can't be sure it's complete)
166 return []
167
168 # Keep only complete lines (up to and including last newline)
169 complete_text = text[: last_newline_idx + 1 ]
170
171 # Split on newlines and filter empty strings
172 lines = [line for line in complete_text.split( " \n " ) if line]
173
174 return lines
175
176
177 def _classify_nova_format (record: dict ) -> FormatType:
178 """Classify Nova-specific format by checking last message structure.
179
180 Args:
181 record: Parsed JSON record with messages field
182
183 Returns:
184 FormatType.NOVA_DPO if last message has candidates field,
185 FormatType.NOVA_SFT if last message has standard content field,
186 FormatType.UNKNOWN otherwise
187 """
188 messages = record.get( "messages" , [])
189 if not messages:
190 return FormatType. UNKNOWN
191
192 last_message = messages[ - 1 ]
193 if "candidates" in last_message:
194 return FormatType. NOVA_DPO
195 elif "content" in last_message and last_message[ "content" ]:
196 return FormatType. NOVA_SFT
197 else :
198 return FormatType. UNKNOWN
199
200
201 def _classify_messages_format (record: dict ) -> FormatType:
202 """Distinguish Nova vs GPT-OSS/HF by inspecting content structure.
203
204 Nova has nested content arrays (list of dicts with 'text' field),
205 GPT-OSS/HF has flat content strings.
206
207 Args:
208 record: Parsed JSON record with messages field
209
210 Returns:
211 FormatType value for the detected format
212 """
213 messages = record.get( "messages" )
214
215 # Critical type checking: messages must be a list
216 if not isinstance (messages, list ):
217 return FormatType. UNKNOWN
218
219 if not messages:
220 return FormatType. UNKNOWN
221
222 first_message = messages[ 0 ]
223
224 # Check if content field exists
225 if "content" not in first_message:
226 return FormatType. UNKNOWN
227
228 content = first_message[ "content" ]
229
230 # Nova: nested content arrays (list of dicts with 'text' field)
231 if isinstance (content, list ):
232 return _classify_nova_format(record)
233 # GPT-OSS/HF: flat content strings
234 elif isinstance (content, str ):
235 return FormatType. GPT_OSS_SFT
236 else :
237 return FormatType. UNKNOWN
238
239
240 def _classify_schema (samples: list[ dict ]) -> FormatType:
241 """Top-level classifier that checks for all 11 supported formats.
242
243 Args:
244 samples: List of parsed JSON records
245
246 Returns:
247 FormatType value for the detected format
248 """
249 if not samples:
250 return FormatType. UNKNOWN
251
252 first = samples[ 0 ]
253 fields = set (first.keys())
254
255 # SageMaker Evaluation: query + response
256 if "query" in fields and "response" in fields:
257 return FormatType. SAGEMAKER_EVAL
258
259 # Verl/RLVR: prompt + (reward_model or extra_info), no completion
260 if "prompt" in fields and ( "reward_model" in fields or "extra_info" in fields):
261 if "completion" not in fields:
262 if isinstance (first[ "prompt" ], list ):
263 return FormatType. VERL
264 return FormatType. VERL_LEGACY
265
266 # Messages-based formats: Nova RLVR, Nova, GPT-OSS
267 if "messages" in fields:
268 if "reference_answer" in fields:
269 return FormatType. NOVA_RLVR
270 return _classify_messages_format(first)
271
272 # DPO: prompt/chosen/rejected
273 if { "prompt" , "chosen" , "rejected" }.issubset(fields):
274 if isinstance (first[ "prompt" ], list ):
275 return FormatType. GPT_OSS_DPO
276 return FormatType. OPEN_WEIGHTS_DPO
277
278 # SFT: prompt/completion
279 if { "prompt" , "completion" }.issubset(fields):
280 if isinstance (first[ "prompt" ], list ):
281 return FormatType. OPEN_WEIGHTS_SFT_CONV
282 return FormatType. OPEN_WEIGHTS_SFT
283
284 return FormatType. UNKNOWN
285
286
287 def _validate_nova_messages (messages: list , line_num: int , is_dpo: bool ) -> list[ValidationError]:
288 """Validate Nova SFT/DPO message structure."""
289 errors = []
290 for msg_idx, msg in enumerate (messages):
291 if "role" not in msg:
292 errors.append(
293 ValidationError(
294 line_number = line_num,
295 error_type = "missing_field" ,
296 message = f "Message { msg_idx } missing required field 'role'" ,
297 )
298 )
299 elif msg[ "role" ] not in [ "user" , "assistant" , "system" ]:
300 errors.append(
301 ValidationError(
302 line_number = line_num,
303 error_type = "invalid_structure" ,
304 message = f "Invalid role ' { msg[ 'role' ] } ' in message { msg_idx } " ,
305 )
306 )
307 if "content" not in msg and "candidates" not in msg:
308 errors.append(
309 ValidationError(
310 line_number = line_num,
311 error_type = "missing_field" ,
312 message = f "Message { msg_idx } missing 'content' or 'candidates'" ,
313 )
314 )
315 if "content" in msg and not isinstance (msg[ "content" ], list ):
316 errors.append(
317 ValidationError(
318 line_number = line_num,
319 error_type = "invalid_structure" ,
320 message = f "Nova format content must be list, got { type (msg[ 'content' ]). __name__ } " ,
321 )
322 )
323 if is_dpo and "candidates" in msg:
324 for cand_idx, candidate in enumerate (msg[ "candidates" ]):
325 if "preferenceLabel" not in candidate:
326 errors.append(
327 ValidationError(
328 line_number = line_num,
329 error_type = "missing_field" ,
330 message = f "DPO message { msg_idx } candidate { cand_idx } missing 'preferenceLabel'" ,
331 )
332 )
333 elif candidate[ "preferenceLabel" ] not in [ "preferred" , "non-preferred" ]:
334 errors.append(
335 ValidationError(
336 line_number = line_num,
337 error_type = "invalid_structure" ,
338 message = f "Invalid preferenceLabel ' { candidate[ 'preferenceLabel' ] } ' in message { msg_idx } candidate { cand_idx } " ,
339 )
340 )
341 return errors
342
343
344 def _validate_gpt_messages (messages: list , line_num: int ) -> list[ValidationError]:
345 """Validate GPT-OSS SFT message structure."""
346 errors = []
347 for msg_idx, msg in enumerate (messages):
348 if "role" not in msg:
349 errors.append(
350 ValidationError(
351 line_number = line_num,
352 error_type = "missing_field" ,
353 message = f "Message { msg_idx } missing required field 'role'" ,
354 )
355 )
356 elif msg[ "role" ] not in [ "user" , "assistant" , "system" ]:
357 errors.append(
358 ValidationError(
359 line_number = line_num,
360 error_type = "invalid_structure" ,
361 message = f "Invalid role ' { msg[ 'role' ] } ' in message { msg_idx } " ,
362 )
363 )
364 if "content" not in msg:
365 errors.append(
366 ValidationError(
367 line_number = line_num,
368 error_type = "missing_field" ,
369 message = f "Message { msg_idx } missing required field 'content'" ,
370 )
371 )
372 elif not isinstance (msg[ "content" ], str ):
373 errors.append(
374 ValidationError(
375 line_number = line_num,
376 error_type = "invalid_structure" ,
377 message = f "GPT-OSS format content must be string, got { type (msg[ 'content' ]). __name__ } " ,
378 )
379 )
380 return errors
381
382
383 def _validate_rlvr_messages (messages: list , line_num: int ) -> list[ValidationError]:
384 """Validate Nova RLVR message structure."""
385 errors = []
386 for msg_idx, msg in enumerate (messages):
387 if "role" not in msg:
388 errors.append(
389 ValidationError(
390 line_number = line_num,
391 error_type = "missing_field" ,
392 message = f "Message { msg_idx } missing required field 'role'" ,
393 )
394 )
395 elif msg[ "role" ] not in [ "user" , "assistant" , "system" ]:
396 errors.append(
397 ValidationError(
398 line_number = line_num,
399 error_type = "invalid_structure" ,
400 message = f "Invalid role ' { msg[ 'role' ] } ' in message { msg_idx } " ,
401 )
402 )
403 if "content" not in msg:
404 errors.append(
405 ValidationError(
406 line_number = line_num,
407 error_type = "missing_field" ,
408 message = f "Message { msg_idx } missing required field 'content'" ,
409 )
410 )
411 elif not isinstance (msg[ "content" ], str ):
412 errors.append(
413 ValidationError(
414 line_number = line_num,
415 error_type = "invalid_structure" ,
416 message = f "Nova RLVR content must be string, got { type (msg[ 'content' ]). __name__ } " ,
417 )
418 )
419 return errors
420
421
422 def _validate_verl_prompt (record: dict , line_num: int ) -> list[ValidationError]:
423 """Validate Verl prompt structure (list of role/content dicts)."""
424 errors = []
425 if "prompt" not in record:
426 errors.append(
427 ValidationError(
428 line_number = line_num,
429 error_type = "missing_field" ,
430 message = "Missing required field 'prompt'" ,
431 )
432 )
433 elif not isinstance (record[ "prompt" ], list ):
434 errors.append(
435 ValidationError(
436 line_number = line_num,
437 error_type = "invalid_structure" ,
438 message = f "Verl field 'prompt' must be list, got { type (record[ 'prompt' ]). __name__ } " ,
439 )
440 )
441 else :
442 for msg_idx, msg in enumerate (record[ "prompt" ]):
443 if not isinstance (msg, dict ) or "role" not in msg or "content" not in msg:
444 errors.append(
445 ValidationError(
446 line_number = line_num,
447 error_type = "invalid_structure" ,
448 message = f "Prompt message { msg_idx } must have 'role' and 'content'" ,
449 )
450 )
451 if "reward_model" not in record and "extra_info" not in record:
452 errors.append(
453 ValidationError(
454 line_number = line_num,
455 error_type = "missing_field" ,
456 message = "Missing required field 'reward_model' or 'extra_info'" ,
457 )
458 )
459 return errors
460
461
462 def _validate_verl_legacy_prompt (record: dict , line_num: int ) -> list[ValidationError]:
463 """Validate Verl Legacy prompt structure (string) and extra_info."""
464 errors = []
465 if "prompt" not in record:
466 errors.append(
467 ValidationError(
468 line_number = line_num,
469 error_type = "missing_field" ,
470 message = "Missing required field 'prompt'" ,
471 )
472 )
473 elif not isinstance (record[ "prompt" ], str ):
474 errors.append(
475 ValidationError(
476 line_number = line_num,
477 error_type = "invalid_structure" ,
478 message = f "Verl Legacy field 'prompt' must be string, got { type (record[ 'prompt' ]). __name__ } " ,
479 )
480 )
481 if "reward_model" not in record and "extra_info" not in record:
482 errors.append(
483 ValidationError(
484 line_number = line_num,
485 error_type = "missing_field" ,
486 message = "Missing required field 'reward_model' or 'extra_info'" ,
487 )
488 )
489 return errors
490
491
492 # Schema-driven format validation specs.
493 # Each entry defines required_fields (field->type mapping) and an optional
494 # message_validator or record_validator for complex per-record checks.
495 # - message_validator: called with (messages_list, line_num) -> list[ValidationError]
496 # Used for formats whose top-level required field is "messages" (list).
497 # - record_validator: called with (record, line_num) -> list[ValidationError]
498 # Used for formats needing whole-record access (verl, verl_legacy).
499 FORMAT_SCHEMAS : dict[FormatType, dict[ str , Any]] = {
500 FormatType. NOVA_SFT : {
501 "required_fields" : { "messages" : list },
502 "message_validator" : lambda msgs, ln: _validate_nova_messages(
503 msgs, ln, is_dpo = False
504 ), # nosemgrep: python.lang.maintainability.return.return-not-in-function -- lambda inside dict literal, not a bare return
505 },
506 FormatType. NOVA_DPO : {
507 "required_fields" : { "messages" : list },
508 "message_validator" : lambda msgs, ln: _validate_nova_messages(
509 msgs, ln, is_dpo = True
510 ), # nosemgrep: python.lang.maintainability.return.return-not-in-function -- lambda inside dict literal, not a bare return
511 },
512 FormatType. NOVA_RLVR : {
513 "required_fields" : { "messages" : list , "reference_answer" : dict },
514 "message_validator" : _validate_rlvr_messages,
515 },
516 FormatType. GPT_OSS_SFT : {
517 "required_fields" : { "messages" : list },
518 "message_validator" : _validate_gpt_messages,
519 },
520 FormatType. GPT_OSS_DPO : {
521 "required_fields" : { "prompt" : list , "chosen" : list , "rejected" : list },
522 "field_error_prefix" : "GPT-OSS DPO" ,
523 },
524 FormatType. OPEN_WEIGHTS_SFT : {
525 "required_fields" : { "prompt" : str , "completion" : str },
526 "field_error_prefix" : "Open Weights SFT" ,
527 },
528 FormatType. OPEN_WEIGHTS_SFT_CONV : {
529 "required_fields" : { "prompt" : list , "completion" : list },
530 "field_error_prefix" : "Open Weights SFT Conv" ,
531 },
532 FormatType. OPEN_WEIGHTS_DPO : {
533 "required_fields" : { "prompt" : str , "chosen" : str , "rejected" : str },
534 "field_error_prefix" : "Open Weights DPO" ,
535 },
536 FormatType. SAGEMAKER_EVAL : {
537 "required_fields" : { "query" : str , "response" : str },
538 "field_error_prefix" : "SageMaker Eval" ,
539 },
540 FormatType. VERL : {
541 "required_fields" : {},
542 "record_validator" : _validate_verl_prompt,
543 },
544 FormatType. VERL_LEGACY : {
545 "required_fields" : {},
546 "record_validator" : _validate_verl_legacy_prompt,
547 },
548 }
549
550
551 def _validate_samples (
552 samples: list[ dict ], expected_format: FormatType, line_numbers: list[ int ]
553 ) -> tuple[ bool , list[ValidationError]]:
554 """Validate that all samples conform to the expected format schema.
555
556 Args:
557 samples: List of parsed JSON records
558 expected_format: Expected FormatType enum value
559 line_numbers: 1-based line numbers corresponding to each sample
560
561 Returns:
562 Tuple of (is_valid, errors) where errors is a list of ValidationError objects
563 """
564 errors = []
565 schema = FORMAT_SCHEMAS .get(expected_format)
566
567 for record, line_num in zip (samples, line_numbers):
568 # Check schema consistency
569 detected_format = _classify_schema([record])
570 if detected_format != expected_format:
571 errors.append(
572 ValidationError(
573 line_number = line_num,
574 error_type = "schema_mismatch" ,
575 message = f "Expected { expected_format.value } but found { detected_format.value } " ,
576 )
577 )
578 continue
579
580 if schema is None :
581 continue
582
583 # Record-level validator (verl, verl_legacy) handles everything
584 if "record_validator" in schema:
585 validator: RecordValidator = schema[ "record_validator" ]
586 errors.extend(validator(record, line_num))
587 continue
588
589 # Check required fields exist with correct types
590 required = schema[ "required_fields" ]
591 prefix: str = schema.get( "field_error_prefix" , "" ) or ""
592 skip_messages = False
593 for field, expected_type in required.items():
594 if field not in record:
595 errors.append(
596 ValidationError(
597 line_number = line_num,
598 error_type = "missing_field" ,
599 message = f "Missing required field ' { field } '" ,
600 )
601 )
602 if field == "messages" :
603 skip_messages = True
604 elif not isinstance (record[field], expected_type):
605 actual = type (record[field]). __name__
606 if field == "messages" :
607 errors.append(
608 ValidationError(
609 line_number = line_num,
610 error_type = "invalid_structure" ,
611 message = f "Field 'messages' must be a list" ,
612 )
613 )
614 skip_messages = True
615 elif prefix:
616 errors.append(
617 ValidationError(
618 line_number = line_num,
619 error_type = "invalid_structure" ,
620 message = f " { prefix } field ' { field } ' must be { expected_type. __name__ } , got { actual } " ,
621 )
622 )
623 else :
624 errors.append(
625 ValidationError(
626 line_number = line_num,
627 error_type = "invalid_structure" ,
628 message = f "Field ' { field } ' must be { expected_type. __name__ } , got { actual } " ,
629 )
630 )
631
632 if skip_messages:
633 continue
634
635 # Message-level validator
636 if "message_validator" in schema:
637 msg_validator: MessageValidator = schema[ "message_validator" ]
638 errors.extend(msg_validator(record[ "messages" ], line_num))
639
640 logger.debug( "Validation found %d error(s)" , len (errors))
641 return ( len (errors) == 0 , errors)
642
643
644 def detect_format (
645 file_path: str , sample_size_bytes: int = 1_048_576 , s3_client = None
646 ) -> FormatDetectionResult:
647 """Detect the format of a JSONL file in S3 or on local disk.
648
649 Samples the first sample_size_bytes of the file and analyzes the structure
650 to determine if it matches one of the 11 supported formats.
651
652 Args:
653 file_path: S3 URI (s3://bucket/key) or local file path
654 sample_size_bytes: Number of bytes to sample (default 1MB = 1,048,576 bytes)
655 s3_client: Optional boto3 S3 client to reuse (ignored for local files)
656
657 Returns:
658 FormatDetectionResult with format type, validation status, and any errors
659 """
660 if file_path.startswith( "s3://" ):
661 lines = _sample_s3_file(file_path, sample_size_bytes, s3_client = s3_client)
662 else :
663 lines = _sample_local_file(file_path, sample_size_bytes)
664
665 # Parse JSON lines and collect parse errors
666 parsed_records = []
667 line_numbers = []
668 errors = []
669
670 for line_num, line in enumerate (lines, start = 1 ):
671 try :
672 parsed_records.append(json.loads(line))
673 line_numbers.append(line_num)
674 except json.JSONDecodeError as e:
675 errors.append(
676 ValidationError(
677 line_number = line_num,
678 error_type = "parse_error" ,
679 message = f "Invalid JSON: { str (e) } " ,
680 )
681 )
682
683 # If no successfully parsed records, return UNKNOWN with parse errors
684 if not parsed_records:
685 confidence = ConfidenceLevel. NONE if errors else ConfidenceLevel. HIGH
686 return FormatDetectionResult(
687 format_type = FormatType. UNKNOWN ,
688 is_valid = len (errors) == 0 ,
689 lines_sampled = len (lines),
690 errors = errors,
691 confidence = confidence,
692 )
693
694 # Classify schema using first successfully parsed record
695 format_type = _classify_schema(parsed_records)
696
697 # Validate all parsed records against detected format
698 is_valid, validation_errors = _validate_samples(parsed_records, format_type, line_numbers)
699 errors.extend(validation_errors)
700
701 # Calculate confidence level
702 if len (errors) == 0 :
703 confidence = ConfidenceLevel. HIGH
704 elif any (err.error_type == "parse_error" for err in errors):
705 confidence = ConfidenceLevel. NONE
706 else :
707 confidence = ConfidenceLevel. LOW
708
709 logger.debug(
710 "Detected format: %s (valid= %s , confidence= %s )" ,
711 format_type.value,
712 is_valid,
713 confidence.value,
714 )
715
716 return FormatDetectionResult(
717 format_type = format_type,
718 is_valid = len (errors) == 0 ,
719 lines_sampled = len (lines),
720 errors = errors,
721 confidence = confidence,
722 )
723
724
725 if __name__ == "__main__" :
726 import argparse
727 import sys
728
729 parser = argparse.ArgumentParser( description = "Detect and validate JSONL file formats" )
730 parser.add_argument( "file_path" , help = "S3 URI (s3://bucket/key) or local file path" )
731 parser.add_argument(
732 "--sample-size" , type = int , default = 1_048_576 , help = "Bytes to sample (default: 1MB)"
733 )
734 parser.add_argument(
735 "--json" , action = "store_true" , help = "Output as JSON instead of human-readable"
736 )
737 args = parser.parse_args()
738
739 try :
740 result = detect_format(args.file_path, args.sample_size)
741
742 if args.json:
743 output = {
744 "format_type" : result.format_type.value,
745 "is_valid" : result.is_valid, # nosemgrep: python.lang.maintainability.is-function-without-parentheses -- dataclass field, not a method
746 "confidence" : result.confidence.value,
747 "lines_sampled" : result.lines_sampled,
748 "errors" : [
749 { "line_number" : e.line_number, "error_type" : e.error_type, "message" : e.message}
750 for e in result.errors
751 ],
752 }
753 print (json.dumps(output, indent = 2 ))
754 else :
755 print ( f "Format: { result.format_type.value } " )
756 print (
757 f "Valid: { '✓' if result.is_valid else '✗' } "
758 ) # nosemgrep: python.lang.maintainability.is-function-without-parentheses -- dataclass field, not a method
759 print ( f "Confidence: { result.confidence.name } " )
760 print ( f "Lines sampled: { result.lines_sampled } " )
761 if result.errors:
762 print ( "Errors:" )
763 for err in result.errors:
764 print ( f " Line { err.line_number } : { err.message } " )
765
766 sys.exit(
767 0 if result.is_valid else 1
768 ) # nosemgrep: python.lang.maintainability.is-function-without-parentheses -- dataclass field, not a method
769 except ( FileNotFoundError , IOError , ValueError ) as e:
770 print ( f "Error: { e } " , file = sys.stderr)
771 sys.exit( 1 )