Compliance for Confluence DC to Cloud Migration Guide

import asyncio
import csv
import json
from urllib.request import Request, urlopen
from urllib.parse import urlencode
from urllib.error import HTTPError
# ===== Configuration =====
BASE_URL = "https://your-confluence-instance"
PAT = "your-personal-access-token"
OUTPUT_FILE = "confluence_pages.csv"
LIMIT = 1
MAX_CONCURRENT = 5
MAX_RETRIES = 3
# Optional: limit to a single space. Set to None for all spaces.
SPACE_KEY = None
# =========================
HEADERS = {
    "Authorization": f"Bearer {PAT}",
    "Accept": "application/json",
}
def _fetch_pages(start):
    params = {
        "type": "page",
        "limit": LIMIT,
        "start": start,
        "expand": "space,metadata.properties.classification",
    }
    if SPACE_KEY:
        params["spaceKey"] = SPACE_KEY
    url = f"{BASE_URL}/rest/api/content?{urlencode(params)}"
    req = Request(url, headers=HEADERS)
    for attempt in range(MAX_RETRIES):
        try:
            with urlopen(req) as resp:
                return json.loads(resp.read())
        except HTTPError as e:
            if e.code >= 500 or e.code == 429:
                wait = 2 ** attempt
                print(f"Retry {attempt + 1}/{MAX_RETRIES} for start={start} (HTTP {e.code}), waiting {wait}s")
                import time; time.sleep(wait)
            else:
                raise
    raise RuntimeError(f"Failed after {MAX_RETRIES} retries for start={start}")
def parse_page(page):
    classification = (
        page.get("metadata", {})
            .get("properties", {})
            .get("classification", {})
            .get("value", {})
            .get("name", {})
            .get("value", "")
    )
    return [page["id"], page["space"]["key"], page["title"], classification]
async def main():
    total = 0
    start = 0
    with open(OUTPUT_FILE, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["pageid", "spacekey", "title", "classification"])
        while True:
            offsets = list(range(start, start + LIMIT * MAX_CONCURRENT, LIMIT))
            results = await asyncio.gather(
                *(asyncio.to_thread(_fetch_pages, s) for s in offsets)
            )
            done = False
            for data in results:
                pages = data.get("results", [])
                if not pages:
                    done = True
                    break
                for page in pages:
                    writer.writerow(parse_page(page))
                total += len(pages)
            print(f"Fetched {total} pages so far...")
            if done:
                break
            start += LIMIT * MAX_CONCURRENT
    print(f"Wrote {total} pages to {OUTPUT_FILE}")
if __name__ == "__main__":
    asyncio.run(main())

import json
import time
import urllib.request
import urllib.parse
import urllib.error
import base64
# ============================================================================
# Configuration
# ============================================================================
CONFLUENCE_BASE_URL = "https://your-domain.atlassian.net"
CONFLUENCE_EMAIL = "email@example.com"
# The documentation for getting a Confluence API key is here:
# https://support.atlassian.com/atlassian-account/docs/manage-api-tokens-for-your-atlassian-account/
CONFLUENCE_API_TOKEN = "your_confluence_api_token"
# This should not be changed
COMPLIANCE_BASE_URL = "https://ac-cloud.com"
# Get the Compliance API key from your instance. Documented here: 
# https://docs.appfox.io/confluence-compliance/rest-api
COMPLIANCE_API_KEY = "your_compliance_api_key"
# Optional: set to a space key like "ENG" to only process pages in that space.
# Set to None to process all current pages.
SPACE_KEY_FILTER = None
# If True, do not apply the level. Just print what would happen.
DRY_RUN = True
# Retry settings for transient errors
MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 2
# ============================================================================
# HTTP helpers
# ============================================================================
def basic_auth_header(email, api_token):
    raw = f"{email}:{api_token}".encode("utf-8")
    encoded = base64.b64encode(raw).decode("utf-8")
    return f"Basic {encoded}"
def make_request(method, url, headers=None, data=None, retries=MAX_RETRIES):
    headers = headers or {}
    if data is not None:
        data = json.dumps(data).encode("utf-8")
        headers["Content-Type"] = "application/json"
    attempt = 0
    while True:
        attempt += 1
        req = urllib.request.Request(url=url, method=method, headers=headers, data=data)
        try:
            with urllib.request.urlopen(req) as response:
                body = response.read().decode("utf-8")
                if body:
                    return response.getcode(), json.loads(body)
                return response.getcode(), None
        except urllib.error.HTTPError as e:
            status = e.code
            body = e.read().decode("utf-8", errors="replace")
            if status in (429, 500, 502, 503, 504) and attempt <= retries:
                print(f"Retrying {method} {url} after HTTP {status} (attempt {attempt}/{retries})")
                time.sleep(RETRY_DELAY_SECONDS * attempt)
                continue
            print(f"HTTP error {status} for {method} {url}")
            print(body)
            raise
        except urllib.error.URLError as e:
            if attempt <= retries:
                print(f"Retrying {method} {url} after network error: {e} (attempt {attempt}/{retries})")
                time.sleep(RETRY_DELAY_SECONDS * attempt)
                continue
            raise
