「All right reserved, any unauthorized reproduction or transfer is prohibitted」
需求背景
公司要求各个项目组内，需要使用统一的 Confluence Matrix 目录构建，由于项目太多，手动复制各个层级的目录到自己的项目下是不现实的，最好是需要使用接口实现。因此我自己写了一个脚本，可以一键创建 Confluence Matirx 目录层级，避免了大家手动执行迁移操作。
自己写的代码实现
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""
@Author : XXXXXX
@Contact : XXXXXX
@File : ConfluenceMigrationPageTool.py
@Create Time: 2025/5/30 18:01
@Description: 一键创建Confluence Matirx目录层级，不需要手动执行
"""
from atlassian import Confluence
# 配置信息
CONFLUENCE_URL = 'https://kone.atlassian.net/wiki'
SOURCE_SPACE_KEY = 'CTF'
DESTINATION_SPACE_KEY = 'GAPP' # 目标空间键更改为 GAPP
USERNAME = 'XXXXXX' # 替换为你的个人邮箱
API_TOKEN = 'XXXXXX # 替换为你的个人API令牌
# 创建 Confluence 实例
confluence = Confluence(
url=CONFLUENCE_URL,
username=USERNAME,
password=API_TOKEN
)
def escape_cql_query(query):
"""Escape special characters in CQL query."""
return query.replace('&', '&').replace('+', '%2B')
def get_page_id_by_title(space_key, title, parent_id=None):
cql_query = f"space={space_key} and title='{escape_cql_query(title)}'"
if parent_id:
cql_query += f" and ancestor={parent_id}"
try:
results = confluence.cql(cql_query)
if results.get('results'):
return results['results'][0]['content']['id']
except Exception as e:
print(f"Failed to execute CQL query '{cql_query}': {e}")
return None
def create_or_get_page(title, space_key, parent_id=None):
page_id = get_page_id_by_title(space_key, title, parent_id)
if page_id:
print(f"Page '{title}' already exists with ID: {page_id}")
return page_id
else:
page_body = '<p>This is a placeholder page.</p>'
try:
new_page = confluence.create_page(
space=space_key,
title=title,
body=page_body,
parent_id=parent_id
)
print(f"Created page: {new_page['title']} with ID: {new_page['id']}")
return new_page['id']
except Exception as e:
print(f"Failed to create page '{title}': {e}")
return None
def process_pages(pages, space_key, parent_id):
for page_dict in pages:
for page, child_pages in page_dict.items():
page_title = page.replace('+', ' ')
page_id = create_or_get_page(page_title, space_key, parent_id)
if page_id is not None:
process_pages(child_pages, space_key, page_id)
def main():
# 如果你正在迁移到一个全新的 Confluence 空间，并且该空间目前没有任何页面，那么 destination_parent_id 就不再需要了。
# 在这种情况下，你可以直接在根目录下创建新的页面。
# # 目标父页面ID（GOV+API）
# destination_parent_id = 95851789
# 源目录结构
source_structure = {
"00.+Product+Overview": [
{"Introduction": []},
{"Product+Roadmap": [
{"Overall": []},
{"Quarterly+planning": []}
]},
{"Project+key+member+responsibility": []}
],
"01.+Requirement": [
{"Business+Requirements": [
{"Policy&Code": []},
{"Marketing+Analyzation": []},
{"Competitors+investigation": []},
{"Business+case": []}
]},
{"Functional+Requirements": [
{"Module-XXX": [
{"Features-XXX+PRD": [
{"Business+Background": []},
{"Role+Authoritarian": []},
{"Requirement+description": []},
{"Business+Workflow": []},
{"Prototype+design": []},
{"Page+elements+definition": []},
{"Log": []},
{"Requirement+Review+Meeting+Summary": []}
]}
]}
]},
{"Non-Functional+Requirements": [
{"Role+Setting": []},
{"Product+Performance": []}
]},
{"Business+value+Review": []}
],
"02.+Engineering": [
{"01.+Architecture": [
{"Tech-arch": []},
{"Business-arch": []},
{"Data-arch": []},
{"Feature-xxx": []}
]},
{"02.+Development": [
{"Frontend-App": []},
{"Frontend-Web": []},
{"Frontend-Mini": []},
{"domain+name": [
{"domain+arch": [
{"features-xxx": []}
]},
{"app+name": [
{"app-name-api": []},
{"design+for+key+feature+1": []}
]}
]}
]},
{"03.+Data+Intelligence": []},
{"04.+Validation+Quality": [
{"Test+Specifications": [
{"Test+ENV": []},
{"Test+Strategy": []},
{"Test+Spec+Documents": []}
]},
{"Test+Cases": []},
{"Test+Reports": []},
{"Automation": [
{"Automation+Strategy": []},
{"Automation+Test+Result": []},
{"Automation+Coverage+Track": []}
]},
{"Non-Function+Test": [
{"Performance+Test": []},
{"Stability+Test": []},
{"Compatibility+test": []},
{"Usability+Test": []}
]},
{"PRD+Leaking+Bug+Retro": []}
]},
{"05.+Data+Services+Products": [
{"KCDP": [
{"PoC": []},
{"Common+Services": []},
{"Data+Engineering": []}
]},
{"Digital+enabled+services+247+services": [
{"Device+view": []},
{"Device+Shadow": []},
{"Dynamic+Scheduling": []},
{"ISN+CN": []}
]}
]}
],
"03.+Application+Security": [
{"Security+Summary": []},
{"Secure+Design": [
{"Security+protocol": []},
{"Common+Reference+design": []}
]},
{"Security+Requirements": []},
{"Security+guideline": []},
{"Security+Certificate": [
{"MLPS+certificate": []},
{"IEC-62443+certificate": []},
{"ISO27001+series": []}
]},
{"Security+Manual": []},
{"Security+Testing": [
{"Security+requirements+verification": []},
{"Hot+findings+mitigation+summary": []},
{"Pen+testing+Summary": []}
]}
],
"04.+Releases": [
{"Release+Calendar": [
{"2025": []},
{"2026": []}
]},
{"Release+Version": [
{"v-x.y.z": [
{"v-x.y.z-git-env-map": []},
{"v-x.y.z-human-resource": []},
{"v-x.y.z-runbook": [
{"v-x.y.z-runbook-result": []}
]},
{"v-x.y.z-dev-to-test": [
{"feature-xxxxxx": []}
]},
{"v-x.y.z-test-report": []},
{"v-x.y.z-security-report": []},
{"v-x.y.z-deploy-approve": []}
]},
{"v-x.y.z.w": []}
]}
],
"05.+Deployment+Operations": [
{"Deployment+Guide": []},
{"CI+CD+Pipeline": []},
{"Monitoring": []},
{"Incident+Management": []},
{"User+Manual+FAQ": []}
],
"06.+Knowledge": [],
"07.+Project+Management": [
{"Process": []},
{"Team+Contacts": []},
{"Team+Availability": []},
{"Team+Member+privilege": [
{"system-xxx": []}
]}
],
"08.+Audit": [],
"09.+Meeting+minutes": [
{"Engineering": [
{"Arch": [
{"yyyy-mm-dd-meeting+topic": []}
]},
{"Dev": []},
{"Algorithm": []},
{"DevOps": []}
]},
{"Design": []},
{"Innovation": []},
{"Cross+team": []}
]
}
for category, subcategories in source_structure.items():
category_title = category.replace('+', ' ')
category_id = create_or_get_page(category_title, DESTINATION_SPACE_KEY)
if category_id is None:
continue
for subcategory_dict in subcategories:
for subcategory, pages in subcategory_dict.items():
subcategory_title = subcategory.replace('+', ' ')
subcategory_id = create_or_get_page(subcategory_title, DESTINATION_SPACE_KEY, category_id)
if subcategory_id is None:
continue
process_pages(pages, DESTINATION_SPACE_KEY, subcategory_id)
if __name__ == "__main__":
main()
利用 AI 模型进行了代码优化
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""
@Author : XXXXXX
@Contact : XXXXXX
@File : ConfluenceMigrationPageTool.py
@Create Time: 2025/5/30 18:01
@Description: 一键创建Confluence Matrix目录层级，不需要手动执行
"""
import os
import logging
from atlassian import Confluence
# 日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 配置信息
CONFLUENCE_URL = os.getenv('CONFLUENCE_URL') or 'https://kone.atlassian.net/wiki'
SOURCE_SPACE_KEY = os.getenv('SOURCE_SPACE_KEY') or 'CTF'
DESTINATION_SPACE_KEY = os.getenv('DESTINATION_SPACE_KEY') or 'GAPP' # 目标空间键更改为 GAPP
USERNAME = os.getenv('CONFLUENCE_USERNAME') or 'XXXXXX' # 替换为个人邮箱
API_TOKEN = os.getenv('CONFLUENCE_API_TOKEN') or 'XXXXXX' # 替换为个人API令牌
def validate_config():
if not CONFLUENCE_URL or not USERNAME or not API_TOKEN or not SOURCE_SPACE_KEY or not DESTINATION_SPACE_KEY:
raise ValueError("Configuration parameters are missing or invalid.")
# 创建 Confluence 实例
confluence = Confluence(
url=CONFLUENCE_URL,
username=USERNAME,
password=API_TOKEN
)
def escape_cql_query(query):
"""Escape special characters in CQL query."""
return query.replace('&', '&').replace('+', '%2B')
def get_page_id_by_title(space_key, title, parent_id=None):
cql_query = f"space={space_key} and title='{escape_cql_query(title)}'"
if parent_id:
cql_query += f" and ancestor={parent_id}"
try:
results = confluence.cql(cql_query)
if results.get('results'):
return results['results'][0]['content']['id']
except Exception as e:
logger.error(f"Failed to execute CQL query '{cql_query}': {e}")
return None
def create_or_get_page(title, space_key, parent_id=None):
"""
Creates a new page in the specified Confluence space with the given title and parent.
If the page already exists, returns its ID.
Args:
title (str): The title of the page to create.
space_key (str): The key of the Confluence space where the page will be created.
parent_id (int, optional): The ID of the parent page under which the new page will be created.
Returns:
int: The ID of the created or existing page.
"""
page_id = get_page_id_by_title(space_key, title, parent_id)
if page_id:
logger.info(f"Page '{title}' already exists with ID: {page_id}")
return page_id
else:
page_body = '<p>This is a placeholder page.</p>'
try:
new_page = confluence.create_page(
space=space_key,
title=title,
body=page_body,
parent_id=parent_id
)
logger.info(f"Created page: {new_page['title']} with ID: {new_page['id']}")
return new_page['id']
except Exception as e:
logger.error(f"Failed to create page '{title}': {e}")
return None
def replace_plus_with_space(text):
"""
Replaces '+' with a space in the given text.
Args:
text (str): The input string containing '+' characters.
Returns:
str: The modified string with spaces instead of '+'.
"""
return text.replace('+', ' ')
def process_pages(pages, space_key, parent_id):
"""
Recursively processes and creates pages based on the provided structure.
Args:
pages (list): A list of dictionaries representing the page structure.
space_key (str): The key of the Confluence space where the pages will be created.
parent_id (int, optional): The ID of the parent page under which the new pages will be created.
"""
for page_dict in pages:
for page, child_pages in page_dict.items():
page_title = replace_plus_with_space(page)
page_id = create_or_get_page(page_title, space_key, parent_id)
if page_id is not None:
process_pages(child_pages, space_key, page_id)
def main():
validate_config()
# 源目录结构
source_structure = {
"00.+Product+Overview": [
{"Introduction": []},
{"Product+Roadmap": [
{"Overall": []},
{"Quarterly+planning": []}
]},
{"Project+key+member+responsibility": []}
],
"01.+Requirement": [
{"Business+Requirements": [
{"Policy&Code": []},
{"Marketing+Analyzation": []},
{"Competitors+investigation": []},
{"Business+case": []}
]},
{"Functional+Requirements": [
{"Module-XXX": [
{"Features-XXX+PRD": [
{"Business+Background": []},
{"Role+Authoritarian": []},
{"Requirement+description": []},
{"Business+Workflow": []},
{"Prototype+design": []},
{"Page+elements+definition": []},
{"Log": []},
{"Requirement+Review+Meeting+Summary": []}
]}
]}
]},
{"Non-Functional+Requirements": [
{"Role+Setting": []},
{"Product+Performance": []}
]},
{"Business+value+Review": []}
],
"02.+Engineering": [
{"01.+Architecture": [
{"Tech-arch": []},
{"Business-arch": []},
{"Data-arch": []},
{"Feature-xxx": []}
]},
{"02.+Development": [
{"Frontend-App": []},
{"Frontend-Web": []},
{"Frontend-Mini": []},
{"domain+name": [
{"domain+arch": [
{"features-xxx": []}
]},
{"app+name": [
{"app-name-api": []},
{"design+for+key+feature+1": []}
]}
]}
]},
{"03.+Data+Intelligence": []},
{"04.+Validation+Quality": [
{"Test+Specifications": [
{"Test+ENV": []},
{"Test+Strategy": []},
{"Test+Spec+Documents": []}
]},
{"Test+Cases": []},
{"Test+Reports": []},
{"Automation": [
{"Automation+Strategy": []},
{"Automation+Test+Result": []},
{"Automation+Coverage+Track": []}
]},
{"Non-Function+Test": [
{"Performance+Test": []},
{"Stability+Test": []},
{"Compatibility+test": []},
{"Usability+Test": []}
]},
{"PRD+Leaking+Bug+Retro": []}
]},
{"05.+Data+Services+Products": [
{"KCDP": [
{"PoC": []},
{"Common+Services": []},
{"Data+Engineering": []}
]},
{"Digital+enabled+services+247+services": [
{"Device+view": []},
{"Device+Shadow": []},
{"Dynamic+Scheduling": []},
{"ISN+CN": []}
]}
]}
],
"03.+Application+Security": [
{"Security+Summary": []},
{"Secure+Design": [
{"Security+protocol": []},
{"Common+Reference+design": []}
]},
{"Security+Requirements": []},
{"Security+guideline": []},
{"Security+Certificate": [
{"MLPS+certificate": []},
{"IEC-62443+certificate": []},
{"ISO27001+series": []}
]},
{"Security+Manual": []},
{"Security+Testing": [
{"Security+requirements+verification": []},
{"Hot+findings+mitigation+summary": []},
{"Pen+testing+Summary": []}
]}
],
"04.+Releases": [
{"Release+Calendar": [
{"2025": []},
{"2026": []}
]},
{"Release+Version": [
{"v-x.y.z": [
{"v-x.y.z-git-env-map": []},
{"v-x.y.z-human-resource": []},
{"v-x.y.z-runbook": [
{"v-x.y.z-runbook-result": []}
]},
{"v-x.y.z-dev-to-test": [
{"feature-xxxxxx": []}
]},
{"v-x.y.z-test-report": []},
{"v-x.y.z-security-report": []},
{"v-x.y.z-deploy-approve": []}
]},
{"v-x.y.z.w": []}
]}
],
"05.+Deployment+Operations": [
{"Deployment+Guide": []},
{"CI+CD+Pipeline": []},
{"Monitoring": []},
{"Incident+Management": []},
{"User+Manual+FAQ": []}
],
"06.+Knowledge": [],
"07.+Project+Management": [
{"Process": []},
{"Team+Contacts": []},
{"Team+Availability": []},
{"Team+Member+privilege": [
{"system-xxx": []}
]}
],
"08.+Audit": [],
"09.+Meeting+minutes": [
{"Engineering": [
{"Arch": [
{"yyyy-mm-dd-meeting+topic": []}
]},
{"Dev": []},
{"Algorithm": []},
{"DevOps": []}
]},
{"Design": []},
{"Innovation": []},
{"Cross+team": []}
]
}
for category, subcategories in source_structure.items():
category_title = replace_plus_with_space(category)
category_id = create_or_get_page(category_title, DESTINATION_SPACE_KEY)
if category_id is None:
continue
for subcategory_dict in subcategories:
for subcategory, pages in subcategory_dict.items():
subcategory_title = replace_plus_with_space(subcategory)
subcategory_id = create_or_get_page(subcategory_title, DESTINATION_SPACE_KEY, category_id)
if subcategory_id is None:
continue
process_pages(pages, DESTINATION_SPACE_KEY, subcategory_id)
if __name__ == "__main__":
main()
对比分析
| 特征 | 原始脚本 | 优化后脚本 |
|---|---|---|
| 安全性 | 敏感信息硬编码 | 使用环境变量存储敏感信息 |
| 日志记录 | 使用 print 语句 |
使用 logging 模块，支持不同级别的日志 |
| 输入验证 | 缺少 | 添加 validate_config 函数验证配置参数 |
| 代码复用 | 字符串替换逻辑分散 | 提取 replace_plus_with_space 函数 |
| 文档字符串 | 缺少 | 为每个函数添加详细的文档字符串 |
| 性能优化 | 无 | 可以通过批量查询等方式进一步优化（未具体实现） |
最终效果
TesterHome 为用户提供「保留所有权利，禁止转载」的选项。
除非获得原作者的单独授权，任何第三方不得转载标注了「All right reserved, any unauthorized reproduction or transfer is prohibitted」的内容，否则均视为侵权。
具体请参见TesterHome 知识产权保护协议。
如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！
No Reply at the moment.