Spaces:
Sleeping
Sleeping
| """CSV parsing utilities for the Annotation Management System.""" | |
| import csv | |
| import hashlib | |
| import os | |
| def sample_id(conversation_cut): | |
| """Return an ID that stays the same when CSV rows are reordered.""" | |
| digest = hashlib.sha256(conversation_cut.strip().encode('utf-8')).hexdigest() | |
| return f'ES-{digest[:12].upper()}' | |
| 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(): | |
| sample = dict(row) | |
| sample['sample_index'] = len(rows) | |
| sample['sample_id'] = sample_id(row['conversation_cut']) | |
| rows.append(sample) | |
| 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 | |