随便说 – Zeno

class MyThread(threading.Thread):
    def __init__(self, page):
        super().__init__()
        self.page = page

    def run(self) -&gt; None:
        while True:
            if self.page.empty():
                break
            self.get_info()

    def get_info(self):
        page = self.page.get()
        spider = BaiduSpider()
        response = spider.search_web(query="欧阳靖", pn=page)
        data = []
        for result in response["results"]:
            url = result.get("url")
            if result.get("url"):
                data.append(url)
                continue
            result_type = result.get("type")
            # 百科处理
            if result.get("result") and result_type != 'total':
                data.append(result.get("result").get("url"))
                continue
            # 其他处理
            if result_type != 'total' and result_type != "related":
                for i in result.get("results"):
                    data.append(i.get('url'))
        print(data)


# 启动线程队列
work_queue = queue.Queue()
# 传参
for i in range(1, 50):
    work_queue.put(i)

threads = []
for j in range(10):
    thread_do = MyThread(work_queue)
    thread_do.start()
    threads.append(thread_do)

for k in threads:
    k.join()

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

class MyThread(threading.Thread):

def __init__(self, page):

super().__init__()

self.page = page

def run(self) -> None:

while True:

if self.page.empty():

break

self.get_info()

def get_info(self):

page = self.page.get()

spider = BaiduSpider()

response = spider.search_web(query="欧阳靖", pn=page)

data = []

for result in response["results"]:

url = result.get("url")

if result.get("url"):

data.append(url)

continue

result_type = result.get("type")

# 百科处理

if result.get("result") and result_type != 'total':

data.append(result.get("result").get("url"))

continue

# 其他处理

if result_type != 'total' and result_type != "related":

for i in result.get("results"):

data.append(i.get('url'))

print(data)

# 启动线程队列

work_queue = queue.Queue()

# 传参

for i in range(1, 50):

work_queue.put(i)

threads = []

for j in range(10):

thread_do = MyThread(work_queue)

thread_do.start()

threads.append(thread_do)

for k in threads:

k.join()

Posted in 随便说Leave a Comment

by Zeno

Posted on 2021年12月29日2021年12月29日

AST对抗某网站的Js抽取型混淆

因为网站比较敏感, 所以具体网站就不说了, 直接说逻辑部分

为了降低数据提取的错误率, 所以使用了python的slimit库对js代码进行提取处理

对网站源码的JavaScript进行分析后, 发现在其中一个script标签内的js代码是利用抽取混淆的, 并用flashvars开头的变量存储

首先用python将该js代码进行提取

response = requests.get(url=url, proxies=proxies)
script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

1 2	response = requests.get(url=url, proxies=proxies) script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

将该部分代码格式化后可以看到键名为mediaDefinitions的值是数组, 数组内存储着videoUrl

在这里可以看到, 链接是被抽取的进行拼接后即可还原真实地址

下面会使用到python的slimit库的ast进行还原

# 将js代码转成结构树
tree = Parser().parse(script)

1 2	# 将js代码转成结构树 tree = Parser().parse(script)

通过smlit的Parser类的parse方法, 对js代码转换为ast结构树

获取到结构树后, 需要自定义类, 并继承ASTVisitor, 自定义访问者遍历节点对节点进行抽取

下面代码我先对mediaDefinitions和qualityItems对象进行抽取还原

class VarStatement_Visitor(ASTVisitor):

    # 自定义访问者，重写VarStatement节点访问逻辑
    def visit_VarStatement(self, node):
        Identifier, Object = node.children()[0].children()
        # 获取flashvars的节点
        if 'flashvars' in Identifier.value:
            for i in Object.properties:
                left, right = i.children()
                # mediaDefinitions数组
                if left.value == '"mediaDefinitions"':
                    # 还原字典
                    for item in right.items:
                        media_data = dict()
                        for medias in item.properties:
                            media_left, media_right = medias.children()
                            if isinstance(media_right, ast.Array):
                                data_list = [i.value for i in media_right.items]
                                media_data[media_left.value[1:-1]] = data_list
                            else:
                                if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':
                                    media_data[media_left.value[1:-1]] = media_right.value
                                else:
                                    media_data[media_left.value[1:-1]] = media_right.value[1:-1]
                        flashvars.append(media_data)
        # 获取qualityItems的节点
        if 'qualityItems' in Identifier.value:
            for i in Object.items:
                media_data = dict()
                for medias in i.properties:
                    media_left, media_right, = medias.children()
                    if isinstance(media_right, ast.Number):
                        media_data[media_left.value[1:-1]] = media_right.value
                    else:
                        media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')
                flashvars.append(media_data)

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

