#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Ginkgo (DUFE) sign-in code query service. Endpoint: GET|POST /dufe/ginkgostu_sign_in params: username, password(base64), url(base64) [commit=true] optional, default true: also POST the code back [mode=auto] optional: fetch the notification list, discover sign-in activities and auto sign-in each one (bot ginkgo parity, consecutive retries); results returned as an aggregated list. mode=auto 时 url 参数可省略。 Listen: 127.0.0.1:9005 (WSGI: waitress) """ import base64 import binascii import html as _html import json import logging import re import sys import time import urllib.parse import uuid from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime from html.parser import HTMLParser try: import requests except ImportError: sys.exit("Missing dependency, run: pip install requests") try: from flask import Flask, request, jsonify except ImportError: sys.exit("Missing dependency, run: pip install flask") try: from waitress import serve except ImportError: sys.exit("Missing dependency, run: pip install waitress") requests.packages.urllib3.disable_warnings() # ------------------------------------------------------------------------------ # Site constants # ------------------------------------------------------------------------------ GINKGO = "https://ginkgostu.dufe.edu.cn" SSO = "https://sso.dufe.edu.cn" CUSTOM_CODE = "kb_dconline" NOTICE_PAGE = GINKGO + "/notice" SYSTEM_PAGE = GINKGO + "/notice/system" PAGE_SIZE = 20 # 单类单次抓取条数（与 bot ginkgo 一致） # 自动签到重试参数（与 bot ginkgo 的 MAX_SIGNIN_ATTEMPTS / SIGNIN_BACKOFF 一致： # 首次 + 最多 5 次重试，每次间隔 3 秒） MAX_SIGNIN_ATTEMPTS = 6 SIGNIN_BACKOFF = (3, 3, 3, 3, 3) TZ8 = timezone(timedelta(hours=8)) UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) log = logging.getLogger("ginkgo") # ------------------------------------------------------------------------------ # Error taxonomy # AuthError -> 401 caller-side credential / captcha problem # NoSignInCode -> 404 activity exists but has no code configured # UpstreamError -> 502 remote service misbehaved / changed shape # anything else -> 500 internal, generic message to the client # ------------------------------------------------------------------------------ class AuthError(Exception): """Authentication failed (bad credentials, captcha rejected, locked account).""" class NoSignInCodeError(Exception): """The numeric-code activity exists but has no code configured.""" class UnsupportedSigninType(Exception): """The activity requires a flow this service cannot safely automate.""" SIGNIN_TYPE_NAMES = { 0: "二维码签到", 1: "数字码签到", 2: "普通签到", 3: "直播签到", } def signin_type_label(value): try: signin_type = int(value) except (TypeError, ValueError): return "未知签到类型" return SIGNIN_TYPE_NAMES.get(signin_type, "未知签到类型(%s)" % signin_type) class UpstreamError(Exception): """Remote service returned an unexpected or erroneous response.""" class RateLimitedError(Exception): """An upstream explicitly refused requests due to throttling.""" # ------------------------------------------------------------------------------ # Rate-limit detection & credential-safe errors # (synced from bot ginkgostu.py 2026-09-23; bot is the baseline) # ------------------------------------------------------------------------------ _RATE_MESSAGE = re.compile( r"too\s+many\s+requests|rate[\s_-]*limit(?:ed|ing|\s+exceeded)?|" r"request(?:s)?\s+throttl|throttl(?:ed|ing)|" r"(?:请求|访问|操作|登录|登陆|接口|提交).{0,12}(?:过于频繁|太频繁|频繁|频率.{0,6}(?:超|限制))|" r"(?:限流|请求次数.{0,8}(?:超限|超出|超过)|访问次数.{0,8}(?:超限|超出|超过))", re.I, ) _RATE_CODES = {"429", "4290", "40029", "90029", "RATE_LIMIT", "RATE_LIMITED", "RATE_LIMIT_EXCEEDED", "TOO_MANY_REQUESTS", "REQUEST_TOO_FREQUENT", "THROTTLED", "THROTTLING", "FREQUENT_REQUESTS"} _FIELDS = ("code", "msg", "message", "error") def _status(value): try: return int(value) except (ValueError, TypeError): return 0 def _explicit(value, depth=0): if depth > 3: return False if isinstance(value, dict): for key in _FIELDS: item = value.get(key) if key == "code" and isinstance(item, (str, int)): if str(item).strip().upper().replace("-", "_") in _RATE_CODES: return True elif isinstance(item, dict): if _explicit(item, depth + 1): return True elif isinstance(item, str) and _RATE_MESSAGE.search(item[:1024]): return True return False class _LoginErrors(HTMLParser): """Extract only explicitly marked login error elements; ignore scripts/notices.""" def __init__(self): super().__init__() self.stack = [] self.messages = [] def handle_starttag(self, tag, attrs): attrs = dict(attrs) marker = (attrs.get("id", "") + " " + attrs.get("class", "")).lower() explicit = bool(re.search(r"(?:^|[\s_-])(?:error|errors|errormsg|errormessage)(?:$|[\s_-])", marker)) ignored = tag in ("script", "style", "template") or bool(self.stack and self.stack[-1][2]) active = explicit or bool(self.stack and self.stack[-1][1]) if tag not in ("input", "br", "img", "meta", "link", "hr", "wbr", "source"): self.stack.append((tag, active, ignored)) def handle_endtag(self, tag): for index in range(len(self.stack) - 1, -1, -1): if self.stack[index][0] == tag: del self.stack[index:] break def handle_data(self, data): if self.stack and self.stack[-1][1] and not self.stack[-1][2]: self.messages.append(data) def is_rate_limited(status, payload=None, text=""): """Check HTTP status and explicit error fields, never arbitrary payload data/HTML.""" status = _status(status) if status == 429 or _explicit(payload): return True # Plain response bodies are evidence only on HTTP errors, and must be short. if status >= 400 and isinstance(text, str): plain = text.strip() if len(plain) <= 1024 and not re.search(r"<[^>]+>", plain): return bool(_RATE_MESSAGE.search(plain)) return False def _valid_retry_after(value): if not isinstance(value, str) or not value.strip(): return False value = value.strip() if re.fullmatch(r"[0-9]+", value): return True try: return parsedate_to_datetime(value).tzinfo is not None except (ValueError, TypeError, OverflowError): return False def check_response(response, what="upstream"): """Return response unchanged, or raise RateLimitedError before caller parsing. `what` is deliberately not interpolated: URLs and upstream bodies may contain credentials. Safe to use as a requests response hook through a small lambda. """ if response is None: return response status = _status(getattr(response, "status_code", 0)) message = "上游请求过于频繁，请稍后重试" if status == 429: raise RateLimitedError(message) headers = getattr(response, "headers", {}) or {} retry_after = next((v for k, v in headers.items() if str(k).lower() == "retry-after"), None) if status >= 400 and _valid_retry_after(retry_after): raise RateLimitedError(message) try: payload = response.json() except (ValueError, TypeError, AttributeError): payload = None text = getattr(response, "text", "") or "" if is_rate_limited(status, payload, text): raise RateLimitedError(message) # Login pages can return HTTP 200 with an explicit error element. Never scan # the whole successful HTML document, inline scripts, or announcement text. if payload is None and isinstance(text, str) and "<" in text: parser = _LoginErrors() try: parser.feed(text[:262144]) except (ValueError, TypeError): pass if _RATE_MESSAGE.search(" ".join(parser.messages)[:1024]): raise RateLimitedError(message) return response def safe_error(exception): """Propagate explicit throttling, otherwise return a credential-free diagnostic. Never stringify arbitrary exceptions: requests URLs, query strings, tokens, passwords and JSON decode snippets can all include secrets. """ if isinstance(exception, RateLimitedError): raise exception response = getattr(exception, "response", None) if response is not None: check_response(response, "request error") status = _status(getattr(response, "status_code", 0)) if status: return "upstream HTTP %d" % status return "upstream request or response error (%s)" % type(exception).__name__ # ============================================================================== # Part 1: Pure-python AES-128 # ============================================================================== _SBOX = [ 0x63,0x7c,0x77,0x7b,0xf2,0x6b,0x6f,0xc5,0x30,0x01,0x67,0x2b,0xfe,0xd7,0xab,0x76, 0xca,0x82,0xc9,0x7d,0xfa,0x59,0x47,0xf0,0xad,0xd4,0xa2,0xaf,0x9c,0xa4,0x72,0xc0, 0xb7,0xfd,0x93,0x26,0x36,0x3f,0xf7,0xcc,0x34,0xa5,0xe5,0xf1,0x71,0xd8,0x31,0x15, 0x04,0xc7,0x23,0xc3,0x18,0x96,0x05,0x9a,0x07,0x12,0x80,0xe2,0xeb,0x27,0xb2,0x75, 0x09,0x83,0x2c,0x1a,0x1b,0x6e,0x5a,0xa0,0x52,0x3b,0xd6,0xb3,0x29,0xe3,0x2f,0x84, 0x53,0xd1,0x00,0xed,0x20,0xfc,0xb1,0x5b,0x6a,0xcb,0xbe,0x39,0x4a,0x4c,0x58,0xcf, 0xd0,0xef,0xaa,0xfb,0x43,0x4d,0x33,0x85,0x45,0xf9,0x02,0x7f,0x50,0x3c,0x9f,0xa8, 0x51,0xa3,0x40,0x8f,0x92,0x9d,0x38,0xf5,0xbc,0xb6,0xda,0x21,0x10,0xff,0xf3,0xd2, 0xcd,0x0c,0x13,0xec,0x5f,0x97,0x44,0x17,0xc4,0xa7,0x7e,0x3d,0x64,0x5d,0x19,0x73, 0x60,0x81,0x4f,0xdc,0x22,0x2a,0x90,0x88,0x46,0xee,0xb8,0x14,0xde,0x5e,0x0b,0xdb, 0xe0,0x32,0x3a,0x0a,0x49,0x06,0x24,0x5c,0xc2,0xd3,0xac,0x62,0x91,0x95,0xe4,0x79, 0xe7,0xc8,0x37,0x6d,0x8d,0xd5,0x4e,0xa9,0x6c,0x56,0xf4,0xea,0x65,0x7a,0xae,0x08, 0xba,0x78,0x25,0x2e,0x1c,0xa6,0xb4,0xc6,0xe8,0xdd,0x74,0x1f,0x4b,0xbd,0x8b,0x8a, 0x70,0x3e,0xb5,0x66,0x48,0x03,0xf6,0x0e,0x61,0x35,0x57,0xb9,0x86,0xc1,0x1d,0x9e, 0xe1,0xf8,0x98,0x11,0x69,0xd9,0x8e,0x94,0x9b,0x1e,0x87,0xe9,0xce,0x55,0x28,0xdf, 0x8c,0xa1,0x89,0x0d,0xbf,0xe6,0x42,0x68,0x41,0x99,0x2d,0x0f,0xb0,0x54,0xbb,0x16, ] _INV_SBOX = [0] * 256 for _i, _v in enumerate(_SBOX): _INV_SBOX[_v] = _i _RCON = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1B, 0x36] def _gmul(a, b): p = 0 for _ in range(8): if b & 1: p ^= a hi = a & 0x80 a = (a << 1) & 0xFF if hi: a ^= 0x1B b >>= 1 return p def _expand_key(key): w = [list(key[4 * i:4 * i + 4]) for i in range(4)] for i in range(4, 44): t = list(w[i - 1]) if i % 4 == 0: t = t[1:] + t[:1] t = [_SBOX[b] for b in t] t[0] ^= _RCON[i // 4 - 1] w.append([w[i - 4][j] ^ t[j] for j in range(4)]) return w def _add_round_key(s, w, rnd): for c in range(4): for r in range(4): s[r][c] ^= w[rnd * 4 + c][r] def _sub_bytes(s, box): for r in range(4): for c in range(4): s[r][c] = box[s[r][c]] def _shift_rows(s): for r in range(1, 4): s[r] = s[r][r:] + s[r][:r] def _inv_shift_rows(s): for r in range(1, 4): s[r] = s[r][-r:] + s[r][:-r] def _mix_columns(s): for c in range(4): a = [s[r][c] for r in range(4)] s[0][c] = _gmul(a[0], 2) ^ _gmul(a[1], 3) ^ a[2] ^ a[3] s[1][c] = a[0] ^ _gmul(a[1], 2) ^ _gmul(a[2], 3) ^ a[3] s[2][c] = a[0] ^ a[1] ^ _gmul(a[2], 2) ^ _gmul(a[3], 3) s[3][c] = _gmul(a[0], 3) ^ a[1] ^ a[2] ^ _gmul(a[3], 2) def _inv_mix_columns(s): for c in range(4): a = [s[r][c] for r in range(4)] s[0][c] = _gmul(a[0],14) ^ _gmul(a[1],11) ^ _gmul(a[2],13) ^ _gmul(a[3],9) s[1][c] = _gmul(a[0],9) ^ _gmul(a[1],14) ^ _gmul(a[2],11) ^ _gmul(a[3],13) s[2][c] = _gmul(a[0],13) ^ _gmul(a[1],9) ^ _gmul(a[2],14) ^ _gmul(a[3],11) s[3][c] = _gmul(a[0],11) ^ _gmul(a[1],13) ^ _gmul(a[2],9) ^ _gmul(a[3],14) class AES128: def __init__(self, key): if len(key) != 16: raise ValueError("AES-128 requires a 16-byte key, got %d" % len(key)) self.w = _expand_key(key) def encrypt_block(self, block): s = [[block[r + 4 * c] for c in range(4)] for r in range(4)] _add_round_key(s, self.w, 0) for rnd in range(1, 10): _sub_bytes(s, _SBOX) _shift_rows(s) _mix_columns(s) _add_round_key(s, self.w, rnd) _sub_bytes(s, _SBOX) _shift_rows(s) _add_round_key(s, self.w, 10) return bytes(s[r][c] for c in range(4) for r in range(4)) def decrypt_block(self, block): s = [[block[r + 4 * c] for c in range(4)] for r in range(4)] _add_round_key(s, self.w, 10) for rnd in range(9, 0, -1): _inv_shift_rows(s) _sub_bytes(s, _INV_SBOX) _add_round_key(s, self.w, rnd) _inv_mix_columns(s) _inv_shift_rows(s) _sub_bytes(s, _INV_SBOX) _add_round_key(s, self.w, 0) return bytes(s[r][c] for c in range(4) for r in range(4)) def _pkcs7_unpad(data): if not data: return data n = data[-1] if 1 <= n <= 16 and data[-n:] == bytes([n]) * n: return data[:-n] return data def _zero_pad(data): if len(data) % 16: data += b"\x00" * (16 - len(data) % 16) return data def aes_ecb_decrypt_b64(cipher_b64, key_text): aes = AES128(key_text[:16].encode("utf-8")) raw = base64.b64decode(cipher_b64) out = b"".join(aes.decrypt_block(raw[i:i + 16]) for i in range(0, len(raw), 16)) return _pkcs7_unpad(out) def aes_cbc_encrypt_b64_zero(plain, key_text): key = key_text[:16].encode("utf-8") aes = AES128(key) iv = key data = _zero_pad(plain.encode("utf-8")) out, prev = b"", iv for i in range(0, len(data), 16): blk = bytes(x ^ y for x, y in zip(data[i:i + 16], prev)) prev = aes.encrypt_block(blk) out += prev return base64.b64encode(out).decode("ascii") # ============================================================================== # Part 2: Session & URL parsing # ============================================================================== def new_session(): s = requests.Session() s.trust_env = False s.verify = False s.headers.update({"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"}) return s def parse_course_url(url): parts = urllib.parse.urlsplit(str(url or "")) m = re.fullmatch( r"/course/detail/(?P[0-9A-Za-z]+)/(?P[^/]+)/" r"(?P\d+)/signin/(?P\d+)/(?P\d+)/?", parts.path) try: port = parts.port except ValueError: port = -1 if (parts.scheme != "https" or parts.username is not None or parts.password is not None or port not in (None, 443) or (parts.hostname or "").lower() != "ginkgostu.dufe.edu.cn" or not m): raise ValueError( "URL is not a sign-in page (missing /signin//)") return m.groupdict() # ============================================================================== # Part 3: Login flow # ============================================================================== def get_unify_config(sess): r = sess.get(GINKGO + "/student/getLoginUnifySet", timeout=20) r.raise_for_status() try: return r.json()["data"] except (ValueError, KeyError) as e: raise UpstreamError("Malformed getLoginUnifySet response: %s" % e) def open_authorize(sess, cfg): params = { "response_type": cfg["responseType"], "client_id": cfg["clientId"], "redirect_uri": cfg["rollBackUrl"], "scope": cfg.get("scope", "") or "", "state": cfg.get("state", "") or "", } r = sess.get(cfg["loginUrl"], params=params, timeout=20, allow_redirects=True) r.raise_for_status() m = re.search(r'name="__token__"\s+value="([^"]+)"', r.text) if not m: raise UpstreamError("CSRF token __token__ not found on login page") return m.group(1), r.url def exploit_slider(sess, referer): url = (SSO + "/auth/widget?layer=image_verify&widget=Slider" "&action=get_verify&type=login_image_verify") r = sess.post(url, data={"width": 280, "height": 155, "block_size": 80}, headers={"X-Requested-With": "XMLHttpRequest", "Referer": referer}, timeout=20) r.raise_for_status() try: d = r.json()["data"] token, cipher = d["token"], d["d"] except (ValueError, KeyError, TypeError) as e: raise UpstreamError("Malformed slider challenge response: %s" % e) x = aes_ecb_decrypt_b64(cipher, token).decode("utf-8", "replace") return {"token": token, "x": x, "cipher": cipher} def submit_login(sess, post_url, csrf, username, password, slider, referer): form = { "fingerprint": uuid.uuid4().hex, "username": username, "password": aes_cbc_encrypt_b64_zero(password, csrf), "verify_token": slider["token"], "verify_code": slider["x"], "__token__": csrf, } return sess.post(post_url, data=form, timeout=25, allow_redirects=False, headers={"Referer": referer, "Origin": SSO, "Content-Type": "application/x-www-form-urlencoded"}) def exchange_token(sess, cfg, code): r = sess.post(GINKGO + "/student/authLogin", data={"code": code}, timeout=20, headers={"Content-Type": "application/x-www-form-urlencoded"}) try: j = r.json() except ValueError: raise UpstreamError("authLogin returned non-JSON") from None if not isinstance(j, dict) or j.get("code") != "2000": raise AuthError("OAuth code exchange refused") return j["data"]["token"] def login(username, password): """Complete SSO login flow. Returns (sess, token). auto 模式一次登录后复用同一 session/token 逐个活动签到， 对齐 bot ginkgo 复用轮询缓存登录态的思路（请求内级）。""" sess = new_session() cfg = get_unify_config(sess) csrf, page_url = open_authorize(sess, cfg) slider = exploit_slider(sess, page_url) resp = submit_login(sess, page_url, csrf, username, password, slider, page_url) loc = resp.headers.get("Location", "") m = re.search(r"[?&]code=([A-Za-z0-9_\-]+)", loc) if not m: raise AuthError( "Login failed: no OAuth code in redirect (HTTP %s)" % resp.status_code) token = exchange_token(sess, cfg, m.group(1)) return sess, token # ============================================================================== # Part 4: Sign-in API # ============================================================================== def _request(sess, method, *args, **kwargs): """Also guard fake sessions and HTTPError.response before any JSON parsing.""" try: response = getattr(sess, method)(*args, **kwargs) except requests.RequestException as exc: raise UpstreamError(safe_error(exc)) from None return check_response(response, "signin") def api_headers(token, fingerprint=None): return { "User-Agent": UA, "Origin": GINKGO, "Referer": GINKGO + "/", "Content-Type": "application/x-www-form-urlencoded", "token": token, "fingerprint": fingerprint or uuid.uuid4().hex, } def call_api(sess, path, data, token): return _request(sess, "post", GINKGO + path, data=data, timeout=20, headers=api_headers(token)) def commit_sign_in(sess, token, sign_in_id, batch_no, referer, code): """ POST /student/signIn/InTheSign to actually check in with the given code. Mirrors the browser request: headers: batchno, fingerprint, token, Referer (course page) body: signInId=&code= Returns (ok: bool, message: str, raw: dict|None). """ headers = api_headers(token) if batch_no: # NOTE: the browser sends a "batchno" header that does not always equal # the batchNo segment of the course URL. We pass the URL segment here; # the server appears not to enforce it. headers["batchno"] = str(batch_no) headers["Referer"] = referer try: r = _request(sess, "post", GINKGO + "/student/signIn/InTheSign", data={"signInId": str(sign_in_id), "code": str(code)}, timeout=20, headers=headers, ) except (requests.RequestException, UpstreamError) as e: return False, "request to InTheSign failed: %s" % safe_error(e), None if r.status_code != 200: return False, "InTheSign returned HTTP %d" % r.status_code, None try: j = r.json() except ValueError: return False, "InTheSign returned non-JSON", None if not isinstance(j, dict): return False, "Malformed InTheSign response", None api_code = str(j.get("code", "")) if api_code == "2000": return True, "accepted", j # Never return/log arbitrary upstream messages or bodies: they may contain secrets. return False, "InTheSign refused the sign-in request", j def fetch_sign_in_info(username, password, course_url, do_commit=True, sess=None, token=None): route = parse_course_url(course_url) sign_in_id = route["signInId"] batch_no = route.get("batchNo") if sess is None or not token: # 无外部登录态时自登录（保持向后兼容）：直接复用上方 SSO 流程 sess, token = login(username, password) # 否则：sess/token 由调用方提供（auto 模式请求内复用同一登录态），跳过 SSO 登录 r = call_api(sess, "/student/signIn/getSignInInfo", {"signInId": sign_in_id}, token) check_response(r, "getSignInInfo") try: j = r.json() except ValueError: raise UpstreamError("getSignInInfo returned non-JSON") from None if not isinstance(j, dict): raise UpstreamError("Malformed getSignInInfo response") data = j.get("data") or {} if not isinstance(data, dict): raise UpstreamError("Malformed getSignInInfo data") model = data.get("signInModel") or {} if not isinstance(model, dict): raise UpstreamError("Malformed getSignInInfo model") code = model.get("code") raw_type = model.get("type") if raw_type is None or raw_type == "": raw_type = route.get("type") try: signin_type = int(raw_type) except (TypeError, ValueError): raise UpstreamError("Malformed sign-in type") from None type_name = signin_type_label(signin_type) if signin_type == 1 and (code is None or code == ""): raise NoSignInCodeError("数字码签到未配置签到码") if signin_type == 2: code = "" result = { "signInId": sign_in_id, "signInCode": code, "type": signin_type, "typeName": type_name, "title": model.get("title"), "status": data.get("signStatus"), } if do_commit: if signin_type not in (1, 2): result["commit"] = { "success": False, "message": "%s暂不支持自动签到，需特定流程" % type_name, } else: canonical_referer = GINKGO + urllib.parse.urlsplit(course_url).path ok, reason, _raw = commit_sign_in( sess, token, sign_in_id, batch_no, canonical_referer, code ) result["commit"] = {"success": ok, "message": reason} return result # ============================================================================== # Part 4b: Auto sign-in (mode=auto) # 通知列表发现签到活动并逐个自动签到 —— 逻辑对齐 bot ginkgo 模块： # fetch_all（通知+互动消息）→ _validated_signin_item（busi + 文本含"签到" + # 可解析课程链接）→ 按 signInId 去重 → 每活动连续重试（首次+最多5次，间隔3s， # token 失效重登录一次）→ 结果聚合为列表返回。 # ============================================================================== _TAG_RE = re.compile(r"<[^>]+>") def _html_to_text(raw): if not raw: return "" text = re.sub(r"(?i)", "\n", raw) text = re.sub(r"(?i)", "\n", text) text = _TAG_RE.sub("", text) text = _html.unescape(text) text = re.sub(r"[ \t\u3000]+", " ", text) text = re.sub(r"\n{2,}", "\n", text) return text.strip() def _parse_ts(value): text = (value or "").strip() if not text: return 0.0 if text.endswith("Z"): text = text[:-1] + "+00:00" m = re.search(r"([+-]\d{2})(\d{2})$", text) if m: text = text[:m.start()] + m.group(1) + ":" + m.group(2) try: dt = datetime.fromisoformat(text) except ValueError: try: dt = datetime.strptime(text[:19], "%Y-%m-%d %H:%M:%S") except ValueError: return 0.0 if dt.tzinfo is None: dt = dt.replace(tzinfo=TZ8) return dt.timestamp() def _normalize_notice(raw): title = (raw.get("title") or "").strip() or "(无标题)" ts = _parse_ts(raw.get("createTime")) return { "kind": "notice", "key": "N%s" % raw.get("id"), "id": str(raw.get("id") or ""), "title": title, "text": _html_to_text(raw.get("content") or ""), "ts": ts, "link": NOTICE_PAGE, "jump": "", } def _normalize_busi(raw): text = _html_to_text(raw.get("content") or "") jump = (raw.get("jumpUrl") or "").strip() if jump.startswith("http"): link = jump elif jump.startswith("/"): link = GINKGO + jump else: link = SYSTEM_PAGE ts = _parse_ts(raw.get("createTime")) return { "kind": "busi", "key": "B%s" % (raw.get("uuid") or ""), "uuid": str(raw.get("uuid") or ""), "id": "", "title": text[:120] or "(无内容)", "text": text, "ts": ts, "link": link, "jump": jump, } def _api_headers(token): return { "User-Agent": UA, "Origin": GINKGO, "Referer": GINKGO + "/", "Content-Type": "application/x-www-form-urlencoded", "token": token, "fingerprint": uuid.uuid4().hex, } def _call_json(sess, token, method, path, params=None, body=None, what="接口"): """消息类 API 调用（对齐 bot _call）：check_response + code 必须为 2000。""" headers = _api_headers(token) if method == "GET": r = sess.get(GINKGO + path, params=params, headers=headers, timeout=20) else: headers["Content-Type"] = "application/json" r = sess.post(GINKGO + path, data=json.dumps(body if body is not None else {}), headers=headers, timeout=20) check_response(r, what) try: payload = r.json() except ValueError: raise UpstreamError("%s returned non-JSON (HTTP %s)" % (what, r.status_code)) if not isinstance(payload, dict): raise UpstreamError("%s JSON must be an object" % what) if is_rate_limited(None, payload): raise RateLimitedError("%s 触发频率限制" % what) if str(payload.get("code")) != "2000": raise UpstreamError("%s returned code=%s" % (what, payload.get("code"))) return payload def fetch_message_all(sess, token, page_size=PAGE_SIZE): """取两类消息（均为全部，含已读），返回 (notices, busi) 归一化列表。""" j1 = _call_json(sess, token, "GET", "/student/message/getMessageAll", params={"pageSize": page_size, "pageNo": 1, "readFlag": 0}, what="通知列表") notices = [_normalize_notice(x) for x in ((j1.get("data") or {}).get("list") or [])] j2 = _call_json(sess, token, "POST", "/student/busiMessage/list", body={}, what="互动消息列表") busi = [_normalize_busi(x) for x in (j2.get("data") or [])] return notices, busi def _validated_signin_item(item): """Return an annotated copy only for a real interactive sign-in activity. 与 bot ginkgo 的 _validated_signin_item 同逻辑：仅互动消息、文本含"签到"、 跳转链接可解析为课程签到路由。query 参数可能携带 token，不入库不回显。""" if (not isinstance(item, dict) or item.get("kind") != "busi" or "签到" not in str(item.get("text") or "")): return None link = str(item.get("link") or "").strip() try: route = parse_course_url(link) except (ValueError, TypeError, AttributeError): return None row = dict(item) # Persist/display only the canonical path: query parameters may carry tokens # and must never reach receipts or push logs. row["link"] = GINKGO + urllib.parse.urlsplit(link).path # signInId is the stable activity identity（同一活动的多条消息去重） row["signin_activity_key"] = "S%s" % route["signInId"] try: row["signin_type"] = int(route["type"]) except (TypeError, ValueError, KeyError): return None row["signin_type_name"] = signin_type_label(row["signin_type"]) return row def _signin_one_with_retry(username, password, row, sess, token): """对单个活动执行签到：失败连续重试（首次 + 最多 5 次，间隔 3s）， token 疑似失效(上游错误/登录错)时重新登录一次后立即重试（对齐 bot refresh 语义）。 与 bot complete_signin 一致：除限流（外层中断）外，任何未确认成功都消耗重试次数； 重试用尽记最终失败。entry["attempts"] 为逐次完整记录列表（每次的结果与原因， relogin 后立即重试的那次带 "relogin": true）。返回 (entry, sess, token)。""" link = row["link"] base = { "activityKey": row.get("signin_activity_key"), "signInId": parse_course_url(link).get("signInId"), "routeType": row.get("signin_type"), "routeTypeName": row.get("signin_type_name"), "link": link, } attempts = 0 refreshed = False relogin_before_next = False # 上一次迭代发生了重登录，下一次尝试标记 relogin last_err = "unknown error" last_data = None attempt_log = [] # 逐次记录：{"n", "success", "message"[, "relogin"]} while attempts < MAX_SIGNIN_ATTEMPTS: attempts += 1 record = {"n": attempts} if relogin_before_next: record["relogin"] = True relogin_before_next = False try: data = fetch_sign_in_info(username, password, link, do_commit=True, sess=sess, token=token) last_data = data commit = data.get("commit") if isinstance(commit, dict) and commit.get("success") is True: record["success"] = True record["message"] = "签到成功" attempt_log.append(record) entry = dict(base) entry.update({ "type": data.get("type"), "typeName": data.get("typeName"), "title": data.get("title"), "signInCode": data.get("signInCode"), "status": data.get("status"), "success": True, "message": "签到成功", "attempts": attempt_log, }) return entry, sess, token message = ((commit or {}).get("message") or "no commit result returned") last_err = str(message) record["success"] = False record["message"] = last_err except RateLimitedError: raise except (UpstreamError, AuthError) as e: # 疑似 token 失效：重登录一次后立即重试（不消耗退避间隔，对齐 bot） if not refreshed: refreshed = True try: sess, token = login(username, password) record["success"] = False record["message"] = str(e) or type(e).__name__ attempt_log.append(record) relogin_before_next = True continue except Exception: pass # 重登录也失败：落入常规退避重试 last_err = str(e) or type(e).__name__ record["success"] = False record["message"] = last_err except NoSignInCodeError as e: last_err = str(e) record["success"] = False record["message"] = last_err except Exception as e: last_err = safe_error(e) record["success"] = False record["message"] = last_err attempt_log.append(record) if attempts < MAX_SIGNIN_ATTEMPTS: time.sleep(SIGNIN_BACKOFF[min(attempts - 1, len(SIGNIN_BACKOFF) - 1)]) entry = dict(base) if isinstance(last_data, dict): entry.update({ "type": last_data.get("type"), "typeName": last_data.get("typeName"), "title": last_data.get("title"), "signInCode": last_data.get("signInCode"), "status": last_data.get("status"), }) entry.update({"success": False, "message": "签到失败（重试 %d 次）: %s" % (attempts, last_err), "attempts": attempt_log}) return entry, sess, token def auto_sign_in_all(username, password): """mode=auto 主流程。返回聚合结果 dict（results 为列表）。""" sess, token = login(username, password) notices, busi = fetch_message_all(sess, token) activities = {} for item in notices + busi: row = _validated_signin_item(item) if row: activities.setdefault(row["signin_activity_key"], row) results = [] rate_limited = False for key in sorted(activities, key=lambda k: activities[k].get("ts") or 0): row = activities[key] try: entry, sess, token = _signin_one_with_retry( username, password, row, sess, token) except RateLimitedError: # 对齐 bot：限流即中断本轮，已完成的结果照常返回 rate_limited = True break results.append(entry) succeeded = sum(1 for r in results if r.get("success")) return { "mode": "auto", "activities": len(activities), "total": len(results), "succeeded": succeeded, "failed": len(results) - succeeded, "rate_limited": rate_limited, "results": results, } # ============================================================================== # Part 5: HTTP service (WSGI app) # ============================================================================== app = Flask(__name__) def _b64_decode(value, field): """ Decode a base64 query/form parameter tolerantly. Handles, in order: * '%3D' / '%2B' / '%20' -- iOS Shortcuts and some HTTP clients URL-encode the base64 payload (notably the trailing '='), so we percent-decode first. * '+' turned into ' ' by application/x-www-form-urlencoded. * trailing newline from `base64` CLI without -w0. * both standard (+/) and urlsafe (-_) alphabets. """ v = urllib.parse.unquote(value or "") v = v.strip() v = v.replace(" ", "+") v = v.replace("\r", "").replace("\n", "") if not v: raise ValueError("Empty base64 for '%s'" % field) try: return base64.b64decode(v, altchars=b"-_", validate=True).decode("utf-8") except (binascii.Error, UnicodeDecodeError, ValueError) as e: raise ValueError("Invalid base64 for '%s': %s" % (field, e)) def _error(http_status, code, msg): return jsonify({"code": code, "msg": msg, "data": None}), http_status @app.route("/dufe/ginkgostu_sign_in", methods=["GET", "POST"]) def ginkgostu_sign_in(): args = request.values username = (args.get("username") or "").strip() password_b64 = args.get("password") or "" url_b64 = args.get("url") or "" commit_arg = (args.get("commit") or "true").strip().lower() do_commit = commit_arg not in ("0", "false", "no") mode = (args.get("mode") or "").strip().lower() if not (username and password_b64): return _error(400, 400, "Missing parameters: username, password(base64)") if mode != "auto" and not url_b64: return _error(400, 400, "Missing parameters: username, password(base64), url(base64)") try: password = _b64_decode(password_b64, "password") course_url = _b64_decode(url_b64, "url") if url_b64 else "" except ValueError as e: return _error(400, 400, str(e)) if mode == "auto": return _auto_response(username, password) log.info("query start user=%s commit=%s", username[:2] + "***", do_commit) try: data = fetch_sign_in_info(username, password, course_url, do_commit=do_commit) if do_commit: # Fail-closed: only an explicit success=true counts as signed in. commit_info = data.get("commit") if not isinstance(commit_info, dict) or commit_info.get("success") is not True: reason = (commit_info.get("message") if isinstance(commit_info, dict) else None) or "no commit result returned" log.warning("sign-in commit failed user=%s reason=%s", username[:2] + "***", reason) return jsonify({ "code": 200, "msg": "sign-in failed: %s" % reason, "data": data, }) log.info("query ok user=%s type=%s", username[:2] + "***", data.get("type")) return jsonify({"code": 200, "msg": "ok", "data": data}) except AuthError: log.warning("authentication failed user=%s", username[:2] + "***") return _error(401, 401, "Authentication failed") except NoSignInCodeError: log.info("numeric code missing user=%s", username[:2] + "***") return _error(404, 404, "数字码签到未配置签到码") except UpstreamError: log.error("upstream response error user=%s", username[:2] + "***") return _error(502, 502, "Upstream response error") except ValueError: log.warning("bad request user=%s", username[:2] + "***") return _error(400, 400, "Invalid request") except requests.RequestException: log.error("upstream request failed user=%s", username[:2] + "***") return _error(502, 502, "Upstream request failed") except Exception as exc: log.error("internal error type=%s", type(exc).__name__) return _error(500, 500, "Internal error") def _auto_response(username, password): """mode=auto：发现 + 自动签到 + 聚合列表返回。""" log.info("auto sign-in start user=%s", username[:2] + "***") try: data = auto_sign_in_all(username, password) except AuthError: log.warning("auto sign-in authentication failed user=%s", username[:2] + "***") return _error(401, 401, "Authentication failed") except RateLimitedError: log.warning("auto sign-in rate limited user=%s", username[:2] + "***") return _error(429, 429, "Upstream rate limited, please retry later") except UpstreamError: log.error("auto sign-in upstream error user=%s", username[:2] + "***") return _error(502, 502, "Upstream response error") except requests.RequestException: log.error("auto sign-in upstream request failed user=%s", username[:2] + "***") return _error(502, 502, "Upstream request failed") except Exception as exc: log.error("auto sign-in internal error type=%s", type(exc).__name__) return _error(500, 500, "Internal error") log.info("auto sign-in done user=%s total=%s succeeded=%s rate_limited=%s", username[:2] + "***", data["total"], data["succeeded"], data["rate_limited"]) if data["rate_limited"]: msg = ("partial: upstream rate limited, %d/%d activities processed" % (data["total"], data["activities"])) elif data["failed"]: msg = "ok with failures: %d/%d succeeded" % (data["succeeded"], data["total"]) else: msg = "ok" return jsonify({"code": 200, "msg": msg, "data": data}) @app.route("/healthz", methods=["GET"]) def healthz(): return jsonify({"code": 200, "msg": "ok", "data": None}) # ============================================================================== # Part 6: WSGI entrypoint # ============================================================================== HOST = "127.0.0.1" PORT = 9005 THREADS = 16 BACKLOG = 128 CONNECTION_LIMIT = 256 CHANNEL_TIMEOUT = 120 def main(): log.info("serving on http://%s:%d (threads=%d)", HOST, PORT, THREADS) serve( app, host=HOST, port=PORT, threads=THREADS, backlog=BACKLOG, connection_limit=CONNECTION_LIMIT, channel_timeout=CHANNEL_TIMEOUT, ident="ginkgo-signin", ) return 0 if __name__ == "__main__": sys.exit(main())