hanhan/study-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(platforms): 学习流程下沉到平台层，engine 只保留编排

Browse Source 
SessionExpired 移到 platforms/errors.py；旧版流程（_study_builtin）搬到 jspx.py，新版 SPA 流程（run_study_course + 视频）搬到 study.py；engine.py 只剩账号级编排，不再 import 具体平台；routes/quiz 去掉平台 star import，session/dump_cookies 缺失时明确报错。顺带修复学习链路三处未定义全局名（jspx/study/quiz）。
This commit is contained in:
study-tool
2026-10-31 14:21:36 +08:00
parent d702830470
commit 7259ab8d96
13 changed files with 700 additions and 647 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
@@ -14,6 +14,8 @@
- 平台拆成插件包 `src/study_tool/platforms/`：`base.py` 定义 `Platform` 契约、`registry.py` 注册表；引擎按 `course["platform"]` 从注册表分发
- HTTP 接口收进 `routes/`（accounts / misc / logs），`docs/architecture.md` 说明分层、平台识别与新增平台的步骤
- 散落的 JS 脚本统一收进 `script/`：运行必需的 `des.js` / `desenc.js` 放 `script/`（`config.py` 的 `DESENC` 指向 `script/desenc.js`），不参与运行的 `dgjy-toolkit.user.js`、`study-axios.mjs` 移到 `script/reference/` 作参考
- 平台契约第一步：新增 `platforms/contract.py` 定义生命周期 ↔ 槽位 ↔ 能力名映射，`Platform.capabilities()` 按已实现槽位推导，`GET /api/platforms` 返回 `capabilities`
- 平台契约第二步：`SessionExpired` 移到 `platforms/errors.py`；旧版 / 新版学习流程分别从 `engine.py` 搬到 `platforms/jspx.py` / `platforms/study.py`，`engine.py` 只剩账号级编排（115 行）；`routes/*`、`quiz.py` 去掉平台 star import，`session` / `dump_cookies` 缺失时明确报错
### 新增
@@ -33,6 +35,7 @@
- 服务重启把上次没跑完的运行标成「已中断」后，那句话会被后续 `set_state` 一起写回库，新一轮运行一起步就顶着「服务重启，上次运行已中断」，库里还留着 `status=studying` 的幽灵记录：`runtime.set_state()` 在状态切入 `discovering` / `studying` 时清掉上一轮的 `error` / `progress` / `finished_at`，`hydrate_runs()` 标记后立即落库
- 服务重启后卡片「课程日志」显示「暂无日志」，SQLite 里的历史读不出来：前端兜底条件原为「内存为空且 status 为空」，而重启后 status 恰好是 `interrupted`（非空），历史永远不读；现在只看内存是否为空
- 删除账号只清了按账号 id 记的 `run_state` / `run_log`，登录日志（key 与 `login_log.username` 记的是用户名 / 手机号）残留：`account_delete()` 改为按账号 id + 用户名一起清，`DELETE /api/accounts/{id}` 同时清掉内存里两个 key 的日志
- 学习链路上三处引用未定义的全局名（`platforms/jspx.py` 的 `extract_study_token` / `StudyClient` / `save_snapshot` 等、`platforms/study.py` 的 `parse_qs`、`quiz.py` 的 `BeautifulSoup` / `urlsplit`）会让课程发现 / 答题直接 `NameError`：`parse_qs`、`BeautifulSoup`、`urlsplit` 改为模块顶部导入，`jspx` ↔ `study` 之间改用函数内延迟导入（模块级会成环）
- 服务重启留下的「已中断」状态在重新登录后仍然压在卡片上，徽标显示「已中断」并挂着旧进度和红条：登录成功后清掉该账号上一轮已结束的运行状态（内存 + SQLite），正在跑（`discovering` / `studying`）的不动，内存日志保留
### 其他
docs/OPTIMIZATION.md
+17
View File
@@ -2,6 +2,23 @@
已实现、但还需要你实测确认的事项。确认没问题后把对应条目删掉。
## 平台流程下沉（契约第二步）
**状态**：已实现，待实测确认。
### 做了什么
- `SessionExpired` 移到 `platforms/errors.py`，`config.py` 继续导出；平台实现不再反向 import 引擎。
- 旧版学习流程（反馈表 / 测验 / 刷课件）从 `engine.py` 搬到 `platforms/jspx.py`；新版 SPA 流程（含真播视频）搬到 `platforms/study.py`。
- `engine.py` 只剩账号级编排（课程循环 / 会话续期 / 按平台分发，115 行），不再 import 任何具体平台模块。
- `routes/*`、`quiz.py` 去掉对 `platforms.jspx` / `platforms.study` 的 star import；平台没实现 `session` / `dump_cookies` 时明确报错，不再退回莞易学的默认实现（原来会静默按 dgjy 过滤 cookie）。
### 怎么测
1. 莞易学：刷新课程 → 发现课程 → 开始学习（快速、常规各跑一次），反馈表 / 测验照常提交。
2. 广东公需课：开始学习，确认进度照常上报。
3. 错误分支：平台缺少 `session` 时，`/api/courses`、`/api/discover` 应返回「还没有实现 session」。
## 添加账号选择平台
**状态**：已实现，待实测确认。
docs/architecture.md
+13 -7
View File
@@ -17,15 +17,18 @@ src/study_tool/
settings_store.py 系统设置读写
llm.py 大模型调用（OpenAI 兼容）
quiz.py 测验解析 / 交卷 / AI 答题（含 /api/quiz/*）
engine.py 学习引擎：登录→发现→按平台分发→worker / 视频 / 提交
engine.py 账号级编排：课程循环、会话续期、按 platform 分发
capture.py 抓包
platforms/ 平台插件（见下）
base.py Platform 契约（id / label / study / entries）
base.py Platform 契约（id / label / 槽位 / capabilities）
contract.py 生命周期 ↔ 槽位 ↔ 能力名映射
errors.py SessionExpired（平台层错误类型）
registry.py 注册表：register / get / detect / all
detect.py 页面代际识别（legacy 旧版 / spa 新版）
guanyixue.py 莞易学平台：合并 jspx 登录入口与 study 学习入口
jspx.py 莞易学旧版页面实现（会话 / 登录 / 课程抓取 / 旧版学习）
study.py 莞易学新版 SPA 实现（StudyClient / 接口学习）
jspx.py 莞易学旧版：会话 / 登录 / 课程抓取 / 旧版学习流程
study.py 莞易学新版 SPA：StudyClient / 接口学习 / 视频流程
gdgxk.py 广东公需课：SSO 换票 + 视频进度心跳
routes/ HTTP 接口
accounts.py 账号 / 登录 / 登录记录
misc.py 课程 / 发现 / 学习 / 停止 / 抓包 / 快照 / 平台列表
@@ -40,11 +43,14 @@ script/reference/ 不参与运行的参考脚本：dgjy-toolkit.user.js
- **入口层**：`app.py`（根）+ `study_tool/app.py`。前者只负责 `sys.path` 与启动，后者组装应用。
- **接口层**：`routes/*` 与 `quiz.py` 里的 `/api/quiz/*`，只做参数校验、调用引擎、返回结果。
- **领域层**：`engine.py`（学习流程编排）、`platforms/*`（平台实现）、`llm.py` / `quiz.py` 的解析逻辑。
- **领域层**：`engine.py`（账号级编排：课程循环 / 会话续期 / 按平台分发）、`platforms/*`（平台实现，含各自的学习流程）、`llm.py` / `quiz.py` 的解析逻辑。
- **基础设施**：`store.py`（SQLite）、`runtime.py`（状态 + 日志）、`config.py`。
依赖方向单向：接口 → 领域 → 基础设施。基础设施不反向 import 接口。
`engine.py` 与 `routes/*` 只用 `registry.get(...)` 取平台实现，不 import 具体平台模块；
平台实现之间只有同平台不同代际会互相 import（`study.py` → `jspx.py`），且用函数内延迟导入避免成环。
## 平台插件与识别
平台实现在 `platforms/`，通过注册表注入，`engine` 只按 `course["platform"]` 分发：
@@ -104,9 +110,9 @@ quiz_key / cookies / plan / project / name`。插件可以只取自己需要的
`id=guanyixue`、`label=莞易学`。两代课程页：
- `platforms/jspx.py`：旧版页面（`tier-tree-one` / `tier3`），负责会话、登录、课程抓取，
学习流程在 `engine._study_builtin`。
以及旧版学习流程（`_study_builtin`：反馈表 + 测验 + 刷课件）。
- `platforms/study.py`：新版 Vue SPA，`StudyClient` + `/train/course/front/*` 接口，
学习流程在 `engine.run_study_course`。
以及新版学习流程（`run_study_course`：接口上报 / 真播视频）。
- `platforms/guanyixue.py`：注册平台并按 `course["generation"]` 分发；
`platforms/detect.py` 只判断页面代际，不判断平台归属。
docs/progress/todo.md
-2
View File
@@ -8,7 +8,5 @@
- 广东公需课：用一门**未学完**的课实测进度上报，确认平台端「已学时长 / 已完成」状态跟着变。
- 广东公需课：**试卷考核**未接入 —— 平台判定通过要「所有教学活动 100% + 试卷考核 ≥60 分」，所以视频报满后课程仍显示「未完成」。
- 广东公需课：视频弹题未接入。
- 平台契约第二步：`SessionExpired` 下沉到 `platforms/`、通用学习流程从 `engine.py` 搬到平台层公共模块，
去掉 `routes/*` 对 `platforms.jspx` 的 star import（新平台漏写 `session` 时会静默按 dgjy 过滤 cookie）。
- 平台契约第三步：把答题下沉为平台能力（`quiz`），`quiz.py` 只留 LLM 与通用交卷；前端按 `/api/platforms`
返回的 `capabilities` 显示答题 / 反馈按钮。
src/study_tool/config.py
+2 -2
View File
@@ -69,5 +69,5 @@ PLAN_STATUS = [
STUDYABLE_STATUS = {"TAKING_PLAN", "NO_BEGAIN_PLAN"}
class SessionExpired(RuntimeError):
"""站点要求重新登录（cookie 已失效）。"""
# 平台层错误类型；老代码 `from study_tool.config import *` 仍然拿得到 SessionExpired
from study_tool.platforms.errors import SessionExpired # noqa: E402
src/study_tool/engine.py
+7 -618
View File
@@ -1,121 +1,20 @@
"""学习引擎：运行状态、课程 worker、视频播放、反馈 / 测验提交。
"""学习引擎：账号级编排（课程循环、会话续期、按平台分发）。
平台自己的学习流程在 platforms/ 里（旧版 jspx / 新版 study），引擎不认识具体站点。
日志统一走 runtime.log()。
"""
import json
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from study_tool import store
from study_tool.config import *
from study_tool.platforms import registry
from study_tool.platforms.jspx import *
from study_tool.platforms.study import *
from study_tool.platforms.errors import SessionExpired
from study_tool.runtime import RUNS, RUNS_LOCK, log, set_state
# ------------------------------------------------------------------ study engine
class CwItem:
def __init__(self, d, cookies):
self.title = d["title"]
href = d["href"]
self.url = href if href.startswith("http") else JSPX + href
self.s = new_session(cookies)
self.rec = None
self.status = "done" if d.get("done") else "pending"
self.last = None
def resolve(self):
try:
r = self.s.get(self.url, timeout=40)
except Exception as e:
return "exc:%s" % e
m = re.search(r"'id':'([0-9A-F]{32})'", r.text)
if not m:
return "no-id"
self.rec = m.group(1)
return "ok"
def start(self):
try:
r = self.s.post(EP_STUDY, data={"id": self.rec, "order": "server_start",
"updatePassPerc": "true"}, timeout=40)
return r.text.strip()[:40]
except Exception as e:
return "EXC:%s" % e
def beat(self):
try:
r = self.s.post(EP_STUDY, data={"id": self.rec, "order": "server_time",
"updatePassPerc": "true",
"videoSchedule": 30}, timeout=40)
return r.text.strip()[:60]
except Exception as e:
return "EXC:%s" % e
def submit_feedback(s, href, key):
url = href if href.startswith("http") else JSPX + href
r = s.get(url, timeout=40)
soup = BeautifulSoup(r.text, "lxml")
fields = {i.get("name"): i.get("value") for i in soup.select("form input[type=hidden]") if i.get("name")}
if fields.get("allowStudy") == "false":
log(key, "反馈表：allowStudy=false，跳过")
return
r2 = s.post(JSPX + "/train/trainee/doFeedbackList.action", data=fields, timeout=40)
qs = BeautifulSoup(r2.text, "lxml")
answers = []
for it in qs.select("div.job-item"):
qtype = it.get("questype")
opts = [sp.get("data-name") for sp in it.select(".job-item-a span")]
ans = ("课程内容充实，讲师讲解清晰，对我的教学实践很有帮助，希望今后能增加更多一线教学案例。"
if qtype == "04" else (opts[0] if opts else "A"))
answers.append({"id": it.get("traineeid"), "quesId": it.get("qumid"),
"quesType": qtype, "quesNo": it.get("quesno"), "answer": ans})
payload = {"planId": fields.get("planId"), "objectId": fields.get("objectId"),
"answers": json.dumps(answers, ensure_ascii=False, separators=(",", ":")),
"submitStatus": "finish"}
rr = s.post(JSPX + "/train/trainee/saveFeedback.action", data=payload, timeout=40)
log(key, "反馈表：%d 题，%s" % (len(answers), "提交成功" if "MSG" not in rr.text else "返回 " + rr.text[:80]))
def submit_quiz(s, href, key, answer_key=None):
url = href if href.startswith("http") else JSPX + href
plan, _ = parse_ids(url)
m = re.search(r"objectId=([0-9A-Fa-f]+)", url)
obj = m.group(1) if m else ""
rr = s.get("%s/train/trainee/showTestQums.action?planId=%s&objectId=%s" % (JSPX, plan, obj), timeout=40)
qs = BeautifulSoup(rr.text, "lxml")
singles = (answer_key or {}).get("singles", {})
multis = (answer_key or {}).get("multis", {})
answers = []
for it in qs.select("div.job-item"):
t, qno = it.get("questype"), it.get("quesno")
opts = [sp.get("data-name") for sp in it.select(".job-item-a span")]
if t == "02":
ans = (multis.get(qno) or ("#".join(opts) if opts else "A"))
else:
ans = singles.get(qno) or (opts[0] if opts else "A")
answers.append({"id": it.get("traineeid"), "quesId": it.get("qumid"),
"quesType": t, "quesNo": qno, "answer": ans})
if not answers:
log(key, "测验：未取到题目，跳过")
return
payload = {"planId": plan, "objectId": obj,
"answers": json.dumps(answers, ensure_ascii=False, separators=(",", ":"))}
r3 = s.post(JSPX + "/train/trainee/endTest.action", data=payload, timeout=40)
r4 = s.get("%s/train/trainee/doTest.action?planId=%s&objectId=%s" % (JSPX, plan, obj), timeout=40)
sc = re.search(r"得分：\s*(\d+)\s*分",
re.sub(r"\s+", " ", BeautifulSoup(r4.text, "lxml").get_text(" ", strip=True)))
log(key, "测验：%d 题，%s，得分 %s" %
(len(answers), "提交成功" if "MSG" not in r3.text else r3.text[:60],
sc.group(1) if sc else "?"))
def study_worker(key, username, cookies, courses, workers, cap, quiz_key,
enable_quiz=False, enable_feedback=True, password="",
mode="fast", parallel=2):
@@ -164,7 +63,9 @@ def run_one_course(key, username, cookies, plan, project, workers, cap, quiz_key
adapter = registry.get(account.get("platform"))
if adapter is None or adapter.discover is None:
raise RuntimeError("平台 %s 没有可用的课程发现实现" % (account.get("platform") or "?"))
s = (adapter.session or new_session)(cookies)
if adapter.session is None:
raise RuntimeError("平台 %s 没有实现 session" % (account.get("platform") or "?"))
s = adapter.session(cookies)
try:
d = adapter.discover(s, plan, project, name)
except SessionExpired:
@@ -185,7 +86,7 @@ def run_one_course(key, username, cookies, plan, project, workers, cap, quiz_key
d = adapter.discover(s, plan, project, name)
# 新平台可能刚在 discover 里换到 study.dgjy.net 的会话（Middle-Token），存回去给后续课程用
set_state(key, course=d["course"], progress=d["progress"], user=d["user"],
cookies=(adapter.dump_cookies or dump_cookies)(s))
cookies=(adapter.dump_cookies(s) if adapter.dump_cookies else []))
log(key, "课程：%s | 进度 %s | %s" % (d["course"], d["progress"], d["info"]))
pf = registry.get(d.get("platform"))
gen = d.get("generation")
@@ -210,515 +111,3 @@ def run_one_course(key, username, cookies, plan, project, workers, cap, quiz_key
quiz_key=quiz_key, cookies=cookies, plan=plan, project=project, name=name)
def _study_builtin(key, s, d, workers=1, cap=30, enable_quiz=False, enable_feedback=False,
mode="fast", parallel=1, quiz_key="", cookies=None, plan="", project="",
name=""):
"""旧版 jspx 的内置流程：反馈表 + 测验 + 刷课件。"""
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
log(key, "反馈表已完成，跳过")
continue
try:
submit_feedback(s, fb["href"], key)
except Exception as e:
log(key, "反馈表异常：%s" % e)
else:
for fb in d["feedback"]:
log(key, "反馈表：已按设置跳过（%s）" % ("已完成" if fb["done"] else "未完成"))
if enable_quiz:
for t in d["test"]:
if t["done"]:
log(key, "测验已完成，跳过")
continue
try:
submit_quiz(s, t["href"], key, quiz_key)
except Exception as e:
log(key, "测验异常：%s" % e)
else:
for t in d["test"]:
if t["done"]:
log(key, "测验：已完成，跳过")
else:
log(key, "测验：⚠ 已按设置跳过答题（避免因正确率不足卡住「完成」），"
"需要时请在页面上手动作答，或打开「自动交测验」开关")
objs = [CwItem(c, cookies) for c in d["courseware"]]
todo = [o for o in objs if o.status != "done"]
done0 = len(objs) - len(todo)
set_state(key, status="studying", total=len(objs), done=done0)
if todo:
log(key, "解析 %d 个学习记录…" % len(todo))
with ThreadPoolExecutor(max_workers=min(workers, len(todo))) as ex:
list(ex.map(lambda x: x.resolve(), todo))
live = [o for o in todo if o.rec]
log(key, "已解析 %d/%d，并发启动会话" % (len(live), len(todo)))
with ThreadPoolExecutor(max_workers=min(workers, len(live))) as ex:
list(ex.map(lambda x: x.start(), live))
t0 = time.time()
while time.time() - t0 < cap * 60:
with RUNS_LOCK:
if (RUNS.get(key) or {}).get("status") == "stopped":
log(key, "已按要求停止")
return
pending = [o for o in live if o.status != "done"]
if not pending:
break
with ThreadPoolExecutor(max_workers=min(workers, len(pending))) as ex:
resps = list(ex.map(lambda x: x.beat(), pending))
newly = []
for o, resp in zip(pending, resps):
o.last = resp
if resp == "study_over":
o.status = "done"
newly.append(o)
el = int(time.time() - t0)
dc = len([o for o in live if o.status == "done"]) + done0
set_state(key, done=dc, elapsed=el)
log(key, "第 %d 轮 %dm%02ds 完成 %d/%d" % (el // 60 + 1, el // 60, el % 60, dc, len(objs)))
for o in newly:
log(key, " ✓ %s" % o.title[:40])
time.sleep(60)
else:
log(key, "所有课件均已完成")
try:
d2 = discover(s, plan, project, name)
set_state(key, progress=d2["progress"], info=d2["info"], cookies=dump_cookies(s))
log(key, "课程结束：%s | %s" % (d2["progress"], d2["info"]))
except Exception as e:
log(key, "复查失败：%s" % e)
def run_study_course(key, s, d, workers, cap, enable_quiz, enable_feedback,
mode="fast", parallel=2):
"""新版平台（study.dgjy.net / Vue SPA）课程：视频走 study/log 接口，
反馈表走 evaluation/result，测验保持手动作答。
mode=fast 只调接口上报（直链 mp4 仍退回真播），速度最快
mode=normal 开无头浏览器真播视频，parallel 控制同时播几个
"""
cookies = dump_cookies(s)
page_url = d.get("page_url") or ""
train_id = d.get("train_id") or ""
client = StudyClient(s, page_url)
# ---- 先确认已加入培训 ----
# 没加入时 study/log 仍然回 200，但服务端不记账：逐个课时重试只会刷一屏「上报未生效」。
if d.get("need_join"):
joined, why = ensure_study_join(s, client, d)
if joined:
log(key, "加入培训：%s" % why)
d["is_join"] = True
else:
log(key, "加入培训被拒：%s" % (why or "未知原因"))
probe = next((c for c in d["courseware"] if not c["done"]), None)
if probe:
good, last = study_complete_period(cookies, page_url, probe)
log(key, "试报 1 个课时：%s" % ("生效" if good else "未生效（%s）" % last))
if not good:
log(key, "该账号对这门课没有写入权限，本次跳过。"
"需要课程管理员把这门课分配到账号下（jspx 在学 ≠ 新版平台已加入）")
return
else:
return
# ---- 反馈表 ----
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
log(key, "反馈表已完成，跳过")
continue
try:
study_submit_feedback(client, fb, key, train_id)
except Exception as e:
log(key, "反馈表异常：%s" % e)
else:
for fb in d["feedback"]:
log(key, "反馈表：已按设置跳过（%s）" % ("已完成" if fb["done"] else "未完成"))
# ---- 测验 ----
for t in d["test"]:
score, passed = t.get("score"), t.get("passed_score")
if t["done"]:
if score is not None and passed is not None:
try:
if float(score) < float(passed):
log(key, "测验：已作答但不及格（%s/%s），需要在页面上重考" % (score, passed))
continue
except Exception:
pass
log(key, "测验：已完成，跳过")
elif enable_quiz:
log(key, "测验：新版平台不代答（%s 需在页面上手动作答）" % t["title"][:30])
else:
log(key, "测验：⚠ 已按设置跳过（%s），需要时请在页面上手动作答" % t["title"][:30])
# ---- 视频课时 ----
objs = d["courseware"]
todo = [c for c in objs if not c["done"]]
done0 = len(objs) - len(todo)
set_state(key, status="studying", total=len(objs), done=done0)
if not objs:
log(key, "没有视频课时")
return
kinds = {}
for c in objs:
kinds[c.get("kind") or "?"] = kinds.get(c.get("kind") or "?", 0) + 1
if kinds:
log(key, "视频类型：%s" % "、".join("%s %d 个" % (k, v) for k, v in kinds.items()))
if not todo:
log(key, "所有课件均已完成")
all_done = True
elif mode == "normal":
got = study_videos_normal(key, cookies, page_url, train_id, todo, parallel, cap, done0)
all_done = got >= len(objs)
set_state(key, done=got)
log(key, "视频课时：%d/%d 已完成" % (got, len(objs)))
else:
got = study_videos_fast(key, cookies, page_url, train_id, todo, workers, cap, done0)
all_done = got >= len(objs)
set_state(key, done=got)
log(key, "视频课时：%d/%d 已完成" % (got, len(objs)))
# ---- 课程容器（periodType 10/13）：子课时学完不会自动完成 ----
if all_done and d.get("containers"):
for ct in d["containers"]:
if ct.get("done"):
continue
try: # 先重算子课程计数：并发上报会丢增量，计数不到容器就不完成
study_cal_final_score(client, ct["kcdm"])
log(key, " 已重算子课程完成计数（kcdm=%s）" % ct["kcdm"][:8])
except Exception as e:
log(key, " 重算子课程完成计数失败：%s" % str(e)[:80])
ok, msg = study_complete_container(client, ct)
log(key, " %s 课程节点「%s」：%s" % ("✓" if ok else "×", ct["name"][:24], msg))
try:
study_cal_final_score(client, train_id)
except Exception:
pass
try:
_reread_study(key, s, d)
except Exception as e:
log(key, "复查失败：%s" % e)
# 平台靠 document.hidden / visibilitychange 判断「用户切走了」，一 hidden 就暂停并记异常。
# 多路真播时把可见性钉死成 visible，不然后台标签页会被判成离开。
VIS_KEEP_JS = """
(() => {
const d = document;
try { Object.defineProperty(d, 'hidden', {get: () => false, configurable: true}); } catch (e) {}
try { Object.defineProperty(d, 'webkitHidden', {get: () => false, configurable: true}); } catch (e) {}
try { Object.defineProperty(d, 'visibilityState', {get: () => 'visible', configurable: true}); } catch (e) {}
d.addEventListener('visibilitychange', e => e.stopImmediatePropagation(), true);
})();
"""
def _classhour_of(url):
m = re.search(r"[?&]classhour=([0-9]+)", url or "")
return m.group(1) if m else ""
def _wait_video(pg, want_id, secs=60):
"""等播放页就绪：("ok", (frame, duration)) / ("redirect", 被切到的课时) / ("timeout", "")。"""
end = time.time() + secs
while time.time() < end:
cur = _classhour_of(pg.url)
if cur and want_id and cur != str(want_id):
return "redirect", cur
for fr in pg.frames:
try:
st = fr.evaluate("""() => {
const v = Array.from(document.querySelectorAll('video')).filter(x => x.duration > 30)[0];
return v ? {d: v.duration} : null; }""")
except Exception:
continue
if st and st.get("d"):
return "ok", (fr, float(st["d"]))
pg.wait_for_timeout(1000)
return "timeout", ""
def _kick_play(frame, seek):
"""保活：静音、1 倍速、必要时 seek 到结尾前 3 秒（seek=False 就是老老实实从头播）。"""
try:
frame.evaluate("""(seek) => {
const v = Array.from(document.querySelectorAll('video')).filter(x => x.duration > 30)[0];
if (!v) return;
v.muted = true;
try { v.playbackRate = 1; } catch (e) {}
if (seek && v.currentTime < v.duration - 5) {
try { v.currentTime = Math.max(0, v.duration - 3); } catch (e) {}
}
if (v.paused) v.play().catch(() => {});
}""", seek)
except Exception:
pass
def _dialog_text(pg):
try:
el = pg.query_selector(".el-message-box__message")
return (el.inner_text() or "").strip() if el else ""
except Exception:
return ""
def play_one_period(key, p, cookies, train_id, item, timeout=900, seek=False, attach_cdp=False):
"""真播一个课时到结尾。返回 (ok, 说明, 平台要求先学的课时 id)。
平台在进入播放页时会先查「是不是轮到这一节」，不按顺序会被弹回并跳到它指定的课时，
这时返回的第三个值就是那一节，调用方据此重排队列。
"""
pid = str(item["period_id"])
url = ("%s/training/courseVideo?classhour=%s&isJoin=1&needJoin=1&courseId=%s"
"&isFront=1&trainCourseId=%s" % (STUDY, pid, train_id, train_id))
posted = {"done": False}
browser, own = None, False
try:
if attach_cdp:
try:
browser = p.chromium.connect_over_cdp("http://127.0.0.1:9222")
except Exception:
browser = None
if browser is None:
browser = p.chromium.launch(headless=True, args=[
"--autoplay-policy=no-user-gesture-required", "--mute-audio",
"--disable-background-timer-throttling", "--disable-renderer-backgrounding"])
own = True
ctx = browser.new_context()
else:
ctx = browser.contexts[0] if browser.contexts else browser.new_context()
for c in cookies:
if "dgjy" in (c.get("domain") or ""):
try:
ctx.add_cookies([{"name": c["name"], "value": c.get("value") or "",
"domain": (c.get("domain") or "").lstrip("."),
"path": c.get("path") or "/"}])
except Exception:
pass
pg = ctx.new_page()
pg.add_init_script(VIS_KEEP_JS)
def on_req(r):
try:
if "period/study/log" in r.url and '"isComplete":1' in (r.post_data or "").replace(" ", ""):
posted["done"] = True
except Exception:
pass
pg.on("request", on_req)
pg.goto(url, wait_until="domcontentloaded", timeout=60000)
status, val = _wait_video(pg, pid, 60)
if status == "redirect":
return False, "平台按顺序放行，页面被切到课时 %s" % val, str(val)
if status != "ok":
return False, "没等到 <video>（可能需要专用播放器的课时）", ""
frame = val[0]
t0 = time.time()
while time.time() - t0 < timeout:
if posted["done"]:
return True, "", ""
msg = _dialog_text(pg)
if "顺序" in msg or "未完成课时" in msg:
return False, "平台按顺序放行：%s" % msg[:30], _classhour_of(pg.url)
_kick_play(frame, seek)
pg.wait_for_timeout(2000)
return False, "超时未等到播放结束上报", ""
except Exception as e:
return False, str(e)[:120], ""
finally:
if own and browser is not None:
try:
browser.close()
except Exception:
pass
def study_videos_fast(key, cookies, page_url, train_id, todo, workers, cap, done0=0):
"""快速学习：直接调 study/log 上报；直链 mp4 接口不认，退回真播。返回完成总数。"""
deadline = time.time() + cap * 60
hard = [c for c in todo if c.get("kind") != "direct"]
soft = [c for c in todo if c.get("kind") == "direct"]
ok = done0
if hard:
log(key, "并发提交 %d 个视频课时（workers=%d）…" % (len(hard), min(workers, len(hard))))
ex = ThreadPoolExecutor(max_workers=min(workers, len(hard)))
futs = {ex.submit(study_complete_period, cookies, page_url, it): it for it in hard}
try:
for fut in as_completed(futs, timeout=max(1, deadline - time.time())):
it = futs[fut]
try:
good, msg = fut.result()
except Exception as e:
good, msg = False, str(e)[:80]
if good:
ok += 1
log(key, " ✓ %s（%ss）" % (it["title"][:40], it.get("duration") or 0))
else:
log(key, " × %s 上报未生效：%s" % (it["title"][:40], msg))
set_state(key, done=ok)
except TimeoutError:
log(key, "已到时间上限（%d 分钟），剩余课时留待下次" % cap)
except Exception as e:
log(key, "批量提交中断：%s" % e)
finally:
ex.shutdown(wait=False, cancel_futures=True)
if soft:
log(key, "另有 %d 个直链视频课时需要真播放到结尾（%s）"
% (len(soft), "、".join(c["title"][:16] for c in soft[:4])))
for it in soft:
if (RUNS.get(key) or {}).get("status") == "stopped":
break
try:
good = play_period_to_end(key, cookies, page_url,
train_id, it["period_id"], it["title"])
except Exception as e:
log(key, " × %s 播放失败：%s" % (it["title"][:30], str(e)[:100]))
continue
if good:
ok += 1
log(key, " ✓ %s（播放到结尾）" % it["title"][:40])
else:
log(key, " × %s 未确认完成" % it["title"][:40])
set_state(key, done=ok)
return ok
def _play_wave(key, cookies, train_id, wave, deadline, seek):
"""一波并发：每路一个独立 playwright + 独立浏览器，互不遮挡。"""
def run(it):
try:
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
return play_one_period(key, p, cookies, train_id, it,
timeout=max(60, deadline - time.time()), seek=seek)
except Exception as e:
return False, str(e)[:100], ""
out = []
ex = ThreadPoolExecutor(max_workers=len(wave))
try:
futs = {ex.submit(run, it): it for it in wave}
for fut in as_completed(futs):
try:
ok, why, wanted = fut.result()
except Exception as e:
ok, why, wanted = False, str(e)[:100], ""
out.append((futs[fut], ok, why, wanted))
finally:
ex.shutdown(wait=False)
return out
def study_videos_normal(key, cookies, page_url, train_id, todo, parallel, cap, done0=0):
"""常规学习：开无头浏览器真播视频，parallel 路并发。
平台可能按课节顺序放行（不在顺序上会被弹回），所以先按并发跑一波；
一旦被弹回就把并发降到 1，并把平台指定的那一节挪到队首，按顺序补回去。
"""
try:
from playwright.sync_api import sync_playwright # noqa: F401
except Exception as e:
log(key, "缺少 playwright，常规学习不可用：%s" % e)
return done0
parallel = max(1, min(int(parallel or 1), 5, len(todo)))
log(key, "常规学习：%d 个课时真播，%d 路并发" % (len(todo), parallel))
deadline = time.time() + cap * 60
ok, pending, order_only, bounced = done0, list(todo), False, {}
while pending and time.time() < deadline:
if (RUNS.get(key) or {}).get("status") == "stopped":
log(key, "已按要求停止")
break
wave, pending = pending[:parallel], pending[parallel:]
log(key, "本轮 %d 路：%s" % (len(wave), "、".join(x["title"][:14] for x in wave)))
for it, good, why, wanted in _play_wave(key, cookies, train_id, wave, deadline, seek=False):
if good:
ok += 1
log(key, " ✓ %s（真播到结尾）" % it["title"][:40])
else:
log(key, " × %s：%s" % (it["title"][:30], why))
if wanted:
order_only = True
n = bounced[it["period_id"]] = bounced.get(it["period_id"], 0) + 1
if n >= 3: # 一直被顺序校验弹回，别再空转
log(key, " × %s：连续被顺序校验弹回，跳过" % it["title"][:30])
else:
i = next((i for i, x in enumerate(pending)
if str(x["period_id"]) == str(wanted)), None)
if i is None: # 平台要的那一节正在播或不在这门课里，先排到队尾
pending.append(it)
else: # 先播平台要的那一节，再回来播这一个
pending.insert(0, pending.pop(i))
set_state(key, done=ok)
if order_only and parallel > 1:
parallel = 1
log(key, "检测到平台按顺序放行，并发降为 1（剩下的逐个播）")
return ok
def play_period_to_end(key, cookies, page_url, train_id, period_id, title, timeout=900):
"""直链视频（fileType=other）：平台只认 onPlayEnd 上报的 isComplete=1。
做法：Playwright 打开课程播放页（带登录 cookie）→ 秒跳到 duration-3 秒 → 播放 →
等接口出现 isComplete=1 → 回读 completeTime 复核。
"""
try:
from playwright.sync_api import sync_playwright
except Exception as e:
log(key, " 缺少 playwright，直链视频无法自动播完：%s" % e)
return False
item = {"period_id": period_id, "title": title}
try:
with sync_playwright() as p:
ok, why, _ = play_one_period(key, p, cookies, train_id, item,
timeout=timeout, seek=True, attach_cdp=True)
except Exception as e:
log(key, " 播放异常：%s" % str(e)[:120])
return False
if not ok:
log(key, " 未播完：%s" % why)
return False
time.sleep(2)
try:
c = StudyClient(new_session(cookies), page_url)
return study_is_done(c, period_id)
except Exception:
return True
def study_sync_gyx(client, train_id):
"""培训里的「课程」是 periodType 10/13 的容器节点，不跟着里面的课时自动完成。
这条同步接口要课程管理员权限：学生账号会回「无权操作该课程」，调一次没坏处，
管理员账号跑的时候能直接把容器刷成完成。
"""
j = client.call("GET", "/course/front/study/sync/gyxCourseData/%s" % train_id)
if j.get("code") == 200:
return "已同步"
return "跳过（%s）" % str(j.get("msg") or ("code=%s" % j.get("code")))[:40]
def _reread_study(key, s, d):
"""课程跑完后按 trainCourseId 直接复查（不重新解析 jspx 页面）。"""
client = StudyClient(s, d.get("page_url") or "")
try:
log(key, "同步莞易学课程完成状态：%s" % study_sync_gyx(client, d["train_id"]))
except Exception as e:
log(key, "同步莞易学课程完成状态失败：%s" % str(e)[:100])
try:
d2 = discover_study(client, d["train_id"], d["course"])
except SessionExpired:
client.bootstrap()
d2 = discover_study(client, d["train_id"], d["course"])
set_state(key, progress=d2["progress"], info=d2["info"], cookies=dump_cookies(s))
log(key, "课程结束：%s | %s" % (d2["progress"], d2["info"]))
src/study_tool/platforms/errors.py
+9
View File
@@ -0,0 +1,9 @@
"""平台层错误类型。
放在这里是为了让平台实现（jspx / study / gdgxk）不必反向 import 引擎；
`config.py` 再导出一次，老的 `from study_tool.config import *` 仍然可用。
"""
class SessionExpired(RuntimeError):
"""站点要求重新登录（cookie 已失效）。"""
src/study_tool/platforms/gdgxk.py
+2 -2
View File
@@ -19,6 +19,7 @@ import requests
from study_tool import runtime, store
from study_tool.platforms import registry
from study_tool.platforms.base import Platform
from study_tool.platforms.errors import SessionExpired
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/153.0.0.0 Safari/537.36")
@@ -35,8 +36,7 @@ def _xhr(s, method, url, **kw):
def _expired(msg):
"""引擎认的是同一个 SessionExpired，这里延迟取避免循环 import。"""
from study_tool.engine import SessionExpired
"""统一错误类型：引擎据此触发续期 / 重新登录。"""
return SessionExpired(msg)
src/study_tool/platforms/jspx.py
+192 -2
View File
@@ -5,6 +5,8 @@
import json
import re
import subprocess
import time
from concurrent.futures import ThreadPoolExecutor
from html import unescape as html_unescape
from urllib.parse import unquote, urljoin, urlsplit
@@ -14,6 +16,7 @@ from bs4 import BeautifulSoup
from study_tool.config import *
from study_tool.platforms import registry
from study_tool.platforms.detect import detect_generation, has_study_marker
from study_tool.runtime import RUNS, RUNS_LOCK, log, set_state
# ------------------------------------------------------------------------ crypto
@@ -298,6 +301,7 @@ def ensure_study_session(s, force=False):
-> {"result":{"url":".../training","token":"<jwt>"}} # 这才是 Middle-Token
成功标准只有一个：cookie 里出现 study.dgjy.net 的 Middle-Token。
"""
from study_tool.platforms.study import _study_token
if force:
for c in list(s.cookies):
if c.name == "Middle-Token":
@@ -365,6 +369,7 @@ def _fetch_train_token(s):
def _exchange_code(s, code):
"""code -> Middle-Token。前端是 POST /mod/gjt/login/<code>，再取 gyx 培训 token。"""
from study_tool.platforms.study import _study_token
paths = ("/mod/gjt/login/%s", "/train/mod/gjt/login/%s")
for path in paths:
try:
@@ -551,6 +556,10 @@ def discover(s, plan, project, name=""):
def _discover_once(s, plan, project, name=""):
# study.py 反向 import 本模块，模块级 import 会成环，用到时再取
from study_tool.platforms.study import (StudyClient, discover_study, extract_study_token,
extract_train_id, save_snapshot,
study_resolve_train_id)
url = "%s/train/trainee/courseStudyMain.action?planId=%s&projectId=%s&trainRole=0" % (JSPX, plan, project)
r = s.get(url, timeout=40)
html = r.text or ""
@@ -627,9 +636,190 @@ def _discover_once(s, plan, project, name=""):
return d
# ------------------------------------------------------------ 旧版学习流程
class CwItem:
def __init__(self, d, cookies):
self.title = d["title"]
href = d["href"]
self.url = href if href.startswith("http") else JSPX + href
self.s = new_session(cookies)
self.rec = None
self.status = "done" if d.get("done") else "pending"
self.last = None
def resolve(self):
try:
r = self.s.get(self.url, timeout=40)
except Exception as e:
return "exc:%s" % e
m = re.search(r"'id':'([0-9A-F]{32})'", r.text)
if not m:
return "no-id"
self.rec = m.group(1)
return "ok"
def start(self):
try:
r = self.s.post(EP_STUDY, data={"id": self.rec, "order": "server_start",
"updatePassPerc": "true"}, timeout=40)
return r.text.strip()[:40]
except Exception as e:
return "EXC:%s" % e
def beat(self):
try:
r = self.s.post(EP_STUDY, data={"id": self.rec, "order": "server_time",
"updatePassPerc": "true",
"videoSchedule": 30}, timeout=40)
return r.text.strip()[:60]
except Exception as e:
return "EXC:%s" % e
def submit_feedback(s, href, key):
url = href if href.startswith("http") else JSPX + href
r = s.get(url, timeout=40)
soup = BeautifulSoup(r.text, "lxml")
fields = {i.get("name"): i.get("value") for i in soup.select("form input[type=hidden]") if i.get("name")}
if fields.get("allowStudy") == "false":
log(key, "反馈表：allowStudy=false，跳过")
return
r2 = s.post(JSPX + "/train/trainee/doFeedbackList.action", data=fields, timeout=40)
qs = BeautifulSoup(r2.text, "lxml")
answers = []
for it in qs.select("div.job-item"):
qtype = it.get("questype")
opts = [sp.get("data-name") for sp in it.select(".job-item-a span")]
ans = ("课程内容充实，讲师讲解清晰，对我的教学实践很有帮助，希望今后能增加更多一线教学案例。"
if qtype == "04" else (opts[0] if opts else "A"))
answers.append({"id": it.get("traineeid"), "quesId": it.get("qumid"),
"quesType": qtype, "quesNo": it.get("quesno"), "answer": ans})
payload = {"planId": fields.get("planId"), "objectId": fields.get("objectId"),
"answers": json.dumps(answers, ensure_ascii=False, separators=(",", ":")),
"submitStatus": "finish"}
rr = s.post(JSPX + "/train/trainee/saveFeedback.action", data=payload, timeout=40)
log(key, "反馈表：%d 题，%s" % (len(answers), "提交成功" if "MSG" not in rr.text else "返回 " + rr.text[:80]))
def submit_quiz(s, href, key, answer_key=None):
url = href if href.startswith("http") else JSPX + href
plan, _ = parse_ids(url)
m = re.search(r"objectId=([0-9A-Fa-f]+)", url)
obj = m.group(1) if m else ""
rr = s.get("%s/train/trainee/showTestQums.action?planId=%s&objectId=%s" % (JSPX, plan, obj), timeout=40)
qs = BeautifulSoup(rr.text, "lxml")
singles = (answer_key or {}).get("singles", {})
multis = (answer_key or {}).get("multis", {})
answers = []
for it in qs.select("div.job-item"):
t, qno = it.get("questype"), it.get("quesno")
opts = [sp.get("data-name") for sp in it.select(".job-item-a span")]
if t == "02":
ans = (multis.get(qno) or ("#".join(opts) if opts else "A"))
else:
ans = singles.get(qno) or (opts[0] if opts else "A")
answers.append({"id": it.get("traineeid"), "quesId": it.get("qumid"),
"quesType": t, "quesNo": qno, "answer": ans})
if not answers:
log(key, "测验：未取到题目，跳过")
return
payload = {"planId": plan, "objectId": obj,
"answers": json.dumps(answers, ensure_ascii=False, separators=(",", ":"))}
r3 = s.post(JSPX + "/train/trainee/endTest.action", data=payload, timeout=40)
r4 = s.get("%s/train/trainee/doTest.action?planId=%s&objectId=%s" % (JSPX, plan, obj), timeout=40)
sc = re.search(r"得分：\s*(\d+)\s*分",
re.sub(r"\s+", " ", BeautifulSoup(r4.text, "lxml").get_text(" ", strip=True)))
log(key, "测验：%d 题，%s，得分 %s" %
(len(answers), "提交成功" if "MSG" not in r3.text else r3.text[:60],
sc.group(1) if sc else "?"))
def _study_builtin(key, s, d, workers=1, cap=30, enable_quiz=False, enable_feedback=False,
mode="fast", parallel=1, quiz_key="", cookies=None, plan="", project="",
name=""):
"""旧版 jspx 的内置流程：反馈表 + 测验 + 刷课件。"""
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
log(key, "反馈表已完成，跳过")
continue
try:
submit_feedback(s, fb["href"], key)
except Exception as e:
log(key, "反馈表异常：%s" % e)
else:
for fb in d["feedback"]:
log(key, "反馈表：已按设置跳过（%s）" % ("已完成" if fb["done"] else "未完成"))
if enable_quiz:
for t in d["test"]:
if t["done"]:
log(key, "测验已完成，跳过")
continue
try:
submit_quiz(s, t["href"], key, quiz_key)
except Exception as e:
log(key, "测验异常：%s" % e)
else:
for t in d["test"]:
if t["done"]:
log(key, "测验：已完成，跳过")
else:
log(key, "测验：⚠ 已按设置跳过答题（避免因正确率不足卡住「完成」），"
"需要时请在页面上手动作答，或打开「自动交测验」开关")
objs = [CwItem(c, cookies) for c in d["courseware"]]
todo = [o for o in objs if o.status != "done"]
done0 = len(objs) - len(todo)
set_state(key, status="studying", total=len(objs), done=done0)
if todo:
log(key, "解析 %d 个学习记录…" % len(todo))
with ThreadPoolExecutor(max_workers=min(workers, len(todo))) as ex:
list(ex.map(lambda x: x.resolve(), todo))
live = [o for o in todo if o.rec]
log(key, "已解析 %d/%d，并发启动会话" % (len(live), len(todo)))
with ThreadPoolExecutor(max_workers=min(workers, len(live))) as ex:
list(ex.map(lambda x: x.start(), live))
t0 = time.time()
while time.time() - t0 < cap * 60:
with RUNS_LOCK:
if (RUNS.get(key) or {}).get("status") == "stopped":
log(key, "已按要求停止")
return
pending = [o for o in live if o.status != "done"]
if not pending:
break
with ThreadPoolExecutor(max_workers=min(workers, len(pending))) as ex:
resps = list(ex.map(lambda x: x.beat(), pending))
newly = []
for o, resp in zip(pending, resps):
o.last = resp
if resp == "study_over":
o.status = "done"
newly.append(o)
el = int(time.time() - t0)
dc = len([o for o in live if o.status == "done"]) + done0
set_state(key, done=dc, elapsed=el)
log(key, "第 %d 轮 %dm%02ds 完成 %d/%d" % (el // 60 + 1, el // 60, el % 60, dc, len(objs)))
for o in newly:
log(key, " ✓ %s" % o.title[:40])
time.sleep(60)
else:
log(key, "所有课件均已完成")
try:
d2 = discover(s, plan, project, name)
set_state(key, progress=d2["progress"], info=d2["info"], cookies=dump_cookies(s))
log(key, "课程结束：%s | %s" % (d2["progress"], d2["info"]))
except Exception as e:
log(key, "复查失败：%s" % e)
# ---------------------------------------------------------------- 平台注册
def _study(key, s, d, **opts):
"""旧版页面的学习流程仍在 engine._study_builtin（反馈表 / 测验 / 刷课件）。"""
from study_tool.engine import _study_builtin
"""旧版页面：反馈表 + 测验 + 刷课件。"""
return _study_builtin(key, s, d, **opts)
src/study_tool/platforms/study.py
+436 -2
View File
@@ -2,13 +2,15 @@
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import parse_qs
from bs4 import BeautifulSoup
from study_tool import store
from study_tool.config import *
from study_tool.platforms.jspx import *
from study_tool.runtime import log
from study_tool.runtime import RUNS, log, set_state
# ------------------------------------------------------- 新版平台：页面识别/提取
@@ -679,10 +681,442 @@ def study_submit_feedback(client, fb, key, train_id):
log(key, "反馈表：提交失败 code=%s %s" % (r.get("code"), str(r.get("msg") or "")[:80]))
# ------------------------------------------------------------ 新版学习流程
def run_study_course(key, s, d, workers, cap, enable_quiz, enable_feedback,
mode="fast", parallel=2):
"""新版平台（study.dgjy.net / Vue SPA）课程：视频走 study/log 接口，
反馈表走 evaluation/result，测验保持手动作答。
mode=fast 只调接口上报（直链 mp4 仍退回真播），速度最快
mode=normal 开无头浏览器真播视频，parallel 控制同时播几个
"""
cookies = dump_cookies(s)
page_url = d.get("page_url") or ""
train_id = d.get("train_id") or ""
client = StudyClient(s, page_url)
# ---- 先确认已加入培训 ----
# 没加入时 study/log 仍然回 200，但服务端不记账：逐个课时重试只会刷一屏「上报未生效」。
if d.get("need_join"):
joined, why = ensure_study_join(s, client, d)
if joined:
log(key, "加入培训：%s" % why)
d["is_join"] = True
else:
log(key, "加入培训被拒：%s" % (why or "未知原因"))
probe = next((c for c in d["courseware"] if not c["done"]), None)
if probe:
good, last = study_complete_period(cookies, page_url, probe)
log(key, "试报 1 个课时：%s" % ("生效" if good else "未生效（%s）" % last))
if not good:
log(key, "该账号对这门课没有写入权限，本次跳过。"
"需要课程管理员把这门课分配到账号下（jspx 在学 ≠ 新版平台已加入）")
return
else:
return
# ---- 反馈表 ----
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
log(key, "反馈表已完成，跳过")
continue
try:
study_submit_feedback(client, fb, key, train_id)
except Exception as e:
log(key, "反馈表异常：%s" % e)
else:
for fb in d["feedback"]:
log(key, "反馈表：已按设置跳过（%s）" % ("已完成" if fb["done"] else "未完成"))
# ---- 测验 ----
for t in d["test"]:
score, passed = t.get("score"), t.get("passed_score")
if t["done"]:
if score is not None and passed is not None:
try:
if float(score) < float(passed):
log(key, "测验：已作答但不及格（%s/%s），需要在页面上重考" % (score, passed))
continue
except Exception:
pass
log(key, "测验：已完成，跳过")
elif enable_quiz:
log(key, "测验：新版平台不代答（%s 需在页面上手动作答）" % t["title"][:30])
else:
log(key, "测验：⚠ 已按设置跳过（%s），需要时请在页面上手动作答" % t["title"][:30])
# ---- 视频课时 ----
objs = d["courseware"]
todo = [c for c in objs if not c["done"]]
done0 = len(objs) - len(todo)
set_state(key, status="studying", total=len(objs), done=done0)
if not objs:
log(key, "没有视频课时")
return
kinds = {}
for c in objs:
kinds[c.get("kind") or "?"] = kinds.get(c.get("kind") or "?", 0) + 1
if kinds:
log(key, "视频类型：%s" % "、".join("%s %d 个" % (k, v) for k, v in kinds.items()))
if not todo:
log(key, "所有课件均已完成")
all_done = True
elif mode == "normal":
got = study_videos_normal(key, cookies, page_url, train_id, todo, parallel, cap, done0)
all_done = got >= len(objs)
set_state(key, done=got)
log(key, "视频课时：%d/%d 已完成" % (got, len(objs)))
else:
got = study_videos_fast(key, cookies, page_url, train_id, todo, workers, cap, done0)
all_done = got >= len(objs)
set_state(key, done=got)
log(key, "视频课时：%d/%d 已完成" % (got, len(objs)))
# ---- 课程容器（periodType 10/13）：子课时学完不会自动完成 ----
if all_done and d.get("containers"):
for ct in d["containers"]:
if ct.get("done"):
continue
try: # 先重算子课程计数：并发上报会丢增量，计数不到容器就不完成
study_cal_final_score(client, ct["kcdm"])
log(key, " 已重算子课程完成计数（kcdm=%s）" % ct["kcdm"][:8])
except Exception as e:
log(key, " 重算子课程完成计数失败：%s" % str(e)[:80])
ok, msg = study_complete_container(client, ct)
log(key, " %s 课程节点「%s」：%s" % ("✓" if ok else "×", ct["name"][:24], msg))
try:
study_cal_final_score(client, train_id)
except Exception:
pass
try:
_reread_study(key, s, d)
except Exception as e:
log(key, "复查失败：%s" % e)
# 平台靠 document.hidden / visibilitychange 判断「用户切走了」，一 hidden 就暂停并记异常。
# 多路真播时把可见性钉死成 visible，不然后台标签页会被判成离开。
VIS_KEEP_JS = """
(() => {
const d = document;
try { Object.defineProperty(d, 'hidden', {get: () => false, configurable: true}); } catch (e) {}
try { Object.defineProperty(d, 'webkitHidden', {get: () => false, configurable: true}); } catch (e) {}
try { Object.defineProperty(d, 'visibilityState', {get: () => 'visible', configurable: true}); } catch (e) {}
d.addEventListener('visibilitychange', e => e.stopImmediatePropagation(), true);
})();
"""
def _classhour_of(url):
m = re.search(r"[?&]classhour=([0-9]+)", url or "")
return m.group(1) if m else ""
def _wait_video(pg, want_id, secs=60):
"""等播放页就绪：("ok", (frame, duration)) / ("redirect", 被切到的课时) / ("timeout", "")。"""
end = time.time() + secs
while time.time() < end:
cur = _classhour_of(pg.url)
if cur and want_id and cur != str(want_id):
return "redirect", cur
for fr in pg.frames:
try:
st = fr.evaluate("""() => {
const v = Array.from(document.querySelectorAll('video')).filter(x => x.duration > 30)[0];
return v ? {d: v.duration} : null; }""")
except Exception:
continue
if st and st.get("d"):
return "ok", (fr, float(st["d"]))
pg.wait_for_timeout(1000)
return "timeout", ""
def _kick_play(frame, seek):
"""保活：静音、1 倍速、必要时 seek 到结尾前 3 秒（seek=False 就是老老实实从头播）。"""
try:
frame.evaluate("""(seek) => {
const v = Array.from(document.querySelectorAll('video')).filter(x => x.duration > 30)[0];
if (!v) return;
v.muted = true;
try { v.playbackRate = 1; } catch (e) {}
if (seek && v.currentTime < v.duration - 5) {
try { v.currentTime = Math.max(0, v.duration - 3); } catch (e) {}
}
if (v.paused) v.play().catch(() => {});
}""", seek)
except Exception:
pass
def _dialog_text(pg):
try:
el = pg.query_selector(".el-message-box__message")
return (el.inner_text() or "").strip() if el else ""
except Exception:
return ""
def play_one_period(key, p, cookies, train_id, item, timeout=900, seek=False, attach_cdp=False):
"""真播一个课时到结尾。返回 (ok, 说明, 平台要求先学的课时 id)。
平台在进入播放页时会先查「是不是轮到这一节」，不按顺序会被弹回并跳到它指定的课时，
这时返回的第三个值就是那一节，调用方据此重排队列。
"""
pid = str(item["period_id"])
url = ("%s/training/courseVideo?classhour=%s&isJoin=1&needJoin=1&courseId=%s"
"&isFront=1&trainCourseId=%s" % (STUDY, pid, train_id, train_id))
posted = {"done": False}
browser, own = None, False
try:
if attach_cdp:
try:
browser = p.chromium.connect_over_cdp("http://127.0.0.1:9222")
except Exception:
browser = None
if browser is None:
browser = p.chromium.launch(headless=True, args=[
"--autoplay-policy=no-user-gesture-required", "--mute-audio",
"--disable-background-timer-throttling", "--disable-renderer-backgrounding"])
own = True
ctx = browser.new_context()
else:
ctx = browser.contexts[0] if browser.contexts else browser.new_context()
for c in cookies:
if "dgjy" in (c.get("domain") or ""):
try:
ctx.add_cookies([{"name": c["name"], "value": c.get("value") or "",
"domain": (c.get("domain") or "").lstrip("."),
"path": c.get("path") or "/"}])
except Exception:
pass
pg = ctx.new_page()
pg.add_init_script(VIS_KEEP_JS)
def on_req(r):
try:
if "period/study/log" in r.url and '"isComplete":1' in (r.post_data or "").replace(" ", ""):
posted["done"] = True
except Exception:
pass
pg.on("request", on_req)
pg.goto(url, wait_until="domcontentloaded", timeout=60000)
status, val = _wait_video(pg, pid, 60)
if status == "redirect":
return False, "平台按顺序放行，页面被切到课时 %s" % val, str(val)
if status != "ok":
return False, "没等到 <video>（可能需要专用播放器的课时）", ""
frame = val[0]
t0 = time.time()
while time.time() - t0 < timeout:
if posted["done"]:
return True, "", ""
msg = _dialog_text(pg)
if "顺序" in msg or "未完成课时" in msg:
return False, "平台按顺序放行：%s" % msg[:30], _classhour_of(pg.url)
_kick_play(frame, seek)
pg.wait_for_timeout(2000)
return False, "超时未等到播放结束上报", ""
except Exception as e:
return False, str(e)[:120], ""
finally:
if own and browser is not None:
try:
browser.close()
except Exception:
pass
def study_videos_fast(key, cookies, page_url, train_id, todo, workers, cap, done0=0):
"""快速学习：直接调 study/log 上报；直链 mp4 接口不认，退回真播。返回完成总数。"""
deadline = time.time() + cap * 60
hard = [c for c in todo if c.get("kind") != "direct"]
soft = [c for c in todo if c.get("kind") == "direct"]
ok = done0
if hard:
log(key, "并发提交 %d 个视频课时（workers=%d）…" % (len(hard), min(workers, len(hard))))
ex = ThreadPoolExecutor(max_workers=min(workers, len(hard)))
futs = {ex.submit(study_complete_period, cookies, page_url, it): it for it in hard}
try:
for fut in as_completed(futs, timeout=max(1, deadline - time.time())):
it = futs[fut]
try:
good, msg = fut.result()
except Exception as e:
good, msg = False, str(e)[:80]
if good:
ok += 1
log(key, " ✓ %s（%ss）" % (it["title"][:40], it.get("duration") or 0))
else:
log(key, " × %s 上报未生效：%s" % (it["title"][:40], msg))
set_state(key, done=ok)
except TimeoutError:
log(key, "已到时间上限（%d 分钟），剩余课时留待下次" % cap)
except Exception as e:
log(key, "批量提交中断：%s" % e)
finally:
ex.shutdown(wait=False, cancel_futures=True)
if soft:
log(key, "另有 %d 个直链视频课时需要真播放到结尾（%s）"
% (len(soft), "、".join(c["title"][:16] for c in soft[:4])))
for it in soft:
if (RUNS.get(key) or {}).get("status") == "stopped":
break
try:
good = play_period_to_end(key, cookies, page_url,
train_id, it["period_id"], it["title"])
except Exception as e:
log(key, " × %s 播放失败：%s" % (it["title"][:30], str(e)[:100]))
continue
if good:
ok += 1
log(key, " ✓ %s（播放到结尾）" % it["title"][:40])
else:
log(key, " × %s 未确认完成" % it["title"][:40])
set_state(key, done=ok)
return ok
def _play_wave(key, cookies, train_id, wave, deadline, seek):
"""一波并发：每路一个独立 playwright + 独立浏览器，互不遮挡。"""
def run(it):
try:
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
return play_one_period(key, p, cookies, train_id, it,
timeout=max(60, deadline - time.time()), seek=seek)
except Exception as e:
return False, str(e)[:100], ""
out = []
ex = ThreadPoolExecutor(max_workers=len(wave))
try:
futs = {ex.submit(run, it): it for it in wave}
for fut in as_completed(futs):
try:
ok, why, wanted = fut.result()
except Exception as e:
ok, why, wanted = False, str(e)[:100], ""
out.append((futs[fut], ok, why, wanted))
finally:
ex.shutdown(wait=False)
return out
def study_videos_normal(key, cookies, page_url, train_id, todo, parallel, cap, done0=0):
"""常规学习：开无头浏览器真播视频，parallel 路并发。
平台可能按课节顺序放行（不在顺序上会被弹回），所以先按并发跑一波；
一旦被弹回就把并发降到 1，并把平台指定的那一节挪到队首，按顺序补回去。
"""
try:
from playwright.sync_api import sync_playwright # noqa: F401
except Exception as e:
log(key, "缺少 playwright，常规学习不可用：%s" % e)
return done0
parallel = max(1, min(int(parallel or 1), 5, len(todo)))
log(key, "常规学习：%d 个课时真播，%d 路并发" % (len(todo), parallel))
deadline = time.time() + cap * 60
ok, pending, order_only, bounced = done0, list(todo), False, {}
while pending and time.time() < deadline:
if (RUNS.get(key) or {}).get("status") == "stopped":
log(key, "已按要求停止")
break
wave, pending = pending[:parallel], pending[parallel:]
log(key, "本轮 %d 路：%s" % (len(wave), "、".join(x["title"][:14] for x in wave)))
for it, good, why, wanted in _play_wave(key, cookies, train_id, wave, deadline, seek=False):
if good:
ok += 1
log(key, " ✓ %s（真播到结尾）" % it["title"][:40])
else:
log(key, " × %s：%s" % (it["title"][:30], why))
if wanted:
order_only = True
n = bounced[it["period_id"]] = bounced.get(it["period_id"], 0) + 1
if n >= 3: # 一直被顺序校验弹回，别再空转
log(key, " × %s：连续被顺序校验弹回，跳过" % it["title"][:30])
else:
i = next((i for i, x in enumerate(pending)
if str(x["period_id"]) == str(wanted)), None)
if i is None: # 平台要的那一节正在播或不在这门课里，先排到队尾
pending.append(it)
else: # 先播平台要的那一节，再回来播这一个
pending.insert(0, pending.pop(i))
set_state(key, done=ok)
if order_only and parallel > 1:
parallel = 1
log(key, "检测到平台按顺序放行，并发降为 1（剩下的逐个播）")
return ok
def play_period_to_end(key, cookies, page_url, train_id, period_id, title, timeout=900):
"""直链视频（fileType=other）：平台只认 onPlayEnd 上报的 isComplete=1。
做法：Playwright 打开课程播放页（带登录 cookie）→ 秒跳到 duration-3 秒 → 播放 →
等接口出现 isComplete=1 → 回读 completeTime 复核。
"""
try:
from playwright.sync_api import sync_playwright
except Exception as e:
log(key, " 缺少 playwright，直链视频无法自动播完：%s" % e)
return False
item = {"period_id": period_id, "title": title}
try:
with sync_playwright() as p:
ok, why, _ = play_one_period(key, p, cookies, train_id, item,
timeout=timeout, seek=True, attach_cdp=True)
except Exception as e:
log(key, " 播放异常：%s" % str(e)[:120])
return False
if not ok:
log(key, " 未播完：%s" % why)
return False
time.sleep(2)
try:
c = StudyClient(new_session(cookies), page_url)
return study_is_done(c, period_id)
except Exception:
return True
def study_sync_gyx(client, train_id):
"""培训里的「课程」是 periodType 10/13 的容器节点，不跟着里面的课时自动完成。
这条同步接口要课程管理员权限：学生账号会回「无权操作该课程」，调一次没坏处，
管理员账号跑的时候能直接把容器刷成完成。
"""
j = client.call("GET", "/course/front/study/sync/gyxCourseData/%s" % train_id)
if j.get("code") == 200:
return "已同步"
return "跳过（%s）" % str(j.get("msg") or ("code=%s" % j.get("code")))[:40]
def _reread_study(key, s, d):
"""课程跑完后按 trainCourseId 直接复查（不重新解析 jspx 页面）。"""
client = StudyClient(s, d.get("page_url") or "")
try:
log(key, "同步莞易学课程完成状态：%s" % study_sync_gyx(client, d["train_id"]))
except Exception as e:
log(key, "同步莞易学课程完成状态失败：%s" % str(e)[:100])
try:
d2 = discover_study(client, d["train_id"], d["course"])
except SessionExpired:
client.bootstrap()
d2 = discover_study(client, d["train_id"], d["course"])
set_state(key, progress=d2["progress"], info=d2["info"], cookies=dump_cookies(s))
log(key, "课程结束：%s | %s" % (d2["progress"], d2["info"]))
# ---------------------------------------------------------------- 学习实现
def _study(key, s, d, **opts):
from study_tool.engine import run_study_course
return run_study_course(key, s, d, opts.get("workers", 1), opts.get("cap", 30),
opts.get("enable_quiz", False), opts.get("enable_feedback", False),
opts.get("mode", "fast"), opts.get("parallel", 2))
src/study_tool/quiz.py
+4 -1
View File
@@ -2,15 +2,18 @@
import json
import re
import time
from urllib.parse import urlsplit
from bs4 import BeautifulSoup
from fastapi.responses import StreamingResponse
from study_tool.webapp import app
from study_tool.config import *
from study_tool.engine import *
from study_tool.llm import *
from study_tool.models import *
from study_tool.platforms import registry
from study_tool.platforms.jspx import new_session, parse_ids
from study_tool.platforms.study import StudyClient
from study_tool.runtime import log
# ------------------------------------------------------------------ quiz
src/study_tool/routes/accounts.py
+1 -2
View File
@@ -9,7 +9,6 @@ from study_tool import store
from study_tool.config import *
from study_tool.models import *
from study_tool.platforms import registry
from study_tool.platforms.jspx import *
from study_tool.runtime import RUNS, RUNS_LOCK, log
def _log_login(username, ok, code, error, elapsed, stage=""):
@@ -87,7 +86,7 @@ def api_login(b: LoginIn):
courses = pf.list_courses(s) if pf.list_courses else []
except Exception as e:
courses, warn = [], str(e)[:200]
cookies = (pf.dump_cookies or dump_cookies)(s)
cookies = pf.dump_cookies(s) if pf.dump_cookies else []
_log_login(username, True, "ok", warn, time.time() - t0, "user=%s" % user)
if acc:
store.account_save_login(acc["id"], user, cookies, courses)
src/study_tool/routes/misc.py
+14 -9
View File
@@ -8,12 +8,11 @@ from study_tool.webapp import app
from study_tool import store
from study_tool.capture import *
from study_tool.config import *
from study_tool.engine import *
from study_tool.engine import study_worker
from study_tool.models import *
from study_tool.platforms import registry
from study_tool.platforms.jspx import *
from study_tool.platforms.study import *
from study_tool.runtime import RUNS, RUNS_LOCK, log
from study_tool.platforms.errors import SessionExpired
from study_tool.runtime import RUNS, RUNS_LOCK, log, set_state
from study_tool.settings_store import load_settings
# ------------------------------------------------------------------ 平台
@@ -74,10 +73,13 @@ def api_courses(b: CookiesIn):
return JSONResponse({"ok": False, "code": "no_platform",
"error": "平台「%s」还没有实现刷新课程，请先在 platforms/ 里补上"
% b.platform}, status_code=200)
s = (pf.session or new_session)(b.cookies)
if pf.session is None:
return JSONResponse({"ok": False, "code": "no_platform",
"error": "平台「%s」还没有实现 session" % b.platform}, status_code=200)
s = pf.session(b.cookies)
if not s.cookies:
return JSONResponse({"ok": False, "code": "no_cookie",
"error": "保存的 cookie 全部被过滤（domain 不含 dgjy），请重新登录"},
"error": "保存的 cookie 全部被过滤（站点域名不匹配），请重新登录"},
status_code=200)
try:
courses = pf.list_courses(s)
@@ -96,7 +98,7 @@ def api_courses(b: CookiesIn):
return {"ok": True, "courses": courses, "user": user,
"in_study": len([c for c in courses if c["in_study"]]),
# 补 jspx 会话时 cookie 会换新（JSESSIONID/route），回传让前端写回
"cookies": (pf.dump_cookies or dump_cookies)(s)}
"cookies": (pf.dump_cookies(s) if pf.dump_cookies else [])}
@app.post("/api/discover")
@@ -105,10 +107,13 @@ def api_discover(b: DiscoverIn):
if pf is None or pf.discover is None:
return JSONResponse({"ok": False, "code": "no_platform",
"error": "平台「%s」还没有实现课程发现" % b.platform}, status_code=200)
s = (pf.session or new_session)(b.cookies)
if pf.session is None:
return JSONResponse({"ok": False, "code": "no_platform",
"error": "平台「%s」还没有实现 session" % b.platform}, status_code=200)
s = pf.session(b.cookies)
try:
d = pf.discover(s, b.plan_id, b.project_id, b.name)
return {"ok": True, **d, "cookies": (pf.dump_cookies or dump_cookies)(s)}
return {"ok": True, **d, "cookies": (pf.dump_cookies(s) if pf.dump_cookies else [])}
except Exception as e:
return JSONResponse({"ok": False, "error": str(e)[:300]}, status_code=200)