python+appium+夜神模拟器监控京东商品价格

需要注意,这里采用了微信登陆的方式,所以需要事先在模拟器登录微信,然后用微信登录京东。京东APP登录一次后就不需要再次登陆了,appium inspector的设置如下:

{
  "platformName": "Android",
  "appium:platformVersion": "7.1.2",
  "appium:deviceName": "127.0.0.1:62001",
  "appium:appPackage": "com.jingdong.app.mall",
  "appium:appActivity": "com.jingdong.app.mall.main.MainActivity",
  "unicodeKeyboard": true,
  "resetKeyboard": true,
  "noReset": true,
  "noSign": true,
  "skipDeviceInitialization": true
}
## 监控购物车感兴趣商品的价格,可以下单
from appium import webdriver
from appium.webdriver.common.touch_action import TouchAction
import time
import re


class Jingdong():
    def __init__(self):
        desired_caps = {}
        # 设备系统
        desired_caps['platformName'] = 'Android'
        # 设备系统版本号
        desired_caps['platformVersion'] = '7.1.2'
        # 设备名称
        desired_caps['deviceName'] = '127.0.0.1:62001'
        ## 要测试的应用的地址
        #desired_caps['app'] = 'xxx.apk'   # 注意是 apk 文件在电脑上的绝对路径
        # 应用的包名
        desired_caps['appPackage'] = 'com.jingdong.app.mall'
        desired_caps['appActivity'] = 'com.jingdong.app.mall.main.MainActivity'
        desired_caps["noReset"] = True
        desired_caps["skipDeviceInitialization"] = True
        self.driver = webdriver.Remote('http://127.0.0.1:4723/wd/hub', desired_caps)  # 地址为appium地址
        time.sleep(1.5)

    def search(self, keyword):
        self.driver.find_element_by_xpath('//android.view.View[@content-desc="我的"]').click()
        time.sleep(1.5)
#         self.driver.find_element_by_xpath('//android.widget.TextView[@content-desc="登录/注册"]').click()
#         time.sleep(1.5)
#         self.driver.find_element_by_xpath("//android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[2]/android.widget.EditText[1]").send_keys(keyword)
#         time.sleep(1.5)
#         self.driver.find_element_by_xpath("//android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.TextView[1]").click()
#         time.sleep(1.5)

        el1 = self.driver.find_element_by_xpath("/hierarchy/android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/android.widget.RelativeLayout/android.widget.FrameLayout[1]/android.widget.RelativeLayout/android.widget.LinearLayout/android.widget.FrameLayout[4]/android.widget.ImageView[2]")
        
        el1.click()
        time.sleep(3)

        try:
            el1_ = self.driver.find_element_by_xpath('//android.widget.LinearLayout[@content-desc="未选中,全选,复选框"]/android.widget.CheckBox')
            el1_.click()
            time.sleep(1.5)
            el1_ = self.driver.find_element_by_xpath('//android.widget.LinearLayout[@content-desc="已选中,全选,复选框"]/android.widget.CheckBox')
            el1_.click()
            time.sleep(1.5)            
        except Exception as e:
            print('已全部选中物品', e)
            el1_ = self.driver.find_element_by_xpath('//android.widget.LinearLayout[@content-desc="已选中,全选,复选框"]/android.widget.CheckBox')
            el1_.click()
            time.sleep(1.5)            
        # //android.widget.RelativeLayout[@content-desc="康师傅方便面红烧牛肉面10袋+香辣牛肉面5+鲜虾鱼板面5+老坛酸菜牛肉面5袋泡面礼盒整箱装"]/android.widget.FrameLayout/android.widget.LinearLayout/android.widget.CheckBox
        el2 = self.driver.find_element_by_xpath("//android.widget.RelativeLayout[@content-desc=\"蒙牛纯甄 常温风味酸牛奶 送礼推荐 200g*24 纯正生牛乳发酵 礼盒装(新老包装随机发货)\"]/android.widget.FrameLayout/android.widget.LinearLayout/android.widget.CheckBox")
        el2.click()
        time.sleep(1.5)
        el3 = self.driver.find_element_by_id("com.jd.lib.cart.feature:id/cart_settle_accounts_but") # com.jd.lib.cart.feature:id/cart_settle_accounts_but
        
        el3.click()
        time.sleep(1.5)
        cart_product_name = self.driver.find_element_by_id('com.jd.lib.settlement.feature:id/q6')
        # 获取模拟器/手机的分辨率(px)
        width = self.driver.get_window_size()['width']
        height = self.driver.get_window_size()['height']
        # print(width, height)
#         TouchAction(self.driver).press(x=372, y=1454).move_to(x=652, y=189).release().perform() 
#         time.sleep(1.5)
#         TouchAction(self.driver).press(x=372, y=1454).move_to(x=652, y=189).release().perform()
        start_x = width // 2 # 滑动的起始点的x坐标,屏幕宽度中心点
        start_y = height // 3 * 2 # 滑动的起始点的y坐标,屏幕高度从上开始到下三分之二处
        distance = height // 2 # y轴滑动距离:屏幕高度一半的距离
        end_x = start_x # 滑动的终点的x坐标
        end_y = start_y-distance # 滑动的终点的y坐标
        #滑动
        self.driver.swipe(start_x, start_y, end_x, end_y)
        
        ## price = self.driver.find_element_by_xpath('//android.widget.TextView[@content-desc=*]').text()
        time.sleep(1.5)
        price = self.driver.find_element_by_id('com.jd.lib.settlement.feature:id/hy').text # 订单总价格 #hy a7p
#         self.driver.findElement(By.xpath("//android.widget.TextView[contains(@text,'本单'])")).
        print('current price: ', price)
        print(re.search("\d+\.\d+", price)[1])
        self.driver.find_element_by_accessibility_id('返回').click()
        time.sleep(1.5)
        try:
            self.driver.find_element_by_id('com.jingdong.app.mall:id/br').click() #忍痛离开
        except Exception as e:
            print(e)

def main():
    jingdong = Jingdong()
    keyword = "酸奶"
    jingdong.search(keyword)

if __name__ == '__main__':
    main()

hive load本地csv新建表

作为分析师,经常遇到建临时表的需求,比如从另外一个系统(比如神策)导出用户数据到csv,然后导入hive表:

use temp;
create table if not exists temp.fast_entrance (dt int, user_id string) row format delimited fields terminated by ',';
hdfs dfs -put 极速处理入口点击但未处理的用户.csv /user/xuzhen.chen/
LOAD DATA INPATH '/user/xuzhen.chen/极速处理入口点击但未处理的用户.csv' OVERWRITE INTO TABLE fast_entrance;

这种需求非常常见,神策圈定具备某些行为特征的用户,然后看这部分用户的属性(hive表)。需要注意的是,考虑到磁盘空间和数据传输效率,这里不能是太大的csv文件