Adicionar src-py/renatoxsr/utils/jsonl.py

src-py/renatoxsr/utils/jsonl.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
+# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+import json
+from pathlib import Path
+from io import BytesIO
+
+def parse_jsonl(file_path):
+    """Parse a JSONL file line by line"""
+    with open(file_path, 'r') as file:
+        for line in file:
+            line = line.strip()
+            if line:
+                try:
+                    yield json.loads(line)
+                except json.JSONDecodeError as e:
+                    print(f"Error parsing line: {e}")
+                    continue
+
+
+def stream_jsonl(file_path: str) -> Iterator[Dict[str, Any]]:
+    """Stream JSONL file for large datasets"""
+    with open(file_path, 'r', encoding='utf-8') as file:
+        for line_num, line in enumerate(file, 1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                yield json.loads(line)
+            except json.JSONDecodeError as e:
+                print(f"Error on line {line_num}: {e}")
+                continue
+
+
+def jsonl_df(file_path):
+    """Convert JSONL to pandas DataFrame"""
+    import pandas as pd
+    data = []
+    with open(file_path, 'r') as file:
+        for line in file:
+            line = line.strip()
+            if line:
+                try:
+                    data.append(json.loads(line))
+                except json.JSONDecodeError:
+                    continue
+    return pd.DataFrame(data)
+
+
+def get_demo_bytes(demo_file_path: str | Path = None):
+    if demo_file_path:
+        return Path(demo_file_path)
+    return BytesIO(
+        # example JSONL file:
+        """{"row1":"col1","row1":"col2"}\n{"row2":"col1","row2":"col2"}""".encode("utf-8"))
+
+
+def demo_parse_jsonl(demo_file_path: str | Path = None):
+    """demo parse_jsonl() function"""
+    demo_io = get_demo_bytes(demo_file_path)
+    for data in parse_jsonl(demo_io):
+        print(data)
+    streaming_parser.py
+    import json
+    from typing import Iterator, Dict, Any
+
+
+def demo_stream_jsonl(demo_file_path: str | Path = None):
+    """Demo stream_jsonl() function"""
+    demo_io = get_demo_bytes(demo_file_path)
+    # Process large files without memory issues
+    for record in stream_jsonl(demo_io):
+        process_record(record)
+    pandas_parser.py
+    
+def demo_jsonl_df(demo_file_path: str | Path = None):
+    """Demo jsonl_df() function"""
+    demo_io = get_demo_bytes(demo_file_path)
+    df = jsonl_df(demo_io)
+    print(df.head())