1.购买服务器阿里云：服务器购买地址https://t.aliyun.com/U/DT4XYh若失效，可用地址

1.购买服务器

阿里云：

服务器购买地址

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.部署教程

2024年最新青龙面板跑脚本教程（一）持续更新中

3.代码如下

import requests
from function.listening_socket import start_all_socketsfrom function.user import get_user_namefrom config import host, api, headers, log_file_name,check_in_sourcesfrom util.file import write_log, read_logfrom util.notice import email_noticefrom 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（课程名列表）。为空=所有正在上课的课都会进入监听队列；不为空=只有名单内的课会被监听，其它课只签到、不监听。

注意：

本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。

历史脚本txt文件获取>>

服务器搭建，人工服务咨询>>

网赚：日进千刀

2025年9月16日星期二

长江雨课堂任务脚本