mirror of
https://github.com/jumpserver/jumpserver.git
synced 2026-05-14 19:03:31 +00:00
1eb8e40d3e
* feat: 账号推送附加参数 * perf: 通过节点 资产 过滤平台api * perf: push automation params * perf: 修改playbook * perf: params serializer * perf: 账号推送playbook 调整 * perf: Automation serializer add params field * perf: params 非必填 * perf: 添加is_params 给前端判断 * perf: is_params bool * perf: 修改push account ansible逻辑 * perf: 修改获取push_kwargs方法 * perf: platform migrate * perf: 修改api * perf: 单个推送 * perf: push account * perf: 修改asset auto_config --------- Co-authored-by: feng <1304903146@qq.com> Co-authored-by: feng626 <57284900+feng626@users.noreply.github.com>
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
import json
|
|
import os
|
|
from functools import partial
|
|
|
|
import yaml
|
|
|
|
|
|
def check_platform_method(manifest, manifest_path):
|
|
required_keys = ['category', 'method', 'name', 'id', 'type']
|
|
less_key = set(required_keys) - set(manifest.keys())
|
|
if less_key:
|
|
raise ValueError("Manifest missing keys: {}, {}".format(less_key, manifest_path))
|
|
if not isinstance(manifest['type'], list):
|
|
raise ValueError("Manifest type must be a list: {}".format(manifest_path))
|
|
return True
|
|
|
|
|
|
def check_platform_methods(methods):
|
|
ids = [m['id'] for m in methods]
|
|
for i, _id in enumerate(ids):
|
|
if _id in ids[i + 1:]:
|
|
raise ValueError("Duplicate id: {}".format(_id))
|
|
|
|
|
|
def generate_serializer(data):
|
|
from common.serializers import create_serializer_class
|
|
params = data.pop('params', None)
|
|
if not params:
|
|
return None
|
|
serializer_name = data['id'].title().replace('_', '') + 'Serializer'
|
|
return create_serializer_class(serializer_name, params)
|
|
|
|
|
|
def get_platform_automation_methods(path):
|
|
methods = []
|
|
for root, dirs, files in os.walk(path, topdown=False):
|
|
for name in files:
|
|
path = os.path.join(root, name)
|
|
if not path.endswith('manifest.yml'):
|
|
continue
|
|
|
|
with open(path, 'r') as f:
|
|
manifest = yaml.safe_load(f)
|
|
check_platform_method(manifest, path)
|
|
manifest['dir'] = os.path.dirname(path)
|
|
manifest['params_serializer'] = generate_serializer(manifest)
|
|
methods.append(manifest)
|
|
|
|
check_platform_methods(methods)
|
|
return methods
|
|
|
|
|
|
def filter_key(manifest, attr, value):
|
|
manifest_value = manifest.get(attr, '')
|
|
if isinstance(manifest_value, str):
|
|
manifest_value = [manifest_value]
|
|
return value in manifest_value or 'all' in manifest_value
|
|
|
|
|
|
def filter_platform_methods(category, tp_name, method=None, methods=None):
|
|
methods = platform_automation_methods if methods is None else methods
|
|
if category:
|
|
methods = filter(partial(filter_key, attr='category', value=category), methods)
|
|
if tp_name:
|
|
methods = filter(partial(filter_key, attr='type', value=tp_name), methods)
|
|
if method:
|
|
methods = filter(lambda x: x['method'] == method, methods)
|
|
return methods
|
|
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
platform_automation_methods = get_platform_automation_methods(BASE_DIR)
|
|
|
|
if __name__ == '__main__':
|
|
print(json.dumps(platform_automation_methods, indent=4))
|