Method test_script

class/crontabModel/scriptModel.py:368–405

@name 测试运行脚本 @author hwliang @param args { script_id 脚本ID } @return dict

(self,args)

        return page

    def test_script(self,args):
        '''
            @name 测试运行脚本
            @author hwliang
            @param args<dict_obj>{
                script_id<int> 脚本ID
            }
            @return dict
        ''&#x27;
        script_start_time = int(time.time())
        script_id = args.get('script_id',0)
        if not script_id: return public.returnMsg(False,'脚本ID不能为空')
        script_info = self._sql.table('scripts').where('script_id=?',(script_id,)).field('script,script_type,return_type').find()
        script_args = args.get('args/s','')
        if script_args: script_args = ' ' + script_args
        if 'is_args' in script_info and script_info['is_args'] and not script_args:
            return public.returnMsg(False,'该脚本需要参数')
        if not script_info: return public.returnMsg(False,'指定脚本不存在')
        tmp_file= '{}/tmp/{}.tmp'.format(public.get_panel_path(),public.md5(script_info['script']))
        try:
            public.writeFile(tmp_file,script_info['script'])
            if script_info['script_type'] == 'bash':
                result = public.ExecShell("bash {}{}".format(tmp_file,script_args))
            elif script_info['script_type'] == 'python':
                python_bin = public.get_python_bin()
                result = public.ExecShell('{} -u {}{}'.format(python_bin,tmp_file,script_args))
            res = result[0].strip()
            if result[1]:
                self.add_task_log(script_id,0,0,0,result[0],result[1],script_start_time,int(time.time()))
                return public.returnMsg(False,'脚本运行错误，请检查脚本代码是否有误: \n{}'.format(result[0] + "\n" +  result[1].split('.tmp:')[-1]))
        except Exception as ex:
            error_msg = public.get_error_info()
            self.add_task_log(script_id,0,0,0,'',error_msg,script_start_time,int(time.time()))
            return public.returnMsg(False,'执行失败,错误信息:\n{}'.format(error_msg))
        finally:
            if os.path.exists(tmp_file): os.remove(tmp_file)
        self.add_task_log(script_id,0,0,1,result[0],result[1],script_start_time,int(time.time()))
        return public.returnMsg(True,"执行成功: \n{}".format(public.xssencode2(res)))