Spaces:
Sleeping
Sleeping
File size: 5,997 Bytes
dbe2c62 |
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 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
from typing import Dict, List, Any
class JSONSchemaExtractor:
def __init__(self, list_policy: str = "first", verbose: bool = True) -> None:
"""
:param list_policy: "first" | "union"
- "first": nếu gặp list các object, lấy schema theo PHẦN TỬ ĐẦU (như bản gốc).
- "union": duyệt mọi phần tử, hợp nhất các field/type.
"""
assert list_policy in ("first", "union"), "list_policy must be 'first' or 'union'"
self.list_policy = list_policy
self.verbose = verbose
self._processed_fields: set[str] = set()
self._full_schema: Dict[str, str] = {}
# =====================================
# 1) Chuẩn hóa kiểu dữ liệu
# =====================================
@staticmethod
def get_standard_type(value: Any) -> str:
if isinstance(value, bool):
return "boolean"
elif isinstance(value, int):
return "number"
elif isinstance(value, float):
return "number"
elif isinstance(value, str):
return "string"
elif isinstance(value, list):
return "array"
elif isinstance(value, dict):
return "object"
elif value is None:
return "null"
return "unknown"
# =====================================
# 2) Hợp nhất kiểu (null / mixed)
# =====================================
def _merge_type(self, key: str, new_type: str, item_index: int) -> None:
"""
Cập nhật self._full_schema[key] theo quy tắc:
- Nếu chưa có: đặt = new_type và log "New: ..."
- Nếu khác:
+ Nếu new_type == "null": giữ kiểu cũ.
+ Nếu kiểu cũ == "null": cập nhật = new_type.
+ Ngược lại: nếu khác nhau và chưa "mixed" => set "mixed" và cảnh báo.
"""
if key not in self._full_schema:
self._full_schema[key] = new_type
self._processed_fields.add(key)
return
old_type = self._full_schema[key]
if old_type == new_type:
return
if new_type == "null":
return
if old_type == "null":
self._full_schema[key] = new_type
return
if old_type != "mixed":
self._full_schema[key] = "mixed"
# =====================================
# 3) Đệ quy trích xuất schema
# =====================================
def _extract_schema_from_obj(self, data: Dict[str, Any], prefix: str, item_index: int) -> None:
"""
Duyệt dict hiện tại, cập nhật _full_schema với kiểu tại key (phẳng),
và nếu là object/array lồng thì đệ quy theo quy tắc gốc.
"""
for key, value in data.items():
new_prefix = f"{prefix}{key}" if prefix else key
vtype = self.get_standard_type(value)
self._merge_type(new_prefix, vtype, item_index)
if isinstance(value, dict):
self._extract_schema_from_obj(value, f"{new_prefix}.", item_index)
elif isinstance(value, list) and value:
first = value[0]
if isinstance(first, dict):
if self.list_policy == "first":
self._extract_schema_from_obj(first, f"{new_prefix}.", item_index)
else: # union
for elem in value:
if isinstance(elem, dict):
self._extract_schema_from_obj(elem, f"{new_prefix}.", item_index)
elif isinstance(first, list):
if self.list_policy == "first":
self._extract_schema_from_list(first, f"{new_prefix}.", item_index)
else:
for elem in value:
if isinstance(elem, list):
self._extract_schema_from_list(elem, f"{new_prefix}.", item_index)
def _extract_schema_from_list(self, data_list: List[Any], prefix: str, item_index: int) -> None:
"""
Hỗ trợ cho trường hợp list lồng list (ít gặp). Duyệt tương tự _extract_schema_from_obj.
"""
if not data_list:
return
first = data_list[0]
if isinstance(first, dict):
if self.list_policy == "first":
self._extract_schema_from_obj(first, prefix, item_index)
else:
for elem in data_list:
if isinstance(elem, dict):
self._extract_schema_from_obj(elem, prefix, item_index)
elif isinstance(first, list):
if self.list_policy == "first":
self._extract_schema_from_list(first, prefix, item_index)
else:
for elem in data_list:
if isinstance(elem, list):
self._extract_schema_from_list(elem, prefix, item_index)
# =====================================
# 4) API chính (data/file)
# =====================================
def create_schema_from_data(self, data: Any) -> Dict[str, str]:
"""
Tạo schema từ biến Python (list | dict).
Giữ log giống bản gốc.
"""
self._processed_fields.clear()
self._full_schema.clear()
data_list = data if isinstance(data, list) else [data]
if not data_list:
raise ValueError("JSON data is empty")
for i, item in enumerate(data_list, 1):
if not isinstance(item, dict):
continue
self._extract_schema_from_obj(item, prefix="", item_index=i)
return dict(self._full_schema)
def schemaRun(self, SegmentDict: str) -> Dict[str, str]:
SchemaDict = self.create_schema_from_data(SegmentDict)
return SchemaDict |