350 lines
12 KiB
Python
350 lines
12 KiB
Python
# coding=utf8
|
|
import json
|
|
from typing import Any, List, NamedTuple, Optional, Tuple
|
|
import json.decoder
|
|
from json.decoder import JSONDecodeError as PyJSONDecodeError, JSONDecoder, py_scanstring
|
|
from json.scanner import py_make_scanner
|
|
from typing import Any, Dict, NamedTuple, Optional, Tuple, Union
|
|
|
|
|
|
class FixResult(NamedTuple):
|
|
success: bool
|
|
line: str
|
|
origin: bool
|
|
|
|
|
|
class JSONFixer:
|
|
def __init__(self, max_try: int = 20, max_stack: int = 3, *, js_style: bool = False) -> None:
|
|
self._max_try = max_try
|
|
self._max_stack = max_stack
|
|
self._js_style = js_style
|
|
self.last_fix: Optional[bool] = None
|
|
self.fix_stack: List[str] = []
|
|
|
|
def fix(self, line: str, *, strict: bool = True) -> FixResult:
|
|
try:
|
|
json.loads(line, strict=strict)
|
|
return FixResult(success=True, line=line, origin=True)
|
|
except Exception:
|
|
pass
|
|
|
|
ok, new_line = self.fixwithtry(line, strict=strict)
|
|
return FixResult(success=ok, line=new_line, origin=False)
|
|
|
|
def fixwithtry(self, line: str, *, strict: bool = True) -> Tuple[bool, str]:
|
|
if self._max_try <= 0:
|
|
return False, line
|
|
|
|
self.fix_stack = []
|
|
self.last_fix = None
|
|
|
|
ok = False
|
|
for _ in range(self._max_try):
|
|
ok, new_line = self.patch_line(line, strict=strict)
|
|
if ok:
|
|
return ok, new_line
|
|
|
|
self.last_fix = line != new_line
|
|
if self.last_fix:
|
|
self.fix_stack.insert(0, new_line)
|
|
self.fix_stack = self.fix_stack[: self._max_stack]
|
|
|
|
line = new_line
|
|
return ok, line
|
|
|
|
def patch_line(self, line: str, *, strict: bool = True) -> Tuple[bool, str]:
|
|
result = decode_line(line, strict=strict)
|
|
if result.success:
|
|
return True, line
|
|
|
|
if isinstance(result.exception, ValueError):
|
|
return self.patch_value_error(line, result.err_info)
|
|
|
|
if isinstance(result.exception, StopIteration):
|
|
return self.patch_stop_iteration(line)
|
|
|
|
if result.exception is None:
|
|
return self.patch_half_parse(line, result.err_info)
|
|
|
|
return False, line
|
|
|
|
def patch_value_error(self, line: str, err_info: Any) -> Tuple[bool, str]:
|
|
if err_info["error"] is None:
|
|
return False, line
|
|
|
|
error = err_info["error"]
|
|
pos = err_info["pos"]
|
|
nextchar = line[pos: pos + 1]
|
|
lastchar = line[pos - 1: pos]
|
|
nextline = line[pos:]
|
|
lastline = line[:pos]
|
|
|
|
if error == errors.StringUnterminatedString:
|
|
return False, insert_line(line, '"', len(line))
|
|
if error == errors.ObjectExceptKey:
|
|
if nextchar == "":
|
|
return False, insert_line(line, "}", pos)
|
|
if nextchar == ":":
|
|
return False, insert_line(line, '""', pos)
|
|
if lastchar in "{," and nextchar == ",":
|
|
return False, remove_line(line, pos, pos + 1)
|
|
if lastchar == "," and nextchar == "}":
|
|
return False, remove_line(line, pos - 1, pos)
|
|
if nextchar in "[{":
|
|
return False, insert_line(line, '"":', pos)
|
|
if self._js_style:
|
|
# find 'abc'
|
|
if nextchar == "'":
|
|
nextline = remove_line(nextline, 0, 1)
|
|
idx = nextline.find(":")
|
|
if idx != -1 and idx != 0 and nextline[idx - 1] == "'":
|
|
nextline = remove_line(nextline, idx - 1, idx)
|
|
|
|
return False, lastline + nextline
|
|
# abc:1 --> "aabc":1
|
|
idx = nextline.find(":")
|
|
if idx != -1:
|
|
line = lastline + insert_line(nextline, '"', idx)
|
|
return False, insert_line(line, '"', pos)
|
|
# TODO process more case "
|
|
return False, insert_line(line, '"', pos)
|
|
if error == errors.ObjectExceptColon:
|
|
return False, insert_line(line, ":", pos)
|
|
if error == errors.ObjectExceptObject:
|
|
if nextchar == "":
|
|
if lastchar == "{":
|
|
return False, insert_line(line, "}", pos)
|
|
return False, insert_line(line, "null}", pos)
|
|
if nextchar == "}":
|
|
return False, insert_line(line, "null", pos)
|
|
# TODO guess more
|
|
return False, insert_line(line, '"', pos)
|
|
if error == errors.ObjectExceptComma:
|
|
if nextchar == "":
|
|
return False, insert_line(line, "}", pos)
|
|
return False, insert_line(line, ",", pos)
|
|
if error == errors.ArrayExceptObject:
|
|
if nextchar == "," and lastchar == "[":
|
|
return False, remove_line(line, pos, pos + 1)
|
|
if nextchar == ",":
|
|
return False, insert_line(line, "null", pos)
|
|
if nextchar == "]":
|
|
return False, remove_line(line, pos - 1, pos)
|
|
if nextchar == "":
|
|
if lastchar == "[":
|
|
return False, insert_line(line, "]", pos)
|
|
return False, insert_line(line, "null]", pos)
|
|
# TODO guess more?
|
|
return False, insert_line(line, "{", pos)
|
|
if error == errors.ArrayExceptComma:
|
|
if len(line) == pos:
|
|
return False, insert_line(line, "]", pos)
|
|
return False, insert_line(line, ",", pos)
|
|
# TODO unknonwn
|
|
return False, line
|
|
|
|
def patch_stop_iteration(self, line: str) -> Tuple[bool, str]:
|
|
# TODO clean
|
|
# TODO fix
|
|
# 1. }]
|
|
# 2. ]}
|
|
# 3. constans
|
|
# 4. -
|
|
# 先 patch 完 {[]}
|
|
# TODO: process number
|
|
if line.startswith("-."):
|
|
new_line = "-0." + line[2:]
|
|
return False, new_line
|
|
# patch
|
|
left = patch_lastest_left_object_and_array(line)
|
|
if left == "":
|
|
if not self.last_fix:
|
|
left = patch_guess_left(line)
|
|
|
|
new_line = left + line
|
|
return False, new_line
|
|
|
|
def patch_half_parse(self, line: str, err_info: Any) -> Tuple[bool, str]:
|
|
obj, end = err_info
|
|
nextline = line[end:].strip()
|
|
nextchar = nextline[:1]
|
|
left = patch_lastest_left_object_and_array(nextline)
|
|
# ??
|
|
if left == "":
|
|
if nextchar == ",":
|
|
left = "["
|
|
elif nextchar == ":" and isinstance(obj, str):
|
|
left = "{"
|
|
else:
|
|
if not self.last_fix:
|
|
left = patch_guess_left(nextline)
|
|
|
|
new_line = left + line[:end] + nextline
|
|
return False, new_line
|
|
|
|
|
|
# TODO better name
|
|
def patch_lastest_left_object_and_array(line: str) -> str:
|
|
# '}]{[' --> '[{}]{['
|
|
pairs = {"}": "{", "]": "["}
|
|
breaks = "{["
|
|
left = ""
|
|
for char in line:
|
|
if char in breaks:
|
|
break
|
|
if char in pairs:
|
|
left = pairs[char] + left
|
|
|
|
return left
|
|
|
|
|
|
# TODO better name
|
|
# TODO 改成 lastest
|
|
# TODO {}}]]]] --> { not [
|
|
def patch_guess_left(line: str) -> str:
|
|
miss_object = line.count("}") - line.count("{")
|
|
miss_array = line.count("]") - line.count("[")
|
|
if miss_object == miss_array == 0:
|
|
if line[-1:] == '"' and line.count('"') == 1:
|
|
return '"'
|
|
elif miss_object >= miss_array:
|
|
return "{"
|
|
else:
|
|
return "["
|
|
return ""
|
|
|
|
|
|
def insert_line(line: str, value: str, pos: int) -> str:
|
|
return line[:pos] + value + line[pos:]
|
|
|
|
|
|
def remove_line(line: str, start: int, end: int) -> str:
|
|
return line[:start] + line[end:]
|
|
|
|
|
|
class JSONDecodeError:
|
|
def __init__(self, parser, message):
|
|
self.message = message
|
|
self.parser = parser
|
|
|
|
def __eq__(self, err):
|
|
return err.parser == self.parser and self.message in err.message
|
|
|
|
|
|
class errors:
|
|
StringInvalidUXXXXEscape = JSONDecodeError(
|
|
"py_scanstring", "Invalid \\uXXXX escape")
|
|
# 2 different case
|
|
StringUnterminatedString = JSONDecodeError(
|
|
"py_scanstring", "Unterminated string starting at")
|
|
StringInvalidControlCharacter = JSONDecodeError(
|
|
"py_scanstring", "Invalid control character")
|
|
StringInvalidEscape = JSONDecodeError("py_scanstring", "Invalid \\escape")
|
|
ObjectExceptColon = JSONDecodeError(
|
|
"JSONObject", "Expecting ':' delimiter")
|
|
ObjectExceptObject = JSONDecodeError("JSONObject", "Expecting value")
|
|
# 2 different case
|
|
ObjectExceptKey = JSONDecodeError(
|
|
"JSONObject", "Expecting property name enclosed in double quotes")
|
|
ObjectExceptComma = JSONDecodeError(
|
|
"JSONObject", "Expecting ',' delimiter")
|
|
ArrayExceptObject = JSONDecodeError("JSONArray", "Expecting value")
|
|
ArrayExceptComma = JSONDecodeError("JSONArray", "Expecting ',' delimiter")
|
|
|
|
@classmethod
|
|
def get_decode_error(cls, parser, message):
|
|
err = JSONDecodeError(parser, message)
|
|
for _, value in cls.__dict__.items():
|
|
if isinstance(value, JSONDecodeError):
|
|
if err == value:
|
|
return value
|
|
return None
|
|
|
|
"""
|
|
01 先不看,不研究
|
|
02 badcase: " --> "" success
|
|
03 控制符 pass
|
|
04 unicode \\u 的 pass
|
|
05 同上
|
|
06 object 后面没有跟随 " , badcase: {abc":1} --> {"abc":1}
|
|
07 object key 后面没有 : , badcase: {"abc"1} --> {"abc":1}
|
|
08 object 开始检测 Value 收到 StopIteration
|
|
08.1 要么后面没有了
|
|
08.2 要么后面不是 "/{/[/n[ull]/t[rue]/f[alse]/number/NaN/Infinity/-Infinity 开头的东西
|
|
-- 08.1 后面补上 null}
|
|
-- 08.2 无脑补一个 "
|
|
09 object 解析完一个 pair 后,下一个不是}, 期待一个 ','
|
|
badcase {"k":1"s":2}
|
|
10 在 09 的基础上解析完{"k":1, 发现下一个不是 ", 这个后面再优化(暂时和 06 一致)
|
|
badcase {"k":1,x":2}
|
|
11 array 开始检测 Value 收到 StopIteration
|
|
11.1 要么后面没有了,补上]
|
|
11.2 同 08.2,无脑补一个{ 看看
|
|
12 array 解析完前一个 object, 需要一个 ,
|
|
这里 nextchar 既不是 ] 也不是, 代表这个 nextchar 的 end 也已经+1 了，所以减 2
|
|
"""
|
|
|
|
|
|
def errmsg_inv(e: ValueError) -> Dict[str, Any]:
|
|
assert isinstance(e, PyJSONDecodeError)
|
|
parser = e.__dict__.get("parser", "")
|
|
errmsg = e.msg
|
|
localerr = errors.get_decode_error(parser, errmsg)
|
|
return {
|
|
"parsers": e.__dict__.get("parsers", []),
|
|
"error": localerr,
|
|
"lineno": e.lineno,
|
|
"colno": e.colno,
|
|
"pos": e.pos,
|
|
}
|
|
|
|
|
|
def record_parser_name(parser: Any) -> Any:
|
|
def new_parser(*args: Any, **kwargs: Any) -> Any:
|
|
try:
|
|
return parser(*args, **kwargs)
|
|
except Exception as e:
|
|
if "parser" not in e.__dict__:
|
|
e.__dict__["parser"] = parser.__name__
|
|
if "parsers" not in e.__dict__:
|
|
e.__dict__["parsers"] = []
|
|
e.__dict__["parsers"].append(parser.__name__)
|
|
raise e
|
|
|
|
return new_parser
|
|
|
|
|
|
def make_decoder(*, strict: bool = True) -> JSONDecoder:
|
|
json.decoder.scanstring = record_parser_name(py_scanstring)
|
|
|
|
decoder = JSONDecoder(strict=strict)
|
|
decoder.parse_object = record_parser_name(decoder.parse_object)
|
|
decoder.parse_array = record_parser_name(decoder.parse_array)
|
|
decoder.parse_string = record_parser_name(py_scanstring)
|
|
decoder.parse_object = record_parser_name(decoder.parse_object)
|
|
|
|
decoder.scan_once = py_make_scanner(decoder)
|
|
return decoder
|
|
|
|
|
|
decoder = make_decoder()
|
|
decoder_unstrict = make_decoder(strict=False)
|
|
|
|
|
|
class DecodeResult(NamedTuple):
|
|
success: bool
|
|
exception: Optional[Exception]
|
|
err_info: Optional[Union[Dict[str, Any], Tuple[Any, Any]]]
|
|
|
|
|
|
def decode_line(line: str, *, strict: bool = True) -> DecodeResult:
|
|
try:
|
|
obj, end = (decoder if strict else decoder_unstrict).scan_once(line, 0)
|
|
ok = end == len(line)
|
|
return DecodeResult(success=ok, exception=None, err_info=(obj, end))
|
|
except StopIteration as e:
|
|
return DecodeResult(success=False, exception=e, err_info=None)
|
|
except ValueError as e:
|
|
err_info = errmsg_inv(e)
|
|
return DecodeResult(success=False, exception=e, err_info=err_info)
|