csv_utils.py

"""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