python UI自動化實戰記錄二：請求接口數據並提取數據

該部分記錄如何獲取預期結果-接口響應數據，分成兩步：

1 獲取數據源接口數據

2 提取后續頁面對比中要用到的數據

並且為了便於后續調用，將接口相關的都封裝到ProjectApi類中。

 

新建python包：apiclass 》 新建python file：api_fund。所有接口相關的操作均放到該文件中。隱去項目相關信息后的代碼如下：

 

1 獲取數據源接口數據

# coding:utf-8

import requests
from common.round_rewrite import round_rewrite #點擊查看該函數 四舍五入
from common.jsonp_to_json import jsonp_to_json #點擊查看該函數 jsonp轉成json

class Fund:
    
    fund_api_url = '接口地址url'

    """四個原始數據接口"""
    def api_strategy(self,day=''):
        """
　　　　　　接口1
        """
        url = Fund.fund_api_url+'api1.json'
        response = requests.get(url,params={"day":day}).json()
        return response

    def api_lastestinfo(self,code):
        """
　　　　　接口2
        """
        url = Fund.fund_api_url+'latestInfo/{0}.json'.format(code)
        response = requests.get(url).json()
        return response

    def api_trends(self,code,pattern,period):
        """
　　　　　接口3　　
　　　　 """
        identifier = "{code}_{pattern}_{period}".format(code=code,pattern=pattern,period=period)
        url = Fund.fund_api_url+"trends/{0}.json".format(identifier)
        jsonpstr = requests.get(url).text 
        jsonstr = jsonp_to_json(jsonpstr)
        return jsonstr

    def api_timeline(self,code):
        """
　　　　　接口4
        """
        url = Fund.fund_api_url+"timeline/{0}.json".format(code)
        response = requests.get(url).json()
        return response

 

2 提取后續頁面對比中要用到的數據

 

接口1比較特別，返回數據是一個list，按時間升序排列。有的頁面需要取最早的數據，有的頁面取最新的數據。

1  用一個參數來標識，該參數設置成list切片步長。1從前往后去，-1從后往前取。

2 samples是一個dict組成的list，要取出每一個dict里的values

samples = sorted([list(ele.values()) for ele in samples]) # 轉變成與頁面數據一致的格式

def get_fund_strategy(self,code,day='',latest=1):
        """提取接口1的數據
        latest:1-取最早的數據；-1-取最新的數據"""
        fund_strategy = self.api_strategy(day) #獲取策略配置接口數據，day之后的數據都會取出來

        for ele in fund_strategy[::latest]:  #1從前往后取最早，-1從后往前取最新
            if ele["code"] == code:
                self.code = ele["code"]
                self.name = ele["name"]
                self.summary = ele["summary"]
                self.memo = ele["memo"]
                samples = ele["samples"]
                self.samples = sorted([list(ele.values()) for ele in samples]) # 轉變成與頁面數據一致的格式
                return #取到則退出循環

 

接口2：一個頁面只需要3M的數據，單獨寫了一個函數； 另一個頁面需要全量數據。

    def get_fund_latestinfo(self,code):
        """提取接口2的數據"""
        fund_lastestinfo = self.api_lastestinfo(code)  
        nav = fund_lastestinfo["nav"]
        navDate = fund_lastestinfo["navDate"][-5:]  
        navChange = fund_lastestinfo["navChange"]
        annualChangeAll = fund_lastestinfo["annualChangeAll"]
        self.navlist = [nav,navDate,navChange,annualChangeAll] 
        percents = fund_lastestinfo["percents"] 
        self.percents_list = [list(ele.values())[1:] for ele in percents] 

 def get_fund_percentM3(self,code):
        """獲取3個月收益率，首頁有該數據"""
        fund_lastestinfo = self.api_lastestinfo(code) 
        self.percentM3 =fund_lastestinfo["percents"][0]["percentM3"] 
        return self.percentM3

 

 

接口3：需要對數值進行判斷，當數值>=0,顯示超出，否則跑輸。

sharprun = "超出" if self.sharpeDiff >= 0 else "跑輸"  　　

將列表里的每一個字典的key轉成中文，方便與頁面數據對比

