akshare获取reits数据,搞笑,数据源测试过没有呀,没有集思录会员还没获取?


描述: 集思录-实时数据-REITs-A股 REITs

限量: 单次返回所有 REITs 的基本信息数据
 



20221107004.jpg

估计是旧的,以前游客就可以获取到数据的了。现在就不行了。
 

20221107003.jpg

 
然后另外一个获取东财的函数,获取了20个就完事了。
结果人间现在超过20家,获取的数据都不全的。
 
akshare,用的还是有点糟心。长期来看,还是自己写靠谱,有些坑,你第一次用就可以发现,但是有些坑,却是埋在那里,像个定时炸弹。
 
 
20221108003.jpg


akshare的代码:

20221108004.jpg


 

 
继续阅读 »


描述: 集思录-实时数据-REITs-A股 REITs

限量: 单次返回所有 REITs 的基本信息数据
 



20221107004.jpg

估计是旧的,以前游客就可以获取到数据的了。现在就不行了。
 

20221107003.jpg

 
然后另外一个获取东财的函数,获取了20个就完事了。
结果人间现在超过20家,获取的数据都不全的。
 
akshare,用的还是有点糟心。长期来看,还是自己写靠谱,有些坑,你第一次用就可以发现,但是有些坑,却是埋在那里,像个定时炸弹。
 
 
20221108003.jpg


akshare的代码:

20221108004.jpg


 

  收起阅读 »

星球文章 获取所有文章 爬虫

挺讽刺的,星球自身没有目录结构,所以浏览文章,只能通过时间线,一些一年前的文章,基本就没有会看到,而且,星球的搜索功能也是一堆bug,处于搜不到的状态。
 
Screenshot_2022_0603_133612.jpg

群里没有人没有吐槽过这个搜索功能的。
 
所以只好自己写个程序把自己的文章抓下来,作为文章目录:

20220603002.png


20220603001.png

 
生成的markdown文件
20220603004.png


每次只需要运行python main.py 就可以拿到最新的星球文章链接了。 
 
需要源码可以在公众号联系~
 
继续阅读 »
挺讽刺的,星球自身没有目录结构,所以浏览文章,只能通过时间线,一些一年前的文章,基本就没有会看到,而且,星球的搜索功能也是一堆bug,处于搜不到的状态。
 
Screenshot_2022_0603_133612.jpg

群里没有人没有吐槽过这个搜索功能的。
 
所以只好自己写个程序把自己的文章抓下来,作为文章目录:

20220603002.png


20220603001.png

 
生成的markdown文件
20220603004.png


每次只需要运行python main.py 就可以拿到最新的星球文章链接了。 
 
需要源码可以在公众号联系~
  收起阅读 »

知识星球获取文章链接与数据


 
既然官方不提供这个功能,只能自己使用爬虫手段获取了,额。

 
既然官方不提供这个功能,只能自己使用爬虫手段获取了,额。

韦世东 python3网络爬虫宝典 勘误

P231:
1. 时间差是正数才是过期
2. 获取权限那里,permission = args[0].permission
不能后面再用get
 
P222:
写入mongodb后,原来的数据会被加入一个_id,值为OjectId,该值是无法被json dumps为string,
所以需要手工把ObjectId 转为str,或者del message['_id'] 将这个键去除。
 

 
继续阅读 »
P231:
1. 时间差是正数才是过期
2. 获取权限那里,permission = args[0].permission
不能后面再用get
 
P222:
写入mongodb后,原来的数据会被加入一个_id,值为OjectId,该值是无法被json dumps为string,
所以需要手工把ObjectId 转为str,或者del message['_id'] 将这个键去除。
 

  收起阅读 »

pyautogui无法再远程桌面最小化或者断线后进行截图

搜索了一圈,似乎无解。
知道的朋友可以私信下我。
 
国外的论坛也找不到答案,只能一直开着屏幕了。。。。
搜索了一圈,似乎无解。
知道的朋友可以私信下我。
 
国外的论坛也找不到答案,只能一直开着屏幕了。。。。

pyppeteer 在AppData下的dev_profile 生成大量文件

具体路径在:
C:\Users\xda\AppData\Local\pyppeteer\pyppeteer\.dev_profile
运行次数多了,这个目录下积累了几十个G的文件。
 

DV0VxpbNtN.png

 
因为每次启动pyppeteer后,如果不指定userData目录,会在dev_profile生成一个新的userData目录,每次大概30MB左右的打小,所以启动的次数,越多,这个文件夹的体积就越大。
其实可以直接删除,然后启动pyppeteer是加上一个参数:userDataDir
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000},
# 'enable-automation':False,
# 'ignoreDefaultArgs':['--enable-automation'],
'ignoreDefaultArgs':True,
}
userDataDir='D:\Temp'
这样每次pyppeteer都会用同一个配置文件,并且还可以把cookies,session文件存在同一个地方,如果登录过的网站,下次可以直接登录,不需要再次输入账号密码。
 
 
继续阅读 »
具体路径在:
C:\Users\xda\AppData\Local\pyppeteer\pyppeteer\.dev_profile
运行次数多了,这个目录下积累了几十个G的文件。
 

DV0VxpbNtN.png

 
因为每次启动pyppeteer后,如果不指定userData目录,会在dev_profile生成一个新的userData目录,每次大概30MB左右的打小,所以启动的次数,越多,这个文件夹的体积就越大。
其实可以直接删除,然后启动pyppeteer是加上一个参数:userDataDir
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000},
# 'enable-automation':False,
# 'ignoreDefaultArgs':['--enable-automation'],
'ignoreDefaultArgs':True,
}
userDataDir='D:\Temp'
这样每次pyppeteer都会用同一个配置文件,并且还可以把cookies,session文件存在同一个地方,如果登录过的网站,下次可以直接登录,不需要再次输入账号密码。
 
  收起阅读 »

pymongo update_one/update_many 返回更新数据的数目

比如有以下的python语句
ret = self.doc.update_one({'announcementId':announcementId},{'$setOnInsert':item},upsert=True)

作用是如果某个id不存在时,则把数据插入文档,如果存在,则不进行任何操作。
 
那么ret是操作的返回结果。
我们可以对返回数据进行核对
 
ret.matched_count, ret.modified_count
前者是匹配到文档的数目,而后者是修改了文档的个数。
 
 
继续阅读 »
比如有以下的python语句
ret = self.doc.update_one({'announcementId':announcementId},{'$setOnInsert':item},upsert=True)

作用是如果某个id不存在时,则把数据插入文档,如果存在,则不进行任何操作。
 
那么ret是操作的返回结果。
我们可以对返回数据进行核对
 
ret.matched_count, ret.modified_count
前者是匹配到文档的数目,而后者是修改了文档的个数。
 
  收起阅读 »

pyppeteer-pdf not support latest version of pypeteer

只能用0.0.25版本的pypeteer。
只好下载一个低版本的chrouium放到本地。

或者使用另一个库
https://github.com/shivanshs9/pdfgen-python
 
只能用0.0.25版本的pypeteer。
只好下载一个低版本的chrouium放到本地。

或者使用另一个库
https://github.com/shivanshs9/pdfgen-python
 

pyppeteer禁用自动化提示栏 --enable-automation参数关闭

在启动的时候加入一个参数即可:
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000},
# 'enable-automation':False,
'ignoreDefaultArgs':['--enable-automation'],
}
)

忽略默认参数:
'ignoreDefaultArgs':['--enable-automation'],
即可,
如果需要去除更多的特征,在列表里面加入更多即可。
如果全部不要,那么 就把它设置为True即可。

 
继续阅读 »
在启动的时候加入一个参数即可:
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000},
# 'enable-automation':False,
'ignoreDefaultArgs':['--enable-automation'],
}
)

忽略默认参数:
'ignoreDefaultArgs':['--enable-automation'],
即可,
如果需要去除更多的特征,在列表里面加入更多即可。
如果全部不要,那么 就把它设置为True即可。

  收起阅读 »

pyppeteer设置浏览器大小

在初始化时加入 defaultViewport字典值即可:
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000}
}
)

更多参数可以查看pyppeteer的源码。
        self.handleSIGINT = options.get('handleSIGINT', True)
self.handleSIGTERM = options.get('handleSIGTERM', True)
self.handleSIGHUP = options.get('handleSIGHUP', True)
self.ignoreHTTPSErrors = options.get('ignoreHTTPSErrors', False)
self.defaultViewport = options.get('defaultViewport', {'width': 800, 'height': 600}) # noqa: E501
self.slowMo = options.get('slowMo', 0)
self.timeout = options.get('timeout', 30000)
self.autoClose = options.get('autoClose', True)

 
继续阅读 »
在初始化时加入 defaultViewport字典值即可:
    browser = await pyppeteer.launch(
{'headless': False,
'userDataDir': UserDataDir,
'defaultViewport': {'width': 1800, 'height': 1000}
}
)

