Files
study-tool/app.py
T
study-tool 549a91af1d fix: 首次进入课程自动同步加入(syncSelf)
首次进入课程时课时树回 code=500「你还没有加入学习…」,前端把它渲染成挡路弹层,
人工点掉后 SPA 自己补发 syncSelf 才落库 —— 所以第二次进同一门课就正常。

- discover_study 返回值补 course_code(trainCourse/getById.courseCode)
- 新增 study_mod_get:/mod/* 只带 cookie + Referer,不带 Bearer
- 新增 study_sync_self:GET /mod/school/wanpeng/syncSelf?courseCode=…
- 新增 study_sync_identity:GET /mod/gjt/roleList 补一刀身份
- ensure_study_join 改为:已加入 → student/join → syncSelf + 轮询 getById → 刷新 jspx 页
- todo.md 收尾,新增 OPTIMIZATION.md 记录待实测确认项
2026-10-28 20:41:44 +08:00

3415 lines
141 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
多账号学习控制台 —— 东莞市教师培训平台 (jspx.dgjy.net)
设计
* 账号、密码、会话 cookie、课程缓存、运行日志、系统设置全部存 **SQLite3**(store.py)
默认库文件 data/study.db,可用 DB_PATH 指定;重启/换机器只搬这一个文件
* 登录后自动抓取 /admin/main.action 的「在学课程」列表
* 勾选课程 -> 并发挂机 -> 自动交反馈表与测验
* 多个账号可同时运行
启动: python app.py 浏览器打开 http://127.0.0.1:8848 (Docker 见 README)
"""
import json
import os
import re
import shutil
import subprocess
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from html import unescape as html_unescape
from urllib.parse import parse_qs, unquote, urljoin, urlsplit
import requests
import uvicorn
from bs4 import BeautifulSoup
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
from pydantic import BaseModel
import store
HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "data")
LOGDIR = os.path.join(DATA, "logs") # 旧版日志目录;只有首次迁移会读它
os.makedirs(LOGDIR, exist_ok=True)
def _find_node():
"""Locate a node.exe: $NODE_BIN > PATH > common install dirs."""
for c in [os.environ.get("NODE_BIN"), shutil.which("node"), shutil.which("node.exe"),
r"D:\Software\devlopSoftware\Node\24.16.0\node.exe",
r"C:\Program Files\nodejs\node.exe",
r"C:\Program Files (x86)\nodejs\node.exe"]:
if c and os.path.exists(c):
return c
return "node"
NODE = _find_node()
DESENC = os.path.join(HERE, "desenc.js")
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")
JSPX = "https://jspx.dgjy.net"
TPASS = "https://tyyh.dgjy.net/tpass"
# 表单提交带上来源头,和浏览器一致(该站有 WAF,字段异常会直接 403)
LOGIN_HEADERS = {"Referer": TPASS + "/login", "Origin": "https://tyyh.dgjy.net"}
AUTHORIZE = ("https://tyyh.dgjy.net/tpass/oauth2.0/authorize?client_id=wpxtgl&scope=all"
"&response_type=code&state=random&redirect_uri=https%3A%2F%2Fdj.dgjy.net%2Fhomepage"
"%2Fremote%2Fopenapi%2Fdgwjt%2Flogin%3Furl%3Dhttps%3A%2F%2Fjspx.dgjy.net%2Ftstudy"
"%2Fpassport%2Flogin.action")
EP_STUDY = JSPX + "/train/trainee/courseware-studyTimeUpdate.action"
# 新版培训平台(莞易学/study.dgjy.net)的课程接口。2026 年起的「慕课」课程不再
# 用 jspx 的 tier-tree 页面,而是 Vue SPA + /train/course/front/* 这组 JSON 接口。
TRAIN_AUTHORIZE_PATH = ("/oauth2.0/authorize?client_id=gyx&scope=all&response_type=code"
"&state=random&redirect_uri=https%3A%2F%2Fstudy.dgjy.net%2F%23%2Flogin%2Fback")
STUDY = os.environ.get("STUDY_BASE", "https://study.dgjy.net").rstrip("/")
DEBUG_DIR = os.path.join(DATA, "debug") # 页面快照的兜底目录(正常存 SQLite 的 snapshots 表)
os.makedirs(DEBUG_DIR, exist_ok=True)
HEX32 = re.compile(r"^[0-9A-Fa-f]{32}$")
# OAuth 授权码:tpass 发的是 ST-369516-xxxx-tpass,带连字符,不能只认 [0-9A-Za-z]
CODE_RE = re.compile(r"(?:[?#&]|\b)code=([0-9A-Za-z._\-]{6,128})")
# 「我的课程」的三个板块 -> 各自的 AJAX 列表接口与 courseType
COURSE_SECTIONS = [
("在学课程", "/train/trainee/studyingCourse.action", "studying"),
("助学课程", "/train/assist/assistanceCourse.action", "assistance"),
("督学课程", "/train/inspect/inspectingCourse.action", "inspecting"),
]
# 每个板块的三种状态(页面上是「进行中 / 未开始 / 已结束」三个页签)
PLAN_STATUS = [
("进行中", "TAKING_PLAN"),
("未开始", "NO_BEGAIN_PLAN"),
("已结束", "FINISHED_PLAN"),
]
# 只有这两种状态还能进去学;已结束的只能查看
STUDYABLE_STATUS = {"TAKING_PLAN", "NO_BEGAIN_PLAN"}
class SessionExpired(RuntimeError):
"""站点要求重新登录(cookie 已失效)。"""
# ------------------------------------------------------------------------ crypto
def str_enc(data: str) -> str:
r = subprocess.run([NODE, DESENC, data, "1", "2", "3"],
capture_output=True, text=True, timeout=30)
if r.returncode != 0:
raise RuntimeError("strEnc failed: %s" % (r.stderr or "")[:200])
return r.stdout.strip()
# ------------------------------------------------------------------- http client
def new_session(cookies=None):
s = requests.Session()
s.headers.update({"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"})
for c in (cookies or []):
if "dgjy" not in (c.get("domain") or ""):
continue
s.cookies.set(c["name"], c.get("value") or "",
domain=(c.get("domain") or "").lstrip("."),
path=c.get("path") or "/")
return s
def dump_cookies(s):
return [{"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
for c in s.cookies]
def whoami(html):
soup = BeautifulSoup(html, "lxml")
hdr = soup.select_one("#header")
if hdr:
m = re.search(r"你好[,,]?\s*(\S+)\s*老师", re.sub(r"\s+", " ", hdr.get_text(" ", strip=True)))
if m:
return m.group(1)
return ""
def parse_ids(text):
plan = project = ""
m = re.search(r"planId=([0-9A-Fa-f]{16,})", text or "")
if m:
plan = m.group(1)
m = re.search(r"projectId=([0-9A-Fa-f]{16,})", text or "")
if m:
project = m.group(1)
return plan, project
def _follow_js_login(s, page):
"""补上浏览器才会执行的两跳,否则只拿得到 CASTGC、拿不到 jspx 的会话。 `oauth2.0/authorize` 链的最后一跳**不是 302**,而是 dj.dgjy.net 上一个
1486 字节的 JS 中转页,里面动态插 <script>:
scriptLogin.src = 'https://passport.dgjy.net/scriptLogin?action=simpleLogin&d1=...&d2=...';
scriptLogin 返回的是 JS,不是跳转:
var loginError=null;
var verifyURL="https://dj.dgjy.net/fpf/login/verify.action?ticket=...&auth=...&sid=...";
location.href=verifyURL;
verify.action 之后才会 302 → loginCall → jspx/tstudy/passport/login.action,
到这一步 jspx.dgjy.net 的 JSESSIONID 才真正下发。
requests 不执行 JS,所以这两跳必须手工走一遍。
"""
m = re.search(r"scriptLogin\.src\s*=\s*'([^']+)'", page or "")
if not m:
return False
r = s.get(html_unescape(m.group(1)), timeout=40)
body = r.text or ""
if re.search(r"loginError\s*=\s*(?!null)", body):
raise RuntimeError("SSO 校验未通过:%s" % body.strip()[:200])
m2 = re.search(r'verifyURL\s*=\s*"([^"]+)"', body)
if not m2:
return False
s.get(html_unescape(m2.group(1)).replace("\\/", "/"), timeout=40, allow_redirects=True)
return True
def _js_redirect_candidates(page):
"""从一个不执行 JS 的响应体里,按出现顺序抽出下一跳 URL。"""
out = []
for pat in (r"scriptLogin\.src\s*=\s*'([^']+)'",
r'scriptLogin\.src\s*=\s*"([^"]+)"',
r'verifyURL\s*=\s*"([^"]+)"',
r"verifyURL\s*=\s*'([^']+)'",
r"location\.href\s*=\s*['\"]([^'\"]+)['\"]",
r"window\.location\.href\s*=\s*['\"]([^'\"]+)['\"]"):
out.extend(re.findall(pat, page or ""))
return [html_unescape(u).replace("\\/", "/") for u in out if u]
def follow_js_hops(s, start_url, stop=None, max_hops=6):
"""通用 SSO 补跳:页面给的不是 302 而是 JS/scriptLogin 时,手工走完。
study.dgjy.net 的授权链和 jspx 不是同一套页面模板,所以这里按「下一跳 URL」
逐个 GET,直到 stop() 为真(例如 Middle-Token 已经下发)或没有下一跳。
"""
cur, seen = start_url, set()
for _ in range(max_hops):
if cur in seen:
break
seen.add(cur)
try:
r = s.get(cur, timeout=40, allow_redirects=True)
except Exception:
break
if stop and stop():
return True
nxt = _js_redirect_candidates(r.text or "")
if not nxt:
break
cur = urljoin(r.url, nxt[0])
return bool(stop and stop())
def _login_error(html, status=200):
"""从 /tpass/login 的响应里取出服务端提示。
提示藏在 <script type="text/html" id="login_template"> 里的
<span id="errormsghide" style="display:none">,BeautifulSoup 的 get_text()
看不见 script 内容,所以只能按原文正则取;锁号是另一张 403 提示页。
"""
text = html or ""
m = re.search(r"验证失败次数过多[^<]*", text)
if m:
return m.group(0).strip()
m = re.search(r'<span[^>]*id="errormsghide"[^>]*>(.*?)</span>', text, re.S)
if m:
msg = re.sub(r"<[^>]+>", " ", m.group(1))
msg = re.sub(r"\s+", " ", html_unescape(msg).replace("\xa0", " ")).strip()
if msg:
return msg
m = re.search(r"(用户名|密码|账号|账户)[^<>]{0,40}(错误|不存在|不正确|锁定)", text)
if m:
return m.group(0).strip()
if status == 403:
return "账号已被锁定(403),约 10 分钟后解锁"
return ""
def parse_accountlist(html):
"""多身份账号:登录响应里内联 `var accountlist = '<json>';`。
每个元素带 USERNAME(身份证)、CODENAME(教师/家长/学生)、UNIT_NAME,
以及该身份自己的 LOGINUSERNAME / LOGINPASSWORD —— 选身份那一步是服务端
把可登录的账号密码回给前端,再带 choosenumber 提交一次。
"""
m = re.search(r"var\s+accountlist\s*=\s*'(.*?)';", html or "", re.S)
if not m:
return []
try:
data = json.loads(html_unescape(m.group(1)))
except Exception:
return []
return data if isinstance(data, list) else []
def _login_as_identity(s, un, lt, ex, it):
"""按 login_dgjy.js 的选身份分支提交:rsa 用该身份的 LOGINUSERNAME+LOGINPASSWORD。"""
name = it.get("LOGINUSERNAME") or ""
pwd = it.get("LOGINPASSWORD") or ""
payload = {"un": un, "ul": str(len(name)), "pl": str(len(pwd)),
"rsa": str_enc(name + pwd + lt), "lt": lt,
"choosenumber": it.get("USERNAME") or "", "_eventId": "submit"}
if ex:
payload["execution"] = ex
return s.post(TPASS + "/login", data=payload, timeout=30, allow_redirects=True,
headers=LOGIN_HEADERS)
def ensure_jspx_session(s):
"""用 CASTGC 走老平台(jspx)的授权链,建立 jspx.dgjy.net 的会话;成功返回 True。
必须先把 jspx 的 JSESSIONID / route 清掉:这两个是粘性会话 cookie,
浏览器里带过来的那份常常指向已经下线的后端节点(route 对不上号),
留着就永远跳 /admin/login.action,看起来像「登录失效」。
"""
try:
r = s.get(JSPX + "/admin/main.action", timeout=30)
if "table-results" in (r.text or ""):
return True
except Exception:
pass
for c in list(s.cookies):
if "jspx" in (c.domain or ""):
s.cookies.clear(c.domain, c.path, c.name)
try:
r = s.get(AUTHORIZE, timeout=40, allow_redirects=True)
_follow_js_login(s, r.text or "")
r2 = s.get(JSPX + "/admin/main.action", timeout=30)
return "table-results" in (r2.text or "")
except Exception:
return False
def _jspx_call(s, fn):
"""跑一个 jspx 请求;会话失效且手上还有 CASTGC 时,清掉粘性 cookie 重新授权后重试一次。
浏览器 / IndexedDB 里存着的 jspx JSESSIONID+route 经常指向已经下线的后端节点,
fetch_admin 看到的就是「跳回 /admin/login.action」,看着像登录过期,
实际 CASTGC 还有效 —— 这种不该让用户重新输密码。
"""
try:
return fn()
except SessionExpired:
if not ensure_jspx_session(s):
raise
return fn()
def login(username: str, password: str):
"""SSO login -> (session, user)。"""
s = new_session()
r = s.get(TPASS + "/login", timeout=30)
lt = re.search(r'id="lt"[^>]*value="([^"]*)"', r.text)
lt = lt.group(1) if lt else ""
ex = re.search(r'name="execution"[^>]*value="([^"]*)"', r.text)
payload = {
"un": username, "pd": password,
"ul": str(len(username)), "pl": str(len(password)),
"rsa": str_enc(username + password + lt),
"lt": lt, "choosenumber": "", "_eventId": "submit",
}
if ex:
payload["execution"] = ex.group(1)
r2 = s.post(TPASS + "/login", data=payload, timeout=30, allow_redirects=True,
headers=LOGIN_HEADERS)
if not any(c.name == "CASTGC" for c in s.cookies):
accts = parse_accountlist(r2.text)
if accts:
# 一个身份证下挂多个身份时,第一次提交只回账号选择页,不会给票据。
# 这里取教师身份(本平台主要场景),没有就取第一个,然后用该身份重提。
pick = next((x for x in accts if "教师" in (x.get("CODENAME") or "")), accts[0])
m = re.search(r'id="lt"[^>]*value="([^"]*)"', r2.text)
ex2 = re.search(r'name="execution"[^>]*value="([^"]*)"', r2.text)
r2 = _login_as_identity(s, username, (m.group(1) if m else lt),
(ex2.group(1) if ex2 else (ex.group(1) if ex else "")), pick)
if not any(c.name == "CASTGC" for c in s.cookies):
msg = _login_error(r2.text, r2.status_code)
raise RuntimeError((msg or "登录失败:未获得登录票据(账号或密码错误?)")[:200])
# 老平台:authorize 链的终点是 JS 中转页,必须手工补两跳才建得起 jspx 会话
if not ensure_jspx_session(s):
raise RuntimeError("SSO 登录未完成:jspx.dgjy.net 没有建立会话(登录链路可能已改版)")
html = fetch_admin(s)
# 同一张 CASTGC 再授权一次给新版平台(client_id=gyx),拿 Middle-Token
ensure_study_session(s)
return s, whoami(html)
def _next_hop_urls(page):
"""不执行 JS 时,从响应体里找出所有可能的下一跳(SSO 中转页/脚本/表单)。"""
out = _js_redirect_candidates(page)
for m in re.finditer(r'(?:src|href)\s*=\s*["\']([^"\']*(?:login|callback|authorize|verify|code)[^"\']*)["\']',
page or "", re.I):
out.append(html_unescape(m.group(1)))
for m in re.finditer(r'<form[^>]+action=["\']([^"\']+)["\']', page or "", re.I):
out.append(html_unescape(m.group(1)))
seen, uniq = set(), []
for u in out:
u = u.replace("\\/", "/")
if u.startswith("http") or u.startswith("/") or "?" in u:
if u not in seen:
seen.add(u)
uniq.append(u)
return uniq
def ensure_study_session(s, force=False):
"""用现有 SSO 票把 study.dgjy.net 的会话补上。
新版平台的登录是 OAuth 授权码换 token(前端 SPA 源码里写死了这几个地址):
GET tyyh.dgjy.net/tpass/oauth2.0/authorize?client_id=gyx
&redirect_uri=https://study.dgjy.net/#/login/back&response_type=code
... -> study.dgjy.net/#/login/back?code=ST-xxxx-tpass
POST study.dgjy.net/mod/gjt/login/<CODE> # 换出前端会话(JSESSIONID)
GET study.dgjy.net/mod/education/gyx/train/front/login
-> {"result":{"url":".../training","token":"<jwt>"}} # 这才是 Middle-Token
成功标准只有一个:cookie 里出现 study.dgjy.net 的 Middle-Token。
"""
if force:
for c in list(s.cookies):
if c.name == "Middle-Token":
try:
s.cookies.clear(c.domain, c.path, c.name)
except Exception:
pass
if _study_token(s):
return True
try:
r = s.get(TPASS + TRAIN_AUTHORIZE_PATH, timeout=40, allow_redirects=True)
except Exception:
return False
cur_url, body = r.url or "", r.text or ""
tried = set()
for _ in range(6):
m = CODE_RE.search(body) or CODE_RE.search(cur_url)
if m and m.group(1) not in tried:
tried.add(m.group(1))
if _exchange_code(s, m.group(1)):
return True
hops = _next_hop_urls(body)
if not hops:
break
nxt = urljoin(cur_url or STUDY, hops[0])
if "code=" in nxt:
code = (CODE_RE.search(nxt) or [None, ""])[1]
if code and code not in tried and _exchange_code(s, code):
return True
try:
r = s.get(nxt, timeout=40, allow_redirects=True)
except Exception:
break
cur_url, body = r.url or "", r.text or ""
return bool(_study_token(s))
def _fetch_train_token(s):
"""SSO 登录后,再用 gyx 会话换培训系统的 token(前端 `_51` 那个接口)。
返回体形如:
{"result":{"url":"https://study.dgjy.net/training","token":"<jwt>"},"resultCode":200}
这个 token 就是培训系统的 `Middle-Token`(前端把它写进 cookie 再发 Bearer)。
"""
try:
r = s.get(STUDY + "/mod/education/gyx/train/front/login", timeout=40,
headers={"Accept": "application/json, text/plain, */*"})
j = r.json()
except Exception:
return False
data = j.get("result") if isinstance(j, dict) else None
if not isinstance(data, dict):
data = j.get("data") if isinstance(j, dict) else None
tok = ""
if isinstance(data, dict):
for k in ("token", "accessToken", "middleToken"):
if data.get(k):
tok = str(data[k])
break
if not tok:
return False
s.cookies.set("Middle-Token", tok, domain=urlsplit(STUDY).hostname, path="/")
return True
def _exchange_code(s, code):
"""code -> Middle-Token。前端是 POST /mod/gjt/login/<code>,再取 gyx 培训 token。"""
paths = ("/mod/gjt/login/%s", "/train/mod/gjt/login/%s")
for path in paths:
try:
rr = s.post(STUDY + (path % code), timeout=40,
headers={"Accept": "application/json, text/plain, */*"})
except Exception:
continue
if _study_token(s):
return True
try:
j = rr.json()
except Exception:
j = {}
ok = isinstance(j, dict) and str(j.get("resultCode")) == "200"
if ok:
# SSO 换票成功后,再换培训系统的 Bearer token(Middle-Token)
if _fetch_train_token(s):
return True
data = j.get("result") or {}
if isinstance(data, dict):
for k in ("token", "accessToken", "middleToken"):
if data.get(k):
s.cookies.set("Middle-Token", str(data[k]),
domain=urlsplit(STUDY).hostname, path="/")
return True
return False
def fetch_admin(s):
"""GET /admin/main.action。
站点会话失效时**不返回 302**,而是 200 + 55 字节的 JS 跳转存根:
<script>location.href="/admin/login.action";</script>
旧代码直接把它当正常页面解析 -> 一个表格都没有 -> 静默返回空列表,
于是「会话失效」「站点改版」「账号确实没有课」被压成同一条提示。
"""
r = s.get(JSPX + "/admin/main.action", timeout=40)
html = r.text or ""
if "/admin/login.action" in html and "table-results" not in html:
# CASTGC 还有效的话,这里能无密码把 jspx 会话补回来
if ensure_jspx_session(s):
r = s.get(JSPX + "/admin/main.action", timeout=40)
html = r.text or ""
if "/admin/login.action" in html and "table-results" not in html:
raise SessionExpired("登录已失效:站点把请求跳回了 /admin/login.action(请重新登录)")
if "table-results" not in html:
raise RuntimeError("课程页结构异常:未找到 table-results 表格(站点可能已改版,页面长度 %d)"
% len(html))
return html
def _parse_course_rows(html, section=""):
"""从「我的课程」列表片段里抽出课程行。
行结构(各板块一致):
<table class="table-results"><tr><th>在学课程</th></tr>
<tr><td><p class="course-tt"><a href="courseStudyMain.action?planId=..&projectId=..">名字</a></p>
<p class="course-dd"><span>班期:..</span><span>分数:82</span><span>成绩:合格</span></p>
</td><td><a class="abtn-orange-big">查看</a></td></tr>
</table>
注意 href 是**不带前导斜杠的相对路径**(`courseStudyMain.action?...`),
相对 `/train/trainee/` 解析,不能直接 `JSPX + href`。
"""
soup = BeautifulSoup(html, "lxml")
out = []
for table in soup.select("table.table-results"):
th = table.select_one("th")
sec = section or (re.sub(r"\s+", " ", th.get_text(" ", strip=True)) if th else "")
# 只按「有没有课程行」判断,不能用 "暂无" in table.get_text():
# 表格里任何一个单元格出现「暂无」都会把整张表(含全部课程)丢掉。
rows = [tr for tr in table.select("tr") if tr.select_one("p.course-tt a")]
for tr in rows:
a = tr.select_one("p.course-tt a")
href = unquote(a.get("href") or "")
plan, project = parse_ids(href)
if not plan or not project:
continue
dd = tr.select_one("p.course-dd")
out.append({
"section": sec,
"name": a.get_text(" ", strip=True),
"info": re.sub(r"\s+", " ", dd.get_text(" ", strip=True)) if dd else "",
"plan_id": plan,
"project_id": project,
"url": urljoin(JSPX + "/train/trainee/", href),
"progress": "",
})
return out
def scrape_courses(s, with_progress=True):
"""把「我的课程」下三个板块 × 三种状态**全部**拉出来,含已学完的。
站点把列表拆成了 AJAX 接口,每个组合一次请求:
/train/trainee/studyingCourse.action?courseType=studying&planStatus=TAKING_PLAN
/train/assist/assistanceCourse.action?courseType=assistance&planStatus=FINISHED_PLAN
/train/inspect/inspectingCourse.action?courseType=inspecting&planStatus=NO_BEGAIN_PLAN
...
只看 /admin/main.action 是看不到「已结束」课程的——那个首页只渲染在学/助学/督学
三张「进行中」的概览表,课一学完就从首页消失了。
返回空列表只代表「三个板块九种状态都没课」;
会话失效或页面改版一律抛异常,让调用方能给出准确提示。
"""
# 先用首页校验会话,避免九个接口各自失败还看不出来
_jspx_call(s, lambda: fetch_admin(s))
out, seen = [], set()
for section, path, ctype in COURSE_SECTIONS:
for label, status in PLAN_STATUS:
url = "%s%s?courseType=%s&planStatus=%s" % (JSPX, path, ctype, status)
try:
html = s.get(url, timeout=40).text
except Exception:
continue
for c in _parse_course_rows(html, section):
key = (c["plan_id"], c["project_id"])
if key in seen:
continue
seen.add(key)
c["course_type"] = ctype
c["status"] = status
c["status_label"] = label
c["in_study"] = status in STUDYABLE_STATUS
out.append(c)
# 进度只在「进行中/未开始」的课上查,已结束的动辄几十门,逐个请求太慢
if with_progress:
for c in out:
if not c["in_study"]:
continue
try:
rr = s.get(c["url"], timeout=30)
p = BeautifulSoup(rr.text, "lxml").select_one("p.loading font")
c["progress"] = p.get_text(strip=True) if p else ""
except Exception:
pass
return out
def parse_jspx_course(html):
"""老版 jspx 课程页(tier-tree)解析。"""
soup = BeautifulSoup(html, "lxml")
course = soup.select_one("h1")
prog = soup.select_one("p.loading font")
info = soup.select_one("div.results-info p.info")
cw, fb, ts = [], [], []
for one in soup.select("div.tier-tree-one"):
ns = one.select_one("p.tier1.has-children span.name")
chapter = re.sub(r"\s+", " ", ns.get_text(" ", strip=True)) if ns else "?"
for p in one.select("p.tier3"):
a = p.select_one("a")
if not a:
continue
href = a.get("href") or ""
txt = re.sub(r"\s+", " ", p.get_text(" ", strip=True))
name = p.select_one("span.name")
title = re.sub(r"\s+", " ", name.get_text(" ", strip=True)) if name else txt[:40]
m = re.search(r"(\d+)分钟", txt)
done = "已完成" in txt
if "courseware.action" in href:
cw.append({"chapter": chapter, "title": title, "href": href, "done": done,
"duration": int(m.group(1)) if m else None})
elif "doFeedback.action" in href:
fb.append({"title": title, "href": href, "done": done})
elif "doTest.action" in href:
ts.append({"title": title, "href": href, "done": done})
return {
"user": whoami(html),
"course": course.get_text(" ", strip=True) if course else "?",
"progress": prog.get_text(strip=True) if prog else "?",
"info": re.sub(r"\s+", " ", info.get_text(" ", strip=True)) if info else "?",
"courseware": cw, "feedback": fb, "test": ts,
}
def discover(s, plan, project, name=""):
"""打开课程页并解析;会话失效时先用 CASTGC 补一次 jspx 授权再重试。"""
return _jspx_call(s, lambda: _discover_once(s, plan, project, name))
def _discover_once(s, plan, project, name=""):
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 ""
if "/admin/login.action" in html and len(html) < 2000:
# 课程页也走同一个判断:CASTGC 有效就无密码续期
if ensure_jspx_session(s):
r = s.get(url, timeout=40)
html = r.text or ""
if "没有权限" in html:
raise RuntimeError("该账号对此课程没有权限")
if "/admin/login.action" in html and len(html) < 2000:
raise SessionExpired("登录已失效:课程页把请求跳回了 /admin/login.action(请重新登录)")
# 课程页 302 到 study 时,token 在跳转 URL 里;哪怕 SPA 外壳返回空也要当成学过的新平台
study_url = "study.dgjy.net" in (r.url or "")
tok = extract_study_token(r.url, r.history)
if tok:
host = urlsplit(r.url).hostname or urlsplit(STUDY).hostname
try:
s.cookies.set("Middle-Token", tok, domain=host, path="/")
except Exception:
pass
if len(html) < 3000 and not (study_url or tok or has_study_marker(html, r.url)):
raise RuntimeError("课程页返回异常(页面长度仅 %d,登录可能已失效)" % len(html))
platform = detect_platform(html, r.url)
if platform == "jspx":
d = parse_jspx_course(html)
if d["courseware"] or d["feedback"] or d["test"]:
d["platform"] = "jspx"
return d
# 老结构但一个活动都没有:多半是「慕课」外链页(页面还在 jspx,课时在 study 平台)
if not has_study_marker(html, r.url):
snap = save_snapshot(plan, project, r.url, html)
raise RuntimeError("课程页能打开,但没有任何课件/反馈表/测验(页面已存 %s)。"
"若是慕课课程,请确认新平台登录态是否有效" % snap)
elif platform != "study":
snap = save_snapshot(plan, project, r.url, html)
raise RuntimeError("无法识别课程页结构(站点可能又改版):页面里既没有 jspx 的 "
"tier-tree,也没有新版 SPA 标记。页面已存 %s,请把它发我" % snap)
# ---- 新版平台(Vue SPA + /train/course/front/* JSON 接口)----
tid = extract_train_id(r.url, html)
user = whoami(html)
client = StudyClient(s, r.url)
def _load():
nonlocal tid
if not tid:
tid = study_resolve_train_id(client, name)
if not tid:
snap = save_snapshot(plan, project, r.url, html)
raise RuntimeError("新版课程页未带 trainCourseId,且按课程名也匹配不到(课程名:%s)。"
"页面已存 %s" % (name or "?", snap))
return discover_study(client, tid, name, user)
try:
d = _load()
except SessionExpired:
client.bootstrap()
tid = extract_train_id(r.url, html)
d = _load()
d["platform"] = "study"
# URL 里可能带 token(凭据),日志/状态里一律打码;鉴权已经写进 cookie
d["page_url"] = re.sub(r"([?&])(token|access_token|securityKey|sign)=[^&]*", r"\1\2=***", r.url or "")
d["plan_id"] = plan
d["project_id"] = project
return d
# ------------------------------------------------------- 新版平台:页面识别/提取
def detect_platform(html, page_url=""):
"""判断课程页是哪一代前端。
jspx 老页:div.tier-tree-one / p.tier3
新版:Vue SPA(<div id="app"> + static/js/app.*.js)或直接带 /training/ 链接
"""
if "tier-tree-one" in html or "tier3" in html:
return "jspx"
if has_study_marker(html, page_url):
return "study"
return "unknown"
def has_study_marker(html, page_url=""):
if page_url and "study.dgjy.net" in page_url:
return True
for k in ("/training/curriculum", "/training/courseVideo", "/training/comment",
"static/js/app.", "study.dgjy.net/training", "study.dgjy.net/#/"):
if k in html:
return True
flat = html.replace('"', "").replace("'", "")
return "<div id=app>" in flat
def extract_train_id(page_url, html):
"""从跳转后的 URL / 页面内链接 / 隐藏域里找新版培训课程 ID(32 位 hex)。"""
q = parse_qs(urlsplit(page_url or "").query)
for k in ("trainCourseId", "courseId", "id"):
for v in q.get(k, []):
if HEX32.match(v or ""):
return v
# 页面里指向 study 的链接 / iframe / JS 跳转
for m in re.finditer(r"(?:https?:)?//[^\s\"'<>]*study\.dgjy\.net/[^\s\"'<>]+", html or ""):
u = m.group(0)
tail = u.split("#", 1)[1] if "#" in u else u
qq = parse_qs(tail.split("?", 1)[1]) if "?" in tail else {}
for k in ("trainCourseId", "courseId", "id"):
for v in qq.get(k, []):
if HEX32.match(v or ""):
return v
# 相对路径 /training/curriculum?id=...
for m in re.finditer(r"/training/(?:curriculum|courseVideo|comment)\?([^\s\"'<>]+)", html or ""):
qq = parse_qs(m.group(1).replace("&amp;", "&"))
for k in ("trainCourseId", "courseId", "id"):
for v in qq.get(k, []):
if HEX32.match(v or ""):
return v
# 隐藏域
try:
soup = BeautifulSoup(html or "", "lxml")
for i in soup.select("input[type=hidden]"):
nm = (i.get("name") or "").lower()
val = (i.get("value") or "").strip()
if HEX32.match(val) and ("course" in nm or nm in ("id", "trainid")):
return val
except Exception:
pass
return ""
def extract_study_token(page_url, history=None):
"""课程页跳转到 study 时,token 直接挂在 URL 上(前端读它写 Middle-Token cookie)。
courseStudyMain.action -> 302 study/openapi/train/front/getById?...
-> 302 study/training/trainDetails?id=..&token=<jwt>
"""
urls = [page_url or ""]
for h in history or []:
try:
urls.append(h.headers.get("Location") or "")
except Exception:
pass
for u in urls:
if not u:
continue
tail = u.split("#", 1)[1] if "#" in u else u
qq = parse_qs(tail.split("?", 1)[1]) if "?" in tail else {}
for k in ("token", "access_token", "Middle-Token"):
v = qq.get(k)
if v and v[0]:
return v[0]
return ""
def save_snapshot(plan, project, url, html):
"""存下原始页面(进 SQLite),方便站点再次改版时定位选择器。
返回给用户的是取回地址 /api/snapshot/<id>;写库失败才退回文件。
"""
safe_url = re.sub(r"([?&])(token|access_token|securityKey|sign)=[^&]*", r"\1\2=***", url or "")
try:
sid = store.snapshot_add(plan, project, safe_url, html or "")
return "/api/snapshot/%d" % sid
except Exception:
pass
name = "course_%s_%s_%s.html" % (re.sub(r"[^\w.-]", "_", str(plan))[:20],
re.sub(r"[^\w.-]", "_", str(project))[:20],
time.strftime("%H%M%S"))
path = os.path.join(DEBUG_DIR, name)
try:
with open(path, "w", encoding="utf-8") as f:
f.write("<!-- final url: %s -->\n" % safe_url)
f.write(html or "")
return path
except Exception:
return "(保存失败)"
# ------------------------------------------------------- 新版平台:HTTP 客户端
def _study_token(s):
for c in s.cookies:
if c.name == "Middle-Token" and "dgjy" in (c.domain or ""):
return c.value
return ""
class StudyClient:
"""新版平台接口客户端。
API 前缀不写死:课程页可能由 jspx 代理、也可能跳到 study.dgjy.net,
所以按「页面 origin/train」→「study.dgjy.net/train」→「gjt/tp_edu_biz/train」
依次试,哪个能返回 200 就用哪个(结果缓存)。
"""
def __init__(self, session, page_url=""):
self.s = session
self.token = _study_token(session)
self.bases = []
origin = ""
if page_url:
u = urlsplit(page_url)
if u.scheme and u.netloc:
origin = "%s://%s" % (u.scheme, u.netloc)
for b in ([origin + "/train"] if origin else []) + [STUDY + "/train"]:
if b not in self.bases:
self.bases.append(b)
if origin and "gjt.dgjy.net" in origin:
for b in (origin + "/tp_edu_biz/train",):
if b not in self.bases:
self.bases.append(b)
self.base = None
self.last_errors = []
def _raw(self, method, base, path, params=None, json_body=None):
h = {"Accept": "application/json, text/plain, */*",
"Referer": STUDY + "/training/curriculum"}
if self.token:
h["Authorization"] = "Bearer " + self.token
r = self.s.request(method, base + path, params=params, json=json_body,
headers=h, timeout=40)
try:
return r.json()
except Exception:
# 不是 JSON:多半是打到了错误前缀(返回登录页/HTML),标 -1 让 call() 换下一个 base
return {"code": -1 if r.status_code < 400 else r.status_code,
"msg": "非 JSON 响应:%s" % (r.text or "")[:120]}
def call(self, method, path, params=None, json_body=None):
bases = ([self.base] if self.base else []) + [b for b in self.bases if b != self.base]
self.last_errors = []
saw_auth = False
for b in bases:
try:
j = self._raw(method, b, path, params, json_body)
except Exception as e:
self.last_errors.append("%s -> %s" % (b, e))
continue
code = j.get("code")
if code in (401, 403) or code == 404 or code == -1:
saw_auth = saw_auth or code in (401, 403)
self.last_errors.append("%s -> %s" % (b, j.get("msg") or ("业务码 %s" % code)))
if self.base == b:
self.base = None
continue
self.base = b
return j
if saw_auth and not self.base:
raise SessionExpired("新版平台没有登录态(Middle-Token 缺失或已过期):请点「重新登录」。%s"
% "; ".join(self.last_errors)[:180])
raise RuntimeError("新版平台接口不可用:%s" % ("; ".join(self.last_errors)[:300] or "无可用地址"))
def bootstrap(self):
"""用现有 SSO 票据强制换一张新的 study.dgjy.net token(旧的可能已过期)。"""
try:
ensure_study_session(self.s, force=True)
except Exception:
pass
self.token = _study_token(self.s)
# ------------------------------------------------------- 新版平台:课程结构
def _flatten_periods(nodes, chapter="", out=None):
out = [] if out is None else out
for n in nodes or []:
if not isinstance(n, dict):
continue
nm = re.sub(r"\s+", " ", n.get("name") or "").strip()
row = dict(n)
row["_chapter"] = chapter
out.append(row)
if isinstance(n.get("children"), list) and n["children"]:
_flatten_periods(n["children"], nm or chapter, out)
return out
def _video_duration(n):
f = n.get("file") or {}
for v in (f.get("videoDuration"), f.get("duration"), f.get("fileDuration"),
n.get("duration"), n.get("videoDuration")):
try:
iv = int(float(v))
if iv > 0:
return iv
except Exception:
continue
return 0
def _period_done(n):
sl = n.get("studyLog") or {}
if sl.get("completeTime"):
return True
try:
return float(sl.get("progress") or 0) >= 1
except Exception:
return False
def _int_or_none(v):
"""平台把次数偶尔给成字符串;取不到就返回 None(页面按「未知」处理)。"""
try:
return int(v)
except (TypeError, ValueError):
return None
# 视频文件类型 -> 播放器通道(来自课程播放页 chunk-3d0ebfea 的 judgeVideoType)
CLOUD_TYPES = {"cloud_video", "middle"} # polyv 云视频:/period/middle/play
MOOC_TYPES = {"mooc_video"} # 网易云课堂:走签名/moc 播放
OTHER_TYPES = {"other"} # 普通 mp4:原生 <video>,ended 即完成
def _norm_name(s):
return re.sub(r"[\s()()【】\[\]·、,,::\-—_/]+", "", s or "").lower()
def _walk_json(o):
if isinstance(o, dict):
yield o
for v in o.values():
yield from _walk_json(v)
elif isinstance(o, list):
for v in o:
yield from _walk_json(v)
def study_resolve_train_id(client, name):
"""课程页没带 ID 时,用「我的课程」/课程列表按名字匹配。"""
target = _norm_name(name)
if not target:
return ""
cands = []
sources = [
("GET", "/course/front/study/studyCenter/myCourse/list/0",
{"pageNum": 1, "pageSize": 200}),
("GET", "/course/front/study/studyCenter/myCourse/list/1",
{"pageNum": 1, "pageSize": 200}),
("GET", "/course/front/trainCourse/list",
{"pageNum": 1, "pageSize": 100, "courseName": name}),
]
for method, path, params in sources:
try:
j = client.call(method, path, params=params)
except SessionExpired:
raise
except Exception:
continue
for rec in _walk_json(j.get("data")):
nm = ""
for k in ("courseName", "trainCourseName", "name", "title"):
if isinstance(rec.get(k), str) and rec.get(k).strip():
nm = rec[k]
break
if not nm:
for k, v in rec.items():
if k.lower().endswith("name") and isinstance(v, str) and v.strip():
nm = v
break
cid = ""
for k in ("trainCourseId", "courseId", "id"):
v = rec.get(k)
if isinstance(v, str) and HEX32.match(v):
cid = v
break
if not cid:
for k, v in rec.items():
if k.lower().endswith("id") and isinstance(v, str) and HEX32.match(v):
cid = v
break
if not nm or not cid:
continue
a = _norm_name(nm)
if a == target:
return cid
if target in a or a in target:
cands.append((cid, nm))
if cands:
return cands[0][0]
return ""
def discover_study(client, train_id, name="", user=""):
info = client.call("GET", "/course/front/trainCourse/getById",
params={"id": train_id}).get("data") or {}
tree = client.call("GET", "/course/front/period/list/tree/studyLog",
params={"courseId": train_id, "isFront": 1}).get("data") or []
if not info and not tree:
raise RuntimeError("新版平台读不到课程结构(trainCourseId=%s)" % train_id)
courseware, feedback, test, containers = [], [], [], []
mooc_nodes, seen = [], set()
def add_video(n, chapter):
pid = str(n.get("id") or "")
if not pid or pid in seen:
return
seen.add(pid)
f = n.get("file") or {}
ftype = str(f.get("fileType") or "")
if ftype in CLOUD_TYPES:
kind = "polyv"
elif ftype in MOOC_TYPES:
kind = "mooc"
else:
kind = "direct"
courseware.append({
"chapter": chapter or "课程",
"title": re.sub(r"\s+", " ", n.get("name") or "") or ("课时 %s" % pid),
"href": "",
"done": _period_done(n),
"duration": _video_duration(n),
"period_id": pid,
"source": "study",
"file_type": ftype,
"kind": kind,
"from_type": str(f.get("fromType") or ""),
"allow_drag": str(n.get("allowVideoDrag") or ""),
})
for n in _flatten_periods(tree):
pt = str(n.get("periodType") or "")
nm = re.sub(r"\s+", " ", n.get("name") or "").strip()
chapter = n.get("_chapter") or ""
if pt == "2":
add_video(n, chapter)
elif pt == "17":
feedback.append({
"title": nm or "培训反馈",
"href": "",
"done": _period_done(n),
"period_id": str(n.get("id") or ""),
"evaluation_id": n.get("evaluationId"),
})
elif pt in ("5", "6"):
tp = n.get("testpaper") or {}
status = str(tp.get("resultStatus") or "")
done = status in ("FINISHED", "REVIEWING", "REVIEWED", "PASS", "UNPASS") \
or _period_done(n)
test.append({
"title": nm or "测验",
"href": "",
"done": done,
"period_id": str(n.get("id") or ""),
"testpaper_id": tp.get("testpaperId"),
"testpaper_result_id": tp.get("testpaperResultId"),
"score": (n.get("studyLog") or {}).get("logPeriodScore"),
"passed_score": tp.get("passedScore"),
"result_status": status,
# 平台的考试次数:allow=总次数,surplus=还剩几次;重考要先看它
"allow_exam_count": _int_or_none(n.get("allowExamCount")),
"surplus_exam_count": _int_or_none(n.get("surplusExamCount")),
# 交卷后是否允许看正确答案/解析(每个测验单独配置)
"exam_after_watch_answer": str(n.get("examAfterWatchAnswer") or ""),
"exam_after_watch_question": str(n.get("examAfterWatchQuestion") or ""),
"complete_limit": str(n.get("completeLimit") or ""),
})
elif pt in ("10", "13") and n.get("kcdm"):
# 10/13 都是「莞易学课程」容器节点,真正的视频课时挂在 kcdm 指向的 MOOC 课程下
mooc_nodes.append((nm or chapter, str(n["kcdm"])))
containers.append({"period_id": str(n.get("id") or ""), "name": nm or chapter,
"kcdm": str(n["kcdm"]), "done": _period_done(n)})
# 莞易学课程节点(periodType=10):真正的视频课时挂在 kcdm 指向的 MOOC 课程下
for mooc_name, mooc_id in mooc_nodes:
try:
sub = client.call("GET", "/course/front/period/list/tree/studyLog",
params={"courseId": mooc_id, "isFront": 1}).get("data") or []
except Exception:
continue
for v in _flatten_periods(sub, mooc_name):
if str(v.get("periodType") or "") == "2":
add_video(v, mooc_name)
st = info.get("study") or {}
if st.get("periodTotal"):
progress = "%s/%s 课时" % (st.get("periodStudy") or 0, st.get("periodTotal"))
else:
progress = "%d/%d 课时" % (len([c for c in courseware if c["done"]]), len(courseware))
bits = []
if info.get("id"):
bits.append("trainCourseId=%s" % info["id"])
if info.get("isJoin") is not None:
bits.append("已加入=%s" % ("是" if str(info.get("isJoin")) == "1" else "否"))
if info.get("needJoin") is not None:
bits.append("需加入=%s" % ("是" if str(info.get("needJoin")) == "1" else "否"))
if info.get("completeRule") is not None:
bits.append("完成规则=%s" % info.get("completeRule"))
return {
"user": user,
"course": info.get("courseName") or name or "?",
"progress": progress,
"info": " | ".join(bits) or "?",
"courseware": courseware,
"feedback": feedback,
"test": test,
"containers": containers,
"train_id": train_id,
# 未加入培训时 study/log 会被静默丢弃,调用方要先补 join
"is_join": str(info.get("isJoin")) == "1",
"need_join": str(info.get("needJoin")) == "1",
# 浏览器「刷新课程页就能加入」靠的就是它:/mod/school/wanpeng/syncSelf?courseCode=<courseCode>
"course_code": str(info.get("courseCode") or ""),
}
# ------------------------------------------------------- 新版平台:动作
def study_period_log(client, period_id, done, seconds=0):
"""播放器两个包:播放中 {isComplete:0, lastProgressTime},播完 {isComplete:1}(不带时长)。"""
body = {"periodId": int(period_id), "isComplete": 1 if done else 0}
if not done:
body["lastProgressTime"] = str(int(seconds or 0))
return client.call("POST", "/course/front/period/study/log", json_body=body)
def study_log_detail(client, period_id, stime, etime):
"""学时台账(播放器暂停/离开时发):stime/etime 是这次学习的起止时间。"""
fmt = "%Y-%m-%d %H:%M:%S"
return client.call("POST", "/course/front/study/log/detail/save",
json_body={"stime": time.strftime(fmt, time.localtime(stime)),
"etime": time.strftime(fmt, time.localtime(etime)),
"periodId": int(period_id)})
def study_is_done(client, period_id):
j = client.call("GET", "/course/front/period/getById", params={"periodId": int(period_id)})
sl = (j.get("data") or {}).get("studyLog") or {}
if sl.get("completeTime"):
return True
try:
return float(sl.get("progress") or 0) >= 1
except Exception:
return False
def study_cal_final_score(client, course_id):
"""按学习日志重算课程的完成计数(PUT)。
并发上报 period/study/log 会丢增量:实测 14 个课时 completeTime 全有,但课程上的
periodStudy 只记到 11,培训那边的「课程」节点就永远不完成。这个接口能把计数补齐。
"""
return client.call("PUT", "/course/front/calFinalScore?courseId=%s" % course_id)
def study_complete_container(client, ct):
"""培训里的「课程」是 periodType 10/13 的容器节点:子课时学完它不会自动完成。
前端从不给这种节点发 study/log(播视频只动子课时),但接口本身没按节点类型拦,
所以这里补一发 isComplete=1,写完回读 completeTime 确认真生效。
"""
pid = ct["period_id"]
try:
client.call("POST", "/course/front/period/study/log",
json_body={"periodId": int(pid), "isComplete": 1})
except Exception as e:
return False, str(e)[:100]
time.sleep(0.5)
try:
j = client.call("GET", "/course/front/period/getById", params={"periodId": int(pid)})
sl = (j.get("data") or {}).get("studyLog") or {}
if sl.get("completeTime"):
return True, "已标记完成(%s)" % sl["completeTime"]
return False, "接口回 200 但 completeTime 仍为空"
except Exception as e:
return False, "回读失败:%s" % str(e)[:80]
def study_refresh_page(s, d):
"""照浏览器的做法把课程页重新走一遍(jspx 课程页 → 带新 token 的 trainDetails)。
新版平台的 isJoin 有时候要等页面真正被打开一次才落库:直接调 student/join 被拒
(「该课程无法直接加入,请联系管理员」),但刷新一次页面它就变 1 了。
所以加入失败后补一次页面访问,再复查。
"""
plan, project = d.get("plan_id") or "", d.get("project_id") or ""
if not (plan and project):
return ""
url = "%s/train/trainee/courseStudyMain.action?planId=%s&projectId=%s&trainRole=0" % (
JSPX, plan, project)
try:
r = s.get(url, timeout=40, allow_redirects=True)
return r.url or ""
except Exception:
return ""
def study_mod_get(client, path, params=None):
"""调 /mod/ 这组接口(不在 /train 下,也不带 Bearer,只认 cookie 会话)。
浏览器课程页刷新时发的就是这组:/mod/student/currentInfo、/mod/school/wanpeng/syncSelf、
/mod/gjt/roleList。用 StudyClient 的会话(cookie)直接 GET 就能通,
而 /train/* 必须额外带 Authorization: Bearer <Middle-Token>。
"""
h = {"Accept": "application/json, text/plain, */*",
"Referer": STUDY + "/training/trainDetails"}
r = client.s.get(STUDY + path, params=params, headers=h, timeout=40)
try:
return r.json()
except Exception:
return {"resultCode": r.status_code, "resultMsg": (r.text or "")[:200]}
def study_sync_self(client, course_code):
"""把当前账号同步进课程 —— 浏览器刷新课程页时真正「加入培训」的那一步。
抓包实录(常用公文写作):
getById -> isJoin=0 needJoin=1
GET /mod/school/wanpeng/syncSelf?courseCode=<code> (耗时约 2s)
getById -> isJoin=1
前端在 isJoin != 1 时就是发这一发把学员同步进课程;
isFreeJoin=0 的慕课课不走 student/join,只能靠它落库。
返回 (ok, msg)。
"""
j = study_mod_get(client, "/mod/school/wanpeng/syncSelf", {"courseCode": course_code})
res = j.get("result") or {}
code = str(j.get("resultCode") or res.get("code") or "")
return code == "200", str(res.get("msg") or j.get("resultMsg") or code or "?")
def study_sync_identity(client):
"""页面上紧随 syncSelf 的身份拉取(/mod/gjt/roleList),个别课程同步后要靠它才认人。
返回 True/False,失败无所谓,只当补一刀。
"""
try:
j = study_mod_get(client, "/mod/gjt/roleList")
return bool(j.get("result")) or str(j.get("resultCode") or "") == "200"
except Exception:
return False
def ensure_study_join(s, client, d):
"""确保这门培训已加入;返回 (ok, 说明)。
顺序(照浏览器课程页的真实行为):
当前状态 → student/join → 被拒就 syncSelf(courseCode) 轮询复查
→ 再不行就刷新一次 jspx 课程页复查。
"""
tid = d.get("train_id") or ""
code = d.get("course_code") or ""
def joined():
j = client.call("GET", "/course/front/trainCourse/getById", params={"id": tid})
return str((j.get("data") or {}).get("isJoin")) == "1"
if d.get("is_join") or joined():
return True, "已加入"
ok, msg = study_join_course(client, tid)
if ok:
return True, "加入成功"
# 浏览器刷新课程页走的就是这条:syncSelf 落库,再重新拉 getById
if code:
sync_ok, sync_msg = study_sync_self(client, code)
for i in range(8):
time.sleep(0.6 if i == 0 else 1.5)
try:
if joined():
return True, "同步课程页后已加入(student/join 被拒:%s;syncSelf=%s)" % (
msg or "-", sync_msg or "-")
except SessionExpired:
raise
except Exception:
pass
if i == 2:
# 页面在 syncSelf 之后还会拉一次身份,个别课程靠它才认人
study_sync_identity(client)
note = "syncSelf:%s" % ("成功" if sync_ok else "失败(%s)" % (sync_msg or "-"))
else:
note = "课程没有 courseCode,无法走同步接口"
study_refresh_page(s, d)
try:
if joined():
return True, "刷新课程页后已加入(student/join 被拒:%s;%s)" % (msg or "-", note)
except SessionExpired:
raise
except Exception:
pass
return False, "%s;%s" % (msg or "未加入", note)
def study_join_course(client, train_id):
"""加入培训(POST /course/front/student/join?courseId=...)。
课程 needJoin=1 且 isJoin=0 时必须先加入,否则 study/log 照样回 200,
但学习记录不会计数 —— 现象就是「每个课时都报成功、复查全是未完成」。
isFreeJoin=0 的课程服务端会直接拒:「该课程无法直接加入,请联系管理员」。
慕课课(isFreeJoin=0)别指望这个接口,要改用 study_sync_self —— 见 ensure_study_join。
返回 (ok, msg)。
"""
j = client.call("POST", "/course/front/student/join", params={"courseId": train_id})
return str(j.get("code")) == "200", str(j.get("msg") or j.get("message") or "")
def study_complete_period(cookies, page_url, item):
"""把一个视频课时报到 complete=1,并以服务端 completeTime 为准。"""
s = new_session(cookies)
c = StudyClient(s, page_url)
pid, dur = item["period_id"], int(item.get("duration") or 0)
last = ""
for attempt in range(1, 4):
try:
study_period_log(c, pid, False, 0)
time.sleep(0.35)
now = time.time()
try: # 学时台账:播放器暂停/离开时都会发,缺它学时可能不记
study_log_detail(c, pid, now - max(60, dur), now)
except Exception:
pass
j = study_period_log(c, pid, True, dur)
last = "code=%s %s" % (j.get("code"), str(j.get("msg") or "")[:60])
except SessionExpired:
raise
except Exception as e:
last = str(e)[:100]
time.sleep(1.2)
try:
if study_is_done(c, pid):
return True, "ok"
except Exception as e:
last = str(e)[:100]
time.sleep(1.0)
return False, last
def study_submit_feedback(client, fb, key, train_id):
eid = fb.get("evaluation_id")
if not eid:
run_log(key, "反馈表:该节点没有 evaluationId,跳过")
return
j = client.call("GET", "/course/front/evaluation/result/getDetail",
params={"evaluationId": eid})
data = j.get("data") or {}
if data.get("resultId"):
run_log(key, "反馈表:已提交过(resultId=%s),跳过" % data["resultId"])
return
items = data.get("items") or []
if not items:
run_log(key, "反馈表:没有题目(evaluationId=%s),跳过" % eid)
return
score_flag = str(data.get("scoreFlag")) != "0"
stars = data.get("starCount") or 10
answers = [{"itemId": it.get("itemId"),
"answer": stars if score_flag else ""} for it in items]
payload = {"periodId": int(fb["period_id"]), "evaluationId": int(eid),
"items": answers, "trainCourseId": train_id}
r = client.call("POST", "/course/front/evaluation/result", json_body=payload)
if r.get("code") == 200:
run_log(key, "反馈表:%d 题,提交成功(%s分)" % (len(answers), stars))
else:
run_log(key, "反馈表:提交失败 code=%s %s" % (r.get("code"), str(r.get("msg") or "")[:80]))
# ------------------------------------------------------------------ study engine
RUNS = {}
RUNS_LOCK = threading.Lock()
def run_log(key, msg):
line = "%s %s" % (time.strftime("%Y-%m-%d %H:%M:%S"), msg)
print("%s [%s]" % (line, key), flush=True) # 同步打到控制台,docker compose logs -f 也能看
with RUNS_LOCK:
st = RUNS.get(key)
if st is not None:
st.setdefault("log", []).append(line)
if len(st["log"]) > 1000:
del st["log"][:300]
st["updated"] = time.time()
store.run_log_add(key, line) # 落 SQLite,重启后 /api/log/history 还能查
def set_state(key, **kw):
with RUNS_LOCK:
st = RUNS.setdefault(key, {})
st.update(kw)
st["updated"] = time.time()
store.run_state_set(key, RUNS.get(key) or {})
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":
run_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)
run_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:
run_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)))
run_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):
"""Run each selected course in sequence for one account.
enable_quiz 默认 False —— 测验答题正确率无法保证,可能因不及格卡住「完成」,故默认跳过。
enable_feedback 默认 True —— 反馈表是主观问卷,没有对错,提交即完成。
mode fast=只调 study/log 接口;normal=开浏览器真播(parallel 路并发)。
"""
try:
set_state(key, status="discovering", log=[], done=0, total=0, started=time.time())
failed = []
run_log(key, "=== 开始:%s(共 %d 门课程)" % (username, len(courses)))
run_log(key, "设置:自动交测验=%s,自动交反馈表=%s,学习方式=%s" %
("开" if enable_quiz else "关", "开" if enable_feedback else "关",
"快速(只调接口)" if mode != "normal" else "常规(真播,%d 路并发)" % parallel))
for ci, c in enumerate(courses, 1):
with RUNS_LOCK:
if (RUNS.get(key) or {}).get("status") == "stopped":
run_log(key, "已按要求停止")
return
run_log(key, "--- [%d/%d] %s" % (ci, len(courses), c.get("name") or c["plan_id"]))
set_state(key, course=c.get("name", ""))
try:
run_one_course(key, username, cookies, c["plan_id"], c["project_id"], workers, cap,
quiz_key, enable_quiz, enable_feedback, c.get("name") or "", password,
mode, parallel)
except Exception as e:
failed.append(c.get("name") or c["plan_id"])
run_log(key, "课程处理异常:%s" % e)
set_state(key, status="finished", finished_at=time.time())
if failed:
run_log(key, "=== 结束:%d/%d 门课程异常(%s)" % (len(failed), len(courses), "、".join(failed)[:120]))
else:
run_log(key, "=== 全部完成")
except Exception as e:
run_log(key, "!! 运行异常:%s" % e)
set_state(key, status="error", error=str(e)[:300])
def run_one_course(key, username, cookies, plan, project, workers, cap, quiz_key,
enable_quiz=False, enable_feedback=True, name="", password="",
mode="fast", parallel=2):
s = new_session(cookies)
try:
d = discover(s, plan, project, name)
except SessionExpired:
# 先试无密码续期(CASTGC 通常还有效),不行再用浏览器里存的账号密码重登
run_log(key, "会话失效,尝试用 SSO 票据续期…")
renewed = ensure_jspx_session(s)
d = None
if renewed:
try:
d = discover(s, plan, project, name)
except SessionExpired:
d = None
if d is None:
if not password:
raise
run_log(key, "SSO 票据已过期,用账号密码重新登录…")
s, _ = login(username, password)
d = discover(s, plan, project, name)
# 新平台可能刚在 discover 里换到 study.dgjy.net 的会话(Middle-Token),存回去给后续课程用
set_state(key, course=d["course"], progress=d["progress"], user=d["user"],
cookies=dump_cookies(s))
run_log(key, "课程:%s | 进度 %s | %s" % (d["course"], d["progress"], d["info"]))
run_log(key, "平台:%s | 活动:课件 %d、反馈表 %d、测验 %d" %
("新版 study.dgjy.net" if d.get("platform") == "study" else "jspx",
len(d["courseware"]), len(d["feedback"]), len(d["test"])))
if d.get("platform") == "study":
run_log(key, "新版课程 ID:%s%s" %
(d.get("train_id") or "?",
(" | 页面:" + (d.get("page_url") or "")[:110]) if d.get("page_url") else ""))
if not d["courseware"] and not d["feedback"] and not d["test"]:
raise RuntimeError("课程里没有任何可识别的活动(课件/反馈表/测验都是 0)")
if d.get("platform") == "study":
run_study_course(key, s, d, workers, cap, enable_quiz, enable_feedback, mode, parallel)
return
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
run_log(key, "反馈表已完成,跳过")
continue
try:
submit_feedback(s, fb["href"], key)
except Exception as e:
run_log(key, "反馈表异常:%s" % e)
else:
for fb in d["feedback"]:
run_log(key, "反馈表:已按设置跳过(%s)" % ("已完成" if fb["done"] else "未完成"))
if enable_quiz:
for t in d["test"]:
if t["done"]:
run_log(key, "测验已完成,跳过")
continue
try:
submit_quiz(s, t["href"], key, quiz_key)
except Exception as e:
run_log(key, "测验异常:%s" % e)
else:
for t in d["test"]:
if t["done"]:
run_log(key, "测验:已完成,跳过")
else:
run_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:
run_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]
run_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":
run_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)
run_log(key, "第 %d 轮 %dm%02ds 完成 %d/%d" % (el // 60 + 1, el // 60, el % 60, dc, len(objs)))
for o in newly:
run_log(key, " ✓ %s" % o.title[:40])
time.sleep(60)
else:
run_log(key, "所有课件均已完成")
try:
d2 = discover(s, plan, project, name)
set_state(key, progress=d2["progress"], info=d2["info"], cookies=dump_cookies(s))
run_log(key, "课程结束:%s | %s" % (d2["progress"], d2["info"]))
except Exception as e:
run_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:
run_log(key, "加入培训:%s" % why)
d["is_join"] = True
else:
run_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)
run_log(key, "试报 1 个课时:%s" % ("生效" if good else "未生效(%s)" % last))
if not good:
run_log(key, "该账号对这门课没有写入权限,本次跳过。"
"需要课程管理员把这门课分配到账号下(jspx 在学 ≠ 新版平台已加入)")
return
else:
return
# ---- 反馈表 ----
if enable_feedback:
for fb in d["feedback"]:
if fb["done"]:
run_log(key, "反馈表已完成,跳过")
continue
try:
study_submit_feedback(client, fb, key, train_id)
except Exception as e:
run_log(key, "反馈表异常:%s" % e)
else:
for fb in d["feedback"]:
run_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):
run_log(key, "测验:已作答但不及格(%s/%s),需要在页面上重考" % (score, passed))
continue
except Exception:
pass
run_log(key, "测验:已完成,跳过")
elif enable_quiz:
run_log(key, "测验:新版平台不代答(%s 需在页面上手动作答)" % t["title"][:30])
else:
run_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:
run_log(key, "没有视频课时")
return
kinds = {}
for c in objs:
kinds[c.get("kind") or "?"] = kinds.get(c.get("kind") or "?", 0) + 1
if kinds:
run_log(key, "视频类型:%s" % "、".join("%s %d 个" % (k, v) for k, v in kinds.items()))
if not todo:
run_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)
run_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)
run_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"])
run_log(key, " 已重算子课程完成计数(kcdm=%s)" % ct["kcdm"][:8])
except Exception as e:
run_log(key, " 重算子课程完成计数失败:%s" % str(e)[:80])
ok, msg = study_complete_container(client, ct)
run_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:
run_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:
run_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
run_log(key, " ✓ %s(%ss)" % (it["title"][:40], it.get("duration") or 0))
else:
run_log(key, " × %s 上报未生效:%s" % (it["title"][:40], msg))
set_state(key, done=ok)
except TimeoutError:
run_log(key, "已到时间上限(%d 分钟),剩余课时留待下次" % cap)
except Exception as e:
run_log(key, "批量提交中断:%s" % e)
finally:
ex.shutdown(wait=False, cancel_futures=True)
if soft:
run_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:
run_log(key, " × %s 播放失败:%s" % (it["title"][:30], str(e)[:100]))
continue
if good:
ok += 1
run_log(key, " ✓ %s(播放到结尾)" % it["title"][:40])
else:
run_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:
run_log(key, "缺少 playwright,常规学习不可用:%s" % e)
return done0
parallel = max(1, min(int(parallel or 1), 5, len(todo)))
run_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":
run_log(key, "已按要求停止")
break
wave, pending = pending[:parallel], pending[parallel:]
run_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
run_log(key, " ✓ %s(真播到结尾)" % it["title"][:40])
else:
run_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: # 一直被顺序校验弹回,别再空转
run_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
run_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:
run_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:
run_log(key, " 播放异常:%s" % str(e)[:120])
return False
if not ok:
run_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:
run_log(key, "同步莞易学课程完成状态:%s" % study_sync_gyx(client, d["train_id"]))
except Exception as e:
run_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))
run_log(key, "课程结束:%s | %s" % (d2["progress"], d2["info"]))
# ------------------------------------------------------- 抓包(排查平台行为)
CAPTURES = {}
def capture_worker(key, cookies, minutes=10):
"""开一个有头浏览器,把这期间所有 /train/** 请求和响应记下来。
用于搞不清「某个状态到底由哪一发请求写成」:手动走一遍(点课程 → 播几秒 → 交评价),
再看抓到的接口和响应,就能定位该自动补哪一步。
"""
try:
from playwright.sync_api import sync_playwright
except Exception as e:
run_log(key, "抓包需要 playwright:%s" % e)
return
recs = CAPTURES.setdefault(key, [])
run_log(key, "抓包开始:浏览器里手动走一遍(点课程 → 播几秒 → 交评价),"
"完事点「停止抓包」,最多 %d 分钟" % minutes)
def on_resp(r):
try:
if r.request.resource_type not in ("xhr", "fetch"):
return
ct = (r.headers or {}).get("content-type", "")
body = ""
if "json" in ct:
try:
body = r.text()[:4000]
except Exception:
body = ""
recs.append({"t": time.strftime("%H:%M:%S"), "method": r.request.method,
"url": r.url, "req": (r.request.post_data or "")[:1000],
"status": r.status, "resp": body})
u = urlsplit(r.url)
run_log(key, " · %s %s%s → %s" % (r.request.method, u.netloc, u.path, r.status))
except Exception:
pass
try:
with sync_playwright() as p:
b = p.chromium.launch(headless=False)
ctx = b.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
# 平台点课程/视频是 window.open 新开标签,必须把新页面也挂上监听,否则只抓到第一页
hooked = set()
def hook(page):
if id(page) in hooked:
return
hooked.add(id(page))
page.on("response", on_resp)
ctx.on("page", hook)
pg = ctx.new_page()
hook(pg)
# /training/ 根路径没有路由,会落到 404;直接开「我的培训」列表页
pg.goto(STUDY + "/training/myTrain", wait_until="domcontentloaded", timeout=60000)
t0 = time.time()
while time.time() - t0 < minutes * 60:
if (RUNS.get(key) or {}).get("capture_stop"):
break
pg.wait_for_timeout(1000)
except Exception as e:
run_log(key, "抓包异常:%s" % str(e)[:150])
finally:
path = ""
if recs:
try:
path = os.path.join(store.DATA_DIR,
"capture-%s-%s.json" % (key, time.strftime("%Y%m%d-%H%M%S")))
with open(path, "w", encoding="utf-8") as f:
json.dump(recs, f, ensure_ascii=False, indent=1)
except Exception as e:
run_log(key, "抓包结果保存失败:%s" % e)
set_state(key, capturing=False, capture_stop=False, capture_file=path)
run_log(key, "抓包结束:%d 条请求%s" % (len(recs), (",已存 " + path) if path else ""))
# ------------------------------------------------------------------------ models
class LoginIn(BaseModel):
username: str = ""
password: str = ""
account_id: str = "" # 只给账号 id 时,用户名/密码从 SQLite 里取,密码不出库
class AccountsIn(BaseModel):
accounts: list = []
class AccountIn(BaseModel):
id: str = ""
username: str = ""
password: str | None = None # 不传 = 保留库里已有的密码
note: str = ""
user: str = ""
cookies: list = []
courses: list = []
selected: list = []
status: str = "idle"
lastError: str = ""
lastErrorAt: float = 0
optQuiz: bool = False
optFeedback: bool = True
loginAt: float = 0
created: float = 0
class CookiesIn(BaseModel):
cookies: list
class DiscoverIn(BaseModel):
cookies: list
plan_id: str
project_id: str
name: str = ""
class StudyIn(BaseModel):
key: str
username: str = ""
password: str = "" # 仅用于会话失效时自动重登,不落盘
cookies: list = []
courses: list = [] # [{name, plan_id, project_id}]
plan_id: str = ""
project_id: str = ""
workers: int = 40
cap_minutes: int = 90
quiz_key: dict | None = None
enable_quiz: bool = False # 默认关闭:答题正确率无法保证
enable_feedback: bool = True # 反馈表无对错,默认开启
class StopIn(BaseModel):
key: str
class CaptureIn(BaseModel):
key: str
cookies: list = []
minutes: int = 10 # 抓包最长分钟数(1~30)
class QuizIn(BaseModel):
cookies: list = []
plan_id: str = ""
project_id: str = ""
name: str = ""
# retake=True:交卷后点「再考一次」,让平台生成新答卷(消耗一次机会)
retake: bool = False
period_id: str = "" # 指定要重考/查看的课时(不传就按原逻辑挑第一个)
class QuizSubmitIn(BaseModel):
cookies: list = []
plan_id: str = ""
object_id: str = ""
platform: str = "jspx"
result_id: str = ""
answers: dict = {} # {"1": "A", "2": "A#C"}
class QuizAnswerIn(BaseModel):
questions: list = [] # [{no,type,title,options:[{key,text}]}]
prompt: str = ""
class LlmCfgIn(BaseModel):
base_url: str = ""
api_key: str = ""
model: str = ""
temperature: float = 0.2
timeout: int = 120
prompt: str = ""
study_mode: str = "" # "" = 不改;fast / normal
study_parallel: int = 0 # 0 = 不改;1~5
class LlmTestIn(BaseModel):
base_url: str = ""
api_key: str = ""
model: str = ""
timeout: int = 60
app = FastAPI(title="多账号学习控制台")
APP_ID = "dgjy-study-console"
# 远程访问令牌:设置了 AUTH_TOKEN 后,页面/接口都要带 ?token=<值>(浏览器会记住 cookie)。
# 不设置 = 不校验(本机使用,或前面已经有反代做鉴权)。/api/health 永远放行,供健康检查用。
AUTH_TOKEN = (os.environ.get("AUTH_TOKEN") or "").strip()
AUTH_COOKIE = "dgjy_auth"
_AUTH_PAGE = """<!doctype html><meta charset="utf-8"><title>需要访问令牌</title>
<style>body{font:14px/1.7 system-ui,"Microsoft YaHei",sans-serif;background:#f5f6f8;color:#1f2430;
display:flex;min-height:100vh;align-items:center;justify-content:center;margin:0}
div{background:#fff;border:1px solid #e3e6ec;border-radius:12px;padding:26px 30px;max-width:520px}
h3{margin:0 0 10px}input{width:100%;padding:8px 10px;border:1px solid #d6dae2;border-radius:8px;
font-size:14px;margin-top:8px}button{margin-top:14px;padding:8px 18px;border:0;border-radius:8px;
background:#2b6cf6;color:#fff;font-size:14px;cursor:pointer}code{background:#f1f3f7;padding:1px 5px;border-radius:4px}</style>
<div><h3>需要访问令牌</h3>
<p>用 <code>/?token=&lt;AUTH_TOKEN&gt;</code> 打开本页,或把令牌填到下面。</p>
<form method="get" action="/"><input name="token" placeholder="访问令牌" autofocus><button>进入</button></form></div>
"""
@app.middleware("http")
async def auth_gate(request: Request, call_next):
if not AUTH_TOKEN:
return await call_next(request)
tok = (request.query_params.get("token")
or request.cookies.get(AUTH_COOKIE)
or request.headers.get("x-auth-token") or "")
if tok == AUTH_TOKEN:
resp = await call_next(request)
if request.query_params.get("token"):
# 带上 ?token= 访问后下发 cookie,页面里所有同源 fetch 会自动带上
resp.set_cookie(AUTH_COOKIE, AUTH_TOKEN, max_age=30 * 24 * 3600,
httponly=True, samesite="lax")
return resp
if request.url.path == "/api/health":
return await call_next(request)
if request.url.path.startswith("/api/"):
return JSONResponse({"ok": False, "code": "unauthorized",
"error": "缺少访问令牌(AUTH_TOKEN)"}, status_code=401)
return HTMLResponse(_AUTH_PAGE, status_code=401)
def hydrate_runs():
"""进程重启后把上次的运行状态读回来;上次没跑完的标记成「已中断」。"""
try:
saved = store.run_state_all()
except Exception:
return
with RUNS_LOCK:
for k, st in saved.items():
if k in RUNS:
continue
st = dict(st)
st.pop("log", None)
if st.get("status") in ("discovering", "studying"):
st["status"] = "interrupted"
st["error"] = "服务重启,上次运行已中断"
RUNS[k] = st
@app.get("/", response_class=HTMLResponse)
def index():
# 页面必须禁止缓存:后端重启或换端口后,旧标签页会拿着缓存里的 HTML 继续跑,
# 页面照常显示、所有接口全 ERR_CONNECTION_REFUSED,看起来就像"登录失败"。
html = open(os.path.join(HERE, "static", "index.html"), encoding="utf-8").read()
return HTMLResponse(html, headers={
"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
"Pragma": "no-cache",
"Expires": "0",
})
@app.get("/api/health")
def api_health():
return {"ok": True, "app": APP_ID, "pid": os.getpid(),
"port": int(os.environ.get("PORT", "8848")), "runs": len(RUNS),
"db": store.db_info(), "auth": bool(AUTH_TOKEN)}
# ------------------------------------------------------------------ settings
SETTINGS_NAME = "llm"
DEFAULT_LLM = {
"base_url": "https://api.openai.com/v1",
"api_key": "",
"model": "gpt-4o-mini",
"temperature": 0.2,
"timeout": 120,
"prompt": "",
}
# 学习方式:fast = 只调 study/log 接口,不开浏览器;normal = 真播视频(可多路并发)
DEFAULT_STUDY = {
"mode": "fast",
"parallel": 2,
}
DEFAULT_ANSWER_PROMPT = (
"你是考试答题助手。根据题干和选项判断正确答案。\n"
"单选题返回一个选项字母;多选题返回多个字母,用 # 连接(如 A#C);判断题只返回 A 或 B(A=正确,B=错误)。\n"
"只输出 JSON:{\"answers\":{\"题号\":\"答案\"}},不要输出解释或多余文字。"
)
TYPE_LABEL = {"01": "单选", "02": "多选", "03": "判断", "04": "问答"}
def load_settings():
data = store.get_setting(SETTINGS_NAME) or {}
llm = dict(DEFAULT_LLM)
study = dict(DEFAULT_STUDY)
if isinstance(data, dict):
llm.update({k: v for k, v in data.items() if k in DEFAULT_LLM and v is not None})
st = data.get("study")
if isinstance(st, dict):
study.update({k: v for k, v in st.items() if v is not None})
study["mode"] = study.get("mode") if study.get("mode") in ("fast", "normal") else "fast"
try:
study["parallel"] = max(1, min(int(study.get("parallel") or 2), 5))
except (TypeError, ValueError):
study["parallel"] = 2
return {"llm": llm, "study": study}
def save_settings(data):
store.set_setting(SETTINGS_NAME, {"llm": data.get("llm") or {},
"study": data.get("study") or {}})
@app.get("/api/settings")
def api_get_settings():
cfg = load_settings()
llm = cfg["llm"]
key = llm.get("api_key") or ""
out = dict(llm)
out["api_key"] = ""
out["has_key"] = bool(key)
out["key_hint"] = (key[:4] + "…" + key[-4:]) if len(key) > 10 else ("已设置" if key else "")
return {"ok": True, "llm": out, "study": cfg["study"]}
@app.post("/api/settings")
def api_set_settings(b: LlmCfgIn):
cur = load_settings()
llm = cur["llm"]
if b.base_url.strip():
llm["base_url"] = b.base_url.strip()
if b.api_key.strip(): # 留空 = 保留原 key,前端不回显明文
llm["api_key"] = b.api_key.strip()
if b.model.strip():
llm["model"] = b.model.strip()
llm["temperature"] = max(0.0, min(2.0, float(b.temperature or 0)))
llm["timeout"] = max(10, min(600, int(b.timeout or 120)))
llm["prompt"] = b.prompt or ""
if b.study_mode in ("fast", "normal"): # 空串 = 不改
cur["study"]["mode"] = b.study_mode
if b.study_parallel: # 0 = 不改
cur["study"]["parallel"] = max(1, min(int(b.study_parallel), 5))
save_settings(cur)
return {"ok": True, "study": cur["study"]}
def llm_chat(messages, cfg, max_tokens=None, json_mode=True):
"""OpenAI 兼容的 /chat/completions。不支持 response_format 的服务端会自动去掉重试一次。"""
base = (cfg.get("base_url") or "").rstrip("/")
if not base:
raise RuntimeError("base_url 为空")
url = base + "/chat/completions"
body = {"model": cfg.get("model") or "gpt-4o-mini", "messages": messages,
"temperature": float(cfg.get("temperature") or 0.2)}
if max_tokens:
body["max_tokens"] = int(max_tokens)
if json_mode:
body["response_format"] = {"type": "json_object"}
headers = {"Authorization": "Bearer " + (cfg.get("api_key") or ""),
"Content-Type": "application/json"}
timeout = int(cfg.get("timeout") or 120)
r = requests.post(url, json=body, headers=headers, timeout=timeout)
if r.status_code == 400 and json_mode:
body.pop("response_format", None)
r = requests.post(url, json=body, headers=headers, timeout=timeout)
if r.status_code >= 400:
body = (r.text or "").strip()
if body.startswith("<"):
body = "网关返回 HTML 错误页(HTTP %s)" % r.status_code
raise RuntimeError("LLM %s: %s" % (r.status_code, body[:200]))
j = r.json()
return (((j.get("choices") or [{}])[0].get("message") or {}).get("content") or "")
@app.post("/api/llm/test")
def api_llm_test(b: LlmTestIn):
cfg = dict(load_settings()["llm"])
if b.base_url.strip():
cfg["base_url"] = b.base_url.strip()
if b.model.strip():
cfg["model"] = b.model.strip()
if b.api_key.strip():
cfg["api_key"] = b.api_key.strip()
cfg["timeout"] = max(10, min(120, int(b.timeout or 60)))
if not cfg.get("api_key"):
return {"ok": False, "error": "没有 api_key"}
t0 = time.time()
try:
reply = llm_chat([{"role": "user", "content": "只回复两个字符:ok"}], cfg,
max_tokens=8, json_mode=False)
return {"ok": True, "ms": int((time.time() - t0) * 1000), "reply": (reply or "")[:80]}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
# ------------------------------------------------------------------ quiz
STUDY_QMAP = {"SINGLECHOICE": "01", "MULTIPLECHOICE": "02", "JUDGMENT": "03",
"TRUEFALSE": "03", "FILLIN": "04", "SHORTANSWER": "04", "ESSAY": "04"}
# 判断题在新版平台里没有 metasData 选项(只有一条 sort/content/option 全 null 的占位),
# 真正可选项是前端写死的「正确 / 错误」,交卷时 replay 用 0/1 下标表示。
# 这里补出 A/B 两个伪选项,页面才有得选,交卷也能把字母换成正确下标。
JUDGE_OPTIONS = [{"key": "A", "text": "正确"}, {"key": "B", "text": "错误"}]
_ANSWER_YES = {"A", "正确", "对", "是", "√", "T", "TRUE", "Y", "YES"}
_ANSWER_NO = {"B", "错误", "错", "否", "×", "X", "F", "FALSE", "N", "NO"}
def answer_tokens(raw, options=None, qtype="01"):
"""把答案文本拆成选项字母列表。
判断题允许模型/用户写「正确 / 错误 / 对 / 错 / √ / × / T / F」,
统一收敛到 A(正确)/ B(错误);有选项时只保留合法字母;
问答题没有选项,原样保留文本。
"""
keys = [o.get("key") for o in (options or []) if o.get("key")]
parts = [p for p in re.split(r"[#,\s、/]+", str(raw or "").strip().upper()) if p]
out, seen = [], set()
for p in parts:
if p in keys:
tok = p
elif qtype == "03":
tok = "A" if p in _ANSWER_YES else ("B" if p in _ANSWER_NO else "")
elif not keys:
tok = p
else:
tok = ""
if tok and tok not in seen:
seen.add(tok)
out.append(tok)
return out
def _plain(s):
return re.sub(r"\s+", " ", BeautifulSoup(s or "", "lxml").get_text(" ", strip=True)).strip()
def study_parse_paper(data):
"""把 /questions/front/testpaper/getById 的 data.questions 拍平成题目列表。
结构:questions = [{type, data: [{seq, questionType, stem, metasData:[{option,content}],
itemResultId, replay}]}]
replay 是已保存答案的**0 基选项下标**数组,例如 ["2"] = 第 3 个选项。
"""
out = []
for grp in data.get("questions") or []:
for it in grp.get("data") or []:
qtype = STUDY_QMAP.get(str(it.get("questionType") or ""), "01")
opts = []
for m in it.get("metasData") or []:
key = str(m.get("option") or "").strip().upper()
if key:
opts.append({"key": key, "text": _plain(m.get("content"))})
opts.sort(key=lambda o: o["key"])
# 判断题:平台不给选项,补上 正确/错误,页面才看得见、交卷才好映射 0/1
if not opts and qtype == "03":
opts = [dict(o) for o in JUDGE_OPTIONS]
picked = ""
rp = it.get("replay")
if rp:
try:
idx = json.loads(rp) if isinstance(rp, str) else rp
picked = "#".join(opts[int(i)]["key"] for i in idx
if 0 <= int(i) < len(opts))
except Exception:
picked = ""
out.append({"no": str(int(it.get("seq") or 0) + 1),
"type": qtype,
"title": _plain(it.get("stem"))[:600],
"options": opts,
"item_id": it.get("itemResultId"),
"picked": picked})
return out
def _idx_letters(raw, opts):
"""review 接口里的 replay/answer 是 JSON 字符串形式的 0 基下标数组。"""
if raw in (None, "", "null"):
return ""
try:
idx = json.loads(raw) if isinstance(raw, str) else raw
except Exception:
return ""
if idx is None:
return ""
if not isinstance(idx, (list, tuple)):
idx = [idx]
out = []
for i in idx:
try:
n = int(i)
except (TypeError, ValueError):
continue
if 0 <= n < len(opts):
out.append(opts[n]["key"])
return "#".join(out)
def _answer_text(raw):
"""填空/问答题的标准答案:可能是 JSON 数组 [{content:…}],也可能就是文本。"""
if raw in (None, ""):
return ""
v = raw
if isinstance(v, str):
try:
v = json.loads(v)
except Exception:
return _plain(v)
if isinstance(v, list):
parts = []
for x in v:
parts.append(_plain(x.get("content") or "") if isinstance(x, dict) else _plain(str(x)))
return " / ".join(p for p in parts if p)
if isinstance(v, dict):
return _plain(v.get("content") or "")
return _plain(str(v))
def study_parse_review(data, show_answers=True):
"""把 /questions/front/testpaper/teacher/review 的题目拍平,带标准答案/得分/解析。
review 的 item 与取题接口不同:
replay = 学生作答的 0 基下标数组(JSON 字符串)
answer = 标准答案:选择题是 0 基下标数组,填空/问答是 [{content:…}] 或文本
answerScore / status / analysis = 得分、对错、解析
show_answers=False 对应测验配置 examAfterWatchAnswer=0:平台不给看答案,
这里也不返回 correct/analysis,避免把不该展示的答案泄露到页面。
"""
out, seq = [], 0
for grp in data.get("questions") or []:
for it in grp.get("data") or []:
seq += 1
qtype = STUDY_QMAP.get(str(it.get("questionType") or ""), "01")
opts = []
for m in it.get("metasData") or []:
key = str(m.get("option") or "").strip().upper()
if key:
opts.append({"key": key, "text": _plain(m.get("content"))})
opts.sort(key=lambda o: o["key"])
if not opts and qtype == "03":
opts = [dict(o) for o in JUDGE_OPTIONS]
picked = (_idx_letters(it.get("replay"), opts)
if qtype in ("01", "02", "03") else _answer_text(it.get("replay")))
correct, is_correct = "", None
if show_answers:
if qtype in ("01", "02", "03"):
correct = _idx_letters(it.get("answer"), opts)
if correct:
is_correct = (set(picked.split("#")) - {""}) == \
(set(correct.split("#")) - {""})
else:
correct = _answer_text(it.get("answer"))
try:
is_correct = float(it.get("answerScore") or 0) > 0
except Exception:
is_correct = None
try:
ascore = float(it.get("answerScore") or 0)
except Exception:
ascore = None
out.append({"no": str(seq),
"type": qtype,
"title": _plain(it.get("stem"))[:600],
"options": opts,
"item_id": it.get("itemResultId"),
"picked": picked,
"correct": correct,
"is_correct": is_correct,
"answer_score": ascore,
"analysis": (_plain(it.get("analysis"))[:600] if show_answers else "")})
return out
def parse_quiz_html(html):
"""解析 jspx 的 showTestQums 页面:div.job-item[questype][quesno][traineeid][qumid]。"""
soup = BeautifulSoup(html or "", "lxml")
out = []
for it in soup.select("div.job-item"):
opts = []
for sp in it.select(".job-item-a span"):
key = (sp.get("data-name") or "").strip()
txt = re.sub(r"\s+", " ", sp.get_text(" ", strip=True))
if key:
opts.append({"key": key.upper(), "text": txt})
stem = ""
node = (it.select_one(".job-item-t") or it.select_one(".job-item-title")
or it.select_one("p.title") or it.select_one("div.title"))
if node:
stem = re.sub(r"\s+", " ", node.get_text(" ", strip=True))
else:
full = re.sub(r"\s+", " ", it.get_text(" ", strip=True))
for o in opts:
if o["text"]:
full = full.replace(o["text"], "")
stem = re.sub(r"\s+", " ", full).strip(" .、0123456789")
# 判断题在页面上通常只有 A/B 两个 radio、没有选项文字,
# 照样补出「正确/错误」,免得只看到一个字母不知道对错
qtype = (it.get("questype") or "01").strip()
if not opts and qtype == "03":
opts = [dict(o) for o in JUDGE_OPTIONS]
out.append({"no": (it.get("quesno") or "").strip(),
"type": qtype,
"title": stem[:600],
"options": opts,
"traineeid": it.get("traineeid"),
"ques_id": it.get("qumid")})
return out
def fetch_quiz_page(s, plan, obj):
url = "%s/train/trainee/showTestQums.action?planId=%s&objectId=%s" % (JSPX, plan, obj)
r = s.get(url, timeout=40)
html = r.text or ""
if "/admin/login.action" in html and len(html) < 2000:
raise SessionExpired("登录已失效,请重新登录后再取题")
return html
@app.post("/api/quiz")
def api_quiz(b: QuizIn):
"""取一次测验的题目(含选项),供页面表格展示。"""
if not b.cookies:
return {"ok": False, "code": "no_cookie", "error": "尚未登录"}
s = new_session(b.cookies)
try:
d = discover(s, b.plan_id, b.project_id, b.name)
except SessionExpired as e:
return {"ok": False, "code": "session_expired", "error": str(e)[:300]}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
tests = d.get("test") or []
if not tests:
return {"ok": False, "error": "这门课没有测验"}
if d.get("platform") == "study":
return _quiz_study(s, d, tests, retake=b.retake, period_id=b.period_id)
t = next((x for x in tests if not x.get("done")), tests[0])
url = t["href"] if t["href"].startswith("http") else JSPX + t["href"]
plan, _ = parse_ids(url)
m = re.search(r"objectId=([0-9A-Fa-f]+)", url)
obj = m.group(1) if m else (b.project_id or "")
try:
html = fetch_quiz_page(s, plan, obj)
except SessionExpired as e:
return {"ok": False, "code": "session_expired", "error": str(e)[:300]}
except Exception as e:
return {"ok": False, "error": "取题失败:%s" % str(e)[:200]}
qs = parse_quiz_html(html)
if not qs:
return {"ok": False, "error": "没解析到题目(页面 %d 字节),站点可能改版" % len(html)}
return {"ok": True, "platform": "jspx", "plan_id": plan, "object_id": obj,
"title": t.get("title") or "测验", "done": bool(t.get("done")),
"questions": qs}
def _quiz_study(s, d, tests, retake=False, period_id=""):
"""新版平台取题 / 重考 / 交卷后看答案。
取题:GET /train/questions/front/testpaper/getById?testpaperResultId=…
重考:PUT /train/questions/front/testpaper/start/do
{"periodId": <课时 id>, "trainCourseId": <trainCourseId>}
-> 新答卷 testpaperResultId(平台有 allowExamCount/surplusExamCount 次数限制)
看答案:GET /train/questions/front/testpaper/teacher/review?testpaperResultId=…
返回 answer(标准答案下标/文本)、replay(学生答案)、answerScore、analysis。
是否展示由测验的 examAfterWatchAnswer 决定,不是所有测验都给看。
这些和页面上「开始考试 / 查看试卷」发的是同一条请求,所以纯 API 也能完成。
"""
def pick_test():
if period_id:
for x in tests:
if str(x.get("period_id") or "") == str(period_id):
return x
return (next((x for x in tests if not x.get("done") and x.get("testpaper_result_id")), None)
or next((x for x in tests if not x.get("done")), None)
or next((x for x in tests if x.get("testpaper_result_id")), None)
or (tests[0] if tests else None))
t = pick_test()
if not t:
return {"ok": False, "code": "no_test", "error": "这门课没有测验"}
client = StudyClient(s, d.get("page_url") or (STUDY + "/training/curriculum"))
allow = t.get("allow_exam_count")
left = t.get("surplus_exam_count")
show_answers = str(t.get("exam_after_watch_answer") or "") == "1"
started = False
def start_paper(pid, err_prefix):
try:
j0 = client.call("PUT", "/questions/front/testpaper/start/do",
json_body={"periodId": int(pid),
"trainCourseId": d.get("train_id") or ""})
except Exception as e:
return None, "%s:%s" % (err_prefix, str(e)[:200])
if str(j0.get("code")) != "200":
return None, "%s:%s" % (err_prefix,
str(j0.get("msg") or j0.get("message") or j0)[:200])
data0 = j0.get("data") or {}
new_rid = str(data0.get("testpaperResultId") or data0.get("resultId") or "")
if not new_rid:
return None, "%s:平台没有返回 testpaperResultId" % err_prefix
return new_rid, ""
# ---- 重考:已交卷且还有剩余次数,start/do 会生成一份新答卷 ----
if retake:
if not t.get("done") and t.get("testpaper_result_id"):
return {"ok": False, "code": "doing",
"error": "还有一份未交卷的答卷,先交卷再重考"}
if not (isinstance(left, int) and left > 0):
used = (allow - left) if isinstance(allow, int) and isinstance(left, int) else "?"
return {"ok": False, "code": "no_attempts",
"error": "考试次数已用完(已用 %s/%s 次),不能再考" % (used, allow)}
pid = str(t.get("period_id") or "").strip()
if not pid:
return {"ok": False, "code": "no_period", "error": "这个测验没有课时 ID,无法重考"}
new_rid, err = start_paper(pid, "重考失败")
if err:
return {"ok": False, "error": err}
old_rid = str(t.get("testpaper_result_id") or "")
if new_rid == old_rid:
return {"ok": False, "code": "no_attempts",
"error": "平台没有生成新答卷(可能次数已用完或需要老师重置)"}
started = True
t = dict(t)
t.update({"testpaper_result_id": new_rid, "done": False, "result_status": "DOING"})
if isinstance(left, int):
left = max(0, left - 1) # 先本地减 1;交卷后重新 discover 会校准
rid = new_rid
else:
rid = str(t.get("testpaper_result_id") or "")
# ---- 第一次取题:还没有答卷就自动创建 ----
if not rid:
if t.get("done"):
return {"ok": False, "code": "no_result",
"error": "这个测验已完成,但平台没有返回答卷 ID,无法再看题目"}
pid = str(t.get("period_id") or "").strip()
if not pid:
return {"ok": False, "code": "no_period",
"error": "这个测验没有课时 ID,无法自动开始答卷"}
rid, err = start_paper(pid, "自动开始答卷失败")
if err:
return {"ok": False, "error": err}
started = True
# ---- 取当前答卷;已交卷的再调 review 拿标准答案 ----
paper = None
try:
j = client.call("GET", "/questions/front/testpaper/getById",
params={"testpaperResultId": rid})
if str(j.get("code")) == "200":
paper = j.get("data") or {}
except Exception:
paper = None
res = (paper or {}).get("result") or {}
status = str(res.get("status") or t.get("result_status") or "").upper()
done = status in ("FINISHED", "REVIEWED", "REVIEWING") or bool(t.get("done"))
qs, review = [], False
if done:
try:
jr = client.call("GET", "/questions/front/testpaper/teacher/review",
params={"testpaperResultId": rid})
if str(jr.get("code")) == "200":
rdata = jr.get("data") or {}
qs = study_parse_review(rdata, show_answers=show_answers)
review = bool(qs)
res = rdata.get("result") or res
paper = rdata
status = str(res.get("status") or status).upper()
except Exception:
pass
if not qs and paper:
qs = study_parse_paper(paper)
if not qs:
return {"ok": False, "error": "接口返回成功,但没有解析到题目"}
score = res.get("answerSumScore")
if score is None:
score = (paper or {}).get("score") if (paper or {}).get("score") is not None else t.get("score")
passed_score = (paper or {}).get("passedScore")
if passed_score is None:
passed_score = t.get("passed_score")
return {"ok": True, "platform": "study", "result_id": rid,
"period_id": t.get("period_id"),
"train_id": d.get("train_id") or "",
"started": started, # True = 本次请求创建了新答卷
"title": t.get("title") or (paper or {}).get("name") or "测验",
"questions": qs,
"done": done,
"review": review, # True = 已交卷,题目里带标准答案/解析
"show_answers": show_answers,
"allow_exam_count": allow,
"surplus_exam_count": left,
"can_retake": bool(done and isinstance(left, int) and left > 0),
"exam_after_watch_answer": t.get("exam_after_watch_answer") or "",
"result_status": status,
"score": score,
"passed_score": passed_score,
"passed": str(res.get("passedStatus") or "") == "1",
"submit_time": res.get("submitTime")}
def _quiz_study_submit(b):
"""新版平台交卷:PUT /train/questions/front/result/submitExam
{"resultId": "<testpaperResultId>",
"resultItemList": [{"fileIds": [], "itemId": "<itemResultId>", "replay": [0,2] | null}]}
replay 是**0 基选项下标**数组,字母答案要先按选项顺序换成下标。
"""
if not b.result_id:
return {"ok": False, "error": "缺少 resultId,请重新点一次「测验」"}
s = new_session(b.cookies)
client = StudyClient(s, STUDY + "/training/curriculum")
try:
j = client.call("GET", "/questions/front/testpaper/getById",
params={"testpaperResultId": b.result_id})
except Exception as e:
return {"ok": False, "error": "交卷前取题失败:%s" % str(e)[:200]}
if str(j.get("code")) != "200":
return {"ok": False, "error": "交卷前取题失败:%s" % str(j.get("msg") or j)[:200]}
qs = study_parse_paper(j.get("data") or {})
if not qs:
return {"ok": False, "error": "交卷前取题失败:没有题目"}
res0 = (j.get("data") or {}).get("result") or {}
st0 = str(res0.get("status") or "").upper()
if st0 in ("FINISHED", "REVIEWED", "REVIEWING"):
sc = res0.get("answerSumScore")
return {"ok": False, "code": "already_finished",
"error": "这份答卷已经交过卷%s;点「再考一次」可以生成新答卷再提交"
% ("(得分 %s)" % sc if sc is not None else "")}
items, answered = [], 0
for q in qs:
keys = [o["key"] for o in q["options"]]
letters = answer_tokens((b.answers or {}).get(q["no"]), q["options"], q["type"])
idx = [keys.index(x) for x in letters if x in keys]
if idx:
answered += 1
items.append({"fileIds": [], "itemId": q["item_id"], "replay": idx or None})
body = {"resultId": b.result_id, "resultItemList": items}
try:
r = client.call("PUT", "/questions/front/result/submitExam", json_body=body)
except Exception as e:
return {"ok": False, "error": "交卷失败:%s" % str(e)[:200]}
if str(r.get("code")) != "200":
return {"ok": False, "error": "交卷失败:%s" % str(r.get("msg") or r)[:200]}
score = None
try:
j2 = client.call("GET", "/questions/front/testpaper/getById",
params={"testpaperResultId": b.result_id})
res = ((j2.get("data") or {}).get("result") or {})
score = res.get("answerSumScore")
except Exception:
pass
return {"ok": True, "answers": answered, "total": len(items), "score": score,
"msg": "交卷成功"}
@app.post("/api/quiz/submit")
def api_quiz_submit(b: QuizSubmitIn):
"""按 quesNo 合并用户/模型的答案后交卷。重新取一次题,保证 traineeid 是当前会话的。"""
if not b.cookies:
return {"ok": False, "error": "尚未登录"}
if b.platform == "study":
return _quiz_study_submit(b)
s = new_session(b.cookies)
try:
html = fetch_quiz_page(s, b.plan_id, b.object_id)
except Exception as e:
return {"ok": False, "error": "重新取题失败:%s" % str(e)[:200]}
qs = parse_quiz_html(html)
if not qs:
return {"ok": False, "error": "重新取题失败,无法提交"}
answers = []
for it in qs:
ans = str((b.answers or {}).get(it["no"]) or "").strip().upper()
if not ans:
ans = it["options"][0]["key"] if it["options"] else "A"
answers.append({"id": it["traineeid"], "quesId": it["ques_id"],
"quesType": it["type"], "quesNo": it["no"], "answer": ans})
payload = {"planId": b.plan_id, "objectId": b.object_id,
"answers": json.dumps(answers, ensure_ascii=False, separators=(",", ":"))}
r3 = s.post(JSPX + "/train/trainee/endTest.action", data=payload, timeout=40)
try:
r4 = s.get("%s/train/trainee/doTest.action?planId=%s&objectId=%s"
% (JSPX, b.plan_id, b.object_id), timeout=40)
text = re.sub(r"\s+", " ", BeautifulSoup(r4.text, "lxml").get_text(" ", strip=True))
m = re.search(r"得分:\s*(\d+)\s*分", text)
score = m.group(1) if m else None
except Exception:
score = None
return {"ok": True, "answers": len(answers), "score": score,
"msg": "提交成功" if "MSG" not in (r3.text or "") else (r3.text or "")[:80]}
def parse_llm_answers(content, questions):
txt = re.sub(r"^```(?:json)?|```$", "", (content or "").strip(), flags=re.M).strip()
m = re.search(r"\{.*\}", txt, re.S)
if not m:
return {}
try:
data = json.loads(m.group(0))
except Exception:
return {}
raw = data.get("answers") if isinstance(data.get("answers"), dict) else data
if not isinstance(raw, dict):
return {}
out = {}
for q in questions:
no = str(q.get("no") or "")
v = raw.get(no, raw.get(no.strip()))
if isinstance(v, list):
v = "#".join(str(x) for x in v)
parts = answer_tokens(v, q.get("options"), str(q.get("type") or "01"))
if parts:
out[no] = "#".join(parts)
return out
@app.post("/api/quiz/answer")
def api_quiz_answer(b: QuizAnswerIn):
"""把题目分批丢给配置好的大模型,返回 {题号: 答案}。
一次发 20 题,很多 OpenAI 兼容网关 100 秒就回 504,所以按 5 题一批串行跑。
"""
cfg = load_settings()["llm"]
if not cfg.get("api_key") or not cfg.get("base_url"):
return {"ok": False, "code": "no_llm",
"error": "还没配置大模型:右上角「设置」→ 填 base_url / api_key / model"}
if not b.questions:
return {"ok": False, "error": "没有题目"}
system = (b.prompt or cfg.get("prompt") or DEFAULT_ANSWER_PROMPT)
answers, errors = {}, []
batch = 5
for i in range(0, len(b.questions), batch):
part = b.questions[i:i + batch]
lines = []
for q in part:
opts = ";".join("%s. %s" % (o.get("key"), o.get("text")) for o in (q.get("options") or []))
lines.append("第%s题(%s)%s\n选项:%s" % (
q.get("no"), TYPE_LABEL.get(str(q.get("type")), "单选"),
q.get("title") or "", opts or "无"))
try:
content = llm_chat([{"role": "system", "content": system},
{"role": "user", "content": "\n\n".join(lines)}], cfg)
answers.update(parse_llm_answers(content, part))
except Exception as e:
errors.append(str(e)[:120])
if not answers:
return {"ok": False, "error": "模型没有返回可用答案:%s" % (";".join(errors)[:300] or "空响应")}
return {"ok": True, "answers": answers, "errors": errors}
def _log_login(username, ok, code, error, elapsed, stage=""):
"""每次登录尝试落一行到 SQLite,便于事后对照平台原话。"""
try:
store.login_log_add(username, ok, code, elapsed, stage, error)
except Exception:
pass
def _login_account(b: LoginIn):
"""解析出一组可登录的凭据:显式传入优先,否则按账号 id 从库里取。"""
acc = store.account_get(b.account_id) if b.account_id else None
username = (b.username or "").strip() or ((acc or {}).get("username") or "")
password = b.password or (store.account_password(b.account_id) if b.account_id else "")
return acc, username, password
@app.post("/api/login")
def api_login(b: LoginIn):
t0 = time.time()
acc, username, password = _login_account(b)
if not username or not password:
err = "没有可用的账号密码:新账号请在「添加账号」里填密码,服务端会把它存进 SQLite"
_log_login(username or b.account_id, False, "no_credential", err, time.time() - t0,
"LoginIn")
return JSONResponse({"ok": False, "code": "no_credential", "error": err}, status_code=200)
try:
s, user = login(username, password)
except Exception as e:
err = str(e)[:300]
_log_login(username, False, "login_failed", err, time.time() - t0, type(e).__name__)
if acc:
store.account_upsert({"id": acc["id"], "username": acc["username"],
"status": "login_failed", "lastError": err,
"lastErrorAt": time.time()})
return JSONResponse({"ok": False, "code": "login_failed", "error": err}, status_code=200)
warn = ""
try:
courses = scrape_courses(s)
except Exception as e:
courses, warn = [], str(e)[:200]
cookies = dump_cookies(s)
_log_login(username, True, "ok", warn, time.time() - t0, "user=%s" % user)
if acc:
store.account_save_login(acc["id"], user, cookies, courses)
return {"ok": True, "user": user, "cookies": cookies, "courses": courses, "warn": warn}
@app.get("/api/loginlog")
def api_loginlog(tail: int = 40):
"""登录尝试回放:前端点「登录」后,这里能看到平台到底回了什么。"""
return {"ok": True, "lines": store.login_log_tail(tail)}
# ------------------------------------------------------------------ accounts
@app.get("/api/accounts")
def api_accounts():
"""全部账号(含会话 cookie / 课程缓存 / 勾选),**不含密码**。"""
return {"ok": True, "accounts": store.account_list()}
@app.post("/api/accounts")
def api_account_save(b: AccountIn):
"""新增或更新一个账号。password 不传(或传空)表示保留库里已有的密码。"""
rec = b.model_dump()
if not (rec.get("password") or "").strip():
rec["password"] = None
if not rec.get("id") and not rec.get("username"):
return JSONResponse({"ok": False, "error": "缺少 username"}, status_code=200)
try:
aid = store.account_upsert(rec)
except Exception as e:
return JSONResponse({"ok": False, "error": "写入 SQLite 失败:%s" % str(e)[:200]},
status_code=200)
return {"ok": True, "id": aid, "account": store.account_get(aid)}
@app.get("/api/accounts/{account_id}")
def api_account_get(account_id: str):
acc = store.account_get(account_id)
if not acc:
return JSONResponse({"ok": False, "error": "账号不存在"}, status_code=200)
return {"ok": True, "account": acc}
@app.delete("/api/accounts/{account_id}")
def api_account_del(account_id: str):
store.account_delete(account_id)
with RUNS_LOCK:
RUNS.pop(account_id, None)
return {"ok": True}
# ------------------------------------------------------------------ 运行日志 / 快照
@app.get("/api/log/history")
def api_log_history(key: str = "", tail: int = 200):
"""从 SQLite 读历史日志;不带 key 时列出每个账号存了多少行。"""
if not key:
return {"ok": True, "keys": store.run_log_keys()}
return {"ok": True, "key": key, "lines": store.run_log_tail(key, tail)}
@app.delete("/api/log/history")
def api_log_history_clear(key: str = ""):
store.run_log_clear(key)
return {"ok": True}
@app.get("/api/snapshots")
def api_snapshots(limit: int = 50):
"""站点改版时存下的课程页原始 HTML 列表。"""
return {"ok": True, "snapshots": store.snapshot_list(limit)}
@app.get("/api/snapshot/{sid}", response_class=HTMLResponse)
def api_snapshot(sid: int):
row = store.snapshot_get(sid)
if not row:
return HTMLResponse("<h3>快照不存在</h3>", status_code=404)
head = ("<!-- id=%s planId=%s projectId=%s final url: %s created=%s -->\n"
% (row["id"], row["plan_id"], row["project_id"], row["url"],
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(row["created"] or 0))))
return HTMLResponse(head + (row["html"] or ""))
@app.post("/api/courses")
def api_courses(b: CookiesIn):
if not b.cookies:
return JSONResponse({"ok": False, "code": "no_cookie",
"error": "本机没有该账号的会话 cookie,请先点「登录」"}, status_code=200)
s = new_session(b.cookies)
if not s.cookies:
return JSONResponse({"ok": False, "code": "no_cookie",
"error": "保存的 cookie 全部被过滤(domain 不含 dgjy),请重新登录"},
status_code=200)
try:
courses = scrape_courses(s)
except SessionExpired as e:
return JSONResponse({"ok": False, "code": "session_expired", "error": str(e)[:300]}, status_code=200)
except Exception as e:
return JSONResponse({"ok": False, "code": "fetch_failed", "error": str(e)[:300]}, status_code=200)
if not courses:
return JSONResponse({"ok": False, "code": "no_course",
"error": "登录态有效,但该账号名下三个板块(在学/助学/督学)"
"的进行中、未开始、已结束里都没有课程"}, status_code=200)
try:
user = whoami(fetch_admin(s))
except Exception:
user = ""
return {"ok": True, "courses": courses, "user": user,
"in_study": len([c for c in courses if c["in_study"]]),
# 补 jspx 会话时 cookie 会换新(JSESSIONID/route),回传让前端写回 IndexedDB
"cookies": dump_cookies(s)}
@app.post("/api/discover")
def api_discover(b: DiscoverIn):
s = new_session(b.cookies)
try:
d = discover(s, b.plan_id, b.project_id, b.name)
return {"ok": True, **d, "cookies": dump_cookies(s)}
except Exception as e:
return JSONResponse({"ok": False, "error": str(e)[:300]}, status_code=200)
@app.post("/api/study")
def api_study(b: StudyIn):
st = RUNS.get(b.key) or {}
if st.get("status") in ("discovering", "studying"):
return JSONResponse({"ok": False, "error": "该账号正在学习中"}, status_code=200)
if not b.cookies:
return JSONResponse({"ok": False, "error": "尚未登录"}, status_code=200)
courses = b.courses or []
if not courses and b.plan_id and b.project_id:
courses = [{"name": "", "plan_id": b.plan_id, "project_id": b.project_id}]
if not courses:
return JSONResponse({"ok": False, "error": "没有选中任何课程"}, status_code=200)
set_state(b.key, status="discovering", log=[], done=0, total=0, started=time.time())
# 会话失效时后台要能自动重登:优先用请求里的密码,没有就按 key 从 SQLite 取
pw = b.password or store.account_password(b.key) or ""
st = load_settings()["study"]
threading.Thread(target=study_worker,
args=(b.key, b.username, b.cookies, courses,
max(1, min(b.workers, 95)), b.cap_minutes, b.quiz_key,
b.enable_quiz, b.enable_feedback, pw,
st.get("mode") or "fast", int(st.get("parallel") or 2)),
daemon=True).start()
return {"ok": True, "courses": len(courses),
"enable_quiz": b.enable_quiz, "enable_feedback": b.enable_feedback,
"study": st}
@app.post("/api/stop")
def api_stop(b: StopIn):
set_state(b.key, status="stopped")
run_log(b.key, "用户请求停止")
return {"ok": True}
@app.post("/api/capture")
def api_capture(b: CaptureIn):
"""开一个可见的浏览器让用户手动走一遍,记录所有 /train/** 请求与响应。"""
if (RUNS.get(b.key) or {}).get("capturing"):
return JSONResponse({"ok": False, "error": "正在抓包"}, status_code=200)
if not b.cookies:
return JSONResponse({"ok": False, "error": "尚未登录"}, status_code=200)
CAPTURES[b.key] = []
set_state(b.key, capturing=True, capture_stop=False, capture_file="")
run_log(b.key, "正在打开浏览器…")
threading.Thread(target=capture_worker,
args=(b.key, b.cookies, max(1, min(int(b.minutes or 10), 30))),
daemon=True).start()
return {"ok": True}
@app.post("/api/capture/stop")
def api_capture_stop(b: StopIn):
set_state(b.key, capture_stop=True)
run_log(b.key, "用户请求停止抓包")
return {"ok": True}
@app.get("/api/capture")
def api_capture_get(key: str = ""):
st = RUNS.get(key) or {}
return {"ok": True, "capturing": bool(st.get("capturing")),
"file": st.get("capture_file") or "",
"records": CAPTURES.get(key) or []}
@app.get("/api/log")
def api_log(key: str, since: int = 0):
st = RUNS.get(key) or {}
log = st.get("log", [])
out = {"lines": log[since:], "next": len(log), "status": st.get("status", ""),
"done": st.get("done", 0), "total": st.get("total", 0),
"course": st.get("course", ""), "progress": st.get("progress", "")}
# discover 可能给账号补上了 study.dgjy.net 的 Middle-Token,回传给前端写回 IndexedDB
if st.get("cookies"):
out["cookies"] = st["cookies"]
return out
@app.get("/api/runs")
def api_runs():
with RUNS_LOCK:
return {k: {kk: vv for kk, vv in v.items() if kk != "log"} for k, v in RUNS.items()}
def _port_open(port, host="127.0.0.1", timeout=0.6):
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout)
return s.connect_ex((host, port)) == 0
def _lan_ip():
"""本机对外的网卡地址,用于启动提示;容器里拿到的是容器 IP,不能当访问地址,返回空。"""
if os.path.exists("/.dockerenv"):
return ""
import socket
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.connect(("223.5.5.5", 80))
return s.getsockname()[0]
except Exception:
return ""
def _probe_self(port):
"""端口已被占用时判断是不是本程序自己在跑,避免重复启动直接报错。"""
import urllib.request
try:
with urllib.request.urlopen("http://127.0.0.1:%d/api/health" % port, timeout=1.5) as r:
return json.loads(r.read().decode("utf-8"))
except Exception:
return None
def main():
port = int(os.environ.get("PORT", "8848"))
# 默认只监听本机;Docker / 远程管理用 HOST=0.0.0.0
host = os.environ.get("HOST", "127.0.0.1")
url = "http://127.0.0.1:%d" % port
# 设了访问令牌时,打印出来的地址一律带上 ?token=:本机访问同样要校验令牌
suffix = ("/?token=%s" % AUTH_TOKEN) if AUTH_TOKEN else ""
store.init() # 建库建表 + 导入旧的 settings.json / logs/*.log
hydrate_runs() # 上次的运行状态读回来(没跑完的标成已中断)
print("SQLite: %s" % store.DB_PATH)
if _port_open(port):
info = _probe_self(port)
if info and info.get("app") == APP_ID:
print("服务已在运行 -> %s%s (PID %s)" % (url, suffix, info.get("pid")))
print("直接打开上面的地址;要重启请先结束 PID %s。" % info.get("pid"))
return
print("[错误] 端口 %d 被其它程序占用(不是本程序)。" % port)
print(" 查占用: netstat -ano | findstr :%d" % port)
print(" 换端口: set PORT=8849 && python app.py")
return
print("多账号学习控制台 -> %s%s" % (url, suffix))
print("IPv6 localhost 同步监听 -> http://[::1]:%d%s" % (port, suffix))
if AUTH_TOKEN:
print("访问令牌 AUTH_TOKEN -> %s" % AUTH_TOKEN)
print("远程访问地址 -> http://%s:%d%s" % (_lan_ip() or "<服务器IP>", port, suffix))
elif host == "0.0.0.0":
print("[提示] 已监听 0.0.0.0 但没有设置 AUTH_TOKEN:同网段谁都能打开,远程部署请设置")
srv4 = uvicorn.Server(uvicorn.Config(app, host=host, port=port,
log_level="warning", access_log=False))
def _serve_v6():
# Windows 上 localhost 优先解析到 ::1。只绑 127.0.0.1 时,
# 浏览器打开 http://localhost:8848 就是 ERR_CONNECTION_REFUSED。
try:
uvicorn.Server(uvicorn.Config(app, host="::1", port=port,
log_level="warning", access_log=False)).run()
except Exception:
pass
threading.Thread(target=_serve_v6, daemon=True, name="uvicorn-ipv6").start()
srv4.run()
if __name__ == "__main__":
main()