class VarStatement_Visitor(ASTVisitor):

# 自定义访问者，重写VarStatement节点访问逻辑

def visit_VarStatement(self, node):

Identifier, Object = node.children()[0].children()

# 获取flashvars的节点

if 'flashvars' in Identifier.value:

for i in Object.properties:

left, right = i.children()

# mediaDefinitions数组

if left.value == '"mediaDefinitions"':

# 还原字典

for item in right.items:

media_data = dict()

for medias in item.properties:

media_left, media_right = medias.children()

if isinstance(media_right, ast.Array):

data_list = [i.value for i in media_right.items]

media_data[media_left.value[1:-1]] = data_list

else:

if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1]

flashvars.append(media_data)

# 获取qualityItems的节点

if 'qualityItems' in Identifier.value:

for i in Object.items:

media_data = dict()

for medias in i.properties:

media_left, media_right, = medias.children()

if isinstance(media_right, ast.Number):

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')

flashvars.append(media_data)

还原mediaDefinitions和qualityItems对象后, 继续创建访问者对抽取链接进行还原

进入的节点仍然是VarStatement, 因为视频的链接由多个变量并且不定数量进行拼接, 所以下面使用递归方式并使用了get_Identifier方法对抽取变量进行获取还原

class Media_Visitor(ASTVisitor):

    def __init__(self, i, *args, **kwargs):
        # 视频所在的序号
        self.i = i
        # 映射关系
        self.identifier = {}
        # 映射顺序
        self.identifiers_list = []
        super(*args, **kwargs)

    # 递归获取映射顺序
    def get_Identifier(self, node, identifiers_list):
        left, right = node.children()
        identifiers_list.append(self.identifier[right.value])
        if isinstance(left, ast.BinOp):
            self.get_Identifier(left, identifiers_list)
        else:
            identifiers_list.append(self.identifier[left.value])

    def visit_VarStatement(self, node):
        Identifier, BinOp = node.children()[0].children()
        # 获取该函数地址的映射顺序
        if 'media_' + str(self.i) in Identifier.value:
            # 计算该视频的真实地址
            self.get_Identifier(BinOp, self.identifiers_list)
            # 填充视频地址
            flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])
        # 根据映射关系进行恢复
        elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):
            if isinstance(BinOp, ast.String):
                self.identifier[Identifier.value] = BinOp.value[1:-1]
            else:
                self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

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

class Media_Visitor(ASTVisitor):

def __init__(self, i, *args, **kwargs):

# 视频所在的序号

self.i = i

# 映射关系

self.identifier = {}

# 映射顺序

self.identifiers_list = []

super(*args, **kwargs)

# 递归获取映射顺序

def get_Identifier(self, node, identifiers_list):

left, right = node.children()

identifiers_list.append(self.identifier[right.value])

if isinstance(left, ast.BinOp):

self.get_Identifier(left, identifiers_list)

else:

identifiers_list.append(self.identifier[left.value])

def visit_VarStatement(self, node):

Identifier, BinOp = node.children()[0].children()

# 获取该函数地址的映射顺序

if 'media_' + str(self.i) in Identifier.value:

# 计算该视频的真实地址

self.get_Identifier(BinOp, self.identifiers_list)

# 填充视频地址

flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])

# 根据映射关系进行恢复

elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):

if isinstance(BinOp, ast.String):

self.identifier[Identifier.value] = BinOp.value[1:-1]

else:

self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

下面是完整代码:

import re
import time
import traceback
import requests_html
# 将js代码转为AST结构树
from slimit.parser import Parser
# 自定义访问者
from slimit.visitors.nodevisitor import ASTVisitor
# 判断类型
from slimit import ast
import os

requests = requests_html.HTMLSession()
flashvars = list()