self.trends = map(lambda line: {"日期":line["date"],"組合市值":line["mv"],"比較基准市值":line["bmv"]},trends) #列表里的字典key英文轉成中文

def get_fund_trends(self,code,pattern,peroid):
        """提取接口3的數據"""
        fund_trends = self.api_trends(code, pattern, peroid)  # 請求接口數據
　　　　
　　　　　"""獲取接口字段值"""
        self.percent = fund_trends["percent"]  
        self.percentDiff = fund_trends["percentDiff"] 
        self.maxDown = fund_trends["maxDown"]  
        self.mdStart = fund_trends["mdStart"]  
        self.mdEnd = fund_trends["mdEnd"]  
        self.startDate = fund_trends["startDate"]  
        try:
            self.sharpe = fund_trends["sharpe"]  
            self.sharpeDiff = fund_trends["sharpeDiff"]  
            sharprun = "超出" if self.sharpeDiff >= 0 else "跑輸"  
            percentRun = "超出" if self.percentDiff >= 0 else "跑輸"
            sharpeDiff = abs(self.sharpeDiff)
        except KeyError: # 夏普比率跨年時今年以來接口無數據，置為空
            sharpe = sharpeDiff =  sharprun = percentRun = ""

        trends = fund_trends["trends"] # 組合漲幅走勢數據
        self.trends = map(lambda line: {"日期":line["date"],"組合市值":line["mv"],"比較基准市值":line["bmv"]},trends) #列表里的字典key英文轉成中文

        result = [code,self.startDate, percentRun, abs(self.percentDiff), self.percent,
                  self.maxDown,sharprun, sharpeDiff, self.sharpe, self.mdStart, self.mdEnd] #與頁面一樣的格式
        return result
  

 

 

接口4：需要取當前和昨天的值計算漲跌幅，並保留2位小數

 """提取原始接口中頁面所需數據"""
    def get_fund_timeline(self,code):
        """提取接口4的數據"""
        fund_timeline = self.api_timeline(code) 

        last = fund_timeline["last"] 
        date = fund_timeline["date"] 
        current = fund_timeline["current"] 
        timeline = fund_timeline["timeline"] 

        rate = (current - last) / last * 100 
        timeline_current = dict(日期=date,實時估值=current,估值漲幅=round_rewrite(rate,2)) 

        timeline_list = []
        for tl in timeline: #分時數據
            dt = tl["dt"]
            nav = tl["nav"]
            rt = (nav - last) / last * 100
            timelinedata = dict(時間=dt,估值=nav,漲跌幅=round_rewrite(rt,2))
            timeline_list.append(timelinedata)
        return timeline_current,timeline_list

 

 

最后一部分，因頁面圖形無法自動化驗證，手工測試相關的函數：

 def mannualtest_timeline(self):
        """分時圖手工測試"""
        print('code：00X00Y')
        scode = input("獲取實時估值 輸入code")
        try:
            """實時估值"""
            current, timeline = self.get_fund_timeline(scode)
            print("實時估值：", current)
            print("分時圖數據：")
            for line in timeline:
                print(line)
        except Exception as e:
            print(e)

 

 def mannualtest_trends(self):
        """走勢圖手工測試"""
        print('code：00X00Y')
        scode = input("獲取組合走勢圖形數據：請輸入code\n")
        pattern = input("投資方式 W（默認） K\n")
        peroiddict = {'1': 'R1M', '2': 'R3M', '3': 'R6M','5': 'R1Y', '6': 'R3Y'}
        peroid = input("投資期限輸入對應數字 %s\n"%peroiddict)

        peroid = peroid.strip()
        pattern = pattern.strip().upper()#去除左右空格后轉大寫

        if pattern != 'K':
            pattern = 'W'#只要不等於K，則默認W
        if peroid in peroiddict.keys():
            peroid = peroiddict[peroid] #在字典里則取對應值
        else:
            peroid = 'R1M' #不在字典取默認R1M

        try:
            self.get_fund_trends(scode, pattern, peroid) #獲取接口數據
            print("組合走勢圖{scode}_{pattern}_{peroid}".format(scode=scode,pattern=pattern,peroid=peroid))
            for line in self.trends:
                print(line)
        except Exception as e:
            print(e)

 

the end！