某app航班信息抓取（1）-Toy模板网

这篇具有很好参考价值的文章主要介绍了某app航班信息抓取（1）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1. 工作环境

python

>python -V
Python 3.10.3

appium

appium desktop 1.15.1

selenium

>pip show selenium
Name: selenium
Version: 4.10.0
Summary:
Home-page: https://www.selenium.dev
Author:
Author-email:
License: Apache 2.0
Location: d:\··\python\310\lib\site-packages
Requires: certifi, trio, trio-websocket, urllib3
Required-by: Appium-Python-Client

android sdk
jdk

>java -version
java version "1.8.0_231"
Java(TM) SE Runtime Environment (build 1.8.0_231-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.231-b11, mixed mode)

mumu模拟器12
vscode
xmind

2. 需求

业务部的同事需要某app上航线下带有优惠的航班及其舱位信息。比如，去哪app上北京到上海航线，有m个航班，然后里边可能有n个航班，这些航班不管是航司或者是平台有些优惠，
优惠类型可能是立减优惠，会员优惠，长者优惠，新人优惠等等····，反正是对于符合对应条件的用户来说表现就是找到对应的航班买票会有价格优惠。以前个人只简单的抓一些资源网站的东西，
静态网页或者动态网页，只是简单的作为了解技术的附带产物。这次指定要抓app的数据，有个客观原因就是航司或者平台可能是为了推广自家移动端产品，有些优惠数据只有在app上才有。所以为了完成需求，就开始找些资料搞。
使用到的技术或者工具已经在上面列出来了，都是按我实际的环境来的。

3. 流程

3.1 分析
抓数据，我个人的理解不谈技术对抗，只谈论抓到目标数据，其实就是分析的过程，在上家公司做一些医疗项目的时候，涉及到药品溯源的项目，很多民营医院图省事都直接入驻了xx放心平台，有实力的医院都找自己的his厂商对接阿里的接口自己做自家的溯源管理系统。因为行业的特殊性，医疗行业的接口字段会很多，门诊数据，住院数据，医嘱数据，结算单数据，基金结算清单数据····少的几十个，一两百个，多个就是四五百、六七百个···随着drg报销政策的推广，那里边的接口也是····扯远了，反正就是当时有个机会搞药品溯源，然后接口呢是阿里提供的，当时因为接口有点多，一些字段什么的就直接抓接口爬数据了，当时算是对于动态网页的数据抓取有个感性的认识，太深的了解就没有了，现在的需求更深刻了，不抓网页，要抓app了，中间也是尝试了好多次，各种思路各种验证，有些太耗时间的就舍弃了，因为肯定要以能拿到目标数据为导向。最终，选择了现在这种方式，在抉择前，对于抓数据，我又稍稍有了更深的认识，不要技术对抗，避免技术对抗，要以和为贵···暴力是最后的手段。所以才有了现在这个过程。直到开始做了以后，整个过程可以总结成俩字：分析。
使用android sdk 带的工具uiautomatorviewer，分析目标数据所在页面结构。详细的分析过程说起来过于琐碎，都在下面的图里了：
某app航班信息抓取（1）
当时做的时候是按两种逻辑进行页面数据抓取的，一种是很惨的那种根据页面元素的相对位置来定位目标元素，另一种就是直接根据id来取，说到这就不得不说刚开始练手的qunar是真的蛋疼，因为根据具体分析想要拿到对应的数据只能按照第一种方式，分析的资料把自己都绕晕了，不过应该也逃不过，先把眼前的需求做了就好。上面的图画好了以后，我就知道这事容易很多了，因为对应数据是可以直接取到的。图片处理过了，因为一些东西容易猜到真名，没搞的无所谓，搞了的心虚，本来之前抓qunar的，人家说那个不紧要，这个着急，然后就上这个了···
3.2 再分析
使用android sdk 带的adb ,分析目标页面路径。目标数据页面的分析一旦完成，接下来就是补全自动抓取页面之间流转的业务。但是这个过程远比上面的步骤轻松的多，下图：
某app航班信息抓取（1）
过程大概是启动应用后进入首页，然后在首页填写出发/到达城市，选择出发日期，点击搜索进入航班结果页面，然后根据总结的带“减”字航班的数据要点击进去查看舱位信息。后续因为需求调整，代码的逻辑也就做了相应的调整，大概的流程就是这样的。

