Confluenceのページをコピーして新しくページを作るサンプル

confluence_page_post.py

import requests
import json
import os
import textwrap
from datetime import datetime
from datetime import timedelta
import re

class ConfluencePagePost:
    '''Class to post page/blog content to Confluence via REST API.'''
    def __init__(self, base_url):
        self.base_url = base_url
        self.headers = {"content-type": "application/json"}

    def set_auth(self, account, api_key):
        self.auth = (account, api_key)

    '''指定のpageIdのBodyをコピーする'''
    def copy_content(self, from_id):
        # pageIdがわかっている場合
        # APIのエンドポイントは /rest/api/content/pageId?status=any
        try:
            # ?expand=body.view を付けないと本文が取得できない....
            response = requests.get(
                self.base_url + "/rest/api/content/{0}?expand=space,body.view,body.storage,ancestors".format(from_id),
                auth = self.auth,
                headers = self.headers)
            response.raise_for_status()
            result = response.json()
            title = result['title']
            value = result["body"]["storage"]["value"]
            space_key = result["space"]["key"]
            type = result["type"]

　　　　　　　         # 親のページを取得する
            ancestors = {}
            if result['ancestors']:
                base = result['ancestors'][-1]
                ancestors = { "type": base['type'], "id": base['id'] }
            id = result['id']
            return {'title': title, 'value': value, 'space_key': space_key, 'type': type, 'id': id,
                    'ancestors': ancestors}
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''元になるページから保持しているラベルを取得する'''
    def copy_labels(self, from_id):
        # pageIdがわかっている場合
        # APIのエンドポイントは /rest/api/content/pageId?status=any
        try:
            # ?expand=body.view を付けないと本文が取得できない....
            response = requests.get(
                self.base_url + "/rest/api/content/{0}/label".format(from_id),
                auth = self.auth,
                headers = self.headers)
            response.raise_for_status()
            result = response.json()
            return result
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''指定のページに指定したラベルをつける'''
    def add_labels(self, page_id, labels):
        try:
            response = requests.post(
                self.base_url + "/rest/api/content/{0}/label".format(page_id),
                auth = self.auth,
                data = json.dumps(labels),
                headers = self.headers)
            response.raise_for_status()
            result = response.json()

            page_url = self.base_url + '/pages/viewpage.action?pageId={0}'.format(page_id)
            return {'status': 'success', 'message': 'labels added', 'url': page_url}
        except requests.exceptions.HTTPError as err:
            return {'status': 'fail', 'message': "blog post failed: {0}".format(err)}
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''指定のページにコメントを追加する'''
    def add_comment(self, page_id, type, comment_body):
        comment_data = {
                "type": "comment",
                "ancestors": [],
                "container": {
                    "id": page_id,
                    "type": type,
                    "status": "current"
                },
                "body": {
                    "storage": {
                        "value": comment_body,
                        "representation": "storage"
                    }
                }
            }

        try:
            response = requests.post(
                self.base_url + "/rest/api/content",
                auth = self.auth,
                data = json.dumps(comment_data),
                headers = self.headers)
            response.raise_for_status()
            result = response.json()

            page_url = self.base_url + '/pages/viewpage.action?pageId={0}'.format(page_id)
            return {'status': 'success', 'message': 'comment added', 'url': page_url}
        except requests.exceptions.HTTPError as err:
            return {'status': 'fail', 'message': "blog post failed: {0}".format(err)}
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''渡されたタイトルの一部、ラベル、スペース、タイプからページを検索する'''
    def search_content_by_term(self, title, label, space, type):
        try:
            cql = "space={space} AND title ~ {title} AND type={type} AND label=\"{label}\" ".format(
                space=space, title=title, label=label, type=type
            )
            # ?expand=body.view を付けないと本文が取得できない....
            response = requests.get(
                self.base_url + "/rest/api/content/search?cql={cql} order by created desc&limit=1".format(cql=cql),
                auth = self.auth,
                headers = self.headers)
            response.raise_for_status()
            results = response.json()
            return int(results['results'][0]['id'])
        except:
            return {'status': 'fail', 'message': "PageId not found by search: {0}".sys.exc_info()[0]}

    '''ページのIDからURLを取得する'''
    def get_page_url_by_page_id(self, page_id):
        return self.base_url + '/pages/viewpage.action?pageId={0}'.format(page_id)

    '''コピーして作った旨のコメント文を作成する'''
    def generate_comment_body_for_copy(self, title, url):
        return "このページは <a href='{url}'>{title}</a> をコピーして作成しています。本文を修正してくださいね。".format(url=url, title=title)

    '''コピーされた本文を元にページを作成する'''
    def create_copy_from_content(self, date_str, from_content):

        title = from_content['title']
        new_title = re.sub('([0-9]{4,8})', date_str, title)

        payload = {
            "type": from_content['type'],
            "title": new_title,
            "space": {"key": from_content['space_key']},
            "ancestors": [from_content["ancestors"]],
            "body": {
                "storage": {
                    "value": from_content['value'],
                    "representation": "storage"
                }
            }
        }

        try:
            response = requests.post(
                self.base_url + "/rest/api/content",
                auth = self.auth,
                data = json.dumps(payload),
                headers = self.headers)
            response.raise_for_status()
            result = response.json()
            page_url = self.base_url + '/spaces/' + payload['space']['key'] + '/pages/' + result['id']
            return {'status': 'success', 'message': 'blog post created', 'url': page_url, 'id': result['id']}
        except requests.exceptions.HTTPError as err:
            return {'status': 'fail', 'message': "blog post failed: {0}".format(err)}
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''親ページから最新の子ページを取得する'''
    def get_latest_child(self, parent_id):
        try:
            childs = requests.get("{0}/rest/api/content/{1}/child/page?limit=100&status=current".format(self.base_url , parent_id),
                          auth = self.auth, headers = self.headers)
            child_result = childs.json()
            return int(child_result['results'][-1]['id'])
        except:
            return {'status': 'fail', 'message': "Copyiable child page is not found: {0}".sys.exc_info()[0]}

    '''ページ作成用のエンドポイントにjsonをPOSTする'''
    def create_page(self, json_data):
        try:
            response = requests.post(
                self.base_url + "/rest/api/content",
                auth = self.auth,
                data = json.dumps(json_data),
                headers = self.headers)
            response.raise_for_status()
            result = response.json()
            page_url = self.base_url + '/spaces/' + json_data['space']['key'] + '/pages/' + result['id']
            return {'status': 'success', 'message': 'blog post created', 'url': page_url}
        except requests.exceptions.HTTPError as err:
            return {'status': 'fail', 'message': "blog post failed: {0}".format(err)}
        except:
            return {'status': 'fail', 'message': "Unexpected error: {0}".sys.exc_info()[0]}

    '''SlackにメッセージをPOSTする'''
    def send_message_to_slack(self, url, message, channel):
        payload_dic = {
            "text": message,
            "username": 'lambda bot',
            "icon_emoji": ':octocat:',
            "channel": channel,
            'link_names': 1
        }
        hook_url = url
        requests.post(hook_url, data = json.dumps(payload_dic))

    def generate_message_to_notice_slack(self, target_user, title, target_url, target_date, diff):
        message = textwrap.dedent('''
    @{target_user} {title} {target_date} のページができました。
    中身を埋めてくださいねー :octocat:
    <{url}>
    ''').format(target_user=target_user, target_date=target_date, title=title, url=target_url,
                diff=diff)
        return message

    
def lambda_handler(event, context):
    # eventオブジェクトからパース

　　 # Confluenceの操作用（親ページ、コンテンツのタイプ、タイトル、x日後に日付をセット）
    parent_id = int(event['parent_id'])
    target_date_range = int(event['date_range'])
    title = event['title']
    content_type = event['type']
    
    # Slackへの通知用
    target_user = event['target_user']
    channel = event['channel']
   
    # 初期化
    from_id = 0 
       
    # ConfluenceのエンドポイントとAPI用のアカウント、キーは環境変数から
    base_url = os.environ["CONFLUENCE_BASE_URL"]
    account = os.environ['ACCOUNT']
    api_key = os.environ['API_KEY']
    # slackのWebHookです (Exp. 'https://hooks.slack.com/services/XXXXXXXX')
    hook_url = os.environ['SLACK_HOOK_URL']

    pagePost = ConfluencePagePost(base_url)
    pagePost.set_auth(account, api_key)

    now = datetime.now()
    target_date_obj = now + timedelta(days=target_date_range)
    target_date = target_date_obj.strftime("%Y%m%d")
    date_str = target_date_obj.strftime("%Y%m%d")

    if content_type == 'page':
        from_id = pagePost.get_latest_child(parent_id)

    if content_type == 'blogpost':
        spaceKey = event['space_key']
        tag = event['tag']
        from_id = pagePost.search_content_by_term(title, tag, spaceKey, 'blogpost')

    # 指定のページをコピーして、ページを作成
    content = pagePost.copy_content(from_id)
    response = pagePost.create_copy_from_content(date_str, content)

　　 # 作成したページに、「xxをコピーして作りました」のコメントを付加
    new_id = response['id']
    from_title = content['title']

    from_url = pagePost.get_page_url_by_page_id(from_id)
    comment_body = pagePost.generate_comment_body_for_copy(from_title, from_url)
    pagePost.add_comment(new_id, content['type'], comment_body)

    # 作成したページに、元のページと同じラベルを設定
    label = pagePost.copy_labels(from_id)
    pagePost.add_labels(new_id, label['results'])

　　 # slackへの通知準備
    to_url = pagePost.get_page_url_by_page_id(new_id)
    to = datetime.now()
        
    message = pagePost.generate_message_to_notice_slack(target_user, title, to_url, target_date)
    pagePost.send_message_to_slack(hook_url, message, channel)

あんまり綺麗とは言えないですが...。
また、ConfluenceのAPIがちょっとわかりにくいのが難点。
ちょっと修正しているので、うまく動かないかもしれません。都度更新します！