347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""
|
||
@project:serviceone
|
||
@File:DynamicModelClass.py
|
||
@IDE:PyCharm
|
||
@Author:徐彬程
|
||
@Date:2023/12/2 18:09
|
||
"""
|
||
import copy
|
||
import os.path
|
||
import re
|
||
from abc import abstractmethod
|
||
from django.db import models
|
||
from rest_framework import serializers
|
||
from django.utils import timezone
|
||
import json
|
||
from django.apps import apps
|
||
import requests
|
||
import configparser # 这里一定要全小写，不然会报找不到模块错误
|
||
from django.conf import settings
|
||
from dependency.classes.tree.treeClass import PythonCodeTreeNode, PythonCodeTreeController
|
||
|
||
cf = configparser.ConfigParser()
|
||
path = os.path.join(settings.BASE_DIR, 'dependency', 'json_data_in_need', 'ZeroCode.conf')
|
||
cf.read(path)
|
||
"""
|
||
data模版
|
||
data = [
|
||
{
|
||
"name": "uuid",
|
||
"field_type": "CharField",
|
||
"options": {
|
||
'max_length': 200
|
||
}
|
||
},
|
||
{
|
||
"name": "path_name",
|
||
"field_type": "CharField",
|
||
"options": {
|
||
"Article",
|
||
"null":True,
|
||
"blank":"True",
|
||
"on_delete":"models.SET_NULL",
|
||
"related_name":"'fileRecord'",
|
||
}
|
||
}
|
||
]
|
||
"""
|
||
|
||
|
||
class BuildDynamicModelClass:
|
||
"""
|
||
工厂模式
|
||
"""
|
||
ZeroCodeModelInfoData = []
|
||
|
||
@classmethod
|
||
def model_atr_transform(cls, data):
|
||
FIELD_TYPES = {
|
||
"CharField": models.CharField,
|
||
"IntegerField": models.IntegerField,
|
||
"JSONField": models.JSONField,
|
||
"ImageField": models.ImageField,
|
||
"IPAddressField": models.IPAddressField,
|
||
"TextField": models.TextField,
|
||
"TimeField": models.TimeField,
|
||
"ForeignKey": models.ForeignKey,
|
||
"ManyToManyField": models.ManyToManyField,
|
||
"DateTimeField": models.DateTimeField,
|
||
"BigIntegerField": models.BigIntegerField,
|
||
"FloatField": models.FloatField
|
||
}
|
||
OPTIONAL_FIELD_TYPES = {
|
||
"models.SET_NULL": models.SET_NULL,
|
||
"models.CASCADE": models.CASCADE,
|
||
"timezone.now": timezone.now
|
||
}
|
||
IGNORE_OPTIONS_LIST = {
|
||
"is_visible": "any",
|
||
"is_editable": "any",
|
||
"form_type": "any",
|
||
"data_show_fields": [],
|
||
"null": "False",
|
||
"blank": "False",
|
||
"primary_key": "False",
|
||
"unique": "False"
|
||
}
|
||
model_fields = {}
|
||
new_data = copy.deepcopy(data)
|
||
for f in new_data:
|
||
field_klass = FIELD_TYPES[f["field_type"]]
|
||
if f["field_type"] == 'ManyToManyField':
|
||
keys_to_keep = ["to", "through"]
|
||
f["options"] = {key: value for key, value in f["options"].items() if key in keys_to_keep}
|
||
continue
|
||
for k, v in f["options"].items():
|
||
if isinstance(v, str):
|
||
if v in OPTIONAL_FIELD_TYPES:
|
||
f["options"][k] = OPTIONAL_FIELD_TYPES[v]
|
||
for ignore_k, ignore_v in IGNORE_OPTIONS_LIST.items():
|
||
if ignore_k in f["options"].keys() and ignore_v == f["options"][ignore_k]:
|
||
del f["options"][ignore_k]
|
||
elif ignore_v == 'any' and ignore_k in f["options"].keys():
|
||
del f["options"][ignore_k]
|
||
model_fields[f["name"]] = field_klass(**f["options"])
|
||
return model_fields
|
||
|
||
@classmethod
|
||
def get_model(cls, table_name):
|
||
fields = cls.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"model_name": table_name,
|
||
}
|
||
})
|
||
# 结构化数据
|
||
progress_data = cls.model_atr_transform(fields["model_info_json"])
|
||
# 构建model类
|
||
try:
|
||
now_model = apps.get_model('dependency', table_name)
|
||
except LookupError:
|
||
now_model = cls.build_model(table_name, progress_data)
|
||
return now_model
|
||
|
||
@classmethod
|
||
def get_model_by_data(cls, table_name, data):
|
||
# 结构化数据
|
||
progress_data = cls.model_atr_transform(data)
|
||
# 构建model类
|
||
try:
|
||
now_model = apps.get_model('dependency', table_name)
|
||
except LookupError:
|
||
now_model = cls.build_model(table_name, progress_data)
|
||
return now_model
|
||
|
||
@classmethod
|
||
def get_model_sign(cls, table_name):
|
||
fields = BuildDynamicModelClass.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"model_name": table_name,
|
||
}
|
||
})
|
||
return fields["sign"]
|
||
|
||
@classmethod
|
||
def get_model_field(cls, table_name):
|
||
fields = BuildDynamicModelClass.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"model_name": table_name,
|
||
}
|
||
})
|
||
return fields["model_info_json"]
|
||
|
||
@classmethod
|
||
def get_model_name_by_sign(cls, sign):
|
||
fields = BuildDynamicModelClass.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"sign": sign,
|
||
}
|
||
})
|
||
return fields["model_name"]
|
||
|
||
@classmethod
|
||
def get_serializers(cls, table_name):
|
||
fields = BuildDynamicModelClass.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"model_name": table_name,
|
||
}
|
||
})
|
||
# 结构化数据
|
||
progress_data = BuildDynamicModelClass.model_atr_transform(fields["model_info_json"])
|
||
# 构建model类
|
||
now_model = cls.get_model_by_data(table_name, fields["model_info_json"])
|
||
# 构建序列化类
|
||
serializers_build = BuildSerializers.build_serializers(table_name + "_serializers",
|
||
now_model,
|
||
{'extend': progress_data})
|
||
return serializers_build
|
||
|
||
@classmethod
|
||
def build_model(cls, table_name, fields, module=''):
|
||
# 获取表格数据
|
||
attrs = fields
|
||
|
||
# 放入表名称
|
||
class Meta:
|
||
app_label = 'dependency'
|
||
db_table = table_name
|
||
|
||
# 设置表名以及相关model属性
|
||
attrs['Meta'] = Meta
|
||
attrs['__module__'] = module
|
||
|
||
# 进行类创建
|
||
model_class = type(table_name, (models.Model,), attrs)
|
||
return model_class
|
||
|
||
@classmethod
|
||
def get_zero_code_model_fields(cls):
|
||
cls.ZeroCodeModelInfoData = []
|
||
if not len(cls.ZeroCodeModelInfoData) == 0:
|
||
return cls.ZeroCodeModelInfoData
|
||
else:
|
||
with open('dependency/json_data_in_need/data.json',encoding='utf-8') as data:
|
||
cls.ZeroCodeModelInfoData = json.load(data)
|
||
return cls.ZeroCodeModelInfoData
|
||
|
||
@classmethod
|
||
def get_model_info(cls, data):
|
||
res = cls.get_zero_code_model_fields()
|
||
if len(res) == 0:
|
||
res = cls.get_zero_code_model_fields()
|
||
progress_data = cls.model_atr_transform(dict(res)["table_fields"])
|
||
try:
|
||
now_model = apps.get_model('dependency', dict(res)["table_name"])
|
||
except LookupError:
|
||
now_model = cls.build_model(dict(res)["table_name"], progress_data)
|
||
data['target_model_info'].update({'model_status': 1})
|
||
if not bool(data['return_type'] == 'True'):
|
||
try:
|
||
|
||
target = now_model.objects.get(**json.loads(json.dumps(data['target_model_info'])))
|
||
except now_model.DoesNotExist as e:
|
||
return str('查找的表不存在')
|
||
else:
|
||
target = now_model.objects.filter(**json.loads(json.dumps(data['target_model_info'])))
|
||
# 构建序列化类
|
||
serializers_build = BuildSerializers.build_serializers('zero_code_model', now_model, {})
|
||
# 序列化类使用
|
||
serializers = serializers_build(target, many=bool(data['return_type'] == 'True'))
|
||
return serializers.data
|
||
|
||
@classmethod
|
||
def get_table_relation(cls, table_sign, now_table):
|
||
return_data = {
|
||
"columns": [],
|
||
"serializers_list": {},
|
||
}
|
||
fields = cls.get_model_info({
|
||
"return_type": "False",
|
||
"target_model_info": {
|
||
"model_name": "middle_index_table",
|
||
}})
|
||
# 结构化数据
|
||
progress_data = BuildDynamicModelClass.model_atr_transform(fields["model_info_json"])
|
||
# 构建model类
|
||
try:
|
||
now_model = apps.get_model('dependency', "middle_index_table")
|
||
except LookupError:
|
||
now_model = BuildDynamicModelClass.build_model("middle_index_table", progress_data)
|
||
target = now_model.objects.all()
|
||
# 构建序列化类
|
||
serializers_build = BuildSerializers.build_serializers("middle_index_table" + "_serializers",
|
||
now_model,
|
||
{"extend": progress_data})
|
||
serializers = serializers_build(target, many=True)
|
||
middle_serializer = None
|
||
middle_model = None
|
||
column_data = None
|
||
for item in serializers.data:
|
||
# 获取关联表信息
|
||
if fields["sign"] == item["table_list"][0] and not item["middle_table_json_info"] == {}:
|
||
middle_data = item["middle_table_json_info"]["middle_json_info"]
|
||
middle_model = cls.get_model_by_data(item["middle_table_json_info"]["middle_name"], middle_data)
|
||
progress_data = BuildDynamicModelClass.model_atr_transform(middle_data)
|
||
middle_serializer = BuildSerializers.build_serializers(item["middle_table_json_info"]["middle_name"],
|
||
middle_model, {'extend': progress_data})
|
||
# 是否关联表判定
|
||
if table_sign in item["table_list"] and not item["middle_table_json_info"] == {} and 'empty' not in item[
|
||
"table_list"]:
|
||
# 写入当前表的预设置字段信息
|
||
if middle_model is not None and middle_serializer is not None:
|
||
try:
|
||
target = middle_model.objects.get(middle_index_table_sign=item["sign"])
|
||
column_data = middle_serializer(target).data
|
||
except middle_model.DoesNotExist:
|
||
print(123)
|
||
item = dict(item)
|
||
# 结构化数据
|
||
middle_progress_data = BuildDynamicModelClass.model_atr_transform(
|
||
item["middle_table_json_info"]["middle_json_info"])
|
||
# 构建model类
|
||
try:
|
||
middle_now_model = apps.get_model('dependency', item["middle_table_json_info"]["middle_name"])
|
||
except LookupError:
|
||
middle_now_model = BuildDynamicModelClass.build_model(item["middle_table_json_info"]["middle_name"],
|
||
middle_progress_data)
|
||
# 构建序列化类
|
||
filter_data = {
|
||
now_table + "_sign": item["sign"]
|
||
}
|
||
middle_target = middle_now_model.objects.filter(**filter_data)
|
||
middle_serializers_build = BuildSerializers.build_serializers(
|
||
item["middle_table_json_info"]["middle_name"] + "_serializers",
|
||
middle_now_model,
|
||
{"extend": middle_progress_data})
|
||
return_data["serializers_list"].update({"middle_table_serializers": middle_serializers_build})
|
||
return_data["serializers_list"].update({"middle_now_model": middle_now_model})
|
||
return_data["serializers_list"].update(
|
||
{"fields": [{x['name']: ""} for x in item["middle_table_json_info"]["middle_json_info"] if
|
||
x['name'] != 'sign']})
|
||
if column_data is not None:
|
||
relation_columns_model = cls.get_model("relation_columns").objects.get(
|
||
sign=column_data["relation_columns_sign"])
|
||
columns_infos = cls.get_serializers("relation_columns")(relation_columns_model).data
|
||
frontend_show_model = cls.get_model("frontend_show").objects.get(
|
||
sign=columns_infos["frontend_show"])
|
||
frontend_show_infos = cls.get_serializers("frontend_show")(frontend_show_model).data
|
||
for columns in columns_infos["columns_infos"]:
|
||
for additional in frontend_show_infos["show_items"]:
|
||
if columns["name"] == additional["field_name"]:
|
||
columns["additional"] = additional["additional"]
|
||
return_data.update({'columns': columns_infos["columns_infos"]})
|
||
return return_data
|
||
|
||
|
||
class BuildSerializers:
|
||
@classmethod
|
||
def build_serializers(cls, table_name, now_model, attrs):
|
||
extra_kwargs = {}
|
||
if 'extend' in attrs.keys():
|
||
for item in attrs['extend'].keys():
|
||
extra_kwargs.update({item: {'validators': []}})
|
||
del attrs["extend"]
|
||
attr = attrs
|
||
else:
|
||
attr = attrs
|
||
|
||
# 创建 Meta 类
|
||
meta_class = type('Meta', (), {
|
||
'model': now_model,
|
||
'fields': '__all__',
|
||
'depth': 1,
|
||
"extra_kwargs": extra_kwargs
|
||
})
|
||
|
||
# 将 Meta 类作为属性传递给 type() 创建的类
|
||
attr['Meta'] = meta_class
|
||
|
||
# 数据验证通过元对象所存放的json中进行读取数据，根传入的参数进行比对
|
||
|
||
model_class = type(table_name, (serializers.ModelSerializer,), attr)
|
||
return model_class
|