QwenModel/text_analysis/tools/bak/tool_1107_final.py

#coding:utf8
import re
from jsonpath_ng import parse
import json
import traceback

def parse_data(raw_data,para):
    all_result = raw_data['data']
    param_split = str(para).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={}
    input=inputdata["input"]
    data=inputdata["data"]
    prompt=input["prompt"]
    if_data=re.findall("@@(.*?)@@",prompt)
    # if_user=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!=[] :
        for rule in if_data:
            try:
                if "#json#" in rule:
                    parm = rule.split("#json#")
                    data1 = parse_data(inputdata, parm[0])
                    data1_json = json.loads(data1)
                    expr = parse(parm[1])
                    result = str([match.value for match in expr.find(data1_json)][0])
                    rep = "@@{}@@".format(rule)
                    # 增加转义字符
                    rep_escaped = re.escape(rep)
                    prompt = re.sub(rep_escaped, result, prompt)
                elif ":" in rule:
                    result = parse_data(inputdata, rule)
                    rep = "@@{}@@".format(rule)
                    rep_escaped = re.escape(rep)
                    prompt = re.sub(rep_escaped, result, prompt)
                else:
                    if rule in data.keys():
                        tmp1=data[rule]
                        rep = "@@{}@@".format(rule)
                        prompt=re.sub(rep,tmp1,prompt)
            except:
                # print(traceback.format_exc())
                rep = "@@{}@@".format(rule)
                prompt = prompt.replace(rep,'')
                logging.info("动态字段获取数据失败。{}-{}".format(rule, traceback.format_exc()))
    logging.info("拼接后的问题：{}".format(prompt))
    res["prompt"]=prompt
    res["authorization"]=input["authorization"]
    res["model"]=input["model"]
    res["temperature"]=input["temperature"]
    res["authorization"]=input["authorization"]
    res["top_p"]=input["top_p"]
    res["n"]=input["n"]
    return res

# 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"
    },
    "address":"http://172.18.1.181:9011/chatGpt/",
    "input":{
        "authorization":"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",
        "top_p":"1",
        "temperature":"1",
        "model":"gpt-3.5-turbo-16k",
        "prompt":"根据下面内容：@@1_Youtube采集:$['content']#json#$['test1']@@。生成一条@@1_Youtube采集:$['Count']@@字的关于中国正面的新闻，标题用title，内容用content，以json格式输出。",
        "n":"1"
    },
    "data":{
        "1_Youtube采集":"{\"isDownload\":\"true\",\"content\":\"{\\\"test1\\\":\\\"22222\\\"}\",\"Count\":\"555\"}"

    },
    "created":1691004265000,
    "module":"ChatGPT",
    "start_tag":"false",
    "multi_branch":0,
    "last_edit":1698927821000,
    "next_app_id":[
        {
            "start_id":316,
            "edge_id":200,
            "end_id":317
        }
    ],
    "transfer_id":3,
    "version":1,
    "blueprint_id":12,
    "scenes_id":12,
    "scenario":{
        "dataloss":1,
        "autoCommitTriggerLast":1,
        "maxErrors":3,
        "autoCommit":1,
        "freshVariables":1
    },
    "wait_condition":[

    ],
    "scheduling":{
        "interval":-1,
        "type":"single"
    },
    "name":"正面引导",
    "businessKey":"78278a5168e45304",
    "id":316,
    "position":[
        100,
        200
    ],
    "describe":"正面引导"
}
    a=get_content(inputdata,"")
    print(a)