python 脚本备份gitlab分支（打tag）

现在使用git作为版本控制的趋势已经很明显， gitlab又是用的比较多的一个开源的git服务端；开发过程中会有很多dev分支，在发布生产后一般会把这些分支备份（打tag）, 然后合并到master分支，之后就可以删除分支了。下面的python脚本 就是自动化实现这一过程的。

需要修改 host为gitlab域名，git_user_name ，git_pwd git登录账号密码，group_project 需要处理的分组和项目，exclude_branch 需要排除的分支；
基于 # GitLab Community Edition 12.10.6版本

可以基于last_commit判断是否已经备份打tag了

"""
用于gitlab已发布版本分支， 打Tag归档, 可选删除归档分支
python3
pip install BeautifulSoup
pip install requests
pip install lxml
"""
import requests
from bs4 import BeautifulSoup
# gitlab的域名  需要自己更改
host = "https://git.xxx.com"
login_index = host + "/users/sign_in"
login_url = host + "/users/auth/ldapmain/callback"
project_index_url = host + "/{}/{}"
branches_api = host + "/api/v4/projects/{}/repository/branches/?page={}&per_page={}"
tags_api = host + "/api/v4/projects/{}/repository/tags/?page={}&per_page={}"
add_tag_url = project_index_url + "/-/tags"
tree_url = project_index_url + "/-/tree/{}"
graphql = host + "/api/graphql"
branches_url = project_index_url + "/branches"
delete_branch_url = branches_url + "/{}"

# git登录的账号密码
git_user_name = "root"
git_pwd = "*****"

# 需要处理的分组和项目, Administrator = root 组名可以打开gitlab项目首页 地址栏显示即是【https://git.xxx.com/组名/项目名】， 建议先fork项目到私人仓库测试，检查没问题再在上游仓库执行
# group_project = {'plat': ['trade'], 'root': ['pay']}
group_project = {'core': ['trade','settle','service-api',]}
# 需要排除创建tag的分支，未发布的分支无需归档
exclude_branch = ['master', 'release', 'dev-v5.8.5', 'dev-v5.8.4', 'dev-v5.8.6', 'dev-v5.8.4.1', 'dev-v5.8.3.5']
# 归档完成是否删除分支,第一次测试请改为False，受保护的分支需要受权限才能删除
delete_bak_branch = True

tag_suffix = '-Tag'
header = {
    'authority': 'git.xxx.com',
    'method': 'POST',
    'path': '/api/graphql',
    'scheme': 'https',
    'accept': '*/*',
    'accept-encoding': 'gzip, deflate, br',
    'accept-language': 'zh-CN,zh;q=0.9',
    'content-type': 'application/json',
    'sec-fetch-dest': 'empty',
    'sec-fetch-mode': 'cors',
    'sec-fetch-site': 'same-origin',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36'
}


def delete_branch(group, project, branch, session):
    if delete_bak_branch:
        # print('开始删除 Group=【%s】 Project=【%s】 Branch=【%s】' % (group, project, branch))
        token = obtain_token(session, branches_url.format(group, project))
        header['x-csrf-token'] = token
        result = session.delete(url=delete_branch_url.format(group, project, branch), headers=header)
        if is_success(result.status_code):
            print('完成删除 Group=【%s】 Project=【%s】 Branch=【%s】' % (group, project, branch))
    pass


def obtain_project(k, v, session):
    result = session.get(project_index_url.format(k, v))
    if not is_success(result.status_code):
        raise Exception('获取项目id失败')
    soup = BeautifulSoup(result.content, 'lxml')
    project_id = soup.body.attrs['data-project-id']
    link = soup.find_all(attrs={'name': 'csrf-token'})[0]
    token = link['content']
    return {'token': token, 'project_id': project_id}


