summaryrefslogtreecommitdiff
path: root/highlight.py
blob: ffcdd2c7f67a2eca5d8638ed6ed5de9efe6bbc95 (plain)
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
import sys
import json
import re
import pathlib
from datetime import timedelta

# ANSI escape codes for terminal colors
class Colors:
    GREEN = '\033[92m'  # Highlight for valid ad text
    RED = '\033[91m'    # Highlight for invalid target ranges
    BLUE = '\033[94m'   # Header color
    RESET = '\033[0m'   # Reset to default color

def timestamp_to_seconds(ts_str: str) -> float:
    """Converts HH:MM:SS.mmm string to total seconds. Returns -1.0 on failure."""
    try:
        parts = ts_str.split(':')
        h, m = int(parts[0]), int(parts[1])
        s_ms = parts[2].split('.')
        s, ms = int(s_ms[0]), int(s_ms[1])
        return h * 3600 + m * 60 + s + ms / 1000.0
    except (ValueError, IndexError):
        return -1.0

def visualize_entry(entry: dict):
    """Parses and prints a single entry from the jsonl file with highlighting."""
    entry_id = entry.get("id", "N/A")
    full_text = entry.get("text", "")
    target_str = entry.get("target", "")

    print(f"{Colors.BLUE}{'='*80}\nID: {entry_id}{Colors.RESET}")

    # --- 1. Parse the transcript from the 'text' field ---
    transcript_lines = []
    # Regex to find the start of the actual transcript text
    transcript_match = re.search(r"Transcript:\n(.*)", full_text, re.DOTALL)
    if not transcript_match:
        print(f"{Colors.RED}Could not find transcript text in entry.{Colors.RESET}")
        return

    raw_transcript = transcript_match.group(1).strip()
    line_pattern = re.compile(r"\[(\d{2}:\d{2}:\d{2}\.\d{3}) --> (\d{2}:\d{2}:\d{2}\.\d{3})\](.*)")
    
    min_transcript_time, max_transcript_time = float('inf'), float('-inf')

    for line in raw_transcript.split('\n'):
        match = line_pattern.match(line)
        if match:
            start_str, end_str, text = match.groups()
            start_sec = timestamp_to_seconds(start_str)
            end_sec = timestamp_to_seconds(end_str)
            transcript_lines.append({"start": start_sec, "end": end_sec, "line": line})
            min_transcript_time = min(min_transcript_time, start_sec)
            max_transcript_time = max(max_transcript_time, end_sec)

    # --- 2. Parse and validate the target ranges ---
    valid_target_ranges = []
    has_invalid_target = False
    
    if target_str:
        for target_line in target_str.strip().split('\n'):
            if '-' not in target_line:
                has_invalid_target = True
                continue
            
            start_target_str, end_target_str = target_line.split('-', 1)
            start_target_sec = timestamp_to_seconds(start_target_str.strip())
            end_target_sec = timestamp_to_seconds(end_target_str.strip())

            # Validation checks
            is_valid_format = start_target_sec != -1.0 and end_target_sec != -1.0
            is_valid_range = is_valid_format and start_target_sec < end_target_sec
            # Check if the target overlaps with the transcript's time frame at all
            is_overlapping = is_valid_range and start_target_sec < max_transcript_time and end_target_sec > min_transcript_time

            if is_overlapping:
                valid_target_ranges.append({"start": start_target_sec, "end": end_target_sec})
            else:
                has_invalid_target = True
    
    # --- 3. Print results with highlighting ---
    print(f"TARGET: ", end="")
    if has_invalid_target:
        print(f"{Colors.RED}{target_str or 'EMPTY'}{Colors.RESET}")
    else:
        print(f"{Colors.GREEN}{target_str or 'EMPTY'}{Colors.RESET}")
    print("-" * 30)

    for line_info in transcript_lines:
        is_ad_line = False
        for target_range in valid_target_ranges:
            # Check for overlap between the line's time and the target's time
            if line_info['start'] < target_range['end'] and line_info['end'] > target_range['start']:
                is_ad_line = True
                break
        
        if is_ad_line:
            print(f"{Colors.GREEN}{line_info['line']}{Colors.RESET}")
        else:
            print(line_info['line'])

def main(filepath: str):
    try:
        p = pathlib.Path(filepath)
        with p.open('r', encoding='utf-8') as f:
            for line in f:
                try:
                    entry = json.loads(line)
                    visualize_entry(entry)
                except json.JSONDecodeError:
                    print(f"{Colors.RED}Error: Could not decode JSON line: {line.strip()}{Colors.RESET}")
    except FileNotFoundError:
        print(f"{Colors.RED}Error: File not found at {filepath}{Colors.RESET}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python highlight.py <path_to_labels.jsonl>")
        sys.exit(1)
    
    main(sys.argv[1])