返回顶部

对启用同步功能的分支版本化数据执行自动协调与提交操作

在本主题中

  1. 适用场景
  2. 代码示例

在 Standard 或 Advanced 许可等级下可用。

外业工作人员将编辑内容同步到分支版本化中包含的数据后，门户管理员和版本管理员可以使用 Python 脚本自动执行复本版本的定期协调。

此工作流适用于管理使用启用了同步功能的要素服务和分支版本化数据创建的复本版本。

了解有关使用带有分支版本化数据的离线地图的详细信息

注：

此工作流特定于分支版本化。 传统版本化需要不同的管理任务。 如果您使用传统版本化，则还可以使用 Python 自动协调与提交操作。

适用场景

下一部分中的 Python 示例将使用要素服务 ArcGIS REST API 和 ArcPy 地理处理工具来协调和提交具有相应验证属性规则的复本版本。 以下内容将介绍何时应使用代码示例以及示例脚本的功能。

注：

此脚本是具有特定数据设置的工作流示例。 您的工作流可能未遵循此确切的数据模型，但是概念和脚本逻辑仍可用于为您的组织自定义自动化脚本。 例如，您可能不使用验证规则，但是您可以更改脚本以移除验证规则的评估。 另外，在某些情况下，如果使用验证规则，则可能需要在运行评估操作之前协调、运行验证并分别提交每个版本，以确保验证包含所有其他编辑者所做的编辑内容。

先决条件：

在本示例中，数据设置如下：

  • 包含分支版本化数据的要素服务必须启用同步功能，并选择为每个下载的地图创建一个版本选项，这需要 ArcGIS Enterprise 10.8.1 或更高版本。
  • 通过离线使用地图，在服务上创建了复本。
  • 该服务上已启用版本管理服务器功能。
  • 在脚本中为门户用户提供的凭据必须适用于默认门户管理员角色或已授予版本管理权限的自定义角色的成员。
  • 在本示例中，数据还应用了验证属性规则并启用了验证服务器功能。 这不是使用复本版本的要求，但其提供了一种确保数据完整性的方法。

  1. 可以使用 ArcGIS REST API 中的 replicas 资源获取要素服务的复本版本列表。
  2. 可以使用 ArcGIS REST API 中的 versionsInfos 资源获取版本的属性。 借助这些属性，可以创建一个过滤器以仅获取评估日期为空或者修改日期晚于上次评估日期（意味着自上次评估以来已进行了编辑）的版本。 将满足过滤器的版本名称附加到将要进行评估的 listOfVersionsToValidate 列表。
  3. 对于评估列表中的每个版本，可以在 ArcGIS REST API 中使用 evaluate 操作来评估服务上的验证属性规则。 如果评估成功并且未返回任何错误，则该版本已准备好进行协调和提交。 如果评估成功但返回了错误，请勿提交编辑内容；将生成一条消息，以便手动检查和修复错误。
  4. 对于已成功进行评估并且没有任何错误的版本，请使用提交编辑内容的选项运行协调版本工具。
  5. 使用仅对所有复本版本进行协调（不提交）选项运行协调版本工具。

代码示例

以下完整 Python 代码示例将完成以上列出的操作。 该脚本包含生成日志文件的选项，该日志文件可捕获每个已完成操作的输出，并且可以在脚本完成后进行查看。

# Import modules
import arcpy, traceback, urllib, json, urllib.request, urllib.parse, os, urllib.error, datetime

# Overwrite the reconcile log output each time the script is run
arcpy.env.overwriteOutput = True

# Script parameters
serviceName = "MyServiceName"
baseURL = "https://MyServer.MyDomain.com"
portalAdmin = "MyAdminUser"
portalAdminPwd = "MyAdmin.Password"
logFileScript = "C:/Logs/validateRecPostScriptLog.txt"
logfileOutputRecPost = 'C:/Logs/reconcile_log.txt' 

# Choose to output a log file for the script
outputScriptReport = True

# Define functions
def openURL(url, params=None):
    """This function used to open a URL and returns the json response"""
    try:
        request_params = {'f':'pjson'}
        if params:
            request_params.update(params)
        encodedParams = urllib.parse.urlencode(request_params)
        request = urllib.request.urlopen(url, encodedParams.encode('UTF-8'))
        response = request.read()
        json_response = json.loads(response)
        return json_response
    except:
        print (traceback.format_exc())

def versionInfo(versionOwner=""):
    """This function queries the versions owned by the versionOwner.
    It returns a list of dictionaries."""
    vmsUrlinfo = "{}/server/rest/services/{}/VersionManagementServer/versionInfos?&ownerFilter={}&includeHidden=&f=json&token={}".format(baseURL, serviceName, versionOwner, token)
    response = openURL(vmsUrlinfo)
    if response['success'] == True:
        versionsDict = response['versions']
        return versionsDict
    else:
        return("Unable to get version info")

def evaluateUrl(validationUrl, validate_params):
    """This function runs evaluate on the validation server
    It returns the json response."""
    evalJsonResp = openURL(validationUrl, validate_params)
    if evalJsonResp['success'] == False:
        return [False, evalJsonResp]
    else:
        return [True, evalJsonResp]

def generateMessage(msg, print_statement=True):
    """This function generates messages as the script runs. If print_statement
    is set to True, print to the screen. If outputScriptReport is set to true,
    write the message to the logfile"""
    if outputScriptReport == True:
        with open(logFileScript, 'a') as log_file:
            log_file.write(msg + "\n")
    if print_statement == True:
        print(msg)