class VarStatement_Visitor(ASTVisitor):

    # 自定义访问者，重写VarStatement节点访问逻辑
    def visit_VarStatement(self, node):
        Identifier, Object = node.children()[0].children()
        # 获取flashvars的节点
        if 'flashvars' in Identifier.value:
            for i in Object.properties:
                left, right = i.children()
                # mediaDefinitions数组
                if left.value == '"mediaDefinitions"':
                    # 还原字典
                    for item in right.items:
                        media_data = dict()
                        for medias in item.properties:
                            media_left, media_right = medias.children()
                            if isinstance(media_right, ast.Array):
                                data_list = [i.value for i in media_right.items]
                                media_data[media_left.value[1:-1]] = data_list
                            else:
                                if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':
                                    media_data[media_left.value[1:-1]] = media_right.value
                                else:
                                    media_data[media_left.value[1:-1]] = media_right.value[1:-1]
                        flashvars.append(media_data)
        # 获取qualityItems的节点
        if 'qualityItems' in Identifier.value:
            for i in Object.items:
                media_data = dict()
                for medias in i.properties:
                    media_left, media_right, = medias.children()
                    if isinstance(media_right, ast.Number):
                        media_data[media_left.value[1:-1]] = media_right.value
                    else:
                        media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')
                flashvars.append(media_data)


class Media_Visitor(ASTVisitor):

    def __init__(self, i, *args, **kwargs):
        # 视频所在的序号
        self.i = i
        # 映射关系
        self.identifier = {}
        # 映射顺序
        self.identifiers_list = []
        super(*args, **kwargs)

    # 递归获取映射顺序
    def get_Identifier(self, node, identifiers_list):
        left, right = node.children()
        identifiers_list.append(self.identifier[right.value])
        if isinstance(left, ast.BinOp):
            self.get_Identifier(left, identifiers_list)
        else:
            identifiers_list.append(self.identifier[left.value])

    def visit_VarStatement(self, node):
        Identifier, BinOp = node.children()[0].children()
        # 获取该函数地址的映射顺序
        if 'media_' + str(self.i) in Identifier.value:
            # 计算该视频的真实地址
            self.get_Identifier(BinOp, self.identifiers_list)
            # 填充视频地址
            flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])
        # 根据映射关系进行恢复
        elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):
            if isinstance(BinOp, ast.String):
                self.identifier[Identifier.value] = BinOp.value[1:-1]
            else:
                self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])
                

def main(url):
    response = requests.get(url=url)
    script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()
    # 将js代码转成结构树
    tree = Parser().parse(script)
    VarStatement_Visitor().visit(tree)
    for i, _ in enumerate(flashvars):
        Media_Visitor(i).visit(tree)
    print(flashvars)

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

90

91

92

93

94

95

96

97

98

99

import re

import time

import traceback

import requests_html

# 将js代码转为AST结构树

from slimit.parser import Parser

# 自定义访问者

from slimit.visitors.nodevisitor import ASTVisitor

# 判断类型

from slimit import ast

import os

requests = requests_html.HTMLSession()

flashvars = list()

class VarStatement_Visitor(ASTVisitor):

# 自定义访问者，重写VarStatement节点访问逻辑

def visit_VarStatement(self, node):

Identifier, Object = node.children()[0].children()

# 获取flashvars的节点

if 'flashvars' in Identifier.value:

for i in Object.properties:

left, right = i.children()

# mediaDefinitions数组

if left.value == '"mediaDefinitions"':

# 还原字典

for item in right.items:

media_data = dict()

for medias in item.properties:

media_left, media_right = medias.children()

if isinstance(media_right, ast.Array):

data_list = [i.value for i in media_right.items]

media_data[media_left.value[1:-1]] = data_list

else:

if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1]

flashvars.append(media_data)

# 获取qualityItems的节点

if 'qualityItems' in Identifier.value:

for i in Object.items:

media_data = dict()

for medias in i.properties:

media_left, media_right, = medias.children()

if isinstance(media_right, ast.Number):

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')

flashvars.append(media_data)

class Media_Visitor(ASTVisitor):

def __init__(self, i, *args, **kwargs):

# 视频所在的序号

self.i = i

# 映射关系

self.identifier = {}

# 映射顺序

self.identifiers_list = []

super(*args, **kwargs)

# 递归获取映射顺序

def get_Identifier(self, node, identifiers_list):

