禅道Bug的一次迁移

一、场景

        平时工作记录在公司禅道上的问题想备份一份到本地，但是又没有公司禅道的数据库信息，有时候出测试报告想批量调整数据方便截图很困难，同时也为了学习禅道数据流转过程，所以有了把缺陷保存到本地一份的想法。

        实际上禅道支持导入缺陷，但是开源版不支持，所以就饶了一圈。

 二、思路：

        1.直接备份整个数据库行不通，考虑通过接口获取缺陷信息，生成sql插入到本地数据库中。

        2.不使用接口获取信息，直接导出缺陷数据，通过获取Excel信息生成sql插入到本地数据库。

这里使用思路一，因为禅道官方提供了二次开发指南，包括基本的接口文档等等，主要是更方便学习以接口的方式操作禅道。

三、操作流程

1.查看禅道官方文档，获取token是第一步

获取Token - 禅道二次开发手册 - 禅道项目管理软件

2.根据提供的文档，确定请求地址和方式，调试后可获取token。注：url_base（前置URL）可以根据登录禅道进行获取。例如：http://192.168.1.1:8080/zentao/user-login-L21heDQxMi8=.html，这个地址就是http://192.168.1.1:8080/zentao

3.获取bug信息。注：在官方手册上只能找到获取产品bug的接口文档，但是我的禅道上没有产品这个概念，只有项目。实际上把products改为projects可以直接获取项目的bug信息，其接口文档的评论区上也提供了其他方法，但是不好用。这里加上status、limit两个参数可以查看所有的bug。

4.信息可以拿到，开始分析数据库结构，主要是步骤中图片文件的关联关系，体现在zt_file和zt_bug两张表。

        对应关系1：zt_bug表中step字段文件名的前缀数字即对应zt_file文件的id

        对应关系2：zt_file表文件pathname和图片路径的关联关系，比如2024/12/1.png对应的就是ZenTao/app/zentao/www/data/upload/1/202412/1.png。注：因为是插入的数据，所以pathname就用的id，实际上禅道会对创建产生的图片文件会进行加密，肯定不会叫这个名字

5.理清接口调用和数据库关联关系后，开始写代码。这里使用python进行操作。代码结构如下：其中获取sql等场景操作都在scene下，其余都是封装的一些方法。

5.1 获取sql所需要的数据信息

from api.zentao import zentao
import json
from common.json_extractor import extractor
from common.read_data import data
class Get_sqlData():def __init__(self):self.data = Nonedef get_sql_data(self,header,json_data):json_data = json.dumps(json_data)res = zentao.zentao_login(data=json_data, headers=header)data.update_ini("../config/setting.ini", "token", "token", res.json()['token'])print(res.json()['token'])header = {"Content-Type": "application/json","token": res.json()['token']}res = zentao.zentao_bug(id="5", headers=header)# print(res.json())title_list = extractor.extract_by_jsonpath(res.json(), "$.bugs[*].title")steps_list = extractor.extract_by_jsonpath(res.json(), "$.bugs[*].steps")# print(steps_list)pattern = r'\{(\d+)\.(?:jpeg|png)\}'extracted_numbers = extractor.extract_numbers_from_img_tags(steps_list, pattern)# print(extracted_numbers)return title_list,steps_list,extracted_numbers
get_sqlData = Get_sqlData()
if __name__ == '__main__':header = {"Content-Type": "application/json"}json_data = {"account" : "admin","password" : "123456"}res = get_sqlData.get_sql_data(header,json_data)print(res)

5.2 生成sql语句，main函数中注释的部分用于切换生成哪个sql语句，这里需要生成zt_file和zt_bug两个表的sql语句。

