我使用标记工具来训练自定义模型。该工具提供了使用它的python脚本。它在终端上运行良好。现在我想把它放在一个逻辑应用程序中的Azure函数中。pdf电子邮件附件将触发表单分析程序模型，然后我将解析JSON响应。到目前为止，我还没能：
- 在Azure函数内成功部署所提供的脚本
- 在逻辑应用程序中使用它来识别我将存入blob或通过电子邮件发送的每一个pdf
这是Azure表单识别器工具提供的脚本：
########### Python Form Recognizer Async Analyze #############
import json
import time
import getopt
import sys
import os
from requests import get, post
def main(argv):
input_file, output_file, file_type = getArguments(argv)
runAnalysis(input_file, output_file, file_type)
def runAnalysis(input_file, output_file, file_type):
# Endpoint URL
endpoint = r"##################################/"
# Subscription Key
apim_key = "####################################"
# Model ID
model_id = "######################################"
# API version
API_version = "v2.1-preview.3"
post_url = endpoint + "/formrecognizer/%s/custom/models/%s/analyze" % (API_version, model_id)
params = {
"includeTextDetails": True
}
headers = {
# Request headers
'Content-Type': file_type,
'Ocp-Apim-Subscription-Key': apim_key,
}
try:
with open(input_file, "rb") as f:
data_bytes = f.read()
except IOError:
print("Inputfile not accessible.")
sys.exit(2)
try:
print('Initiating analysis...')
resp = post(url = post_url, data = data_bytes, headers = headers, params = params)
if resp.status_code != 202:
print("POST analyze failed:n%s" % json.dumps(resp.json()))
quit()
print("POST analyze succeeded:n%s" % resp.headers)
print
get_url = resp.headers["operation-location"]
except Exception as e:
print("POST analyze failed:n%s" % str(e))
quit()
n_tries = 15
n_try = 0
wait_sec = 5
max_wait_sec = 60
print()
print('Getting analysis results...')
while n_try < n_tries:
try:
resp = get(url = get_url, headers = {"Ocp-Apim-Subscription-Key": apim_key})
resp_json = resp.json()
if resp.status_code != 200:
print("GET analyze results failed:n%s" % json.dumps(resp_json))
quit()
status = resp_json["status"]
if status == "succeeded":
if output_file:
with open(output_file, 'w') as outfile:
json.dump(resp_json, outfile, indent=2, sort_keys=True)
print("Analysis succeeded:n%s" % json.dumps(resp_json, indent=2, sort_keys=True))
quit()
if status == "failed":
print("Analysis failed:n%s" % json.dumps(resp_json))
quit()
# Analysis still running. Wait and retry.
time.sleep(wait_sec)
n_try += 1
wait_sec = min(2*wait_sec, max_wait_sec)
except Exception as e:
msg = "GET analyze results failed:n%s" % str(e)
print(msg)
quit()
print("Analyze operation did not complete within the allocated time.")
def getArguments(argv):
input_file = ''
file_type = ''
output_file = ''
try:
opts, args = getopt.gnu_getopt(argv, "ht:o:", [])
except getopt.GetoptError:
printCommandDescription(2)
for opt, arg in opts:
if opt == '-h':
printCommandDescription()
if len(args) != 1:
printCommandDescription()
else:
input_file = args[0]
for opt, arg in opts:
if opt == '-t':
if arg not in ('application/pdf', 'image/jpeg', 'image/png', 'image/tiff', 'image/bmp'):
print('Type ' + file_type + ' not supported')
sys.exit()
else:
file_type = arg
if opt == '-o':
output_file = arg
try:
open(output_file, 'a')
except IOError:
print("Output file not creatable")
sys.exit(2)
if not file_type:
file_type = inferrType(input_file)
return (input_file, output_file, file_type)
def inferrType(input_file):
filename, file_extension = os.path.splitext(input_file)
if file_extension == '':
print('File extension could not be inferred from inputfile. Provide type as an argument.')
sys.exit()
elif file_extension == '.pdf':
return 'application/pdf'
elif file_extension == '.jpeg':
return 'image/jpeg'
elif file_extension == '.bmp':
return 'image/bmp'
elif file_extension == '.png':
return 'image/png'
elif file_extension == '.tiff':
return 'image/tiff'
else:
print('File extension ' + file_extension + ' not supported')
sys.exit()
def printCommandDescription(exit_status=0):
print('analyze.py <inputfile> [-t <type>] [-o <outputfile>]')
print
print('If type option is not provided, type will be inferred from file extension.')
sys.exit(exit_status)
if __name__ == '__main__':
main(sys.argv[1:])
根据您的需求，您需要创建azure函数，并在函数中编写经过一些修改的python代码。
首先在VS代码中用python创建azure函数。您创建的python函数应该如下所示：
def main(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
...............
然后，您需要将上面提供的python代码编写到经过一些修改的函数中。将代码融合到函数HttpRequest代码体中。
之后，将函数从local部署到azure。
然后，您可以调用逻辑应用程序中的函数。