ececuteBafterA

# -*- coding: UTF-8 -*-
import sys
import os
sys.path.append('/usr/local/bin')
import json
import requests
import logging
import time
import datetime
from impala.dbapi import connect
import pymysql
from get_database_info import get_database_info
import logging as logger
from functools import reduce
import re


proxy = {
    'http': '代理',
    'https': '代理'
}

class AzkabanHelper:

    def __init__(self, url, username, password):
        self.url = url
        self.username = username
        self.password = password
        self.login()

    def login(self):
        '''
        登录azkaban获取sessionID
        :return: sessionID
        '''
        params = {
            'action': 'login',
            'username': self.username,
            'password': self.password
        }
        r = requests.post(url=self.url, data=params, verify=False)
        if r.ok:
            res = json.loads(r.text)
            self.sessionId = res['session.id']
            logging.info("登录Azkaban成功，获取sessionID:{}".format(self.sessionId))
        else:
            logging.error("Azkaban登录异常！")
            exit(1)

    def execFlow(self, project, flow, **kwargs):
        '''
        执行Azkaban Flow
        :param project: 工程名
        :param flow: flow名
        :param sessionId: 登录的session
        :param kwargs: 传参
        :return:
        '''

        url = '%s/executor?ajax=executeFlow&session.id=%s&project=%s&flow=%s' % (
            self.url, self.sessionId, project, flow)

        for key, value in kwargs.items():
            url = url + '&flowOverride[%s]=%s' % (key, value)

        r = requests.get(url, verify=False)
        if r.ok:
            res = json.loads(r.text)
            if "error" in res:
                logging.error(res["error"])
            else:
                logging.info(
                    "调用 Azkaban任务： project:{project} flow:{flow} execid:{execid} message:{message}".format(**res))
                return res
        else:
            logging.error("调用Azkaban任务失败！")

    def fetchExecutions(self, project, flow, length=10):
        '''
        获取指定Project和flow的执行状态
        :param execid:
        :param sessionId:
        :return:
        '''
        url = "%s/manager?ajax=fetchFlowExecutions&session.id=%s&project=%s&flow=%s&start=0&length=%s" % (
            self.url, self.sessionId, project, flow, length)
        r = requests.get(url, verify=False)
        if r.ok:
            res = json.loads(r.text)
            if "error" in res:
                logger.error(res["error"])
            else:
                return res
        else:
            logger.error("获取任务信息错误！")

    def fetchExecutionsInfo(self, exec_id):
        '''
        获取指定执行id的执行状态
        :param execid:
        :param sessionId:
        :return:
        '''
        url = "%s/executor?ajax=fetchexecflow&session.id=%s&execid=%s" % (self.url, self.sessionId, exec_id)
        r = requests.get(url, verify=False)
        if r.ok:
            res = json.loads(r.text)
            if "error" in res:
                logger.error(res["error"])
            else:
                return res
        else:
            logger.error("获取任务信息错误！")

    def executeBAfterA(self):
        '''
        在某个时间点检查A工作流是否执行，如果A工作流没有执行，则将报警信息发送到企业微信
        如果A工作流已执行，则自动执行B工作流
        :return:
        '''
        time_now = time.strftime("%Y-%m-%d", time.localtime(time.time()))
        #找到工作流A的日志,length是获取日志长度
        fetch_executions = azkaban_client.fetchExecutions('工作流A的project_name', '工作流A的flow_name', length=200)
        exec_list = list(filter(lambda x: x[2] > time_now,[(x['execId'],time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(x['startTime']) / 1000)),
                                  time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(x['endTime']) / 1000)),
                                  int(x['endTime']) - int(x['startTime'])) for x in fetch_executions['executions']]))
        res_f1=[]
        #判断工作流某个节点是否执行完成
        job_name = '定义监控工作流里的某个节点是否执行完成，一般是监控最后的end节点'
        for exec_detail in exec_list:
            detail_info = azkaban_client.fetchExecutionsInfo(exec_detail[0])
            res_f = list(filter(lambda x: x[1] == 'SUCCEEDED', list(map(lambda x: (x['id'], x['status']), filter(lambda x: x['id'] == job_name, detail_info['nodes'])))))
        if res_f != []:
            res_f1.append(res_f)
        success_times=len(res_f1)
        #如果A工作流执行成功过1次或以上，就执行B工作流，否则发出预警
        if success_times > 0:
            azkaban_client.execFlow('自定义project_name','自定义flow_name')
        else:
            #print('A工作流今天未执行，暂不执行B工作流')
            wechat_client.markdown('A工作流今天未执行，暂不执行B工作流')

class WechatWorkWebhook:
    '''
    企业微信机器人
    '''
    headers = {"Content-Type": "text/plain"}

    def __init__(self, webhook_url, proxy=None):
        self.webhook_url = webhook_url
        self.proxy = proxy

    def markdown(self, markdown):
        if markdown == '':
            pass
        else:
            data = {
                "msgtype": "markdown",
                "markdown": {
                    "content": markdown
                }
            }
            if self.proxy:
                return requests.post(self.webhook_url, headers=self.headers, json=data, proxies=self.proxy).json()
            return requests.post(self.webhook_url, headers=self.headers, json=data).json()

        def news(self, articles):
            data = {
                "msgtype": "news",
                "news": {
                    "articles": articles
                }
            }
            if self.proxy:
                return requests.post(self.webhook_url, headers=self.headers, json=data, proxies=self.proxy).json()
            return requests.post(self.webhook_url, headers=self.headers, json=data).json()

        def media(self, media_id):
            data = {
                "msgtype": "file",
                "file": {
                    "media_id": media_id
                }
            }
            if self.proxy:
                return requests.post(self.webhook_url, headers=self.headers, json=data, proxies=self.proxy).json()
            return requests.post(self.webhook_url, headers=self.headers, json=data).json()

        def upload_media(self, file_path):
            upload_url = self.webhook_url.replace('send', 'upload_media') + '&type=file'
            if self.proxy:
                return requests.post(upload_url, proxies=self.proxy, files=[('media', open(file_path, 'rb'))]).json()
            return requests.post(upload_url, files=[('media', open(file_path, 'rb'))]).json()

        def file(self, file_path):
            media_id = self.upload_media(file_path)['media_id']
            return self.media(media_id)

if __name__ == '__main__':

    azkabanInfo = {
        "url": "azkaban地址",
        "username": "azkaban账号",
        "password": "azkaban密码"
    }
    url='企业微信机器人地址'
    wechat_client = WechatWorkWebhook(url, proxy)
    azkaban_client = AzkabanHelper(**azkabanInfo)
    azkaban_client.login()

    azkaban_client.executeBAfterA()