当前位置: 首页 > news >正文

工具层handle_replace

 工具层handle_replace

1. 删除换行符和空格

自定义方法名所用模块/函数解决的问题
__handle_str

for循环

replace函数

数据清洗:清理请求参数中的 \n 和空格,确保JSON格式合法,避免解析错误

2. 提取需替换的参数名

自定义方法名所用模块/函数解决的问题
__get_replace_keys
  • re模块:Python的正则表达式操作模块
  • findall函数:返回字符串中所有匹配正则的子串(列表形式)
  • 正则表达式#(\w.+?)#:匹配任意字符至少一次,但遇到第一个结束标记(如 #)就停止,​避免跨越多余内容;
  1. \w → 匹配字母/数字/下划线
  2. .+? → 非贪婪匹配任意字符(至少一个)
通过正则 #(\w.+?)# 找出用#包裹起来的参数名并提取出来识别为动态参数(如 #user_name#

3. 动态设置类属性值

自定义方法名条件语句所用模块/函数
__get_data_and_set_attributeif key in user_info:取 user_info[key] 的值,并设置为HandleAttr 类的属性setattr():Python 内置函数,动态设置对象的属性值
elif key == "sessionUUID":生成形如 a1b2c3d4-1234-5678-90ab-cdef12345678 的UUID并存入类属性uuid.uuid4():uuid 标准模块,生成全局唯一标识符(UUID 格式字符串)
elif key == "partyCode" or "bizPayNo":生成形如 1620000000000 的毫秒级时间戳并存入类属性(若属性不存在)time.time():time 标准模块,生成唯一业务编码(如 partyCodebizPayNo),避免订单号或流水号重复
elif key == "mobile":生成形如 13812345678 的手机号并存入类属性(若属性不存在)handlePhone.get_phone():自定义类方法,生成未注册的手机号

 4. SQL参数替换

