Python脚本之准备测试环境的用户数据

news2024/9/27 21:28:23

本文为博主原创，未经授权，严禁转载及使用。
本文链接：https://blog.csdn.net/zyooooxie/article/details/127645678

这期是讲述下我准备测试环境用户数据的经历。

【实际这篇博客推迟发布N个月】

个人博客：https://blog.csdn.net/zyooooxie

【以下所有内容仅为个人项目经历，如有不同，纯属正常】

经历

之前测试环境的用户我造的时候 user_id是Faker生成的；union_id直接取了 user_id的前15位，有天我测缓存时，发现有问题：union_id出现了重复的情况，所以我清掉所有相关记录后，想着重新造一批数据：还是用测试环境本就有的用户 + 调整union_id后的值。

就找大佬捞了 100w用户信息，只要2个字段：手机号（明文）+ user_id。因为我这边项目要用到加密手机号（需要请求一个加密接口），我就写了脚本来跑： https://blog.csdn.net/zyooooxie/article/details/100106284。


# 当晚跑了后，我查看本地日志，

# 第一条是 读取txt用户信息，放入队列后，队列的长度；
19:51:08-wc_performance_test.py-main_1101-line:96-INFO:1000000

# 第二条 是成功加密，已经写入新的txt_1 的长度；
22:38:12-wc_performance_test.py-write_txt-line:210-INFO:998746

# 第三条是没有拿到响应的（我设置 超时时间为0.5s），写入新的txt_2 的长度。
22:38:13-wc_performance_test.py-write_txt-line:210-INFO:1254

# 就 重新跑 失败的这些，第一条是 新读取txt_2 用户信息，放入队列后，队列的长度；
22:47:03-wc_performance_test.py-main_1101-line:96-INFO:1254

# 第二条 到 第五条是 每次起100个线程，跑过程中 队列的长度 逐渐变小，当队列长度小于线程数，我是直接将其扔到 失败list； 
22:47:10-wc_performance_test.py-main_1101-line:96-INFO:354
22:47:10-wc_performance_test.py-main_1101-line:96-INFO:254
22:47:11-wc_performance_test.py-main_1101-line:96-INFO:154
22:47:11-wc_performance_test.py-main_1101-line:96-INFO:54
22:47:11-wc_performance_test.py-main_1101-line:102-INFO:小于线程数：54

# 第六条是 成功加密，已经写入新的txt_3 的长度；
22:47:11-wc_performance_test.py-write_txt-line:210-INFO:994

# 最后一条是 没有拿到响应的（我设置 超时时间为0.5s）+ 小于线程数时的，写入新的txt_4 的长度。
22:47:11-wc_performance_test.py-write_txt-line:210-INFO:260

此时，我将 txt_3的内容复制到 txt_1时，复制完了，我看 整个txt_1的行数超了100w，这不扯呢？明明还有260条没加密的，咋可能就超了？

发现、解决问题

去除我不小心多复制的记录

Counter类

ABC = [(k, v) for k, v in abc.items() if v != 1]

在这里插入图片描述

去除重复手机号

ABC = [d[0] for d in ABC]

在这里插入图片描述

去除非11位手机号

all_data_list_2 = [adl[0] for adl in all_data_list if len(str(adl[0])) != 11]

在这里插入图片描述

去除异常user-id

all_data_list_3 = [adl[1] for adl in all_data_list if len(adl[1]) <= 15]

在这里插入图片描述

代码

"""
@blog: https://blog.csdn.net/zyooooxie
@qq: 153132336
"""

def test_1101():
    """
    用户信息txt文件 去重
    :return:
    """
    
    all_data_list = read_txt_func(file=r'D:\100W用户4.txt', get_all='True')
    Log.info(len(all_data_list))

    all_data_list_2 = [adl[0] for adl in all_data_list if len(str(adl[0])) != 11]
    Log.info(len(all_data_list_2))
    Log.info(all_data_list_2)

    # all_data_list_2 = dict.fromkeys(all_data_list_2)

    all_data_list_3 = [adl[1] for adl in all_data_list if len(adl[1]) <= 15]
    Log.info(len(all_data_list_3))
    Log.info(all_data_list_3)

    # all_data_list_2 =[(str(adl[0]), *adl[1:]) for adl in all_data_list if adl[0] not in all_data_list_2]
    # Log.info(len(all_data_list_2))

    all_data_list_1 = [adl[-1] for adl in all_data_list if adl[-1].endswith('%3D')]
    Log.info(len(all_data_list_1))

    # abc = Counter(all_data_list_1)

    # ABC = [(k, v) for k, v in abc.items() if v != 1]
    # Log.info(ABC)

    # ABC = [k for k, v in abc.items() if v == 1]
    # ABC = {k: None for k, v in abc.items() if v == 1}

    # all_data_list_2 = [tuple(adl) for adl in all_data_list if adl[0] in ABC]
    # all_data_list_2 = [(str(adl[0]), *adl[1:]) for adl in all_data_list if adl[0] in ABC]
    # Log.info(len(all_data_list_2))

    # ABC = [d[0] for d in ABC]
    #
    # ABC.extend([k for k, v in abc.items() if v == 1])
    # Log.info(len(ABC))
    # ABC = [(str(d[0]), *d[1:]) for d in ABC]
    #
    # write_txt(all_data_list_2)


def write_txt(data: Union[str, list]):
    """
    将加密结果 写入文件（可一条一条写入，也可一次性全部写入）
    :param data:单独一条记录str or  全部记录list
    :return:
    """
    file = r'D:\param_{}.txt'.format(random.randint(1, 99999))

    ls = os.linesep

    with open(file, mode='a', encoding='utf-8') as f:
        if isinstance(data, str):
            # f.write(''.join([data, ls]))
            f.write(''.join([data, '\n']))

        elif isinstance(data, list):
            Log.info(len(data))

            data = [''.join([d, '\n']) if type(d) == str else ''.join([','.join([*d]), '\n']) for d in data]

            f.writelines(data)