利用jira及confluence的API进行批量操作(查找/更新/导出/备份/删除等)

这篇具有很好参考价值的文章主要介绍了利用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')

 文章来源地址https://www.toymoban.com/news/detail-468677.html

到了这里,关于利用jira及confluence的API进行批量操作(查找/更新/导出/备份/删除等)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于mybatis进行批量更新

    MyBatis是一种基于Java的持久层框架,提供了一种优雅的方式来进行数据库操作。对于批量更新数据操作,MyBatis 提供了两种方法:使用 foreach 标签和 batch 执行器。 使用 foreach 标签 使用 foreach 标签时,需要编写 SQL 语句,使用 ${} 或 #{} 占位符传递参数。示例: id:SQL 语句的标

    2024年02月17日
    浏览(37)
  • C++中利用OpenCV进行图像批量处理

    想要对大量图像进行简单处理,我们可以利用代码实现。 OpenCV作为开源的图像处理库,安装方便,容易上手,功能强大,受到了很多人的喜爱。 笔者正在参加全国大学生智能汽车竞赛。由于放假在家,家中没有铺设赛道的条件,我找到了一款上位机,可以将智能车的图像导

    2024年02月03日
    浏览(51)
  • Elasticsearch Java API 的使用-更新索引(update & upset)与 Bulk的批量更新

    Java更新索引(update upset) update 更新使用UpdateRequest(update类型更新,只能更新) upset 要用IndexRequest设定添加文档,UpdateRequest设定更新文档,设定upset执行有则修改无则更新(upset类型更新,文档不存在时创建) 基于Bulk的批量更新(update upset) 动态的更新一个 documents 中的任

    2024年02月11日
    浏览(50)
  • 利用Sqlmap API接口联动Google Hacking批量SQL注入检测

    目录 前言 slghack自动化搜集URL Sqlmap API 脚本slghack_sqli 挖掘SQL注入漏洞的一种方式就是通过Google Hacking搜索那些可能存在SQL的URL,然后手工的探测注入点。但是现在的SQL注入漏洞的网站是比较少的了,所以这样一个一个手工测效率有一点低。 sqlmap比较好的一点是可批量扫描多个

    2024年04月14日
    浏览(35)
  • 利用ChatGPT如何进行批量长文本处理工具GPTBAT

    大家好,我是技术宅小伙,今天要跟大家分享一下我之前写的 GPT 长文本处理程序。当时我写完后就把它放到 Hog 上了,因为最开始是为了自己用,所以后来就忘掉了。最近有同学把它翻出来用,然后经常来问我,说不知道这个东西怎么用。其实在我看来这个挺简单的,但是如

    2023年04月22日
    浏览(49)
  • 数据批量操作:如何在HBase中进行批量操作

    HBase是一个分布式、可扩展、高性能的列式存储系统,基于Google的Bigtable设计。它是Hadoop生态系统的一部分,可以与HDFS、MapReduce、ZooKeeper等其他组件集成。HBase适用于大规模数据存储和实时数据访问场景,如日志记录、实时数据分析、实时数据流处理等。 在HBase中,数据以列族

    2024年04月23日
    浏览(32)
  • Java API批量操作Elasticsearch

    @Test public void batchAddIndex() throws IOException { BulkRequestBuilder bulkRequest = client .prepareBulk(); bulkRequest.add( client .prepareIndex( “batch_test1” , “batch” , “1” ) .setSource( jsonBuilder () .startObject() .field( “user” , “lzq” ) .field( “postDate” , new Date()) .field( “message” , “trying out Elasticsearch”

    2024年04月09日
    浏览(45)
  • HBase Java API 开发:批量操作 第2关:批量删除数据

    删除单行数据 删除一行数据很简单,我们来看个示例: 这段代码就可以删除行键为 row1 的行。 删除多行数据 如何删除多行数据呢? 相信你已经猜到了,既然 get() 方法有重载方法,那应该 delete() 方法也有,确实: 这样就可以删除多行数据啦。 编程要求 还等啥,亲自试一试

    2024年02月05日
    浏览(56)
  • 数据库高效批量更新操作 MERGE INTO

    使用 UPDATE 批量更新大量的数据,会出现效率低下,有时候甚至卡死的情况,后面通过使用 MERGE INTO 代替 UPDATE 执行批量更新,会提升执行效率。 原理:因为 UPDATE 关联子查询写法,是选到一条做一次子查询,这种写法得更新少,走好的索引才行,MERGE 写法是执行完 ON 后一次

    2024年02月17日
    浏览(42)
  • 对分库分表进行批量操作

    对ShardingJDBC基础了解:https://blog.csdn.net/m0_63297646/article/details/131894472 对批量操作案例:https://blog.csdn.net/m0_63297646/article/details/131843517 分为db0和db1两个库,每个库都有三张订单表,分表键根据年份【year】,分库键根据店铺id【store_id】 在db0中存在两张学生信息表,分表键根据年

    2024年02月10日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包