left, right = node.children()

identifiers_list.append(self.identifier[right.value])

if isinstance(left, ast.BinOp):

self.get_Identifier(left, identifiers_list)

else:

identifiers_list.append(self.identifier[left.value])

def visit_VarStatement(self, node):

Identifier, BinOp = node.children()[0].children()

# 获取该函数地址的映射顺序

if 'media_' + str(self.i) in Identifier.value:

# 计算该视频的真实地址

self.get_Identifier(BinOp, self.identifiers_list)

# 填充视频地址

flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])

# 根据映射关系进行恢复

elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):

if isinstance(BinOp, ast.String):

self.identifier[Identifier.value] = BinOp.value[1:-1]

else:

self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

def main(url):

response = requests.get(url=url)

script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

# 将js代码转成结构树

tree = Parser().parse(script)

VarStatement_Visitor().visit(tree)

for i, _ in enumerate(flashvars):

Media_Visitor(i).visit(tree)

print(flashvars)

Posted in 随便说Leave a Comment

by Zeno

Posted on 2021年12月29日2021年12月29日

[python] Django宝塔线上部署踩坑记录

原本的Django本地运行的时候是正常的, 当部署线上的时候就一大堆问题出现了.

因为我对Linux并不熟悉, 所以就选择了宝塔的python项目管理器来部署

在部署上线之前, 要添加几个文件.

常规的要在项目根目录加上uwsgi.ini文件, 文件内容如下

[uwsgi]
master = true
processes = 1
threads = 2
chdir = /www/wwwroot/api
wsgi-file = /www/wwwroot/api/zeno_api/wsgi.py  # wsgi文件的绝对路径
http = 0.0.0.0:520  # 0.0.0.0:端口号
logto = /www/wwwroot/api/logs/error.log  # 日志保存路径
chmod-socket = 660
vacuum = true
max-requests = 1000

1

2

3

4

5

6

7

8

9

10

11

[uwsgi]

master = true

processes = 1

threads = 2

chdir = /www/wwwroot/api

wsgi-file = /www/wwwroot/api/zeno_api/wsgi.py # wsgi文件的绝对路径

http = 0.0.0.0:520 # 0.0.0.0:端口号

logto = /www/wwwroot/api/logs/error.log # 日志保存路径

chmod-socket = 660

vacuum = true

max-requests = 1000

以上内容需要注意的是, wsgi-file填写的是wsgi.py文件的绝对路径加上文件夹名字. 网上很多都是只填路径的, 部署后会直接报错. 而wsgi.py是Django在建项目的时候就自带了的, 不需修改

部署成功之后, 如果有域名可以映射到域名上, 点击上面的映射填上域名即可, 映射成功之后可以使用域名访问

如果有使用Django自带的admin后台, 会发现项目部署后admin后台的样式会没了. 找了好几天百度都没有解决, 后来还是看官方文档解决的…

首先是要在settings.py下加上STATIC_ROOT, 然后在这个路径下手动创建一个static文件夹

STATIC_ROOT = '/www/wwwroot/api/zeno_api/static/'

1	STATIC_ROOT = '/www/wwwroot/api/zeno_api/static/'

然后再去主app下的urls.py修改以下代码

from django.conf import settings
from django.conf.urls.static import static

urlpatterns = [
    # ... the rest of your URLconf goes here ...
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

1

2

3

4

5

6

from django.conf import settings

from django.conf.urls.static import static

urlpatterns = [

# ... the rest of your URLconf goes here ...

] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

修改完之后去服务器切换到你Django所使用的python环境, 去到你Django项目的目录, 然后运行以下命令

python manage.py collectstatic

1	python manage.py collectstatic

官方文档链接: https://docs.djangoproject.com/zh-hans/3.2/howto/static-files/

注: 记得每次修改后重启一下服务才能生效

Posted in 随便说Tagged django, PythonLeave a Comment

by Zeno

Posted on 2021年12月29日2021年12月29日

某车号标准加密

这次分析的是某车号的一个列表请求头参数加密

网站: aHR0cHM6Ly9oYW8ueWljaGUuY29tLw==

接口: aHR0cHM6Ly9oYW8ueWljaGUuY29tL3NpdGVfd2ViL2hhby9hcGkvZ2V0X2xhdGVzdF9hcnRpY2xlX2xpc3Q=

参数分析:

初步抓包发现部分可疑参数:

x-sign: 目测疑似md5加密

CIGUID, UserGuid,x-user-guid: 相等, 看似uuid之类的随机生成id

至于其他的x-city-id, x-ip-address应该是记录该次请求的ip和ip所在地域, 其他参数并无可疑之处

直接全局搜索x-sign之后只有一个结果, 点击进去查看发现就只有一个地方有x-sign:

打断点后重新发起请求发现r值是32位大写的字符串, 并且下面的sign加密是需要用到该值的, 所以需要分析r值的获取方法

分析o的方法之后发现这个值可以固定, 是由e.headerEncryptKeys对象里面根据请求头的x-platform获取的不一样的固定值, 因为他的默认请求是phone所对应的value所以r就可以固定为该值

u值就是由字符串拼接而成的, e.cid也是固定默认值601, i是请求体的json格式, r是上面所获得的固定值, e.timestamp自然就是13位的时间戳

u值获取之后, 就到了下面的yicheUtils.md5方法, 看名字是md5, 但是需要验证一下是否md5和是否标准加密. 可以在console里面试一下加密固定值的md5(例如123456) 是否跟标准md5加密一致

确认是标准md5加密算法, 本次分析完毕.

再加一句, 里面的x-user-guid其实是随机的py的uuid随便搞

Posted in 随便说Tagged Python, 爬虫Leave a Comment

by Zeno

Posted on 2021年7月6日2021年7月6日

Stative Verbs Used in the Continuous Form

Some verbs describe a state (and are known as stative), and they aren’t usually used in the continuous (or progressive) form. These verbs include know, believe, like, recognize, etc.

These verbs are not used in the continuous form because the state lasts for a longer period of time, is not an action, and does not have a definite start or finish.

I like classical music. (NOT ~~I’m liking~~ classical music.)

I believe in you. (NOT ~~I’m believing~~ in you.)

We can use some stative verbs like feel, look, hear, taste, see in the continuous form when describing an action or where there is a clear beginning and end.

I feel good doing this. (sense / state)

I’m not feeling well. (action with a clear beginning and end)

I hear a lot of noise. (sense)

I’m hearing a lot of noise. (action with a clear beginning and end)

I can see him. (sense)

I’m seeing him tonight. (action to meet someone with a clear beginning and end)

I think it’s a good idea. (sense / state)

I’m thinking of moving. (action of trying to reach a decision with a clear beginning and end)

I can taste cinnamon. (sense)

I’m tasting the dessert. (action of tasting with a clear beginning and end)

You look good. (sense)

I’m looking for a new job. (action of searching for something with a clear beginning and end)

Posted in 随便说Leave a Comment

by Lii025

Posted on 2020年12月29日2021年6月10日2020年12月29日2021年6月10日

密码保护：叽-12

Posted in 随便说Tagged 满

by Zeno

Posted on 2020年8月30日2020年8月29日2020年8月30日2020年8月29日

【Python】爬虫(Xpath):批量爬取站长免费简历

批量爬取站长免费简历, 可以自行选择爬取页数.

from lxml import etree
import requests
import os


# 封装解析下载函数
def cv_down(tree, headers):
    cv_href = tree.xpath('//div[@class="sc_warp  mt20"]/div/div/div/a/@href')
    for href in cv_href:
        act_response = requests.get(url=href, headers=headers).text
        act_tree = etree.HTML(act_response)
        cv_title = act_tree.xpath('//div[@class="ppt_tit clearfix"]/h1/text()')
        cv_title = cv_title[0].encode('ISO-8859-1').decode('utf-8') + '.rar'
        dow_url = act_tree.xpath('//div[@class="clearfix mt20 downlist"]/ul/li[1]/a/@href')[0]
        doc = requests.get(url=dow_url, headers=headers).content
        cv_path = './免费简历/' + cv_title
        with open(cv_path, 'wb') as fp:
            fp.write(doc)
            print(cv_title, '下载完成！！！')


# 检查文件夹是否存在,并创建文件夹
if not os.path.exists('./免费简历'):
    os.mkdir('./免费简历')

first_page_url = 'http://sc.chinaz.com/jianli/free.html'
headers = {
    "User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.92 Safari/537.36'
}

tree = etree.HTML(requests.get(url=first_page_url, headers=headers).text)
page_limit = tree.xpath('//div[@class="pagination fr clearfix clear"]/a[8]/b/text()')[0]

while True:
    print(page_limit + " pages at most")
    page = input("Please enter how many page you want: ")

    if page.isdigit() and 1 <= int(page) <= int(page_limit):
        for i in range(1, int(page) + 1):
            if i == 1:
                cv_down(tree, headers)
            else:
                other_page_url = 'http://sc.chinaz.com/jianli/free_' + str(i) + '.html'  # 一页以上的特殊性需要重新制定链接
                response = requests.get(url=other_page_url, headers=headers).text
                tree = etree.HTML(response)
                cv_down(tree, headers)
        break
    else:
        print("You can only enter right numbers!!!")
        continue

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

from lxml import etree

import requests

import os

# 封装解析下载函数

def cv_down(tree, headers):

cv_href = tree.xpath('//div[@class="sc_warp mt20"]/div/div/div/a/@href')

for href in cv_href:

act_response = requests.get(url=href, headers=headers).text

act_tree = etree.HTML(act_response)

cv_title = act_tree.xpath('//div[@class="ppt_tit clearfix"]/h1/text()')

cv_title = cv_title[0].encode('ISO-8859-1').decode('utf-8') + '.rar'

dow_url = act_tree.xpath('//div[@class="clearfix mt20 downlist"]/ul/li[1]/a/@href')[0]

doc = requests.get(url=dow_url, headers=headers).content

cv_path = './免费简历/' + cv_title

with open(cv_path, 'wb') as fp:

fp.write(doc)

print(cv_title, '下载完成！！！')

# 检查文件夹是否存在,并创建文件夹

if not os.path.exists('./免费简历'):

os.mkdir('./免费简历')

first_page_url = 'http://sc.chinaz.com/jianli/free.html'

headers = {

"User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.92 Safari/537.36'

}

tree = etree.HTML(requests.get(url=first_page_url, headers=headers).text)

page_limit = tree.xpath('//div[@class="pagination fr clearfix clear"]/a[8]/b/text()')[0]

while True:

print(page_limit + " pages at most")

page = input("Please enter how many page you want: ")

if page.isdigit() and 1 <= int(page) <= int(page_limit):

for i in range(1, int(page) + 1):

if i == 1:

cv_down(tree, headers)

else:

other_page_url = 'http://sc.chinaz.com/jianli/free_' + str(i) + '.html' # 一页以上的特殊性需要重新制定链接

response = requests.get(url=other_page_url, headers=headers).text

tree = etree.HTML(response)

cv_down(tree, headers)

break

else:

print("You can only enter right numbers!!!")

continue

Posted in 随便说Tagged Python, Xpath, 爬虫Leave a Comment

by Zeno

Posted on 2020年8月29日2020年8月29日

【Python】爬虫(Xpath):批量爬取站长免费PPT模版

本来是打算写一个异步爬取的,但是暂时还没有发现是什么问题好像没有实现异步

import requests
from lxml import etree
import os
import asyncio
import aiohttp


def get_ppt_list(page_need, headers):
    for i in range(1, int(page_need)+1):
        if i == 1:
            url = 'http://sc.chinaz.com/ppt/free.html'
            list_page_text = requests.get(url=url, headers=headers).text
            list_page_tree = etree.HTML(list_page_text)
            list_ppt_url = list_page_tree.xpath('//*[@id="main"]/div/div/a/@href')
            tasks = []
            for url in list_ppt_url:
                c = get_ppt(url, headers)
                task = asyncio.ensure_future(c)
                tasks.append(task)
            loop = asyncio.get_event_loop()
            loop.run_until_complete(asyncio.wait(tasks))

        else:
            url = 'http://sc.chinaz.com/ppt/free_{}.html'.format(i)
            list_page_text = requests.get(url=url, headers=headers).text
            list_page_tree = etree.HTML(list_page_text)
            list_ppt_url = list_page_tree.xpath('//*[@id="main"]/div/div/a/@href')
            tasks = []
            for url in list_ppt_url:
                c = get_ppt(url, headers)
                task = asyncio.ensure_future(c)
                tasks.append(task)
            loop = asyncio.get_event_loop()
            loop.run_until_complete(asyncio.wait(tasks))


async def get_ppt(url, headers):
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url, headers=headers) as ppt_page:
            ppt_page_text = await ppt_page.text()
            ppt_page_tree = etree.HTML(ppt_page_text)
            title = ppt_page_tree.xpath('//div[@class="ppt_tit clearfix"]/h1/text()')[0]
            title = title + '.rar'
            ppt_path = './站长PPT/' + title
            down_url = ppt_page_tree.xpath('//*[@id="down"]/div[2]/ul/li[3]/a/@href')[0]
            async with await session.get(url=down_url, headers=headers) as ppt:
                ppt = await ppt.read()
                with open(ppt_path, 'wb') as fp:
                    fp.write(ppt)
                print("{} 下载成功!!!".format(title))