def recPostVersions(versionList, post):
    """This function runs the Reconcile Versions GP tool to reconcile
    and optionally post to the feature service"""
    if post == True:
        postVersion = "POST"
    elif post == False:
        postVersion = "NO_POST"
    # Reconcile and post the replica versions 
    # This tool is set to abort if there are conflicts and detects conflicts by object
    arcpy.ReconcileVersions_management(featureService,
                                        'ALL_VERSIONS',
                                        'sde.DEFAULT',
                                         versionList,
                                        'NO_LOCK_ACQUIRED',
                                        'ABORT_CONFLICTS',
                                        'BY_OBJECT',
                                        'FAVOR_EDIT_VERSION',
                                        postVersion,
                                        'KEEP_VERSION',
                                        logfileOutputRecPost)
    generateMessage(arcpy.GetMessages()+"\n")
    
# Start execution
generateMessage('Starting Validation/Reconcile/Post Automation Script... {:%Y-%b-%d %H:%M:%S}\n'.format(datetime.datetime.now()))

# Sign in to ArcGIS Enterprise    
signIntoPortal = arcpy.SignInToPortal(baseURL+"/portal", portalAdmin, portalAdminPwd)
generateMessage("Signed into ArcGIS Enterprise {} as user {}".format(baseURL+"/portal", portalAdmin))

# Get the token returned by the SignIntoPortal arcpy function to use for making REST requests
token = signIntoPortal['token']

# Build the feature service URL
featureService = "{}/server/rest/services/{}/FeatureServer".format(baseURL, serviceName)

# Get a list of the replica versions from the REST endpoint
listOfRepVersions = []
replicaVersionsURL = featureService + "/replicas?returnVersion=true&f=pjson"
repVersionsJson = openURL(replicaVersionsURL, signIntoPortal)
for repVersion in repVersionsJson:
    versionName = repVersion['replicaVersion']
    listOfRepVersions.append(versionName)
    
# Create an empty list to append version names to validate
listOfVersionsToValidate = []

# Iterate through each version returned by the versionInfo() function to find 
# the versions that need to be validated that are also in the listOfRepVersions list
for version in versionInfo():
    print("")    
    # Parse the version info response, which is a python dictionary/json
    # If the version name is sde.DEFAULT, pass since we do not want to evaluate the default version
    if version['versionName'] == "sde.DEFAULT":
        pass
    # If the modifiedDate property is null, pass
    elif version['modifiedDate'] == "None":
        pass
    # If the evaluation date is null, append the version name to the list to listOfVersions to be evaluated
    elif version['evaluationDate'] == None:
        if version['versionName'] in listOfRepVersions:
            listOfVersionsToValidate.append(version['versionName'])
    # If the evaluation date is not null, but it has been modifed since the last evaluation, add it to the list to be validated
    elif version['evaluationDate'] != None and version['modifiedDate'] > version['evaluationDate']:
        if version['versionName'] in listOfRepVersions:
            listOfVersionsToValidate.append(version['versionName'])
    # If none of these conditions are met
    else:
        generateMessage("Version {} will not be validated.".format(version['versionName']))
            
# Validate versions
generateMessage('The following versions will be validated: {}\n'.format(listOfVersionsToValidate))

# Create lists to contain versions where the validation passed or failed
failEval = []
passEval = []

# For each version in the list of versions, build the json request needed to validate
for i in listOfVersionsToValidate:
    validate_params = { "gdbVersion": i,
             "sessionId": "",
             "evaluationArea": "",
             "changesInVersion": "true",
             "selection": "",
             "evaluationType": '["validationRules"]',
             "returnEdits": "true",
             "async": "false",
             "f": "pjson",
             "token": token
    }
    # Build the REST URL used to validate the service
    validationUrl = baseURL + "/server/rest/services/"+ serviceName +"/ValidationServer/evaluate"
    
    # Call the evalVersion() function to validate the version
    evalVersion = evaluateUrl(validationUrl, validate_params)
    
    # If the evaluate failed, append to the failEval list
    if evalVersion[0] == False:
        generateMessage("Evalution of version {} failed".format(i))
        generateMessage(str(evalVersion[1]))
        failEval.append(i)
        
    # If the evaluate passed, check to see if errors were returned    
    elif evalVersion[0] == True:
        # If errors are returned, only reconcile this version
        if evalVersion[1]['errorsIdentified'] != 0:
            generateMessage("{} Errors were identified in version {}.\nThe version will be reconciled but will not be posted.\n".format((str(evalVersion[1]['errorsIdentified'])),i))
            generateMessage(str(evalVersion[1]), False)
        # If errors were not found this version can be posted
        else:
            generateMessage("Evaluation of version {} passed with no errors identified.\nThis version will be reconciled and posted.\n".format(i))
            generateMessage(str(evalVersion[1]))
            passEval.append(i)

# The versions that passed validation should be reconciled/posted
generateMessage('\nThe following versions passed validation and will be reconciled and posted: {}\n'.format(passEval))

# Run recPostVersions on the list of versions that passed evaluation with the option to post
recPostVersions(passEval, True)

# Open the reconcile log file and append the results to our report
with open(logfileOutputRecPost, 'r') as openRecLog:
    generateMessage(openRecLog.read(), False)
    
# Run recPostVersions with the option to reconcile all replica versions, no post
recPostVersions(listOfRepVersions, False)

# Open the reconcile log file and append the results to our report
with open(logfileOutputRecPost, 'r') as openRecLog:
    generateMessage(openRecLog.read(), False)

# Script execution complete
generateMessage('Validate, Reconcile, & Post Script Complete.')

相关主题

  • 连接到分支版本
  • 管理分支版本
  • 查看分支版本变更
  • 协调和提交对分支版本的编辑
  • 管理分支版本冲突

有关此主题的反馈?

在本主题中
  1. 适用场景
  2. 代码示例