+ 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 "
" 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("&", "&"))
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=
"""
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/；写库失败才退回文件。
"""
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("\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：原生