Python3批量創建Crowd用戶並分配組

背景
 

遷移 Crowd 完成後（之前採用 LDAP 方式，新遷移 Crowd 不採用），需要批量創建公司所有員工的用戶以及分配組，手工創建以及之前 Postman 的方式還是比較低效。

Python 在 N 多年前入門，寫了幾個爬蟲腳本後，再也沒用過，借這個機會順便再熟悉下 Python 腳本。

歸根結底的原因就是：本人很懶~

Crowd Api
 

https://docs.atlassian.com/atlassian-crowd/3.2.0/REST/

如下示例是基於 Crowd 3.2.0 版本的 Api，不同版本間的 Api 稍有差異。

# 添加用戶
 $ curl -u "application-name:password" -X POST -H "Content-Type: application/json" -H "Accept: application/json" -d "{"name" : "test.user", "display-name" : "Test User", "active" : true, "first-name" : "Test", "email" : "test.user@ourdomain.com", "last-name" : "User", "password" : {"value" : "mypassword"} }" http://localhost:8095/crowd/rest/usermanagement/1/user

# 用戶添加到組
 $ curl -u "application-name:password" -X POST -H "Content-Type: application/json" -d "{"name" : "all-users"}" http://localhost:8095/crowd/rest/usermanagement/1/user/group/direct?username=daodaotest
 

注意：此處-u的參數為 Crowd 中應用（Application）的用戶名和密碼，Crowd 的管理員是不能添加用戶。

Python 實現腳本
 

實現添加 Crowd 用戶，用戶添加到指定組，讀取 csv 文件批量添加用戶和設定的多個組。

crowdUsers.csv 用戶數據 csv 文件

name,displayName,email
 daodaotest1,daodaotest1,daodaotest1@daodaotest.com
 daodaotest2,daodaotest2,daodaotest2@daodaotest.com
 daodaotest3,daodaotest3,daodaotest3@daodaotest.com
 ......
 

addCrowdUsers.py 批量添加 Crowd 用戶和用戶組腳本

  #!/usr/bin/python  # -*- coding: UTF-8 -*-  #  # Filename     addCrowdUsers.py  # Revision     0.0.1  # Date       2020/5/14  # Author      jiangliheng  # Email      jiang_liheng@163.com  # Website     https://jiangliheng.github.io/  # Description   批量添加 Crowd 用戶和用戶組    import requests  from requests.auth import HTTPBasicAuth  import csv  from itertools import islice    # 請求 headers  headers = {    'Accept': 'application/json',    'Content-type': 'application/json',  }    # crowd 訪問基礎路徑  base_url='http://localhost:8095'    # 添加用戶的默認用戶組和密碼  auth_username='application-name'  auth_password='password'    # 用戶默認密碼  password='daodaotest'    def addUser(name,displayName,email):    """    添加單用戶      :param name: 登錄用戶，建議拼音全稱，如：jiangliheng    :param displayName: 顯示名稱，建議中文全稱，如：蔣李恆    :param email: 郵箱地址    :return: status_code 狀態碼,text 響應報文信息    """      # 請求 json 數據    data = '{       "name" :"' + name + '",       "email" : "' + email + '",       "active" : true,       "first-name" : "' + displayName + '",       "last-name" : "' + displayName + '",       "display-name" : "'+ displayName + '",       "password" : {         "value" : "' + password + '"       }     }'      # 發起請求    # 解決中文亂碼問題 data.encode("utf-8").decode("latin1")    response = requests.post(      base_url + '/crowd/rest/usermanagement/1/user',      headers=headers,      auth=HTTPBasicAuth(auth_username,auth_password),      data=data.encode("utf-8").decode("latin1")    )      # 狀態碼    status_code=response.status_code    # 響應報文信息    text=response.text      # 狀態判斷    if str(status_code).startswith("2"):      print("%s 用戶添加成功，狀態碼：%s ，響應報文信息：%s" % (name,status_code,text))    else:      print("%s 用戶添加失敗，狀態碼：%s ，響應報文信息：%s" % (name,status_code,text))      # 返回 狀態碼，響應報文信息    return status_code,text    def addGroup(username,groupname):    """    用戶添加到組      :param username: 登錄用戶，建議拼音全稱，如：jiangliheng    :param groups: 用戶組，用逗號隔開，如：bitbucket-users,bamboo-users    :return: status_code 狀態碼,text 響應報文信息    """      # 請求 json 數據    data = '{       "name" :"' + groupname + '"     }'      # 發起請求    response = requests.post(      base_url + '/crowd/rest/usermanagement/1/user/group/direct?username='+username,      headers=headers,      auth=HTTPBasicAuth(auth_username,auth_password),      data=data    )      # 狀態碼    status_code=response.status_code    # 響應報文信息    text=response.text      # 狀態判斷    if str(status_code).startswith("2"):      print("%s 用戶添加組 %s 成功，狀態碼：%s ，響應報文信息：%s" % (username,groupname,status_code,text))    else:      print("%s 用戶添加組 %s 失敗，狀態碼：%s ，響應報文信息：%s" % (username,groupname,status_code,text))      # 返回 狀態碼，響應報文信息    return status_code,text    def addUserByCsv(csvfile):    """    通過 CSV 文件批量添加用戶，並加到組      :param filename: Crowd 用戶 csv 文件    """      # 批量讀取 csv 的用戶    with open(csvfile, 'r', encoding='utf-8') as f:      fieldnames = ("name", "displayName", "email")      reader = csv.DictReader(f, fieldnames)        for row in islice(reader, 1, None):        print("批量添加用戶 %s" % (row["name"]))        # 添加用戶        addUser(row["name"],row["displayName"],row["email"])        # 添加多個組        addGroup(row["name"],"all-users")        addGroup(row["name"],"bitbucket-users")        addGroup(row["name"],"confluence-users")        addGroup(row["name"],"jira-software-users")        addGroup(row["name"],"sonar-users")        f.close()    def main():    # 通過 CSV 文件批量添加用戶，並加到組    addUserByCsv("crowdUsers.csv")      # 添加單用戶    # addUser("daodaotest","叨叨軟件測試","daodaotest@daodaotest.com")      # 添加用戶到組    # addGroup("daodaotest","all-users")    if __name__ == "__main__":    main()