51208f7295
- construct_get_data_sql: 两个路径都支持c.dbname前缀 - get_code_desc: params中dbname优先用c.dbname - models/*.json codes中添加dbname字段即可跨库引用
460 lines
13 KiB
Python
460 lines
13 KiB
Python
import os
|
||
import sys
|
||
import codecs
|
||
import json
|
||
import argparse
|
||
|
||
from appPublic.dictObject import DictObject
|
||
from appPublic.folderUtils import listFile, _mkdir
|
||
from appPublic.myTE import MyTemplateEngine
|
||
from appPublic.argsConvert import ArgsConvert
|
||
from xls2ddl.xlsxData import xlsxFactory
|
||
from xls2ddl.tmpls import (
|
||
data_browser_tmpl,
|
||
get_data_tmpl,
|
||
data_new_tmpl,
|
||
data_update_tmpl,
|
||
data_delete_tmpl,
|
||
check_changed_tmpls
|
||
)
|
||
|
||
"""
|
||
usage:
|
||
xls2crud.py dbname models_dir uidir
|
||
"""
|
||
|
||
def build_dbdesc(models_dir: list) -> dict:
|
||
print(f'{models_dir=}')
|
||
mdirs = []
|
||
if isinstance(models_dir, list):
|
||
mdirs = models_dir
|
||
else:
|
||
mdirs = [models_dir]
|
||
db_desc = {}
|
||
for models_dir in mdirs:
|
||
for f in listFile(models_dir, suffixs=['.xlsx']):
|
||
print(f'{f} handle ...')
|
||
x = xlsxFactory(f)
|
||
d = x.get_data()
|
||
tbname = d.summary[0].name
|
||
db_desc.update({tbname:d})
|
||
for f in listFile(models_dir, suffixs=['.json']):
|
||
print(f'{f} handle ...')
|
||
with codecs.open(f, 'r', 'utf-8') as fh:
|
||
d = json.load(fh)
|
||
tbname = d['summary'][0]['name']
|
||
db_desc.update({tbname:d})
|
||
return db_desc
|
||
|
||
def subtable2toolbar(desc):
|
||
if desc.subtables is None:
|
||
return
|
||
if desc.toolbar is None:
|
||
desc.toolbar = DictObject(**{
|
||
"tools":[]
|
||
})
|
||
tools = desc.toolbar.tools or []
|
||
binds = desc.binds or []
|
||
for st in desc.subtables:
|
||
tools.append({
|
||
"selected_row": True,
|
||
"name": st.subtable,
|
||
"icon":"{{entire_url('/imgs/" + st.subtable + ".svg')}}",
|
||
"label": st.title
|
||
})
|
||
bind = {
|
||
"wid":"self",
|
||
"event":st.subtable,
|
||
"actiontype": "urlwidget",
|
||
"target":"PopupWindow",
|
||
"popup_options":{
|
||
"title":st.title or st.subtable,
|
||
"icon": "{{entire_url('/appbase/get_icon.dspy')}}?id="+st.subtable,
|
||
"resizable": True,
|
||
"height":"70%",
|
||
"width":"70%"
|
||
},
|
||
"params_mapping": {
|
||
"mapping":{
|
||
"id": st.field,
|
||
"referer_widget":"referer_widget"
|
||
},
|
||
"need_other":False
|
||
},
|
||
"options":{
|
||
"method":"POST",
|
||
"params":{},
|
||
"url":st.url or "{{entire_url('../" + st.subtable + "')}}"
|
||
}
|
||
}
|
||
if st.mapping:
|
||
bind['params_mapping']['mapping'].update(st.mapping)
|
||
binds.append(bind);
|
||
|
||
desc.toolbar.tools = tools
|
||
desc.binds = binds
|
||
|
||
def build_crud_ui(crud_data: dict, dbdesc: dict):
|
||
uidir = crud_data.output_dir
|
||
tables = [ k for k in dbdesc.keys() ]
|
||
desc = DictObject(**dbdesc[crud_data.tblname].copy())
|
||
desc.update(crud_data.params.copy())
|
||
subtable2toolbar(desc)
|
||
binds = desc.binds or []
|
||
if desc.relation:
|
||
desc.checkField = 'has_' + desc.relation.param_field
|
||
binds.append({
|
||
"wid":"self",
|
||
"event":"row_check_changed",
|
||
"actiontype":"urlwidget",
|
||
"target":"self",
|
||
"options":{
|
||
"params":{},
|
||
"url":"{{entire_url('check_changed.dspy')}}"
|
||
}
|
||
})
|
||
desc.bindsstr = json.dumps(binds, indent=4, ensure_ascii=False)
|
||
desc.update({
|
||
"tblname":crud_data.tblname,
|
||
"dbname":crud_data.dbname
|
||
})
|
||
build_table_crud_ui(uidir, desc)
|
||
|
||
def build_table_crud_ui(uidir: str, desc: dict) -> None:
|
||
_mkdir(uidir)
|
||
build_data_browser(uidir, desc)
|
||
if desc.relation:
|
||
build_check_changed(uidir, desc)
|
||
else:
|
||
build_data_new(uidir, desc)
|
||
build_data_update(uidir, desc)
|
||
build_data_delete(uidir, desc)
|
||
build_get_data(uidir, desc)
|
||
|
||
def alter_field(field:dict, desc:DictObject) -> dict:
|
||
name = field['name']
|
||
ret = field.copy()
|
||
alters = desc.browserfields.alters
|
||
if alters:
|
||
[ ret.update(alters[k]) for k in alters.keys() if k == name ]
|
||
return ret
|
||
|
||
def field_list(desc: dict) -> list:
|
||
fs = []
|
||
for f in desc.fields:
|
||
if desc.codes and f.name in [c.field for c in desc.codes]:
|
||
d = get_code_desc(f, desc)
|
||
else:
|
||
d = setup_ui_info(f, confidential_fields=desc.confidential_fields or [])
|
||
"""
|
||
use alters to modify fields
|
||
"""
|
||
d = alter_field(d, desc)
|
||
fs.append(d)
|
||
return fs
|
||
|
||
def get_code_desc(field: dict, desc: dict) -> dict:
|
||
d = DictObject(**field.copy())
|
||
if not desc.codes:
|
||
return None
|
||
for c in desc.codes:
|
||
if d.name == c.field:
|
||
d.label = d.title or d.name
|
||
d.uitype = 'code'
|
||
d.valueField = d.name
|
||
d.textField = d.name + '_text'
|
||
dbname_val = "{{get_module_dbname('" + getattr(c, 'dbname', desc.modulename) + "')}}"
|
||
d.params = {
|
||
'dbname':dbname_val,
|
||
'table':c.table,
|
||
'tblvalue':c.valuefield,
|
||
'tbltext':c.textfield,
|
||
'valueField':d.valueField,
|
||
'textField':d.textField
|
||
}
|
||
if c.cond:
|
||
d.params['cond'] = c.cond
|
||
ac = ArgsConvert('[[', ']]')
|
||
vars = ac.findAllVariables(c.cond)
|
||
for v in vars:
|
||
d.params[v] = '{{params_kw.' + v + '}}'
|
||
d.dataurl = "{{entire_url('/appbase/get_code.dspy')}}"
|
||
return d
|
||
return None
|
||
|
||
def setup_ui_info(field:dict, confidential_fields=[]) ->dict:
|
||
d = DictObject(**field.copy())
|
||
if d.length:
|
||
length = int(d.length)
|
||
d.cwidth = length if length < 18 else 18
|
||
if d.cwidth < 4:
|
||
d.cwidth = 4;
|
||
else:
|
||
d.length = 0
|
||
|
||
if d.type == 'date':
|
||
d.uitype = 'date'
|
||
d.length = 0
|
||
elif d.type == 'time':
|
||
d.uitype = 'time'
|
||
d.length = 0
|
||
elif d.type in ['int', 'short', 'long', 'longlong']:
|
||
d.uitype = 'int'
|
||
d.length = 0
|
||
elif d.type == 'text':
|
||
d.uitype = 'text'
|
||
elif d.type in ['float', 'double', 'decimal']:
|
||
d.uitype = 'float'
|
||
else:
|
||
if d.name in confidential_fields:
|
||
d.uitype = 'password'
|
||
elif d.name.endswith('_date') or d.name.endswith('_dat'):
|
||
d.uitype = 'date'
|
||
d.length = 0
|
||
else:
|
||
d.uitype = 'str'
|
||
d.datatype = d.type
|
||
d.label = d.title or d.name
|
||
return d
|
||
|
||
def construct_get_data_sql(desc: dict) -> str:
|
||
shortnames = [c for c in 'bcdefghjklmnopqrstuvwxyz']
|
||
infos = []
|
||
if desc.relation and desc.codes:
|
||
param_field = "${" + desc.relation.param_field + "}$"
|
||
for code in desc.codes:
|
||
if code.field == desc.relation.outter_field:
|
||
code_tbl = f"{code.dbname}.{code.table}" if getattr(code, 'dbname', None) else code.table
|
||
return f"""select '$[{desc.relation.param_field}]$' as {desc.relation.param_field},
|
||
case when b.{desc.relation.param_field} is NULL then 0 else 1 end has_{desc.relation.param_field},
|
||
a.{code.valuefield} as {code.field},
|
||
a.{code.textfield} as {code.field}_text
|
||
from {code_tbl} a left join
|
||
(select * from {desc.tblsql or desc.tblname} where {desc.relation.param_field} ={param_field}) b
|
||
on a.{code.valuefield} = b.{code.field}
|
||
"""
|
||
if not desc.codes or len(desc.codes) == 0:
|
||
return f"select * from {desc.tblsql or desc.tblname} where 1=1 " + ' [[filterstr]]'
|
||
|
||
for i, c in enumerate(desc.codes):
|
||
shortname = shortnames[i]
|
||
cond = '1 = 1'
|
||
if c.cond:
|
||
cond = c.cond
|
||
code_tbl = f"{c.dbname}.{c.table}" if getattr(c, 'dbname', None) else c.table
|
||
csql = f"""(select {c.valuefield} as {c.field},
|
||
{c.textfield} as {c.field}_text from {code_tbl} where {cond})"""
|
||
infos.append([f'{shortname}.{c.field}_text', f"{csql} {shortname} on a.{c.field} = {shortname}.{c.field}"])
|
||
bt = f'(select * from {desc.summary[0].name} where 1=1' + " [[filterstr]]) a"
|
||
infos.insert(0, ['a.*', bt])
|
||
fields = ', '.join([i[0] for i in infos])
|
||
tables = ' left join '.join([i[1] for i in infos])
|
||
return f"""select {fields}
|
||
from {tables}"""
|
||
|
||
def filter_backslash(s):
|
||
if s is None:
|
||
return s
|
||
ls = s.split('\\/')
|
||
return '/'.join(ls)
|
||
|
||
def extract_filter_vars(filterjson, field_map):
|
||
"""递归提取 data_filter 中的 var 和对应的 field，返回 {var: field_info}"""
|
||
if not filterjson:
|
||
return {}
|
||
|
||
result = {}
|
||
|
||
if isinstance(filterjson, dict):
|
||
# 逻辑操作符：AND, OR, NOT
|
||
for key in ['AND', 'OR', 'NOT', 'and', 'or', 'not']:
|
||
if key in filterjson:
|
||
sub = filterjson[key]
|
||
if isinstance(sub, list):
|
||
for item in sub:
|
||
result.update(extract_filter_vars(item, field_map))
|
||
elif isinstance(sub, dict):
|
||
result.update(extract_filter_vars(sub, field_map))
|
||
|
||
# 单个条件：field + op + var
|
||
if 'field' in filterjson and 'var' in filterjson:
|
||
var_name = filterjson['var']
|
||
field_name = filterjson['field']
|
||
if field_name in field_map:
|
||
result[var_name] = field_map[field_name].copy()
|
||
|
||
elif isinstance(filterjson, list):
|
||
for item in filterjson:
|
||
result.update(extract_filter_vars(item, field_map))
|
||
|
||
return result
|
||
|
||
def build_filter_field_list(desc) -> list:
|
||
"""从 data_filter 提取所有 var，根据 field 从 fields/alters 获取字段定义，
|
||
构造 InlineForm 搜索字段。field name 用 var。"""
|
||
if not desc.data_filter:
|
||
return []
|
||
|
||
# 构建 field name -> field info 的映射
|
||
field_map = {}
|
||
for f in field_list(desc):
|
||
field_map[f['name']] = f
|
||
|
||
# 从 data_filter 提取 var -> field info
|
||
var_field_map = extract_filter_vars(desc.data_filter, field_map)
|
||
|
||
codes_map = {}
|
||
if desc.codes:
|
||
for c in desc.codes:
|
||
cfield = c.field if isinstance(c, dict) else getattr(c, 'field', None)
|
||
if cfield:
|
||
codes_map[cfield] = c
|
||
|
||
modulename = getattr(desc, 'modulename', '')
|
||
result = []
|
||
|
||
for var_name, field_info in var_field_map.items():
|
||
field_name = field_info['name']
|
||
field_title = field_info.get('title', field_name)
|
||
field_uitype = field_info.get('uitype', 'str')
|
||
|
||
field_def = {
|
||
'name': var_name,
|
||
'uitype': field_uitype,
|
||
'placeholder': field_title,
|
||
'cwidth': 15
|
||
}
|
||
|
||
if field_uitype == 'code' and field_name in codes_map:
|
||
c = codes_map[field_name]
|
||
ctable = c.table if isinstance(c, dict) else getattr(c, 'table', '')
|
||
cvaluefield = c.valuefield if isinstance(c, dict) else getattr(c, 'valuefield', '')
|
||
ctextfield = c.textfield if isinstance(c, dict) else getattr(c, 'textfield', '')
|
||
ccond = (c.cond if isinstance(c, dict) else getattr(c, 'cond', None))
|
||
field_def['valueField'] = var_name
|
||
field_def['textField'] = var_name + '_text'
|
||
field_def['params'] = {
|
||
'dbname': "{{get_module_dbname('" + modulename + "')}}",
|
||
'table': ctable,
|
||
'tblvalue': cvaluefield,
|
||
'tbltext': ctextfield,
|
||
'valueField': var_name,
|
||
'textField': var_name + '_text'
|
||
}
|
||
if ccond:
|
||
field_def['params']['cond'] = ccond
|
||
field_def['dataurl'] = "{{entire_url('/appbase/get_code.dspy')}}?prepend_all=1"
|
||
|
||
result.append(field_def)
|
||
|
||
return result
|
||
|
||
def build_data_browser(pat: str, desc: dict):
|
||
desc = desc.copy()
|
||
desc.fieldliststr = json.dumps(field_list(desc), ensure_ascii=False, indent=4)
|
||
desc.filter_fields = build_filter_field_list(desc)
|
||
desc.filterfieldstr = json.dumps(desc.filter_fields, ensure_ascii=False, indent=4)
|
||
# 有 data_filter 时自动注入搜索按钮
|
||
if desc.data_filter:
|
||
toolbar = desc.get('toolbar', None)
|
||
if not toolbar:
|
||
toolbar = {'tools': []}
|
||
desc.toolbar = toolbar
|
||
tools = toolbar.get('tools', [])
|
||
# 确保 filter 按钮在最前面且不重复
|
||
tools = [t for t in tools if t.get('name') != 'filter']
|
||
tools.insert(0, {'name': 'filter', 'label': '搜索'})
|
||
toolbar['tools'] = tools
|
||
e = MyTemplateEngine([])
|
||
s = e.renders(data_browser_tmpl, desc)
|
||
with codecs.open(os.path.join(pat, f'index.ui'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
def extract_validation_rules(desc) -> str:
|
||
"""Extract validation rules from field alters. Returns JSON string or empty."""
|
||
rules = {}
|
||
alters = desc.get('browserfields', {}).get('alters', {})
|
||
if not alters:
|
||
return ''
|
||
for fname, alter in alters.items():
|
||
if 'rules' in alter and alter['rules']:
|
||
rules[fname] = alter['rules']
|
||
if not rules:
|
||
return ''
|
||
return json.dumps(rules, ensure_ascii=False)
|
||
|
||
def build_data_new(pat: str, desc: dict):
|
||
e = MyTemplateEngine([])
|
||
desc = desc.copy()
|
||
desc['validation_rules'] = extract_validation_rules(desc)
|
||
s = e.renders(data_new_tmpl, desc)
|
||
with codecs.open(os.path.join(pat, f'add_{desc.tblname}.dspy'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
def build_data_update(pat: str, desc: dict):
|
||
e = MyTemplateEngine([])
|
||
desc = desc.copy()
|
||
desc['validation_rules'] = extract_validation_rules(desc)
|
||
s = e.renders(data_update_tmpl, desc)
|
||
with codecs.open(os.path.join(pat, f'update_{desc.tblname}.dspy'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
def build_data_delete(pat: str, desc: dict):
|
||
e = MyTemplateEngine([])
|
||
desc = desc.copy()
|
||
s = e.renders(data_delete_tmpl, desc)
|
||
with codecs.open(os.path.join(pat, f'delete_{desc.tblname}.dspy'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
def build_get_data(pat: str, desc: dict):
|
||
e = MyTemplateEngine([])
|
||
desc = desc.copy()
|
||
desc.sql = construct_get_data_sql(desc)
|
||
s = e.renders(get_data_tmpl, desc)
|
||
with codecs.open(os.path.join(pat, f'get_{desc.tblname}.dspy'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
def build_check_changed(pat:str, desc:dict):
|
||
e = MyTemplateEngine([])
|
||
desc = desc.copy()
|
||
s = e.renders(check_changed_tmpls, desc)
|
||
with codecs.open(os.path.join(pat, 'check_changed.dspy'), 'w', "utf-8") as f:
|
||
f.write(filter_backslash(s))
|
||
|
||
if __name__ == '__main__':
|
||
"""
|
||
crud_json has following format
|
||
{
|
||
"tblname",
|
||
"params"
|
||
}
|
||
"""
|
||
parser = argparse.ArgumentParser('xls2crud')
|
||
parser.add_argument('-m', '--models_dir')
|
||
parser.add_argument('-o', '--output_dir')
|
||
parser.add_argument('modulename')
|
||
parser.add_argument('files', nargs='*')
|
||
args = parser.parse_args()
|
||
if len(args.files) < 1:
|
||
print(f'Usage:\n{sys.argv[0]} [-m models_dir] [-o output_dir] json_file ....\n')
|
||
sys.exit(1)
|
||
ns = {k:v for k, v in os.environ.items()}
|
||
for fn in args.files:
|
||
print(f'handle {fn}')
|
||
crud_data = {}
|
||
with codecs.open(fn, 'r', 'utf-8') as f:
|
||
a = json.load(f)
|
||
ac = ArgsConvert('${','}$')
|
||
a = ac.convert(a,ns)
|
||
crud_data = DictObject(**a)
|
||
if args.models_dir:
|
||
crud_data.models_dir = args.models_dir
|
||
models_dir = crud_data.models_dir
|
||
if args.output_dir:
|
||
tblname = crud_data.alias or crud_data.tblname
|
||
crud_data.output_dir = os.path.join(args.output_dir, tblname)
|
||
crud_data.params.modulename = args.modulename
|
||
dbdesc = build_dbdesc(models_dir)
|
||
build_crud_ui(crud_data, dbdesc)
|
||
|