此工具将根据预设的结构，在指定的目标空间中创建页面层级。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@File : ConfluenceMatrixCreationWebService.py
@Create Time: 2025/9/26 10:00
@Description: WEB网页服务：提供一个简单的Web界面来输入目标空间和父页面ID，并显示实时日志。
"""
from flask import Flask, render_template_string, request, jsonify, session
from flask.sessions import SecureCookieSessionInterface
import os
from atlassian import Confluence
import logging
import time
import threading
import queue
import json
# --- 配置信息
CONFLUENCE_URL = 'https://kone.atlassian.net/wiki'
SOURCE_SPACE_KEY = 'CTF'
DESTINATION_SPACE_KEY_DEFAULT = 'NGDTU' # 默认目标空间键
USERNAME = XXXXXX'' # 你的邮箱
API_TOKEN = XXXXXX' # 你的API令牌
# 配置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
app = Flask(__name__)
app.secret_key = os.urandom(24) # Flask session 密钥
session_serializer = SecureCookieSessionInterface().get_signing_serializer(app)
# 用于存储每个会话的实时日志队列
log_queues = {}
def get_or_create_log_queue(session_id):
if session_id not in log_queues:
log_queues[session_id] = queue.Queue()
return log_queues[session_id]
def clear_log_queue(session_id):
if session_id in log_queues:
del log_queues[session_id]
def escape_cql_text(text):
"""Escape text for use in CQL string literals."""
return text.replace("'", "''").replace("\\", "\\\\")
def get_page_id_by_title(confluence_instance, space_key, title, parent_id=None):
escaped_title = escape_cql_text(title)
cql_query = f"space='{space_key}' and title='{escaped_title}'"
if parent_id:
cql_query += f" and ancestor={parent_id}"
try:
results = confluence_instance.cql(cql_query)
if results.get('results'):
page_id = int(results['results'][0]['content']['id'])
app.logger.info(f"页面 '{title}' 在空间 {space_key} 中已存在，ID: {page_id}") # 记录日志
return page_id
else:
app.logger.info(f"在空间 {space_key} 中未找到页面 '{title}'") # 记录日志
return None
except Exception as e:
app.logger.error(f"执行CQL查询'{cql_query}'失败: {e}") # 记录错误日志
return None
return None
def create_or_get_page(confluence_instance, title, space, parent_id=None, session_id=None):
page_id = get_page_id_by_title(confluence_instance, space, title, parent_id)
if page_id:
msg = f"[INFO] 页面 '{title}' 已存在，ID: {page_id}"
app.logger.info(msg)
if session_id:
get_or_create_log_queue(session_id).put(msg)
return page_id
else:
page_body = '<p>此页面由Matrix创建工具自动生成。</p>'
try:
new_page = confluence_instance.create_page(
space=space,
title=title,
body=page_body,
parent_id=parent_id
)
msg = f"[INFO] 已创建页面: {new_page['title']}，ID: {new_page['id']}"
app.logger.info(msg)
if session_id:
get_or_create_log_queue(session_id).put(msg)
return int(new_page['id'])
except Exception as e:
msg = f"[ERROR] 创建页面 '{title}' 失败: {e}"
app.logger.error(msg)
if session_id:
get_or_create_log_queue(session_id).put(msg)
return None
def process_pages(confluence_instance, pages, space, parent_id, session_id):
for page_dict in pages:
for page, child_pages in page_dict.items():
page_title = page.replace('+', ' ')
page_id = create_or_get_page(confluence_instance, page_title, space, parent_id, session_id)
if page_id is not None:
if session_id:
get_or_create_log_queue(session_id).put(f"[INFO] 已处理页面: {page_title} (ID: {page_id})")
process_pages(confluence_instance, child_pages, space, page_id, session_id)
else:
if session_id:
get_or_create_log_queue(session_id).put(f"[ERROR] 处理页面失败: {page_title}")
def migration_worker(session_id, space, parent_id, source_structure):
"""在后台线程中执行迁移任务"""
try:
# 每次请求创建新的 Confluence 实例
confluence_local = Confluence(
url=CONFLUENCE_URL,
username=USERNAME,
password=API_TOKEN
)
# 发送开始信息
get_or_create_log_queue(session_id).put(f"[INFO] 开始迁移到空间: {space}")
if parent_id:
get_or_create_log_queue(session_id).put(f"[INFO] 使用父页面 ID: {parent_id} 作为根级别。")
else:
get_or_create_log_queue(session_id).put(f"[INFO] 未提供父页面 ID，将在空间根级别创建。")
# --- 执行创建逻辑 ---
top_level_parent_id = parent_id if parent_id is not None else None
for category, subcategories in source_structure.items():
category_title = category.replace('+', ' ') # 解码 URL 编码的空格
category_id = create_or_get_page(confluence_local, category_title, space, top_level_parent_id, session_id) # 使用 top_level_parent_id
if category_id is None:
log_msg = f"[ERROR] 无法创建/获取分类 '{category_title}'，跳过其子项。"
app.logger.error(log_msg) # 同时记录到服务器日志
get_or_create_log_queue(session_id).put(log_msg)
continue
app.logger.info(f"成功处理分类 '{category_title}', ID: {category_id}") # 记录成功日志
get_or_create_log_queue(session_id).put(f"[INFO] 成功处理分类 '{category_title}', ID: {category_id}")
for subcategory_dict in subcategories:
for subcategory, pages in subcategory_dict.items():
subcategory_title = subcategory.replace('+', ' ') # 解码
subcategory_id = create_or_get_page(confluence_local, subcategory_title, space, category_id, session_id)
if subcategory_id is None:
log_msg = f"[ERROR] 无法创建/获取子分类 '{subcategory_title}'，跳过其子项。"
app.logger.error(log_msg) # 同时记录到服务器日志
get_or_create_log_queue(session_id).put(log_msg)
continue
app.logger.info(f"成功处理子分类 '{subcategory_title}', ID: {subcategory_id}") # 记录成功日志
get_or_create_log_queue(session_id).put(f"[INFO] 成功处理子分类 '{subcategory_title}', ID: {subcategory_id}")
process_pages(confluence_local, pages, space, subcategory_id, session_id)
final_log_message = "[INFO] 迁移过程完成。"
app.logger.info(final_log_message) # 记录完成日志
get_or_create_log_queue(session_id).put(final_log_message)
except Exception as e:
error_msg = f"[ERROR] 迁移过程中发生异常: {str(e)}"
app.logger.error(error_msg)
get_or_create_log_queue(session_id).put(error_msg)
# 发送结束信号
get_or_create_log_queue(session_id).put("__MIGRATION_DONE__")
@app.route('/')
def index():
html_content = '''
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<title>Confluence Matrix 目录层级创建工具</title>
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "Helvetica Neue", Helvetica, Arial, "PingFang SC", "Microsoft YaHei", sans-serif; /* 设计字体 */
margin: 0;
padding: 20px;
background-color: #f0f2f5; /* 设计背景色 */
}
.container {
max-width: 800px; /* 设计宽度 */
margin: 0 auto;
background: white;
padding: 30px;
border-radius: 8px;
box-shadow: 0 4px 12px rgba(0,0,0,0.1); /* 设计阴影 */
}
h2 {
margin-top: 0;
color: #172b4d; /* 设计标题色 */
font-size: 1.8em; /* 设计标题大小 */
text-align: center;
}
p {
color: #6b778c; /* 设计段落色 */
line-height: 1.5; /* 设计行高 */
}
label {
display: block;
margin-top: 15px;
font-weight: 600;
color: #172b4d; /* 设计标签色 */
font-size: 0.95em;
}
input[type="text"], input[type="number"] {
width: 100%;
padding: 10px 12px;
margin-top: 6px;
box-sizing: border-box;
border: 1px solid #dfe1e6; /* 设计边框色 */
border-radius: 4px;
font-size: 0.95em;
background-color: #fafbfc; /* 设计输入框背景 */
color: #172b4d; /* 设计输入框文字色 */
}
input[type="text"]:focus, input[type="number"]:focus {
outline: none;
border-color: #0052cc; /* 设计焦点边框色 */
box-shadow: 0 0 0 2px rgba(0, 82, 204, 0.1); /* 设计焦点阴影 */
}
button {
margin-top: 20px;
padding: 12px 24px;
background-color: #0052cc; /* 设计按钮背景色 */
color: white; /* 设计按钮文字色 */
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 1em;
font-weight: 500;
}
button:hover {
background-color: #003d99; /* 设计按钮悬停色 */
}
button:disabled {
background-color: #a5adba; /* 设计按钮禁用色 */
cursor: not-allowed;
}
.note {
margin-top: 25px;
padding: 15px;
background-color: #e3f2fd; /* 设计 note 背景色 */
border-left: 4px solid #1890ff; /* 设计 note 边框色 */
border-radius: 0 4px 4px 0;
}
.note h4 {
margin-top: 0;
color: #0052cc; /* 设计 note 标题色 */
font-size: 1.1em;
}
.status {
margin-top: 20px;
padding: 12px;
border-radius: 4px;
font-size: 0.95em;
display: none; /* 默认隐藏 */
}
.status.success {
background-color: #e6ffed; /* 设计成功背景色 */
border: 1px solid #28a745; /* 设计成功边框色 */
color: #1d8a50; /* 设计成功文字色 */
}
.status.error {
background-color: #fff5f5; /* 设计错误背景色 */
border: 1px solid #dc3545; /* 设计错误边框色 */
color: #c53030; /* 设计错误文字色 */
}
.status.info {
background-color: #e6f7ff; /* 设计信息背景色 */
border: 1px solid #1890ff; /* 设计信息边框色 */
color: #172b4d; /* 设计信息文字色 */
}
#resultLog {
margin-top: 20px;
padding: 15px;
background-color: #262626; /* 设计深色背景 */
border: 1px solid #42526e; /* 设计边框色 */
border-radius: 4px;
height: 400px; /* 增加高度 */
overflow-y: auto; /* 添加垂直滚动条 */
font-family: "Monaco", "Menlo", "Ubuntu Mono", monospace; /* 设计等宽字体 */
font-size: 0.9em; /* 设计字体大小 */
display: none; /* 默认隐藏 */
color: #ffffff; /* 设计白色文字 */
}
/* 滚动条样式 (可选) */
#resultLog::-webkit-scrollbar {
width: 10px;
}
#resultLog::-webkit-scrollbar-track {
background: #333;
}
#resultLog::-webkit-scrollbar-thumb {
background: #555;
border-radius: 5px;
}
#resultLog::-webkit-scrollbar-thumb:hover {
background: #666;
}
.log-entry {
padding: 3px 0; /* 调整行内边距 */
line-height: 1.4; /* 调整行高 */
}
.log-timestamp {
color: #aaa; /* 设计时间戳颜色 */
margin-right: 8px;
}
.log-level {
font-weight: bold;
margin-right: 8px;
text-transform: uppercase; /* 级别大写 */
font-size: 0.85em; /* 级别字体稍小 */
}
.log-level.INFO {
color: #1890ff; /* 信息级别颜色 */
}
.log-level.WARN, .log-level.WARNING {
color: #fa8c16; /* 警告级别颜色 */
}
.log-level.ERROR {
color: #f5222d; /* 错误级别颜色 */
}
.log-message {
color: #ddd; /* 设计消息颜色 */
}
code {
background-color: #e1e8f0; /* 设计代码块背景色 */
padding: 2px 4px;
border-radius: 3px;
font-weight: bold;
}
</style>
</head>
<body>
<div class="container">
<h2>Confluence Matrix 目录层级创建工具</h2>
<p>此工具将根据预设的结构，在指定的目标空间中创建页面层级。请根据需要填写以下信息：</p>
<label for="spaceKey">目标空间 KEY:</label>
<input type="text" id="spaceKey" placeholder="例如：ChinaIT" />
<p style="font-size: 0.85em; color: #6b778c; margin-top: 4px; margin-bottom: 16px;">
请填写 Confluence 空间地址中 <code>/spaces/</code> 和 <code>/</code> 之间的部分。<br>
示例：<code>https://XXX.atlassian.net/wiki/spaces/ChinaIT/overview</code> → KEY 为 <strong>ChinaIT</strong>。
</p>
<label for="parentId">目标父页面 ID (可选):</label>
<input type="number" id="parentId" placeholder="例如：150805887" />
<p style="font-size: 0.85em; color: #6b778c; margin-top: 4px; margin-bottom: 16px;">
请填写 Confluence 页面地址中 <code>/pages/</code> 后面的数字 ID。<br>
示例：<code>https://XXX.atlassian.net/wiki/spaces/ChinaIT/pages/150805887/SFM+Smart+Field+Mobility</code> → ID 为 <strong>150805887</strong>。<br>
留空表示在目标空间的根目录下创建新页面（不挂载到任何父页面）。
</p>
<button id="runButton" onclick="runMigration()">执行创建目录结构</button>
<div id="statusMessage" class="status"></div>
<div id="resultLog"></div>
<div class="note">
<h4>说明：</h4>
<p>如果目标空间是全新的（没有任何页面），请留空 <code>目标父页面 ID</code>，脚本将在空间根目录下创建一级页面。</p>
<p>如果您希望将目录结构创建在现有页面下，请填写该页面的 ID 到 <code>目标父页面 ID</code>。</p>
</div>
</div>
<script>
let logPollingInterval;
let currentSessionId = null;
async function runMigration() {
const runButton = document.getElementById('runButton');
const statusDiv = document.getElementById('statusMessage');
const resultLogDiv = document.getElementById('resultLog');
const spaceKey = document.getElementById('spaceKey').value.trim();
const parentId = document.getElementById('parentId').value.trim();
if (!spaceKey) {
showStatus('请填写目标空间 Key', 'error');
return;
}
// 禁用按钮防止重复点击
runButton.disabled = true;
runButton.textContent = '执行中...';
// 隐藏之前的状态和日志
statusDiv.style.display = 'none';
resultLogDiv.style.display = 'none';
resultLogDiv.innerHTML = ''; // Clear previous logs
try {
const response = await fetch('/run_migration', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
spaceKey: spaceKey,
parentId: parentId || null
})
});
const data = await response.json();
if (response.ok) {
showStatus(data.message, 'info');
currentSessionId = data.session_id; // 保存会话ID
resultLogDiv.style.display = 'block'; // 显示日志区域
startPollingLogs(); // 开始轮询日志
} else {
showStatus(data.error || '请求失败', 'error');
if (data.log && data.log.length > 0) {
const logContainer = document.getElementById('resultLog');
data.log.forEach(log => {
addLogEntry(logContainer, log);
});
resultLogDiv.style.display = 'block'; // 显示日志区域
}
runButton.disabled = false;
runButton.textContent = '执行创建目录结构';
}
} catch (error) {
console.error('Error:', error);
showStatus('网络错误或服务器无响应', 'error');
runButton.disabled = false;
runButton.textContent = '执行创建目录结构';
}
}
function startPollingLogs() {
if (logPollingInterval) {
clearInterval(logPollingInterval);
}
logPollingInterval = setInterval(pollLogs, 500); // 每500ms检查一次
}
async function pollLogs() {
if (!currentSessionId) return;
try {
const response = await fetch(`/get_log/${currentSessionId}`, {
method: 'GET',
});
if (response.ok) {
const data = await response.json();
const logContainer = document.getElementById('resultLog');
if (data.logs && data.logs.length > 0) {
for (const log of data.logs) {
if (log === "__MIGRATION_DONE__") {
clearInterval(logPollingInterval);
const runButton = document.getElementById('runButton');
runButton.disabled = false;
runButton.textContent = '执行创建目录结构';
showStatus('迁移过程完成。', 'success');
return; // 停止轮询
}
addLogEntry(logContainer, log);
}
// 自动滚动到底部
logContainer.scrollTop = logContainer.scrollHeight;
}
} else {
console.error('获取日志失败');
}
} catch (error) {
console.error('轮询日志出错:', error);
}
}
function addLogEntry(container, logMessage) {
// Extract timestamp, level, and message from log string
// Expected format: [LEVEL] Message or [TIMESTAMP - LEVEL] Message
// For this specific script, we expect [LEVEL] Message
// 使用 24 小时制时间格式
const now = new Date();
const hours = String(now.getHours()).padStart(2, '0');
const minutes = String(now.getMinutes()).padStart(2, '0');
const seconds = String(now.getSeconds()).padStart(2, '0');
const timestamp = `${hours}:${minutes}:${seconds}`;
let level = 'INFO'; // Default level
let message = logMessage;
const match = logMessage.match(/^\[([^\]]+)\]\s*(.*)/);
if (match) {
level = match[1];
message = match[2];
}
const entryDiv = document.createElement('div');
entryDiv.className = 'log-entry';
const timestampSpan = document.createElement('span');
timestampSpan.className = 'log-timestamp';
timestampSpan.textContent = `[${timestamp}]`;
const levelSpan = document.createElement('span');
levelSpan.className = `log-level ${level}`;
levelSpan.textContent = `[${level}]`;
const messageSpan = document.createElement('span');
messageSpan.className = 'log-message';
messageSpan.textContent = message;
entryDiv.appendChild(timestampSpan);
entryDiv.appendChild(levelSpan);
entryDiv.appendChild(messageSpan);
container.appendChild(entryDiv);
}
function showStatus(message, type) {
const statusDiv = document.getElementById('statusMessage');
statusDiv.textContent = message;
statusDiv.className = `status ${type}`;
statusDiv.style.display = 'block'; // 显示状态区域
}
</script>
</body>
</html>
'''
return render_template_string(html_content)
@app.route('/run_migration', methods=['POST'])
def run_migration():
data = request.json
space = data.get('spaceKey', DESTINATION_SPACE_KEY_DEFAULT)
parent_id_str = data.get('parentId')
if not space:
return jsonify({'error': '目标空间 Key 是必需的。'}), 400
parent_id = None
if parent_id_str is not None and parent_id_str != '':
try:
parent_id = int(parent_id_str)
except (ValueError, TypeError):
return jsonify({'error': '目标父页面 ID 必须是有效的整数。'}), 400
# --- 目录结构定义 ---
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": []}
]
}
# 获取当前请求的 session ID
s = session_serializer.dumps(dict(session))
session_id = s.split('.')[0] # 提取 session ID (签名前的部分)
# 清理可能存在的旧队列
clear_log_queue(session_id)
# 启动后台线程执行迁移
thread = threading.Thread(target=migration_worker, args=(session_id, space, parent_id, source_structure))
thread.daemon = True
thread.start()
return jsonify({'message': '迁移已启动，正在后台运行...', 'session_id': session_id})
@app.route('/get_log/<session_id>')
def get_log(session_id):
q = get_or_create_log_queue(session_id)
logs = []
while not q.empty():
try:
log_entry = q.get_nowait()
logs.append(log_entry)
except queue.Empty:
break
return jsonify({'logs': logs})
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=9000)
