study-tool
302 lines
11 KiB
Python
302 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
study.dgjy.net（莞易学 / 东莞教师培训）视频课时 —— 自动完成 + 完成度核验
|
||
|
||
机制（逆向结论）
|
||
* 播放器是网易云课堂 EduPlayer，进度上报接口：
|
||
POST /train/course/front/period/study/log
|
||
body {"periodId":4133,"isComplete":0|1,"lastProgressTime":"<秒>"}
|
||
* onTimeupdate 时若 currentTime > maxTime 才上报（isComplete=0）
|
||
* 视频播到结尾触发 onPlayEnd -> POST {"periodId":x,"isComplete":1} ← 这才是"完成"
|
||
* onSeek 守卫：仅当 n.newData !== duration 且 allowVideoDrag === "0" 且 newData > maxTime
|
||
时才会重新拉签名把进度拉回去。所以「拖到正好结尾」不会被拦。
|
||
* 上报后 10 秒自动跳到下一课时。
|
||
|
||
完成度核验（唯一可信来源）
|
||
GET /train/course/front/period/list/tree/studyLog?courseId=<MOOC课程ID>&isFront=1
|
||
每个课时节点里的 studyLog.completeTime 非空 = 平台真的记为完成。
|
||
|
||
用法
|
||
python study_video.py verify <trainCourseId> # 只核验，不动数据
|
||
python study_video.py run <trainCourseId> # 完成所有未完成的视频课时
|
||
"""
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
|
||
import requests
|
||
|
||
# 代码在 src/study_tool 包里，命令行脚本自己把 src 加进 sys.path
|
||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))
|
||
|
||
from study_tool import store # noqa: E402
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
BASE = "https://study.dgjy.net"
|
||
KV_SESSION = "study_session" # 会话存在 SQLite 的 kv 表（和网页端同一份库）
|
||
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36 Edg/154.0.0.0")
|
||
CDP = "http://127.0.0.1:9222"
|
||
|
||
|
||
def load_session():
|
||
"""从 SQLite 读上次抓到的会话；没有返回 None。"""
|
||
sess = store.kv_get(KV_SESSION)
|
||
return sess if isinstance(sess, dict) and sess.get("token") else None
|
||
|
||
|
||
|
||
# ------------------------------------------------------------------ 会话 / 认证
|
||
|
||
def capture_session():
|
||
"""从已登录的浏览器里取 Middle-Token 与 cookie。"""
|
||
from playwright.sync_api import sync_playwright
|
||
with sync_playwright() as p:
|
||
b = p.chromium.connect_over_cdp(CDP)
|
||
cookies = b.contexts[0].cookies()
|
||
token = ""
|
||
for c in cookies:
|
||
if c["name"] == "Middle-Token" and "study.dgjy.net" in (c.get("domain") or ""):
|
||
token = c["value"]
|
||
if not token:
|
||
raise SystemExit("没找到 Middle-Token —— 请先在浏览器里登录 study.dgjy.net")
|
||
sess = {"token": token,
|
||
"cookies": [{"name": c["name"], "value": c["value"],
|
||
"domain": c["domain"], "path": c["path"]} for c in cookies]}
|
||
store.kv_set(KV_SESSION, sess)
|
||
return sess
|
||
|
||
|
||
def api(sess=None):
|
||
if sess is None:
|
||
sess = load_session() or capture_session()
|
||
s = requests.Session()
|
||
s.headers.update({"User-Agent": UA, "Accept": "application/json, text/plain, */*",
|
||
"Content-Type": "application/json",
|
||
"Authorization": "Bearer " + sess["token"],
|
||
"Referer": BASE + "/training/courseVideo"})
|
||
for c in sess["cookies"]:
|
||
if "dgjy.net" in (c.get("domain") or ""):
|
||
s.cookies.set(c["name"], c["value"] or "",
|
||
domain=(c.get("domain") or "").lstrip("."), path=c.get("path") or "/")
|
||
return s
|
||
|
||
|
||
def check_auth(s):
|
||
r = s.get(BASE + "/train/getInfo", timeout=30)
|
||
try:
|
||
return r.json().get("code") == 200
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ------------------------------------------------------------------ 课程结构
|
||
|
||
def course_info(s, train_id):
|
||
return s.get(BASE + "/train/course/front/trainCourse/getById?id=%s" % train_id,
|
||
timeout=30).json().get("data") or {}
|
||
|
||
|
||
def period_tree(s, course_id, is_front="1"):
|
||
r = s.get(BASE + "/train/course/front/period/list/tree/studyLog?courseId=%s&isFront=%s"
|
||
% (course_id, is_front), timeout=30)
|
||
return r.json().get("data") or []
|
||
|
||
|
||
def flatten(nodes, out=None, parent=""):
|
||
out = [] if out is None else out
|
||
for n in nodes or []:
|
||
sl = n.get("studyLog") or {}
|
||
out.append({
|
||
"periodId": n.get("id"),
|
||
"name": n.get("name") or "",
|
||
"parent": parent,
|
||
"periodType": str(n.get("periodType") or ""),
|
||
"duration": n.get("duration"),
|
||
"periodScore": n.get("periodScore"),
|
||
"allowVideoDrag": n.get("allowVideoDrag"),
|
||
"completeLimit": n.get("completeLimit"),
|
||
"kcdm": n.get("kcdm"),
|
||
"completeTime": sl.get("completeTime"),
|
||
"logPeriodScore": sl.get("logPeriodScore"),
|
||
"lastProgressTime": sl.get("lastProgressTime"),
|
||
})
|
||
flatten(n.get("children"), out, n.get("name") or parent)
|
||
return out
|
||
|
||
|
||
def video_periods(s, train_id):
|
||
"""返回 (培训课程信息, MOOC课程ID, 视频课时列表)"""
|
||
info = course_info(s, train_id)
|
||
top = flatten(period_tree(s, train_id, ""))
|
||
mooc = ""
|
||
for t in top:
|
||
if t["kcdm"]:
|
||
mooc = t["kcdm"]
|
||
break
|
||
vids = []
|
||
if mooc:
|
||
vids = [p for p in flatten(period_tree(s, mooc, "1")) if p["periodType"] == "2"]
|
||
return info, mooc, vids
|
||
|
||
|
||
# ------------------------------------------------------------------ 核验
|
||
|
||
def report(info, vids):
|
||
print()
|
||
print("课程：%s" % info.get("courseName", "?"))
|
||
st = info.get("study") or {}
|
||
print("进度：%s/%s 课时 规则 completeRule=%s" %
|
||
(st.get("periodStudy"), st.get("periodTotal"), info.get("completeRule")))
|
||
print()
|
||
print("%-8s %-40s %-8s %-6s %-20s %s" %
|
||
("periodId", "课时名", "时长(s)", "拖拽", "completeTime", "判定"))
|
||
done = 0
|
||
for v in vids:
|
||
ok = bool(v["completeTime"])
|
||
done += ok
|
||
print("%-8s %-40s %-8s %-6s %-20s %s" %
|
||
(v["periodId"], v["name"][:40], v["duration"], v["allowVideoDrag"],
|
||
v["completeTime"] or "-", "✅ 已完成" if ok else "❌ 未完成"))
|
||
print()
|
||
print("视频课时：%d/%d 已完成" % (done, len(vids)))
|
||
return done, len(vids)
|
||
|
||
|
||
# ------------------------------------------------------------------ 自动完成
|
||
|
||
def complete_one(pg, train_id, train_course, period_id, timeout=180):
|
||
"""打开视频页 -> 拖到结尾 -> 播放 -> 等 isComplete=1 上报。返回 True/False。"""
|
||
url = ("%s/training/courseVideo?classhour=%d&isJoin=1&needJoin=1&courseId=%s&isFront=1"
|
||
"&trainCourseId=%s" % (BASE, period_id, train_course, train_id))
|
||
posted = {"done": False}
|
||
|
||
def on_req(r):
|
||
try:
|
||
if "study/log" in r.url and "detail" not in r.url:
|
||
body = r.post_data or ""
|
||
if '"isComplete":1' in body.replace(" ", ""):
|
||
posted["done"] = True
|
||
except Exception:
|
||
pass
|
||
|
||
pg.on("request", on_req)
|
||
pg.goto(url, wait_until="domcontentloaded", timeout=60000)
|
||
|
||
dur = None
|
||
for _ in range(40):
|
||
time.sleep(1.5)
|
||
st = pg.evaluate("""() => {
|
||
const v = Array.from(document.querySelectorAll('video')).filter(x=>x.duration>100)[0];
|
||
return v ? {dur: v.duration, ct: v.currentTime} : null;
|
||
}""")
|
||
if st and st["dur"] > 100:
|
||
dur = st["dur"]
|
||
break
|
||
if not dur:
|
||
print(" 播放器未就绪")
|
||
return False
|
||
|
||
# 拖到结尾前 4 秒（绕开 onSeek 守卫，等价于"拖到正好结尾"）
|
||
pg.evaluate("""() => {
|
||
const v = Array.from(document.querySelectorAll('video')).filter(x=>x.duration>100)[0];
|
||
v.muted = true;
|
||
v.currentTime = Math.max(0, v.duration - 4);
|
||
}""")
|
||
time.sleep(1)
|
||
|
||
btn = pg.query_selector(".j-bigplaybtn") or pg.query_selector(".j-playbtn")
|
||
if btn and btn.bounding_box():
|
||
bx = btn.bounding_box()
|
||
pg.mouse.click(bx["x"] + bx["width"] / 2, bx["y"] + bx["height"] / 2)
|
||
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
time.sleep(2)
|
||
if posted["done"]:
|
||
return True
|
||
st = pg.evaluate("""() => {
|
||
const v = Array.from(document.querySelectorAll('video')).filter(x=>x.duration>100)[0];
|
||
return v ? {ct: Math.round(v.currentTime), ended: v.ended, paused: v.paused} : null;
|
||
}""")
|
||
# 防挂机弹窗「是否继续观看视频?」
|
||
try:
|
||
ok = pg.query_selector(".el-message-box__btns .el-button--primary")
|
||
if ok:
|
||
ok.click()
|
||
except Exception:
|
||
pass
|
||
# 播放结束但还没上报 -> 再补一次播放
|
||
if st and st["ended"] and not posted["done"]:
|
||
btn = pg.query_selector(".j-bigplaybtn") or pg.query_selector(".j-playbtn")
|
||
if btn and btn.bounding_box():
|
||
bx = btn.bounding_box()
|
||
pg.mouse.click(bx["x"] + bx["width"] / 2, bx["y"] + bx["height"] / 2)
|
||
return posted["done"]
|
||
|
||
|
||
def run(train_id, only_missing=True):
|
||
sess = load_session() or capture_session()
|
||
s = api(sess)
|
||
if not check_auth(s):
|
||
print("会话失效，重新抓取…")
|
||
sess = capture_session()
|
||
s = api(sess)
|
||
if not check_auth(s):
|
||
raise SystemExit("认证失败：请先在浏览器登录 study.dgjy.net")
|
||
|
||
info, mooc, vids = video_periods(s, train_id)
|
||
print("=== 处理前核验 ===")
|
||
report(info, vids)
|
||
|
||
todo = [v for v in vids if not v["completeTime"]]
|
||
if not todo:
|
||
print("\n所有视频课时均已完成，无需处理。")
|
||
return
|
||
|
||
print("\n待完成 %d 个课时：" % len(todo))
|
||
for v in todo:
|
||
print(" %s %s (%ss)" % (v["periodId"], v["name"][:40], v["duration"]))
|
||
|
||
from playwright.sync_api import sync_playwright
|
||
with sync_playwright() as p:
|
||
b = p.chromium.connect_over_cdp(CDP)
|
||
ctx = b.contexts[0]
|
||
pg = ctx.new_page()
|
||
for i, v in enumerate(todo, 1):
|
||
print("\n[%d/%d] %s %s" % (i, len(todo), v["periodId"], v["name"][:40]))
|
||
try:
|
||
ok = complete_one(pg, train_id, train_id, v["periodId"])
|
||
except Exception as e:
|
||
print(" 异常:", e)
|
||
ok = False
|
||
print(" 上报 isComplete=1 :", "✅" if ok else "❌")
|
||
time.sleep(4)
|
||
try:
|
||
pg.close()
|
||
except Exception:
|
||
pass
|
||
|
||
print("\n=== 处理后核验（以平台数据为准）===")
|
||
time.sleep(3)
|
||
info2, _, vids2 = video_periods(s, train_id)
|
||
report(info2, vids2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 3:
|
||
print(__doc__)
|
||
sys.exit(1)
|
||
mode, tid = sys.argv[1], sys.argv[2]
|
||
if mode == "verify":
|
||
sess = load_session() or capture_session()
|
||
s = api(sess)
|
||
info, mooc, vids = video_periods(s, tid)
|
||
print("MOOC 课程ID:", mooc)
|
||
report(info, vids)
|
||
elif mode == "run":
|
||
run(tid)
|
||
else:
|
||
print(__doc__)
|