代码小整理3

2023-07-27 4427 words 9 minutes

Contents

📱 Android UI 自动化 (uiautomator2)

电脑和手机需要在同一局域网环境下

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42


# coding: utf-8
import uiautomator2 as u2
import os

# 连接设备
def connect_device():
    """通过adb连接设备"""
    tmp = os.popen('adb devices')
    serial_number = tmp.readlines()[1].split()[0]
    return u2.connect(serial_number)

# 初始化设备连接
d = connect_device()

# 返回主页
d.press('home')
d.sleep(0.1)
d.press('home')
d.sleep(1)

# 获取屏幕尺寸并滑动
x, y = d.window_size()
x1 = x * 0.9
y1 = y * 0.5
x2 = x * 0.1
d.swipe(x1, y1, x2, y1)
d.sleep(1)

# 点击文件夹
try:
    d.xpath('//*[@content-desc="文件夹：amuse"]/android.widget.ImageView[1]').click()
except:
    d.xpath('//*[@content-desc="amuse 小文件夹"]/android.widget.ImageView[1]').click()

# 点击应用
d(text='命运-冠位指定').click()
d.sleep(1)
d.click(0.37, 0.776)

# 停止应用
d.app_stop("com.bilibili.fgo.uc")
d.uiautomator.stop()

游戏自动化 adb

1
2
3
4
5


adb shell input keyevent 3   # home键
adb shell input keyevent 4   # 返回键
adb shell input tap x y      # 触击屏幕第x行，第y列的点
adb shell getprop ro.product.model # 获取手机型号
adb shell wm size            # 获取屏幕分辨率

pip3 install weditor==0.6.3

参考资料：
遇到的问题：https://blog.csdn.net/folcan/article/details/123441086
使用教程：https://blog.csdn.net/pirate5211/article/details/128111141
adb安装：https://blog.csdn.net/x2584179909/article/details/108319973
wifi连接：https://blog.csdn.net/zhuan_long/article/details/104323961

🎮 Pygame 横版跑酷游戏

设置游戏内的常量

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


WIDTH = 1200
HEIGHT = 508
SIZE = (WIDTH, HEIGHT)
score = None
my_font = None
my_font1 = None
sur_game_over = None
bg = None
role = None
sur_object = None
game_object = None
object_list = []
clock = None
game_state = None  # 0:游戏中, 1:游戏结束

游戏角色类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


class Role:
    def __init__(self, surface=None, y=None):
        self.surface = surface
        self.y = y
        self.w = surface.get_width() / 12
        self.h = surface.get_height() / 2
        self.current_frame = -1
        self.state = 0  # 0:跑步, 1:跳跃, 2:连续跳跃
        self.g = 1  # 重力加速度
        self.vy = 0  # y轴速度
        self.vy_start = -20  # 起跳初速度

    def get_rect(self):
        return (0, self.y + 12, self.w, self.h)

障碍物类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


import random
import pygame

class GameObject:
    def __init__(self, surface, x=0, y=0):
        self.surface = surface
        self.x = x
        self.y = y
        self.w = 100
        self.h = 100
        self.current_frame = random.randint(0, 6)

    def get_rect(self):
        return (self.x, self.y, self.w, self.h)

    def collision(self, rect1, rect2):
        """碰撞检测"""
        if (rect2[0] >= rect1[2] - 20 or
            rect1[0] + 40 >= rect2[2] or
            rect1[1] + rect1[3] < rect2[1] + 20 or
            rect2[1] + rect2[3] < rect1[1] + 20):
            return False
        return True

背景类

1
2
3
4
5
6


class Background:
    def __init__(self, surface):
        self.surface = surface
        self.dx = -10
        self.w = surface.get_width()
        self.rect = surface.get_rect()

辅助功能函数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55


def handle_jump():
    """处理跳跃输入"""
    if role.state == 0:  # 一段跳
        role.state = 1
        role.vy = role.vy_start
    elif role.state == 1:  # 二段跳
        role.state = 2
        role.vy = role.vy_start

