前言
本来一直在用这个频道的推送来追linux版本更新(以即时更新内核来爽),但是上个月这个频道不知怎么的似乎死了,到现在整整一个月过去了,4.19的longterm已经更新了5个版本,可它还是一点动静也没有。
那就自己写一个吧😋
开工
首先,要用什么来写,第一个想到的便是python,虽说别的东西也能写,但是一个小脚本配合丰富的第三方库,实现起来应该是相当简单的
不过python平时基本没摸过,那就来面向搜索引擎编程吧(不过u1s1,这东西真的面向搜索引擎就能轻松写起来..)
库
这种小脚本主要需要用到两类库吧,一类是用来发请求的,另一类是用来解析的(虽说其实自己手动解析字符串也不是不行。。)
发请求的库选定的是著名的,有45k star的 requests
解析库则是另一个著名的 Beautiful Soup 4
安装这些库,pip install requests bs4
pip是可以用代理的,比如pip install xxx --proxy="http://127.0.0.1:1080"
里面的http
在某个pip更新之后,就变成必须的了,否则直接报错
当然,在bash/cygwin环境下可以通过bash profile中增加https_proxy
和http_proxy
变量来全局代理
正式开工
开发环境:Windows + Notepad++
先导包
import requests as req
from bs4 import BeautifulSoup
py有一种特别的from import语句
from后面的东西是包名(文件夹+文件),而import后面的东西则是函数/类
as则是取个昵称,取完昵称之后原来的名字就不认了
有时候也可以用import xxx.*
来把函数/类导进来,但是据说这是要靠对应包的内置配置指定?
然后看看要爬的是啥
目标:kernel.org
说人话就是release这个table下面的每一行的第二列
程序要做的事情大概就是,先找到release这个table,然后在里面遍历每一行,把每一行的第二列放入一个数组(列表)然后返回
总的来说,这个页面的结构还是相当鲜明的(好评)
发送请求
https://docs.python-requests.org/en/master/api/#requests.get
res = req.get("https://kernel.org")
(这里我在import的时候把requests库重命名为req了)
嗯,一行就搞定了,返回的是一个requests.Response
对象,我将其命名为res
解析请求
刚刚拿到了requests.Response
的一个对象,这个对象可以有许多的操作
https://docs.python-requests.org/en/master/api/#requests.Response
此处只需要使用其text属性,拿出返回的正文信息
然后用它来构造一个BeautifulSoup对象
bs = BeautifulSoup(res.text)
嗯,其实这样就可以了,但是会抛出一个warning,此时只需要按照提示再加上一个参数即可(这种是叫 关键字参数? 不过去掉那个features=
也是没有任何问题的)
bs = BeautifulSoup(res.text,features="html.parser")
之后,就可以跟着官方文档开干了
https://www.crummy.com/software/BeautifulSoup/bs4/doc.zh/
按照上面所说的思路,首先找到那个release的table<table id="release">
根据文档,我们可以使用find
方法,并指定id
该方法返回符合搜索条件的第一个东西,由于我们要找的东西只有一个,因此不需要使用能返回所有结果的find_all
方法
(find_all
返回一个包含所有结果的list迭代器(list_iterator),当作数组用for循环遍历即可)
table = bs.find("table", id="releases")
之后,对这个table里面的东西进行遍历(遍历每一行)
根据文档,可以使用table.children
来获取其子元素列表(其实也是个list_iterator)
在这之前,可以先了解一下遍历出来的东西可能有哪几种类型
- Tag
对应html里的tag (比如div、a、h1都是) - NavigableString
可遍历的字符串,可以理解为字符内容,比如<h1>aaa</h1>
里面装着的aaa - BeautifulSoup
整个文档,当成Tag就完事了 - Comment
文档中的注释内容,可以视作特殊类型的NavigableString
然后,我们开始遍历
ret = []
table = bs.find("table", id="releases")
for row in table.children:
ret.append(row.find_all("td")[1].string)
来解释一下这里的逻辑吧,上面说过了.children
返回其所有的子节点的一个迭代器(可以当作列表),这里面的子节点应该就是行(<tr>
)了,我们在for循环中对每一行进行遍历,寻找行中所有列(<td>
)节点,取出第二列的值(就是上面所说的NavigableString)
运行一下,炸了。。AttributeError: 'NavigableString' object has no attribute 'find_all'
wtf,这是怎么回事,我们遍历出来的行,居然是个NavigableString?然后对这个行进行一个find_all
就炸了
奇了怪了
接着我在循环中打印了一下变量的类型
for row in table.children:
print(type(row))
????
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
<class 'bs4.element.Tag'>
<class 'bs4.element.NavigableString'>
嗯?居然是Tag和NavigableString的混合体?
很明显,那些Tag是我们的目标<tr>
接着,来一下.contents
,根据官方文档,这将会把子节点以列表形式输出,好让我看看那些NavigableString都是啥
table = bs.find("table", id="releases")
print(table.contents)
结果:
['\n', <tr align="left">
<td>mainline:</td>
懂了,虽然chrome里看源码,table里只有<tr>
,但是实际上中间依然夹杂了一些空行,这些空行被作为字符对象也就是NavigableString输出了
有两种解决方法,一种是进行一下类型判断
from bs4.element import *
for row in table.children:
print(type(row))
if type(row) != Tag:
continue
ret.append(row.find_all("td")[1].string)
需要import
一下bs4.element
,不然这玩意儿不认
还有一种是对table进行find_all
搜索,只搜索Tag节点
for row in table.find_all("tr"):
ret.append(row.find_all("td")[1].string)
(最后我用了第一种,别问为什么)
好啦,这样我们就能获得装着版本号的列表了
['5.12-rc7', '5.11.15', '5.10.31', '5.4.113', '4.19.188', '4.14.231', '4.9.267', '4.4.267', 'next-20210416']
比较版本
别忘了要求,这是一个抓取linux版本更新的bot,也就是说,它需要知道哪些版本更新了
思路很简单,一个新的版本列表,一个旧的版本列表,假如新列表里有东西在旧列表里没有,那些东西就是新的版本内容
def list_cmp(list_old, list_new):
ret = []
for element in list_new:
if list_old.count(element) == 0:
ret.append(element)
return ret
telegram 推送
首先准备好
- 一个bot (可以参阅 https://zhuanlan.zhihu.com/p/59228574)
- 一个频道(推送消息的)
然后,把你的bot拉进频道里,并且设置为管理员
然后,在频道里发送一则消息,通过私聊转发给你的bot
之后,浏览器打开 https://api.telegram.org/bot<你的token>/getUpdates
(把<你的token>
整体换成token,别留尖括号)
然后,应该可以看到刚刚发给bot的消息的json格式内容,找到其中的forward_from_chat
部分,后面紧跟着一个id
,把id后面的数字记录下来(可能是负数,很正常),这在给telegram api发请求时会用到(人家要知道你要发给哪个频道啊喂)
然后,诶嘿,使用requests
库发post请求就是啦
(requests库被我重命名为req了,将就着看)
chat_id = ""
bot_id = ""
def send(what):
status = 0
while status != 200:
status = req.post("https://api.telegram.org/bot" + bot_id + "/sendMessage",
{"parse_mode":"Markdown", "text":what, "chat_id":chat_id}).status_code
post
方法的第二个参数为post的内容,其中的第一个为telegram的解析模式(为了能够显示粗体,链接之类的,可以指定为markdown或者html,看习惯啦),第二个为消息内容,第三个是要发送到哪个对话里(上面获取的chat_id)
当然,telegram的markdown有些奇怪😢,似乎有点不正宗,比如粗体是*aaa*
而不是**aaaa**
(第二种直接不认)
组装
倒数第二步,组装上面的逻辑,形成完整代码😋
#! /usr/bin/python3
import requests as req
import time
import logging
from bs4.element import *
from bs4 import BeautifulSoup
chat_id = "-1001174556663"
bot_id = "941245555:AAERjsDUPak_P5SJ2HeA4do6S8OVOyBpdi8"
logging.basicConfig(level=logging.INFO,
filename='/root/kernel_org/log',
filemode='w',
format=
"%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s"
)
def fetch_once():
ret = []
try:
res = req.get("https://kernel.org")
bs = BeautifulSoup(res.text,features="html.parser")
table = bs.find("table", id="releases")
for row in table.children:
if type(row) != Tag:
continue
ret.append(row.find_all("td")[1].string)
logging.info(ret)
except BaseException as e:
logging.error(e)
return None
return ret
def list_cmp(list_old, list_new):
ret = []
for element in list_new:
if list_old.count(element) == 0:
ret.append(element)
return ret
def send(what):
status = 0
while status != 200:
logging.info("Try sending "+what)
status = req.post("https://api.telegram.org/bot" + bot_id + "/sendMessage",
{"parse_mode":"Markdown", "text":what, "chat_id":chat_id}).status_code
last_result = fetch_once()
while True:
time.sleep(300)
next_result = fetch_once()
if next_result is None:
continue
for item in list_cmp(last_result, next_result):
send("*New Linux Kernel Release*\n["+item+"](https://www.kernel.org/)")
last_result = next_result
引入了logging方便记录日志,time用来设置多久爬取一次页面,解析页面的部分用try catch罩了起来以防止意外情况(比如这页面突然抽了导致脚本崩溃)
部署
写一个systemd启动😋
[Unit]
Description=Korg service
After=network.target
Wants=network.target
[Service]
Type=simple
ExecStart=/usr/bin/python3 <脚本路径>
[Install]
WantedBy=multi-user.target
丢到/etc/systemd/system/kernel.org.service
然后systemctl enable kernel.org.service
systemctl start kernel.org.service
大功告成
可以通过systemctl status kernel.org.service
或者日志文件内容来查看状态
只要enable了,重启后服务也会自动启动的,无需担心
成果展示
额外的踩坑
- CR LF的传统恶心问题,由于我是在Windows下面写的脚本,Linux下面部署,因此中间需要用
dos2unix
转换一下,否则直接./脚本名
的时候会出现bash报错/usr/bin/python3^M
这种第一行的可执行文件路径找不到的问题(当然,直接调用python是没事的,用bash跑才会这样)