# ============================================================================
# Compliance API
# ============================================================================
def get_compliance_levels():
    url = f"{COMPLIANCE_BASE_URL}/compliance/api/v1/level"
    headers = {
        "accept": "application/json",
        "x-api-key": COMPLIANCE_API_KEY,
    }
    status, data = make_request("GET", url, headers=headers)
    if status != 200:
        raise RuntimeError(f"Failed to get compliance levels. HTTP {status}")
    levels_by_name = {}
    for level in data:
        name = level.get("name", "").strip().lower()
        if name:
            levels_by_name[name] = level
    print(f"Loaded {len(levels_by_name)} compliance levels")
    return levels_by_name
def apply_page_level(page_id, level_id):
    url = f"{COMPLIANCE_BASE_URL}/compliance/api/v1/page-level"
    headers = {
        "accept": "application/json",
        "x-api-key": COMPLIANCE_API_KEY,
    }
    body = {
        "pageId": str(page_id),
        "levelId": level_id,
    }
    if DRY_RUN:
        print(f"DRY RUN: would apply levelId={level_id} to pageId={page_id}")
        return
    status, data = make_request("POST", url, headers=headers, data=body)
    print(f"Applied page level to page {page_id} (HTTP {status})")
    if data is not None:
        print(json.dumps(data, indent=2))
# ============================================================================
# Confluence API
# ============================================================================
def confluence_headers():
    return {
        "Accept": "application/json",
        "Authorization": basic_auth_header(CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN),
    }
def get_space_id_from_key(space_key):
    """
    Looks up a Confluence space ID from a space key using the v2 spaces endpoint.
    """
    url = f"{CONFLUENCE_BASE_URL}/wiki/api/v2/spaces?keys={urllib.parse.quote(space_key)}"
    status, data = make_request("GET", url, headers=confluence_headers())
    if status != 200:
        raise RuntimeError(f"Failed to look up space '{space_key}'. HTTP {status}")
    results = data.get("results", [])
    if not results:
        raise RuntimeError(f"No space found for key '{space_key}'")
    if len(results) > 1:
        print(f"Warning: multiple spaces found for key '{space_key}', using the first one")
    space_id = results[0].get("id")
    if not space_id:
        raise RuntimeError(f"Space '{space_key}' did not include an id")
    return str(space_id)
def get_pages(space_id=None):
    """
    Yields current Confluence pages using cursor-based pagination.
    Uses _links.next and follows the cursor from that URL.
    """
    if space_id:
        url = f"{CONFLUENCE_BASE_URL}/wiki/api/v2/pages?space-id={urllib.parse.quote(space_id)}"
    else:
        url = f"{CONFLUENCE_BASE_URL}/wiki/api/v2/pages"
    while url:
        status, data = make_request("GET", url, headers=confluence_headers())
        if status != 200:
            raise RuntimeError(f"Failed to get pages. HTTP {status}")
        results = data.get("results", [])
        for page in results:
            yield page
        next_link = data.get("_links", {}).get("next")
        if next_link:
            if next_link.startswith("http://") or next_link.startswith("https://"):
                url = next_link
            else:
                url = CONFLUENCE_BASE_URL + next_link
        else:
            url = None
def get_page_properties(page_id):
    url = f"{CONFLUENCE_BASE_URL}/wiki/api/v2/pages/{page_id}/properties"
    status, data = make_request("GET", url, headers=confluence_headers())
    if status != 200:
        raise RuntimeError(f"Failed to get properties for page {page_id}. HTTP {status}")
    return data.get("results", [])
def get_page_classification(page_id):
    properties = get_page_properties(page_id)
    for prop in properties:
        if prop.get("key") == "classification":
            value = prop.get("value", {})
            name_value = value.get("name", {}).get("value")
            if name_value:
                return name_value.strip()
    return None
# ============================================================================
# Main processing
# ============================================================================
def main():
    print("Starting page classification sync")
    levels_by_name = get_compliance_levels()
    space_id = None
    if SPACE_KEY_FILTER:
        print(f"Looking up space ID for space key: {SPACE_KEY_FILTER}")
        space_id = get_space_id_from_key(SPACE_KEY_FILTER)
        print(f"Using space ID: {space_id}")
    total_pages = 0
    pages_with_classification = 0
    pages_skipped_no_classification = 0
    pages_skipped_no_matching_level = 0
    pages_applied = 0
    pages_failed = 0
    for page in get_pages(space_id=space_id):
        total_pages += 1
        page_id = page.get("id")
        title = page.get("title", "(untitled)")
        print(f"\nProcessing page {page_id}: {title}")
        try:
            classification = get_page_classification(page_id)
            if not classification:
                print(f"SKIP page {page_id}: no classification property found")
                pages_skipped_no_classification += 1
                continue
            pages_with_classification += 1
            print(f"Found classification: {classification}")
            normalized = classification.strip().lower()
            level = levels_by_name.get(normalized)
            if not level:
                print(f"SKIP page {page_id}: no matching compliance level found for '{classification}'")
                pages_skipped_no_matching_level += 1
                continue
            level_id = level["id"]
            level_name = level["name"]
            print(f"Matched compliance level: {level_name} ({level_id})")
            apply_page_level(page_id, level_id)
            pages_applied += 1
        except Exception as e:
            print(f"FAILED page {page_id}: {e}")
            pages_failed += 1
    print("\nFinished")
    print(f"Total pages seen: {total_pages}")
    print(f"Pages with classification: {pages_with_classification}")
    print(f"Pages skipped (no classification): {pages_skipped_no_classification}")
    print(f"Pages skipped (no matching level): {pages_skipped_no_matching_level}")
    print(f"Pages applied: {pages_applied}")
    print(f"Pages failed: {pages_failed}")
if __name__ == "__main__":
    main()