1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/DT4XYh若失效,可用地址
阿里云:
服务器购买地址
https://t.aliyun.com/U/DT4XYh
若失效,可用地址
https://www.aliyun.com/activity/wuying/dj?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 requests
from function.listening_socket import start_all_sockets
from function.user import get_user_name
from config import host, api, headers, log_file_name,check_in_sources
from util.file import write_log, read_log
from util.notice import email_notice
from util.timestamp import get_now
# 获取正在进行的
def get_listening():
response = requests.get(host + api["get_listening"], headers=headers)
if response.status_code == 200:
response_data = response.json()
return response_data["data"]
else:
return None
# 获取正在进行的课堂并且签到、写日志 新的签到方法
def get_listening_classes_and_sign(filtered_courses: list):
response = get_listening()
name = get_user_name()
# 短期存储查看PPT用的JWT、lessonId等信息
on_lesson_list = []
if response is None:
return None
else:
classes = list(response["onLessonClassrooms"])
if len(classes) == 0:
print("\n无课")
return
else:
print("\n发现上课")
for item in classes:
course_name = item["courseName"]
lesson_id = item["lessonId"]
response_sign = check_in_on_listening(lesson_id)
if response_sign.status_code == 200:
status = "签到成功"
print(course_name, status)
data = response_sign.json()["data"]
socket_jwt = data["lessonToken"]
jwt = response_sign.headers["Set-Auth"]
# print(jwt)
identity_id = data["identityId"]
def queue_on_listening_task():
on_lesson_list.append({
"ppt_jwt": jwt,
"socket_jwt": socket_jwt,
"lesson_id": lesson_id,
"identity_id": identity_id,
"course_name": course_name
})
if len(filtered_courses) == 0: # 无需过滤 全部进入监听队列
queue_on_listening_task()
else: # 只监听符合条件的课程 其余的就签到
if course_name in filtered_courses:
queue_on_listening_task()
# 将签到信息写入文件顶部
new_log = {
"id": lesson_id,
"title": course_name,
"name": course_name,
"time": get_now(),
"student": name,
"status": status,
"url": "https://changjiang.yuketang.cn/m/v2/lesson/student/" + str(lesson_id)
}
write_log(log_file_name, new_log)
else:
print("失败", response_sign.status_code, response_sign.text)
# 所有签到完成后,进行死循环巡查,检查是否出现答题
start_all_sockets(on_lesson_list)
# for item in on_lesson_list:
# start_socket_ppt(ppt_jwt=item["ppt_jwt"],socket_jwt=item["socket_jwt"], lesson_id=item["lesson_id"], identity_id=item["identity_id"])
# 获取正在进行的考试
def check_exam():
response = get_listening()
if response is None:
return None
else:
exams = list(response["upcomingExam"])
if len(exams) == 0:
print("无考试")
return
else:
print("发现考试")
print(exams)
# 发邮件提醒
email_notice(subject="雨课堂考试提醒", content="请打开雨课堂")
return
# 传入lessonId 签到
def check_in_on_listening(lesson_id):
sign_data = {
"source": check_in_sources["二维码"],
"lessonId": str(lesson_id),
"joinIfNotIn": True
}
response_sign = requests.post(host + api["sign_in_class"], headers=headers, json=sign_data)
return response_sign
# 是否已经签过(写入日志)
def has_in_checked(lesson_id):
logs = read_log(log_file_name)
if logs and logs[-1]["id"] == lesson_id:
print("已签过")
return True
else:
return False
# 收到的课程列表前check_num个全部进行签到、写日志 旧的签到方法 弃用
def check_in_on_latest(check_num=1):
data = {
"size": check_num,
"type": [],
"beginTime": None,
"endTime": None
}
name = get_user_name()
# 检查收到消息
response = requests.post(host + api["get_received"], headers=headers, json=data)
if response.status_code == 200:
response_data = response.json()
# 提取 `data` 列表中的第一个元素信息
if "data" in response_data and response_data["data"]:
courseware_info = response_data["data"][0]
courseware_id = courseware_info.get("coursewareId")
courseware_title = courseware_info.get("coursewareTitle")
course_name = courseware_info.get("courseName")
if has_in_checked(courseware_id):
return
print("标题:", courseware_title)
print("名称:", course_name)
response_sign = check_in_on_listening(courseware_id)
if response_sign.status_code == 200:
status = "签到成功"
print(name, status)
# 将签到信息写入文件顶部
new_log = {
"id": courseware_id,
"title": courseware_title,
"name": course_name,
"time": get_now(),
"student": name,
"status": status,
"url": "https://changjiang.yuketang.cn/m/v2/lesson/student/" + str(courseware_id)
}
write_log(log_file_name, new_log)
else:
print("失败", response_sign.status_code, response_sign.text)
else:
print("没有找到数据")
else:
print("请求失败:", response.status_code, response.text)
解析
该脚本为长江雨课堂的自动签到与课堂监听脚本。
主要作用
拉取正在上课/即将考试的信息,自动为正在上的课签到,把结果写入日志,并把需要监听互动(如PPT答题/课堂问答)的课程放入WebSocket 监听队列。
发现"即将考试"时邮件提醒。
依赖/外部模块
start_all_sockets(on_lesson_list)
: 开启/聚合所有课程的监听(WebSocket),用于实时答题等。get_user_name()
: 获取当前用户名称。config
:host、api、headers、log_file_name、check_in_sources
等配置。write_log、read_log
: 读写签到日志。email_notice
: 发送邮件通知。get_now
: 当前时间戳/格式化时间。
关键数据
从"正在上课"的接口里拿到
courseName、lessonId
等;签到成功后从响应中取:lessonToken(socket_jwt)
:用于课堂互动长连。响应头
Set-Auth(ppt_jwt)
:用于查看PPT接口授权。identityId
:身份标识。on_lesson_list
: 临时队列,收集待监听的课程参数(jwt、lessonId、课程名)。
方法职责
get_listening()
GET
onLessonClassrooms/upcomingExam
的聚合数据。成功返回
data
字段;失败返回None
。get_listening_classes_and_sign(filtered_courses: list)
成功:打印"签到成功",从响应里取
socket_jwt / ppt_jwt / identity_id
,按过滤规则决定是否加入on_lesson_list
(filtered_courses
为空=全部监听;不为空=只监听名单内课程,其余仅签到)。同时把签到结果写入日志(包含课程名、时间、学生名、课堂URL)。
调
get_listening()
拿到正在上课列表。对每个课堂执行
check_in_on_listening(lesson_id)
:全部签到完毕后,调用
start_all_sockets(on_lesson_list)
开始统一监听互动。check_exam()
从
get_listening()
里查看upcomingExam
。有考试则打印、邮件提醒(
email_notice
),无则提示"无考试"。check_in_on_listening(lesson_id)
source
(来自check_in_sources["二维码"]
)、lessonId
、joinIfNotIn=True
(未加入课堂则先加入)。POST 签到接口,payload 里包含:
返回原始
response
供上层判断。has_in_checked(lesson_id)
读日志
read_log
,如果最后一条的id
与入参一致,判定"已签过",用于旧方法的去重。
check_in_on_latest(check_num=1)
POST 拉取最近收到的课堂(按
check_num
条)。取第一条,调用
has_in_checked
去重后,用check_in_on_listening
签到、写日志。仅做"最近课堂"的一次性签到,不含 WebSocket 监听。
日志内容(写入)
id、title、name(课程名)、time(get_now)、student、status、url
用于之后的"是否已签过"判断和记录留痕。
运行流程(简述)
get_listening_classes_and_sign(filtered_courses)
—— 主入口:拉取正在上课 → 逐个签到 → 记录日志 → 组装监听参数 → 统一启动
start_all_sockets
。可选:
check_exam()
定期检查将来考试并邮件提示。
过滤逻辑说明:传入
filtered_courses
(课程名列表)。为空=所有正在上课的课都会进入监听队列;不为空=只有名单内的课会被监听,其它课只签到、不监听。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论