Select to view content in your preferred language

Update the URL of a service in a web map Python 3.x REST

980
0
08-11-2019 07:35 AM
HenryLindemann
Esri Contributor
1 0 980

Usually when updating a URL on portal I use ArcGIS Online Assistant,  it is a excellent resource for quick maintenance, but in this case I had to update hundreds of Web Maps and using AGO Assistant would have been time consuming.

I initially started by looking at the ArcGIS API for python version 1.6.2, but unfortunately I had no luck in updating the services with the API, so searching the web for a scrip to update web map services I found this one Example: Update the URL of a service in a web map—Portal for ArcGIS (10.7 and 10.7.1) | ArcGIS Enter... but the script is still in python 2.7 and its end is near  Python 2.7 Countdown.

So I took the time to update the script using the standard library as far as possible but I do like  requests · PyPI to handle my web calls.

Well here is the result I hope someone finds it helpful

This Crawls trough all your portals content and updates the DNS

It goes without saying use with caution and at own risk.

# Original Source: 
Example: Update the URL of a service in a web map—Portal for ArcGIS (10.7 and 10.7.1) | ArcGIS Enter...

import urllib.request
import json
import requests
from urllib.parse import urlencode
import logging
import os


def header():
    return {
        'content-type': "application/x-www-form-urlencoded"
        , 'accept': "application/json"
        , 'cache-control': "no-cache"
    }


def get_token(portal_admin_url, portal_username, portal_password, root_url):
    # Generate Token
    url = f"{portal_admin_url}/sharing/rest/generateToken"

    payload = {
        'f': 'json'
        , 'username': portal_username
        , 'password': portal_password
        , 'client': 'referer'
        , 'referer': f'{root_url}'
        , 'expiration': '100'
    }

    return json.loads(
        requests.request(
            "POST"
            , url
            , data=payload
            , headers=header()
            , verify=False
        ).text
    )['token']


def search_portal(portal_url, query=None, totalResults=None, sortField='numviews',
                 sortOrder='desc', token=None):

    all_results = []
    if not totalResults or totalResults > 100:
        num_results = 100
    else:
        num_results = totalResults

    results = __search__(
        portal_url
        , query
        , num_results
        , sortField
        , sortOrder
        , 0
        , token
    )

    if not 'error' in results.keys():
        if not totalResults:
            totalResults = results['total'] # Return all of the results.
        all_results.extend(results['results'])
        while (results['nextStart'] > 0 and
               results['nextStart'] < totalResults):
            # Do some math to ensure it only
            # returns the total results requested.
            num_results = min(totalResults - results['nextStart'] + 1, 100)
            results = __search__(
                portalUrl=portal_url
                , query=query
                , sortField=sortField
                , sortOrder=sortOrder
                , token=token
                , start=results['nextStart']
            )
            all_results.extend(results['results'])
        return all_results
    else:
        logging.info(results['error']['message'])
        return results


def __search__(portalUrl, query=None, numResults=1000, sortField='numviews',
               sortOrder='desc', start=0, token=None):
    '''Retrieve a single page of search results.'''
    params = {
        'q': query
        , 'num': numResults
        , 'sortField': sortField
        , 'sortOrder': sortOrder
        , 'f': 'json'
        , 'start': start
    }

    if token:
        # Adding a token provides an authenticated search.
        params['token'] = token

    return json.loads(
        requests.request(
            "POST"
            , f'{portalUrl}/sharing/rest/search?'
            , data=params
            , headers=header()
            , verify=False
        ).text
    )


def webmap_get_data_or_info(webmapId, portalUrl, token, data_or_info):
    # Query either the rest info or the rest data
    try:
        params = {
            'token': token
            , 'f': 'json'
        }
        logging.info(f'Getting Info for: {webmapId}')
        item_data_req = requests.request(
            "POST"
            , f'{portalUrl}/sharing/content/items/{webmapId}{data_or_info}?'
            , data=params
            , headers=header()
            , verify=False
        ).text

        logging.info(item_data_req)
        return json.loads(item_data_req)
    except Exception as e:
        logging.error(str(e))



def update_webmap_with_folder(webmapId, portalUrl, token, new_data, owner, folder=None):
    # Update dns in webmap located in root or a folder
    params = {'token': token}
    data = {'text': str(new_data)}
    folder = "" if folder is None else f"/{folder}/"
    mod_request = f"{portalUrl}/sharing/rest/content/users/{owner}{folder}items/{webmapId}/update?" \
        f"{urllib.parse.urlencode(params)}"
    logging.info(mod_request)
    # set verify to False if certificates are not valid.
    try:
        logging.info(
            requests.request(
                "POST"
                , mod_request
                , data=data
                , headers=header()
                , verify=True
            ).text
        )
    except Exception as e:
        logging.error(e)


def update_webmap_service(webmapId, old_dns, new_dns, token, portalUrl):
    '''Replaces the URL for a specified map service in a web map.'''

    itemString = webmap_get_data_or_info(
        webmapId
        , portalUrl
        , token
        , '/data'
    )

    for layer in itemString['operationalLayers']:
        try:
            url = layer['url']
            logging.info(f"Searching item ---- {url}")

        except Exception as e:
            logging.error(e)
            return True

        dns = str(url).split('/')
        if dns[2].lower() == old_dns.lower():
            dns[2] = new_dns.lower()
            logging.info('/'.join(dns))
            layer['url'] = '/'.join(dns)

    item_info = webmap_get_data_or_info(
        webmapId
        , portalUrl
        , token
        , ""
    )
    logging.info(itemString)
    logging.info('Updating: ' + item_info['title'])

    update_webmap_with_folder(
        webmapId
        , portalUrl
        , token
        , itemString
        , item_info['owner']
        , item_info['ownerFolder']
    )


def write_data(content):
    # filter out esri_ accounts
    # filter out false positive matches for web apps
    data_to_write = [json.loads(json.dumps(item)) for item in content
                     if item['owner'][:5] != 'esri_'
                     or item['type'] != 'Web Mapping Application']

    with open(f'{get_script_dir()}//content.txt', 'w') as content_file:
        json.dump(data_to_write
                  , content_file
                  , indent=2)


def read_data():
    # Reads item data and returns a list of item id's
    with open(f'{get_script_dir()}//content.txt', 'r') as content_file:
        data = json.load(content_file)
        item_id = []
        for record in data:
            if record is None:
                continue
            item_id.append(record['id'])
        return item_id


def setup_log(folder, level, filemode):
    logging.basicConfig(
        level=level
        , filename=f"{folder}//log.txt"
        , filemode=filemode
    ) if os.path.isdir(folder) else os.mkdir(folder)


def get_script_dir():
    script_path = os.path.realpath(__file__)
    working_dir = script_path.split('\\')[0:-1]
    return '\\'.join(working_dir)


if __name__ == '__main__':

    setup_log(
        f'{get_script_dir()}\\log'
        , logging.DEBUG
        , 'w'
    )

    portal_url = 'https://gisportal.gauteng.gov.za/portal'
    old_dns = 'gisportal.gauteng.gov.za'
    new_dns = 'gisportal.gauteng.gov.za'
    username = 'hlindemann'
    password = 'Br@aks1809'
    referer = 'https://gisportal.gauteng.gov.za/portal'

    # Get a token for the source Portal for ArcGIS.
    token = get_token(
        portal_url
        , username
        , password
        , referer
    )

    # Get a list of the content matching the query.
    write_data(
        search_portal(
            portal_url=portal_url
            , query='type:"Web Map"'
            , token=token
        )
    )
    input('Review content.txt before you continue')
    for id in read_data():
        update_webmap_service(
            id
            , old_dns
            , new_dns
            , token
            , portal_url
        )

Labels