Usage:
python3 copy_blogs.py --base-url https://www.mos-eisley.dk --target-space BT --source-spaces familieblog --source-spaces familieblog --token <PAT Token> --parent-id 372376751 |
Made by Claude:
#!/usr/bin/env python3
"""
Copy blog posts on a Confluence Data Center instance into a target space as
PAGES, prefixing each title with the blog's date (YYYY-MM-DD). Attachments
and labels are copied too.
Auth: Personal Access Token (Bearer). Set CONFLUENCE_PAT or use --token.
Examples:
# Dry run: all blogs from every space -> pages in space ARCHIVE
python copy_blogs.py --base-url https://wiki.example.com --target-space ARCHIVE --dry-run
# Only blogs from spaces TEAMA and TEAMB, placed under a parent page
python copy_blogs.py --base-url https://wiki.example.com --target-space ARCHIVE \
--source-spaces TEAMA TEAMB --parent-id 123456
"""
import argparse
import os
import sys
import tempfile
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
PAGE_SIZE = 25 # keep small: bodies are expanded
def make_session(token, verify):
s = requests.Session()
s.headers.update({
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"Content-Type": "application/json",
})
s.verify = verify
retry = Retry(total=5, backoff_factor=1,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=None)
s.mount("https://", HTTPAdapter(max_retries=retry))
s.mount("http://", HTTPAdapter(max_retries=retry))
return s
def paged(session, url, params):
"""Yield results from a Confluence list/search endpoint using start/limit."""
start = 0
while True:
p = dict(params, start=start, limit=PAGE_SIZE)
r = session.get(url, params=p)
r.raise_for_status()
results = r.json().get("results", [])
if not results:
return
yield from results
start += len(results)
def build_cql(source_spaces, target_space):
parts = ["type = blogpost"]
if source_spaces:
keys = ",".join(f'"{k}"' for k in source_spaces)
parts.append(f"space in ({keys})")
else:
# avoid re-copying blogs that live in the target space itself
parts.append(f'space != "{target_space}"')
return " and ".join(parts) + " order by created asc"
def existing_titles(session, base, space):
titles = set()
for c in paged(session, f"{base}/rest/api/content",
{"spaceKey": space, "type": "page"}):
titles.add(c["title"])
return titles
def copy_attachments(session, base, src_id, dst_id, dry_run=False):
"""Copy the latest version of every attachment from src_id to dst_id.
Returns (copied, failed)."""
copied = failed = 0
url = f"{base}/rest/api/content/{src_id}/child/attachment"
for att in paged(session, url, {"expand": "version,metadata"}):
name = att["title"]
if dry_run:
print(f" attachment: {name}")
copied += 1
continue
download = att["_links"]["download"]
if not download.startswith("http"):
download = base + download
meta = att.get("metadata", {}) or {}
media = (meta.get("mediaType")
or (att.get("extensions", {}) or {}).get("mediaType")
or "application/octet-stream")
comment = meta.get("comment") or ""
try:
# Stream to a temp file so large attachments don't fill memory
with session.get(download, stream=True) as r:
r.raise_for_status()
with tempfile.SpooledTemporaryFile(max_size=50 * 1024 * 1024) as tmp:
for chunk in r.iter_content(1024 * 1024):
tmp.write(chunk)
tmp.seek(0)
up = session.post(
f"{base}/rest/api/content/{dst_id}/child/attachment",
# Content-Type None => requests sets multipart boundary itself
headers={"X-Atlassian-Token": "no-check", "Content-Type": None},
files={"file": (name, tmp, media)},
data={"comment": comment, "minorEdit": "true"},
)
if up.status_code != 200:
raise RuntimeError(f"HTTP {up.status_code} {up.text[:200]}")
copied += 1
print(f" attachment ok: {name}")
except Exception as e: # keep going on a single bad attachment
failed += 1
print(f" attachment FAIL: {name}: {e}")
return copied, failed
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--base-url", required=True,
help="e.g. https://wiki.example.com (include context path if any)")
ap.add_argument("--target-space", required=True, help="space key to copy into")
ap.add_argument("--source-spaces", nargs="*", default=None,
help="limit to these space keys (default: all spaces except target)")
ap.add_argument("--token", default=os.environ.get("CONFLUENCE_PAT"),
help="Personal Access Token (or env CONFLUENCE_PAT)")
ap.add_argument("--parent-id", help="parent page id in target space")
ap.add_argument("--add-source-note", action="store_true",
help="prepend a line linking back to the original blog")
ap.add_argument("--no-labels", action="store_true", help="don't copy labels")
ap.add_argument("--no-attachments", action="store_true",
help="don't copy attachments")
ap.add_argument("--no-verify", action="store_true", help="disable TLS verification")
ap.add_argument("--dry-run", action="store_true", help="list what would be created")
args = ap.parse_args()
if not args.token:
sys.exit("No token: pass --token or set CONFLUENCE_PAT")
base = args.base_url.rstrip("/")
s = make_session(args.token, not args.no_verify)
# Check target space exists / is accessible
r = s.get(f"{base}/rest/api/space/{args.target_space}")
if r.status_code != 200:
sys.exit(f"Cannot access target space {args.target_space}: HTTP {r.status_code}")
taken = existing_titles(s, base, args.target_space)
print(f"Target space has {len(taken)} existing page(s)")
cql = build_cql(args.source_spaces, args.target_space)
print(f"CQL: {cql}")
created = skipped = failed = 0
att_ok = att_fail = 0
for blog in paged(s, f"{base}/rest/api/content/search", {
"cql": cql,
"expand": "body.storage,history,space,metadata.labels"}):
blog_id = blog["id"]
src_space = blog["space"]["key"]
date = blog["history"]["createdDate"][:10] # YYYY-MM-DD
new_title = f"{date} - {blog['title']}"
# Idempotency: skip if an identical title already exists in the target
if new_title in taken:
print(f"SKIP (exists) {new_title}")
skipped += 1
continue
body = blog["body"]["storage"]["value"]
if args.add_source_note:
link = f"{base}/pages/viewpage.action?pageId={blog_id}"
body = (f'<p><em>Copied from <a href="{link}">original blog post</a> '
f'in space {src_space}.</em></p>') + body
payload = {
"type": "page",
"title": new_title,
"space": {"key": args.target_space},
"body": {"storage": {"value": body, "representation": "storage"}},
}
if args.parent_id:
payload["ancestors"] = [{"id": args.parent_id}]
if args.dry_run:
print(f"DRY [{src_space}] {new_title}")
taken.add(new_title)
if not args.no_attachments:
a, f = copy_attachments(s, base, blog_id, None, dry_run=True)
att_ok += a
continue
r = s.post(f"{base}/rest/api/content", json=payload)
# Title clash within the target space? Disambiguate once and retry.
if r.status_code == 400 and "already exists" in r.text.lower():
payload["title"] = f"{new_title} ({src_space}-{blog_id})"
r = s.post(f"{base}/rest/api/content", json=payload)
if r.status_code != 200:
print(f"FAIL [{src_space}] {new_title}: HTTP {r.status_code} {r.text[:200]}")
failed += 1
continue
new = r.json()
taken.add(payload["title"])
created += 1
print(f"OK [{src_space}] {payload['title']} -> id {new['id']}")
if not args.no_labels:
labels = blog.get("metadata", {}).get("labels", {}).get("results", [])
if labels:
lr = s.post(f"{base}/rest/api/content/{new['id']}/label",
json=[{"prefix": "global", "name": l["name"]} for l in labels])
if lr.status_code != 200:
print(f" warn: labels not copied ({lr.status_code})")
if not args.no_attachments:
a, f = copy_attachments(s, base, blog_id, new["id"])
att_ok += a
att_fail += f
time.sleep(0.1) # be gentle with the server
print(f"\nDone. pages created={created} skipped={skipped} failed={failed}; "
f"attachments copied={att_ok} failed={att_fail}"
+ (" (dry run)" if args.dry_run else ""))
if __name__ == "__main__":
main() |