この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
Confluence特定ワークスペース内の全添付ファイルについてバックアップを試みる機会があり、個別ダウンロードは漏れの発生が想定できるために一括でのダウンロードをやってみました。
抑えておきたい点としては以下2点。
- 文書ツリーの構成を維持して、どの文書の添付ファイルかすぐわかるようにする
- 添付ファイルが存在するページに限った文書ツリーを生成する
効率よく行うため、atlassian-python-apiを併用しています。
また、ページングの処理についてatlassian-python-apiはカバーしていないため、以下のリポジトリも流用しています。
必要なディレクトリツリーの構築
文書ツリー構成をローカルで再現します。具体的にはConfluence上でこうなっている場合は、
ローカル上ではこのように再現したいところです。
atlassian-python-apiでは親ページのステータスを返す関数が実装されていますが、文書ルートからの親子関係を通しで取得する関数がありません。ConfluenceのREST APIとしては取れないわけではなく、このWrapperでは直接の親ページ一つだけを返す処理となっています。具体的には以下2つ。
- get_parent_content_id
- get_parent_content_title
def get_parent_content_id(self, page_id):
"""
Provide parent content id from page id
:type page_id: str
:return:
"""
parent_content_id = None
try:
parent_content_id = (self.get_page_by_id(page_id=page_id, expand="ancestors").get("ancestors") or {})[
-1
].get("id") or None
except Exception as e:
log.error(e)
return parent_content_id
def get_parent_content_title(self, page_id):
"""
Provide parent content title from page id
:type page_id: str
:return:
"""
parent_content_title = None
try:
parent_content_title = (self.get_page_by_id(page_id=page_id, expand="ancestors").get("ancestors") or {})[
-1
].get("title") or None
except Exception as e:
log.error(e)
return parent_content_titleこの処理を参考に文書ルートから該当ページまでのツリー構成をとってみます。
def create_absolute_path(title_keys):
_ancestors = (confluence.get_page_by_id(page_id=page_id, expand="ancestors").get("ancestors") or {})
_absolute_title_keys = []
for ancestor in _ancestors:
if not 'title' in ancestor:
continue
_title = ancestor['title'].replace('/', '_')
_absolute_title_keys.append(_title)
return "/".join(_absolute_title_keys)スラッシュをアンダースコアに置き換えているのは、ページ名にはスラッシュを含めることができて且つ含まれているとそこでディレクトリが作成されてしまい、本来の文書ツリーと異なる構成になるためです。
添付ファイルのダウンロード
添付ファイルのリンクを元にダウンロードするだけです。認証を入れておかないと正常にダウンロードができません。
headers = {
"Accept": "application/json"
}
r = requests.request(
"GET",
"{}/wiki{}".format(BASE_URL, attachment["_links"]["download"]),
stream=True,
headers=headers,
auth=(USER, TOKEN)
)
if r.status_code == 200:
try:
with open(attachment_file_path, 'w') as f:
f.write(r.content)
except TypeError:
with open(attachment_file_path, 'wb') as f:
r.raw.decode_content = True
f.write(r.content)添付ファイルがバイナリなのかテキストなのかはAPIのレスポンスでは判断できないため、エラーが発生したらバイナリとして落とすように変更します。
実際のコード
幾つか省いている部分はありますが、おおよそ以下の通りです。
from atlassian import Confluence
import os
import requests
import json
from dotenv import load_dotenv
from pathlib import Path
import datetime
env_path = Path('.') / '.env'
load_dotenv(dotenv_path=env_path)
# INSERT "USER", "TOKEN", "BASE_URL" HERE
USER = os.environ["CONFLUENCE_USER"]
TOKEN = os.environ["CONFLUENCE_TOKEN"]
BASE_URL = os.environ["CONFLUENCE_BASEURL"]
logging.basicConfig(format="%(message)s")
logger = logging.getLogger(name=__name__)
logger.setLevel(level=logging.CRITICAL)
confluence = Confluence(url=BASE_URL, username=USER, password=TOKEN)
def call(path):
next_link = path
while next_link is not None:
response = requests.request(
"GET",
BASE_URL + '/wiki' + next_link,
headers=headers,
auth=(USER, TOKEN)
)
if not response.ok:
response.raise_for_status
try:
next_link = json.loads(response.text)["_links"]["next"]
except KeyError:
next_link = None
results = json.loads(response.text)["results"]
for result in results:
yield result
def get_pages_for(space):
next_link = "/rest/api/space/" + space["key"] + "/content/page"
return call(next_link)
def get_attachments_for(page):
next_link = "/rest/api/content/" + page["id"] + "/child/attachment"
return call(next_link)
def create_absolute_path(title_keys):
_ancestors = (confluence.get_page_by_id(page_id=page_id, expand="ancestors").get("ancestors") or {})
_absolute_title_keys = []
for ancestor in _ancestors:
if not 'title' in ancestor:
continue
_title = ancestor['title'].replace('/', '_')
_absolute_title_keys.append(_title)
return "/".join(_absolute_title_keys)
print(datetime.datetime.now())
err_list = open('err_space.csv', 'a', encoding='utf-8')
for page in get_pages_for({"key":"XXX"}):
page_count = page_count + 1
if not page or not 'title' in page:
continue
page_id = page['id']
absolute_path = create_absolute_path([page["title"]])
attachment_count = 0
headers = {
"Accept": "application/json"
}
# Get attachments from each page
print(f"\rcount: {page_count: <0} \n page: {page_id: <20}\n attachment: {attachment_count: <20}\033[2A", end='')
for attachment in get_attachments_for(page):
attachment_count += 1
attachment_file = attachment["title"]
attachment_file_path = f"attachments/{absolute_path}/{attachment_file}"
if os.path.exists(attachment_file_path):
continue
os.makedirs(f"attachments/{absolute_path}", exist_ok=True)
print(f"\rcount: {page_count: <20} \n page: {page_id: <20} \n attachment: {attachment_count: <20}\033[2A", end='')
r = requests.request(
"GET",
"{}/wiki{}".format(BASE_URL, attachment["_links"]["download"]),
stream=True,
headers=headers,
auth=(USER, TOKEN)
)
if r.status_code == 200:
try:
with open(attachment_file_path, 'w') as f:
f.write(r.content)
except TypeError:
with open(attachment_file_path, 'wb') as f:
r.raw.decode_content = True
f.write(r.content)
except:
err_list.write("{},{},{},{}\n".format(page['id'], page['title'], attachment['id'], attachment['title']))
print(datetime.datetime.now())ページと添付ファイルのリンク取得を独自実装にしているのは、ライブラリ側のロジックがgeneratorを使っていないためです。
実際に実行すると以下のようなログ出力になるはずです。
あとがき
事前にローカル環境にて添付ファイルが収まる程の空き容量があることを確認しておきましょう。以下のライブラリを実行することで把握できます。
ライブラリ側の実装が大量のページ及び大量の添付ファイルを対象とした場合には最適と言い難く、かつページツリー構成等を再現する手段もなかったりと、必要に応じて独自実装が求められることは確かです。
ブラウザ上の管理画面から個別にダウンロードするよりは遥かに漏れがなく且つ確実です。そう頻繁に行うことではありませんが、必要に迫られて5〜6桁の数の添付ファイルを前にせざるえなくなった場合等の参考になれば幸いです。