"""CSV parsing utilities for the Annotation Management System.""" import csv import json import os def parse_csv_file(filepath): """Parse a CSV file and return list of row dicts with only conversation samples.""" rows = [] with open(filepath, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: # Only include rows that have conversation_cut data if row.get('conversation_cut', '').strip(): rows.append(dict(row)) return rows def count_samples(filepath): """Count the number of valid conversation samples in a CSV file.""" count = 0 with open(filepath, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if row.get('conversation_cut', '').strip(): count += 1 return count def get_sample(filepath, index): """Get a specific sample by index from a CSV file.""" rows = parse_csv_file(filepath) if 0 <= index < len(rows): return rows[index] return None def scan_csv_folder(folder_path): """Scan a folder and return list of CSV file info dicts.""" files = [] if not os.path.isdir(folder_path): return files for fname in sorted(os.listdir(folder_path)): if fname.lower().endswith('.csv'): fpath = os.path.join(folder_path, fname) files.append({ 'filename': fname, 'filepath': fpath, 'size_bytes': os.path.getsize(fpath), }) return files