|
|
#coding:utf8import reimport jsonfrom jsonpath_ng import jsonpath, parse
def parse_data(raw_data,field): parm = field.split("#") all_result = raw_data['data'] if "json" in field: param_split = str(parm[0]).split(":") datasourcestr = all_result[param_split[0]] datasource = json.loads(datasourcestr) # 创建 JsonPath 表达式对象 expr = parse(param_split[1]) # 使用表达式来选择 JSON 元素 match = [match.value for match in expr.find(datasource)] data1_json = json.loads(match[0]) expr = parse(parm[2]) match = [match.value for match in expr.find(data1_json)] result = match[0] else: all_result = raw_data['data'] param_split = str(parm[0]).split(":") datasourcestr = all_result[param_split[0]] datasource = json.loads(datasourcestr) # 创建 JsonPath 表达式对象 expr = parse(param_split[1]) # 使用表达式来选择 JSON 元素 match = [match.value for match in expr.find(datasource)] result = match[0] return result
def promptPro(prompt,allPromptSim): p="" for promptSim in allPromptSim["documents"][0]: p+=promptSim p+="\n" if len(prompt)+len(p)>50000: p=p[0:50000-len(prompt)] res = "已知信息：\n{}\n参考上述已知信息回答问题：\n{}".format(p, prompt) return res
# def parse_data(raw_data,url):# all_result = raw_data['data']# param_split = str(url).split(":")# datasourcestr = all_result[param_split[0]]# datasource = json.loads(datasourcestr)# # 创建 JsonPath 表达式对象# expr = parse(param_split[1])# # 使用表达式来选择 JSON 元素# match = [match.value for match in expr.find(datasource)]# val = match[0]# return val
def get_content(inputdata,logging): """
重新组装参数 :param inputdata:原json数据 :return: 组装的prompt及其他参数 """
res={} admin=inputdata["metadata"]["admin"] data=inputdata["data"] prompt=admin["prompt"] if_user=re.findall("{{(.*)}}",prompt) if_data=re.findall("@@(.*)@@",prompt) if if_user != []: user_data=inputdata["metadata"]["user"] if if_user[0] in user_data.keys(): tmp=user_data[if_user[0]] prompt=re.sub("{{(.*)}}",tmp,prompt) if if_data!=[] and if_data[0] in data.keys(): tmp1=data[if_data[0]] prompt=re.sub("@@(.*)@@",tmp1,prompt) res["prompt"]=prompt res["authorization"]=admin["authorization"] res["model"]=admin["model"] res["temperature"]=admin["temperature"] res["authorization"]=admin["authorization"] res["top_p"]=admin["top_p"] res["n"]=admin["n"] return res
if __name__=="__main__": inputdata={ "output":{ "id":"id", "content":"content" }, "input":{ "chunkSize":"1_Youtube采集:$['file']#json#$['chunkSize']", "content":"1_Youtube采集:$['file']#json#$['content']", "fieldName":"1_Youtube采集:$['file']#json#$['fieldName']" }, "data":{ "1_Youtube采集":"{\"file\": \"{\\\"chunkSize\\\": \\\"500\\\", \\\"content\\\": \\\"广东省人民政府关于公布广东省行政许可事项清单（2023年版）的通知，广东省人民政府关于公布广东省行政许可事项 清单（2023年版）的通知 粤府〔2023〕101号 各地级以上市人民政府，各县（市、区）人民政府，省政府各部门、各直属机构，中直驻粤有关单位： 为深入贯彻习近平新时代中国特色社会主义思想，全面贯彻党的二十大精神，认真落实党中央、国务院决策部署，按照《国务院办公厅关于公布〈法律、行政法规、国务院决定设定的行政许可事项清单（2023年版）〉的通知》（国办发〔2023〕5号）要求，结合我省行政审批制度改革情况，省政府办公厅会同有关部门对行政许可事项调整情况进行汇总，修订形成《广东省行政许可事项清单（2023年版）》，现予公布。 各地各部门要认真对照《广东省行政许可事项清单（2023年版）》及时调整完善本地本系统行政许可事项清单和实施规范，严格依照清单实施行政许可，不断强化实施情况动态评估和全程监督，扎实有效做好全面实行行政许可事项清单管理工作。 广东省人民政府 2023年12月19日 附件：广东省行政许可事项清单（2023年版）\\\", \\\"fieldName\\\": \\\"field_test\\\", \\\"fileType\\\": \\\"mp4\\\", \\\"fileUrl\\\": \\\"http://172.18.1.130:9985/group33/default/20230828/15/48/1/lKTZNen6aak.mp4\\\", \\\"ossPath\\\": \\\"/group33/default/20230828/15/48/1/lKTZNen6aak.mp4\\\"}\"}" }, "created":1691004265000, "module":"ASR", "start_tag":"false", "multi_branch":0, "last_edit":1693417201000, "next_app_id":[ { "start_id":154, "edge_id":75, "end_id":155 } ], "transfer_id":3, "version":1, "blueprint_id":4, "scenes_id":5, "scenario":{ "dataloss":1, "autoCommitTriggerLast":1, "maxErrors":3, "autoCommit":1, "freshVariables":1 }, "wait_condition":[
], "scheduling":{ "interval":-1, "type":"single" }, "name":"ASR", "businessKey":"19615b029da477fb", "id":154, "position":[ 100, 200 ], "describe":"ASR识别"} a=parse_data(inputdata,inputdata["input"]["fieldName"]) print(a)
|
