由于申请的电信宽带IP不是固定的,每次变了IP又不知道,需要有个IP变更提醒的小功能。
思路是这样的,通过路由器的日志功能来收集IP变更或者断网提醒,开始我是打算直接搭建日志服务器来接收日志,但是发现断网的情况下不会给获取到的IP的关键日志发送到外网的服务器
。但是本地又可以接收到这条关键日志。所以就想到了钉钉的 Webhook 接入自定义服务!
02 申请机器人
✕
额,偷懒吧。电脑手机都可以。注意的是机器人必须在群里面添加,首先我们需要几个工具人的小伙伴!建了群直接踢掉~
添加机器人的步骤,直接看官网!
https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq
我用的是加签模式,大家根据需求自己搞哦
03 签名设置
✕
如果不是用的签名可以跳过这个步骤~
我们新建的py文件,名字就用jqr就好!
#python 3.8
import time
import hmac
import hashlib
import base64
import urllib.parse
timestamp = str(round(time.time() * 1000))
secret = '你的签填在这里'
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
print(timestamp)
print(sign)
上面是官网给提供的签名计算代码。
我们把下面添加一个return 给两个关键参数当做返回值就好啦!
04 运行测试
✕
接下来我们就要发送请求了!来测试下是否能发送数据!
在我们的jqr.py 下面新建一个函数!
import requests
import json
def fxx(nr="没有东西呀~"):
# 请求头,这个一定要有!不然报错!
headers = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
# 获取 timestamp 盒 sign 如果不用签 可以省略!
timestamp, sign = signs(bq)
# 合并请求地址 如果不用签 可以省略后面的 timestamp 和 sign!
# https://oapi.dingtalk.com/robot/send?access_token=XXXXXX×tamp=XXX&sign=XXX
url = 'https://oapi.dingtalk.com/robot/send?access_token=你的token×tamp=' + timestamp + '&sign=' + sign
# 测试看下URL 拼接是否有问题
# print(url)
# 发送文本内容 ,at的地方根据自己可以用或者不用
payload = {
"msgtype": "text",
"text": {
"content": nr
},
"at": {
"atMobiles": [
"你的手机号"
],
"isAtAll": 'false'
},
}
# 字典转换为 json 数据
jpayload = json.dumps(payload)
# 发送请求
strhtml = requests.post(url, data=jpayload, headers=headers)
# 转换返回数据
content = json.loads(strhtml.text)
# 打印返回数据
print(content)
fxx()
这样你的机器人就会给你发送一个 :没有东西呀~
05 日志收集+推送到钉钉完整代码
✕
需要把两个文件放在同一个目录呀!
jqr.py内容
#python 3.8
# 签名使用
import time
import hmac
import hashlib
import base64
import urllib.parse
# 请求使用
import requests
import json
# 加密签名
def signs():
timestamp = str(round(time.time() * 1000))
secret = '我居然给自己的签发上来了!95a8a6e0f3801af36e88c119334a15b9eb0968ee65c6c97f5a677afb4ca'
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
# print(timestamp)
# print(sign)
return (timestamp, sign)
def fxx(nr="没有东西呀~"):
# 请求头
headers = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
# 获取 timestamp 盒 sign
timestamp, sign = signs()
# 合并请求地址
# https://oapi.dingtalk.com/robot/send?access_token=XXXXXX×tamp=XXX&sign=XXX
url = 'https://oapi.dingtalk.com/robot/send?access_token=你的access_token×tamp=' + timestamp + '&sign=' + sign
# print(url)
# 发送文本内容
payload = {
"msgtype": "text",
"text": {
"content": nr
},
"at": {
"atMobiles": [
"需要艾特的人手机号"
],
"isAtAll": 'false'
},
}
jpayload = json.dumps(payload)
strhtml = requests.post(url, data=jpayload, headers=headers)
content = json.loads(strhtml.text)
# print(content)
if __name__ == '__main__':
fxx()
log.py内容
#python 3.8
import socket
import jqr
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 把下面的IP修改为你的
server.bind(('192.168.1.136', 514))
def SyslogToFile():
while True:
data = server.recvfrom(1024)
(LogMsg, host) = data
# 我这里tp-link 是GBK 编码!!如果你是utf-8就修改下
LogMsg = LogMsg.decode('GBK')
print(LogMsg)
jqr.fxx(LogMsg)
SyslogToFile()