feat: 接入广东公需课平台（jsglpt 登录 + jsxx 进度心跳）
- 新增 platforms/gdgxk.py：base64 登录 → ssoItlmsGxkTicket 换票进 jsxx → /uc/store/courseRegister 取课程 → updateViewTime1 心跳上报进度（无需真播） - platforms/__init__.py 注册；配套抓包脚本与自测脚本 - CHANGELOG / OPTIMIZATION / todo 同步，版本升到 0.0.2
This commit is contained in:
study-tool
@@ -0,0 +1,127 @@
|
||||
"""广东省教师继续教育平台抓包工具（一次性取证用，不参与运行）。
|
||||
|
||||
用法：
|
||||
python script/capture-jsglpt.py [分钟数]
|
||||
|
||||
行为：开一个有头 Chromium（持久 profile，登录态会保留），打开 jsglpt 登录页，
|
||||
把你在窗口里操作期间产生的 XHR/fetch 全量落到 data/capture-jsglpt-<时间>.jsonl，
|
||||
同时每 3 秒把 cookie 快照写到 data/capture-jsglpt-cookies.json。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
DATA = os.path.join(ROOT, "data")
|
||||
LOGIN = "https://jsglpt.gds.edu.cn/login.jsp"
|
||||
MINUTES = float(sys.argv[1]) if len(sys.argv) > 1 else 40
|
||||
|
||||
os.makedirs(DATA, exist_ok=True)
|
||||
STAMP = time.strftime("%Y%m%d-%H%M%S")
|
||||
JSONL = os.path.join(DATA, "capture-jsglpt-%s.jsonl" % STAMP)
|
||||
COOKIES = os.path.join(DATA, "capture-jsglpt-cookies.json")
|
||||
PROFILE = os.path.join(DATA, ".gd-profile")
|
||||
|
||||
out = open(JSONL, "a", encoding="utf-8")
|
||||
seen = 0
|
||||
|
||||
|
||||
def emit(rec):
|
||||
global seen
|
||||
seen += 1
|
||||
rec["n"] = seen
|
||||
out.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||||
out.flush()
|
||||
print(" %-4s %s %s -> %s" % (rec.get("rt", "doc"), rec.get("method", ""),
|
||||
rec.get("url", "")[:150], rec.get("status", "")), flush=True)
|
||||
|
||||
|
||||
def on_response(r):
|
||||
try:
|
||||
req = r.request
|
||||
rt = req.resource_type
|
||||
if rt not in ("xhr", "fetch", "document"):
|
||||
return
|
||||
h = {}
|
||||
try:
|
||||
h = dict(r.headers or {})
|
||||
except Exception:
|
||||
pass
|
||||
body = ""
|
||||
# 只留文本类响应，视频/图片跳过
|
||||
if not (r.url or "").lower().split("?")[0].endswith((".mp4", ".m3u8", ".ts", ".png", ".jpg", ".gif", ".woff", ".woff2", ".css", ".js")):
|
||||
try:
|
||||
body = r.text()[:8000]
|
||||
except Exception:
|
||||
body = "<binary>"
|
||||
emit({
|
||||
"t": time.strftime("%H:%M:%S"),
|
||||
"rt": rt,
|
||||
"method": req.method,
|
||||
"url": r.url,
|
||||
"req_headers": dict(req.headers or {}),
|
||||
"req": (req.post_data or "")[:4000],
|
||||
"status": r.status,
|
||||
"resp_headers": h,
|
||||
"resp": body,
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
print("抓包输出：%s" % JSONL, flush=True)
|
||||
with sync_playwright() as p:
|
||||
ctx = p.chromium.launch_persistent_context(
|
||||
PROFILE, headless=False,
|
||||
args=["--disable-blink-features=AutomationControlled"],
|
||||
viewport={"width": 1440, "height": 900},
|
||||
)
|
||||
hooked = set()
|
||||
|
||||
def hook(page):
|
||||
if id(page) in hooked:
|
||||
return
|
||||
hooked.add(id(page))
|
||||
page.on("response", on_response)
|
||||
print("[页面] %s" % (page.url or "about:blank"), flush=True)
|
||||
|
||||
ctx.on("page", hook)
|
||||
pg = ctx.pages[0] if ctx.pages else ctx.new_page()
|
||||
hook(pg)
|
||||
try:
|
||||
pg.goto(LOGIN, wait_until="domcontentloaded", timeout=60000)
|
||||
except Exception as e:
|
||||
print("打开登录页失败：%s" % e, flush=True)
|
||||
|
||||
last = 0.0
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < MINUTES * 60:
|
||||
try:
|
||||
now = time.time()
|
||||
if now - last > 3:
|
||||
last = now
|
||||
with open(COOKIES, "w", encoding="utf-8") as f:
|
||||
json.dump(ctx.cookies(), f, ensure_ascii=False, indent=1)
|
||||
ctx.pages[0].wait_for_timeout(1000)
|
||||
except Exception:
|
||||
# 用户把所有页面关掉了，收工
|
||||
break
|
||||
try:
|
||||
with open(COOKIES, "w", encoding="utf-8") as f:
|
||||
json.dump(ctx.cookies(), f, ensure_ascii=False, indent=1)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
ctx.close()
|
||||
except Exception:
|
||||
pass
|
||||
out.close()
|
||||
print("抓包结束：%d 条 → %s" % (seen, JSONL), flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,41 @@
|
||||
"""广东公需课自测：用浏览器抓到的 cookie 走一遍 列表 → 发现 → 心跳。
|
||||
|
||||
用法：PYTHONPATH=src python script/test-gdgxk.py [课件id]
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
import uuid
|
||||
|
||||
from study_tool.platforms import gdgxk
|
||||
|
||||
cid = sys.argv[1] if len(sys.argv) > 1 else ""
|
||||
cookies = json.load(open("data/capture-jsglpt-cookies.json", encoding="utf-8"))
|
||||
s = gdgxk.new_session(cookies)
|
||||
|
||||
courses = gdgxk.list_courses(s)
|
||||
print("课程数：", len(courses))
|
||||
for c in courses:
|
||||
print(" ", c["plan_id"], c["name"])
|
||||
|
||||
if not cid and courses:
|
||||
cid = courses[0]["plan_id"]
|
||||
|
||||
d = gdgxk.discover(s, cid, "")
|
||||
print("\n课程：", d["course"])
|
||||
print("进度：", d["progress"])
|
||||
print("视频：", d["info"])
|
||||
cw = d["courseware"][0]
|
||||
print("课件：", cw["id"], "| 总长", cw["total"], "| 已看", cw["viewTime"])
|
||||
|
||||
sid = uuid.uuid4().hex
|
||||
sign = gdgxk._sign(s, {"aid": cw["aid"], "id": cw["id"], "sid": sid}, cw["uid"])
|
||||
print("\n播放注册签名：", sign)
|
||||
print("服务端已看秒数：", gdgxk._view_time(s, cw, cw["uid"], sign))
|
||||
|
||||
r = gdgxk._xhr(s, "POST",
|
||||
gdgxk.X + "/" + cw["aid"] + "/study/video/user/" + cw["id"] + "/updateViewTime1",
|
||||
data={"_method": "PUT", "aid": cw["aid"], "isLimit": "false", "md5": sign["md5"],
|
||||
"expires": sign["time"], "md5Uid": cw["uid"],
|
||||
"lastViewTime": str(min(cw["viewTime"] + 60, cw["total"])),
|
||||
"playbackSessionId": sid})
|
||||
print("\n心跳响应：", r.status_code, r.text[:300])
|
||||
Reference in New Issue
Block a user