url = 'http://sc.chinaz.com/ppt/free.html'
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.92 Safari/537.36',
}

list_page_text = requests.get(url=url, headers=headers).text
list_page_tree = etree.HTML(list_page_text)
page_limit = list_page_tree.xpath('//div[@class="pagination fr clearfix clear"]/a[8]/b/text()')[0]


while True:
    print("该模板页数上限是{}, 请输入您所需要的总页数".format(page_limit))
    page_need = input("<<<")
    if page_need.isdigit() and 1 <= int(page_need) <= int(page_limit):
        if not os.path.exists('./站长PPT'):
            os.mkdir('./站长PPT')
        get_ppt_list(page_need, headers)
        break
    else:
        print("请重新输入正确的页数")
        continue

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

import requests

from lxml import etree

import os

import asyncio

import aiohttp

def get_ppt_list(page_need, headers):

for i in range(1, int(page_need)+1):

if i == 1:

url = 'http://sc.chinaz.com/ppt/free.html'

list_page_text = requests.get(url=url, headers=headers).text

list_page_tree = etree.HTML(list_page_text)

list_ppt_url = list_page_tree.xpath('//*[@id="main"]/div/div/a/@href')

tasks = []

for url in list_ppt_url:

c = get_ppt(url, headers)

task = asyncio.ensure_future(c)

tasks.append(task)

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

else:

url = 'http://sc.chinaz.com/ppt/free_{}.html'.format(i)

list_page_text = requests.get(url=url, headers=headers).text

list_page_tree = etree.HTML(list_page_text)

list_ppt_url = list_page_tree.xpath('//*[@id="main"]/div/div/a/@href')

tasks = []

for url in list_ppt_url:

c = get_ppt(url, headers)

task = asyncio.ensure_future(c)

tasks.append(task)

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

async def get_ppt(url, headers):

async with aiohttp.ClientSession() as session:

async with await session.get(url=url, headers=headers) as ppt_page:

ppt_page_text = await ppt_page.text()

ppt_page_tree = etree.HTML(ppt_page_text)

title = ppt_page_tree.xpath('//div[@class="ppt_tit clearfix"]/h1/text()')[0]

title = title + '.rar'

ppt_path = './站长PPT/' + title

down_url = ppt_page_tree.xpath('//*[@id="down"]/div[2]/ul/li[3]/a/@href')[0]

async with await session.get(url=down_url, headers=headers) as ppt:

ppt = await ppt.read()

with open(ppt_path, 'wb') as fp:

fp.write(ppt)

print("{} 下载成功!!!".format(title))

url = 'http://sc.chinaz.com/ppt/free.html'

headers = {

'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.92 Safari/537.36',

}

list_page_text = requests.get(url=url, headers=headers).text

list_page_tree = etree.HTML(list_page_text)

page_limit = list_page_tree.xpath('//div[@class="pagination fr clearfix clear"]/a[8]/b/text()')[0]

while True:

print("该模板页数上限是{}, 请输入您所需要的总页数".format(page_limit))

page_need = input("<<<")

if page_need.isdigit() and 1 <= int(page_need) <= int(page_limit):

if not os.path.exists('./站长PPT'):

os.mkdir('./站长PPT')

get_ppt_list(page_need, headers)

break

else:

print("请重新输入正确的页数")

continue

Posted in 随便说Tagged aiohttp, asyncio, Python, Xpath, 异步, 爬虫Leave a Comment