Python學習day17 任務發佈

時間 2019-12-10

標籤 python 學習 day17 day 任務發佈欄目 Python 简体版

原文原文鏈接

Created on 2017年7月17日html

第1課本節內容 8minutespython

　　任務編排系統開發git

　　架構思路/實現方式介紹github

　　項目實現shell

　　接口認證數據庫

　　擴展：django

　　　　Python的類是什麼json

　　　　模板語言的本質api

第2課任務編排系統架構 46minutes架構

　　發任務讓機器執行

　　gitlab

　　github

　　svn

　　git.oschina.net 碼雲

　　對ManytoMany追加列，能夠在Models字段中添加如下內容：

　　nane = models.ManytoMany(through='table name')

　　一個任務系統的表結構：

　　UserType Userinfo Admin Usergroup

　　HostStatus Host TaskTemplate TaskType

　　ExecuteType Task TaskHoststatus Tasklog

第3課任務編排後臺管理功能介紹一 25minutes

第4課任務編排後臺管理功能介紹二 15minutes

　　任務後臺系統頁面書寫實現

第5課任務編排後臺管理之任務列表 10minutes

第6課任務編排後臺管理之建立任務一 46minutes

from django import forms, templatetags
from pip._vendor.requests.utils import is_valid_cidr
from django.shortcuts import render_to_response
from cgitb import html
from django.template.backends.django import Template
from _codecs import register

#---------froms-----------------------------------------------
class TaskForm(forms.Form):
    name = forms.CharField(max_length=30,
                        error_messages={'required':u'任務名稱不能爲空'},
                        widget = forms.TextInput(attrs={'class':
                        'form-control no radius','placeholder':
                        u'任務名稱'}))
    task_type = forms.IntegerField(error_messages={'required':u'任務類型不能爲空'},
                                widget=forms.widgets.Select(
                               choices=models.TaskType.objects.all().order_by('id').values_list('id','caption'),
                                attrs={'class': 'form-control no radius'}))
    hosts = forms.CharField(error_messages={'required':u'任務類型不能爲空'},
                            widget=forms.widgets.SelectMultiple(
                            choices=models.Hosts.objects.all().order_by('id').values_list('id','hostname'),
                            attrs={'class': 'form-control no radius','multiple':"multiple"}))
    kick_off_time = forms.CharField(max_length=30,
                            error_messages={'required':u'執行時間不能爲空'},
                            widget=forms.DateTimeInput(
                            attrs={'class': 'form-control no radius',
                                   'placeholder':u'執行時間','id':'kick_off_at'
                                    }))
    def __init__(self):
        #每執行一次都從數據庫中更新，不然每次都是原先數據，新數據不顯示
        #靜態字段執行後會寫入內存
        self.fields['hosts'].widget.choices = models.Hosts.objects.all().order_by('id').values_list('id','hostname')
 
#-------------views------------------------------------------------

def add_task(request):
    pass
def create(request):
    form_obj = TaskForm()
    if request.method == 'POST':
        form_obj = TaskForm(request.POST)
        if form_obj.is_valid:
            print form_obj.clean()
            raw_data = form_obj.clean()
            add_task(raw_data)  #將Form的內容添加到數據庫中
        else:
            #顯示錯誤信息
            print form_obj.errors.as_data()  
            return render_to_response('index.html',
                                      {'model',form_obj,
                                       'message':form_obj.errors.as_data()})

#-------------- 自定義錯誤模板----------------------------------------

文件目錄在APP下，模塊名：templatetags,文件名可定義

#form_tag:

form django import template
register = template.Library()

@register.simple_tag
def error_message(arg):
    if arg:
        return arg[0][0]
    else:
        return ''

上面模塊可顯示第一個參數

html使用 {% load  form_tag %}
而後能夠使用：{% error_message 變量.字段1 %}

第7課任務編排後臺管理之建立任務二 23 minutes

第8課任務編排後臺管理之建立任務三 7 minutes

from django.db import transaction
def add_task(request):
    try:
        with transaction.atomic():
            hosts = data['hosts']
            del data['hosts']
            del data['task_template']
            data['task_type'] = models.TaskType.objects.get(id=data['task_type'])
            data['execute_type'] = models.ExecuteType.objects.get(id=data['execute_type'])
            task_obj = models.Task.objects.create(**data)
　　　　　　　hosts = hosts.replace("'",'"')
            hosts = hosts.replace('u',' ')
            '''
            hosts = '[u"1",u"2",u"3"]'
            Json不能load帶U的字符
             同時Loads時注意表裏面必須是雙引號，外面是單引號，不然出錯
               >>> a
                '[u"1",u"2",u"3"]'
                >>> a = a.replace("u",' ')
                >>> json.loads(a)
                ['1', '2', '3']              
                >>> a =  '[u"1",u"2",u"3"]'
                >>> json.loads(a)
                Traceback (most recent call last):
                  File "<stdin>", line 1, in <module>
                  File "C:\Python27\lib\json\__init__.py", line 310, in loads
                    return _default_decoder.decode(s)
                  File "C:\Python27\lib\json\decoder.py", line 346, in decode
                    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
                  File "C:\Python27\lib\json\decoder.py", line 364, in raw_decode
                    raise ValueError("No JSON object could be decoded")
                ValueError: No JSON object could be decoded

>>>         
            '''
            #這裏去U改爲雙引號後才能Loads
            host_list = models.Host.objects.filter(id__in=json.loads(hosts))
            for item in host_list:
                models.TaskHostStatus.objects.create(status=0,task=task_obj,host=item)

