利用jira及confluence的API進行批次操作（查詢/更新/匯出/備份/刪除等）

前言：

近期因為某些原因需要批次替換掉 jira 和 confluence中的特定關鍵字，而且在替換前還希望進行備份（以便後續恢復）和匯出（方便檢視）
atlassian官方的api介紹檔案太簡陋，很多傳參都沒有進一步的描述說明，過程中踩了不少的坑...
故現將相關程式碼分享下，希望有類似需求的朋友能用得上，直接上程式碼：

from jira import JIRA
import requests
import re

'''
用途: jira單的查詢、匯出、更新、刪除等操作
author: tony
date: 2023
'''

class jira_tools():

    # jira API
    base_url = "http://your-jira-url.com/"
    username = "your_username"
    password = "your_password"
    jira = JIRA(base_url,basic_auth=(username, password))

    # 搜尋鍵碼和替換關鍵字
    search_keyword = '查詢關鍵詞'
    replace_keyword = '替換關鍵詞'

    def jira_search(self):
        '''查詢標題和正文中包含特定關鍵字的issue
        返回一個list,list中的元素為jira issue物件<class 'jira.resources.Issue'>
        '''
        # 拼接jql,可按需修改(此處為搜尋專案REQ和TREQ中的標題or描述中包含特定關鍵詞的issue)
        jql_query = 'project in (REQ,TREQ) AND (summary ~ "{0}" or description ~ "{0}") ORDER BY updated DESC'.format(self.search_keyword)
        # jql_query = 'summary ~ "{0}" or description ~ "{0}" ORDER BY updated DESC'.format(self.search_keyword)
        # jql_query = 'id = BUG-44257'
        
        # 每頁的大小(應該最大隻支援50)
        page_size = 50

        # 初始化起始索引和總體issues列表
        start_at = 0
        all_issues = []

        while True:
            # 執行查詢並獲取當前頁的問題
            issues = self.jira.search_issues(jql_query, startAt=start_at, maxResults=page_size)
            # 將當前頁的issues新增到總體issues列表
            all_issues.extend(issues)
            # 檢查是否已獲取所有issues
            if len(issues) < page_size:
                break
            # 更新起始索引以獲取下一頁
            start_at += page_size
        return all_issues

    def jira_export(self, issue_id, issue_summary):
        # 頁面上抓到的匯出介面(需要先行在瀏覽器上登入)
        export_url = 'http://your-jira-url.com/si/jira.issueviews:issue-word/{0}/{0}.doc'.format(issue_id)

        #替換掉標題中可能存在的特殊關鍵字,避免儲存檔案失敗
        issue_summary = re.sub(r'[【】|（）()\\/:：<>*]', '', issue_summary)
        filename = 'D:/jira_bak/{0}_{1}.doc'.format(issue_id, issue_summary)  # 下載後儲存的檔名

        response = requests.get(export_url)

        if response.status_code == 200:
            try:
                with open(filename, 'wb') as f:
                    f.write(response.content)
                print('issue匯出成功！')
            except Exception as e:
                print('issue匯出失敗~失敗原因:{0}'.format(e))

    def jira_replace(self,issues):
        '''替換issue標題和正文中的特定關鍵字'''
        for issue in issues:
            issue_id = issue.key
            issue_obj = self.jira.issue(issue_id)
            # 獲取原始標題和描述
            old_summary = issue_obj.fields.summary
            old_description = issue_obj.fields.description
            # 先匯出word
            self.jira_export(issue_id, old_summary)
            # 替換關鍵字
            new_summary = old_summary.replace(self.search_keyword, self.replace_keyword)
            # 更新問題的標題和描述(description)
            if old_description: # 描述可能為空
                new_description = old_description.replace(self.search_keyword, self.replace_keyword)
                issue_obj.update(summary=new_summary, description=new_description)
            else:
                issue_obj.update(summary=new_summary)
            # 更新問題的標題和描述
            print("{0}-{1} 關鍵詞替換成功".format(issue_id, old_summary))
    
    def jira_delete(self, issue_id):
        '''刪除特定的issue'''
        try:
            # 獲取issue
            issue = self.jira.issue(issue_id)
            # 刪除issue
            issue.delete()
            print("{0}刪除成功".format(issue_id))
        except Exception as e:
            print("{0}刪除失敗:{1}".format(issue_id, e))

