306 lines
12 KiB
Python
306 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
精确时间选课自动化脚本
|
||
在指定时间的抢课时段，进行高频请求；其余时间则低频监控。
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import json
|
||
import threading
|
||
from datetime import datetime, timedelta
|
||
import schedule
|
||
import signal
|
||
import sys
|
||
|
||
# ====================================================================================
|
||
# ██████████████████████████████ 用户配置区域 ██████████████████████████████
|
||
#
|
||
# 使用前请务必更新这里的 Cookie 和 data 信息！
|
||
#
|
||
# ====================================================================================
|
||
CONFIG = {
|
||
# 目标URL（注意更改profileId！）
|
||
'url': 'https://jwxt.neuq.edu.cn/eams/stdElectCourse!batchOperator.action?profileId=1422',
|
||
|
||
# 请求头 (一般无需修改)
|
||
'headers': {
|
||
'Accept': 'text/html, */*; q=0.01',
|
||
'Accept-Language': 'en-US,en;q=0.9',
|
||
'Connection': 'keep-alive',
|
||
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
|
||
'Origin': 'https://jwxt.neuq.edu.cn',
|
||
'Referer': 'https://jwxt.neuq.edu.cn/eams/stdElectCourse!defaultPage.action?electionProfile.id=1422',
|
||
'Sec-Fetch-Dest': 'empty',
|
||
'Sec-Fetch-Mode': 'cors',
|
||
'Sec-Fetch-Site': 'same-origin',
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
|
||
'X-Requested-With': 'XMLHttpRequest',
|
||
},
|
||
|
||
# Cookie (!!重要!! 必须替换为你的有效Cookie)
|
||
'cookies': {
|
||
'semester.id': '85',
|
||
'JSESSIONID': 'YOURCOOKIE1',
|
||
'SERVERNAME': 'c1',
|
||
'JSESSIONID': 'YOURCOOKIE2',
|
||
'GSESSIONID': 'YOURCOOKIE3'
|
||
|
||
},
|
||
|
||
# 请求数据 (!!重要!! 必须替换为你要选择的课程ID)
|
||
'data': '',
|
||
|
||
# 时间配置，替换为你课程开始抢的时间
|
||
'target_date': "2025-08-05", # 目标日期 (格式: YYYY-MM-DD)
|
||
'rush_hours': ["3:15", "10:00", "13:00", "14:00", "15:00", "16:00"], # 抢课时间点
|
||
'rush_interval': 0.6, # 抢课时请求间隔(秒)
|
||
'normal_interval': 10.0, # 平时监控间隔(秒)
|
||
'rush_duration': 10, # 每次抢课持续时间(分钟)
|
||
'rush_start_offset': 5 # 提前几分钟开始抢课(分钟)
|
||
}
|
||
|
||
|
||
class PreciseCourseScheduler:
|
||
def __init__(self, config):
|
||
self.config = config
|
||
self.url = self.config['url']
|
||
self.headers = self.config['headers']
|
||
self.cookies = self.config['cookies']
|
||
self.data = self.config['data']
|
||
self.target_date = self.config['target_date']
|
||
self.rush_hours = self.config['rush_hours']
|
||
self.rush_interval = self.config['rush_interval']
|
||
self.normal_interval = self.config['normal_interval']
|
||
|
||
self.is_running = False
|
||
self.current_mode = "normal"
|
||
self.rush_thread = None
|
||
self.normal_thread = None
|
||
self.stop_event = threading.Event()
|
||
self.rush_active = threading.Event() # 用于在抢课时暂停正常模式
|
||
|
||
self.total_requests = 0
|
||
self.success_count = 0
|
||
self.course_closed_count = 0
|
||
self.already_selected_count = 0
|
||
self.error_count = 0
|
||
self.session = requests.Session()
|
||
|
||
def make_request(self) -> dict:
|
||
"""执行单次请求"""
|
||
start_time = time.time()
|
||
request_time = datetime.now()
|
||
|
||
try:
|
||
response = self.session.post(
|
||
self.url, headers=self.headers, cookies=self.cookies,
|
||
data=self.data, timeout=5
|
||
)
|
||
response_time = time.time() - start_time
|
||
content = response.text
|
||
status, is_success = self._analyze_response(content)
|
||
|
||
self.total_requests += 1
|
||
if is_success:
|
||
self.success_count += 1
|
||
if status == 'course_closed':
|
||
self.course_closed_count += 1
|
||
elif status == 'already_selected':
|
||
self.already_selected_count += 1
|
||
else:
|
||
self.error_count += 1
|
||
|
||
return {
|
||
'timestamp': request_time.isoformat(), 'status_code': response.status_code,
|
||
'response_time': response_time, 'content': content, 'status': status,
|
||
'is_success': is_success, 'mode': self.current_mode
|
||
}
|
||
except Exception as e:
|
||
self.total_requests += 1
|
||
self.error_count += 1
|
||
return {
|
||
'timestamp': request_time.isoformat(), 'status_code': 0,
|
||
'response_time': time.time() - start_time, 'content': f"请求异常: {str(e)}",
|
||
'status': 'error', 'is_success': False, 'mode': self.current_mode
|
||
}
|
||
|
||
def _analyze_response(self, content: str) -> tuple:
|
||
"""分析响应内容"""
|
||
if '当前选课不开放' in content: return 'course_closed', True
|
||
if '已投放' in content: return 'already_selected', True
|
||
if '请不要过快点击' in content or '点击过快' in content: return 'too_fast', False
|
||
if not content.strip(): return 'empty', False
|
||
return 'unknown', False
|
||
|
||
def _log_request(self, result: dict):
|
||
"""记录请求日志"""
|
||
status_emoji = {'course_closed': '🔒', 'already_selected': '✅', 'too_fast': '⚡️',
|
||
'error': '❌', 'empty': '📄', 'unknown': '❓'}.get(result['status'], '❓')
|
||
mode_emoji = "🚀" if result['mode'] == "rush" else "🐌"
|
||
success_mark = "✔️" if result['is_success'] else "✖️"
|
||
|
||
print(f"{result['timestamp']} {mode_emoji} {result['mode'].upper()} {success_mark} "
|
||
f"Code:{result['status_code']} RTT:{result['response_time']:.3f}s "
|
||
f"{status_emoji} {result['status']}")
|
||
|
||
if result['status'] in ['error', 'unknown', 'too_fast'] and result['content']:
|
||
preview = result['content'][:100].replace('\n', ' ').replace('\r', ' ')
|
||
print(f" └── 内容: {preview}...")
|
||
|
||
if self.total_requests % 10 == 0: self._print_stats()
|
||
|
||
def _print_stats(self):
|
||
"""打印统计信息"""
|
||
success_rate = (self.success_count / self.total_requests * 100) if self.total_requests > 0 else 0
|
||
print(f"\n📊 统计 (总计 {self.total_requests} 次请求):")
|
||
print(f" 成功率: {success_rate:.1f}% ({self.success_count}/{self.total_requests}) | "
|
||
f"不开放: {self.course_closed_count} | 已投放: {self.already_selected_count} | "
|
||
f"错误: {self.error_count}\n")
|
||
|
||
def rush_mode_worker(self):
|
||
"""抢课模式工作线程"""
|
||
duration = self.config['rush_duration']
|
||
print(f"🚀 进入抢课模式，持续 {duration} 分钟，间隔 {self.rush_interval}s")
|
||
self.rush_active.set()
|
||
self.current_mode = "rush"
|
||
|
||
end_time = time.time() + (duration * 60)
|
||
while time.time() < end_time and not self.stop_event.is_set():
|
||
result = self.make_request()
|
||
self._log_request(result)
|
||
if result['status'] == 'already_selected':
|
||
print("✅ 选课成功！已投放！脚本将继续运行，可按 Ctrl+C 停止。")
|
||
|
||
if self.stop_event.wait(self.rush_interval): break
|
||
|
||
print("🚀 抢课模式结束")
|
||
self.current_mode = "normal"
|
||
self.rush_active.clear()
|
||
|
||
def normal_mode_worker(self):
|
||
"""正常模式工作线程"""
|
||
print(f"🐌 正常监控模式已启动，间隔 {self.normal_interval}s")
|
||
self.current_mode = "normal"
|
||
|
||
# 同步到下一个10秒周期
|
||
now = datetime.now()
|
||
wait_seconds = self.normal_interval - (now.timestamp() % self.normal_interval)
|
||
if self.stop_event.wait(wait_seconds): return
|
||
|
||
while not self.stop_event.is_set():
|
||
if self.rush_active.is_set():
|
||
if self.stop_event.wait(1): break
|
||
continue
|
||
|
||
result = self.make_request()
|
||
self._log_request(result)
|
||
if self.stop_event.wait(self.normal_interval): break
|
||
|
||
print("🐌 正常监控模式结束")
|
||
|
||
def schedule_rush_session(self, target_time: str):
|
||
"""安排一次抢课会话"""
|
||
|
||
def start_rush():
|
||
if self.rush_thread and self.rush_thread.is_alive():
|
||
print(f"⚠️ 上一个抢课会话仍在进行中，跳过 {target_time}")
|
||
return
|
||
|
||
print(f"⏸️ 抢课时间到，正常监控将自动暂停...")
|
||
self.rush_thread = threading.Thread(target=self.rush_mode_worker, daemon=True)
|
||
self.rush_thread.start()
|
||
|
||
offset = self.config['rush_start_offset']
|
||
target_dt = datetime.strptime(f"{self.target_date} {target_time}", "%Y-%m-%d %H:%M")
|
||
start_dt = target_dt - timedelta(minutes=offset)
|
||
start_time_str = start_dt.strftime("%H:%M")
|
||
|
||
print(f"🗓️ 已安排抢课任务: {target_time} (将于 {start_time_str} 启动)")
|
||
schedule.every().day.at(start_time_str).do(start_rush)
|
||
|
||
def start_scheduler(self):
|
||
"""启动调度器"""
|
||
print("✅ 精确时间选课调度器启动")
|
||
print(f"🎯 目标日期: {self.target_date}")
|
||
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
if today != self.target_date:
|
||
print(f"⚠️ 注意: 当前日期({today})不是目标日期({self.target_date})。脚本将在目标日期自动激活。")
|
||
|
||
for rush_hour in self.rush_hours: self.schedule_rush_session(rush_hour)
|
||
print("-" * 60)
|
||
|
||
self.is_running = True
|
||
self.normal_thread = threading.Thread(target=self.normal_mode_worker, daemon=True)
|
||
self.normal_thread.start()
|
||
|
||
try:
|
||
while self.is_running:
|
||
if datetime.now().strftime("%Y-%m-%d") == self.target_date:
|
||
schedule.run_pending()
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
print("\n🛑 收到中断信号，正在停止...")
|
||
self.stop()
|
||
|
||
def stop(self):
|
||
"""停止调度器"""
|
||
self.is_running = False
|
||
self.stop_event.set()
|
||
|
||
print("...正在停止所有线程...")
|
||
if self.normal_thread and self.normal_thread.is_alive(): self.normal_thread.join(timeout=5)
|
||
if self.rush_thread and self.rush_thread.is_alive(): self.rush_thread.join(timeout=5)
|
||
|
||
schedule.clear()
|
||
self._print_final_stats()
|
||
print("✅ 调度器已安全停止")
|
||
|
||
def _print_final_stats(self):
|
||
"""打印最终统计"""
|
||
print("\n" + "=" * 25 + " 最终统计报告 " + "=" * 25)
|
||
print(f"总请求数: {self.total_requests}, 成功: {self.success_count}, 错误: {self.error_count}")
|
||
if self.total_requests > 0:
|
||
print(f"成功率: {self.success_count / self.total_requests * 100:.2f}%")
|
||
print(f"选课结果: [不开放: {self.course_closed_count}次], [已投放: {self.already_selected_count}次]")
|
||
print("=" * 68)
|
||
|
||
|
||
def signal_handler(signum, frame):
|
||
"""优雅地处理退出信号"""
|
||
# 这个函数现在是备用的，因为主循环的 KeyboardInterrupt 处理更直接
|
||
print(f"\n🚨 收到信号 {signum}，准备退出...")
|
||
# 全局变量 `scheduler` 可能还未定义，所以需要检查
|
||
if 'scheduler' in globals() and scheduler.is_running:
|
||
scheduler.stop()
|
||
sys.exit(0)
|
||
|
||
|
||
def main():
|
||
global scheduler
|
||
|
||
# 检查依赖
|
||
try:
|
||
import schedule
|
||
except ImportError:
|
||
print("❌ 缺少依赖包 `schedule`，请运行: pip install schedule")
|
||
sys.exit(1)
|
||
|
||
scheduler = PreciseCourseScheduler(CONFIG)
|
||
|
||
print("=" * 60)
|
||
print(" 欢迎使用精确时间选课自动化脚本")
|
||
print(" 作者: Galaxy")
|
||
print(" 按 Ctrl+C 可随时停止运行")
|
||
print("=" * 60)
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
|
||
scheduler.start_scheduler()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |