| from __future__ import annotations
|
|
|
| import base64
|
| import json
|
| import re
|
| import time
|
| from dataclasses import dataclass
|
|
|
| from selenium.common.exceptions import NoSuchElementException, WebDriverException
|
| from selenium.webdriver.common.by import By
|
| from selenium.webdriver.support.wait import WebDriverWait
|
|
|
| import onnx_inference
|
| import webdriver_utils
|
| from core.config import CAPTCHA_MODEL_PATH, CHECK_RESULT_JS_PATH, CONFIG, SELECT_COURSE_JS_PATH
|
| from core.database import Database
|
| from core.security import decrypt_secret
|
|
|
|
|
| URL_LOGIN = 'http://id.scu.edu.cn/enduser/sp/sso/scdxplugin_jwt23?enterpriseId=scdx&target_url=index'
|
| URL_SELECT_COURSE = 'http://zhjw.scu.edu.cn/student/courseSelect/courseSelect/index'
|
| 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/',
|
| )
|
|
|
|
|
| class TaskStoppedError(RuntimeError):
|
| """Raised when a running task is stopped by the user or the service."""
|
|
|
|
|
| @dataclass(frozen=True)
|
| class CourseTarget:
|
| course_id: str
|
| course_index: str
|
|
|
|
|
| class CourseTaskRunner:
|
| def __init__(self, database: Database, context):
|
| self.database = database
|
| self.context = context
|
| self.driver = None
|
| self.web_wait: WebDriverWait | None = None
|
| self.select_course_js = SELECT_COURSE_JS_PATH.read_text(encoding='utf-8')
|
| self.check_result_js = CHECK_RESULT_JS_PATH.read_text(encoding='utf-8')
|
| self.captcha_solver = onnx_inference.CaptchaONNXInference(model_path=str(CAPTCHA_MODEL_PATH))
|
|
|
| def run(self) -> None:
|
| task = self.database.get_task(self.context.task_id)
|
| if not task:
|
| raise RuntimeError('任务不存在，无法启动。')
|
| user = self.database.get_user_by_id(task['user_id'])
|
| if not user:
|
| raise RuntimeError('用户不存在，无法执行选课任务。')
|
|
|
| payload = json.loads(task['task_payload'])
|
| courses = [CourseTarget(**item) for item in payload.get('courses', [])]
|
| if not courses:
|
| self.database.set_task_status(self.context.task_id, 'failed', last_error='没有可执行的课程。')
|
| return
|
|
|
| plain_password = decrypt_secret(user['encrypted_password'])
|
| self.context.log('info', f'任务启动，目标课程 {len(courses)} 门。')
|
| self.context.log('info', '正在启动浏览器环境。')
|
| self.driver = webdriver_utils.configure_browser()
|
| self.web_wait = WebDriverWait(self.driver, CONFIG.page_ready_timeout, 0.4)
|
|
|
| try:
|
| self._login(student_id=user['student_id'], password=plain_password)
|
| self._catch_courses(courses)
|
| finally:
|
| if self.driver:
|
| try:
|
| self.driver.quit()
|
| except WebDriverException:
|
| pass
|
|
|
| def _login(self, *, student_id: str, password: str) -> None:
|
| assert self.driver and self.web_wait
|
| self.context.log('info', '正在登录川大统一认证。')
|
| captcha_failures = 0
|
| other_failures = 0
|
|
|
| while True:
|
| self._raise_if_stopped()
|
| self.driver.get(URL_LOGIN)
|
| webdriver_utils.wait_for_ready(self.web_wait)
|
| std_box, password_box, captcha_box, login_button = self._get_login_fields()
|
| std_box.clear()
|
| std_box.send_keys(student_id)
|
| password_box.clear()
|
| password_box.send_keys(password)
|
|
|
| captcha_raw, captcha_b64 = self._read_captcha_image()
|
| if captcha_failures >= CONFIG.captcha_auto_attempts:
|
| self.context.log('warning', '自动验证码识别已到阈值，等待人工输入。')
|
| captcha_text = self.context.request_captcha(captcha_b64)
|
| if not captcha_text:
|
| raise RuntimeError('验证码等待超时，登录已终止。')
|
| self.context.log('info', '收到人工验证码，继续登录。')
|
| else:
|
| captcha_text = self.captcha_solver.classification(captcha_raw)
|
| self.context.log('debug', f'验证码模型输出：{captcha_text}')
|
|
|
| captcha_box.clear()
|
| captcha_box.send_keys(captcha_text)
|
| login_button.click()
|
|
|
| if self._wait_for_login_success():
|
| self.context.log('info', '登录成功。')
|
| return
|
|
|
| error_text = self._read_login_error_text()
|
| if error_text:
|
| self.context.log('warning', f'登录返回：{error_text}')
|
| else:
|
| self.context.log('warning', '未捕获到明确登录错误，准备重试。')
|
|
|
| if '密码错误' in error_text or '用户名或密码错误' in error_text or '用户不存在' in error_text:
|
| raise RuntimeError('学号或密码错误，请先在后台或用户页更新账号信息。')
|
|
|
| if '验证码' in error_text:
|
| captcha_failures += 1
|
| else:
|
| other_failures += 1
|
|
|
| if other_failures >= 8:
|
| raise RuntimeError('登录多次失败，当前无法稳定进入教务系统。')
|
|
|
| time.sleep(1.0)
|
|
|
| def _get_login_fields(self):
|
| assert self.driver
|
| return (
|
| self.driver.find_element(
|
| 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',
|
| ),
|
| self.driver.find_element(
|
| 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',
|
| ),
|
| self.driver.find_element(
|
| By.XPATH,
|
| '//*[@id="app"]/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[3]/div/div/div/div/input',
|
| ),
|
| self.driver.find_element(
|
| By.XPATH,
|
| '//*[@id="app"]/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[4]/div/button',
|
| ),
|
| )
|
|
|
| def _read_captcha_image(self) -> tuple[bytes, str]:
|
| assert self.driver
|
| source = self.driver.find_element(
|
| By.XPATH,
|
| '//*[@id="app"]/div[1]/div/div[2]/div/div[1]/div[2]/div[2]/div/form/div[3]/div/div/img',
|
| ).get_attribute('src')
|
| if 'base64,' not in source:
|
| raise RuntimeError('未能读取验证码图片。')
|
| encoded = source.split('base64,', 1)[1]
|
| return base64.b64decode(encoded), encoded
|
|
|
| def _wait_for_login_success(self) -> bool:
|
| assert self.driver
|
| for _ in range(10):
|
| current_url = self.driver.current_url
|
| if any(current_url.startswith(prefix) for prefix in LOGIN_SUCCESS_PREFIXES):
|
| return True
|
| time.sleep(0.5)
|
| return False
|
|
|
| def _read_login_error_text(self) -> str:
|
| assert self.driver
|
| try:
|
| error_box = self.driver.find_element(By.XPATH, '/html/body/div[2]')
|
| return error_box.text.strip()
|
| except NoSuchElementException:
|
| return ''
|
|
|
| def _goto_select_course(self) -> None:
|
| assert self.driver and self.web_wait
|
| self.driver.get(URL_SELECT_COURSE)
|
| webdriver_utils.wait_for_ready(self.web_wait)
|
| body_text = self.driver.find_element(By.TAG_NAME, 'body').text
|
| if '非选课' in body_text:
|
| raise RuntimeError('当前不在选课时间，教务系统返回了非选课页面。')
|
|
|
| def _catch_courses(self, courses: list[CourseTarget]) -> None:
|
| remaining = {(course.course_id, course.course_index) for course in courses}
|
| round_index = 0
|
| while remaining:
|
| self._raise_if_stopped()
|
| round_index += 1
|
| self.context.log('info', f'开始第 {round_index} 轮检索，剩余 {len(remaining)} 门课程。')
|
| round_successes = 0
|
| for tab_name, tab_id in (('方案选课', 'faxk'), ('自由选课', 'zyxk')):
|
| if not remaining:
|
| break
|
| round_successes += self._process_tab(tab_name, tab_id, remaining)
|
|
|
| if not remaining:
|
| self.context.log('info', '所有目标课程已完成。')
|
| self.database.set_task_status(
|
| self.context.task_id,
|
| 'success',
|
| completed_count=len(courses),
|
| last_error='',
|
| )
|
| return
|
|
|
| if round_successes == 0:
|
| self.context.log('debug', f'本轮未命中课程，{CONFIG.task_poll_interval:.1f}s 后继续。')
|
| self._sleep_with_stop(CONFIG.task_poll_interval)
|
|
|
| def _process_tab(self, tab_name: str, tab_id: str, remaining: set[tuple[str, str]]) -> int:
|
| assert self.driver and self.web_wait
|
| self._goto_select_course()
|
| self.driver.find_element(By.XPATH, f'//*[@id="{tab_id}"]').click()
|
| webdriver_utils.wait_for_ready(self.web_wait)
|
|
|
| selected_targets: list[tuple[str, str]] = []
|
| for course_id, course_index in list(remaining):
|
| self._raise_if_stopped()
|
| if self._search_and_mark_current_tab(course_id, course_index):
|
| selected_targets.append((course_id, course_index))
|
| self.context.log('info', f'{tab_name} 命中课程 {course_id}_{course_index}，已勾选。')
|
|
|
| if not selected_targets:
|
| return 0
|
|
|
| self.driver.find_element(By.XPATH, '//*[@id="submitButton"]').click()
|
| results = self._read_submit_results()
|
| successes = 0
|
| for result in results:
|
| course_key = self._extract_course_key(result['subject'])
|
| if result['result'] and course_key in remaining:
|
| remaining.remove(course_key)
|
| successes += 1
|
| completed = self._current_completed_count(remaining)
|
| self.database.update_task_progress(self.context.task_id, completed)
|
| self.context.log('info', f"选课成功：{result['subject']}")
|
| elif result['result']:
|
| self.context.log('info', f"选课成功，但未能精确匹配目标项：{result['subject']}")
|
| else:
|
| self.context.log('warning', f"选课失败：{result['subject']}，原因：{result['detail']}")
|
| return successes
|
|
|
| def _search_and_mark_current_tab(self, course_id: str, course_index: str) -> bool:
|
| assert self.driver and self.web_wait
|
| self.driver.switch_to.frame('ifra')
|
| try:
|
| course_id_box = self.driver.find_element(By.XPATH, '//*[@id="kch"]')
|
| query_button = self.driver.find_element(By.XPATH, '//*[@id="queryButton"]')
|
| course_id_box.clear()
|
| course_id_box.send_keys(course_id)
|
| query_button.click()
|
| self.web_wait.until(
|
| lambda driver: driver.execute_script(
|
| "return document.getElementById('queryButton').innerText.indexOf('正在') === -1"
|
| )
|
| )
|
| finally:
|
| self.driver.switch_to.default_content()
|
|
|
| self.web_wait.until(lambda driver: driver.execute_script("return document.getElementById('ifra') != null"))
|
| time.sleep(0.15)
|
| return self.driver.execute_script(self.select_course_js, f'{course_id}_{course_index}') == 'yes'
|
|
|
| def _read_submit_results(self) -> list[dict]:
|
| assert self.driver and self.web_wait
|
| self.web_wait.until(
|
| lambda driver: driver.execute_script("return document.getElementById('xkresult') != null")
|
| )
|
| time.sleep(0.3)
|
| raw = self.driver.execute_script(self.check_result_js)
|
| return json.loads(raw)
|
|
|
| def _extract_course_key(self, subject: str) -> tuple[str, str]:
|
| match = re.search(r'(\d{5,})_(\d{2})', subject)
|
| if match:
|
| return (match.group(1), match.group(2))
|
| numbers = re.findall(r'\d+', subject)
|
| if len(numbers) >= 2:
|
| return (numbers[-2], numbers[-1].zfill(2))
|
| return ('', '')
|
|
|
| def _current_completed_count(self, remaining: set[tuple[str, str]]) -> int:
|
| task = self.database.get_task(self.context.task_id)
|
| total = int(task['total_count']) if task else 0
|
| return max(0, total - len(remaining))
|
|
|
| def _sleep_with_stop(self, seconds: float) -> None:
|
| deadline = time.monotonic() + seconds
|
| while time.monotonic() < deadline:
|
| self._raise_if_stopped()
|
| time.sleep(0.2)
|
|
|
| def _raise_if_stopped(self) -> None:
|
| if self.context.should_stop():
|
| raise TaskStoppedError()
|
|
|
