MB-X Bilibili Pipeline
6 days ago ef678d7f95e5fc6348ad7b7ac3a89275d915342d
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
from __future__ import annotations
 
import json
from typing import Any
 
 
class StrictJsonError(ValueError):
    pass
 
 
def _pairs(values: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in values:
        if key in result:
            raise StrictJsonError(f"duplicate key: {key}")
        result[key] = value
    return result
 
 
def loads(payload: bytes) -> Any:
    try:
        text = payload.decode("utf-8", errors="strict")
        return json.loads(text, object_pairs_hook=_pairs, parse_constant=lambda value: (_ for _ in ()).throw(StrictJsonError(value)))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise StrictJsonError("invalid strict UTF-8 JSON") from exc
 
 
def canonical_bytes(value: Any, *, newline: bool = False) -> bytes:
    payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return payload + (b"\n" if newline else b"")