前言：
近期因为某些原因需要批量替换掉 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')
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/1.jpg&w=218&h=124&zc=1)
