"""
莞工教务系统 - 核心API模块
============================
被 CLI (dgut.py) 和 GUI (dgut-gui/dgut_gui.py) 共用。
自适应探测选课类型标签页，不依赖硬编码的课程类型。
"""
import json
import re
import subprocess
import time
from typing import Any, Optional
ORIGIN = "https://jwx.dgut.edu.cn"
BROWSER_ACT = "browser-act"
SESSION_NAME = "jwx-s1"
# 已知的标签页ID → iframe页面 → API 映射
# 新增选课类型时只需在此添加
TAB_REGISTRY: dict[str, dict[str, Optional[str]]] = {
"选课学分情况": {"id": "xkzx", "iframe": "/xsxk/xsxk_tzsm", "api": None},
"必修选课": {"id": "bxxksfkf", "iframe": "/xsxkkc/getBxxkxx", "api": "/xsxkkc/xsxkXxxk"},
"选修选课": {"id": "xxxksfkf", "iframe": "/xsxkkc/getXxxk", "api": "/xsxkkc/xsxkXxxk"},
"本学期计划选课": {"id": "bxqxksfkf", "iframe": "/xsxkkc/getBxqjxk", "api": "/xsxkkc/xsxkXxxk"},
"公共选修": {"id": "ggksfkf", "iframe": "/xsxkkc/getGgxxk", "api": "/xsxkkc/xsxkXxxk"},
"跨年级选课": {"id": "knjxksfkf", "iframe": "/xsxkkc/getKnjxk", "api": "/xsxkkc/xsxkXxxk"},
"跨专业选课": {"id": "kzyxksfkf", "iframe": "/xsxkkc/getKzyxk", "api": "/xsxkkc/xsxkXxxk"},
"辅修选课": {"id": "fxzyxksfkf", "iframe": "/xsxkkc/getFxzy", "api": "/xsxkkc/xsxkXxxk"},
"重修选课": {"id": "cxxkkf", "iframe": "/xsxkkc/comeInCxxk", "api": None},
"体育选课": {"id": "tyxksfkf", "iframe": "/xsxkkc/newcomeInTyxk", "api": None},
"选课组选课": {"id": "xkzxksfkf", "iframe": "/xsxkkc/getXkzxk", "api": None},
}
# ─── 底层通信 ────────────────────────────────────────────────────────────
def _browser_eval(js_code: str) -> str:
escaped = js_code.replace("'", "'\\''")
cmd = f"{BROWSER_ACT} --session {SESSION_NAME} eval '{escaped}'"
try:
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or result.stdout.strip())
return result.stdout.strip()
except FileNotFoundError:
raise RuntimeError("browser-act not found")
def fetch_api(method: str, path: str, body: Optional[str] = None,
referer: Optional[str] = None) -> Any:
"""在浏览器上下文中发 fetch 请求（自动携带 cookie）。"""
url = f"{ORIGIN}{path}"
headers: dict[str, str] = {"X-Requested-With": "XMLHttpRequest"}
if referer:
headers["Referer"] = referer
opts = {"method": method, "credentials": "include", "headers": headers}
if body:
opts["body"] = body
opts["headers"]["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
opts_json = json.dumps(opts)
js = (
f"(async()=>{{"
f"try{{const r=await fetch('{url}',{opts_json});const t=await r.text();"
f"try{{return JSON.stringify({{ok:!0,data:JSON.parse(t)}})}}"
f"catch(e){{return JSON.stringify({{ok:!0,raw:t}})}}}}"
f"catch(e){{return JSON.stringify({{ok:!1,error:e.message}})}}}})()"
)
output = _browser_eval(js)
result = json.loads(output)
if not result.get("ok"):
raise RuntimeError(result.get("error", "unknown error"))
return result.get("data") or result.get("raw", {})
def fetch_html(path: str) -> str:
return fetch_api("GET", path)
def _browser_click_enter_button() -> bool:
"""在浏览器中点击"进入选课"按钮。"""
js = """
(function(){
var f = document.querySelector('iframe[id=NEW_XSD_PYGL_XKGL_NXSXKZX]');
if(!f) return 'no iframe';
var d = f.contentDocument;
var btns = d.querySelectorAll('button');
for(var b of btns) {
if(b.textContent.includes('进入选课')) { b.click(); return 'ok'; }
}
return 'btn not found';
})()
"""
result = _browser_eval(js)
time.sleep(3)
return "ok" in result
def ensure_in_course_selection() -> bool:
"""
确保浏览器已在选课界面内。若在列表页则自动点击"进入选课"。
"""
try:
r = subprocess.run(
f"{BROWSER_ACT} --session {SESSION_NAME} state",
shell=True, capture_output=True, text=True, timeout=15,
)
out = r.stdout
if "selectBottom" in out and "进入选课" not in out:
return True
if "进入选课" in out:
return _browser_click_enter_button()
return False
except Exception:
return False
# ─── 业务 API ────────────────────────────────────────────────────────────
def get_cycles() -> list[dict]:
"""获取可选课的选课轮次列表。"""
return fetch_api("GET", "/xsxk/xklc_list_data?xkmc=").get("data", [])
def discover_tabs(jx0502zbid: str) -> list[dict]:
"""
动态探测选课界面开放了哪些标签页。
通过分析 selectBottom 页面的 HTML 和 JavaScript 实现。
返回: [{"id": str, "label": str, "iframe": str, "api": str|None}, ...]
"""
html = fetch_html(f"/xsxk/selectBottom?jx0502zbid={jx0502zbid}&sfylxkstr=")
# 方式1: 从 选修选课 或 'xxxksfkf' 提取
tabs: list[dict] = []
seen_ids: set[str] = set()
for m in re.finditer(
r"""]*?onclick="jpbxxk\(['']{1}(.+?)['']{1}\)"[^>]*?>([^<]+)""",
html,
):
tid, label = m.group(1), m.group(2).strip()
if tid in seen_ids:
continue
seen_ids.add(tid)
known = TAB_REGISTRY.get(label, {})
tabs.append({
"id": tid,
"label": label,
"iframe": known.get("iframe"),
"api": known.get("api"),
})
# 方式2: 从 jpbxxk 函数解析 iframe 映射
iframe_map: dict[str, str] = {}
for m in re.finditer(
r"case\s+'([^']+)':\s*[^#]*?#selectTable\)\.attr\('src',\s*\"([^\"]+)\"",
html,
):
iframe_map[m.group(1)] = m.group(2)
for tab in tabs:
if not tab["iframe"]:
tab["iframe"] = iframe_map.get(tab["id"])
# 方式3: 对新标签尝试自动推断 API
for tab in tabs:
if tab["api"] is None and tab.get("iframe"):
# 如果 iframe 路径包含 getXxx 模式，推测对应 xsxkXxx API
iframe = tab["iframe"]
m = re.match(r"/xsxkkc/get(\w+)", iframe)
if m:
suffix = m.group(1)
tab["api"] = f"/xsxkkc/xsxk{suffix}"
return tabs
def get_courses(api_path: str, referer: str) -> list[dict]:
"""获取某选课类型的课程列表。"""
post_data = (
"sEcho=1&iColumns=14&sColumns=&iDisplayStart=0&iDisplayLength=50"
"&mDataProp_0=kch&mDataProp_1=kcmc&mDataProp_2=skfsmc"
"&mDataProp_3=fzmc&mDataProp_4=ktmc&mDataProp_5=xf"
"&mDataProp_6=skls&mDataProp_7=sksj&mDataProp_8=skdd"
"&mDataProp_9=xqmc&mDataProp_10=syrs&mDataProp_11=skfsmc"
"&mDataProp_12=ctsm&mDataProp_13=czOper"
)
query = "?1=1&kcxx=&skls=&skfs=" if "?" not in api_path else ""
data = fetch_api("POST", f"{api_path}{query}", body=post_data, referer=referer)
return data.get("aaData", [])
def select_course(kcid: str, jx0404id: str, referer: str) -> dict:
"""选课。"""
params = f"kcid={kcid}&cfbs=null&jx0404id={jx0404id}&xkzy=&trjf=&sfsyjc=&sfkvtj="
return fetch_api("GET", f"/xsxkkc/xxxkOper?{params}", referer=referer)
def drop_course(jx0404id: str, referer: str) -> dict:
"""退课。"""
return fetch_api("GET", f"/xsxkjg/xstkOper?jx0404id={jx0404id}", referer=referer)
def ensure_session() -> bool:
"""检查 browser-act session 是否存在。"""
try:
r = subprocess.run(f"{BROWSER_ACT} session list", shell=True,
capture_output=True, text=True, timeout=10)
return SESSION_NAME in r.stdout
except FileNotFoundError:
return False