(资料图片仅供参考)

近期因为某些原因需要批量替换掉 jira 和 confluence中的特定关键字，而且在替换前还希望进行备份（以便后续恢复）和导出（方便查看）atlassian官方的api介绍文档太简陋，很多传参都没有进一步的描述说明，过程中踩了不少的坑...故现将相关代码分享下，希望有类似需求的朋友能用得上，直接上代码：

from jira import JIRAimport requestsimport re"""用途: jira单的查找、导出、更新、删除等操作author: tonydate: 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对象        """        # 拼接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 requestsimport re,osimport pandas as pdfrom atlassian import Confluence  # pip install atlassian-python-api"""用途: confluence的查找、备份/导出、更新、删除、恢复等相关操作author: tonydate: 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")