def update_role():
    """更新角色状态"""
    # 应用重力
    role.vy += role.g
    role.y += role.vy

    # 地面检测
    if role.y >= HEIGHT - role.h:
        role.y = HEIGHT - role.h
        role.vy = 0
        role.state = 0

    # 更新动画帧
    role.current_frame = (role.current_frame + 0.2) % 12

def update_objects():
    """更新障碍物状态"""
    global object_list, score

    # 移动障碍物
    for obj in object_list:
        obj.x += bg.dx
        obj.current_frame = (obj.current_frame + 0.1) % 7

    # 移除屏幕外的障碍物
    object_list = [obj for obj in object_list if obj.x > -100]

    # 随机生成新障碍物
    add_object()

def check_collisions():
    """检测碰撞"""
    global game_state

    role_rect = role.get_rect()

    for obj in object_list:
        obj_rect = obj.get_rect()
        if obj.collision(role_rect, obj_rect):
            game_state = 1  # 游戏结束
            break

def update_score():
    """更新分数"""
    global score
    score += 1

游戏功能函数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


def init_game():
    """初始化游戏所有组件和状态"""
    global bg, role, clock, game_state, sur_object, sur_game_over
    global score, my_font, my_font1, object_list, screen, sur_bg

    # pygame初始化
    pygame.init()
    screen = pygame.display.set_mode(SIZE)
    pygame.display.set_caption('天天跑酷')

    # 游戏状态初始化
    score = 0
    game_state = 0
    object_list = []

    # 加载资源
    my_font = pygame.font.Font("game/font/consola.ttf", 32)
    my_font1 = pygame.font.Font("game/font/consola.ttf", 64)
    clock = pygame.time.Clock()

    # 加载图像资源
    sur_bg = pygame.image.load("game/image/bg.bmp").convert_alpha()
    sur_bg = pygame.transform.scale(sur_bg, (WIDTH, HEIGHT))
    bg = Background(sur_bg)

    sur_game_over = pygame.image.load("game/image/gameover.bmp").convert_alpha()
    sur_game_over = pygame.transform.scale(sur_game_over, (WIDTH, HEIGHT))

    sur_role = pygame.image.load("game/image/role.png").convert_alpha()
    role = Role(sur_role, HEIGHT - 85)

    sur_object = pygame.image.load("game/image/object.png").convert_alpha()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


def add_object():
    """随机生成障碍物"""
    global sur_object, object_list
    rate = 4  # 生成概率

    if not random.randint(0, 300) < rate:
        return

    y = random.choice([HEIGHT-100, HEIGHT-200, HEIGHT-300, HEIGHT-400])
    new_object = GameObject(sur_object, WIDTH + 40, y)
    object_list.append(new_object)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


def update_view(screen):
    """更新游戏画面"""
    # 渲染背景
    screen.blit(bg.surface, [-bg.dx, 0])
    screen.blit(bg.surface, [WIDTH - bg.dx, 0])

    # 渲染分数
    text_sur = my_font.render(f"score: {score}", True, (128, 128, 128))
    screen.blit(text_sur, (500, 20))

    # 渲染角色
    screen.blit(role.surface, [0, role.y],
                [int(role.current_frame) * role.w, 0, role.w, role.h])

    # 渲染障碍物
    for obj in object_list:
        screen.blit(obj.surface, [obj.x, obj.y],
                   [int(obj.current_frame) * obj.w, 0, obj.w, obj.h])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


def judge_state(screen):
    """根据游戏状态渲染不同画面"""
    global game_state

    if game_state == 0:  # 游戏中
        update_view(screen)
    elif game_state == 1:  # 游戏结束
        screen.blit(sur_game_over, [0, 0])
        text_sur = my_font1.render(f"GameOver Score: {score}", True, (255, 0, 0))
        screen.blit(text_sur, (WIDTH/2 - 350, HEIGHT/2 + 150))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


