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=2019052.部署教程
3.代码如下
import osimport refrom functools import wrapsfrom bark import barkfrom rsa import rsa_encryptfrom requests import Sessionfrom parsel import Selectordef 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 occurredif i == n:raise Exception(f"Retry for {n} times. Still get error: {repr(e)}")return wrapperreturn decoratordef 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, valuereturn wrapperreturn decoratordef 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").textsel = 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:breakres = 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_passwordPOST 登录;断言页面包含"安全退出"判断登录成功。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会话。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论