4. 代码

4.1 工具类

# excel表格工具类
# 生成excel表格

class ExcelUtils:
    '表格工具类，提供各种excel生成与读取方法'

    # 根据表头和表格数据生成excel表格
    # excelTitle 表头
    # excelData 表格数据
    # sheetName sheet名称
    # docName
    def doExportExcel (excelHead,excelData,sheetName,docName):
        # 新建工作簿，编码格式为utf-8
        workbook = xlwt.Workbook(encoding='utf-8')
        # sheet name
        worksheet = workbook.add_sheet(sheetName)
        # 设置表头
        for i in range(len(excelHead)):
            worksheet.write(0, i, excelHead[i])

        currentDateTime=datetime.datetime.now()
   

        # 遍历赋值
        # i 表示行数 j 表示列数
        for i in range(len(excelData)):
            # print("表格行数："+str(len(excelData)))
            # print("内容是："+excelData)
            # print(excelData[i][0])
            for j in range(len(excelData[i])):
                # print("内容是："+excelData[i][j])
                # print("i="+str(i))
                # print("j="+str(j))
                worksheet.write(i+1, j, excelData[i][j])

        # 表格名称由固定字符加当前时间共同组成
        workbook.save(docName+currentDateTime.strftime('%Y-%m-%d-%H-%M-%S')+'.xls')
        print("导出excel表格")

# eu=ExcelUtils.doExportExcel(QunarConstant.getSpecialPriceExcelHead(),"b")

4.2 常量类

class  qunarCommonConstant():
    '常用字面量'
# 表格表头
    SPECIAL_PRICE_EXCEl_HEAD=['始发城市','始发机场','目的城市','目的机场','航司','航班','起飞时间','优势1','优势2','价格','剩余票数']
    
    AIR_SPECIAL_PRICE_EXCEL_HEAD=['始发城市','始发机场','目的城市','目的机场','航班','起飞时间','立减','价格','舱位']

    @classmethod
    def getSpecialPriceExcelHead(self):
        return  self.SPECIAL_PRICE_EXCEl_HEAD
    @classmethod
    def getAirSpecialPriceExcelHead(self):
        return  self.AIR_SPECIAL_PRICE_EXCEL_HEAD

4.3 业务代码

# # 打开指定表格,待遍历航线
excelDoc= xlrd.open_workbook(filename=r'xx航线.xlsx')

# # 获取sheet表格
tableSheet = excelDoc.sheets()[0]
# # 获取sheet中有效行数
line_rows=tableSheet.nrows
# 准备自动化配置信息

desired_caps={
    # 真机信息
    # #移动设备平台
    # 'platformName':'Android',
    # #平台OS版本号
    # 'plathformVersion':'10',
    # #设备的名称
    # 'deviceName':'DT1901A',
    # 模拟器
    'platformName':'Android',
    'plathformVersion':'12',
    # 'deviceName':'127.0.0.1:16384',
     'deviceName':'127.0.0.1:16384',
    #提供被测app的信息-包名，入口信息
    'appPackage':'com.xx.xx',
    # app
    'appActivity':'xxx.WelcomeActivity',
    # 'appActivity':'com.mqunar.atom.uc.access.activity.UCQuickLoginActivity',
    #   mResumedActivity: ActivityRecord{152521f u0 com.Qunar/com.mqunar.atom.uc.access.activity.UCQuickLoginActivity t13735}
    #确保自动化之后不重置app
    'noReset':True,
    #设置session的超时时间，单位秒
    'newCommandTimeout':600000,
    #更换底层驱动
    'automationName':'UiAutomator2',
    # 不停止正在测试的应用程序的进程，默认为false。为true时，
    # appium 将不会在adb shell am start调用中包含-S标志，不需要重新启动。
    'dontStopAppOnReset': True
    # 'unicodeKeyboard':True,#修改手机的输入法，UI2不需要设置
    # 'resetKeyboard':True#自动化结束之后将输入法还原
}
#初始化driver对象-用于控制手机
 