def update_logic():
    """更新游戏逻辑"""
    global game_state, score, object_list

    # 处理事件
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            pygame.quit()
            sys.exit()
        elif event.type == pygame.KEYDOWN:
            if game_state == 0:  # 游戏中
                if event.key == pygame.K_SPACE:
                    handle_jump()
            elif game_state == 1:  # 游戏结束
                if event.key == pygame.K_SPACE:
                    init_game()  # 重新开始游戏

    if game_state == 0:
        # 更新角色状态
        update_role()

        # 更新障碍物
        update_objects()

        # 检测碰撞
        check_collisions()

        # 更新分数
        update_score()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


def main_game_loop():
    """游戏主循环"""
    init_game()

    screen.blit(bg.surface, [0, 0])

    while True:
        clock.tick(60)  # 60 FPS
        judge_state(screen)
        update_logic()
        pygame.display.flip()

if __name__ == "__main__":
    main_game_loop()

🕷️ Scrapy 爬虫框架

📚 参考资料

https://blog.csdn.net/weixin_44505901/article/details/105041462
https://docker-practice.github.io/zh-cn/install/ubuntu.html
https://www.cnblogs.com/shaosks/p/6950358.html
https://blog.csdn.net/jbluxun/article/details/119716823
https://blog.csdn.net/weixin_44173603/article/details/108272819
https://blog.csdn.net/wzp7081/article/details/111404445

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


# 安装Scrapy
pip install Scrapy

# 创建新项目
scrapy startproject --logfile="./logf.log" --loglevel=WARNING project_name

# 示例：创建B站爬虫项目
scrapy startproject bilibili
cd ./bilibili
scrapy genspider bili_spider "bilibili.com"

全局命令	项目命令
fetch - 显示爬虫过程	bench - 测试本地硬件性能
runspider - 运行.py 爬虫文件	genspider - 创建爬虫文件
settings - 查看设置	check - 爬虫测试
shell - 启动交互终端	crawl - 启动爬虫
view - 下载并查看网页	list - 列出可用爬虫文件
startproject - 创建项目	edit - 编辑爬虫文件
version - 显示版本	parse - 分析指定网址

关于一个招标网站的爬取

目录结构：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


yunnan_ggzy/
    ├── yunnan_ggzy/
    │   ├── spiders/
    │   │   └── yn_trade.py
    │   ├── __init__.py
    │   ├── items.py
    │   ├── middlewares.py
    │   ├── pipelines.py
    │   └── settings.py
    └── scrapy.cfg

items.py 文件中，定义需要的字段

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


import scrapy


# 继承自Item父类
class YunnanGgzyItem(scrapy.Item):
    # 例如，我们需要类型、标题、时间、内容等等
    type = scrapy.Field()
    title = scrapy.Field()
    date = scrapy.Field()
    contents = scrapy.Field()

pipelines.py 文件中对 items 中的字段进行加工处理
主要针对爬取的数据进行处理，并将其保存到一个 Excel 文件中

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58


import openpyxl