更多参数可以查看pyppeteer的源码。
        self.handleSIGINT = options.get('handleSIGINT', True)
self.handleSIGTERM = options.get('handleSIGTERM', True)
self.handleSIGHUP = options.get('handleSIGHUP', True)
self.ignoreHTTPSErrors = options.get('ignoreHTTPSErrors', False)
self.defaultViewport = options.get('defaultViewport', {'width': 800, 'height': 600}) # noqa: E501
self.slowMo = options.get('slowMo', 0)
self.timeout = options.get('timeout', 30000)
self.autoClose = options.get('autoClose', True)

  收起阅读 »

pyppeteer下载chromedriver失败的解决办法

因为网址变了!
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='storage.googleapis.com', port=443): Max retries exceeded with url: /chromium-browser-snapshots/Win_
x64/588429/chrome-win32.zip (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x00000000037D3880>: Failed to establish a new connect
ion: [Errno 11004] getaddrinfo failed'))
用浏览器看了,发现根本打不开。
然后谷歌了一下它的镜像,实际链接为: 
https://commondatastorage.googleapis.com/chromium-browser-snapshots/Win_x64/575458/chrome-win32.zip
 
所以,哎,这个库真的有点烂。
直接下载上面的链接,然后解压到本地路径,然后在设置一个环境变量指向这个目录即可。
 
继续阅读 »
因为网址变了!
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='storage.googleapis.com', port=443): Max retries exceeded with url: /chromium-browser-snapshots/Win_
x64/588429/chrome-win32.zip (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x00000000037D3880>: Failed to establish a new connect
ion: [Errno 11004] getaddrinfo failed'))
用浏览器看了,发现根本打不开。
然后谷歌了一下它的镜像,实际链接为: 
https://commondatastorage.googleapis.com/chromium-browser-snapshots/Win_x64/575458/chrome-win32.zip
 
所以,哎,这个库真的有点烂。
直接下载上面的链接,然后解压到本地路径,然后在设置一个环境变量指向这个目录即可。
  收起阅读 »

requests使用断点续传时注意要加stream=True,不然你的硬盘可能会爆掉

res = requests.get(url, stream=True, headers=headers, verify=False)
with open(dst, "ab") as f:
dp = Down_progress(file_size, first_size, dst)
dp.start()
chunk_size = 1024
for chunk in res.iter_content(chunk_size = chunk_size):
if chunk:
f.write(chunk)
dp.update(chunk_size)
 
如果不加stream=True,那么你的硬盘很可能就不停被写入,文件会变得无比巨大,最后磁盘空间不够死机。
不要问我为什么知道。
继续阅读 »
res = requests.get(url, stream=True, headers=headers, verify=False)
with open(dst, "ab") as f:
dp = Down_progress(file_size, first_size, dst)
dp.start()
chunk_size = 1024
for chunk in res.iter_content(chunk_size = chunk_size):
if chunk:
f.write(chunk)
dp.update(chunk_size)
 
如果不加stream=True,那么你的硬盘很可能就不停被写入,文件会变得无比巨大,最后磁盘空间不够死机。
不要问我为什么知道。 收起阅读 »

不用一行代码 下载雪球嘉年华视频

最近一年一度的雪球嘉年华在深圳举行,曾经去过一次,后来觉得里面都是卖基金,卖私募,后面就没有再去了。
听说今年着重分享一些观念,抱着好奇心,就打算下载几部来看看。
 
雪球网站很简单,只要找到下载链接就可以下载了。
 
第一步。打开一个视频播放的页面,比如大金链的 
 
11737544 粉丝主会场 | 巅峰对谈:金牛双子星主动VS量化
 
https://xueqiu.com/video/5285890810945319765
 

chrome_Z6q4VbZu5I1.png

 
右键,查看源码,然后在源码里面试着查找 mp4,flv,webp等流媒体字样。
 
在这里找到一个了:

chrome_WYjENd4ifc.png

 
但是这个视频下载地址有很多转义字符:
http:\u002F\u002F1256122120.vod2.myqcloud.com\u002F53ad1740vodtranscq1256122120\u002F17ebe5145285890810945319765\u002Fv.f20.mp4

直接在浏览器是无法直接打开的。
可以直接替换\u002f 为一个斜杠 \  就可以了。
如果嫌麻烦,可以在浏览器里面,按下F12,在console页面里面输入上面的地址,前后加个双引号,然后回车,就可以得到完整的地址了。
 

mstsc_v7jXwdJbSy.png

原创文章,转载请注明出处
 
http://30daydo.com/article/44119
 
继续阅读 »
最近一年一度的雪球嘉年华在深圳举行,曾经去过一次,后来觉得里面都是卖基金,卖私募,后面就没有再去了。
听说今年着重分享一些观念,抱着好奇心,就打算下载几部来看看。
 
雪球网站很简单,只要找到下载链接就可以下载了。
 
第一步。打开一个视频播放的页面,比如大金链的 
 
11737544 粉丝主会场 | 巅峰对谈:金牛双子星主动VS量化
 
https://xueqiu.com/video/5285890810945319765
 

chrome_Z6q4VbZu5I1.png

 
右键,查看源码,然后在源码里面试着查找 mp4,flv,webp等流媒体字样。
 
在这里找到一个了:

chrome_WYjENd4ifc.png

 
但是这个视频下载地址有很多转义字符:
http:\u002F\u002F1256122120.vod2.myqcloud.com\u002F53ad1740vodtranscq1256122120\u002F17ebe5145285890810945319765\u002Fv.f20.mp4

直接在浏览器是无法直接打开的。
可以直接替换\u002f 为一个斜杠 \  就可以了。
如果嫌麻烦,可以在浏览器里面,按下F12,在console页面里面输入上面的地址,前后加个双引号,然后回车,就可以得到完整的地址了。
 

mstsc_v7jXwdJbSy.png

原创文章,转载请注明出处
 
http://30daydo.com/article/44119
  收起阅读 »

P站 视频下载 JS加密URL 【JS逆向】

P站,你懂的网站,黄黑风格的。
它的真实下载地址是隐藏在JS加密当中的。
 
接下来我们看看如何找到它的真实地址。
 
首先随便打开一个视频:
我就找一个python学习的视频(什么鬼,上面还有python学习? 是的有的,还有很多数学题目在上面讲解的呢)

chrome_cID7nwLJtU.png

 
然后按F12
找到一个疑似的下载地址

chrome_d6YqqHw23u.png


试下拷贝直接去打开,发现无法打开,显示403 fobiden。
但是看url,应该就是最终的下载url的格式,通过最后的随机数控制播放权限。
 
然后打开页面的源码
 
chrome_TkTA0Gu2mZ.png



chrome_xQ9PuglQfb.png



 
在里面找下有没有一些mp4等的字符, 然后发现有个720p,1080p不同分辨率格式的字段,我们把这个JavaScript的代码扣下来,然后保存为p_hub.js
 
用vs code 或者其他编辑器打开

sublime_text_fYK2l4s4NA.png

看到了他们生成过程了吗? 
 
然后尝试用nodejs运行一下。
什么都没有输出。
 
因为上面代码并没有任何输出语句,我们在最后加一个 console.log(quality_720p); 就会有输出的了。

cmd_HJhTndtvh8.png

看到了吗?
然后拿这个地址去试试,看能否播放和下载。
果然,浏览器里面出现了一个, 额, python学习的页面,一个单独的视频页面,所以可以直接右键,弹出一个保存视频的菜单,然后可以直接下载了。
 
用代码requesets.get(url) 保存text.content 写入文件,就可以把视频保存到本地了。
 
源码:
#!/usr/bin/env python
# http://30daydo.com
import os
import re
import js2py
import requests
from lxml import etree
from clint.textui import progress
import fire
from loguru import logger

file='crawler'
logger.add(
"logs/%s.log" % file,
format="{time:MM-DD HH:mm:ss} {level} {message}",
)

headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36",
}
proxies = {}

# 如果代理不稳定,不推荐使用
# local proxy service
# proxies example:
# proxies = {
# "http": "socks5://127.0.0.1:1080",
# "https": "socks5://127.0.0.1:1080",
# }

def list_page(url):
logger.info("crawling : %s" % url)
resp = requests.get(url, headers=headers, proxies=proxies, verify=False)
html = etree.HTML(resp.text)

buff = '//*[@class="phimage"]/a/'
names = html.xpath(f"{buff}@href")
urls = html.xpath(f"{buff}img/@data-mediabook")
for i in range(len(urls)):
try:
url = urls
[i] name = re.findall("=ph(\w+)", names[i])[-1]
logger.info(f"{url} {name}")
download(url, name, "webm")
except Exception as err:
logger.error(err)


