0%

记一次成功的购物app爬虫

Posted on Edited on

与其说爬虫写的成功，不如说这个 app 实在太弱了 ^_^

之前阿妈在网上某 P2P 平台投资遇到资本家谎称破产卷款跑路留下漫天的债务。无奈阿妈为了追回本金，只能去和平台走清算流程，最后换回了等额的“金豆”（代金券）。这些代金券只能用于在某个 app 中购买一些丝毫没有性价比可言的产品。说实话我觉得挺惨的。

更惨的是，或许是因为清算的人实在太多了，app 里凡是智商税稍低一点的硬通货（比如手机、耳机和一些大牌家电），都属于一进货就被抢购一空的情况。而根据观察，每次进货的时间不固定，进货量又特别少。于是我就萌生了写个爬虫脚本挂机的想法，一有货就通知阿妈，彻底解放阿妈的双手。

常识告诉我所有和利益直接打交道的平台，一般都是网络攻防的重灾区。然而或许是这个平台真的没钱了或者只是因为太小众了，这个 app 简直弱爆了（说实话比上大 sso 还弱）。

探索

我不会移动端开发，也不会 Java。所以去做一个 app 上的爬虫可能全都是 from scratch 的。

但是常识告诉我，这个 app 大概率也只不过是个 web 而已，所以为了划归问题，获取 url 就成了当务之急。

apktool 和 classes-dex2jar：

首先想到的办法就是对 apk 拆包，直接分析源码拿到 url 。apktool 和 classes-dex2jar 就是两个经典的轻量级 apk 分析工具（shell 脚本）。根据后来的使用，前者似乎更倾向于把 apk 中的资源（音频、图标、视频等）分离出来，后者则只对 class.dex 做拆包得到源码。所以从理论上来说，后者更符合我的需求。

遗憾的是，并没有找到想要的 url 。究其原因，或许是这个 url 是运行时动态获取的，不保存在 apk 里。

抓包工具：

然后在 hecate 的提示下，我换了一种思路。既然反向工程很复杂，获取 url 又是外部动作，为何不简单的抓个包呢？后续选用的是 NetCapture 的抓包工具，可以直接在手机上进行抓包。原理异常简单，开一个系统代理，然后一切网络访问就都对 NetCapture 透明了。

于是我轻而易举的抓到了这个商城的 url，我现在能成功在浏览器上访问这个商城了。

爬取

然后，经过长达 10 分钟的探索，我发现这个商城简直太离谱了：

  • 居然没有任何异步加载和加密 script！
  • 居然没有任何令人头疼的证书和 token！
  • 居然不检查 header！
  • 居然不需要登录就能访问商城首页！
  • 居然不需要登录就能搜索特定关键字的商品！
  • 搜索功能的 post 的 response 是裸的 json！

爬虫脚本 10 分钟就给整好了，甚至没开 pycharm，文本编辑器写的。

剩下的就是做简单的异常处理，找个售罄的商品测试测试，找个未售罄的商品再测试测试。封装一下，加个邮件提醒，一气呵成。整个脚本从拿到 url 后一小时就搞定了。

部署

放本地太占地方，直接丢服务器上跑了。用 xftp 把文件传服务器上，然后 nohup 一下。

1
sudo nohup python3 main.py &
2
sudo ps -ef | grep main.py

然后就等着收邮件吧 QwQ，但愿阿妈能买到她心仪的商品 hhh

1
import requests
2
from time import sleep
3
from email.mime.text import MIMEText
4
from email.utils import formataddr
5
import smtplib
6
7
ENV_TEST = True
8
SEARCH_URL = 'the secret url'
9
JSON = {
10
    "xxx": "xxx",
11
    "keyword": None,
12
}
13
HEADER = {
14
    "User-Agent": "xxx mobile phone agent"
15
}
16
KEY_WORDS = ["key_word_1", "key_word_2"]
17
vis = [False for _ in range(len(KEY_WORDS))]
18
19
my_sender = ''  # 发件人邮箱账号
20
my_pass = ''  # 发件人邮箱口令
21
my_user = ''  # 收件人邮箱账号
22
23
24
def mail(title, content, coding_type, retry_flag=False):
25
    print('Mailing...', title)
26
    while True:
27
        try:
28
            msg = MIMEText(content, coding_type, 'utf-8')
29
            msg['From'] = formataddr(['someone', my_sender])
30
            msg['To'] = formataddr(['me', my_user])
31
            msg['Subject'] = title
32
            server = smtplib.SMTP_SSL('smtp.qq.com', 465)
33
            server.login(my_sender, my_pass)
34
            server.sendmail(my_sender, [my_user, ], msg.as_string())
35
            server.quit()
36
        except Exception as err:
37
            print(f"\n{err}")
38
            if not retry_flag:
39
                break
40
            sleep(5)
41
            continue
42
        return
43
44
45
def main_loop():
46
    try:
47
        for i in range(len(KEY_WORDS)):
48
            json = JSON
49
            json["keyword"] = KEY_WORDS[i]
50
            response = requests.post(SEARCH_URL, json=JSON, headers=HEADER, timeout=5)
51
            assert response.status_code == 200, f"status_code = {response.status_code}"
52
            result = eval(response.text.replace("null", "None").replace("false", "False").replace("true", "True"))
53
            result = result['result']['dataList']
54
            assert len(result) == 1, f"len(result) = {len(result)}"
55
            x = result[0]
56
            if x['inventory'] != 0 and not vis[i]:
57
                print()
58
                print("=" * 20)
59
                print(f"发现进货！ {x['productName']}")
60
                print(f"进货数量： {x['inventory']}")
61
                print("=" * 20)
62
                mail(
63
                    title=f"进货{x['inventory']} {KEY_WORDS[i]}",
64
                    content=f"发现进货！ {x['productName']} <br> 进货数量： {x['inventory']}",
65
                    coding_type="HTML",
66
                    retry_flag=True,
67
                )
68
                vis[i] = True
69
            elif x['inventory'] == 0 and vis[i]:
70
                print()
71
                print("=" * 20)
72
                print(f"发现售空！ {x['productName']}")
73
                print("=" * 20)
74
                mail(
75
                    title=f"售空 {KEY_WORDS[i]}",
76
                    content=f"发现售空！ {x['productName']}",
77
                    coding_type="HTML",
78
                    retry_flag=False,
79
                )
80
                vis[i] = False
81
            sleep(2)
82
    except Exception as err:
83
        print(f"\n{err}")
84
        return False
85
    return True
86
87
88
if __name__ == '__main__':
89
    if ENV_TEST:
90
        print("env test!")
91
        sleep(10)
92
        mail(
93
            title=f"脚本开始运行",
94
            content=f"脚本已经开始运行...",
95
            coding_type="HTML",
96
            retry_flag=True,
97
        )
98
    attempt_cnt = 0
99
    success_cnt = 0
100
    continuous_err_cnt = 0
101
    while True:
102
        success = main_loop()
103
        print(f"{f'x{attempt_cnt % 10}'[success]}", end="", flush=True)
104
        attempt_cnt += 1
105
        if attempt_cnt % 10 == 0:
106
            print()
107
        if attempt_cnt % 500 == 0:
108
            mail(
109
                title=f"脚本运行正常",
110
                content=f"attempt_cnt: {attempt_cnt} <br> success_cnt: {success_cnt}",
111
                coding_type="HTML",
112
                retry_flag=True,
113
            )
114
        if not success:
115
            continuous_err_cnt += 1
116
            if continuous_err_cnt >= 10:
117
                mail(
118
                    title=f"脚本异常退出",
119
                    content=f"attempt_cnt: {attempt_cnt} <br>  success_cnt: {success_cnt}",
120
                    coding_type="HTML",
121
                    retry_flag=True,
122
                )
123
                break
124
            sleep(3)
125
        else:
126
            success_cnt += 1
127
            continuous_err_cnt = 0
128
            sleep(1)