#  appium 2.0 请求改变了
# driver=webdriver.Remote('http://127.0.0.1:4723/',desired_caps)
driver=webdriver.Remote('http://127.0.0.1:4723/wd/hub',desired_caps)
print("手机应用连接成功")
print("程序已启动")
    # 手机动作
touch1=TouchAction(driver)
    # 导出表格数据
excelData =[]
    # 航线计数
lineCount=0
    # 待遍历航线列表
    # queryDataList=[]
    # 踢出表头
for i in range (line_rows-1):
  
    # 只有第一次进入的时候需要强制等待
    # 首页加载时需要强制等待，保证后续逻辑的顺利进行
    if(i==0):
        time.sleep(3)
    lineCount+=1
    print("======正在查询第%s条行航线======" % lineCount)
    try:

        searchResultGoBackFlag=False
        # 始发城市
        iataExcelDptCityRow=tableSheet.row(i+1)[0].value
        # queryDataList.append(iataExcelDptCityRow)
        # 目的城市
        iataExcelArrCityRow=tableSheet.row(i+1)[1].value
        # queryDataList.append(iataExcelArrCityRow)
        # time.sleep(5)
        # 进入程序自动操作流程
        # 显示等待加载 设置出发城市
        tv_start=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        # tv_start.click() 不推荐 该指令触发容易发生问题

        # tap 在指定元素上敲击 推荐
        touch1.tap(element=tv_start).perform()
        list_tv_citySearch=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        touch1.tap(element=list_tv_citySearch).perform()
        list_tv_citySearch=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        list_tv_citySearch.send_keys(iataExcelDptCityRow)
        dept_tv_title=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        touch1.tap(element=dept_tv_title).perform()
        #起始城市输入完毕
        print("起始城市输入:%s,完毕" % iataExcelDptCityRow)
        # 开始输入目的城市
        view_booking_tv_back=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        touch1.tap(element=view_booking_tv_back).perform()
        city_list_tv_citySearch=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        touch1.tap(element=city_list_tv_citySearch).perform()
        city_list_et_citySearch=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        city_list_et_citySearch.send_keys(iataExcelArrCityRow)
        arr_tv_title=WebDriverWait(driver,10).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        touch1.tap(element=arr_tv_title).perform()
        print("目的城市输入:%s,完毕" % iataExcelArrCityRow)

        # 设置起飞日期

        # tv_month_start=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        # # # 进入选择页面
        # # # tv_month_start.click()
        # touch1.tap(element=tv_month_start).perform()
        # ActionChains(driver).move_to_element(tv_month_start).click(tv_month_start).perform()
        # calendar_month_view=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
        # print(calendar_month_view[0].text)
        # print("======设置起飞日期======")
        # 显示等待加载 点击搜索
        booking_llyt_querybtn=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
        # booking_llyt_querybtn.click()
        touch1.tap(element=booking_llyt_querybtn).perform()
        # print("点击搜索")

        try:
            # 进入搜索结果页面
            # 获取所有带"减" 字的航班信息，然后点击进去抓取会员特惠舱位信息
            list_data_rv=WebDriverWait(driver,3).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
            # 强制等待，xxapp，不知道是因为网络问题还是故意设置页面加载，要延时完成
            time.sleep(3)
            print("======带减字航班共:%s条======" %str(len(list_data_rv)))
            # touch1.tap(element=list_data_rv[0]).perform()
            goBackFlag=False
            # 遍历计数
            countNum=0
            for list_element_data in list_data_rv:
                excelElement=[]
                countNum+=1
                print("======正在抓取立减数据，条数：%s======" % countNum)
            
                # 点击第一个立减航班，进入舱位页面
                try:
                        
                            time.sleep(0.5)
                            touch1.tap(element=list_element_data).perform()

                            # 进入立减结果页面后
                            # 业务处理 1、等待加载 2、获取立减列表并进行遍历，判断其立减优惠是否是会员优惠，如果是，则抓取否则跳过执行下一条判断
                            # 等待页面加载
                            # time.sleep(0.5)
                            try:
                                # 判断立减数据中是否包含"会员",如果包含则进行抓取，否则不进行数据提取
                                try:
                                    tips_content_iv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                    tips_content=tips_content_iv.text
                                except:
                                    tips_content="自定义"
                                    print("不是会员优惠，跳出抓取")
                                finally:
                                    try:
                                        if(tips_content.find('会员')!=-1):
                                        # 判断立减舱位条数
                                            order_cut_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                            length_order_cut_tv=len(order_cut_tv)
                                            print("=========立减舱位信息条数：%s=========" %length_order_cut_tv)
                                            # 立减舱位数等于1的时候
                                            if(length_order_cut_tv==1):
                                                print("===========走线路1===============")
                                                # 显示等待加载 起飞日期
                                                flight_go_date=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(flight_go_date.text)

                                                # 显示等待加载 出发机场
                                                dep_term_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(dep_term_tv.text)
                                                # # 显示等待加载 到达机场
                                                # arr_term_tv=WebDriverWait(driver,5).until(lambda x:list_element_data.find_element(by=By.ID,value='resources-id'))
                                                arr_term_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(arr_term_tv.text)
                                                # # 显示等待加载 价格
                                                # details_price_tv=WebDriverWait(driver,5).until(lambda x:list_element_data.find_element(by=By.ID,value='resources-id'))
                                                details_price_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(details_price_tv.text)
                                                # # 显示等待加载 立减
                                                # order_cut_tv=WebDriverWait(driver,5).until(lambda x:list_element_data.find_element(by=By.ID,value='resources-id'))
                                                order_cut_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(order_cut_tv.text)
                                                # # 显示等待加载 航班
                                                # flight_no=WebDriverWait(driver,5).until(lambda x:list_element_data.find_element(by=By.ID,value='resources-id'))
                                                flight_no=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(flight_no.text)
                                                # # 显示等待加载 舱位
                                                details_code_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                # details_code_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(details_code_tv.text)
                                                # 出发城市
                                                excelElement.append(iataExcelDptCityRow)
                                                # 出发机场
                                                excelElement.append(dep_term_tv.text)
                                                # 目的城市
                                                excelElement.append(iataExcelArrCityRow)
                                                # 目的机场
                                                excelElement.append(arr_term_tv.text)
                                                # 航班
                                                excelElement.append(flight_no.text)
                                                # 起飞时间
                                                excelElement.append(flight_go_date.text)
                                                # 立减
                                                excelElement.append(order_cut_tv.text)
                                                # 价格
                                                excelElement.append(details_price_tv.text)
                                                # 舱位
                                                excelElement.append(details_code_tv.text)
                                                excelData.append(excelElement)
                                            elif (length_order_cut_tv>1):
                                                print("=======================走线路2==================")
                                                # 显示等待加载 起飞日期
                                                flight_go_date=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(flight_go_date[0].text)

                                                # 显示等待加载 出发机场
                                                dep_term_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(dep_term_tv[0].text)
                                                # # 显示等待加载 到达机场
                                                arr_term_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(arr_term_tv[0].text)
                                                # # 显示等待加载 价格
                                                details_price_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(details_price_tv[0].text)
                                                # # 显示等待加载 立减
                                                order_cut_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(order_cut_tv[0].text)
                                                # # 显示等待加载 航班
                                                flight_no=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                print(flight_no[0].text)
                                                # # 显示等待加载 舱位
                                                details_code_tv=WebDriverWait(driver,5).until(lambda x:x.find_elements(by=By.ID,value='resources-id'))
                                                # details_code_tv=WebDriverWait(driver,5).until(lambda x:x.find_element(by=By.ID,value='resources-id'))
                                                print(details_code_tv[0].text)
                                                # 出发城市
                                                excelElement.append(iataExcelDptCityRow)
                                                # 出发机场
                                                excelElement.append(dep_term_tv[0].text)
                                                # 目的城市
                                                excelElement.append(iataExcelArrCityRow)
                                                # 目的机场
                                                excelElement.append(arr_term_tv[0].text)
                                                # 航班
                                                excelElement.append(flight_no[0].text)
                                                # 起飞时间
                                                excelElement.append(flight_go_date[0].text)
                                                # 立减
                                                excelElement.append(order_cut_tv[0].text)
                                                # 价格
                                                excelElement.append(details_price_tv[0].text)
                                                # 舱位
                                                excelElement.append(details_code_tv[0].text)
                                                excelData.append(excelElement)
                                        else :
                                            print("立减为非会员优惠票，不进行抓取")
                                    except:
                                        print("====立减舱位只抓取第一条=====")
                            except:
                                print("立减舱位信息抓取失败，进行下一个舱位抓取")
                            finally:
                                # 目前只抓去一个立减信息即可，其他立减都时一样的
                                driver.back()
                                break
                                
                        
                 
                except:
                 
                    print("======立减数据抓取出错了，进行下一个航班======")
                finally:
                    # 增加强制等待，防止连续执行上一步退出app
                    # time.sleep(0.5)
                    driver.back()
                    break
       
        except:
            goBackFlag=True
            print("======该航班没有立减优惠，跳过进行下一个航班查询,返回上一层：%s======" % goBackFlag)
        finally:
             # 增加强制等待，防止连续执行上一步退出app
            if(goBackFlag):
                driver.back()
    except:
        searchResultGoBackFlag=True
        # print("航线遍历出错了，进行下次循环")
        print("======第%s行航线数据抓取失败,进入下一航线抓取======" % lineCount)
        
    finally:
    
    #     # 等待时间，保证app顺利启动
        # pass
   
        # time.sleep(0.5)
        if(searchResultGoBackFlag):
             driver.back()
        # driver.back()
        # time.sleep(5)
        if (i>=10 and i % 10==0):
            print("为保证抓取数据程序顺利跑完，每查询10次，进行程序重新加载")
            driver.launch_app()
        #     ExcelUtils.ExcelUtils.doExportExcel(qunarCommonConstant.getAirSpecialPriceExcelHead(),excelData,'xxApp','xxApp')
    