# # 查詢、備份/替換
# j = jira_tools()
# issues = j.jira_search()
# issues_id_list = [ issue.key for issue in issues]
# print(len(issues_id_list),issues_id_list)
# j.jira_replace(issues)

# 刪除
# j=jira_tools()
# j.jira_delete('TREQ-18431')

import requests
import re,os
import pandas as pd
from atlassian import Confluence  # pip install atlassian-python-api

'''
用途: confluence的查詢、備份/匯出、更新、刪除、恢復等相關操作
author: tony
date: 2023
'''

def save_content_to_file(filename, content, file_format='txt'):
    '''儲存內容到檔案'''
    if file_format=='pdf':
        directory = 'D:/wiki_bak/pdf/'
        filename = directory + filename + '.pdf'
    else:
        directory = 'D:/wiki_bak/txt/'
        filename = directory + filename + '.txt'
    try:
        os.makedirs(directory, exist_ok=True)
        with open(filename, 'wb' if file_format == 'pdf' else 'w', encoding='utf-8' if file_format != 'pdf' else None) as file:
            file.write(content)
        print("內容已儲存到檔案{0}".format(filename))
    except Exception as e:
        print("{0} 檔案儲存時失敗:{1}".format(filename, e))

class wiki_tools():
    # Confluence API
    base_url = "http://your-confluence-url.com/"
    search_url = base_url + "/rest/api/search"
    content_url = base_url + "/rest/api/content"
    username = "your_username"
    password = "your_password"
    wiki_replace_record = 'D:/wiki_bak/wiki_replace_record.csv' #處理過的檔案概況

    # 搜尋鍵碼和替換關鍵字
    search_keyword = '"查詢關鍵詞"'  # 將搜尋詞用""號擴起來表示進行整詞匹配,不會被confluence拆分成多個單詞進行匹配
    replace_keyword = '替換關鍵詞'


    def wiki_search(self):
        '''查詢confluence檔案
        查詢關鍵詞：
            search_keyword
        returns:
            list:匹配檔案的content_id(即URL上的pageId)
        '''
        content_id_list = []  # 用於記錄檔案id
        start = 0
        limit = 100
        total_size = 0

        while start <= total_size:
            # 構建搜尋請求的URL
            search_url = "{0}?cql=type=page and (title~'{1}' OR text~'{2}')&start={3}&limit={4}".format(
                self.search_url, self.search_keyword, self.search_keyword, start, limit)
            # 傳送搜尋請求
            response = requests.get(search_url, auth=(self.username, self.password))
            search_results = response.json()
            total_size = search_results['totalSize']
            
            # 提取當前頁匹配的檔案 id
            page_content_id_list  = [ result['content']['id'] for result in search_results["results"]]
            content_id_list.extend(page_content_id_list)

            start += limit

        return content_id_list


    def wiki_replace(self,content_id):
        '''替換confluence檔案中的關鍵字'''
        # 獲取檔案正文部分內容
        # https://community.atlassian.com/t5/Confluence-questions/How-to-edit-the-page-content-using-rest-api/qaq-p/904345
        content_url = self.content_url + "/" + content_id + "?expand=body.storage,version,history"
 
        content_response = requests.get(content_url, auth=(self.username, self.password))

        if content_response.status_code == 200:
            content_data = content_response.json()

            # 獲取檔案最新的版本號
            latest_version = content_data["version"]["number"]

            # 獲取檔案的建立者
            createdBy = content_data["history"]["createdBy"]["displayName"]

            # 獲取檔案的建立時間 eg: 2023-05-30T11:02:44.000+08:00
            createdDate = content_data["history"]["createdDate"].split('T')[0]

            # 獲取檔案的標題
            old_title = content_data["title"]
            # 替換掉標題中的特殊字元,避免無法作為檔案命名
            old_title = re.sub(r'[【】|（）()\\/:：<>*]', '', old_title)

            # 獲取檔案的正文
            old_body = content_data["body"]["storage"]["value"]

            # 儲存檔案標題和正文內容(檔名稱: contentid_title, 檔案內容: body),以便後續恢復
            save_content_to_file(content_id + "_" + old_title, old_body)

            # 記錄所有處理過的檔案概要資訊到csv檔案(mode='a'即追加模式寫入)
            pd.DataFrame(data=[[content_id, old_title, createdBy, createdDate]]).to_csv(self.wiki_replace_record, encoding='utf-8', index=None, mode='a', header=None)

            # 匯出檔案內容為pdf(方便直接檢視)
            try:
                self.wiki_export_pdf(content_id, old_title + '_' + createdBy + '_' + createdDate)
            except Exception as e:
                # 有些檔案較大可能會超時
                print("{0}檔案匯出時發生異常:{1}".format(content_id, e))

            # 避免出現無效更新造成version無謂增加
            if self.search_keyword in old_title or self.search_keyword in old_body:
                # 替換檔案標題和正文中的關鍵字
                new_title = old_title.replace(self.search_keyword, self.replace_keyword)
                new_body = old_body.replace(self.search_keyword, self.replace_keyword)
        
                # 更新檔案
                update_data = {
                    "title": new_title,
                    "type": content_data["type"],
                    "version":{
                        "number": latest_version + 1  # 使用最新版本號加1
                    },
                    "body": {
                        "storage": {
                            "value": new_body,
                            "representation": "storage"
                        }
                    }
                }
                update_response = requests.put(content_url, auth=(self.username, self.password), json=update_data)

                if update_response.status_code == 200:
                    print("替換成功：", old_title)
                else:
                    print("替換失敗：", old_title)
            else:
                print("檔案中未包含關鍵字:{0},無需更新".format(self.search_keyword))


    def wiki_update_from_file(self, content_id, title, body):
        '''指定內容更新'''
        content_url = self.content_url + "/" + content_id + "?expand=body.storage,version"
        content_response = requests.get(content_url, auth=(self.username, self.password))

        if content_response.status_code == 200:
            content_data = content_response.json()

            # 獲取檔案最新的版本號
            latest_version = content_data["version"]["number"]

            # 更新檔案
            update_data = {
                "title": title,
                "type": content_data["type"],
                "version":{
                    "number": latest_version + 1  # 使用最新版本號加1
                },
                "body": {
                    "storage": {
                        "value": body,
                        "representation": "storage"
                    }
                }
            }
            update_response = requests.put(content_url, auth=(self.username, self.password), json=update_data)
            
            if update_response.status_code == 200:
                print("恢復成功：", title)
            else:
                print("恢復失敗：", title)


    def wiki_restore(self, path="D:/wiki_bak/txt/"):
        '''根據備份的body檔案恢復對應的confluence檔案'''
        # 獲取指定路徑下的所有檔案
        files = os.listdir(path)
        for file_name in files:
            # 根據檔名解析content_id、標題 ( 形如: contentid_title.txt )
            content_id = file_name.split('_')[0]
            title = file_name.split('_')[1].replace('.txt','')
            file_path = os.path.join(path, file_name)
            # 讀取備份檔案並恢復
            if os.path.isfile(file_path):
                print('開始處理',file_path)
                with open(file_path, 'r') as file:
                    content = file.read()
                    self.wiki_update_from_file(content_id, title, content)


    def wiki_export_pdf(self, content_id, filename):
        '''利用atlassian-python-api庫匯出pdf'''
        confluence = Confluence(
            url=self.base_url,
            username=self.username,
            password=self.password)
        page = confluence.get_page_by_id(page_id=content_id)
        response = confluence.get_page_as_pdf(page['id'])
        save_content_to_file(filename, content=response, file_format='pdf')


    def wiki_delete(self,content_id):
        '''利用atlassian-python-api庫刪除特定檔案'''
        confluence = Confluence(
            url=self.base_url,
            username=self.username,
            password=self.password)
        try:
            confluence.remove_content(content_id)
            print("檔案 {0} 刪除成功".format(content_id))
        except Exception as e:
            print("檔案 {0} 刪除失敗: {1}".format(content_id, e))


# w = wiki_tools()
# # 批次查詢&替換wiki檔案,同時備份替換前的內容
# contentid_list = w.wiki_search()
# print(contentid_list)
# for i in contentid_list:
#     print("----開始處理：{0}----".format(i))
#     w.wiki_replace(i)

# # 根據備份的檔案恢復wiki檔案內容
# w.wiki_restore()

# # 刪除特定的檔案 
# w.wiki_delete('137295690')