def detail_page(url):
s = requests.Session()
resp = s.get(url, headers=headers, proxies=proxies, verify=False)
html = etree.HTML(resp.content)

title = "".join(html.xpath("//h1//text()")).strip()
logger.info(title)

js_temp = html.xpath("//script/text()")
for j in js_temp:
if "flashvars" in j:
videoUrl = exeJs(j)

download(videoUrl, title, "mp4")
continue


def exeJs(js):
flashvars = re.findall("flashvars_\d+", js)[0]
js = "\n\t".join(js.split("\n\t")[:-5]).strip()

js = js.replace("// var nextVideoObject = flashvars_['nextVideo'];",'')
js+=flashvars
res = js2py.eval_js(js)

if res.quality_720p:
return res.quality_720p
elif res.quality_480p:
return res.quality_480p
elif res.quality_240p:
return res.quality_240p
else:
logger.error("parse url error")


def download(url, name, filetype):
logger.info(f"{url} {name} {filetype}")
filepath = "%s/%s.%s" % (filetype, name, filetype)
if os.path.exists(filepath):
logger.info("this file had been downloaded :: %s" % filepath)
return
else:
response = requests.get(url, headers=headers, proxies=proxies, stream=True)
with open(filepath, "wb") as file:
total_length = int(response.headers.get("content-length"))
for ch in progress.bar(
response.iter_content(chunk_size=2391975),
expected_size=(total_length / 1024) + 1,
):
if ch:
file.write(ch)

logger.info("download success :: %s" % filepath)


def run(_arg=None):
paths = ["webm", "mp4"]
for path in paths:
if not os.path.exists(path):
os.mkdir(path)
if _arg == "webm":
# https://www.pornhub.com/categories
urls = [
# "https://www.pornhub.com/video?o=tr",
# "https://www.pornhub.com/video?o=ht",
# "https://www.pornhub.com/video?o=mv",
"https://www.pornhub.com/video",
]
for url in urls:
list_page(url)
elif _arg == "mp4":
with open("download.txt", "r") as file:
keys = list(set(file.readlines()))
logger.info(keys)
keys += [d.strip(".webm") for d in os.listdir("webm/")]
for key in keys:
if not key.strip():
continue
url = "https://www.pornhub.com/view_v ... ot%3B % key.strip()
logger.info("url: {}", url)
detail_page(url)
else:
_str = """
tips:
python crawler.py webm
- 下载热门页面的缩略图,路径为webm文件夹下

python crawler.py mp4
- 该命令会下载webm文件下对应的mp4文件
- 也可以将目标地址写入download.txt中
"""
logger.info(_str)
return
logger.info("finish !")


if __name__ == "__main__":
fire.Fire(run)
[/i][/i]


[i]原创文章,
转载请注明出处:
http://30daydo.com/article/44115 

 
[/i]
[i]
3lDOnb5.jpg
[/i]
继续阅读 »
P站,你懂的网站,黄黑风格的。
它的真实下载地址是隐藏在JS加密当中的。
 
接下来我们看看如何找到它的真实地址。
 
首先随便打开一个视频:
我就找一个python学习的视频(什么鬼,上面还有python学习? 是的有的,还有很多数学题目在上面讲解的呢)

chrome_cID7nwLJtU.png

 
然后按F12
找到一个疑似的下载地址

chrome_d6YqqHw23u.png


试下拷贝直接去打开,发现无法打开,显示403 fobiden。
但是看url,应该就是最终的下载url的格式,通过最后的随机数控制播放权限。
 
然后打开页面的源码
 
chrome_TkTA0Gu2mZ.png



chrome_xQ9PuglQfb.png



 
在里面找下有没有一些mp4等的字符, 然后发现有个720p,1080p不同分辨率格式的字段,我们把这个JavaScript的代码扣下来,然后保存为p_hub.js
 
用vs code 或者其他编辑器打开

sublime_text_fYK2l4s4NA.png

看到了他们生成过程了吗? 
 
然后尝试用nodejs运行一下。
什么都没有输出。
 
因为上面代码并没有任何输出语句,我们在最后加一个 console.log(quality_720p); 就会有输出的了。

cmd_HJhTndtvh8.png

看到了吗?
然后拿这个地址去试试,看能否播放和下载。
果然,浏览器里面出现了一个, 额, python学习的页面,一个单独的视频页面,所以可以直接右键,弹出一个保存视频的菜单,然后可以直接下载了。
 
用代码requesets.get(url) 保存text.content 写入文件,就可以把视频保存到本地了。
 
源码:
#!/usr/bin/env python
# http://30daydo.com
import os
import re
import js2py
import requests
from lxml import etree
from clint.textui import progress
import fire
from loguru import logger

file='crawler'
logger.add(
"logs/%s.log" % file,
format="{time:MM-DD HH:mm:ss} {level} {message}",
)

headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36",
}
proxies = {}

# 如果代理不稳定,不推荐使用
# local proxy service
# proxies example:
# proxies = {
# "http": "socks5://127.0.0.1:1080",
# "https": "socks5://127.0.0.1:1080",
# }

def list_page(url):
logger.info("crawling : %s" % url)
resp = requests.get(url, headers=headers, proxies=proxies, verify=False)
html = etree.HTML(resp.text)

buff = '//*[@class="phimage"]/a/'
names = html.xpath(f"{buff}@href")
urls = html.xpath(f"{buff}img/@data-mediabook")
for i in range(len(urls)):
try:
url = urls
[i] name = re.findall("=ph(\w+)", names[i])[-1]
logger.info(f"{url} {name}")
download(url, name, "webm")
except Exception as err:
logger.error(err)


def detail_page(url):
s = requests.Session()
resp = s.get(url, headers=headers, proxies=proxies, verify=False)
html = etree.HTML(resp.content)

title = "".join(html.xpath("//h1//text()")).strip()
logger.info(title)

js_temp = html.xpath("//script/text()")
for j in js_temp:
if "flashvars" in j:
videoUrl = exeJs(j)

download(videoUrl, title, "mp4")
continue


def exeJs(js):
flashvars = re.findall("flashvars_\d+", js)[0]
js = "\n\t".join(js.split("\n\t")[:-5]).strip()

js = js.replace("// var nextVideoObject = flashvars_['nextVideo'];",'')
js+=flashvars
res = js2py.eval_js(js)

if res.quality_720p:
return res.quality_720p
elif res.quality_480p:
return res.quality_480p
elif res.quality_240p:
return res.quality_240p
else:
logger.error("parse url error")


def download(url, name, filetype):
logger.info(f"{url} {name} {filetype}")
filepath = "%s/%s.%s" % (filetype, name, filetype)
if os.path.exists(filepath):
logger.info("this file had been downloaded :: %s" % filepath)
return
else:
response = requests.get(url, headers=headers, proxies=proxies, stream=True)
with open(filepath, "wb") as file:
total_length = int(response.headers.get("content-length"))
for ch in progress.bar(
response.iter_content(chunk_size=2391975),
expected_size=(total_length / 1024) + 1,
):
if ch:
file.write(ch)

logger.info("download success :: %s" % filepath)


def run(_arg=None):
paths = ["webm", "mp4"]
for path in paths:
if not os.path.exists(path):
os.mkdir(path)
if _arg == "webm":
# https://www.pornhub.com/categories
urls = [
# "https://www.pornhub.com/video?o=tr",
# "https://www.pornhub.com/video?o=ht",
# "https://www.pornhub.com/video?o=mv",
"https://www.pornhub.com/video",
]
for url in urls:
list_page(url)
elif _arg == "mp4":
with open("download.txt", "r") as file:
keys = list(set(file.readlines()))
logger.info(keys)
keys += [d.strip(".webm") for d in os.listdir("webm/")]
for key in keys:
if not key.strip():
continue
url = "https://www.pornhub.com/view_v ... ot%3B % key.strip()
logger.info("url: {}", url)
detail_page(url)
else:
_str = """
tips:
python crawler.py webm
- 下载热门页面的缩略图,路径为webm文件夹下

python crawler.py mp4
- 该命令会下载webm文件下对应的mp4文件
- 也可以将目标地址写入download.txt中
"""
logger.info(_str)
return
logger.info("finish !")


if __name__ == "__main__":
fire.Fire(run)
[/i][/i]


[i]原创文章,
转载请注明出处:
http://30daydo.com/article/44115 

 
[/i]
[i]
3lDOnb5.jpg
[/i] 收起阅读 »

简单快速下载知乎视频

现在的知乎也有视频内容,目前看来没有任何反爬措施。可以简单通过以下方式用浏览器下载:
 
1. 打开视频前按F12
2. 播放视频
3. 查看F12的网络选项
4. 找到 https://vdn3.vzuu.com 的url
5. 对应的整个url链接就是视频的真实下载地址。把url复制到浏览器打开,然后右键另存为本地视频就可以了 
 