逻辑步骤自定义方法名所用模块/函数解决的问题
replace_sqlre.findall()
getattr()
替换SQL中的动态参数(如 #partyCode#),生成可执行SQL语句
5. 执行SQL并注入属性__execute_sql_and_setattrmysql.get_data_dict()
setattr()
执行预处理后的SQL,将查询结果映射到类属性(如用户ID、订单号),供后续接口请求使用
6. 整合属性设置流程__set_attributeast.literal_eval()
self.replace_sql()
self.__execute_sql_and_setattr
统一处理属性来源:
- 配置数据
- SQL查询结果
- 动态生成值,确保参数替换前数据就绪
7. 主流程:生成最终请求体replace_dataast.literal_eval()
self.__handle_str
self.__set_attribute
入口方法:
1. 清洗原始数据 → 2. 提取参数名 → 3. 设置属性 → 4. 替换参数 → 5. 生成JSON请求体
{
    #时间戳
    "t": "",
    #用户名
    "principal":"#user_name#",
    #密码
    "credentials":"123456a",
    #类似于access-token
    "sessionUUID": session_uuid,
    #验证码
    "imageCode":"lemon"
}
逻辑:
1、传入请求参数data,excel中读取到的请求参数
2、删除请求参数中的换行符和空格
3、获取需要替换的参数名称
4、通过参数名称获取到参数的值,获取到值统一设置为类属性
5、将参数的值替换掉参数名称和#号
"""

import re  #正则表达式会用
import ast
import uuid
import time


from tools.handle_path import data_dir
from tools.handle_excel import HandleExcel
from conf.setting import user_info
from tools.handle_attribute import HandleAttr
from tools.handle_phone import HandlePhone
from tools.handle_db import mysql
from tools.handle_logs import myLog



class HandleReplace:
    def __init__(self):
        self.handle_phone =  HandlePhone()

    # 删除换行符和空格键
    # data(str): excel中获取的请求参数
    # return data:去掉空格后的请求参数
    def __handle_str(self,data:str):
        for str_data in ["\n"," "]:
            data = data.replace(str_data,"")
        return data

    # 获取需要替换参数的名称
    def __get_replace_keys(self,data:str):
        #"#(\w.+?)#"正则表达式
        key_list = re.findall("#(\w.+?)#",data)
        return key_list

    # 根据数据的来源,获取数据,设置为类属性
        def __get_data_and_set_attribute(self,key:str,user_info):
        if key in user_info:
            # 取出来设置为类属性
            setattr(HandleAttr,key,str(user_info[key]))
        elif key == "sessionUUID":
            # 特殊处理,通过脚本生成sessionUUID
            setattr(HandleAttr,key,str(uuid.uuid4()))
        # 添加产品的唯一字段partyCode,单独处理
        elif key == "partyCode" or key == "bizPayNo":
        # 添加产品数据库断言的时候,partyCode要使用参数替换时候的partyCode,不能在sql语句替换的时候再去生产partyCode
            if hasattr(HandleAttr,key):
                pass
            else:
                setattr(HandleAttr,key,str(int(time.time()*1000)))
        # 如果是手机号(注册用户),单独生产未注册的手机号
        elif key == "mobile":
            if hasattr(HandleAttr, key):
                pass
            else:
                print("生成为注册的手机号")
                phone = self.handle_phone.get_phone()
                setattr(HandleAttr,key,str(phone))

  
    #替换sql语句中的参数
    def replace_sql(self,sql,assert_db_info):
        # 获取需要替换的参数的名称,返回类型list
        key_list = self.__get_replace_keys(data=sql)
        if len(key_list)>0:
            # 获取key的值,然后设置为类属性
            for key in key_list:
                self.__get_data_and_set_attribute(key=key,user_info=assert_db_info)
            #从类属性中查询到key的值,然后替换
            for key in key_list:
                sql = sql.replace(f"#{key}#", str(getattr(HandleAttr, key)))
            return sql
        else:
            print("不需要替换sql")
            return sql


    # 执行sql语句,将sql语句查询出来的key,val设置为类属性的key,val
    def __execute_sql_and_setattr(self,sql):
        # 调sql执行方法,获取到数据
        # 拿到数据并设置为属性
        result = mysql.get_data_dict(sql=sql)
        for dict_data in result:
            for key,value in dict_data.items():
                setattr(HandleAttr,key,str(value))


    # 调用数据来源设置属性方法,设置类属性
    def __set_attribute(self,key_list,setup_sql,setup_sql_info):
        """
        : param setup_sql (str):接口执行之前,需要从通过sql查询来获取替换数据,可以兼容多条[sql1,sql2]
        :param key_list: ["user_name","session_uuid"]
        :return:
        """
        if setup_sql:
            for sql in ast.literal_eval(setup_sql):
                # 替换sql语句中需要替换的参数
                new_sql = self.replace_sql(sql=sql,assert_db_info=setup_sql_info)
                # 执行sql语句,并设置为属性
                self.__execute_sql_and_setattr(sql=new_sql)

        # 只对excel请求参数进行属性设置
        for key in key_list:
            #key的值的来源配置文件
            #通过脚本生成的数据
            #响应结果
            #数据库
            self.__get_data_and_set_attribute(key=key,user_info=user_info)

    # 获取参数,不同的数据,来源是不是不一样,然后设置为类属性
    # data (str): excel中读取到的请求参数
    def replace_data(self,data,setup_sql,setup_sql_info):
        try:
            myLog.info(msg="replace_data参数替换入参:")
            myLog.info(msg=f"data:{data},{type(data)}")
            myLog.info(msg=f"setup_sql:{setup_sql},{type(setup_sql)}")
            myLog.info(msg=f"setup_sql_info:{setup_sql_info},{type(setup_sql_info)}")
            if data:
                # 删除换行符和空格键,返回类型str
                new_data = self.__handle_str(data=data)
                # 获取需要替换的参数的名称,返回类型list
                key_list = self.__get_replace_keys(data=new_data)
                myLog.info(msg=f"需要替换的参数名称:{key_list}")
                if len(key_list)>0:
                    # 当 len(key_list)>0 说明 参数里面需要替换数据
                    # 获取参数,并设置为类属性,无返回值
                    self.__set_attribute(key_list=key_list,setup_sql=setup_sql,setup_sql_info=setup_sql_info)
                    # 从类属性中通过key_list里面的key获取值,然后替换请求参数
                    for key in key_list:
                        new_data = new_data.replace(f"#{key}#",str(getattr(HandleAttr,key)))
                    new_data = ast.literal_eval(new_data)
                    new_data["t"] = int(time.time()*1000)
                    myLog.info(msg=f"参数替换完成后返回数据:{new_data},{type(new_data)}")
                    return new_data
                else:
                    new_data = ast.literal_eval(new_data)
                    new_data["t"] = int(time.time() * 1000)
                    myLog.info(msg=f"不需要参数替换,返回数据:{new_data},{type(new_data)}")
                    return new_data
            else:
                myLog.info(msg="没有请求参数,不需要做任何事情,返回空的dict")
                return {}
        except Exception as e:
            myLog.error(msg=f"replace_data参数替换方法执行报错,错误原因{e}")
            myLog.exception(e) #将错误信息收集到日志文件
            raise TypeError #手动抛出异常给框架


if __name__ == '__main__':
    print(case_list[0])
    cl = HandleReplace()
    cl.replace_data(data=case_list[0]["data"])

未待完续! 

相关文章:

  • Sensodrive机器人力控关节模组SensoJoint在海洋垃圾清理机器人中的拓展应用
  • vector容器以及deque
  • 嵌入式c学习六
  • 硬件基础(5):(2)二极管分类
  • 如何编译鲁班猫(LubanCat 1N)固件
  • HTTP/HTTPS
  • 前端小食堂 | Day18 - 身份认证の八卦阵
  • LORA 中的 梯度外积是什么意思; 方差和协方差的实际含义:衡量变量的离散程度和变量间的线性相关性
  • 信息系统运行管理员教程4--信息系统软件运维
  • 【Linux———线程精讲】
  • electron js node vscode 调试electron
  • 实验4 Vue.js路由实验
  • Can通信流程
  • 【Linux网络-数据链路层】以太网(以太网帧格式|MAC地址+模拟一次性局域网通信+MTU)+ARP协议
  • 09 python函数(上)
  • UE学习记录part7
  • 免密登录远程服务器shell脚本
  • 【Linux】:守护进程化
  • 工具层handle_excel
  • Java-泛型总结
  • 马上评|“AI神医宇宙”欺诈,连演员都不请了
  • 气温“过山车”现象未来或更频繁且更剧烈
  • 加拿大警方:已确认有9人在温哥华驾车撞人事件中遇难
  • 最高法报告重申保护创新主体权益:加大侵权损害赔偿力度
  • 铜钴巨头洛阳钼业一季度净利润同比大增九成,最新宣布超30亿元收购黄金资产
  • 全国首例!上市公司董监高未履行公开增持承诺,投资者起诉获赔