第9課任務編排Agent實現分析 27 minutes

#--------------------------------------------------------------------------
Agent

import json
import uuid

from lib.plugins import PluginApi
from lib.commons import log
from log.core import securty
import config
import commands

class Program:
    def __init__(self):
        self.host = config.configuations['host']
        self.port = config.configuations['port']
        self.resource = config.configuations['resource']
        self.timeout = config.configuations['timeout']

    def process(self):
        data = self.get_task()
        #這裏能夠判斷，命令是腳本仍是命令，若是是命令直接執行，不需寫入文件
        #若是是腳本，寫入文件再執行
        file_name = self.write_file(data)
        retsult = self.execute(file_name)

    def get_task(self):
        params = urllib.urlencode({'data':json.dumps(
                                {'hostname':c1.salt.com })})
        result = self.url_request(params,'GET')
        return result

    def wirte_file(self):
        file_name = str(uuid.uuid())+.'py'
        f = file(file_name,'w')
        f.write(data)
        f.close()
        return file_name

    def execute(self,file_name):
        shell_command = 'python %s'%(file_name)
        status,output = commands.getstatusoutput(shell_command)
        pirnt  'output:==========>'
        print output

    def url_request(self,params,methos):
        original = None
        headers = {'Content-type':
        "application/x-www/from-urlencoded","Accept":
        "text/json","SecurtyKey":securty.create_ai_key()}
        try:
            conn = httplib.HTTPConnection(self.host,self.port,self.timeout)
            conn.request(metod,self.resource,params,headers)
            response = conn.getresponse()
            original = response.read()
        except Exception,e:
            log.write_error_log('[htp],%s' %e)
        return original

第10課任務編排之API驗證 38 minutes

client:

發送 md5(key+datetime)|datetime

server:

1.接收請求，分割字符串(加密碼，客戶端時間）

2.若是當前時間－客戶端時間>5s,請求失效

3.md5(客戶端時間+key),生成加密碼

4.比對加密碼是否一致

#--------------------------------------------------------------

#securty -----生成加密串

import time
import config 
import hashlib


def create_api_key():
    hash_obj = hashlib.md5()
    key = config.configration['key']
    time_span = time.time()
    hash_obj.updata("%s|%f"%(key,time_span))
    encryption = hash_obj.hexdigest()
    result = '%s|%f'%(encryption,time_span)  
    return result    

#----------------views----------------------------------------------------     

from django.shortcuts import HttpResponse

def api_auth(func):
    def wrapper(request):
        securty_key = request.META.get('HTTP_SECURTYKEY',None)
        if not securty_key:
            return HttpResponse('認證失敗')
        if not auth_api_valid(securty_key):
            return HttpResponse('認證失敗')
        return func(request)     
    return wrapper


#對接收的數據進行MD5匹配
def auth_api_valid(data):
    try:
        encryption ,time_span = data.split('|')
        time_span = float(time_span)
        if (time.time()-time_span)>5:
            return False
        hash_obj = hashlib.md5()
        hash_obj.update("%s|%f"%(key,time_span))
        if hash_obj.hexdigest() = encryption:
            return True
        else:
            return False
    except Exception,e:
        pass
    return False

'''
這裏能夠將Key,5等放到一個配置文件裏，再直接調用，可方便更改管理

'''
@api_auth #使用裝飾器來檢查是否有KEY
def handle_server_info(request):
    ret = {'status':0,'message':''}
    return HttpResponse(json.dumps(ret))

第11課擴展之類是什麼（上） 34 minutes

第12課擴展之類是什麼（下） 2 minutes

python中一切都是對象，類自己也是對象，類是由type產生的。
class Foo
　　pass

如下兩種方式都是同樣的結果
1. Bar = type('Bar',(object,),{'name':123,'Func':Hello})
2. class Bar:
    name = 123

def Hello(self):
    print 'hello'

既然這樣，那麼對於定義的類來講，只要定義了一個類，就調用一次 type類的構造函數，如何驗證？
__metaclass__  能夠指定類是由那個type來產生的
class MyClass(type):
    def __init__(self,name,bases,dicts):
        print name
    def __call__(self, *args, **kwargs):
        return type.__call__(self, *args, **kwargs)
        
class C1:  #是MyClass的實例
    __metaclass__ = MyClass

class C2:  #是MyClass的實例
    __metaclass__ = MyClass

c1 = C1()  #執行Call方法

class Test(object):
    def __init__(self):
        print 'init'
    def __call__(self):
        print 'call'

t1 = Test() #執行Init
t1()  #執行Call

第13課擴展之模板語言的本質 23 minutes

模板語言就是經過如下的方式來實現的

#!usr/bin/env python

#coding:utf-8

下面的Name至關於後臺發過來的字典變量
namespace = {'name':'wupeiqi','data':[18,73,84]}
code =  '''def hellocute():return  "name %s ,age %d" %(name,data[0],) '''
#經過如下兩個命令執行上面的字符串函數
func = compile(code, '<string>', "exec")
exec func in namespace 
result = namespace['hellocute']() 
print result

第14課總結 15 minutes