流程

公司之前Jenkins的构建结果通知用的是微信公众号推送,但领导觉得短信更靠谱,想要改用短信通知.咱也不敢犟哪个更好,只是默默实现了这些需求.

所幸给公司搞得基建还算是比较全面,让每个员工都接入Windows AD域控,因此也就是说通过AD域接口,可以获取到员工的信息;而 Jenksin 和 GitLab 同样接入了AD域,因此思路流程是这样的:

实现

人生苦短,脚本的事情还是用 Python 分分钟实现.

用了第三方模块python-ldap,python-gitlab,python-jenkins,并且 SMS 的发送由阿里云提供,因此还用到了aliyun-python-sdk-core-v3.

如果 Jobs 用的是 Pipeline 或 Jenkinsfile ,则可以直接在构建流程中判断之前是否失败,但通常很多项目需要使用自由模式,那么就需要对每个阶段进行自定义的判断,这都是比较容易实现的.

根据当前 Job 获取 Git 仓库地址:

import jenkins

def get_git_by_job(jenkins_server:jenkins.Jenkins, job_name:str)->dict:
    config = jenkins_server.get_job_config(job_name)
    root = xml.dom.minidom.parseString(config)
    for first_level in root.childNodes:
        git_url = first_level.getElementsByTagName('scm')[0].getElementsByTagName('url')[0].firstChild.data
        if 'JOB_NAME' in git_url:
            git_url = git_url.replace('JOB_NAME', job_name)
        if isinstance(git_url, str):
            return {'is_success': True, 'job_name': job_name,
                    'git_url': git_url}
        else:
            print('Get Git_Url Failed.')
    print('Fatal: Not Exist git url, Please Check Job Name: %s' % job_name)
    return {'is_success': False, 'job_name': job_name}

然后获取 GitLab 中的 Maintainer:

import gitlab

class GitLab:
    def __init__(self,addr:str,token:str):
        self.addr = addr
        self.token = token
        self.gl_server = gitlab.Gitlab(self.addr,self.token)
        self.gl_server.auth()

    def get_proj_id(self,proj_name:str):
        proj = self.gl_server.projects.get(proj_name)
        return proj.get_id()

    def get_proj_maintainers(self, proj_name):
        proj = self.gl_server.projects.get(proj_name)
        members = proj.members.all(all=True)
        maintainer = []
        for m in members:
            if m['access_level'] == gitlab.MAINTAINER_ACCESS:  # maintainer
                user_attr = self.gl_server.users.get(m['id']).attributes
                maintainer.append(user_attr)
            elif m['access_level'] == gitlab.OWNER_ACCESS:  # owner
                user_attr = self.gl_server.users.get(m['id']).attributes
                maintainer.append(user_attr)
            else:
                pass

        return maintainer

通过 AD域 获取负责人的手机号:

import ldap

class LDAP
    def __init__(self,addr:str,base_dn:str,bind_dn:str,bind_pwd:str):
        self.addr = addr
        self.base = base_dn
        conn = ldap.initialize(addr)
        conn.simple_bind_s(bind_dn, bind_pwd)
        self.ldap = conn
        self.simple_attr = ["mail", "mobile", "displayName"]

    def get_user_simple_attr(self, sAMAccountName=None, name=None, uid=None) -> list:
        ret = None
        if name != None:
            ret = self.ldap.search(self.base, ldap.SCOPE_SUBTREE, "name=" + name, self.simple_attr)
        elif sAMAccountName != None:
            ret = self.ldap.search(self.base, ldap.SCOPE_SUBTREE, "sAMAccountName=" + sAMAccountName, self.simple_attr)
        elif uid != None:
            ret = self.ldap.search(self.base, ldap.SCOPE_SUBTREE, "uid=" + name, self.simple_attr)
        ret_type, ret_data = self.ldap.result(ret, 0)
        return ret_data

    def get_user_contact_info(self, sAMAccountName) -> dict:
        resp = {"name": sAMAccountName,
                "phone_number": '',
                "mail": '',
                "displayName": ""}
        data = self.get_user_simple_attr(sAMAccountName=sAMAccountName)
        if data[0][1]["mobile"]:
            resp['phone_number'] = data[0][1]["mobile"][0].decode('utf-8')
        if data[0][1]["mail"]:
            resp['mail'] = data[0][1]["mail"][0].decode('utf-8')

        resp["displayName"] = data[0][1]["displayName"][0].decode('utf-8')
        return resp

最后调用 SMS 提供商的 API 发送最后的构建结果即可.