class YunnanGgzyPipeline:
    def process_item(self, item, spider):
        # 载入或创建 Excel 文件
        try:
            wb = openpyxl.load_workbook('./excel_test.xlsx')
            # 创建名为 type 的 sheet 页
            if item['type'] not in wb.sheetnames:
                wb.create_sheet(item['type'])

        except Exception as e:
            print(e)
            print('新建 excel 文件')

            wb = openpyxl.Workbook() # 新建工作簿
            wb.create_sheet(item['type'])
            #table = data.get_sheet_by_name('Sheet1') # 获得指定名称页

        # 在工作表中写入表头
        table = wb[item['type']] # 获得当前活跃的工作页，默认为第一个工作页
        table.cell(1,1,'标题') # 行，列，值。从1开始计数的
        table.cell(1,2,'日期')
        table.cell(1,3,'内容')

        # 先遍历工作表中的数据，收集已有的标题、日期和内容
        unique_title = []
        unique_date = []
        unique_contents = []
        firstrow_flag = True
        for row in table.iter_rows():
            if firstrow_flag:
                firstrow_flag = False
                continue
            unique_title.append(row[0].value)
            unique_date.append(row[1].value)
            unique_contents.append(row[2].value)
        # 将三个 unique 列表数据转换成字典形式
        unique_dict = dict(zip(unique_title, zip(unique_date, unique_contents)))
        # 向 unique 字典中补充 item 中的新数据
        if item['title'] not in unique_dict.keys():
            unique_dict[item['title']] = (item['date'], item['contents'])
        else:
            if len(item['contents']) > len(unique_dict.get(item['title'])[1]):
                unique_dict[item['title']] = (item['date'], item['contents'])

        # 删除原 table 中的数据，并写入新数据
        table.delete_rows(2, table.max_row)
        nrows = table.max_row    # 获得行数
        for item in unique_dict.items():
            table.cell(nrows+1, 1).value = item[0]
            table.cell(nrows+1, 2).value = item[1][0]
            table.cell(nrows+1, 3).value = item[1][1]
            nrows += 1
        wb.save('./excel_test.xlsx')

        return item

settings.py 文件中，对 piplines 或者中间件进行设置
也可以配置 useragent 或者 ippool

重点的 spiders/yn_trade.py 文件中
创建 YnTradeSpider 类，继承自 scrapy.Spider

主要流程
先通过 scrapy.Request 发送 post 请求得到标题时间等 json 数据
之后发送 Splash 请求，得到 动态内容 的 html 页面
start_requests 方法
定义了一个 data 字典，存放需要的字段值，并将其转为 json 格式
随机选择 useragent
通过 scrapy.Request 发送 post 请求，callback 调用下面定义的 parse 方法
parse 方法
用于解析上面 scrapy.Request 请求得到的 JSON 响应
将请求返回的 JSON 字符串加载为字典对象 info_dict
进行数据的整理，提取需要的内容之后再次发送 SplashRequest 请求，获取动态页面的内容
parse_dynamic_content 方法
解析动态页面响应的内容
从 response 对象中提取存放在 meta 字典中的 item
使用 etree.HTML() 方法解析返回的 HTML
通过 XPath 提取目标内容
将提取到的内容 needContent 存入 item[‘contents’] 中

SplashRequest 请求相比于普通的请求，可以加载 js 文件实现动态页面的获取

🌍 NASA 数据下载自动化

需要注册账号
账号：nihao_2022
密码：Abc123456

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89


from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver import ActionChains
from time import sleep
import requests


# URL = input("请输入你的需要下载数据的页面网址: ")
# begintime, endtime = input('请输入想要的时间段,形式如: 2021-02-26,2021-03-01 ').split(',')
# region = input('请输入想要的位置,默认形式为: -180,-90,180,90 ')

# 配置测试参数
URL = 'https://disc.gsfc.nasa.gov/datasets/NHSNOWM_001/summary'
begin_time, end_time = '2013-02-26', '2013-03-01'
region = '-180,-90,180,90'

browser = webdriver.Edge() #设置 msedgedriver 所在路径
wait = WebDriverWait(browser, 15, 0.5) #设置等待时间，等待网页加载完成
browser.get(URL) #向目标网站发送请求
sleep(3)

# 模拟鼠标点击 Subset/Get Data
# 先等待网页直到 get data 按钮加载出来
get_data_button = wait.until(EC.presence_of_element_located((By.XPATH, '//*[@id="intro-dt-5"]/div[3]/span/button'))) #利用 Xpath 找到 get data 的位置
get_data_button.click() #鼠标进行点击
sleep(1)
# 下面的操作基本类似

