最近工作中有個數據傳輸接口需要測試,
業務需求:
通過核對接口接收的數據條目數、是否等於接口傳輸條目數,印證數據接口的正確性。
前提條件:
本接口是實時傳輸。
接口每次傳輸的數據是全量數據。
解決方法:
因為是全量傳輸,所以需要根據傳輸時間對接收數據進行篩選,
把接口當次傳輸的數據找出來,並進行數據條目求和計算。
看接收數是否和傳輸數一致。
筆者嘗試用非正則、正則表達式兩種方法分別解決此問題。
首先,需要對接收的數據進行格式處理:
1、 獲取傳輸數據
通過開發給的接口URL,在網頁上得到接口傳輸的數據信息。復制粘貼該頁面所有數據到后綴為txt的文本文檔中。
如圖顯示:
得到的接口傳輸的數據,並沒有根據某個字段信息進行排序,所以數據顯示沒有什么規律性。
2、 分析數據格式
數據樣式如下:
原數據較長且比較復雜,不利於閱讀和梳理思路,所以筆者精簡了數據,只留下兩條,並用紅色框拆解出來數據的格式。
可以看出來,數據是按照json串的格式進行傳輸的。美化后,顯示如下。看着就更清晰了。
3、 拆解數據,把源數據變成python中的字典數據類型
將json串轉換成dict類型,需要引入一個json包
使用包中的json.loads的方法,將數據轉化成dict類型。代碼如下:
import json
with open("e:\\test\\sourcedata.txt",encoding="utf-8") as fp:
content = fp.read()
print(type(content)) #查看讀取json串后的數據類型
data_dict = json.loads(content)
print(type(data_dict)) #查看處理json串后的數據類型
執行結果如下:
然后把得到的data_dict存放到新的文檔loadsdata.txt中,代碼如下:
with open("e:\\test\\loadsdata.txt",'w',encoding="utf-8") as fp2:
fp2.write(str(data_dict))
print(type(data_dict)) #查看存放在新文檔后的數據類型
執行結果如下:
4、 獲取所需的時間信息
時間信息是shelf_version字段,而該字段存在於data_dict字典中"data"這個key對應的value值里。
這個value值是一個列表,列表里嵌套了多個字典。所以要再次對數據進行處理,取出value值。
代碼如下:
data = data_dict['data']
結果取出value列表(列表內容嵌套多個字典)。存放在新的文件newdata中,顯示如下:
數據格式整理好之后,終於可以開始數據篩選工作了。
一、非正則方法
1、 日期比對
筆者只想查找2019-03-21這一天的10點-12點的時間段內,總共傳輸過來多少條數據。
定義一個函數,用來篩選符合條件的日期。
代碼如下:
def get_data(x):
if x['shelf_version'] >'2019-03-21 10:00' and x['shelf_version'] <'2019-03-21 12:00' :
return x
使用filter函數,對newdata中的數據進行篩選,並把結果存放在新的list里,然后記錄在result結果文件中,代碼如下:
result=list(filter(get_data,newdata))
print(result)
with open("e:\\test\\result.txt",'w',encoding="utf-8") as fp1:
fp1.write(str(result))
結果如下:
根據時間段篩選數據成功。
2、 計算篩選出的數據條目
因為篩選出的數據類型為list,所以通過len函數就可以得到篩選結果的總數。(如果不是list格式,想必大家都知道可以用for循環或者count函數求合計值,達到一樣的效果)
代碼如下:
print("滿足條數的數據:",len(list(filter(get_data,newdata))))
得出的數據,和傳輸的數據條目一致。此接口的傳輸是成功的。
二、正則表達式匹配方法
1、 日期比對
列表中嵌套的每個字典中都有一個’shelf_version’字段(也就是想獲取的時間信息字段),使用for循環,循環取出。循環次數是newdata列表長度。
代碼如下:
for i in range(len(newdata)):
date=newdata[i]['shelf_version']
每次循環取出時間的時候,使用正則表達式進行時間比對。需要進行判斷,匹配成功的,才能寫入結果文檔中。
正則表達式如下(兩種都可以,在for循環內部執行):
mt = re.findall(r"2019-03-21 1[0-1]:.*",date)
#mt = re.findall(r"2019-03-21 1[0-1]:\d{2}:\d{2}",date)
匹配的是2019-03-21 10點至12點之間的數據。
寫入結果文檔代碼如下(在for循環內部執行):
if date in mt:
with open("e:\\test\\result.txt",'a',encoding="utf-8") as fp1:
fp1.write(str(mt))
print(mt)
此處需要注意,由於是放在for循環中,多次打開文檔,所以打開文檔寫入模式,應該是“a”追加寫。否則每次只能保存一個數據比對結果。
並且,調試程序的時候,應該每次執行前都刪掉result.txt文檔、重新生成,否則執行代碼時,調試的所有結果都會記錄在這個文檔中,會造成代碼錯誤的誤導。
執行結果如下:
根據時間段篩選數據成功,篩選出的是2019-03-21 10點至12點之間的數據。
2、 計算篩選出的數據條目
如果想要計算匹配成功的條目數,可以用一個變量count,每次寫入文檔的時候,計數加一。(此處代碼相信大家都會,略過)
得出的數據,和傳輸的數據條目一致。與第一種方法計算出的合計值相同,正則匹配方法成功。
由於是新手,可能實現方法有點復雜,還請大神們有更好的篩選方法,能不吝賜教,在評論區留言。歡迎批評指正。