def main():
    session = requests.Session()
    # 获取token
    token = obtain_token(session, login_index)
    # 登录
    user_name = login(session, token)
    for k in group_project:
        for v in group_project[k]:
            tags = []
            branches = []
            for p in range(1, 10):
                pair = obtain_project(k, v, session)
                header['x-csrf-token'] = pair['token']
                branch_result = session.get(url=branches_api.format(pair['project_id'], p, 100), headers=header)
                tag_result = session.get(url=tags_api.format(pair['project_id'], p, 100), headers=header)
                if not is_success(branch_result.status_code) or not is_success(tag_result.status_code):
                    raise Exception('obtain branch and tags error')
                branch_page = branch_result.json()
                tag_page = tag_result.json()
                have_more = False
                if branch_page:
                    for x in branch_page:
                        branches.append(x['name'])
                    have_more = True
                if tag_page:
                    for y in tag_page:
                        tags.append(y['name'])
                    have_more = True
                if not have_more:
                    break
            deal_branches = set(branches) - set(exclude_branch)
            print('过滤之后待处理分支列表=%s' % (str(deal_branches)))
            for branch in deal_branches:
                rel_add_tag_url = add_tag_url.format(k, v)
                add_token_url = rel_add_tag_url + '/new'
                new_tag = obtain_tag_name(branch, tags, k, v, session)
                if not new_tag:
                    delete_branch(k, v, branch, session)
                    continue
                add_tag_param = {'utf8': '✓', 'tag_name': new_tag, 'ref': branch,
                                 'message': '自动归档创建Tag-by-' + user_name,
                                 'authenticity_token': obtain_token(session, add_token_url)}
                # print('开始归档 Group=【%s】 Project=【%s】 Branch=【%s】Tag=【%s】' % (k, v, branch, new_tag))
                result = session.post(rel_add_tag_url, add_tag_param)
                if is_success(result.status_code):
                    print('完成归档 Group=【%s】 Project=【%s】 Branch=【%s】Tag=【%s】' % (k, v, branch, new_tag))
                    delete_branch(k, v, branch, session)
                else:
                    print('归档失败 Group=【%s】 Project=【%s】 Branch=【%s】Tag=【%s】' % (k, v, branch, new_tag))

    print("over")


def is_success(code):
    if 200 <= code < 300:
        return True
    else:
        return False


def check_commit(branch, branch_last_commit, tag_name, tag_last_commit):
    code1 = branch_last_commit.status_code
    code2 = tag_last_commit.status_code
    flag = is_success(code1) and is_success(code2)
    if not flag:
        # 检查失败， 认为不是同一个分支
        print('请求获取最后提交记录请求失败branch_last_commit=【%d】【%d】' % (code1, code2))
        return False
    try:
        branch_json = branch_last_commit.json()
        tag_json = tag_last_commit.json()
        sha1 = branch_json['data']['project']['repository']['tree']['lastCommit']['sha']
        sha2 = tag_json['data']['project']['repository']['tree']['lastCommit']['sha']
        if sha1 == sha2:
            print('判断Branch=【%s】 Tag=【%s】最新Commit=【%s】记录相同，不再生成新的Tag' % (branch, tag_name, sha1))
            return True
    except Exception as e:
        print(e)
    return False


