写几个青龙面板的脚本（Python脚本）

自己用的python脚本，主要是根据自己的需求写的脚本，之前本人论文投稿，专门搞了个脚本监控我的论文cnki啥时候收录

脚本1、获取公网ip，达到远程访问的目的，可以导入青龙本地的推送模块notify.py

import requests, json, re, os

import datetime

QYWX_AM = ["", "", "", ""]#替换成自己的企业微信应用的信息

CORPID = QYWX_AM[0]

CORPSECRET = QYWX_AM[1]

AGENTID = QYWX_AM[3]

touser = QYWX_AM[2]

def get_access_token():

url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"

values = {

"corpid": CORPID,

"corpsecret": CORPSECRET,

}

req = requests.post(url, params=values)

data = json.loads(req.text)

return data["access_token"]

def push_text(title="企业微信", content="内容"):

send_url = (

"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="

+ get_access_token()

)

tm = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

message = tm + 'rn' + title + 'rn' + 'rn' + content

send_values = {

"touser": touser,

"msgtype": "text",

"agentid": AGENTID,

"text": {"content": message},

"safe": "0",

}

send_msges = bytes(json.dumps(send_values), "utf-8")

respone = requests.post(send_url, send_msges)

respone = respone.json()

if respone["errcode"] == 0:

return ("企业微信机器人推送成功！")

else:

return ("企业微信机器人推送失败！")

def IPchaxun():

headers = {

'accept': '*/*',

'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',

'cache-control': 'no-cache',

'content-type': 'application/json;charset=UTF-8',

'origin': 'https://www.lddgo.net',

'pragma': 'no-cache',

'priority': 'u=1, i',

'referer': 'https://www.lddgo.net/',

'sec-ch-ua': '"Chromium";v="130", "Microsoft Edge";v="130", "Not?A_Brand";v="99"',

'sec-ch-ua-mobile': '?0',

'sec-ch-ua-platform': '"Windows"',

'sec-fetch-dest': 'empty',

'sec-fetch-mode': 'cors',

'sec-fetch-site': 'same-site',

'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0',

}

response = requests.get('https://openapi.lddgo.net/base/gtool/api/v1/GetIp', headers=headers)

response = response.json()

res = response["data"]["ip"]

print(res)

return (res)

ip = IPchaxun()

s1="http://"+ip+":631/printers/Brother_MFC-7360"

s2="青龙面板："+ip+":5700"

sms=s1 + 'rn' + 'rn' + s2

print(push_text("家庭打印机提醒", sms))

脚本2、本人上班要走高速，我很关注高速口信息，接口湖北高速，使用青龙自带的notify 推送信息

import requests,json

from notify import send

def get_satation_info(eventid="T20250112171533531603652369"):

url = 'https://www.02712122.com/HuBeiGSTAppAPIServer/roadpoi/getEvent'

headers = {

"accept": "application/json, text/plain, /",

"accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",

"authorization": "123123",

"content-type": "application/x-www-form-urlencoded",

"origin": "https://www.02712122.com",

"referer": "https://www.02712122.com/HubeiOfficialAccounts/serverHall/CharterPollicy?eventid=T20250112171533531603652369&longitude=112.93886&latitude=28.22778",

"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36 Edg/133.0.0.0"

}

data="eventid="+eventid

response = requests.post(url, headers=headers, data=data)

res = response.json()

# note = res["data"]["name"] + " " + res["data"]["reportout"]

note = res["data"]["reportout"]

# print(note)

return note

def get_keyword_to_satation_info(keyword="凤凰山站"):

url = "https://www.02712122.com/HuBeiGSTAppAPIServer/roadpoi/getPoiList"

headers = {

"accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",

"accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",

"authorization": "123123",

"content-type": "application/x-www-form-urlencoded",

"origin": "https://www.02712122.com",

"referer": "https://www.02712122.com/HubeiOfficialAccounts/serverHall/FreePolicy",

"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36 Edg/133.0.0.0"

}

data = {

"longitude": "112.93886",

"latitude": "28.22778",

"page": "1",

"size": "9999",

"searchtype": "",

"keyword": keyword,

"roadoldid": "",

"areaid": "",

"lv": ""

}

