Skip to content

Instantly share code, notes, and snippets.

@monsterxcn
Last active August 21, 2021 01:05
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save monsterxcn/f0a7a083ee65d52b1400596ace43687b to your computer and use it in GitHub Desktop.
Save monsterxcn/f0a7a083ee65d52b1400596ace43687b to your computer and use it in GitHub Desktop.
Nonebot2 HEU 平安行动打卡插件,基于 Playwright 的模拟浏览器操作,无需 formData 等数据。但是务必保证事先填写并提交了一遍正确数据!代码仅学习使用
import nonebot
from nonebot import require
from nonebot.adapters.cqhttp import Bot, Message
from nonebot.adapters.cqhttp.event import MessageEvent
from nonebot.log import logger
from nonebot.plugin import on_command
from nonebot.typing import T_State
import time
from playwright.async_api import async_playwright
# 使用前需要自行安装 playwright 和 nonebot_plugin_apscheduler
# 然后使用 python3 -m playwright install 给 playwright 安装浏览器包
# 将此文件放入 src/plugins 文件夹下重启 nonebot2 机器人
customers = {
# "QQ 号码": ["HEU 认证学号", "HEU 认证密码", "名字"]
"123456789": ["2018010101", "01016789", "张三"],
"789456123": ["2018010102", "08084567", "李四"],
}
ADMIN = "管理员QQ号码" # 用于提供报错
GROUPNUMBER = "大伙的QQ群号码" # 用于提醒用户
ckMatch = on_command('打卡', priority=1)
@ckMatch.handle()
async def send_checkin(bot: Bot, event: MessageEvent, state: T_State):
id = str(event.get_user_id())
if customers[id] is not None:
# 如果发送「打卡」命令的用户 QQ 在上方已经有对应的 HEU 认证学号和密码则自动取出
await ckMatch.send(f"嘿您猜怎么着?我已经通过意念检测到你的学号和密码了,稍等!", at_sender=True)
(username, password) = (customers[id][0], customers[id][1])
else:
# 如果发送「打卡」命令的用户 QQ 在上方没有对应的 HEU 认证学号和密码,根据输入判断
input = event.get_plaintext().split(' ', 2)
if len(input) == 2:
# 检测到输入格式为「打卡 2018020202 pAsSwOrD」依次取出学号和密码
(username, password) = (input[0], input[1])
await ckMatch.send(Message("好的,收到你的学号和密码了,正在处理!over over.."), at_sender=True)
else:
# 检测到无法识别的格式,返回提示,结束
await ckMatch.finish(Message("[CQ:face,id=244] 不告诉我名字和密码怎么行呢!请使用如下格式提交:\n打卡[空格]学号[空格]密码"), at_sender=True)
# 异步执行单次打卡任务
msg = await checkin(username, password, "single")
if msg is not None:
# 执行成功,返回提示
await ckMatch.finish(Message(msg), at_sender=True)
else:
# 执行出错了,返回了一个 None,实测大部分时候是因为夜间与学校网络连接不好、服务器模拟操作未响应等
await ckMatch.finish(Message("服务器返回了空消息,可能与校园网络连接断开![CQ:face,id=146]"), at_sender=True)
# 打卡程序,采用 playwright 模拟浏览器操作
# https://playwright.dev/python/docs/intro#installation
async def checkin(username, password, mode):
loginCAS = "https://cas.hrbeu.edu.cn/cas/login?service=http://jkgc.hrbeu.edu.cn/infoplus/form/JSXNYQSBtest/start"
try:
msgRun = "Yes, master.\n"
async with async_playwright() as p:
browser = await p.chromium.launch()
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 我打开了浏览器\n'
page = await browser.new_page()
await page.goto(loginCAS)
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 我打开了 HEU 认证页\n'
await page.wait_for_selector('#login-submit')
await page.fill('#username', username)
await page.fill('#password', password)
await page.click('#login-submit')
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 我填上了信息并登录\n'
await page.wait_for_selector('#V1_CTRL82')
await page.check('#V1_CTRL82')
assert await page.is_checked('#V1_CTRL82') is True
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 我勾选了本人承诺信息可靠\n'
await page.click('text="确认填报"')
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 我点击了确认填报\n'
await page.wait_for_selector('button.dialog_button.default.fr')
await page.click('button.dialog_button.default.fr')
msgRun += f'[{time.strftime("%H:%M:%S", time.localtime())}] 好,等待返回结果\n\n'
# 刷新页面以便获取表单提交状态
await page.reload()
await page.wait_for_selector('#title_content > nobr')
if mode == "single":
# 如果传入 mode 为 single 则输出上述执行过程
msg = msgRun
else:
# 否则(用于后面定时任务)不输出
msg = ""
# 最终表单地址为 page.url,这里用 CAS 页面跳转供外网访问
msg += f'https://cas.hrbeu.edu.cn/cas/login?service={page.url}\n\n'
msg += f'任务结束于 {time.strftime("%H:%M:%S", time.localtime())},最终表单链接已经生成,你可以登录检查看看。'
if "已完成" in await page.text_content('#title_content > nobr'):
msg += f'总之,我觉得:[CQ:face,id=76] 打卡成功!'
else:
msg += f'总之,我觉得有哪里不对劲![CQ:face,id=146]'
return msg
except Exception as e:
logger.error(str(e))
return None
# 定义一个服务器时间 6 时 6 分 6 秒执行的定时任务,结果直接发送群内
scheduler = require("nonebot_plugin_apscheduler").scheduler
@scheduler.scheduled_job("cron", hour=6, minute=6, second=6)
async def auto_checkin():
# 获取 bot 信息
(bot,) = nonebot.get_bots().values()
for customer in customers:
# 拼接 CQ 码 at 某人,如果人员不在群里则会显示 at 事先定义的名字
msg = f"[CQ:at,qq={customer},name={customers[str(customer)][2]}] "
try:
# 传入了最后一个参数 mode=auto,打卡任务将不输出执行过程提示
msg += await checkin(customers[str(customer)][0], customers[str(customer)][1], 'auto')
await bot.send_group_msg(group_id=GROUPNUMBER, message=Message(msg))
except Exception as e:
logger.error(str(e))
# 如果定时任务执行出错了将向管理员用户私信报错信息
await bot.send_private_msg(user_id=ADMIN, message=f"{customer}(学号{customers[str(customer)][2]}) 今日打卡定时任务执行出错了!")
# 有的时候觉得机器人一堆消息发在群里难看,可以用合并转发
# 定义一个服务器时间 8 时 8 分 8 秒执行的定时任务,结果以合并转发形式发送群内
@scheduler.scheduled_job("cron", hour=8, minute=8, second=8)
async def auto_checkin_fwd():
# 获取 bot 信息
(bot,) = nonebot.get_bots().values()
fwdMsg = []
for customer in customers:
try:
msg = await checkin(customers[str(customer)][0], customers[str(customer)][1], 'auto')
# 使用 .append() 方法向 fwdMsg 添加节点
# https://docs.go-cqhttp.org/cqcode/#合并转发消息节点
fwdMsg.append({"type": "node", "data": {"name": f"{customers[str(customer)][2]} 平安行动", "uin": "3100017665", "content": msg}})
except Exception as e:
logger.error(str(e))
# 如果定时任务执行出错了将向管理员用户私信报错信息
await bot.send_private_msg(user_id=ADMIN, message=f"{customers[str(customer)][2]}(学号{customers[str(customer)][0]}) 今日打卡定时任务执行出错了!\n{str(e)}")
# 合并转发群消息 fwdMsg 已经构造好了,尝试发送
try:
await bot.send_group_forward_msg(group_id=GROUPNUMBER, messages=Message(fwdMsg))
except Exception as e:
logger.error(str(e))
# 如果合并转发群消息出错了将向管理员用户私信报错信息,出现此报错 可能 表明打卡成功
await bot.send_private_msg(user_id=ADMIN, message=f"打卡任务执行成功,但合并转发群消息好像出错了!\n{str(e)}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment