| 59 | |
| 60 | |
| 61 | def read_jsonl(file_path): |
| 62 | with open(file_path, 'r') as f: |
| 63 | content = f.read() |
| 64 | |
| 65 | # 用来存储完整的 JSON 对象 |
| 66 | json_objects = [] |
| 67 | |
| 68 | # 变量用于跟踪大括号的嵌套层级 |
| 69 | brace_count = 0 |
| 70 | start = 0 |
| 71 | |
| 72 | # 遍历文件内容 |
| 73 | for idx, char in enumerate(content): |
| 74 | if char == '{': |
| 75 | # 增加嵌套层级 |
| 76 | if brace_count == 0: |
| 77 | start = idx # 记录开始位置 |
| 78 | brace_count += 1 |
| 79 | elif char == '}': |
| 80 | # 减少嵌套层级 |
| 81 | brace_count -= 1 |
| 82 | if brace_count == 0: |
| 83 | # 当嵌套层级归零时，意味着我们找到了一个完整的 JSON 对象 |
| 84 | json_str = content[start:idx+1] |
| 85 | try: |
| 86 | # 将字符串解析为字典 |
| 87 | json_objects.append(json.loads(json_str)) |
| 88 | except json.JSONDecodeError as e: |
| 89 | print(f"Error parsing JSON: {e}") |
| 90 | continue |
| 91 | action_dicts = [d for d in json_objects if 'action' in d] |
| 92 | aim_landmark = [d for d in json_objects if 'aim_landmark' in d] |
| 93 | |
| 94 | if len(aim_landmark) == 0: |
| 95 | aim_landmark = [{'aim_landmark':{"error":None}}] |
| 96 | return action_dicts,aim_landmark[-1] |
| 97 | |
| 98 | |
| 99 | |