继续阅读 »
现在的知乎也有视频内容,目前看来没有任何反爬措施。可以简单通过以下方式用浏览器下载:
 
1. 打开视频前按F12
2. 播放视频
3. 查看F12的网络选项
4. 找到 https://vdn3.vzuu.com 的url
5. 对应的整个url链接就是视频的真实下载地址。把url复制到浏览器打开,然后右键另存为本地视频就可以了 
  收起阅读 »

集思录用户名密码JS加密流程解密 【JS加密破解教程一】

之前帮他人部署的一个集思录爬取的程序,忽然获取的数据不一样了。 查看日志后发现,现在的登录用户名和密码做了一层加密后再post提交。
 
而且提交的内容每次都固定,第一种最傻的方式就是每次提交就把加密的用户名和密码提交上去。


5960fc43cefbc020698b2d4b928274f8.png


 
当然,对于有钻研的读者,可能想看看其具体的加密方式。
那么就按照流程,通过断点与搜索,找到其加密方法。
首先在上面的截图中很明显就知道,这个字符加密应该是aes,因为它的提交字段中aes:1
 
按F12,走完整个登录流程。
然后搜索password字样的地方。
 
在index.html首页中找到一处:

chrome_3IrUrGEXxY.png

 
然后在该password的地方打个断点,然后跳转到jslencode 的地方。

YEtNA7EKa0.png

跳转到的地方是这里。
 
然后我们浏览一下这个JS页面,尝试把整个JS代码抠出来。
放到我们的调试软件中,比较常用的是鬼鬼JS调试器。(有需要的可以关注wx公众号下载:回复 鬼鬼JS 即可)

鬼鬼JS调试工具7.5_J47FALG4mD_.png

这个调试器的好处是,可以很方便格式化JS代码,然后输入你要调试的字符,然后点击运行,可以当场拿到结果,等到结果ok了的话,就可以用python 的pyexecjs执行。
 