from get_sql_data import get_sqlDataclass GenerateSql():def __init__(self):self.data = Nonedef generate_sql_bug(self,sql_template,title_list=None,step_list=None,file_id_list=None):if file_id_list==None:sql_statements = []for title,step in zip(title_list,step_list):# 使用str.format()方法替换占位符sql_statement = sql_template.format(title=title.replace("'", "''"),  # 转义单引号以防止SQL语法错误steps=step.replace("'", "''")  # 同样转义HTML标签中的单引号)sql_statements.append(sql_statement)return sql_statementselse:sql_statements = []for id in file_id_list:sql_statement = sql_template.format(id=id.replace("'", "''"),  # 转义单引号以防止SQL语法错误)sql_statements.append(sql_statement)return sql_statementsgenerate_sql = GenerateSql()
if __name__ == '__main__':sql_template = """INSERT INTO `zentao`.`zt_bug` (`product`, `title`, `severity`, `pri`, `type`,`steps`, `status`, `openedBy`, `openedDate`, `openedBuild`,`assignedTo`, `assignedDate`, `resolvedBy`, `resolution`,`resolvedBuild`, `resolvedDate`, `closedBy`, `closedDate`,`lastEditedBy`, `lastEditedDate`) VALUES (1, '{title}', 3, 3, 'codeerror','{steps}','closed', 'admin', '2024-12-09 17:06:17', 'trunk','admin', '2024-12-10 16:33:15','admin', 'fixed', 'trunk', '2024-12-10 00:00:00','admin', '2024-12-10 00:00:00','admin', '2024-12-10 16:33:15');"""sql_template_file = """INSERT INTO `zentao`.`zt_file` ( `id`, `pathname`, `title`, `extension`, `objectType`, `objectID`, `addedBy`, `addedDate`, `downloads`, `extra`, `deleted` )VALUES( {id}, '202412/{id}.png', 'image.png', 'png', '', 0, 'admin', '2024-12-10 00:00:00', 0, '', '0' );"""
#    print(type(sql))header = {"Content-Type": "application/json"}data = {"account" : "admin","password" : "123456"}data_list = get_sqlData.get_sql_data(header,data)# sql = generate_sql.generate_sql_bug(sql_template,title_list=data_list[0],step_list=data_list[1])# for i in sql:#     print(i)sql =generate_sql.generate_sql_bug(sql_template_file,file_id_list=data_list[2])for i in sql:print(i)

5.3 下载图片至本地禅道路径

from concurrent.futures import ThreadPoolExecutor, as_completed
from common.read_data import data
from get_sql_data import get_sqlData
import os
import requests
from urllib.parse import urlparse, unquote
from common.logger import logger
from PIL import Image
import iodef download_image(url, save_path, id):# 确保目录存在if not os.path.exists(save_path):os.makedirs(save_path)try:token = data.load_ini("../config/setting.ini")['token']['token']header = {"token": token}response = requests.get(url, headers=header, stream=True)if response.status_code == 200:try:img = Image.open(io.BytesIO(response.content))filename = f"{id}.png"img = img.convert("RGB")  # 确保图片模式兼容JPEGfile_path = os.path.join(save_path, filename)img.save(file_path, 'JPEG')logger.info(f"Downloaded and saved as JPEG: {url} to {filename}")except Exception as e:logger.error(f"Error processing image from {url}: {e}")else:logger.error(f"Failed to download image {url}")with open(file_path, 'wb') as file:file.write(response.content)print(f"Downloaded: {filename}")except Exception as e:print(f"Failed to download {url}. Error: {e}")def download_images(url_list, save_path, id_list, max_workers=5):# 使用线程池并发下载图片with ThreadPoolExecutor(max_workers=max_workers) as executor:futures = [executor.submit(download_image, url, save_path, id) for url, id in zip(url_list, id_list)]for future in as_completed(futures):future.result()  # 获取结果并处理异常（如果有的话）if __name__ == '__main__':header = {"Content-Type": "application/json"}json_data = {"account": "admin","password": "123456"}res = get_sqlData.get_sql_data(header, json_data)id = res[2]url_base = data.load_ini("../config/setting.ini")['host_Portal_System_Test']['api_root_url']url_list = [url_base + f"/file-read-{i}.png" for i in id]print(url_list)path = r"D:\ZenTao\app\zentao\www\data\upload\1\202412"# 调用函数下载图片download_images(url_list, path, id_list=id)