# 模拟鼠标悬停在 Refine Date Range
time_select_button = wait.until(EC.presence_of_element_located((By.XPATH, "//span[@class='selectionSummary ng-binding']")))
ActionChains(browser).move_to_element(time_select_button).perform()
time_select_button.click()
# 找到填写起始 From 和结束时间 To 的输入框
begin_loc = browser.find_element(By.XPATH, "//input[@name='start']")
end_loc = browser.find_element(By.XPATH, "//input[@name='end']")
sleep(1)
# 先清除输入框中原本的时间，再将所需的时间传入
begin_loc.clear()
begin_loc.send_keys(begintime)
sleep(1)
end_loc.clear()
end_loc.send_keys(endtime)
sleep(1)

# 同理模拟鼠标悬停在输入区域信息，但是有些页面不存在可以输入位置区域的栏，故这里使用 try...except...
try:
    # 尝试寻找可以输入 region 的地方
    region_button = wait.until(EC.presence_of_element_located((By.XPATH, "//span[@class='row-3']")))
    ActionChains(browser).move_to_element(region_button).perform()
    region_button.click()
    region_loc = browser.find_element(By.XPATH, "//input[@name='inputBox']") #利用 Xpath 找到填写区域信息的栏
    # 填写所需的区域信息
    region_loc.clear()
    region_loc.send_keys(region)
    sleep(1)
except:
    pass

# 上面设置完后，点击绿色的 Get Data 按钮
getdata_button = browser.find_element(By.XPATH, "//button[@id='intro-subset-modal-9']") #利用 Xpath找到绿色的 Get Data按钮
getdata_button.click() #鼠标进行点击
sleep(1)


# 这里需要先等待搜索完成，否则将无法得到需要的下载链接
# 如果时间跨度大的话需要修改延长下面的等待时间，防止网页还没加载全就进行点击
sleep(20)
# 鼠标悬停在 Instructions for downloading
list_button = wait.until(EC.presence_of_element_located((By.XPATH, "//span[@class='popover-anchor']"))) #利用 Xpath 找到 Instructions for downloading
ActionChains(browser).move_to_element(list_button).perform() #将鼠标移动过去
list_links = wait.until(EC.presence_of_element_located((By.LINK_TEXT, "Download the list of links"))) #利用 link_text 找到下载 txt 链接的位置
list_of_links = list_links.get_attribute(name='href') #【获取标签里面真正需要的 txt 下载链接】
sleep(1)

# 下载得到存放着数据下载链接的 txt 文件
result = requests.get(list_of_links) #向刚刚获取的链接发送 get 请求
filename = 'listoflinks.txt'
try:
    result.raise_for_status() #抛出请求的状态码，常见的有 200、401、403、404等等
    with open(filename, 'wb') as f:
        f.write(result.content)
    print('txt 文件已经下载好，文件名为：' + FILENAME)
except:
    print('请求发送失败，返回的状态码为：' + str(result.status_code))

# 关闭浏览器
browser.quit()

遍历 txt 文件下载数据文件：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


import requests

def download_data_files():
    filename = 'listoflinks.txt'
    with open(filename) as data:
        for idx, url in enumerate(data, 1):
            if idx == 1:  # 跳过标题行
                continue

            url = url.strip('\n')  # 由于 txt文件读取出来后每一行后面自带‘\n’，故需要去掉
            file_id = '_'.join(url.split('/')[-3:])  # 观察链接，先将其按照‘/’进行拆分，再获取后面倒数三个字符串，最后利用‘-’进行拼接

            try:
                response = requests.get(url)
                if response.status_code == 200:
                    with open(file_id, 'wb') as f:
                        f.write(response.content)
                    print(f'第{idx}个文件下载完成: {file_id}')
                else:
                    print(f'第{idx}个文件请求失败: {response.status_code}')
            except Exception as e:
                print(f'第{idx}个文件下载异常: {str(e)}')

download_data_files()