ExcelUtils.doExportExcel(qunarCommonConstant.getAirSpecialPriceExcelHead(),excelData,'xxApp','xxApp')
print("抓取完毕")
# 断掉服务链接
driver.quit()

代码是按照分析线性逻辑实现的，中间因为没搞过，每次出了问题就立马查资料改正，总算了满足了需求。现在回想起来，感觉得把知识点总结一下，这是以后应该会常用的。

针对页面元素查找要使用显示等待的方式，提升抓数据的周期效率
安卓的连续两次返回上一页等于退出程序，driver.back()。终于从另一个层面认识到了这个事实，因为忘记了这个，导致中间测试运行的时候总是无故退出程序。搞的我都怀疑人生了，后来灵光一闪，想起来了，最后验证了确实是这个原因。所以这个印象很深刻。
由第二条的因素，所以必须要考虑目标app无故闪退。所以，要加上定量执行 driver.launch_app()，这个方法是保证无人值守的时候程序可以完整的跑完全程的重要方法。

4.4 测试运行
因为我也是第一次抓取手机app的数据，磕磕绊绊，总算是写出来，最初的时候每次跑数据，都是要开着录屏，记录运行过程中的出现的问题，然后分析，修改，分析，修改···最终完成了上面的抓取业务。

最初跑程序的时候，人一直在看着，后来加了很多处理，防止抓取数据业务中断的处理。目前可以按照给定的航线表格，完成所有的结果搜索及目标数据抓取。

5. 运行

经过几天的完整航线抓取（大概5000条航线），最长的一次是连续跑了19个多小时，最终抓到了指定的目标数据。中间抓数据过程中还发现几个问题，尤为突出的是程序不定时的弹出公告通知，还有莫名奇妙的网络延迟，当然还有一些更具体的问题也因为业务的动态调整没有去研究处理办法，比如自动滑动屏幕定位到指定元素然后进行单击或者别的操作。这些肯定是要解决的。随后再说吧。
结果：

某app航班信息抓取（1）文章来源地址https://www.toymoban.com/news/detail-582067.html