本文为博主原创，未经授权，严禁转载及使用。
本文链接：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]
all_data_list_2 = [adl[0] for adl in all_data_list if len(str(adl[0])) != 11]
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)
从没想到 最初的txt文件中的用户信息 会有坑；换个角度想想：同一用户手机号，多条user-id，应该是注销后，再次注册；非11位手机号，是港澳台、海外用户；
最后生成的txt：
本文链接：https://blog.csdn.net/zyooooxie/article/details/127645678
个人博客 https://blog.csdn.net/zyooooxie