response = requests.post(url, headers=headers, data=data)

response = response.text

res = json.loads(response)

satations = res["data"]

# print(satations)

satation_info = ""

for i in range(len(satations)):

satation = satations[i]["name"]

if "eventid" in satations[i]:

eventid = satations[i]["eventid"]

info = get_satation_info(eventid)

else:

eventid = "False"

info="正常"

satation_info += satation + " " + info + "nn"

# print(satation,satation_info)

return satation_info

saa = get_keyword_to_satation_info("嘉鱼北") + get_keyword_to_satation_info("凤凰山")

print(saa)

send("上下班的高速口状态", saa)

脚本3、关注好价信息，修改keywords，使用青龙自带的notify 推送

import re

import os

import sys

import json

import requests

import datetime

from notify import send

def Txt_Create(path="zklog.txt", msg="msg"): # path是指定文件路径，msg是写入的文件内容

if os.path.isfile("zklog.txt"):

# print('文件已存在')

pass

else:

txt_file = open("zklog.txt", 'w')

txt_file.write(msg)

# print('文件创建成功')

Txt_Create()

def YSF():

header = {

"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",

"Accept-Encoding": "deflate",

"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",

"Cache-Control": "max-age=0",

"Connection": "keep-alive",

"Cookie": "timezone=8; mochu_us_notice_alert=1",

"Host": "new.ixbk.net",

"Upgrade-Insecure-Requests": "1",

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

}

re0 = requests.get("http://new.ixbk.net/", headers=header)

re0.encoding = 'utf-8'

re0 = re0.text

if "置顶" in re0:

re2=re0.split("置顶")

print("置顶帖数量："+str(len(re2)-1))

re0=re2[-1]

# print(re0)

CB = re.compile(r'href="(.*?)" title=(.*?)" target="_blank" data-catename="(.*?)" data-content="(.*?)"') # 查找数字

CBS = CB.findall(re0)

print(len(CBS))

if len(CBS) > 10:

keywords =["YSF", "ysf", "云闪付", "招行", "掌上生活", "建行", "建行生活", "招商银行", "建设银行", "cmb","放水","E卡","好价","红包",'秒杀', '放水', '福利', '神价', '冲好价', '撸神券', 'bug', '刚中', '实测', '无门槛', '手慢无', '好价', '大猫', '大毛', 'BUG', '漏洞', '洗衣液', '收纳', '1号店', '一号店', '0元', '零元', '湿厕纸', 'plus', '小号','放shui', '防水', '吧机', '家电', '水壶', '小米', '全品', '9.9-9.8', '11-10', '洗衣珠', '凝珠', '监控', '摄像头', '云台', '1元', '一元', '指纹锁', '碰一碰', '碰一下']

back = ""

for aa in range(1,len(CBS)):

a = CBS[aa][3]

print(a)

# while any(keyword in a for keyword in keywords)

if any(keyword in a for keyword in keywords):

print("http://new.ixbk.net"+CBS[aa][0]+" "+CBS[aa][1]+" "+CBS[aa][2]+" "+CBS[aa][3])

back = back + "http://new.ixbk.net" + CBS[aa][0] + "t" + CBS[aa][1] + "t" + CBS[aa][3] + "t"

break

else:

back = ""

return back

def compare():

temp = open("zklog.txt", 'r', encoding="utf-8").read().strip()

a = YSF().strip()

# print(temp,a)

if a != "":

if temp == a:

print("未发现目标")

else:

print("发现目标")

print(a)

with open("zklog.txt", "w", encoding="utf-8") as fp:

fp.write(a)

fp.close()

a1 = a.split("t")

print(send("PUSH信息", a1[1] + "rn" + "rn" + a1[2] + "rn" + "rn" + a1[0] + "rn" + "rn"))

else:

print("被网页锁定")

compare()

方法：a把脚本添加到脚本管理中去

b、将脚本自己添加到任务中去 task XXX.py

作者声明本文无利益相关，欢迎值友理性交流，和谐讨论～