fix: Accept raw JSON encodings
- Добавлена мягкая загрузка сырых JSON-файлов с корневым массивом точек. - Исправляются trailing commas перед закрывающими скобками и числа вида 1173. без дробной части. - При открытии файлов в UTF-16, UTF-16 LE/BE, CP1251 или CP1252 содержимое автоматически конвертируется и сохраняется в UTF-8. - Добавлены smoke-проверки для сырого circle_centers JSON, CP1251 и UTF-16 LE без BOM.
This commit is contained in:
parent
3699586cc2
commit
207d6ca86c
97
models.py
97
models.py
@ -40,6 +40,101 @@ _LED_POINT_KNOWN_FIELDS = {
|
||||
}
|
||||
|
||||
|
||||
def load_json_document(path: str | Path) -> dict[str, Any]:
|
||||
path = Path(path)
|
||||
text, was_converted = _read_text_as_utf8(path)
|
||||
if was_converted:
|
||||
path.write_text(text, encoding="utf-8")
|
||||
data = _loads_lenient_json(text)
|
||||
if isinstance(data, list):
|
||||
return {"circles": data}
|
||||
if isinstance(data, dict):
|
||||
return data
|
||||
raise ValueError("JSON must contain an object or an array of LED points.")
|
||||
|
||||
|
||||
def _read_text_as_utf8(path: Path) -> tuple[str, bool]:
|
||||
raw = path.read_bytes()
|
||||
try:
|
||||
text = raw.decode("utf-8-sig")
|
||||
if _looks_like_json_text(text):
|
||||
return text, False
|
||||
except UnicodeDecodeError:
|
||||
pass
|
||||
|
||||
for encoding in ("utf-16", "utf-16-le", "utf-16-be", "cp1251", "cp1252"):
|
||||
try:
|
||||
text = raw.decode(encoding)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if _looks_like_json_text(text):
|
||||
return text.lstrip("\ufeff"), True
|
||||
return raw.decode("utf-8", errors="replace"), True
|
||||
|
||||
|
||||
def _looks_like_json_text(text: str) -> bool:
|
||||
if "\x00" in text[:64]:
|
||||
return False
|
||||
stripped = text.lstrip("\ufeff \t\r\n")
|
||||
return stripped.startswith("{") or stripped.startswith("[")
|
||||
|
||||
|
||||
def _loads_lenient_json(text: str) -> Any:
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError as original_error:
|
||||
fixed_text = _sanitize_raw_json_text(text)
|
||||
if fixed_text == text:
|
||||
raise original_error
|
||||
try:
|
||||
return json.loads(fixed_text)
|
||||
except json.JSONDecodeError:
|
||||
raise original_error
|
||||
|
||||
|
||||
def _sanitize_raw_json_text(text: str) -> str:
|
||||
result: list[str] = []
|
||||
in_string = False
|
||||
escaped = False
|
||||
length = len(text)
|
||||
index = 0
|
||||
while index < length:
|
||||
char = text[index]
|
||||
if in_string:
|
||||
result.append(char)
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
escaped = True
|
||||
elif char == '"':
|
||||
in_string = False
|
||||
index += 1
|
||||
continue
|
||||
|
||||
if char == '"':
|
||||
in_string = True
|
||||
result.append(char)
|
||||
elif char == "," and _next_non_space(text, index + 1) in {"}", "]"}:
|
||||
pass
|
||||
elif char == "." and result and result[-1].isdigit() and not _next_char_is_digit(text, index + 1):
|
||||
result.append(".0")
|
||||
else:
|
||||
result.append(char)
|
||||
index += 1
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def _next_non_space(text: str, start: int) -> str | None:
|
||||
for index in range(start, len(text)):
|
||||
if not text[index].isspace():
|
||||
return text[index]
|
||||
return None
|
||||
|
||||
|
||||
def _next_char_is_digit(text: str, start: int) -> bool:
|
||||
return start < len(text) and text[start].isdigit()
|
||||
|
||||
|
||||
@dataclass
|
||||
class LedPoint:
|
||||
uid: int
|
||||
@ -149,7 +244,7 @@ class MappingDocument:
|
||||
def load(self, path: str | Path) -> None:
|
||||
self.clear()
|
||||
self.path = Path(path)
|
||||
data = json.loads(self.path.read_text(encoding="utf-8-sig"))
|
||||
data = load_json_document(self.path)
|
||||
self.extra_payload = {
|
||||
key: value
|
||||
for key, value in data.items()
|
||||
|
||||
54
qa_smoke.py
54
qa_smoke.py
@ -76,6 +76,9 @@ PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
def main() -> None:
|
||||
tests: list[Callable[[], None]] = [
|
||||
test_load_supported_json_files,
|
||||
test_load_raw_json_with_trailing_commas_and_bare_decimal_points,
|
||||
test_non_utf8_json_is_converted_to_utf8_on_load,
|
||||
test_utf16_le_json_without_bom_is_converted_to_utf8_on_load,
|
||||
test_line_limit_rejects_fifth_line,
|
||||
test_line_indexes_are_scoped_per_line,
|
||||
test_remove_selected_leds_blocks_and_lines,
|
||||
@ -219,6 +222,57 @@ def test_load_supported_json_files() -> None:
|
||||
assert len(doc.lines) <= MAX_LINES
|
||||
|
||||
|
||||
def test_load_raw_json_with_trailing_commas_and_bare_decimal_points() -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
path = Path(temp_dir) / "raw_circle_centers.json"
|
||||
path.write_text(
|
||||
"""
|
||||
[
|
||||
{"id":1,"x":1173.,"y":2159.019396,"row":35,"column":64,},
|
||||
{"id":2,"x":1207.,"y":17.,"row":36,"column":1,}
|
||||
]
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
doc = MappingDocument()
|
||||
doc.load(path)
|
||||
|
||||
assert doc.collection_name == "circles"
|
||||
assert len(doc.points) == 2
|
||||
assert doc.points[0].x == 1173.0
|
||||
assert doc.points[0].col == 64
|
||||
assert doc.points[1].y == 17.0
|
||||
|
||||
|
||||
def test_non_utf8_json_is_converted_to_utf8_on_load() -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
path = Path(temp_dir) / "cp1251.json"
|
||||
text = '{"circles":[{"id":1,"x":1.,"y":2.,"row":0,"column":0,"label":"точка"}]}'
|
||||
path.write_bytes(text.encode("cp1251"))
|
||||
|
||||
doc = MappingDocument()
|
||||
doc.load(path)
|
||||
|
||||
assert len(doc.points) == 1
|
||||
assert doc.points[0].extra_payload["label"] == "точка"
|
||||
assert path.read_bytes().decode("utf-8") == text
|
||||
|
||||
|
||||
def test_utf16_le_json_without_bom_is_converted_to_utf8_on_load() -> None:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
path = Path(temp_dir) / "utf16le.json"
|
||||
text = '[{"id":1,"x":1173.,"y":17.,"row":35,"column":64}]'
|
||||
path.write_bytes(text.encode("utf-16-le"))
|
||||
|
||||
doc = MappingDocument()
|
||||
doc.load(path)
|
||||
|
||||
assert len(doc.points) == 1
|
||||
assert doc.points[0].x == 1173.0
|
||||
assert path.read_bytes().decode("utf-8") == text
|
||||
|
||||
|
||||
def test_line_limit_rejects_fifth_line() -> None:
|
||||
with temp_document_path() as path:
|
||||
doc = MappingDocument()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user