def obtain_tag_name(branch, tags, group, project_name, session):
    tag_name = branch + tag_suffix
    project_path = group + '/' + project_name
    if tag_name in tags:
        params = "{\"operationName\":\"pathLastCommit\",\"variables\":{\"projectPath\":\"" + project_path + "\",\"ref\":\"" + branch + "\",\"path\":\"\"},\"query\":\"query pathLastCommit($projectPath: ID!, $path: String, $ref: String!) {\\n  project(fullPath: $projectPath) {\\n    repository {\\n      tree(path: $path, ref: $ref) {\\n        lastCommit {\\n          sha\\n          title\\n          description\\n          message\\n          webUrl\\n          authoredDate\\n          authorName\\n          authorGravatar\\n          author {\\n            name\\n            avatarUrl\\n            webUrl\\n            __typename\\n          }\\n          signatureHtml\\n          pipelines(ref: $ref, first: 1) {\\n            edges {\\n              node {\\n                detailedStatus {\\n                  detailsPath\\n                  icon\\n                  tooltip\\n                  text\\n                  group\\n                  __typename\\n                }\\n                __typename\\n              }\\n              __typename\\n            }\\n            __typename\\n          }\\n          __typename\\n        }\\n        __typename\\n      }\\n      __typename\\n    }\\n    __typename\\n  }\\n}\\n\"}"
        token = obtain_token(session, project_index_url.format(group, project_name))
        header['x-csrf-token'] = token
        branch_last_commit = session.request(method='POST', url=graphql, data=params.encode('utf-8'), headers=header)
        tree_token = obtain_token(session, tree_url.format(group, project_name, tag_name))
        tree_params = "{\"operationName\":\"pathLastCommit\",\"variables\":{\"projectPath\":\"" + project_path + "\",\"ref\":\"" + tag_name + "\",\"path\":\"\"},\"query\":\"query pathLastCommit($projectPath: ID!, $path: String, $ref: String!) {\\n  project(fullPath: $projectPath) {\\n    repository {\\n      tree(path: $path, ref: $ref) {\\n        lastCommit {\\n          sha\\n          title\\n          description\\n          message\\n          webUrl\\n          authoredDate\\n          authorName\\n          authorGravatar\\n          author {\\n            name\\n            avatarUrl\\n            webUrl\\n            __typename\\n          }\\n          signatureHtml\\n          pipelines(ref: $ref, first: 1) {\\n            edges {\\n              node {\\n                detailedStatus {\\n                  detailsPath\\n                  icon\\n                  tooltip\\n                  text\\n                  group\\n                  __typename\\n                }\\n                __typename\\n              }\\n              __typename\\n            }\\n            __typename\\n          }\\n          __typename\\n        }\\n        __typename\\n      }\\n      __typename\\n    }\\n    __typename\\n  }\\n}\\n\"}"
        header['x-csrf-token'] = tree_token
        tag_last_commit = session.request(method='POST', url=graphql, data=tree_params.encode('utf-8'), headers=header)
        flag = check_commit(branch, branch_last_commit, tag_name, tag_last_commit)
        if flag:
            return None
        suffix = '({})'
        count = 0
        while True:
            count = count + 1
            new_tag_name = tag_name + suffix.format(count)
            if new_tag_name not in tags:
                return new_tag_name
            else:
                tree_token = obtain_token(session, tree_url.format(group, project_name, new_tag_name))
                tree_params = "{\"operationName\":\"pathLastCommit\",\"variables\":{\"projectPath\":\"" + project_path + "\",\"ref\":\"" + new_tag_name + "\",\"path\":\"\"},\"query\":\"query pathLastCommit($projectPath: ID!, $path: String, $ref: String!) {\\n  project(fullPath: $projectPath) {\\n    repository {\\n      tree(path: $path, ref: $ref) {\\n        lastCommit {\\n          sha\\n          title\\n          description\\n          message\\n          webUrl\\n          authoredDate\\n          authorName\\n          authorGravatar\\n          author {\\n            name\\n            avatarUrl\\n            webUrl\\n            __typename\\n          }\\n          signatureHtml\\n          pipelines(ref: $ref, first: 1) {\\n            edges {\\n              node {\\n                detailedStatus {\\n                  detailsPath\\n                  icon\\n                  tooltip\\n                  text\\n                  group\\n                  __typename\\n                }\\n                __typename\\n              }\\n              __typename\\n            }\\n            __typename\\n          }\\n          __typename\\n        }\\n        __typename\\n      }\\n      __typename\\n    }\\n    __typename\\n  }\\n}\\n\"}"
                header['x-csrf-token'] = tree_token
                tag_last_commit = session.request(method='POST', url=graphql, data=tree_params.encode('utf-8'),
                                                  headers=header)
                flag = check_commit(branch, branch_last_commit, tag_name, tag_last_commit)
                if flag:
                    return None
    else:
        return tag_name


def login(session, token):
    params = {'utf8': '✓', 'username': git_user_name, 'password': git_pwd, 'remember_me': '1',
              'authenticity_token': token}
    response = session.post(login_url, params)
    if not is_success(response.status_code):
        raise Exception('登录git失败')
    soup = BeautifulSoup(response.content, 'lxml')
    user_name = soup.find_all(attrs={'class': 'user-name'})[0].text
    return user_name


def obtain_token(session, url):
    response = session.get(url)
    if not is_success(response.status_code):
        raise Exception('获取Token失败')
    content = response.text
    soup = BeautifulSoup(content, 'lxml')
    link = soup.find_all(attrs={'name': 'csrf-token'})[0]
    token = link['content']
    return token


if __name__ == '__main__':
    main()