先点击代码格式化,然后在输入框里找到函数的入口:
jslencode(text, aes_key),好了,现在就把我们的密码 XXXXXX,和aes_key 放进去就可以了。
aes_key 在之前的index.html就能找到。
var A397151C04723421F = '397151C04723421F';
function doLogin(){
var data = $('#login_form').serializeObjectToJson();
data['_post_type'] = 'ajax';
data['aes'] = 1;
data['user_name'] = jslencode(data['user_name'], A397151C04723421F);
data['password'] = jslencode(data['password'], A397151C04723421F);
$.post('/account/ajax/login_process/', data, function(rst){
on_login_error_processer(rst);
}, 'json');

var A397151C04723421F = '397151C04723421F';
 
jslencode(‘jisilupassword’, '397151C04723421F')
右下角有个系统引擎运行。
 
得到结果:
1d1bd2b22b8cc5c09ad8ce8f1e69b87f
 
对比一下第一张图里面的的post请求,发现是一样的。那么现在的JS解密就成功了一大半了。 接着我们就写python代码执行这段JS脚本。
 
 
尝试直接把JS放到一个文件,然后编写python代码
# -*- coding: utf-8 -*-
# @Time : 2020/11/27 12:15
# @File : js_executor.py
# @Author : Rocky C@www.30daydo.com
import execjs

def main():
encode_user = ctx.call('jslencode', user, key)
encode_password = ctx.call('jslencode', password, key)
print(encode_user)
print(encode_password)


if __name__ == '__main__':
main()

然后发现报错: 说CryptoJS没有定义,那么我们看看代码。
(function(root, factory) {
if (typeof exports === "object") {
module.exports = exports = factory()
} else {
if (typeof define === "function" && define.amd) {
define(, factory)
} else {
root.CryptoJS = factory()
}
}
}(this, function() {
var CryptoJS = CryptoJS || (function(Math, undefined) {
var create = Object.create || (function() {

发现这一行此时CryptoJS的定义

var CryptoJS = CryptoJS || (function(Math, undefined) { var create = Object.create || (function() {
 
那么我们把这一行上面的全部删除。最后的调试后,能够执行的JS代码如下,并保存为 集思录.js 文件,调用上面的python文件。
 
集思录.js
var CryptoJS = CryptoJS || (function(Math, undefined) {
var create = Object.create || (function() {
function F() {}
return function(obj) {
var subtype;
F.prototype = obj;
subtype = new F();
F.prototype = null;
return subtype
}
}());
var C = {};
var C_lib = C.lib = {};
var Base = C_lib.Base = (function() {





下面的就跟上面的一模一样了:
源文件地址:
https://www.jisilu.cn/static/js/crypto-js-3.3.0-min.js
然后python运行后得到加密后的aes数据。
 
对比一下鬼鬼JS调试器的结果,一样的。
OK,手工。
这里汇聚了平时整理的JS破解工作流,大家可以参考参考。 
 https://github.com/Rockyzsu/JS-Reverse 

原创文章,转载请注明出处:
http://30daydo.com/article/44109
喜欢的朋友可以加入星球探讨:
继续阅读 »
之前帮他人部署的一个集思录爬取的程序,忽然获取的数据不一样了。 查看日志后发现,现在的登录用户名和密码做了一层加密后再post提交。
 
而且提交的内容每次都固定,第一种最傻的方式就是每次提交就把加密的用户名和密码提交上去。


5960fc43cefbc020698b2d4b928274f8.png


 
当然,对于有钻研的读者,可能想看看其具体的加密方式。
那么就按照流程,通过断点与搜索,找到其加密方法。
首先在上面的截图中很明显就知道,这个字符加密应该是aes,因为它的提交字段中aes:1
 
按F12,走完整个登录流程。
然后搜索password字样的地方。
 
在index.html首页中找到一处:

chrome_3IrUrGEXxY.png

 
然后在该password的地方打个断点,然后跳转到jslencode 的地方。

YEtNA7EKa0.png

跳转到的地方是这里。
 
然后我们浏览一下这个JS页面,尝试把整个JS代码抠出来。
放到我们的调试软件中,比较常用的是鬼鬼JS调试器。(有需要的可以关注wx公众号下载:回复 鬼鬼JS 即可)

鬼鬼JS调试工具7.5_J47FALG4mD_.png

这个调试器的好处是,可以很方便格式化JS代码,然后输入你要调试的字符,然后点击运行,可以当场拿到结果,等到结果ok了的话,就可以用python 的pyexecjs执行。
 
先点击代码格式化,然后在输入框里找到函数的入口:
jslencode(text, aes_key),好了,现在就把我们的密码 XXXXXX,和aes_key 放进去就可以了。
aes_key 在之前的index.html就能找到。
var A397151C04723421F = '397151C04723421F';
function doLogin(){
var data = $('#login_form').serializeObjectToJson();
data['_post_type'] = 'ajax';
data['aes'] = 1;
data['user_name'] = jslencode(data['user_name'], A397151C04723421F);
data['password'] = jslencode(data['password'], A397151C04723421F);
$.post('/account/ajax/login_process/', data, function(rst){
on_login_error_processer(rst);
}, 'json');

var A397151C04723421F = '397151C04723421F';
 
jslencode(‘jisilupassword’, '397151C04723421F')
右下角有个系统引擎运行。
 
得到结果:
1d1bd2b22b8cc5c09ad8ce8f1e69b87f
 
对比一下第一张图里面的的post请求,发现是一样的。那么现在的JS解密就成功了一大半了。 接着我们就写python代码执行这段JS脚本。
 
 
尝试直接把JS放到一个文件,然后编写python代码
# -*- coding: utf-8 -*-
# @Time : 2020/11/27 12:15
# @File : js_executor.py
# @Author : Rocky C@www.30daydo.com
import execjs

def main():
encode_user = ctx.call('jslencode', user, key)
encode_password = ctx.call('jslencode', password, key)
print(encode_user)
print(encode_password)


if __name__ == '__main__':
main()

然后发现报错: 说CryptoJS没有定义,那么我们看看代码。
(function(root, factory) {
if (typeof exports === "object") {
module.exports = exports = factory()
} else {
if (typeof define === "function" && define.amd) {
define(, factory)
} else {
root.CryptoJS = factory()
}
}
}(this, function() {
var CryptoJS = CryptoJS || (function(Math, undefined) {
var create = Object.create || (function() {

发现这一行此时CryptoJS的定义

var CryptoJS = CryptoJS || (function(Math, undefined) { var create = Object.create || (function() {
 
那么我们把这一行上面的全部删除。最后的调试后,能够执行的JS代码如下,并保存为 集思录.js 文件,调用上面的python文件。
 
集思录.js
var CryptoJS = CryptoJS || (function(Math, undefined) {
var create = Object.create || (function() {
function F() {}
return function(obj) {
var subtype;
F.prototype = obj;
subtype = new F();
F.prototype = null;
return subtype
}
}());
var C = {};
var C_lib = C.lib = {};
var Base = C_lib.Base = (function() {





下面的就跟上面的一模一样了:
源文件地址:
https://www.jisilu.cn/static/js/crypto-js-3.3.0-min.js
然后python运行后得到加密后的aes数据。
 
对比一下鬼鬼JS调试器的结果,一样的。
OK,手工。
这里汇聚了平时整理的JS破解工作流,大家可以参考参考。 
 https://github.com/Rockyzsu/JS-Reverse 

原创文章,转载请注明出处:
http://30daydo.com/article/44109
喜欢的朋友可以加入星球探讨:
收起阅读 »

asyncio 异步爬取vs requests同步爬取 性能对比

 首先是异步爬取:
import sys
sys.path.append('..')
import asyncio
import datetime
import aiohttp
import re
import time
from parsel import Selector
from configure.settings import DBSelector
from common.BaseService import BaseService

SLEEP = 2

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}

URL_MAP = {'home_page': 'https://holdle.com/stocks/industry', 'base': 'https://holdle.com'}


class AsyncMongo():
def __init__(self):
self.DB = DBSelector()
self.client = self.DB.mongo(location_type='qq', async_type=True)
self.db = self.client['db_stock']

async def update(self, table,data):
self.doc= self.db[table]
await self.doc.insert_many(data)


class Holdle(BaseService):

def __init__(self):
super(Holdle, self).__init__()
self.data_processor = AsyncMongo()
self.tables_list =['ROE','Cash_Ratio','Gross_Margin','Operation_Margin','Net_Profit_Ratio','Dividend_ratio']

async def home_page(self):
start = time.time()
async with aiohttp.ClientSession() as session:
async with session.get(url=URL_MAP['home_page'], headers=headers) as response:
html = await response.text() # 这个阻塞
resp = Selector(text=html)
industries = resp.xpath('//ul[@class="list-unstyled"]/a')
task_list = []
for industry in industries:
json_data = {}
industry_url = industry.xpath('.//@href').extract_first()
industry_name = industry.xpath('.//li/text()').extract_first()
industry_name = industry_name.replace('-', '').strip()
json_data['industry_url'] = industry_url
json_data['industry_name'] = industry_name

task = asyncio.ensure_future(self.detail_list(session, industry_url, json_data))
task_list.append(task)

await asyncio.gather(*task_list)
end = time.time()

print(f'time used {end - start}')

async def detail_list(self, session, url, json_data):

async with session.get(URL_MAP['base'] + url, headers=headers) as response:
response = await response.text()
await self.parse_detail(response, json_data)

async def parse_detail(self, html, json_data=None):
resp = Selector(text=html)
industry=json_data['industry_name']
tables = resp.xpath('//table[@class="table table-bordered"]')
if len(tables)!=6:
raise ValueError

for index,table in enumerate(self.tables_list):
rows = tables[index].xpath('.//tr')
result = []
for row in rows[1:]:
stock_name = row.xpath('.//td[1]/text()').extract_first()
value = row.xpath('.//td[2]/text()').extract_first()
value = float(value)
d={'industry':industry,'name':stock_name,'value':value,'crawltime':datetime.datetime.now()}
result.append(d)
await self.data_processor.update(table,result)


app = Holdle()
loop = asyncio.get_event_loop()
loop.run_until_complete(app.home_page())

爬完并且入库,用时大约为35s
 
使用requests爬取
# -*- coding: utf-8 -*-
# @Time : 2020/11/24 21:42
# @File : sync_spider.py
# @Author : Rocky C@www.30daydo.com
import requests
import sys
sys.path.append('..')
import asyncio
import datetime
import aiohttp
import re
import time
from parsel import Selector
from configure.settings import DBSelector
from common.BaseService import BaseService

SLEEP = 2

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}

URL_MAP = {'home_page': 'https://holdle.com/stocks/industry', 'base': 'https://holdle.com'}


class Holdle(BaseService):

def __init__(self):
super(Holdle, self).__init__()

self.DB = DBSelector()
self.client = self.DB.mongo(location_type='qq', async_type=True)
self.session = requests.Session()

def run(self):
start = time.time()

response = self.session.get(url=URL_MAP['home_page'], headers=headers)
html = response.text # 这个阻塞
resp = Selector(text=html)
industries = resp.xpath('//ul[@class="list-unstyled"]/a')
for industry in industries:
json_data = {}
industry_url = industry.xpath('.//@href').extract_first()
industry_name = industry.xpath('.//li/text()').extract_first()
json_data['industry_url'] = industry_url
json_data['industry_name'] = industry_name
self.detail_list(industry_url, json_data)

end = time.time()
print(f'time used {end-start}')

def detail_list(self, url, json_data):

response = self.session.get(URL_MAP['base']+url, headers=headers)
response =response.text
self.parse_detail(response, json_data)

def parse_detail(self, html, json_data=None):
resp = Selector(text=html)
title =resp.xpath('//title/text()').extract_first()
print(title)


app = Holdle()
app.run()
用时约160s,而且这里还省略了mongo入库的时间。上面异步爬取里面包含了异步存入mongo。
 
所以单从网络IO性能上来说,异步是比纯同步要快很多。
但是,async的生态做得不是太好,第三方的异步框架做得也不够完善。 
 
因为如果系统中引入了异步,很多耗时的地方也是需要使用异步的写法和框架,不然会导致系统的控制权没有被正确转移。
 
水文一篇。
完毕
 
继续阅读 »
 首先是异步爬取:
import sys
sys.path.append('..')
import asyncio
import datetime
import aiohttp
import re
import time
from parsel import Selector
from configure.settings import DBSelector
from common.BaseService import BaseService

SLEEP = 2

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}

URL_MAP = {'home_page': 'https://holdle.com/stocks/industry', 'base': 'https://holdle.com'}


class AsyncMongo():
def __init__(self):
self.DB = DBSelector()
self.client = self.DB.mongo(location_type='qq', async_type=True)
self.db = self.client['db_stock']

async def update(self, table,data):
self.doc= self.db[table]
await self.doc.insert_many(data)


class Holdle(BaseService):

def __init__(self):
super(Holdle, self).__init__()
self.data_processor = AsyncMongo()
self.tables_list =['ROE','Cash_Ratio','Gross_Margin','Operation_Margin','Net_Profit_Ratio','Dividend_ratio']

async def home_page(self):
start = time.time()
async with aiohttp.ClientSession() as session:
async with session.get(url=URL_MAP['home_page'], headers=headers) as response:
html = await response.text() # 这个阻塞
resp = Selector(text=html)
industries = resp.xpath('//ul[@class="list-unstyled"]/a')
task_list = []
for industry in industries:
json_data = {}
industry_url = industry.xpath('.//@href').extract_first()
industry_name = industry.xpath('.//li/text()').extract_first()
industry_name = industry_name.replace('-', '').strip()
json_data['industry_url'] = industry_url
json_data['industry_name'] = industry_name

task = asyncio.ensure_future(self.detail_list(session, industry_url, json_data))
task_list.append(task)

await asyncio.gather(*task_list)
end = time.time()

print(f'time used {end - start}')

async def detail_list(self, session, url, json_data):

async with session.get(URL_MAP['base'] + url, headers=headers) as response:
response = await response.text()
await self.parse_detail(response, json_data)

async def parse_detail(self, html, json_data=None):
resp = Selector(text=html)
industry=json_data['industry_name']
tables = resp.xpath('//table[@class="table table-bordered"]')
if len(tables)!=6:
raise ValueError

for index,table in enumerate(self.tables_list):
rows = tables[index].xpath('.//tr')
result = []
for row in rows[1:]:
stock_name = row.xpath('.//td[1]/text()').extract_first()
value = row.xpath('.//td[2]/text()').extract_first()
value = float(value)
d={'industry':industry,'name':stock_name,'value':value,'crawltime':datetime.datetime.now()}
result.append(d)
await self.data_processor.update(table,result)


app = Holdle()
loop = asyncio.get_event_loop()
loop.run_until_complete(app.home_page())

爬完并且入库,用时大约为35s
 
使用requests爬取
# -*- coding: utf-8 -*-
# @Time : 2020/11/24 21:42
# @File : sync_spider.py
# @Author : Rocky C@www.30daydo.com
import requests
import sys
sys.path.append('..')
import asyncio
import datetime
import aiohttp
import re
import time
from parsel import Selector
from configure.settings import DBSelector
from common.BaseService import BaseService

SLEEP = 2

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}

URL_MAP = {'home_page': 'https://holdle.com/stocks/industry', 'base': 'https://holdle.com'}


class Holdle(BaseService):

def __init__(self):
super(Holdle, self).__init__()

self.DB = DBSelector()
self.client = self.DB.mongo(location_type='qq', async_type=True)
self.session = requests.Session()

def run(self):
start = time.time()

response = self.session.get(url=URL_MAP['home_page'], headers=headers)
html = response.text # 这个阻塞
resp = Selector(text=html)
industries = resp.xpath('//ul[@class="list-unstyled"]/a')
for industry in industries:
json_data = {}
industry_url = industry.xpath('.//@href').extract_first()
industry_name = industry.xpath('.//li/text()').extract_first()
json_data['industry_url'] = industry_url
json_data['industry_name'] = industry_name
self.detail_list(industry_url, json_data)

end = time.time()
print(f'time used {end-start}')

def detail_list(self, url, json_data):

response = self.session.get(URL_MAP['base']+url, headers=headers)
response =response.text
self.parse_detail(response, json_data)

def parse_detail(self, html, json_data=None):
resp = Selector(text=html)
title =resp.xpath('//title/text()').extract_first()
print(title)


app = Holdle()
app.run()
用时约160s,而且这里还省略了mongo入库的时间。上面异步爬取里面包含了异步存入mongo。
 
所以单从网络IO性能上来说,异步是比纯同步要快很多。
但是,async的生态做得不是太好,第三方的异步框架做得也不够完善。 
 
因为如果系统中引入了异步,很多耗时的地方也是需要使用异步的写法和框架,不然会导致系统的控制权没有被正确转移。
 
水文一篇。
完毕
  收起阅读 »

免费代理ip与收费的代理ip

在爬虫采集数据过程中,经常会遇到ip被限制,这也是常见的问题。因为网站都会根据ip访问的频率进行判断封锁ip,这是反爬虫机制的策略,如果拥有大量的ip资源,自然会突破ip限制的反爬虫策略。
 
曾经有尝试过使用免费的代理ip来搭建代理池,可是免费的代理ip不仅资源少,而且可用率、高匿性、速度等都极差,每次使用都需要借助第三方软件进行检查是否可用,严重影响效率,根本满足不了任务的需求。
 
收费的代理ip与免费的代理ip差距非常大,不仅拥有海量的ip资源,可用率、高匿性、速度都是极好。操作简单工作效率既然提高上去了。经过多家的测试,最终选择了性价比最高的万变ip。高质量的优质代理ip才可以真正用来防止爬虫被封锁,如果使用普通代理,爬虫的真实IP还是会暴露。新获取一批新IP
继续阅读 »
在爬虫采集数据过程中,经常会遇到ip被限制,这也是常见的问题。因为网站都会根据ip访问的频率进行判断封锁ip,这是反爬虫机制的策略,如果拥有大量的ip资源,自然会突破ip限制的反爬虫策略。
 
曾经有尝试过使用免费的代理ip来搭建代理池,可是免费的代理ip不仅资源少,而且可用率、高匿性、速度等都极差,每次使用都需要借助第三方软件进行检查是否可用,严重影响效率,根本满足不了任务的需求。
 
收费的代理ip与免费的代理ip差距非常大,不仅拥有海量的ip资源,可用率、高匿性、速度都是极好。操作简单工作效率既然提高上去了。经过多家的测试,最终选择了性价比最高的万变ip。高质量的优质代理ip才可以真正用来防止爬虫被封锁,如果使用普通代理,爬虫的真实IP还是会暴露。新获取一批新IP 收起阅读 »

Python爬虫学习者需要注意什么?

在这个大数据时代里,学习Python网络爬虫的人越来越多,在学习过程中难免会遇到一些问题,学习者最常见的问题就是遇到ip被限制,因为你在爬虫采集数据过程中,同一个ip频繁的对网站进行访问,会给对方服务器会造成压力, 那么网站就根据ip访问的频率来限制你的ip,这也是反爬虫机制常见的一种判断。


最常见的解决方法就是使用大量的ip,就是借着代理ip保证IP被封时有替换IP可用,永远保持着续航能力。这里推荐51代理ip,作为一家提供代理IP的专业服务商,万变ip代理拥有强大的技术团队运营维护,全高匿系统所产生的高匿ip不仅安全稳定、而且速度快, 以及与爬虫用户多年来合作的宝贵经验,是Python爬虫首选代理IP。

Python是一种全栈计算机程序设计语言,全栈,顾名思义,应用范围广。你可能听说过很多编程语言,例如C语言,Java语言等,众所周知,这些语言都非常难学,更别说景桐使用了。而python不一样,比如完成一个Web服务,C语言要写1000行代码,Java要写100行,而python可能只要写20行。对!这就是差距!目前由于python“简单易懂”,已逐步成为网络爬虫主流语言。

在初学python爬虫时,很多程序员会被一些“小问题”阻碍脚步,为避免大家再次犯同样的错误,加快学习进程,在爬取网站信息时一定要使用大量代理IP。好用的代理IP服务商,

高效率的爬虫工作离不开ip代理的支持,这就是ip代理越来越受欢迎的原因!收藏举报投诉
继续阅读 »
在这个大数据时代里,学习Python网络爬虫的人越来越多,在学习过程中难免会遇到一些问题,学习者最常见的问题就是遇到ip被限制,因为你在爬虫采集数据过程中,同一个ip频繁的对网站进行访问,会给对方服务器会造成压力, 那么网站就根据ip访问的频率来限制你的ip,这也是反爬虫机制常见的一种判断。


最常见的解决方法就是使用大量的ip,就是借着代理ip保证IP被封时有替换IP可用,永远保持着续航能力。这里推荐51代理ip,作为一家提供代理IP的专业服务商,万变ip代理拥有强大的技术团队运营维护,全高匿系统所产生的高匿ip不仅安全稳定、而且速度快, 以及与爬虫用户多年来合作的宝贵经验,是Python爬虫首选代理IP。

Python是一种全栈计算机程序设计语言,全栈,顾名思义,应用范围广。你可能听说过很多编程语言,例如C语言,Java语言等,众所周知,这些语言都非常难学,更别说景桐使用了。而python不一样,比如完成一个Web服务,C语言要写1000行代码,Java要写100行,而python可能只要写20行。对!这就是差距!目前由于python“简单易懂”,已逐步成为网络爬虫主流语言。

在初学python爬虫时,很多程序员会被一些“小问题”阻碍脚步,为避免大家再次犯同样的错误,加快学习进程,在爬取网站信息时一定要使用大量代理IP。好用的代理IP服务商,

高效率的爬虫工作离不开ip代理的支持,这就是ip代理越来越受欢迎的原因!收藏举报投诉 收起阅读 »

Python爬虫虎牙平台主播的图片代码

今天万变ip就为大家分享一下,Python爬虫是如何爬取虎牙平台的主播图片。这里我们主要爬取虎牙星秀的主播图片,并下载保存到本地。在爬取之前,我们可先打开目标页面,F12审查元素,查找我们需要的信息所在的位置。

代码如下:

import urllib.request

import re

import os

# 全局变量用来记录图片的编号

gl_z = 0

def down_img(url1):

"""下载图片"""

# 处理图片链接,拼接http:

url = "https:" + re.sub(r"\?", "", url1)

global gl_z

print(url)

# 请求链接

response = urllib.request.urlopen(url)

# 读取内容

data = response.read()

# 切片取出图片名称

file_name = url[url.rfind('/') + 1:]

# 生成列表

a = [x for x in range(10000)]

# 打开文件用以写入

file = open(os.path.join("photo3", "img" + file_name + str(a[gl_z]) + ".jpg"), "wb")

file.write(data)

# 关闭文件

file.close()

# 编号加1

gl_z += 1

if __name__ == '__main__':

# 要抓去信息的网址

home = """http://www.huya.com/g/xingxiu"""

# 模拟请求头

headers = {

"Host": "www.huya.com",

"User-Agent": "agent信息"

}

# 构造好请求对象 将请求提交到服务器 获取的响应就是到首页的html代码

request = urllib.request.Request(url=home, headers=headers)

response = urllib.request.urlopen(request)

# 读取抓到的内容并解码

html_data = response.read().decode()

"""huyaimg.msstatic.com/avatar/1054/db/6590aa9bcf98e12e5d809d371e46cc_180_135.jpg

"""

# 使用正则 从首页中 提取出所有的图片链接

img_list = re.findall(r"//huyaimg\.msstatic\.com.+\.jpg\?", html_data)

print(img_list)

# 取出每张图片进行下载

for img_url in img_list:

print(img_url)

down_img(img_url)
继续阅读 »
今天万变ip就为大家分享一下,Python爬虫是如何爬取虎牙平台的主播图片。这里我们主要爬取虎牙星秀的主播图片,并下载保存到本地。在爬取之前,我们可先打开目标页面,F12审查元素,查找我们需要的信息所在的位置。

代码如下:

import urllib.request

import re

import os

# 全局变量用来记录图片的编号

gl_z = 0

def down_img(url1):

"""下载图片"""

# 处理图片链接,拼接http:

url = "https:" + re.sub(r"\?", "", url1)

global gl_z

print(url)

# 请求链接

response = urllib.request.urlopen(url)

# 读取内容

data = response.read()

# 切片取出图片名称

file_name = url[url.rfind('/') + 1:]

# 生成列表

a = [x for x in range(10000)]

# 打开文件用以写入

file = open(os.path.join("photo3", "img" + file_name + str(a[gl_z]) + ".jpg"), "wb")

file.write(data)

# 关闭文件

file.close()

# 编号加1

gl_z += 1

if __name__ == '__main__':

# 要抓去信息的网址

home = """http://www.huya.com/g/xingxiu"""

# 模拟请求头

headers = {

"Host": "www.huya.com",

"User-Agent": "agent信息"

}

# 构造好请求对象 将请求提交到服务器 获取的响应就是到首页的html代码

request = urllib.request.Request(url=home, headers=headers)

response = urllib.request.urlopen(request)

# 读取抓到的内容并解码

html_data = response.read().decode()

"""huyaimg.msstatic.com/avatar/1054/db/6590aa9bcf98e12e5d809d371e46cc_180_135.jpg

"""

# 使用正则 从首页中 提取出所有的图片链接

img_list = re.findall(r"//huyaimg\.msstatic\.com.+\.jpg\?", html_data)

print(img_list)

# 取出每张图片进行下载

for img_url in img_list:

print(img_url)

down_img(img_url) 收起阅读 »

Python爬虫基本框架

基本爬虫框架包括五个模块,万变ip来告诉你:

1. 爬虫调度器负责统筹其他四个模块协调工作。

2. URL管理器负责管理URL链接,包括已爬取的链接和未爬取的链接。

3. HTML下载器用于从URL管理器中获取未爬取的链接并下载其HTML网页。

4. HTML解析器用于解析HTML下载器下载的HTML网页,获取URL链接交给URL管理器,提取要获取的数据交给数据存储器。

5. 数据存储器用于将HTML解析器解析出来的数据存储到数据库或文件。
继续阅读 »
基本爬虫框架包括五个模块,万变ip来告诉你:

1. 爬虫调度器负责统筹其他四个模块协调工作。

2. URL管理器负责管理URL链接,包括已爬取的链接和未爬取的链接。

3. HTML下载器用于从URL管理器中获取未爬取的链接并下载其HTML网页。

4. HTML解析器用于解析HTML下载器下载的HTML网页,获取URL链接交给URL管理器,提取要获取的数据交给数据存储器。

5. 数据存储器用于将HTML解析器解析出来的数据存储到数据库或文件。 收起阅读 »

Python爬虫如何防止ip被封?

Python爬虫采集数据过程中,经常遇到网站的反爬虫机制,当相同的ip频繁的对服务器进行访问,就很容易出现ip被限制的,这就是常见的一种反爬虫策略,ip被限制会影响工作的进度甚至无法再继续工作,那么如何去避免呢?
 
第一种:降低访问的速度,我们可以使用 time模块中的sleep,使程序每运行一次后就睡眠1s,这样可以很有效的降低ip被封机率,但是效率效果不是很高,一般是用于量小的采集任务。
 
第二种:使用类似万变ip代理这样的优质换ip软件,这也是爬虫工作者最常用的手段之一,通过代理ip来伪装我们的ip,隐藏本地真实的ip地址,让目标服务器无法识别是相同ip发出的请求,这样就很有效的防止ip被封。突破了ip的限制,采集数据的任务就会顺利,工作效率自然会提高!
 
 
 
继续阅读 »
Python爬虫采集数据过程中,经常遇到网站的反爬虫机制,当相同的ip频繁的对服务器进行访问,就很容易出现ip被限制的,这就是常见的一种反爬虫策略,ip被限制会影响工作的进度甚至无法再继续工作,那么如何去避免呢?
 
第一种:降低访问的速度,我们可以使用 time模块中的sleep,使程序每运行一次后就睡眠1s,这样可以很有效的降低ip被封机率,但是效率效果不是很高,一般是用于量小的采集任务。
 
第二种:使用类似万变ip代理这样的优质换ip软件,这也是爬虫工作者最常用的手段之一,通过代理ip来伪装我们的ip,隐藏本地真实的ip地址,让目标服务器无法识别是相同ip发出的请求,这样就很有效的防止ip被封。突破了ip的限制,采集数据的任务就会顺利,工作效率自然会提高!
 
 
  收起阅读 »

python asyncio aiohttp motor异步爬虫例子 定时抓取bilibili首页热度网红

使用的异步库: aiohttp【http异步库】,motor【mongo异步库】

AsyncIOMotorClient(connect_uri) 
motor连接带用户名和密码的方法和pymongo一致。

connect_uri = f'mongodb://{user}:{password}@{host}:{port}'
# -*- coding: utf-8 -*-
# website: http://30daydo.com
# @Time : 2020/9/22 10:07
# @File : bilibili_hot_anchor.py

# 异步爬取首页与列表

import asyncio
import datetime
import aiohttp
import re
import time
from motor.motor_asyncio import AsyncIOMotorClient
from parsel import Selector
from settings import _json_data

SLEEP = 60 * 10
INFO = _json_data['mongo']['arm']
host = INFO['host']
port = INFO['port']
user = INFO['user']
password = INFO['password']
connect_uri = f'mongodb://{user}:{password}@{host}:{port}'

client = AsyncIOMotorClient(connect_uri)

db = client['db_parker']

home_url = 'https://www.bilibili.com/ranking'
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}


def convertor(number_str):
'''
将小数点的万变为整数
:param number_str:
:return:
'''
number = re.search('(\d+\.+\d+)', number_str)
if number:
number = float(number.group(1))
if re.search('万', number_str):
number = int(number * 10000)
else:
number = 0

return number


async def home_page():
async with aiohttp.ClientSession() as session:
while True:
start = time.time()
async with session.get(url=home_url, headers=headers) as response:
html = await response.text()
resp = Selector(text=html)
items = resp.xpath('//ul[@class="rank-list"]/li')
for item in items:
json_data = {}
number = item.xpath('.//div[@class="num"]/text()').extract_first()
info = item.xpath('.//div[@class="info"][1]')
title = info.xpath('.//a/text()').extract_first()

detail_url = info.xpath('.//a/@href').extract_first()
play_number = info.xpath('.//div[@class="detail"]/span[1]/text()').extract_first()
viewing_number = info.xpath('.//div[@class="detail"]/span[2]/text()').extract_first()

json_data['number'] = int(number)
json_data['title'] = title
json_data['play_number'] = convertor(play_number)
json_data['viewing_number'] = convertor(viewing_number)

json_data['url'] = detail_url

task = asyncio.create_task(detail_list(session, detail_url, json_data))
# await detail_url()
end = time.time()
print(f'time used {end-start}')
await asyncio.sleep(SLEEP) # 暂停10分钟
print(f'sleep for {SLEEP}')


async def detail_list(session, url, json_data):
async with session.get(url, headers=headers) as response:
response = await response.text()
await parse_detail(response, json_data)


async def parse_detail(html, json_data=None):
resp = Selector(text=html)
info = resp.xpath('//div[@id="v_desc"]/div[@class="info open"]/text()').extract_first()

if not info:
info = '这个家伙很懒'

json_data['info'] = info.strip()
current = datetime.datetime.now()
json_data['crawltime'] = current
await db['bilibili'].update_one({'url': json_data['url']}, {'$set': json_data}, True, True)


loop = asyncio.get_event_loop()
loop.run_until_complete(home_page())




爬取的数据图:
Studio_3T_wIGBpehcXS.png

 
原创文章,转载请注明:http://30daydo.com/article/605
需要源码,可私信。

mmexport1585743466421.jpg

 
继续阅读 »
使用的异步库: aiohttp【http异步库】,motor【mongo异步库】

AsyncIOMotorClient(connect_uri) 
motor连接带用户名和密码的方法和pymongo一致。

connect_uri = f'mongodb://{user}:{password}@{host}:{port}'
# -*- coding: utf-8 -*-
# website: http://30daydo.com
# @Time : 2020/9/22 10:07
# @File : bilibili_hot_anchor.py

# 异步爬取首页与列表

import asyncio
import datetime
import aiohttp
import re
import time
from motor.motor_asyncio import AsyncIOMotorClient
from parsel import Selector
from settings import _json_data

SLEEP = 60 * 10
INFO = _json_data['mongo']['arm']
host = INFO['host']
port = INFO['port']
user = INFO['user']
password = INFO['password']
connect_uri = f'mongodb://{user}:{password}@{host}:{port}'

client = AsyncIOMotorClient(connect_uri)

db = client['db_parker']

home_url = 'https://www.bilibili.com/ranking'
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2'}


def convertor(number_str):
'''
将小数点的万变为整数
:param number_str:
:return:
'''
number = re.search('(\d+\.+\d+)', number_str)
if number:
number = float(number.group(1))
if re.search('万', number_str):
number = int(number * 10000)
else:
number = 0

return number


async def home_page():
async with aiohttp.ClientSession() as session:
while True:
start = time.time()
async with session.get(url=home_url, headers=headers) as response:
html = await response.text()
resp = Selector(text=html)
items = resp.xpath('//ul[@class="rank-list"]/li')
for item in items:
json_data = {}
number = item.xpath('.//div[@class="num"]/text()').extract_first()
info = item.xpath('.//div[@class="info"][1]')
title = info.xpath('.//a/text()').extract_first()

detail_url = info.xpath('.//a/@href').extract_first()
play_number = info.xpath('.//div[@class="detail"]/span[1]/text()').extract_first()
viewing_number = info.xpath('.//div[@class="detail"]/span[2]/text()').extract_first()

json_data['number'] = int(number)
json_data['title'] = title
json_data['play_number'] = convertor(play_number)
json_data['viewing_number'] = convertor(viewing_number)

json_data['url'] = detail_url

task = asyncio.create_task(detail_list(session, detail_url, json_data))
# await detail_url()
end = time.time()
print(f'time used {end-start}')
await asyncio.sleep(SLEEP) # 暂停10分钟
print(f'sleep for {SLEEP}')


async def detail_list(session, url, json_data):
async with session.get(url, headers=headers) as response:
response = await response.text()
await parse_detail(response, json_data)


async def parse_detail(html, json_data=None):
resp = Selector(text=html)
info = resp.xpath('//div[@id="v_desc"]/div[@class="info open"]/text()').extract_first()

if not info:
info = '这个家伙很懒'

json_data['info'] = info.strip()
current = datetime.datetime.now()
json_data['crawltime'] = current
await db['bilibili'].update_one({'url': json_data['url']}, {'$set': json_data}, True, True)


loop = asyncio.get_event_loop()
loop.run_until_complete(home_page())




爬取的数据图:
Studio_3T_wIGBpehcXS.png

 
原创文章,转载请注明:http://30daydo.com/article/605
需要源码,可私信。

mmexport1585743466421.jpg

  收起阅读 »

爬虫nike登录流程抓包分析

<占坑> 敬请期待。
<占坑> 敬请期待。

深圳住房公积金验证码 识别破解

1212.PNG

 
http://gjj.sz.gov.cn/fzgn/zfcq/index.html
 
比较常规的验证码,使用keras全连接层,cv切割后每个字符只需要20个样本就达到准确率99%。
需要模型或者代码的私聊。
继续阅读 »
1212.PNG

 
http://gjj.sz.gov.cn/fzgn/zfcq/index.html
 
比较常规的验证码,使用keras全连接层,cv切割后每个字符只需要20个样本就达到准确率99%。
需要模型或者代码的私聊。 收起阅读 »

薅“疫情公益”羊毛,黑产恶意爬取各大出版社电子书上万册

疫情以来,所有企业都上班延期选择在线复工,在我们居家自我隔离期间,极验观察爬虫却没有消停,反而爬虫行为更加活跃且更胜往常。本周五,我们和无糖信息一起聊聊线上爬虫的“疫情”。
 
 
疫情以来,所有企业都上班延期选择在线复工,在我们居家自我隔离期间,极验观察爬虫却没有消停,反而爬虫行为更加活跃且更胜往常。本周五,我们和无糖信息一起聊聊线上爬虫的“疫情”。
 
 

scrapy在settings中定义变量不能包含小写!

如果变量名包含小写字母,那么你的变量会被过滤掉,在scrapy编码的其他地方都会无法被识别。
比如定义了一个叫 Redis_host = '192.168.1.1',的值
 
然后在spider中,如果你调用self.settings.get('Redis_host')
那么返回值是 None。
 
如果用REDIS_HOST定义,那么就可以正确返回它的值。
 
如果你一定要用小写,也有其他方法可正常调用。
先导入settings文件
fromt xxxx import setttings # xxx为项目名
 
host = settings.Redis_host # 直接导入一个文件的形式来调用是可以的
继续阅读 »
如果变量名包含小写字母,那么你的变量会被过滤掉,在scrapy编码的其他地方都会无法被识别。
比如定义了一个叫 Redis_host = '192.168.1.1',的值
 
然后在spider中,如果你调用self.settings.get('Redis_host')
那么返回值是 None。
 
如果用REDIS_HOST定义,那么就可以正确返回它的值。
 
如果你一定要用小写,也有其他方法可正常调用。
先导入settings文件
fromt xxxx import setttings # xxx为项目名
 
host = settings.Redis_host # 直接导入一个文件的形式来调用是可以的 收起阅读 »

etree.strip_tags的用法

直接从官方文档那里拿过来,发现这个函数功能还挺不错的。
它把参数中的标签从源htmlelement中删除,并且把里面的标签文本给合并进来。
 
举个例子:
from lxml.html import etree
from lxml.html import fromstring, HtmlElement

test_html = '''<p><span>hello</span><span>world</span></p>'''
test_element = fromstring(test_html)
etree.strip_tags(test_element,'span') # 清除span标签
etree.tostring(test_element)

因为上述操作直接应用于test_element上的,所以test_element的值已经被修改了。
 
所以现在test_element 的值是 
b'<p>helloworld</p>'

原创文章,转载请注明出处
http://30daydo.com/article/553
 
继续阅读 »
直接从官方文档那里拿过来,发现这个函数功能还挺不错的。
它把参数中的标签从源htmlelement中删除,并且把里面的标签文本给合并进来。
 
举个例子:
from lxml.html import etree
from lxml.html import fromstring, HtmlElement

test_html = '''<p><span>hello</span><span>world</span></p>'''
test_element = fromstring(test_html)
etree.strip_tags(test_element,'span') # 清除span标签
etree.tostring(test_element)

因为上述操作直接应用于test_element上的,所以test_element的值已经被修改了。
 
所以现在test_element 的值是 
b'<p>helloworld</p>'

原创文章,转载请注明出处
http://30daydo.com/article/553
  收起阅读 »

mumu模拟器adb无法识别

因为端口号被mumu改了。
 
            <Forwarding name="ADB_PORT" proto="1" hostip="127.0.0.1" hostport="7555" guestport="5555"/>
 
在mumu浏览器里面可以看到这个配置信息。
 
adb connect 127.0.0.1:7555
然后adb shell 就可以了。
 
配置文件名是:myandrovm_vbox86.nemu
继续阅读 »
因为端口号被mumu改了。
 
            <Forwarding name="ADB_PORT" proto="1" hostip="127.0.0.1" hostport="7555" guestport="5555"/>
 
在mumu浏览器里面可以看到这个配置信息。
 
adb connect 127.0.0.1:7555
然后adb shell 就可以了。
 
配置文件名是:myandrovm_vbox86.nemu 收起阅读 »

aiohttp异步下载图片

保存图片的时候不能用自带的open函数打开文件,需要用到异步io库 aiofiles来打开
url = 'http://xyhz.huizhou.gov.cn/static/js/common/jigsaw/images/{}.jpg'
headers={'User-Agent':'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'}
async def getPage(num):

async with aiohttp.ClientSession() as session:
async with session.get(url.format(num),headers=headers) as resp:
if resp.status==200:
f= await aiofiles.open('{}.jpg'.format(num),mode='wb')
await f.write(await resp.read())
await f.close()

loop = asyncio.get_event_loop()
tasks = [getPage(i) for i in range(5)]
loop.run_until_complete(asyncio.wait(tasks))

原创文章,
转载请注明出处:
http://30daydo.com/article/537
 
继续阅读 »
保存图片的时候不能用自带的open函数打开文件,需要用到异步io库 aiofiles来打开
url = 'http://xyhz.huizhou.gov.cn/static/js/common/jigsaw/images/{}.jpg'
headers={'User-Agent':'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'}
async def getPage(num):

async with aiohttp.ClientSession() as session:
async with session.get(url.format(num),headers=headers) as resp:
if resp.status==200:
f= await aiofiles.open('{}.jpg'.format(num),mode='wb')
await f.write(await resp.read())
await f.close()

loop = asyncio.get_event_loop()
tasks = [getPage(i) for i in range(5)]
loop.run_until_complete(asyncio.wait(tasks))

原创文章,
转载请注明出处:
http://30daydo.com/article/537
  收起阅读 »