AEM Cloud Service | Python script to query for published items on one instance and to publish these items on another instance (instance migration)

aem-cloud-migration-republish.py

#!/usr/bin/python3

import requests
import json

def load_config(filename="config.json"):
    try:
        with open(filename, 'r') as file:
            config = json.load(file)
        return config
    except FileNotFoundError:
        print(f"Error: Config file '{filename}' not found.")
        return None
    except json.JSONDecodeError:
        print(f"Error: Unable to parse JSON in config file '{filename}'.")
        return None

def execute_aem_search_published_items(base_url, local_developer_token):
    query_url = f"{base_url}/bin/querybuilder.json"
    headers = {
        'Authorization': f'Bearer {local_developer_token}',
    }
    params = {
        'path': '/content',
        'type': '[nt:unstructured]',
        'property': 'jcr:content/cq:lastReplicationAction',
        'property.value': 'Activate',
        'p.limit': '-1',
        'p.hits': 'selective',
        'p.properties': 'jcr:path',
        'orderby': 'jcr:path',
        'orderby.sort': 'desc',
        'p.guessTotal': 'true',
    }

    response = requests.get(query_url, params=params, headers=headers)

    if response.status_code == 200:
        data = response.json()
        # Extract only the 'jcr:path' values
        paths = [item.get('jcr:path') for item in data.get('hits', [])]
        return paths
    else:
        print(f"Error executing query: {response.status_code}")
        return None

def publish_paths_to_aem(target_base_url, local_developer_token, paths):
    replicate_url = f"{target_base_url}/bin/replicate"
    headers = {
        'Authorization': f'Bearer {local_developer_token}',
    }

    success_logs = []
    error_logs = []

    for path in paths:
        params = {
            'cmd': 'Activate',
            'path': path,
            'agentId': 'publish',
        }

        response = requests.post(replicate_url, params=params, headers=headers)

        result = {"path": path}

        if response.status_code == 200:
            result["status"] = "Published successfully"
            success_logs.append(result)
            save_to_file(success_logs, "publish_success_logs.json")
            print(f"O: Publishing {path} was successfull.")
        else:
            result["status"] = f"Error publishing: {response.status_code}"
            error_logs.append(result)
            save_to_file(error_logs, "publish_error_logs.json")
            print(f"X: Publishing {path} was NOT succsesful.")
        
        

def save_to_file(data, filename):
    with open(filename, 'w') as file:
        json.dump(data, file, indent=2)

def main():
    config = load_config()
    if config:
        source_aem_base_url = config.get("source_aem_base_url", "")
        source_local_developer_token = config.get("source_local_developer_token", "")
        target_aem_base_url = config.get("target_aem_base_url", "")
        target_local_developer_token = config.get("target_local_developer_token", "")

        # Query for published elements
        paths = execute_aem_search_published_items(source_aem_base_url, source_local_developer_token)
        if paths:
            save_to_file(paths, "aem_published-paths.json")

            # Publish paths to target AEM instance
            publish_paths_to_aem(target_aem_base_url, target_local_developer_token, paths)

if __name__ == "__main__":
    main()

config.json

{
  "source_aem_base_url": "https://author-pXXXXXX-eXXXXXX.adobeaemcloud.com",
  "source_local_developer_token": "eyJhbGciOiJSUzI1...",
  "target_aem_base_url": "https://author-pYYYYYY-eYYYYYY.adobeaemcloud.com",
  "target_local_developer_token": "1dSI6Imltc19uYTE..."
}

requirements.txt

requests
json