1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
import sys
import os
import json
import subprocess
import re
from datetime import timedelta
# --- Configuration ---
HIGHLIGHT_SCRIPT_NAME = "highlight.py"
GOOD_DATASET_FILE = "dataset.jsonl"
BAD_FILES_LOG = "errors.txt"
PROGRESS_FILE = ".curation_progress.log"
# --- Unbuffered Input (Cross-Platform) ---
try:
# Unix-like systems (Linux, macOS)
import tty, termios
def getch():
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
try:
tty.setraw(sys.stdin.fileno())
ch = sys.stdin.read(1)
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
return ch
except ImportError:
# Windows
import msvcrt
def getch():
return msvcrt.getch().decode()
# --- Main Curation Logic ---
def review_files(file_paths: list[str]):
"""
Iterates through files, displays them for review, and sorts them based on user input.
"""
if not os.path.exists(HIGHLIGHT_SCRIPT_NAME):
print(f"Error: The visualization script '{HIGHLIGHT_SCRIPT_NAME}' was not found in this directory.")
sys.exit(1)
# Load the set of already processed files
processed_files = set()
if os.path.exists(PROGRESS_FILE):
with open(PROGRESS_FILE, 'r', encoding='utf-8') as f:
processed_files = {line.strip() for line in f}
total_files = len(file_paths)
for i, filepath in enumerate(file_paths):
if filepath in processed_files:
print(f"Skipping already processed file: {filepath}")
continue
print("-" * 80)
print(f"Reviewing file ({i + 1}/{total_files}): {filepath}")
# 1. Run highlight.py and pipe its output to the user's pager
try:
pager_cmd = os.environ.get("PAGER", "less").split()
highlight_process = subprocess.Popen(
["python", HIGHLIGHT_SCRIPT_NAME, filepath],
stdout=subprocess.PIPE,
text=True,
encoding='utf-8'
)
pager_process = subprocess.Popen(pager_cmd, stdin=highlight_process.stdout)
highlight_process.stdout.close()
return_code = pager_process.wait()
if return_code != 0:
print(f"Warning: Pager exited with code {return_code}.")
except FileNotFoundError:
print(f"Error: Pager command not found. Is '{os.environ.get('PAGER', 'less')}' installed?")
continue
except Exception as e:
print(f"An error occurred while running the subprocess: {e}")
continue
# 2. Prompt for user decision
while True:
print("\nWas this file's labeling correct? (y/n): ", end="", flush=True)
choice = getch().lower()
print(choice)
if choice == 'y':
print(f" -> Appending {filepath} to {GOOD_DATASET_FILE}")
try:
with open(filepath, 'r', encoding='utf-8') as infile, \
open(GOOD_DATASET_FILE, 'a', encoding='utf-8') as outfile:
for line in infile:
outfile.write(line)
except Exception as e:
print(f"Error appending to dataset file: {e}")
break
elif choice == 'n':
print(f" -> Logging {filepath} to {BAD_FILES_LOG}")
try:
with open(BAD_FILES_LOG, 'a', encoding='utf-8') as errfile:
errfile.write(filepath + '\n')
except Exception as e:
print(f"Error writing to log file: {e}")
break
elif choice in ['\x03', 'q']: # Handle Ctrl+C or 'q' to quit
print("\nExiting review process.")
sys.exit(0)
else:
print("Invalid input. Please press 'y' for yes or 'n' for no.")
# Log the file as processed after a decision has been made
with open(PROGRESS_FILE, 'a', encoding='utf-8') as f:
f.write(filepath + '\n')
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python curate_dataset.py <file1.jsonl> <file2.jsonl> ...")
print("Or: python curate_dataset.py /path/to/*/*.jsonl")
sys.exit(1)
review_files(sys.argv[1:])
print("-" * 80)
print("Curation complete.")
|
