238 lines
7.5 KiB
Python
238 lines
7.5 KiB
Python
import t, c
|
||
from stdint import *
|
||
import memhub
|
||
import string
|
||
import viperio
|
||
|
||
# ============================================================
|
||
# JSON 值类型枚举
|
||
# ============================================================
|
||
JSON_NULL: t.CDefine = 0
|
||
JSON_BOOL: t.CDefine = 1
|
||
JSON_INT: t.CDefine = 2
|
||
JSON_FLOAT: t.CDefine = 3
|
||
JSON_STRING: t.CDefine = 4
|
||
JSON_ARRAY: t.CDefine = 5
|
||
JSON_OBJECT: t.CDefine = 6
|
||
|
||
|
||
# ============================================================
|
||
# JsonValue - JSON 值节点
|
||
# 使用 mpool 分配:JsonValue(pool) 从池中分配内存
|
||
# ============================================================
|
||
class JsonValue:
|
||
vtype: t.CInt
|
||
bool_val: t.CInt
|
||
int_val: t.CInt64T
|
||
float_val: t.CDouble
|
||
str_val: t.CChar | t.CPtr
|
||
next: 'JsonValue' | t.CPtr
|
||
key: t.CChar | t.CPtr
|
||
child: 'JsonValue' | t.CPtr
|
||
child_count: t.CSizeT
|
||
|
||
def __new__(self, pool: memhub.MemManager | t.CPtr):
|
||
"""从 mpool 分配 JsonValue 结构体内存并零初始化"""
|
||
ptr: JsonValue | t.CPtr = pool.alloc(JsonValue.__sizeof__())
|
||
if ptr:
|
||
string.memset(ptr, 0, JsonValue.__sizeof__())
|
||
return ptr
|
||
|
||
def type(self) -> t.CInt:
|
||
return self.vtype
|
||
|
||
def is_null(self) -> bool:
|
||
return self.vtype == JSON_NULL
|
||
|
||
def is_bool(self) -> bool:
|
||
return self.vtype == JSON_BOOL
|
||
|
||
def is_int(self) -> bool:
|
||
return self.vtype == JSON_INT
|
||
|
||
def is_float(self) -> bool:
|
||
return self.vtype == JSON_FLOAT
|
||
|
||
def is_string(self) -> bool:
|
||
return self.vtype == JSON_STRING
|
||
|
||
def is_array(self) -> bool:
|
||
return self.vtype == JSON_ARRAY
|
||
|
||
def is_object(self) -> bool:
|
||
return self.vtype == JSON_OBJECT
|
||
|
||
def as_bool(self) -> bool:
|
||
return self.bool_val != 0
|
||
|
||
def as_int(self) -> t.CInt64T:
|
||
return self.int_val
|
||
|
||
def as_float(self) -> t.CDouble:
|
||
return self.float_val
|
||
|
||
def as_string(self) -> t.CChar | t.CPtr:
|
||
return self.str_val
|
||
|
||
def __len__(self) -> t.CSizeT:
|
||
return self.child_count
|
||
|
||
def __getitem__(self, key: t.CChar | t.CPtr) -> 'JsonValue' | t.CPtr:
|
||
"""Python风格取值:v["key"] 取对象字段"""
|
||
if self.vtype != JSON_OBJECT:
|
||
return None
|
||
if key == None:
|
||
return None
|
||
cur: JsonValue | t.CPtr = self.child
|
||
while cur != None:
|
||
if cur.key != None:
|
||
if cur.key == key:
|
||
return cur
|
||
cur = cur.next
|
||
return None
|
||
|
||
def __setitem__(self, key: t.CChar | t.CPtr, val: 'JsonValue' | t.CPtr):
|
||
"""Python风格赋值:v["key"] = val"""
|
||
if self.vtype != JSON_OBJECT or key == None or val == None:
|
||
return
|
||
cur: JsonValue | t.CPtr = self.child
|
||
while cur != None:
|
||
if cur.key != None:
|
||
if cur.key == key:
|
||
cur.vtype = val.vtype
|
||
cur.bool_val = val.bool_val
|
||
cur.int_val = val.int_val
|
||
cur.float_val = val.float_val
|
||
cur.str_val = val.str_val
|
||
cur.child = val.child
|
||
cur.child_count = val.child_count
|
||
return
|
||
cur = cur.next
|
||
# key 不存在,追加新子节点
|
||
klen: t.CSizeT = string.strlen(key)
|
||
kbuf: t.CChar | t.CPtr = self._pool_alloc(klen + 1)
|
||
if kbuf != None:
|
||
string.memcpy(kbuf, key, klen + 1)
|
||
val.key = kbuf
|
||
self._append_child(val)
|
||
|
||
def get_item(self, index: t.CSizeT) -> 'JsonValue' | t.CPtr:
|
||
"""按索引取数组元素:v.get_item(0)"""
|
||
if self.vtype != JSON_ARRAY and self.vtype != JSON_OBJECT:
|
||
return None
|
||
cur: JsonValue | t.CPtr = self.child
|
||
i: t.CSizeT = 0
|
||
while cur != None:
|
||
if i == index:
|
||
return cur
|
||
cur = cur.next
|
||
i += 1
|
||
return None
|
||
|
||
def _pool_alloc(self, size: t.CSizeT) -> t.CChar | t.CPtr:
|
||
"""内部方法:从当前对象的 mpool 分配内存(预留,暂返回 None)"""
|
||
return None
|
||
|
||
def _append_child(self, child: 'JsonValue' | t.CPtr):
|
||
if self.child == None:
|
||
self.child = child
|
||
else:
|
||
cur: JsonValue | t.CPtr = self.child
|
||
while cur.next != None:
|
||
cur = cur.next
|
||
cur.next = child
|
||
self.child_count += 1
|
||
|
||
|
||
# ============================================================
|
||
# 构造函数:创建各类 JsonValue(从 mpool 分配)
|
||
# ============================================================
|
||
|
||
def null(pool: memhub.MemManager | t.CPtr) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_NULL
|
||
return v
|
||
|
||
def bool_val(pool: memhub.MemManager | t.CPtr, val: bool) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_BOOL
|
||
v.bool_val = 1 if val else 0
|
||
return v
|
||
|
||
def int_val(pool: memhub.MemManager | t.CPtr, val: t.CInt64T) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_INT
|
||
v.int_val = val
|
||
return v
|
||
|
||
def float_val(pool: memhub.MemManager | t.CPtr, val: t.CDouble) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_FLOAT
|
||
v.float_val = val
|
||
return v
|
||
|
||
def string_val(pool: memhub.MemManager | t.CPtr, val: t.CChar | t.CPtr) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_STRING
|
||
if val != None:
|
||
slen: t.CSizeT = string.strlen(val)
|
||
buf: t.CChar | t.CPtr = pool.alloc(slen + 1)
|
||
if buf != None:
|
||
string.memcpy(buf, val, slen + 1)
|
||
v.str_val = buf
|
||
return v
|
||
|
||
def array(pool: memhub.MemManager | t.CPtr) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_ARRAY
|
||
return v
|
||
|
||
def object(pool: memhub.MemManager | t.CPtr) -> JsonValue | t.CPtr:
|
||
v: JsonValue | t.CPtr = JsonValue(pool)
|
||
if v == None: return None
|
||
v.vtype = JSON_OBJECT
|
||
return v
|
||
|
||
# 向数组追加元素
|
||
def array_append(pool: memhub.MemManager | t.CPtr, arr: JsonValue | t.CPtr, item: JsonValue | t.CPtr):
|
||
if arr == None or item == None: return
|
||
if arr.vtype != JSON_ARRAY: return
|
||
arr._append_child(item)
|
||
|
||
# 向对象设置键值对
|
||
def object_set(pool: memhub.MemManager | t.CPtr, obj: JsonValue | t.CPtr, key: t.CChar | t.CPtr, val: JsonValue | t.CPtr):
|
||
if obj == None or key == None or val == None: return
|
||
if obj.vtype != JSON_OBJECT: return
|
||
cur: JsonValue | t.CPtr = obj.child
|
||
while cur != None:
|
||
if cur.key != None:
|
||
if cur.key == key:
|
||
cur.vtype = val.vtype
|
||
cur.bool_val = val.bool_val
|
||
cur.int_val = val.int_val
|
||
cur.float_val = val.float_val
|
||
cur.str_val = val.str_val
|
||
cur.child = val.child
|
||
cur.child_count = val.child_count
|
||
return
|
||
cur = cur.next
|
||
klen: t.CSizeT = string.strlen(key)
|
||
kbuf: t.CChar | t.CPtr = pool.alloc(klen + 1)
|
||
if kbuf != None:
|
||
string.memcpy(kbuf, key, klen + 1)
|
||
val.key = kbuf
|
||
obj._append_child(val)
|
||
|
||
|
||
# ============================================================
|
||
# 导入子模块
|
||
# ============================================================
|
||
from .__parser import parse
|
||
from .__writer import write
|