from __future__ import annotations import base64 import json import re import time import urllib.error import urllib.request from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from dataclasses import dataclass from pathlib import Path from typing import Any, Callable from selenium.common.exceptions import NoSuchElementException, TimeoutException, WebDriverException from selenium.webdriver.common.by import By from selenium.webdriver.remote.webdriver import WebDriver from selenium.webdriver.remote.webelement import WebElement from selenium.webdriver.support.wait import WebDriverWait import onnx_inference import webdriver_utils from core.db import Database from core.login_modes import LOGIN_MODE_LABELS, LOGIN_MODE_UNIFIED, LOGIN_MODE_ZHJW_DIRECT, normalize_login_mode from core.task_modes import ( TASK_RUN_MODE_LABELS, TASK_RUN_MODE_OPTIMIZED, TASK_RUN_MODE_STABLE, normalize_task_run_mode, ) URL_LOGIN = "http://id.scu.edu.cn/enduser/sp/sso/scdxplugin_jwt23?enterpriseId=scdx&target_url=index" URL_ZHJW_LOGIN = "http://zhjw.scu.edu.cn/login" URL_SELECT_COURSE = "http://zhjw.scu.edu.cn/student/courseSelect/courseSelect/index" URL_CURRICULUM_CALLBACK = "http://zhjw.scu.edu.cn/student/courseSelect/thisSemesterCurriculum/callback" LOGIN_SUCCESS_PREFIXES = ( "http://zhjw.scu.edu.cn/index", "http://zhjw.scu.edu.cn/", "https://zhjw.scu.edu.cn/index", "https://zhjw.scu.edu.cn/", ) CATEGORY_META = { "plan": {"label": "方案选课", "tab_id": "faxk"}, "free": {"label": "自由选课", "tab_id": "zyxk"}, } ACCOUNT_LOGIN_TAB_SELECTORS = [ (By.CSS_SELECTOR, ".login-tab .login-tab-item:nth-child(3)"), (By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' login-tab-item ') and (normalize-space(.)='账号登录' or normalize-space(.)='Account')]"), (By.XPATH, "//*[self::div or self::span or self::a or self::li or self::button][normalize-space(.)='账号登录']"), (By.XPATH, "//*[self::div or self::span or self::a or self::li or self::button][normalize-space(.)='Account']"), ] LOGIN_STUDENT_SELECTORS = [ (By.CSS_SELECTOR, ".login-content input[type='text']:not([placeholder*='验证码']):not([placeholder*='Captcha'])"), (By.CSS_SELECTOR, "input[placeholder='请输入学(工)号']"), (By.CSS_SELECTOR, "input[placeholder='Campus ID']"), (By.XPATH, "//*[@id='app']//form//input[@type='text']"), (By.XPATH, "//*[@id='app']/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[1]/div/div/div[2]/div/input"), ] LOGIN_PASSWORD_SELECTORS = [ (By.CSS_SELECTOR, ".login-content input[type='password']"), (By.CSS_SELECTOR, "input[type='password'][placeholder='请输入密码']"), (By.CSS_SELECTOR, "input[type='password'][placeholder='Password']"), (By.CSS_SELECTOR, "input[placeholder='请输入密码']"), (By.CSS_SELECTOR, "input[placeholder='Password']"), (By.XPATH, "//*[@id='app']//form//input[@type='password']"), (By.XPATH, "//*[@id='app']/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[2]/div/div/div[2]/div/input"), ] LOGIN_CAPTCHA_INPUT_SELECTORS = [ (By.CSS_SELECTOR, ".captcha-box input"), (By.CSS_SELECTOR, "input[placeholder='请输入验证码']"), (By.CSS_SELECTOR, "input[placeholder='Captcha code']"), (By.CSS_SELECTOR, "input[placeholder*='Captcha']"), (By.XPATH, "//*[@id='app']/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[3]//input"), (By.XPATH, "//*[@id='app']//form//input[contains(@placeholder, '验证码')]"), (By.XPATH, "//*[@id='app']//form//input[contains(translate(@placeholder, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'captcha')]"), ] LOGIN_CAPTCHA_IMAGE_SELECTORS = [ (By.CSS_SELECTOR, "img.captcha-img"), (By.CSS_SELECTOR, ".captcha-box img"), (By.XPATH, "//*[@id='app']/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[3]//img"), (By.XPATH, "//*[@id='app']//form//img"), ] LOGIN_BUTTON_SELECTORS = [ (By.CSS_SELECTOR, "button.login-btn"), (By.XPATH, "//button[normalize-space(.)='登录']"), (By.XPATH, "//button[normalize-space(.)='Log on']"), (By.XPATH, "//*[@id='app']//form//button"), (By.XPATH, "//*[@id='app']/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[4]/div/button"), ] ZHJW_LOGIN_STUDENT_SELECTORS = [ (By.ID, "input_username"), (By.CSS_SELECTOR, "input[name='username']"), (By.CSS_SELECTOR, "input[type='text']"), ] ZHJW_LOGIN_PASSWORD_SELECTORS = [ (By.ID, "input_password"), (By.CSS_SELECTOR, "input[name='password']"), (By.CSS_SELECTOR, "input[type='password']"), ] ZHJW_LOGIN_CAPTCHA_INPUT_SELECTORS = [ (By.ID, "input_checkcode"), (By.CSS_SELECTOR, "input[name='checkcode']"), (By.CSS_SELECTOR, "input[name*='captcha' i]"), (By.CSS_SELECTOR, "input[id*='captcha' i]"), ] ZHJW_LOGIN_CAPTCHA_IMAGE_SELECTORS = [ (By.ID, "captchaImg"), (By.CSS_SELECTOR, "img[src*='captcha' i]"), (By.CSS_SELECTOR, "img[id*='captcha' i]"), ] ZHJW_LOGIN_BUTTON_SELECTORS = [ (By.ID, "loginButton"), (By.CSS_SELECTOR, "input[type='submit']"), (By.CSS_SELECTOR, "button[type='submit']"), (By.XPATH, "//*[self::button or self::input][contains(normalize-space(.), '登录') or @value='登录']"), ] LOGIN_2FA_SEND_CODE_SELECTORS = [ (By.CSS_SELECTOR, ".otp-layout .btn-content button"), (By.CSS_SELECTOR, ".btn-content button"), (By.CSS_SELECTOR, ".security-auth-content button"), (By.XPATH, "//button[.//span[normalize-space(.)='获取验证码'] or normalize-space(.)='获取验证码']"), (By.XPATH, "//button[.//span[normalize-space(.)='Send email'] or normalize-space(.)='Send email']"), (By.XPATH, "//button[contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'send')]"), (By.XPATH, "//*[self::button or self::a or @role='button'][contains(., '获取验证码') or contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'send')]"), ] LOGIN_2FA_CODE_INPUT_SELECTORS = [ (By.CSS_SELECTOR, ".sec-factor-six-ipt .inner input[type='text']"), (By.CSS_SELECTOR, ".security-auth-content input[type='text']"), (By.CSS_SELECTOR, "input[type='text'][maxlength='1']"), ] LOGIN_2FA_PRIMARY_CODE_INPUT_SELECTOR = ".sec-factor-six-ipt .inner input[type='text']" LOGIN_2FA_VALIDATION_REQUEST_PATH = "/2factor/otpcode" LOGIN_2FA_SEND_SMS_REQUEST_PATH = "/2factor/send_sms" LOGIN_2FA_NETWORK_TRACE_PATHS = ( LOGIN_2FA_SEND_SMS_REQUEST_PATH, LOGIN_2FA_VALIDATION_REQUEST_PATH, "/2factor/sms/idv/user_info", "/commons/user_password_info", "/commons/role_permissions", ) SUBMIT_CAPTCHA_IMAGE_SELECTORS = [ (By.XPATH, "//div[contains(@class,'dialog') or contains(@class,'modal') or contains(@class,'popup')]//img[contains(@src,'base64') or contains(translate(@src,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'captcha')]"), (By.XPATH, "//img[contains(@src,'base64')]"), (By.XPATH, "//img[contains(translate(@alt,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'captcha') or contains(translate(@class,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'captcha')]"), ] SUBMIT_CAPTCHA_INPUT_SELECTORS = [ (By.XPATH, "//input[contains(@placeholder,'验证码')]"), (By.XPATH, "//input[contains(translate(@id,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'captcha')]"), (By.XPATH, "//input[contains(translate(@name,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'captcha')]"), ] SUBMIT_CAPTCHA_BUTTON_SELECTORS = [ (By.XPATH, "//button[contains(.,'确定') or contains(.,'提交') or contains(.,'确认') or contains(.,'验证')]"), (By.XPATH, "//input[(@type='button' or @type='submit') and (contains(@value,'确定') or contains(@value,'提交') or contains(@value,'确认') or contains(@value,'验证'))]"), ] MODE_DOWNGRADE_THRESHOLD = 3 STABLE_UI_SETTLE_SECONDS = 0.2 OPTIMIZED_UI_SETTLE_SECONDS = 0.08 LOGIN_2FA_POST_SUBMIT_TIMEOUT_SECONDS = 360 LOGIN_2FA_WAIT_LOG_INTERVAL_SECONDS = 15 LOGIN_2FA_SSO_ADVANCE_AFTER_SECONDS = 12 class FatalCredentialsError(Exception): pass class RecoverableAutomationError(Exception): pass @dataclass(slots=True) class TaskResult: status: str error: str = "" class CourseBot: def __init__( self, *, config, store: Database, task_id: int, user: dict, password: str, logger: Callable[[str, str], None], login_2fa_code_provider: Callable[[str, Any], str] | None = None, ) -> None: self.config = config self.store = store self.task_id = task_id self.user = user self.password = password self.logger = logger self.login_2fa_code_provider = login_2fa_code_provider self.root_dir = Path(__file__).resolve().parent.parent self.select_course_js = (self.root_dir / "javascript" / "select_course.js").read_text(encoding="utf-8") self.check_result_js = (self.root_dir / "javascript" / "check_result.js").read_text(encoding="utf-8") self.captcha_solver = onnx_inference.CaptchaONNXInference( model_path=str(self.root_dir / "ocr_provider" / "captcha_model.onnx") ) if getattr(self.captcha_solver, "ddddocr", None) is None: detail = str(getattr(self.captcha_solver, "ddddocr_error", "") or "未加载 DDDDOCR") self.logger("WARNING", f"DDDDOCR 未启用，将使用旧 ONNX 验证码模型兜底。详情: {detail}") task = self.store.get_task(task_id) or {} self.selection_attempts = int(task.get("total_attempts") or 0) self.requested_mode = normalize_task_run_mode(task.get("requested_mode")) self.effective_mode = normalize_task_run_mode(task.get("effective_mode")) self.login_mode = normalize_login_mode(task.get("login_mode")) self.use_browser_proxy = bool(int(task.get("use_proxy") or 0)) self.login_mode_fallback_to_unified = False self.mode_failure_count = 0 self.active_category: str | None = None self.select_course_page_ready = False self.stop_event = None self._login_2fa_network_events: dict[str, dict[str, Any]] = {} self.browser_proxy_disabled_reason = "" def run(self, stop_event) -> TaskResult: self.stop_event = stop_event self._refresh_effective_mode_from_store(log_change=False) initial_courses = self.store.list_courses_for_user(self.user["id"]) self.logger( "INFO", "任务运行配置：" f"请求模式={self._mode_label(self.requested_mode)}，" f"当前模式={self._mode_label(self.effective_mode)}，" f"登录方式={self._login_mode_label(self.login_mode)}，" f"代理={'启用' if self.use_browser_proxy else '关闭'}，" f"刷新间隔={self._current_poll_interval_seconds()} 秒，" f"待选课程={self._course_batch_summary(initial_courses)}。", ) exhausted_sessions = 0 while not stop_event.is_set(): courses = self.store.list_courses_for_user(self.user["id"]) if not courses: self.logger("INFO", "当前没有待选课程，任务直接完成。") return TaskResult(status="completed") driver: WebDriver | None = None session_ready = False session_error_count = 0 self._invalidate_course_page_cache() session_proxy_url = "" try: self._ensure_browser_proxy_available() session_proxy_url = self._current_browser_proxy_url() enable_login_2fa_bypass = self.login_mode == LOGIN_MODE_UNIFIED and self.effective_mode == TASK_RUN_MODE_OPTIMIZED self.logger( "INFO", f"准备启动 Selenium 会话，浏览器重启计数 {exhausted_sessions}/{self.config.selenium_restart_limit}，" f"代理={self._browser_proxy_label()}。", ) driver = webdriver_utils.configure_browser( chrome_binary=self.config.chrome_binary, chromedriver_path=self.config.chromedriver_path, page_timeout=self.config.browser_page_timeout, proxy_url=session_proxy_url, enable_login_2fa_bypass=enable_login_2fa_bypass, ) wait = WebDriverWait(driver, self.config.browser_page_timeout, 0.5) if enable_login_2fa_bypass: self.logger("INFO", "优化版已启用统一认证 2FA 绕过脚本；如未生效将继续使用短信验证码辅助流程。") self.logger("INFO", f"Selenium 会话已启动，准备执行{self._login_mode_label(self.login_mode)}登录。") while not stop_event.is_set(): try: if not session_ready: self._login_for_task(driver, wait) session_ready = True self._invalidate_course_page_cache() self.logger("INFO", f"{self._login_mode_label(self.login_mode)}登录完成，准备进入选课流程。") self._refresh_effective_mode_from_store() current_courses = self.store.list_courses_for_user(self.user["id"]) if not current_courses: self.logger("INFO", "待选课程已全部完成或被移除，任务完成。") return TaskResult(status="completed") round_stats = self._run_selection_round(driver, wait, stop_event, current_courses) if round_stats["not_selectable"]: session_error_count = 0 self._sleep_with_cancel( stop_event, self._current_poll_interval_seconds(), "当前不是选课时段或选课入口暂不可用，等待下一轮检查", ) continue attempts = round_stats["attempts"] successes = round_stats["successes"] all_done = round_stats["all_done"] remaining_courses = self.store.list_courses_for_user(self.user["id"]) if attempts == 0: self.logger( "INFO", f"本轮未产生提交尝试，剩余待选课程={self._course_batch_summary(remaining_courses)}。", ) elif successes == 0: self.logger( "INFO", f"本轮完成：尝试 {attempts} 门，成功 0 门，剩余待选课程={self._course_batch_summary(remaining_courses)}。", ) else: self.logger( "INFO", f"本轮完成：尝试 {attempts} 门，成功 {successes} 门，剩余待选课程={self._course_batch_summary(remaining_courses)}。", ) session_error_count = 0 if exhausted_sessions: self.logger("INFO", "Selenium 会话已恢复稳定，清零浏览器重启计数。") exhausted_sessions = 0 if all_done or not remaining_courses: self.logger("INFO", "所有待选课程已处理完成，任务完成。") return TaskResult(status="completed") self._sleep_with_cancel( stop_event, self._current_poll_interval_seconds(), "本轮结束，等待下一轮选课检查", ) except FatalCredentialsError as exc: self.logger("ERROR", str(exc)) return TaskResult(status="failed", error=str(exc)) except RecoverableAutomationError as exc: if session_proxy_url and self._should_fallback_browser_proxy(exc): self.browser_proxy_disabled_reason = "Chrome 通过代理访问教务/认证页面失败" self.logger( "WARNING", "检测到代理访问教务或统一认证页面失败，本任务将回退直连并重启 Selenium 会话。" f"原因: {exc}", ) break if self.effective_mode != TASK_RUN_MODE_STABLE: self._register_mode_failure(f"Selenium 可恢复异常: {exc}") self._invalidate_course_page_cache() session_ready = False session_error_count += 1 self.logger( "WARNING", f"Selenium 可恢复异常 {session_error_count}/{self.config.selenium_error_limit}: {exc}", ) if session_error_count < self.config.selenium_error_limit: self._sleep_with_cancel( stop_event, self.config.task_backoff_seconds, "等待后重试当前 Selenium 会话", ) continue exhausted_sessions += 1 if exhausted_sessions >= self.config.selenium_restart_limit: message = ( f"Selenium 会话连续重启 {exhausted_sessions} 次仍失败，任务停止。" f"最后原因: {exc}" ) self.logger("ERROR", message) return TaskResult(status="failed", error=message) self.logger( "WARNING", f"当前 Selenium 会话已连续异常 {self.config.selenium_error_limit} 次，" f"准备重启浏览器 {exhausted_sessions}/{self.config.selenium_restart_limit}。", ) break except Exception as exc: # pragma: no cover - defensive fallback if self.effective_mode != TASK_RUN_MODE_STABLE: self._register_mode_failure(f"未预期异常: {exc}") self._invalidate_course_page_cache() session_ready = False session_error_count += 1 self.logger( "WARNING", f"Selenium 未预期异常 {session_error_count}/{self.config.selenium_error_limit}: {exc}", ) if session_error_count < self.config.selenium_error_limit: self._sleep_with_cancel( stop_event, self.config.task_backoff_seconds, "等待后重试当前流程", ) continue exhausted_sessions += 1 if exhausted_sessions >= self.config.selenium_restart_limit: message = ( f"Selenium 会话连续重启 {exhausted_sessions} 次仍失败，任务停止。" f"最后原因: {exc}" ) self.logger("ERROR", message) return TaskResult(status="failed", error=message) self.logger( "WARNING", f"当前流程连续异常达到上限，准备重启 Selenium 会话 " f"{exhausted_sessions}/{self.config.selenium_restart_limit}。", ) break finally: if driver is not None: try: webdriver_utils.quit_browser(driver) self.logger("INFO", "Selenium 会话已关闭。") except Exception: pass self.logger("INFO", "任务收到停止信号，已在安全节点退出。") return TaskResult(status="stopped") def _refresh_effective_mode_from_store(self, *, log_change: bool = True) -> None: latest_task = self.store.get_task(self.task_id) if latest_task is None: return previous_effective = self.effective_mode self.requested_mode = normalize_task_run_mode(latest_task.get("requested_mode")) self.effective_mode = normalize_task_run_mode(latest_task.get("effective_mode")) latest_login_mode = normalize_login_mode(latest_task.get("login_mode")) if self.login_mode_fallback_to_unified and latest_login_mode == LOGIN_MODE_ZHJW_DIRECT: self.login_mode = LOGIN_MODE_UNIFIED else: self.login_mode = latest_login_mode self.use_browser_proxy = bool(int(latest_task.get("use_proxy") or 0)) if log_change and previous_effective != self.effective_mode: self.logger( "INFO", f"任务模式已更新：当前模式={self._mode_label(self.effective_mode)}，请求模式={self._mode_label(self.requested_mode)}。", ) def _invalidate_course_page_cache(self) -> None: self.select_course_page_ready = False self.active_category = None def _run_selection_round(self, driver: WebDriver, wait: WebDriverWait, stop_event, current_courses: list[dict]) -> dict[str, Any]: if self.effective_mode == TASK_RUN_MODE_STABLE: return self._run_stable_round(driver, wait, stop_event, current_courses) return self._run_optimized_round(driver, wait, stop_event, current_courses) def _run_stable_round(self, driver: WebDriver, wait: WebDriverWait, stop_event, current_courses: list[dict]) -> dict[str, Any]: self.logger( "INFO", f"开始选课轮次：模式={self._mode_label(self.effective_mode)}，待处理课程={self._course_batch_summary(current_courses)}。", ) if not self._goto_select_course(driver, wait): self._invalidate_course_page_cache() return {"attempts": 0, "successes": 0, "all_done": False, "not_selectable": True} attempts = 0 successes = 0 all_done = False grouped_courses = self._group_courses_by_category(current_courses) for category, items in grouped_courses.items(): if not items: continue for course in items: if stop_event.is_set(): break if not self._is_course_still_queued(course): continue attempts += 1 attempt_number = self._register_course_attempt() if self._attempt_single_course( driver, wait, course, attempt_number=attempt_number, reuse_page=False, reuse_category=False, ): successes += 1 self._maybe_probe_current_curriculum(driver, attempt_number=attempt_number) if not self.store.list_courses_for_user(self.user["id"]): all_done = True break if stop_event.is_set() or all_done: break return {"attempts": attempts, "successes": successes, "all_done": all_done, "not_selectable": False} def _run_optimized_round(self, driver: WebDriver, wait: WebDriverWait, stop_event, current_courses: list[dict]) -> dict[str, Any]: self.logger( "INFO", f"开始选课轮次：模式={self._mode_label(self.effective_mode)}，复用选课页面批量处理，待处理课程={self._course_batch_summary(current_courses)}。", ) if not self._ensure_select_course_page(driver, wait): return {"attempts": 0, "successes": 0, "all_done": False, "not_selectable": True} attempts = 0 successes = 0 all_done = False grouped_courses = self._group_courses_by_category(current_courses) for category, items in grouped_courses.items(): active_items = [course for course in items if self._is_course_still_queued(course)] if not active_items: continue if stop_event.is_set(): break self._ensure_category_open(driver, wait, category) for course in active_items: if stop_event.is_set(): break if not self._is_course_still_queued(course): continue attempts += 1 attempt_number = self._register_course_attempt() if self._attempt_single_course( driver, wait, course, attempt_number=attempt_number, reuse_page=True, reuse_category=True, ): successes += 1 self._maybe_probe_current_curriculum(driver, attempt_number=attempt_number) if not self.store.list_courses_for_user(self.user["id"]): all_done = True break if all_done: break return {"attempts": attempts, "successes": successes, "all_done": all_done, "not_selectable": False} @staticmethod def _group_courses_by_category(courses: list[dict]) -> dict[str, list[dict]]: grouped_courses = {"plan": [], "free": []} for course in courses: grouped_courses.setdefault(course["category"], []).append(course) return grouped_courses def _course_batch_summary(self, courses: list[dict]) -> str: grouped_courses = self._group_courses_by_category(courses) plan_count = len(grouped_courses.get("plan") or []) free_count = len(grouped_courses.get("free") or []) return f"{len(courses)} 门（方案选课 {plan_count}，自由选课 {free_count}）" def _course_key(self, course: dict) -> str: return self._normalize_course_identity( str(course.get("course_id") or ""), str(course.get("course_index") or ""), ) or f'{course.get("course_id", "")}_{course.get("course_index", "")}' @staticmethod def _raw_course_key(course: dict) -> str: return f"{str(course.get('course_id') or '').strip()}_{str(course.get('course_index') or '').strip()}" @staticmethod def _is_already_selected_detail(detail: str) -> bool: text = str(detail or "").strip() if not text: return False normalized_text = text.lower() success_tokens = ( "已选", "已在课表", "已经选", "重复", "无需", "成功", "selected", "already", "duplicate", "success", ) return any(token in text for token in success_tokens[:6]) or any( token in normalized_text for token in success_tokens[6:] ) def _ensure_select_course_page(self, driver: WebDriver, wait: WebDriverWait, *, force_reload: bool = False) -> bool: if force_reload or not self.select_course_page_ready: if not self._goto_select_course(driver, wait): self._invalidate_course_page_cache() return False self.select_course_page_ready = True self.active_category = None return True def _ensure_category_open(self, driver: WebDriver, wait: WebDriverWait, category: str) -> None: if not self.select_course_page_ready: self._ensure_select_course_page(driver, wait, force_reload=True) if self.active_category == category: return self._open_category_tab(driver, wait, category) self.active_category = category def _on_mode_activity_success(self, *, reset_mode_failures: bool = True) -> None: if reset_mode_failures: self.mode_failure_count = 0 def _register_mode_failure(self, reason: str) -> None: if self.effective_mode == TASK_RUN_MODE_STABLE: return self.mode_failure_count += 1 self.logger( "WARNING", f"{reason}；模式异常计数 {self.mode_failure_count}/{MODE_DOWNGRADE_THRESHOLD}。", ) if self.mode_failure_count >= MODE_DOWNGRADE_THRESHOLD: self._downgrade_to_stable(reason) def _downgrade_to_stable(self, reason: str) -> None: if self.effective_mode == TASK_RUN_MODE_STABLE: return previous_mode = self.effective_mode self.effective_mode = TASK_RUN_MODE_STABLE self.mode_failure_count = 0 self.store.update_task_mode(self.task_id, effective_mode=TASK_RUN_MODE_STABLE) self._invalidate_course_page_cache() self.logger( "WARNING", f"连续异常达到阈值，已从 {self._mode_label(previous_mode)} 降级为 {self._mode_label(TASK_RUN_MODE_STABLE)}。原因: {reason}", ) @staticmethod def _mode_label(mode: str) -> str: return TASK_RUN_MODE_LABELS.get(mode, mode) @staticmethod def _login_mode_label(mode: str) -> str: return LOGIN_MODE_LABELS.get(mode, mode) def _browser_proxy_label(self) -> str: if not self.use_browser_proxy: return "关闭" proxy_url = self._current_browser_proxy_url() if not proxy_url: if self.browser_proxy_disabled_reason: return f"已回退直连（{self.browser_proxy_disabled_reason}）" return "未配置" try: proxy = webdriver_utils.parse_browser_proxy(proxy_url) except Exception: return "配置无效" return "未启用" if proxy is None else proxy.display_label def _current_browser_proxy_url(self) -> str: if not self.use_browser_proxy: return "" if self.browser_proxy_disabled_reason: return "" return str(getattr(self.config, "browser_proxy_url", "") or "").strip() def _browser_proxy_probe_targets(self) -> list[tuple[str, int, str]]: targets = [("zhjw.scu.edu.cn", 80, "/login")] if self.login_mode == LOGIN_MODE_UNIFIED: targets.insert(0, ("id.scu.edu.cn", 80, "/")) return targets def _ensure_browser_proxy_available(self) -> None: proxy_url = self._current_browser_proxy_url() if not proxy_url: return passed_details: list[str] = [] for target_host, target_port, request_path in self._browser_proxy_probe_targets(): ok, detail = webdriver_utils.probe_browser_proxy( proxy_url, target_host=target_host, target_port=target_port, request_path=request_path, timeout_seconds=8, ) if not ok: self.browser_proxy_disabled_reason = "代理预检失败" self.logger("WARNING", f"浏览器代理预检失败，已回退直连。详情: {detail}。") return passed_details.append(detail) if passed_details: self.logger("INFO", f"浏览器代理预检通过：{'；'.join(passed_details)}。") return def _should_fallback_browser_proxy(self, exc: Exception) -> bool: text = str(exc or "").lower() return ( "chrome-error://chromewebdata" in text or "页面打开失败" in text or "页面加载失败" in text or "页面加载超时" in text or "chrome 返回空白页面" in text or "data:," in text or "about:blank" in text or "this site can" in text or "err_tunnel" in text or "err_proxy" in text or "err_socks" in text ) def _ensure_account_login_form(self, driver: WebDriver, wait: WebDriverWait) -> None: if ( self._find_first_visible_optional(driver, LOGIN_STUDENT_SELECTORS, timeout=1) and self._find_first_visible_optional(driver, LOGIN_PASSWORD_SELECTORS, timeout=1) ): return account_tab = self._find_first_visible_optional(driver, ACCOUNT_LOGIN_TAB_SELECTORS, timeout=20) if account_tab is None: raise RecoverableAutomationError(f"页面元素未找到: 账号登录页签。{self._page_snapshot(driver, include_body=True)}") try: account_tab.click() except WebDriverException: driver.execute_script("arguments[0].click();", account_tab) try: wait.until( lambda current_driver: ( self._find_first_visible_optional(current_driver, LOGIN_STUDENT_SELECTORS, timeout=0) is not None and self._find_first_visible_optional(current_driver, LOGIN_PASSWORD_SELECTORS, timeout=0) is not None ) ) except TimeoutException as exc: raise RecoverableAutomationError(f"账号登录表单未出现。{self._page_snapshot(driver, include_body=True)}") from exc def _login_for_task(self, driver: WebDriver, wait: WebDriverWait) -> None: if self.login_mode == LOGIN_MODE_ZHJW_DIRECT: self._login_zhjw_direct(driver, wait) return self._login(driver, wait) def _login_zhjw_direct(self, driver: WebDriver, wait: WebDriverWait) -> None: for attempt in range(1, self.config.login_retry_limit + 1): self._open_page(driver, wait, URL_ZHJW_LOGIN, f"教务直连登录页（第 {attempt} 次）", log_on_success=True) student_box = self._find_first_visible(driver, ZHJW_LOGIN_STUDENT_SELECTORS, "教务直连学号输入框", timeout=8) password_box = self._find_first_visible(driver, ZHJW_LOGIN_PASSWORD_SELECTORS, "教务直连密码输入框", timeout=8) captcha_box = self._find_first_visible(driver, ZHJW_LOGIN_CAPTCHA_INPUT_SELECTORS, "教务直连验证码输入框", timeout=8) captcha_image = self._find_first_visible(driver, ZHJW_LOGIN_CAPTCHA_IMAGE_SELECTORS, "教务直连验证码图片", timeout=8) login_button = self._find_first_visible(driver, ZHJW_LOGIN_BUTTON_SELECTORS, "教务直连登录按钮", timeout=8) student_box.clear() student_box.send_keys(str(self.user.get("student_id") or "")) password_box.clear() password_box.send_keys(self.password) captcha_text = self._solve_captcha_text(captcha_image, scene="教务直连登录") self.logger( "INFO", f"教务直连登录尝试 {attempt}/{self.config.login_retry_limit}，验证码 OCR 输出[{self._captcha_solver_engine_label()}]: {captcha_text}", ) captcha_box.clear() captcha_box.send_keys(captcha_text) self.logger("INFO", f"教务直连登录尝试 {attempt}/{self.config.login_retry_limit}，准备提交登录表单。") submit_method = self._trigger_non_blocking_action( driver, login_button, label="教务直连登录按钮", allow_form_submit=True, prefer_click=True, ) self.logger("INFO", f"教务直连登录表单已触发提交，方式={submit_method}，开始等待登录结果。") state, error_message = self._wait_for_zhjw_direct_login_outcome(driver, timeout_seconds=12) if state == "success": return if error_message: self.logger("WARNING", f"教务直连登录失败，第 {attempt} 次尝试，提示: {error_message}") if self._is_zhjw_direct_credentials_error(error_message): self.logger( "WARNING", "教务直连页明确提示账号或密码错误，已停止直连重试并回退统一认证，避免触发教务直连账户锁定。", ) self.login_mode_fallback_to_unified = True self.login_mode = LOGIN_MODE_UNIFIED self._login(driver, wait) return else: self.logger("WARNING", f"教务直连登录失败，第 {attempt} 次尝试，未读取到明确错误提示。{self._page_snapshot(driver, include_body=True)}") time.sleep(0.8) self.logger( "WARNING", "教务直连登录连续失败，疑似验证码识别不稳定或教务登录页暂不可用；本任务将回退统一认证登录。", ) self.login_mode_fallback_to_unified = True self.login_mode = LOGIN_MODE_UNIFIED self._login(driver, wait) def _wait_for_zhjw_direct_login_outcome(self, driver: WebDriver, timeout_seconds: int = 12) -> tuple[str, str]: deadline = time.monotonic() + max(1, timeout_seconds) started_at = time.monotonic() last_error = "" while time.monotonic() < deadline: try: current_url = driver.current_url or "" except WebDriverException: time.sleep(0.3) continue if self._is_login_success_url(current_url) and "/login" not in current_url: return "success", "" last_error = self._read_login_error(driver) if ( last_error and time.monotonic() - started_at >= 1.5 and not self._is_passive_second_factor_prompt(last_error) and not self._is_zhjw_direct_login_prompt(last_error) ): return "error", last_error time.sleep(0.35) return "unknown", last_error @staticmethod def _is_zhjw_direct_login_prompt(message: str) -> bool: text = re.sub(r"\s+", "", str(message or "")) if not text: return False prompts = { "验证码", "请输入验证码", "学号", "用户名", "密码", "请输入用户名", "请输入密码", "登录", } return text in prompts @staticmethod def _is_zhjw_direct_credentials_error(message: str) -> bool: text = re.sub(r"\s+", "", str(message or "")) if not text: return False credential_tokens = ( "用户密码错误", "用户名或密码错误", "账号或密码错误", "账户或密码错误", "密码错误", "用户不存在", "账号不存在", ) return any(token in text for token in credential_tokens) def _login(self, driver: WebDriver, wait: WebDriverWait) -> None: for attempt in range(1, self.config.login_retry_limit + 1): self._open_page(driver, wait, URL_LOGIN, f"登录页（第 {attempt} 次）", log_on_success=True) self._ensure_account_login_form(driver, wait) std_id_box = self._find_first_visible(driver, LOGIN_STUDENT_SELECTORS, "登录学号输入框", timeout=6) password_box = self._find_first_visible(driver, LOGIN_PASSWORD_SELECTORS, "登录密码输入框", timeout=6) captcha_box = self._find_first_visible(driver, LOGIN_CAPTCHA_INPUT_SELECTORS, "登录验证码输入框", timeout=6) login_button = self._find_first_visible(driver, LOGIN_BUTTON_SELECTORS, "登录按钮", timeout=6) captcha_image = self._find_first_visible(driver, LOGIN_CAPTCHA_IMAGE_SELECTORS, "登录验证码图片", timeout=6) captcha_text = self._solve_captcha_text(captcha_image, scene="登录") self.logger("INFO", f"登录尝试 {attempt}/{self.config.login_retry_limit}，验证码 OCR 输出[{self._captcha_solver_engine_label()}]: {captcha_text}") std_id_box.clear() std_id_box.send_keys(self.user["student_id"]) password_box.clear() password_box.send_keys(self.password) captcha_box.clear() captcha_box.send_keys(captcha_text) self.logger("INFO", f"登录尝试 {attempt}/{self.config.login_retry_limit}，准备提交登录表单。") submit_method = self._trigger_non_blocking_action( driver, login_button, label="登录表单", allow_form_submit=True, ) self.logger("INFO", f"登录表单已触发提交，方式={submit_method}，开始等待登录结果。") state, error_message = self._wait_for_login_outcome(driver, timeout_seconds=10) if state == "success": self.logger("INFO", f"登录成功，耗费 {attempt} 次尝试。") return if state == "second_factor": self.logger("INFO", "检测到统一认证短信二次验证页，准备进入短信验证码协助流程。") self._complete_second_factor_login(driver, wait) state, error_message = self._wait_for_login_outcome( driver, timeout_seconds=LOGIN_2FA_POST_SUBMIT_TIMEOUT_SECONDS, detect_second_factor=False, ) if state == "success": self.logger("INFO", f"短信认证通过，登录成功，耗费 {attempt} 次尝试。") return raise FatalCredentialsError( error_message or f"短信验证码提交后未进入教务系统，任务已停止，请重新启动任务。{self._page_snapshot(driver, include_body=True)}" ) if any(token in error_message for token in ("用户名或密码错误", "密码错误", "账号或密码错误", "用户不存在")): raise FatalCredentialsError("学号或密码错误，任务已停止，请在面板中更新后重新启动。") if error_message: self.logger("WARNING", f"登录失败，第 {attempt} 次尝试: {error_message}") else: self.logger( "WARNING", f"登录失败，第 {attempt} 次尝试，未读取到明确错误提示。{self._page_snapshot(driver, include_body=True)}", ) time.sleep(0.6) raise RecoverableAutomationError("连续多次登录失败，可能是验证码识别失败、页面异常或系统暂时不可用。") def _complete_second_factor_login(self, driver: WebDriver, wait: WebDriverWait) -> None: if self.login_2fa_code_provider is None: raise RecoverableAutomationError("检测到短信认证，但当前没有可用的验证码提交通道。") self._reset_login_2fa_network_trace(driver) phone_mask = self._read_second_factor_phone_mask(driver) self.logger("INFO", f"短信认证页面已就绪，手机号={phone_mask}，准备点击获取验证码。") send_button = self._find_first_visible(driver, LOGIN_2FA_SEND_CODE_SELECTORS, "短信验证码发送按钮", timeout=6) submit_method = self._trigger_non_blocking_action( driver, send_button, label="短信验证码发送按钮", allow_form_submit=False, ) self._wait_for_login_2fa_network_activity(driver, LOGIN_2FA_SEND_SMS_REQUEST_PATH, timeout_seconds=4) self.logger( "INFO", f"已触发短信验证码发送，方式={submit_method}，手机号={phone_mask}，网络={self._login_2fa_network_summary() or '暂无'}。", ) try: self.logger("INFO", "已进入短信验证码等待状态，等待用户或管理员在面板提交 6 位验证码。") code = self.login_2fa_code_provider(phone_mask, self.stop_event) except TimeoutError as exc: raise FatalCredentialsError(str(exc)) from exc except RuntimeError as exc: raise FatalCredentialsError(str(exc)) from exc normalized_code = re.sub(r"\D", "", str(code or "")) if not re.fullmatch(r"\d{6}", normalized_code): raise FatalCredentialsError("收到的短信验证码格式无效，任务已停止，请重新启动任务。") if self.stop_event is not None and self.stop_event.is_set(): raise FatalCredentialsError("短信验证码等待已停止。") request_count_before = self._count_login_2fa_network_events(LOGIN_2FA_VALIDATION_REQUEST_PATH) self.logger("INFO", "已收到面板提交的短信验证码，准备填入统一认证页面。") submit_method = self._fill_second_factor_code(driver, normalized_code) if not self._wait_for_login_2fa_network_activity( driver, LOGIN_2FA_VALIDATION_REQUEST_PATH, timeout_seconds=5, previous_count=request_count_before, ): submit_method = self._trigger_second_factor_component_validation(driver, normalized_code) if submit_method: self._wait_for_login_2fa_network_activity( driver, LOGIN_2FA_VALIDATION_REQUEST_PATH, timeout_seconds=5, previous_count=request_count_before, ) else: submit_method = self._dispatch_second_factor_keyboard_fallback(driver, normalized_code) self._wait_for_login_2fa_network_activity( driver, LOGIN_2FA_VALIDATION_REQUEST_PATH, timeout_seconds=5, previous_count=request_count_before, ) validation_events = self._latest_login_2fa_network_events(LOGIN_2FA_VALIDATION_REQUEST_PATH) if len(validation_events) > request_count_before: latest_event = validation_events[-1] status = latest_event.get("status") or latest_event.get("failure") or latest_event.get("state") or "unknown" self.logger( "INFO", f"短信验证码校验请求已触发，方式={submit_method}，状态={status}，网络={self._login_2fa_network_summary()}，等待统一认证跳转。", ) else: self.logger( "WARNING", f"短信验证码已填入，但未检测到统一认证校验请求，方式={submit_method}，网络={self._login_2fa_network_summary() or '暂无'}。继续等待页面结果。", ) def _read_second_factor_phone_mask(self, driver: WebDriver) -> str: script = """ const body = document.body ? (document.body.innerText || '') : ''; const match = body.match(/\\d{3}[\\s-]*(?:\\*|•|·|x|X){2,}[\\s-]*\\d{4}/); return match ? match[0].replace(/[\\s-]/g, '').replace(/[•·xX]/g, '*') : ''; """ deadline = time.monotonic() + 4 while time.monotonic() < deadline: try: phone_mask = str(driver.execute_script(script) or "").strip() except WebDriverException: phone_mask = "" if phone_mask: return phone_mask self._drain_login_2fa_network_events(driver) time.sleep(0.25) return "绑定手机" def _fill_second_factor_code(self, driver: WebDriver, code: str) -> str: inputs = self._find_second_factor_code_inputs(driver) if len(inputs) != 6: raise RecoverableAutomationError(f"短信验证码输入框数量异常，当前找到 {len(inputs)} 个。") self.logger("INFO", "已定位 6 个短信验证码输入框，开始逐位填入。") try: self._set_second_factor_input_values(driver, inputs, "") except WebDriverException: if self._has_left_second_factor_route(driver): return "route-changed" native_error: WebDriverException | None = None try: for index, digit in enumerate(code): input_box = inputs[index] input_box.click() input_box.send_keys(digit) time.sleep(0.08) if self._second_factor_input_values_match(driver, inputs, code): self.logger("INFO", "短信验证码已逐位填入统一认证页面，并完成 6 位输入框校验。") return "native-per-field" self.logger("WARNING", "短信验证码逐位填入后输入框校验未通过，准备切换到 DOM 事件回退方式。") except WebDriverException as exc: if self._has_left_second_factor_route(driver): return "route-changed" native_error = exc try: self._set_second_factor_input_values(driver, inputs, code) if self._second_factor_input_values_match(driver, inputs, code): self.logger("INFO", "短信验证码已通过 DOM 事件回退方式填入统一认证页面，并完成 6 位输入框校验。") return "dom-events" if self._has_left_second_factor_route(driver): return "route-changed" raise RecoverableAutomationError("短信验证码输入后校验失败。") except WebDriverException as exc: if self._has_left_second_factor_route(driver): return "route-changed" raise RecoverableAutomationError("短信验证码输入失败。") from (native_error or exc) def _find_second_factor_code_inputs(self, driver: WebDriver) -> list[WebElement]: primary_inputs = self._visible_elements_by_css(driver, LOGIN_2FA_PRIMARY_CODE_INPUT_SELECTOR) if len(primary_inputs) == 6: return primary_inputs visible_inputs = self._find_visible_elements(driver, LOGIN_2FA_CODE_INPUT_SELECTORS) if len(visible_inputs) == 6: return visible_inputs scoped_inputs = [] for element in visible_inputs: try: in_code_group = driver.execute_script( "return Boolean(arguments[0].closest('.sec-factor-six-ipt'));", element, ) if in_code_group: scoped_inputs.append(element) except WebDriverException: continue return scoped_inputs if len(scoped_inputs) == 6 else visible_inputs @staticmethod def _visible_elements_by_css(driver: WebDriver, selector: str) -> list[WebElement]: visible_elements: list[WebElement] = [] seen_ids: set[str] = set() try: elements = driver.find_elements(By.CSS_SELECTOR, selector) except WebDriverException: return visible_elements for element in elements: try: if element.id in seen_ids or not element.is_displayed(): continue seen_ids.add(element.id) visible_elements.append(element) except WebDriverException: continue return visible_elements def _has_left_second_factor_route(self, driver: WebDriver) -> bool: try: current_url = driver.current_url or "" except WebDriverException: return False if self._is_login_success_url(current_url): return True normalized_url = current_url.lower() return bool( current_url and "/second/auth" not in normalized_url and "#/second/auth" not in normalized_url ) def _set_second_factor_input_values(self, driver: WebDriver, inputs: list[WebElement], code: str) -> None: driver.execute_script( """ const inputs = arguments[0]; const code = String(arguments[1] || ''); const setNativeValue = (input, value) => { const proto = Object.getPrototypeOf(input); const descriptor = proto ? Object.getOwnPropertyDescriptor(proto, 'value') : null; if (descriptor && typeof descriptor.set === 'function') { descriptor.set.call(input, value); } else { input.value = value; } }; const keyEvent = (input, type, digit) => { input.dispatchEvent(new KeyboardEvent(type, { key: digit, code: 'Digit' + digit, keyCode: digit.charCodeAt(0), which: digit.charCodeAt(0), bubbles: true, cancelable: true })); }; inputs.forEach((input, index) => { const digit = code.charAt(index) || ''; input.focus(); if (digit) { keyEvent(input, 'keydown', digit); keyEvent(input, 'keypress', digit); if (typeof InputEvent === 'function') { input.dispatchEvent(new InputEvent('beforeinput', { inputType: 'insertText', data: digit, bubbles: true, cancelable: true })); } } setNativeValue(input, digit); if (typeof InputEvent === 'function') { input.dispatchEvent(new InputEvent('input', { inputType: digit ? 'insertText' : 'deleteContentBackward', data: digit || null, bubbles: true, cancelable: true })); } else { input.dispatchEvent(new Event('input', { bubbles: true })); } if (digit) { keyEvent(input, 'keyup', digit); } input.dispatchEvent(new Event('change', { bubbles: true })); }); """, inputs, code, ) def _second_factor_input_values_match(self, driver: WebDriver, inputs: list[WebElement], code: str) -> bool: try: current_value = str( driver.execute_script( """ return Array.from(arguments[0]) .slice(0, 6) .map((input) => String(input.value || '').trim()) .join(''); """, inputs, ) or "" ) except WebDriverException: if self._has_left_second_factor_route(driver): return True return False return current_value == code def _reset_login_2fa_network_trace(self, driver: WebDriver) -> None: self._login_2fa_network_events = {} self._drain_login_2fa_network_events(driver) def _wait_for_login_2fa_network_activity( self, driver: WebDriver, path: str, *, timeout_seconds: float, previous_count: int | None = None, ) -> bool: if previous_count is None: previous_count = self._count_login_2fa_network_events(path) deadline = time.monotonic() + max(0.2, timeout_seconds) while time.monotonic() < deadline: self._drain_login_2fa_network_events(driver) if self._count_login_2fa_network_events(path) > previous_count: return True try: current_url = driver.current_url or "" except WebDriverException: current_url = "" if self._is_login_success_url(current_url): return True if current_url and "/second/auth" not in current_url and "#/second/auth" not in current_url: return True time.sleep(0.2) return False def _drain_login_2fa_network_events(self, driver: WebDriver) -> None: try: raw_entries = driver.get_log("performance") except (ValueError, WebDriverException): return for entry in raw_entries: try: message = json.loads(str(entry.get("message") or "{}")).get("message") or {} except (TypeError, json.JSONDecodeError): continue method = str(message.get("method") or "") params = message.get("params") if isinstance(message.get("params"), dict) else {} request_id = str(params.get("requestId") or "") if not request_id: continue if method == "Network.requestWillBeSent": request = params.get("request") if isinstance(params.get("request"), dict) else {} url = str(request.get("url") or "") matched_path = self._matched_login_2fa_trace_path(url) if not matched_path: continue self._login_2fa_network_events[request_id] = { "path": matched_path, "method": str(request.get("method") or ""), "url": self._sanitize_login_2fa_url(url), "state": "sent", "status": "", "failure": "", "timestamp": float(params.get("timestamp") or 0), "wall_time": time.time(), } elif method == "Network.responseReceived": response = params.get("response") if isinstance(params.get("response"), dict) else {} url = str(response.get("url") or "") event = self._login_2fa_network_events.get(request_id) if event is None: matched_path = self._matched_login_2fa_trace_path(url) if not matched_path: continue event = { "path": matched_path, "method": "", "url": self._sanitize_login_2fa_url(url), "timestamp": float(params.get("timestamp") or 0), "wall_time": time.time(), } self._login_2fa_network_events[request_id] = event event["state"] = "response" event["status"] = str(response.get("status") or "") event["mime"] = str(response.get("mimeType") or "") event["from_cache"] = bool(response.get("fromDiskCache") or response.get("fromPrefetchCache")) elif method == "Network.loadingFinished": event = self._login_2fa_network_events.get(request_id) if event is not None: event["state"] = "finished" event["encoded_length"] = int(params.get("encodedDataLength") or 0) elif method == "Network.loadingFailed": event = self._login_2fa_network_events.get(request_id) if event is not None: event["state"] = "failed" event["failure"] = str(params.get("errorText") or "") event["canceled"] = bool(params.get("canceled")) def _matched_login_2fa_trace_path(self, url: str) -> str: text = str(url or "") for path in LOGIN_2FA_NETWORK_TRACE_PATHS: if path in text: return path return "" @staticmethod def _sanitize_login_2fa_url(url: str) -> str: try: parts = urlsplit(str(url or "")) except ValueError: return str(url or "").split("?", 1)[0][:140] return f"{parts.scheme}://{parts.netloc}{parts.path}"[:140] def _count_login_2fa_network_events(self, path: str) -> int: return len(self._latest_login_2fa_network_events(path)) def _latest_login_2fa_network_events(self, path: str) -> list[dict[str, Any]]: events = [ event for event in self._login_2fa_network_events.values() if str(event.get("path") or "") == path ] events.sort(key=lambda item: float(item.get("wall_time") or 0)) return events def _login_2fa_network_summary(self) -> str: chunks = [] for path in LOGIN_2FA_NETWORK_TRACE_PATHS: events = self._latest_login_2fa_network_events(path) if not events: continue latest = events[-1] state = str(latest.get("state") or "unknown") status = str(latest.get("status") or "").strip() failure = str(latest.get("failure") or "").strip() suffix = status or failure or state chunks.append(f"{path}#{len(events)}:{suffix}") return "; ".join(chunks) def _trigger_second_factor_component_validation(self, driver: WebDriver, code: str) -> str: try: method = driver.execute_script( """ const code = String(arguments[0] || ''); const root = document.querySelector('.sec-factor-six-ipt'); if (!root || !/^\\d{6}$/.test(code)) return ''; const inputs = Array.from(root.querySelectorAll('input[type="text"]')) .filter((input) => input.offsetParent !== null || input.getClientRects().length > 0) .slice(0, 6); if (inputs.length !== 6) return ''; inputs.forEach((input, index) => { input.focus(); input.value = code.charAt(index); input.dispatchEvent(new Event('input', { bubbles: true })); input.dispatchEvent(new Event('change', { bubbles: true })); }); const findVue = (start) => { let node = start; while (node) { let vm = node.__vue__; while (vm) { if (typeof vm.vilidOtpCode === 'function' || typeof vm.inputCode === 'function') { return vm; } vm = vm.$parent; } node = node.parentElement; } return null; }; const vm = findVue(root); if (vm && typeof vm.vilidOtpCode === 'function') { vm.vilidOtpCode(); return 'vue-vilidOtpCode'; } if (vm && typeof vm.inputCode === 'function') { vm.inputCode(5, new KeyboardEvent('keyup', { key: code.charAt(5), code: 'Digit' + code.charAt(5), keyCode: code.charCodeAt(5), which: code.charCodeAt(5), bubbles: true, cancelable: true })); return 'vue-inputCode'; } return ''; """, code, ) except WebDriverException: return "" return str(method or "") def _dispatch_second_factor_keyboard_fallback(self, driver: WebDriver, code: str) -> str: try: method = driver.execute_script( """ const code = String(arguments[0] || ''); const root = document.querySelector('.sec-factor-six-ipt'); if (!root || !/^\\d{6}$/.test(code)) return ''; const inputs = Array.from(root.querySelectorAll('input[type="text"]')) .filter((input) => input.offsetParent !== null || input.getClientRects().length > 0) .slice(0, 6); if (inputs.length !== 6) return ''; const fireKey = (input, type, digit) => { input.dispatchEvent(new KeyboardEvent(type, { key: digit, code: 'Digit' + digit, keyCode: digit.charCodeAt(0), which: digit.charCodeAt(0), bubbles: true, cancelable: true })); }; inputs.forEach((input, index) => { const digit = code.charAt(index); input.focus(); fireKey(input, 'keydown', digit); fireKey(input, 'keypress', digit); input.value = digit; input.dispatchEvent(new InputEvent('input', { inputType: 'insertText', data: digit, bubbles: true, cancelable: true })); fireKey(input, 'keyup', digit); input.dispatchEvent(new Event('change', { bubbles: true })); }); return 'dom-keyup'; """, code, ) except WebDriverException: return "" return str(method or "") def _goto_select_course(self, driver: WebDriver, wait: WebDriverWait) -> bool: self._open_page(driver, wait, URL_SELECT_COURSE, "选课页") if self._is_session_expired(driver): raise RecoverableAutomationError("检测到登录会话已失效，准备重新登录。") body_text = self._safe_body_text(driver) if "非选课" in body_text or "未到选课时间" in body_text: self.logger("INFO", body_text.strip() or "当前不是选课时段。") return False return True def _attempt_single_course( self, driver: WebDriver, wait: WebDriverWait, course: dict, *, attempt_number: int, reuse_page: bool = False, reuse_category: bool = False, ) -> bool: category_name = CATEGORY_META[course["category"]]["label"] course_key = self._course_key(course) self.logger( "INFO", f"开始尝试课程：类别={category_name}，课程={course_key}，累计尝试={attempt_number}，模式={self._mode_label(self.effective_mode)}。", ) if reuse_page: if not self._ensure_select_course_page(driver, wait): self.logger("INFO", f"当前不是可选课状态，跳过课程 {course_key}。") return False else: if not self._goto_select_course(driver, wait): self._invalidate_course_page_cache() self.logger("INFO", f"当前不是可选课状态，跳过课程 {course_key}。") return False self.select_course_page_ready = True self.active_category = None if reuse_category: self._ensure_category_open(driver, wait, course["category"]) else: self._open_category_tab(driver, wait, course["category"]) self.active_category = course["category"] found_target = self._query_and_mark_course(driver, wait, course, course_key) if not found_target: self._on_mode_activity_success() self.logger("INFO", f"查询结果中未发现目标课程 {course_key}，本轮跳过。") return False self.logger("INFO", f"已定位目标课程 {course_key}，准备提交选课请求。") results = self._submit_with_optional_captcha(driver, wait, course_key) if not results: self._on_mode_activity_success() self.logger("WARNING", f"课程 {course_key} 提交后未读取到选课结果。") return False satisfied = False for result in results: detail = (result.get("detail") or "").strip() subject = (result.get("subject") or course_key).strip() if result.get("result"): self.logger("SUCCESS", f"提交结果：课程={subject}，状态=成功。") satisfied = True elif self._is_already_selected_detail(detail): self.logger("INFO", f"提交结果：课程={subject}，状态=已在课表或无需重复处理，详情={detail or '无'}。") satisfied = True else: self.logger("WARNING", f"提交结果：课程={subject}，状态=失败，详情={detail or '未返回详情'}。") if satisfied: self.store.remove_course_by_identity( self.user["id"], course["category"], course["course_id"], course["course_index"], ) self._on_mode_activity_success() return True self._on_mode_activity_success() return False def _query_and_mark_course(self, driver: WebDriver, wait: WebDriverWait, course: dict, course_key: str) -> bool: raw_course_key = self._raw_course_key(course) candidate_keys = [raw_course_key] if course_key and course_key not in candidate_keys: candidate_keys.append(course_key) self.logger("INFO", f"正在查询课程 {course_key}，查询条件=课程号 {course['course_id']}，匹配候选={', '.join(candidate_keys)}。") try: wait.until(lambda current_driver: current_driver.find_element(By.ID, "ifra")) driver.switch_to.frame("ifra") course_id_box = self._find(driver, By.ID, "kch") search_button = self._find(driver, By.ID, "queryButton") course_id_box.clear() course_id_box.send_keys(course["course_id"]) search_button.click() wait.until( lambda current_driver: current_driver.execute_script( "const button = document.getElementById('queryButton'); return !button || !/正在|查询中|加载中|loading/i.test(button.innerText || '')" ) ) except TimeoutException as exc: self._invalidate_course_page_cache() raise RecoverableAutomationError( f"课程查询超时，未能确认查询按钮恢复。{self._page_snapshot(driver, include_body=True)}" ) from exc finally: driver.switch_to.default_content() time.sleep(self._ui_settle_seconds()) self._wait_for_course_result_table(driver, wait) try: found_target = False matched_key = "" for candidate_key in candidate_keys: if driver.execute_script(self.select_course_js, candidate_key) == "yes": found_target = True matched_key = candidate_key break except WebDriverException as exc: self._invalidate_course_page_cache() raise RecoverableAutomationError( f"读取课程查询结果失败。{self._page_snapshot(driver, include_body=True)}" ) from exc if found_target: self.logger("INFO", f"查询完成：课程={course_key}，匹配键={matched_key}，结果=找到目标课程。") else: diagnostic = self._course_result_table_diagnostic(driver) self.logger("INFO", f"查询完成：课程={course_key}，结果=未找到目标课程。{diagnostic}") return found_target def _wait_for_course_result_table(self, driver: WebDriver, wait: WebDriverWait) -> None: try: wait.until( lambda current_driver: current_driver.execute_script( """ const iframe = document.getElementById('ifra'); const doc = iframe ? (iframe.contentDocument || iframe.contentWindow.document) : null; return Boolean(doc && doc.getElementById('xirxkxkbody')); """ ) ) except TimeoutException as exc: self._invalidate_course_page_cache() raise RecoverableAutomationError( f"课程查询结果表未准备好。{self._page_snapshot(driver, include_body=True)}" ) from exc def _course_result_table_diagnostic(self, driver: WebDriver) -> str: try: payload = driver.execute_script( """ const iframe = document.getElementById('ifra'); const doc = iframe ? (iframe.contentDocument || iframe.contentWindow.document) : null; const rows = doc ? Array.from(doc.querySelectorAll('#xirxkxkbody tr')) : []; const samples = rows.slice(0, 5).map((row) => { const cells = row.getElementsByTagName('td'); return cells.length >= 3 ? (cells[2].innerText || cells[2].innerHTML || '').trim() : ''; }).filter(Boolean); return JSON.stringify({ row_count: rows.length, samples }); """ ) data = json.loads(payload) if isinstance(payload, str) else payload except Exception: return "未能读取结果表诊断信息。" row_count = int((data or {}).get("row_count") or 0) samples = [str(item).replace("\n", " ").strip() for item in ((data or {}).get("samples") or [])] sample_text = "；".join(samples)[:240] if samples else "无" return f"结果表行数={row_count}，样例={sample_text}。" def _ui_settle_seconds(self) -> float: if self.effective_mode == TASK_RUN_MODE_STABLE: return STABLE_UI_SETTLE_SECONDS return OPTIMIZED_UI_SETTLE_SECONDS def _register_course_attempt(self) -> int: self.selection_attempts += 1 self.store.increment_task_attempts(self.task_id) return self.selection_attempts def _maybe_probe_current_curriculum(self, driver: WebDriver, *, attempt_number: int) -> None: if attempt_number <= 0 or attempt_number % 20 != 0: return self.logger("INFO", f"累计选课尝试 {attempt_number} 次，开始回查本学期课表。") try: matched_courses = self._reconcile_selected_courses_via_callback(driver) except Exception as exc: self.logger("WARNING", f"回查本学期课表失败: {exc}") return if matched_courses: self.logger("INFO", f"回查完成，新增确认 {len(matched_courses)} 门已在课表中的课程。") else: self.logger("INFO", "回查完成，暂未发现新增已选课程。") def _reconcile_selected_courses_via_callback(self, driver: WebDriver) -> list[dict]: payload = self._fetch_current_curriculum_payload(driver) return self._reconcile_selected_courses_from_payload(payload) def _fetch_current_curriculum_payload(self, driver: WebDriver) -> dict: try: cookies = driver.get_cookies() except WebDriverException as exc: raise RecoverableAutomationError(f"读取浏览器会话 Cookie 失败: {exc}") from exc if not cookies: raise RecoverableAutomationError("当前浏览器会话没有可用 Cookie，无法回查本学期课表。") cookie_header = "; ".join( f"{cookie['name']}={cookie['value']}" for cookie in cookies if cookie.get("name") and cookie.get("value") ) if not cookie_header: raise RecoverableAutomationError("当前浏览器会话没有可用 Cookie，无法回查本学期课表。") try: referer = driver.current_url or URL_SELECT_COURSE except WebDriverException: referer = URL_SELECT_COURSE request = urllib.request.Request( URL_CURRICULUM_CALLBACK, headers={ "Accept": "application/json, text/plain, */*", "Cookie": cookie_header, "Referer": referer, "User-Agent": webdriver_utils.DEFAULT_USER_AGENT, "X-Requested-With": "XMLHttpRequest", }, ) try: with urllib.request.urlopen(request, timeout=20) as response: status = getattr(response, "status", 200) body = response.read().decode("utf-8", errors="replace") except urllib.error.URLError as exc: raise RecoverableAutomationError(f"访问本学期课表回查接口失败: {exc}") from exc if int(status) >= 400: raise RecoverableAutomationError(f"本学期课表回查接口返回异常状态码: {status}") try: payload = json.loads(body) except json.JSONDecodeError as exc: snippet = body.strip().replace("\n", " ")[:180] raise RecoverableAutomationError(f"本学期课表回查接口返回了非 JSON 内容: {snippet}") from exc if not isinstance(payload, dict): raise RecoverableAutomationError("本学期课表回查接口返回的数据结构不是对象。") return payload def _reconcile_selected_courses_from_payload(self, payload: dict) -> list[dict]: selected_map = self._extract_curriculum_course_entries(payload) if not selected_map: return [] matched_courses: list[dict] = [] current_courses = self.store.list_courses_for_user(self.user["id"]) for course in current_courses: course_key = self._normalize_course_identity(course["course_id"], course["course_index"]) detail = selected_map.get(course_key) if not detail: continue self.store.remove_course_by_identity( self.user["id"], course["category"], course["course_id"], course["course_index"], ) course_name = str(detail.get("courseName") or course_key).strip() status_name = str(detail.get("selectCourseStatusName") or detail.get("selectCourseStatusCode") or "已在课表中").strip() self.logger( "SUCCESS", f"回查确认课程已在本学期课表中: {course_name} ({course_key})，状态={status_name}，已自动从队列移除。", ) matched_courses.append({"course_key": course_key, "course_name": course_name, "status_name": status_name}) return matched_courses def _extract_curriculum_course_entries(self, payload: dict) -> dict[str, dict]: selected_map: dict[str, dict] = {} for entry in payload.get("xkxx") or []: if not isinstance(entry, dict): continue for raw_key, detail in entry.items(): if not isinstance(detail, dict): continue normalized_key = self._normalize_callback_course_key(raw_key, detail) if normalized_key: selected_map[normalized_key] = detail return selected_map @staticmethod def _normalize_course_identity(course_id: str, course_index: str) -> str: normalized_course_id = str(course_id or "").strip().upper() normalized_course_index = str(course_index or "").strip().upper() if not normalized_course_id or not normalized_course_index: return "" return f"{normalized_course_id}_{normalized_course_index}" def _normalize_callback_course_key(self, raw_key: str, detail: dict | None = None) -> str: raw_text = str(raw_key or "").strip() if raw_text and "_" in raw_text: course_id, course_index = raw_text.rsplit("_", 1) normalized = self._normalize_course_identity(course_id, course_index) if normalized: return normalized detail_id = detail.get("id") if isinstance(detail, dict) else None if isinstance(detail_id, dict): return self._normalize_course_identity( str(detail_id.get("coureNumber") or ""), str(detail_id.get("coureSequenceNumber") or ""), ) return "" def _is_course_still_queued(self, course: dict) -> bool: return any(item["id"] == course["id"] for item in self.store.list_courses_for_user(self.user["id"])) def _current_poll_interval_seconds(self) -> int: latest_user = self.store.get_user(self.user["id"]) if latest_user is not None: self.user = latest_user try: interval = int((latest_user or self.user).get("refresh_interval_seconds") or self.config.poll_interval_seconds) except (TypeError, ValueError, AttributeError): interval = int(self.config.poll_interval_seconds) return max(1, min(120, interval)) def _submit_with_optional_captcha(self, driver: WebDriver, wait: WebDriverWait, course_key: str) -> list[dict]: submit_button = self._find(driver, By.ID, "submitButton") submit_method = self._trigger_non_blocking_action( driver, submit_button, label=f"选课提交 {course_key}", allow_form_submit=True, ) self._invalidate_course_page_cache() self.logger("INFO", f"已触发选课提交，课程={course_key}，方式={submit_method}。") for attempt in range(1, self.config.submit_captcha_retry_limit + 1): state = self._wait_for_submit_state(driver, wait) if state == "result": self.logger("INFO", f"提交后检测到结果页，课程={course_key}。") return self._read_result_page(driver, wait) if state == "captcha": self.logger("INFO", f"提交后出现验证码，课程={course_key}，第 {attempt}/{self.config.submit_captcha_retry_limit} 次处理。") if not self._solve_visible_submit_captcha(driver): raise RecoverableAutomationError("提交验证码识别或输入控件未准备好。") continue self.logger("WARNING", f"提交后暂未出现结果或验证码，课程={course_key}，继续等待下一次检测。") raise RecoverableAutomationError("提交后多次等待仍未出现结果或验证码。") def _wait_for_submit_state(self, driver: WebDriver, wait: WebDriverWait) -> str: script = """ const visible = (el) => { if (!el) return false; const style = window.getComputedStyle(el); const rect = el.getBoundingClientRect(); return style.display !== 'none' && style.visibility !== 'hidden' && rect.width > 0 && rect.height > 0; }; if (document.querySelector('#xkresult tbody tr')) return 'result'; const input = Array.from(document.querySelectorAll('input')).find((el) => { const text = `${el.placeholder || ''} ${el.id || ''} ${el.name || ''} ${el.className || ''}`; return visible(el) && /验证码|captcha/i.test(text); }); const img = Array.from(document.querySelectorAll('img')).find((el) => { const text = `${el.src || ''} ${el.id || ''} ${el.alt || ''} ${el.className || ''}`; return visible(el) && (/captcha/i.test(text) || (el.src || '').includes('base64') || (el.naturalWidth >= 50 && el.naturalWidth <= 240 && el.naturalHeight >= 20 && el.naturalHeight <= 120)); }); const button = Array.from(document.querySelectorAll('button, input[type="button"], input[type="submit"]')).find((el) => { const text = `${el.innerText || ''} ${el.value || ''}`; return visible(el) && /确定|提交|确认|验证/.test(text); }); if (input && img && button) return 'captcha'; return 'pending'; """ try: wait.until(lambda current_driver: current_driver.execute_script(script) != "pending") except TimeoutException: return "pending" return str(driver.execute_script(script)) def _solve_visible_submit_captcha(self, driver: WebDriver) -> bool: image = self._find_first_visible_optional(driver, SUBMIT_CAPTCHA_IMAGE_SELECTORS, timeout=3) input_box = self._find_first_visible_optional(driver, SUBMIT_CAPTCHA_INPUT_SELECTORS, timeout=3) button = self._find_first_visible_optional(driver, SUBMIT_CAPTCHA_BUTTON_SELECTORS, timeout=3) if not image or not input_box or not button: return False captcha_text = self._solve_captcha_text(image, scene="提交") self.logger("INFO", f"提交验证码 OCR 输出[{self._captcha_solver_engine_label()}]: {captcha_text}") input_box.clear() input_box.send_keys(captcha_text) submit_method = self._trigger_non_blocking_action( driver, button, label="提交验证码确认按钮", allow_form_submit=True, ) self.logger("INFO", f"提交验证码确认已触发，方式={submit_method}。") time.sleep(1.0) return True def _open_category_tab(self, driver: WebDriver, wait: WebDriverWait, category: str) -> None: tab_id = CATEGORY_META[category]["tab_id"] tab = self._find(driver, By.ID, tab_id) tab.click() webdriver_utils.wait_for_ready(wait, allow_interactive=True) try: wait.until( lambda current_driver: current_driver.execute_script( """ const iframe = document.getElementById('ifra'); const doc = iframe ? (iframe.contentDocument || iframe.contentWindow.document) : null; return Boolean(doc && (doc.readyState === 'interactive' || doc.readyState === 'complete')); """ ) ) except TimeoutException as exc: self._invalidate_course_page_cache() raise RecoverableAutomationError( f"{CATEGORY_META[category]['label']}页签 iframe 未准备好。{self._page_snapshot(driver, include_body=True)}" ) from exc self.active_category = category time.sleep(self._ui_settle_seconds()) def _read_result_page(self, driver: WebDriver, wait: WebDriverWait) -> list[dict]: try: webdriver_utils.wait_for_ready(wait, allow_interactive=True) wait.until( lambda current_driver: current_driver.execute_script( """ const node = document.querySelector('#xkresult tbody tr'); return Boolean(node); """ ) ) return json.loads(driver.execute_script(self.check_result_js)) except Exception as exc: raise RecoverableAutomationError( f"读取选课结果失败，页面结构可能发生变化。{self._page_snapshot(driver, include_body=True)}" ) from exc def _open_page( self, driver: WebDriver, wait: WebDriverWait, url: str, label: str, *, allow_interactive: bool = True, log_on_success: bool = False, ) -> None: last_error: RecoverableAutomationError | None = None for open_attempt in range(1, 3): timed_out = webdriver_utils.open_with_recovery(driver, url) if timed_out: self.logger("WARNING", f"{label} 页面加载超时，尝试停止页面并继续执行。") try: ready_state = webdriver_utils.wait_for_ready(wait, allow_interactive=allow_interactive) except TimeoutException as exc: last_error = RecoverableAutomationError(f"{label} 页面加载失败。{self._page_snapshot(driver, include_body=True)}") if open_attempt >= 2: raise last_error from exc self.logger("WARNING", f"{label} 第 {open_attempt} 次打开失败，准备立即重试。原因: {last_error}") time.sleep(1.0) continue if self._is_chrome_error_page(driver): last_error = RecoverableAutomationError(f"{label} 页面打开失败，Chrome 返回网络错误。{self._page_snapshot(driver, include_body=True)}") if open_attempt >= 2: raise last_error self.logger("WARNING", f"{label} 第 {open_attempt} 次打开失败，准备立即重试。原因: {last_error}") time.sleep(1.0) continue if self._is_blank_browser_page(driver): last_error = RecoverableAutomationError(f"{label} 页面打开失败，Chrome 返回空白页面。{self._page_snapshot(driver, include_body=True)}") if open_attempt >= 2: raise last_error self.logger("WARNING", f"{label} 第 {open_attempt} 次打开失败，准备立即重试。原因: {last_error}") time.sleep(1.0) continue if log_on_success or timed_out or open_attempt > 1: self.logger("INFO", f"{label} 页面已打开，readyState={ready_state}。{self._page_snapshot(driver, include_body=timed_out)}") return if last_error is not None: raise last_error def _wait_for_login_outcome( self, driver: WebDriver, timeout_seconds: int = 10, *, detect_second_factor: bool = True, ) -> tuple[str, str]: deadline = time.monotonic() + max(1, timeout_seconds) last_error = "" wait_started_at = time.monotonic() next_progress_log_at = wait_started_at + LOGIN_2FA_WAIT_LOG_INTERVAL_SECONDS sso_advance_attempted = False last_logged_url = "" while time.monotonic() < deadline: if not detect_second_factor: self._drain_login_2fa_network_events(driver) try: current_url = driver.current_url or "" except WebDriverException: time.sleep(0.3) continue if not detect_second_factor and current_url != last_logged_url: last_logged_url = current_url self.logger( "INFO", f"短信认证后当前页面: {self._sanitize_snapshot_url(current_url) or '-'}。", ) if self._is_login_success_url(current_url): return "success", "" if detect_second_factor and self._is_second_factor_page(driver): return "second_factor", "" last_error = self._read_login_error(driver) if not detect_second_factor and last_error and self._is_second_factor_failure_message(last_error): return "error", last_error if ( not detect_second_factor and not sso_advance_attempted and self._has_successful_login_2fa_validation() and time.monotonic() - wait_started_at >= LOGIN_2FA_SSO_ADVANCE_AFTER_SECONDS ): sso_advance_attempted = self._advance_second_factor_sso(driver) if sso_advance_attempted: time.sleep(0.8) continue if ( last_error and not detect_second_factor and self._is_second_factor_page(driver) and not self._is_second_factor_failure_message(last_error) ): time.sleep(0.4) continue if last_error and not detect_second_factor and self._is_passive_second_factor_prompt(last_error): time.sleep(0.4) continue if last_error: return "error", last_error if not detect_second_factor and time.monotonic() >= next_progress_log_at: elapsed_seconds = int(time.monotonic() - wait_started_at) remaining_seconds = max(0, int(deadline - time.monotonic())) self.logger( "INFO", "短信验证码已提交，仍在等待统一认证跳转。" f" 已等待 {elapsed_seconds} 秒，剩余约 {remaining_seconds} 秒，" f"网络={self._login_2fa_network_summary() or '暂无'}。", ) next_progress_log_at = time.monotonic() + LOGIN_2FA_WAIT_LOG_INTERVAL_SECONDS time.sleep(0.4) if not detect_second_factor: self._drain_login_2fa_network_events(driver) final_error = self._read_login_error(driver) if not detect_second_factor and self._is_passive_second_factor_prompt(final_error): network_summary = self._login_2fa_network_summary() final_error = ( "短信验证码校验未完成或已超时，任务已停止，请重新启动任务。" f" 2FA网络摘要: {network_summary or '暂无'}。{self._page_snapshot(driver, include_body=True)}" ) return "unknown", final_error def _has_successful_login_2fa_validation(self) -> bool: events = self._latest_login_2fa_network_events(LOGIN_2FA_VALIDATION_REQUEST_PATH) if not events: return False latest_event = events[-1] failure = str(latest_event.get("failure") or "").strip() if failure: return False try: status = int(str(latest_event.get("status") or "0")) except ValueError: return False return 200 <= status < 300 def _advance_second_factor_sso(self, driver: WebDriver) -> bool: self.logger( "INFO", "短信验证码校验请求已成功，但统一认证尚未自动跳转；" "正在主动访问教务选课入口以触发 SSO 回调。", ) try: timed_out = webdriver_utils.open_with_recovery(driver, URL_SELECT_COURSE) except WebDriverException as exc: self.logger("WARNING", f"主动触发 SSO 回调失败: {exc}") return False if timed_out: self.logger("WARNING", "主动触发 SSO 回调时页面加载超时，已停止加载并继续等待登录结果。") else: self.logger("INFO", f"主动触发 SSO 回调完成。{self._page_snapshot(driver)}") return True def _is_second_factor_page(self, driver: WebDriver) -> bool: try: body_text = self._safe_body_text(driver) except Exception: body_text = "" normalized_body = body_text.lower() if "短信认证" in body_text or "sms verification" in normalized_body: return True if "security code" in normalized_body and "/second/auth" in (driver.current_url or ""): return True if self._find_first_visible_optional(driver, LOGIN_2FA_SEND_CODE_SELECTORS, timeout=0) is not None: return True return len(self._find_visible_elements(driver, LOGIN_2FA_CODE_INPUT_SELECTORS)) == 6 @staticmethod def _is_passive_second_factor_prompt(message: str) -> bool: text = str(message or "") if not text: return False if CourseBot._is_second_factor_failure_message(text): return False normalized_text = text.lower() return ( "短信认证" in text or "获取验证码" in text or "6位安全码" in text or "sms verification" in normalized_text or "security code" in normalized_text or "authentication code" in normalized_text ) @staticmethod def _is_second_factor_failure_message(message: str) -> bool: text = str(message or "") normalized_text = text.lower() return any(token in text for token in ("错误", "失败", "过期", "不正确", "无效")) or any( token in normalized_text for token in ("error", "failed", "expired", "incorrect", "invalid") ) def _read_login_error(self, driver: WebDriver) -> str: script = """ const visible = (node) => { if (!node) return false; const style = window.getComputedStyle(node); const rect = node.getBoundingClientRect(); return style.display !== 'none' && style.visibility !== 'hidden' && rect.width > 0 && rect.height > 0; }; const selectors = [ '.el-message', '.ivu-message', '.ivu-notice', '.ivu-form-item-error-tip', '[role="alert"]', '.message', '.toast', '.el-form-item__error', '.error' ]; for (const selector of selectors) { for (const node of Array.from(document.querySelectorAll(selector))) { const text = (node.innerText || '').trim(); if (visible(node) && text) { return text; } } } const nodes = Array.from(document.querySelectorAll('body span, body div')); for (const node of nodes) { const text = (node.innerText || '').trim(); if (!text || !visible(node)) continue; if (/验证码|密码|用户|账号|认证|校验|安全码/.test(text)) return text; if (/verification|captcha|code|password|account|user|invalid|expired|failed|incorrect|error|authentication|security/i.test(text)) return text; } return ''; """ try: raw_message = driver.execute_script(script) or "" except WebDriverException: return "" return re.sub(r"\s+", " ", str(raw_message)).strip() def _trigger_non_blocking_action( self, driver: WebDriver, element: WebElement, *, label: str, allow_form_submit: bool = False, prefer_click: bool = False, ) -> str: script = """ const target = arguments[0]; const allowFormSubmit = Boolean(arguments[1]); const preferClick = Boolean(arguments[2]); const form = allowFormSubmit && target ? (target.form || target.closest('form')) : null; let method = 'unavailable'; if (preferClick && target && typeof target.click === 'function') { method = 'scheduled-js-click'; } else if (form && typeof form.requestSubmit === 'function') { method = 'scheduled-requestSubmit'; } else if (target && typeof target.click === 'function') { method = 'scheduled-js-click'; } else if (target) { method = 'scheduled-dispatch-click'; } else if (form && typeof form.submit === 'function') { method = 'scheduled-form-submit'; } if (!target && !form) { return 'unavailable'; } window.setTimeout(() => { const dispatchFallback = (node) => { if (!node) return false; ['pointerdown', 'mousedown', 'mouseup', 'click'].forEach((type) => { node.dispatchEvent(new MouseEvent(type, { bubbles: true, cancelable: true, view: window, })); }); return true; }; try { if (target && typeof target.scrollIntoView === 'function') { target.scrollIntoView({block: 'center', inline: 'center'}); } } catch (error) {} if (preferClick) { try { if (target && typeof target.click === 'function') { target.click(); return; } } catch (error) {} try { if (dispatchFallback(target)) { return; } } catch (error) {} } try { if (form && typeof form.requestSubmit === 'function') { form.requestSubmit(target || undefined); return; } } catch (error) {} try { if (target && typeof target.click === 'function') { target.click(); return; } } catch (error) {} try { if (dispatchFallback(target)) { return; } } catch (error) {} try { if (form && typeof form.submit === 'function') { form.submit(); } } catch (error) {} }, 0); return method; """ try: method = str(driver.execute_script(script, element, allow_form_submit, prefer_click) or "").strip() if method and method != "unavailable": return method self.logger("WARNING", f"{label} 的 JS 非阻塞提交未找到可用方式，回退到原生点击。") except Exception as exc: self.logger("WARNING", f"{label} 的 JS 非阻塞提交失败，回退到原生点击。原因: {exc}") try: element.click() return "native-click" except TimeoutException: self.logger("WARNING", f"{label} 的原生点击触发后页面响应超时，已执行 window.stop() 并继续等待。") self._stop_loading(driver) return "native-click-timeout" except WebDriverException as exc: raise RecoverableAutomationError(f"{label} 触发失败: {exc}") from exc @staticmethod def _stop_loading(driver: WebDriver) -> None: try: driver.execute_script("window.stop();") except Exception: pass def _captcha_solver_engine_label(self) -> str: engine = str(getattr(self.captcha_solver, "last_engine_name", "") or "").strip() return engine or "unknown" def _solve_captcha_text(self, image_element: WebElement, *, scene: str) -> str: last_candidate = "" for attempt in range(1, 3): raw_text = self.captcha_solver.classification(self._extract_image_bytes(image_element)) normalized = re.sub(r"[^0-9A-Za-z]", "", str(raw_text or "")).strip().lower() if len(normalized) >= 4: return normalized[:4] last_candidate = normalized self.logger("WARNING", f"{scene}验证码 OCR 输出异常，第 {attempt} 次结果: {raw_text!r}") try: image_element.click() time.sleep(0.4) except Exception: pass if len(last_candidate) >= 3: return last_candidate[:4] raise RecoverableAutomationError(f"{scene}验证码 OCR 未能识别出有效内容。") def _is_login_success_url(self, url: str) -> bool: if not url: return False if any(url.startswith(prefix) for prefix in LOGIN_SUCCESS_PREFIXES): return True return "zhjw.scu.edu.cn" in url and "id.scu.edu.cn" not in url def _is_session_expired(self, driver: WebDriver) -> bool: current_url = driver.current_url or "" if "id.scu.edu.cn" in current_url: return True password_box = self._find_first_visible_optional(driver, LOGIN_PASSWORD_SELECTORS, timeout=1) return password_box is not None def _is_chrome_error_page(self, driver: WebDriver) -> bool: try: current_url = driver.current_url or "" except WebDriverException: return False return current_url.startswith("chrome-error://") def _is_blank_browser_page(self, driver: WebDriver) -> bool: try: current_url = driver.current_url or "" except WebDriverException: return False return current_url == "about:blank" or current_url.startswith("data:,") def _safe_body_text(self, driver: WebDriver) -> str: try: body = self._find(driver, By.TAG_NAME, "body") return body.text or "" except RecoverableAutomationError: return "" def _page_snapshot(self, driver: WebDriver, *, include_body: bool = False) -> str: script = """ const body = document.body ? (document.body.innerText || '') : ''; return JSON.stringify({ url: window.location.href || '', title: document.title || '', readyState: document.readyState || '', body: body.replace(/\\s+/g, ' ').trim().slice(0, 180) }); """ try: raw = driver.execute_script(script) data = json.loads(raw) if isinstance(raw, str) else raw except Exception: current_url = getattr(driver, "current_url", "") or "" return f" url={self._sanitize_snapshot_url(current_url) or '-'}" snapshot_url = self._sanitize_snapshot_url(str(data.get("url") or "")) parts = [ f"url={snapshot_url or '-'}", f"title={data.get('title') or '-'}", f"readyState={data.get('readyState') or '-'}", ] body_excerpt = self._sanitize_snapshot_text((data.get("body") or "").strip()) if include_body and body_excerpt: parts.append(f"body={body_excerpt}") return " " + " | ".join(parts) @staticmethod def _sanitize_snapshot_url(raw_url: str) -> str: text = str(raw_url or "") if not text: return "" try: parts = urlsplit(text) except ValueError: return re.sub(r"([?&][^=&]*(?:token|loginParams|credential|ticket|code|state|session)[^=]*=)[^&]+", r"\1[redacted]", text, flags=re.I) if not parts.query: return text sanitized_pairs = [(key, "[redacted]") for key, _value in parse_qsl(parts.query, keep_blank_values=True)] return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(sanitized_pairs), parts.fragment)) @classmethod def _sanitize_snapshot_text(cls, text: str) -> str: def replace_url(match: re.Match) -> str: return cls._sanitize_snapshot_url(match.group(0)) return re.sub(r"https?://[^\s<>\"']+", replace_url, str(text or "")) def _extract_image_bytes(self, image_element: WebElement) -> bytes: source = image_element.get_attribute("src") or "" if "base64," in source: return base64.b64decode(source.split("base64,", 1)[1]) return image_element.screenshot_as_png def _find_first_visible(self, driver: WebDriver, selectors: list[tuple[str, str]], label: str, timeout: int = 0): element = self._find_first_visible_optional(driver, selectors, timeout=timeout) if element is None: raise RecoverableAutomationError(f"页面元素未找到: {label}。{self._page_snapshot(driver, include_body=True)}") return element @staticmethod def _find_visible_elements(driver: WebDriver, selectors: list[tuple[str, str]]) -> list[WebElement]: visible_elements: list[WebElement] = [] seen_ids: set[str] = set() for by, value in selectors: try: elements = driver.find_elements(by, value) except WebDriverException: continue for element in elements: try: element_id = element.id if element_id in seen_ids or not element.is_displayed(): continue seen_ids.add(element_id) visible_elements.append(element) except WebDriverException: continue return visible_elements @staticmethod def _find_first_visible_optional(driver: WebDriver, selectors: list[tuple[str, str]], timeout: int = 0): deadline = time.monotonic() + max(0, timeout) while True: for by, value in selectors: try: elements = driver.find_elements(by, value) except WebDriverException: continue for element in elements: try: if element.is_displayed(): return element except WebDriverException: continue if timeout <= 0 or time.monotonic() >= deadline: return None time.sleep(0.2) @staticmethod def _find(driver: WebDriver, by: str, value: str): try: return driver.find_element(by, value) except NoSuchElementException as exc: raise RecoverableAutomationError(f"页面元素未找到: {value}") from exc except WebDriverException as exc: raise RecoverableAutomationError(f"浏览器操作失败: {value}") from exc def _sleep_with_cancel(self, stop_event, seconds: int, reason: str) -> None: if seconds <= 0: return self.logger("INFO", f"{reason} 大约 {seconds} 秒。") for _ in range(seconds): if stop_event.is_set(): return time.sleep(1)