| 1 | import json |
| 2 | |
| 3 | def try_parse(json_string: str): |
| 4 | try: |
| 5 | return json.loads(json_string) |
| 6 | except json.JSONDecodeError: |
| 7 | return DirtyJson.parse_string(json_string) |
| 8 | |
| 9 | |
| 10 | def parse(json_string: str): |
| 11 | return DirtyJson.parse_string(json_string) |
| 12 | |
| 13 | |
| 14 | def stringify(obj, **kwargs): |
| 15 | return json.dumps(obj, ensure_ascii=False, **kwargs) |
| 16 | |
| 17 | |
| 18 | class DirtyJson: |
| 19 | def __init__(self): |
| 20 | self._reset() |
| 21 | |
| 22 | def _reset(self): |
| 23 | self.json_string = "" |
| 24 | self.index = 0 |
| 25 | self.current_char = None |
| 26 | self.result = None |
| 27 | self.stack = [] |
| 28 | self.completed = False |
| 29 | self._parsing_started = False |
| 30 | |
| 31 | def _pop_stack(self, root_closed: bool = False): |
| 32 | """Pop from the parsing stack and mark completed only on an explicit root close.""" |
| 33 | self.stack.pop() |
| 34 | if root_closed and self._parsing_started and not self.stack: |
| 35 | self.completed = True |
| 36 | |
| 37 | @staticmethod |
| 38 | def parse_string(json_string): |
| 39 | parser = DirtyJson() |
| 40 | return parser.parse(json_string) |
| 41 | |
| 42 | def parse(self, json_string): |
| 43 | self._reset() |
| 44 | self.json_string = json_string |
| 45 | |
| 46 | # Add bounds checking to prevent IndexError |
| 47 | if not json_string: |
| 48 | # Return None for empty strings |
| 49 | return None |
| 50 | |
| 51 | self.index = self.get_start_pos(self.json_string) |
| 52 | |
| 53 | # Ensure index is within bounds |
| 54 | if self.index >= len(self.json_string): |
| 55 | # If start position is beyond string length, return None |
| 56 | return None |
| 57 | |
| 58 | self.current_char = self.json_string[self.index] |
| 59 | self._parse() |
| 60 | return self.result |
| 61 | |
| 62 | def feed(self, chunk): |
| 63 | self.json_string += chunk |
| 64 | if not self.current_char and self.json_string: |
| 65 | self.current_char = self.json_string[0] |
| 66 | self._parse() |
| 67 | return self.result |
| 68 | |
| 69 | def _advance(self, count=1): |
| 70 | self.index += count |
| 71 | if self.index < len(self.json_string): |
| 72 | self.current_char = self.json_string[self.index] |
| 73 | else: |
| 74 | self.current_char = None |
| 75 | |
| 76 | def _skip_whitespace(self): |
| 77 | while self.current_char is not None: |
| 78 | if self.current_char.isspace(): |
| 79 | self._advance() |
| 80 | elif ( |
| 81 | self.current_char == "/" and self._peek(1) == "/" |
| 82 | ): # Single-line comment |
| 83 | self._skip_single_line_comment() |
| 84 | elif ( |
| 85 | self.current_char == "/" and self._peek(1) == "*" |
| 86 | ): # Multi-line comment |
| 87 | self._skip_multi_line_comment() |
| 88 | else: |
| 89 | break |
| 90 | |
| 91 | def _skip_single_line_comment(self): |
| 92 | while self.current_char is not None and self.current_char != "\n": |
| 93 | self._advance() |
| 94 | if self.current_char == "\n": |
| 95 | self._advance() |
| 96 | |
| 97 | def _skip_multi_line_comment(self): |
| 98 | self._advance(2) # Skip /* |
| 99 | while self.current_char is not None: |
| 100 | if self.current_char == "*" and self._peek(1) == "/": |
| 101 | self._advance(2) # Skip */ |
| 102 | break |
| 103 | self._advance() |
| 104 | |
| 105 | def _parse(self): |
| 106 | if self.completed and not self.stack: |
| 107 | return |
| 108 | if self.result is None: |
| 109 | self.result = self._parse_value() |
| 110 | else: |
| 111 | self._continue_parsing() |
| 112 | |
| 113 | def _continue_parsing(self): |
| 114 | while self.current_char is not None: |
| 115 | if self.completed and not self.stack: |
| 116 | return |
| 117 | if isinstance(self.result, dict): |
| 118 | self._parse_object_content() |
| 119 | elif isinstance(self.result, list): |
| 120 | self._parse_array_content() |
| 121 | elif isinstance(self.result, str): |
| 122 | self.result = self._parse_string() |
| 123 | else: |
| 124 | break |
| 125 | |
| 126 | def _parse_value(self): |
| 127 | self._skip_whitespace() |
| 128 | if self.current_char == "{": |
| 129 | # Only treat doubled braces as a wrapper at the root; nested objects |
| 130 | # must keep their closing braces paired correctly. |
| 131 | if not self.stack and self._peek(1) == "{": # Handle {{ |
| 132 | self._advance(2) |
| 133 | return self._parse_object() |
| 134 | elif self.current_char == "[": |
| 135 | return self._parse_array() |
| 136 | elif self.current_char in ['"', "'", "`"]: |
| 137 | if self._peek(2) == self.current_char * 2: # type: ignore |
| 138 | return self._parse_multiline_string() |
| 139 | return self._parse_string() |
| 140 | elif self.current_char and ( |
| 141 | self.current_char.isdigit() or self.current_char in ["-", "+"] |
| 142 | ): |
| 143 | return self._parse_number() |
| 144 | elif self._match("true"): |
| 145 | return True |
| 146 | elif self._match("false"): |
| 147 | return False |
| 148 | elif self._match("null") or self._match("undefined"): |
| 149 | return None |
| 150 | elif self.current_char: |
| 151 | return self._parse_unquoted_string() |
| 152 | return None |
| 153 | |
| 154 | def _match(self, text: str) -> bool: |
| 155 | # first char should match current char |
| 156 | if not self.current_char or self.current_char.lower() != text[0].lower(): |
| 157 | return False |
| 158 | |
| 159 | # peek remaining chars |
| 160 | remaining = len(text) - 1 |
| 161 | if self._peek(remaining).lower() == text[1:].lower(): |
| 162 | self._advance(len(text)) |
| 163 | return True |
| 164 | return False |
| 165 | |
| 166 | def _parse_object(self): |
| 167 | obj = {} |
| 168 | self._advance() # Skip opening brace |
| 169 | self.stack.append(obj) |
| 170 | self._parsing_started = True |
| 171 | self._parse_object_content() |
| 172 | return obj |
| 173 | |
| 174 | def _parse_object_content(self): |
| 175 | while self.current_char is not None: |
| 176 | self._skip_whitespace() |
| 177 | if self.current_char == "}": |
| 178 | # Root-level wrapper outputs may end in "}}"; nested objects must |
| 179 | # still close one brace at a time. |
| 180 | if len(self.stack) == 1 and self._peek(1) == "}": # Handle }} |
| 181 | self._advance(2) |
| 182 | else: |
| 183 | self._advance() |
| 184 | self._pop_stack(root_closed=True) |
| 185 | return |
| 186 | if self.current_char is None: |
| 187 | self._pop_stack() |
| 188 | return # End of input reached while parsing object |
| 189 | |
| 190 | key = self._parse_key() |
| 191 | value = None |
| 192 | self._skip_whitespace() |
| 193 | |
| 194 | if self.current_char == ":": |
| 195 | self._advance() |
| 196 | value = self._parse_value() |
| 197 | elif self.current_char is None: |
| 198 | value = None # End of input reached after key |
| 199 | else: |
| 200 | value = self._parse_value() |
| 201 | |
| 202 | self.stack[-1][key] = value |
| 203 | |
| 204 | self._skip_whitespace() |
| 205 | if self.current_char == ",": |
| 206 | self._advance() |
| 207 | continue |
| 208 | elif self.current_char != "}": |
| 209 | if self.current_char is None: |
| 210 | self._pop_stack() |
| 211 | return # End of input reached after value |
| 212 | continue |
| 213 | |
| 214 | def _parse_key(self): |
| 215 | self._skip_whitespace() |
| 216 | if self.current_char in ['"', "'"]: |
| 217 | return self._parse_string(is_key=True) |
| 218 | else: |
| 219 | return self._parse_unquoted_key() |
| 220 | |
| 221 | def _parse_unquoted_key(self): |
| 222 | result = "" |
| 223 | while ( |
| 224 | self.current_char is not None |
| 225 | and not self.current_char.isspace() |
| 226 | and self.current_char not in [":", ",", "}", "]"] |
| 227 | ): |
| 228 | result += self.current_char |
| 229 | self._advance() |
| 230 | return result |
| 231 | |
| 232 | def _parse_array(self): |
| 233 | arr = [] |
| 234 | self._advance() # Skip opening bracket |
| 235 | self.stack.append(arr) |
| 236 | self._parsing_started = True |
| 237 | self._parse_array_content() |
| 238 | return arr |
| 239 | |
| 240 | def _parse_array_content(self): |
| 241 | while self.current_char is not None: |
| 242 | self._skip_whitespace() |
| 243 | if self.current_char == "]": |
| 244 | self._advance() |
| 245 | self._pop_stack(root_closed=True) |
| 246 | return |
| 247 | value = self._parse_value() |
| 248 | self.stack[-1].append(value) |
| 249 | self._skip_whitespace() |
| 250 | if self.current_char == ",": |
| 251 | self._advance() |
| 252 | # handle trailing commas, end of array |
| 253 | self._skip_whitespace() |
| 254 | if self.current_char is None or self.current_char == "]": |
| 255 | if self.current_char == "]": |
| 256 | self._advance() |
| 257 | self._pop_stack(root_closed=True) |
| 258 | return |
| 259 | elif self.current_char != "]": |
| 260 | self._pop_stack() |
| 261 | return |
| 262 | |
| 263 | def _parse_string(self, is_key: bool = False): |
| 264 | result = "" |
| 265 | quote_char = self.current_char |
| 266 | self._advance() # Skip opening quote |
| 267 | while self.current_char is not None: |
| 268 | if self.current_char == quote_char: |
| 269 | if self._is_closing_quote(is_key): |
| 270 | break |
| 271 | result += self.current_char |
| 272 | self._advance() |
| 273 | continue |
| 274 | |
| 275 | if self.current_char == "\\": |
| 276 | self._advance() |
| 277 | if self.current_char in ['"', "'", "\\", "/", "b", "f", "n", "r", "t"]: |
| 278 | result += { |
| 279 | "b": "\b", |
| 280 | "f": "\f", |
| 281 | "n": "\n", |
| 282 | "r": "\r", |
| 283 | "t": "\t", |
| 284 | }.get(self.current_char, self.current_char) |
| 285 | elif self.current_char == "u": |
| 286 | self._advance() # Skip 'u' |
| 287 | unicode_char = "" |
| 288 | # Try to collect exactly 4 hex digits |
| 289 | for _ in range(4): |
| 290 | if self.current_char is None or not self.current_char.isalnum(): |
| 291 | # If we can't get 4 hex digits, treat it as a literal '\u' followed by whatever we got |
| 292 | return result + "\\u" + unicode_char |
| 293 | unicode_char += self.current_char |
| 294 | self._advance() |
| 295 | try: |
| 296 | result += chr(int(unicode_char, 16)) |
| 297 | except ValueError: |
| 298 | # If invalid hex value, treat as literal |
| 299 | result += "\\u" + unicode_char |
| 300 | continue |
| 301 | else: |
| 302 | result += self.current_char |
| 303 | self._advance() |
| 304 | if self.current_char == quote_char: |
| 305 | self._advance() # Skip closing quote |
| 306 | return result |
| 307 | |
| 308 | def _is_closing_quote(self, is_key: bool) -> bool: |
| 309 | next_index = self._skip_padding_from(self.index + 1) |
| 310 | if next_index >= len(self.json_string): |
| 311 | return True |
| 312 | |
| 313 | next_char = self.json_string[next_index] |
| 314 | if is_key: |
| 315 | return next_char in [":", ",", "}", "]"] |
| 316 | |
| 317 | if next_char in [",", "}", "]"]: |
| 318 | return True |
| 319 | |
| 320 | return self._looks_like_missing_comma_before_key(next_index) |
| 321 | |
| 322 | def _looks_like_missing_comma_before_key(self, index: int) -> bool: |
| 323 | if not self.stack or not isinstance(self.stack[-1], dict): |
| 324 | return False |
| 325 | if index >= len(self.json_string) or self.json_string[index] not in ['"', "'"]: |
| 326 | return False |
| 327 | |
| 328 | quote_char = self.json_string[index] |
| 329 | index += 1 |
| 330 | while index < len(self.json_string): |
| 331 | char = self.json_string[index] |
| 332 | if char == "\\": |
| 333 | index += 2 |
| 334 | continue |
| 335 | if char == quote_char: |
| 336 | next_index = self._skip_padding_from(index + 1) |
| 337 | return ( |
| 338 | next_index < len(self.json_string) |
| 339 | and self.json_string[next_index] == ":" |
| 340 | ) |
| 341 | if char in ["\n", "\r", "{", "}", "[", "]", ","]: |
| 342 | return False |
| 343 | index += 1 |
| 344 | |
| 345 | return False |
| 346 | |
| 347 | def _skip_padding_from(self, index: int) -> int: |
| 348 | while index < len(self.json_string): |
| 349 | char = self.json_string[index] |
| 350 | if char.isspace(): |
| 351 | index += 1 |
| 352 | elif char == "/" and index + 1 < len(self.json_string): |
| 353 | next_char = self.json_string[index + 1] |
| 354 | if next_char == "/": |
| 355 | index += 2 |
| 356 | while index < len(self.json_string) and self.json_string[index] != "\n": |
| 357 | index += 1 |
| 358 | elif next_char == "*": |
| 359 | end = self.json_string.find("*/", index + 2) |
| 360 | if end == -1: |
| 361 | return len(self.json_string) |
| 362 | index = end + 2 |
| 363 | else: |
| 364 | break |
| 365 | else: |
| 366 | break |
| 367 | return index |
| 368 | |
| 369 | def _parse_multiline_string(self): |
| 370 | result = "" |
| 371 | quote_char = self.current_char |
| 372 | self._advance(3) # Skip first quote |
| 373 | while self.current_char is not None: |
| 374 | if self.current_char == quote_char and self._peek(2) == quote_char * 2: # type: ignore |
| 375 | self._advance(3) # Skip first quote |
| 376 | break |
| 377 | result += self.current_char |
| 378 | self._advance() |
| 379 | return result.strip() |
| 380 | |
| 381 | def _parse_number(self): |
| 382 | number_str = "" |
| 383 | while self.current_char is not None and ( |
| 384 | self.current_char.isdigit() |
| 385 | or self.current_char in ["-", "+", ".", "e", "E"] |
| 386 | ): |
| 387 | number_str += self.current_char |
| 388 | self._advance() |
| 389 | try: |
| 390 | return int(number_str) |
| 391 | except ValueError: |
| 392 | return float(number_str) |
| 393 | |
| 394 | def _parse_unquoted_string(self): |
| 395 | result = "" |
| 396 | while self.current_char is not None and self.current_char not in [ |
| 397 | ":", |
| 398 | ",", |
| 399 | "}", |
| 400 | "]", |
| 401 | ]: |
| 402 | result += self.current_char |
| 403 | self._advance() |
| 404 | self._advance() |
| 405 | return result.strip() |
| 406 | |
| 407 | def _peek(self, n): |
| 408 | peek_index = self.index + 1 |
| 409 | result = "" |
| 410 | for _ in range(n): |
| 411 | if peek_index < len(self.json_string): |
| 412 | result += self.json_string[peek_index] |
| 413 | peek_index += 1 |
| 414 | else: |
| 415 | break |
| 416 | return result |
| 417 | |
| 418 | def get_start_pos(self, input_str: str) -> int: |
| 419 | chars = ["{", "[", '"'] |
| 420 | indices = [input_str.find(char) for char in chars if input_str.find(char) != -1] |
| 421 | return min(indices) if indices else 0 |