1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/E8o0aM若失效,可用地址
阿里云:
服务器购买地址
https://t.aliyun.com/U/E8o0aM
若失效,可用地址
https://www.aliyun.com/daily-act/ecs/activity_selection?source=5176.29345612&userCode=49hts92d
腾讯云:
https://curl.qcloud.com/wJpWmSfU
若失效,可用地址
https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console
华为云
https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=201905
2.部署教程
3.代码如下
import os
import re
from functools import wraps
from bark import bark
from rsa import rsa_encrypt
from requests import Session
from parsel import Selector
def retry(n: int):
"""如果遇到报错,重新尝试n次"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for i in range(1, n + 1):
try:
return func(*args, **kwargs)
except Exception as e:
# 重试了n次依然有错误,直接抛出一个错误
# 这会触发连锁的错误 During handling of the above exception, another exception occurred
if i == n:
raise Exception(
f"Retry for {n} times. Still get error: {repr(e)}"
)
return wrapper
return decorator
def warn_wrapper(threshold: float):
"""在函数返回值低于threshold的时候发出警告"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
name, value = func(*args, **kwargs)
if value < threshold:
bark(
title=name,
body=str(value),
group="FDU",
token=os.getenv("bark_token"),
icon="https://assets.elearning.fudan.edu.cn/cdn/brand/20250930/xiaohui.gif",
)
return name, value
return wrapper
return decorator
def login(se: Session):
res = se.get("https://uis.fudan.edu.cn/authserver/login")
sel = Selector(res.text)
hidden_input = dict(
zip(
sel.css("#casLoginForm > input[type=hidden]::attr(name)").getall(),
sel.css("#casLoginForm > input[type=hidden]::attr(value)").getall(),
)
)
res = se.post(
url="https://uis.fudan.edu.cn/authserver/login",
data={
"username": os.getenv("fudan_username"),
"password": os.getenv("fudan_password"),
**hidden_input,
},
)
assert "安全退出" in res.text, "登陆失败"
def get_balance(se: Session):
"""
获取一卡通余额
"""
content = se.get("https://ecard.fudan.edu.cn/epay/myepay/index").text
sel = Selector(text=content)
name = sel.css(".custname::text").get()[3:].replace("!", "")
balance = sel.css(
"div.payway-box-bottom > div:nth-child(1) > p:nth-child(1)::text"
).get()
return (name + "一卡通余额", float(balance))
def get_dom_elec_surplus(se: Session):
"""
查询宿舍电费余量
示例输出:
{
'e': 0,
'm': '操作成功',
'd': {
'xq': '北区',
'ting': False,
'xqid': '10',
'roomid': '1173',
'tingid': '',
'realname': '张杨',
'ssmc': '999号楼',
'fjmc': '909A',
'fj_update_time': '',
'fj_used': 15788.2,
'fj_all': 16016.4,
'fj_surplus': 228.2,
't_update_time': 0,
't_used': 0,
't_all': 0,
't_surplus': 0
}
}
"""
# 不断重定向,直到拿到lck参数
res = se.get(
r"https://zlapp.fudan.edu.cn/fudanelec/wap/default/info", allow_redirects=False
)
while res.status_code == 302 or res.status_code == 301:
location = res.headers.get("Location")
if "lck=" in location:
break
res = se.get(location, allow_redirects=False)
# 提取lck参数
location = res.headers.get("Location")
lck = re.search("lck=(.*?)&", location).group(1)
res = se.get(location)
# 请求认证方法
queryAuthMethods = se.post(
"https://id.fudan.edu.cn/idp/authn/queryAuthMethods",
json={"entityId": "https://zlapp.fudan.edu.cn", "lck": lck},
)
# 使用账号密码认证
passwd_way = queryAuthMethods.json()["data"][0]
assert passwd_way["moduleCode"] == "userAndPwd"
authChainCode = passwd_way["authChainCode"]
# 请求rsa公钥
getJsPublicKey = se.get("https://id.fudan.edu.cn/idp/authn/getJsPublicKey")
pub_key = getJsPublicKey.json()["data"]
# 构造登录参数
d = {
"authModuleCode": "userAndPwd",
"authChainCode": authChainCode,
"entityId": "https://zlapp.fudan.edu.cn",
"requestType": "chain_type",
"lck": lck,
"authPara": {
"loginName": os.getenv("fudan_username"),
"password": rsa_encrypt(os.getenv("fudan_password"), pub_key),
"verifyCode": "",
},
}
# 提交登录请求
loginToken = se.post(
"https://id.fudan.edu.cn/idp/authn/authExecute", json=d, allow_redirects=False
)
# 登录完成之后请求zlapp的授权
res = se.post(
"https://id.fudan.edu.cn/idp/authCenter/authnEngine?locale=zh-CN",
data={"loginToken": loginToken.json()["loginToken"]},
)
# 授权平台会给我们一个直链
auth_link = (
re.search('locationValue = "(.*?)"', res.text).group(1).replace("&", "&")
)
# 访问直链,获取电费信息
res = se.get(auth_link).json()
return (
res["d"]["xq"] + res["d"]["ssmc"] + res["d"]["fjmc"] + "电量余量",
float(res["d"]["fj_surplus"]),
)
if __name__ == "__main__":
se = Session()
login(se)
print(get_balance(se))
print(get_dom_elec_surplus(se))
该脚本为复旦大学自动查询脚本。
主要作用
使用学校统一认证登录后,自动查询并输出两项信息:
一卡通余额;
宿舍用电剩余度数。
当余额/电量低于预设阈值时,通过 Bark 推送告警。所有关键请求自带失败重试机制。
主要方法
retry(n)
(装饰器)
给被装饰的函数增加最多 n 次重试。函数抛错会自动重试,超出次数仍失败则抛出统一异常。warn_wrapper(threshold)
(装饰器)
约定被装饰函数返回(name, value)
。若value < threshold
,用bark(...)
发送告警(标题为name
,正文为数值)。login(se: Session)
访问https://uis.fudan.edu.cn/authserver/login
获取隐藏表单字段,携带fudan_username / fudan_password
POST 登录;断言页面包含"安全退出"判断登录成功。get_balance(se: Session)
(带@retry(3)
与@warn_wrapper(20)
)
访问一卡通首页https://ecard.fudan.edu.cn/epay/myepay/index
,用 CSS 选择器解析姓名与余额,返回("xxx一卡通余额", 余额)
;余额低于 20 触发 Bark 告警。get_dom_elec_surplus(se: Session)
(带@retry(3)
与@warn_wrapper(30)
)
走电费查询的统一身份认证链路:连续请求直到拿到
lck
;调用
queryAuthMethods
获取认证方式与authChainCode
;获取 RSA 公钥,用
rsa_encrypt(...)
对密码加密;提交
authExecute
获得loginToken
,再到authnEngine
换取跳转直链;访问直链得到电费 JSON,取
fj_surplus
(宿舍剩余电量)。
返回("校区+楼名+房间电量余量", 剩余度数)
;低于 30 触发 Bark 告警。__main__
区域
创建requests.Session()
;先login(se)
,再分别打印get_balance(se)
与get_dom_elec_surplus(se)
的结果。
依赖/环境变量:
需要
bark_token, fudan_username, fudan_password
等环境变量;依赖
bark
推送、rsa_encrypt
加密、parsel.Selector
解析、requests
会话。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论