134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
|
||
import random
|
||
import string
|
||
from hashlib import md5
|
||
from typing import Any, Dict, List, Union, Optional, get_origin, get_args, Type
|
||
from pydantic import BaseModel
|
||
|
||
def gen_random_str(length=10, prefix="", suffix=""):
|
||
"""
|
||
生成随机字符串，默认长度为10，可添加前缀和后缀
|
||
"""
|
||
random_str = prefix + ''.join(random.choices(string.ascii_letters + string.digits, k=length)) + suffix
|
||
return md5(random_str.encode()).hexdigest()
|
||
|
||
|
||
def split_generator(iterable, chunk_size=50):
|
||
"""
|
||
将可迭代对象拆分为多个子列表
|
||
"""
|
||
chunk = []
|
||
for item in iterable:
|
||
chunk.append(item)
|
||
if len(chunk) == chunk_size:
|
||
yield chunk
|
||
chunk = []
|
||
if chunk:
|
||
yield chunk
|
||
|
||
async def async_split_generator(iterable, chunk_size=50):
|
||
"""
|
||
将可迭代对象拆分为多个子列表
|
||
"""
|
||
chunk = []
|
||
async for item in iterable:
|
||
chunk.append(item)
|
||
if len(chunk) == chunk_size:
|
||
yield chunk
|
||
chunk = []
|
||
if chunk:
|
||
yield chunk
|
||
|
||
async def async_generator_to_list(iterable):
|
||
"""
|
||
将可迭代对象转换为列表
|
||
"""
|
||
return [item async for item in iterable]
|
||
|
||
TYPE_MAPPING = {
|
||
"str": "string",
|
||
"int": "number",
|
||
"float": "number",
|
||
"bool": "boolean",
|
||
"list": "array",
|
||
"dict": "object",
|
||
"NoneType": "null", # 注意：None 的类型名是 'NoneType'
|
||
}
|
||
|
||
def transform_pydantic_to_list(
|
||
cls: Type[BaseModel],
|
||
prefix: str = "" # 新增：当前路径前缀
|
||
) -> List[Dict[str, Any]]:
|
||
if not issubclass(cls, BaseModel):
|
||
raise TypeError(f"{cls} is not a Pydantic BaseModel")
|
||
|
||
result = []
|
||
exclude_filter = getattr(cls, "__exclude_filter__", [])
|
||
meta_mapping = getattr(cls, "__meta_mapping__", {})
|
||
|
||
for field_name, field_info in cls.model_fields.items():
|
||
if field_name in exclude_filter:
|
||
continue
|
||
|
||
# 构建当前字段的完整路径
|
||
current_path = f"{prefix}.{field_name}" if prefix else field_name
|
||
|
||
field_meta = meta_mapping.get(field_name, {}) or {}
|
||
|
||
raw_type = field_info.annotation
|
||
children = []
|
||
display_type_name = "any"
|
||
|
||
origin = get_origin(raw_type)
|
||
args = get_args(raw_type)
|
||
|
||
# 处理 Optional[T]（即 Union[T, None]）
|
||
if origin is Union and type(None) in args:
|
||
# 提取非 None 的类型
|
||
non_none_types = [arg for arg in args if arg is not type(None)]
|
||
if len(non_none_types) == 1:
|
||
inner_type = non_none_types[0]
|
||
else:
|
||
inner_type = raw_type # 多类型 Union，暂不深入处理
|
||
else:
|
||
inner_type = raw_type
|
||
|
||
# 判断是否为 List[BaseModel]
|
||
list_origin = get_origin(inner_type)
|
||
if list_origin is list:
|
||
item_type = get_args(inner_type)[0] if get_args(inner_type) else Any
|
||
if isinstance(item_type, type) and issubclass(item_type, BaseModel):
|
||
children = transform_pydantic_to_list(item_type, prefix=current_path)
|
||
display_type_name = "list"
|
||
elif isinstance(inner_type, type) and issubclass(inner_type, BaseModel):
|
||
children = transform_pydantic_to_list(inner_type, prefix=current_path)
|
||
display_type_name = "object"
|
||
else:
|
||
# 基础类型：获取类型名
|
||
if inner_type is type(None):
|
||
display_type_name = "NoneType"
|
||
elif hasattr(inner_type, '__name__'):
|
||
display_type_name = inner_type.__name__
|
||
else:
|
||
display_type_name = str(inner_type)
|
||
|
||
frontend_type = TYPE_MAPPING.get(display_type_name, "string")
|
||
|
||
item = {
|
||
"value": current_path, # ✅ 使用层级路径
|
||
"label": field_info.description or field_name,
|
||
"type": frontend_type,
|
||
**field_meta,
|
||
}
|
||
if children:
|
||
item["children"] = children
|
||
result.append(item)
|
||
|
||
return result
|
||
|
||
if __name__ == "__main__":
|
||
print(gen_random_str())
|
||
print(gen_random_str())
|
||
print(gen_random_str())
|
||
print(gen_random_str())
|