async def hierarchical(data): # Create a dictionary to hold the output data output = {} rows_data = data.get('rows') rows_data = sorted(rows_data, key=lambda x: x['create_at'], reverse=False) # Process the input data for item in rows_data: parentid = item['parentid'] k = item['k'] v = item['v'] # Create a new dictionary for the item new_item = {'parentid': parentid, 'k': k, 'v': v, 'id': item['id'], 'codeid': item['codeid']} # Check if the item has a parent if not parentid: # If it doesn't have a parent, add it to the output dictionary output[k] = new_item else: # If it has a parent, add it to the parent's 'son' list parent = '' if '-' in parentid or len(parent) > 1: data_id = parentid.split('-') parent = output for index, value in enumerate(data_id): if index == len(data_id) - 1: parent = parent.get(value) else: parent = parent.get(value).get('son') if not parent: parent = output[parentid] if 'son' not in parent: parent['son'] = {} parent['son'][k] = new_item output = await convert_dict_to_list(output) return output async def convert_dict_to_list(data): result = [] for key, value in data.items(): if isinstance(value, dict): value = [value] if isinstance(value, list): for item in value: result.append({k: v for k, v in item.items() if k != "son"}) if "son" in item: result[-1]["son"] = await convert_dict_to_list(item["son"]) else: result.append({key: value}) return result async def appCodesSearch(ns={}): """ search new appcodes `id` VARCHAR(32) 'id', `name` VARCHAR(255) '编码名称', `hierarchy_flg` VARCHAR(1) '多级标志', `del_flg` VARCHAR(1) DEFAULT '0' comment '删除标志', `create_at` TIMESTAMP comment '创建时间戳' :param ns: :return: """ db = DBPools() async with db.sqlorContext('kboss') as sor: try: ns['del_flg'] = '0' ns['sort'] = 'create_at' ns['order'] = 'desc' ns['page'] = ns.get('page') if ns.get('page') else 1 if ns.get('kv'): if not ns.get('codeid'): return { "status": False, "msg": "appCodes_kv search failed, the codeid is empty", "data": "" } app_code_result = await sor.R('appcodes_kv', ns) app_code_result = await hierarchical(app_code_result) else: app_code_result = await sor.R('appcodes',ns) return { "status": True, "msg": "appCodes search success", "data": app_code_result } except Exception as e: return { "status": False, "msg": "appCodes search failed", "data": "" } ret = await appCodesSearch(params_kw) return ret