高效学习利器：打造智能化的长江雨课堂习题备份系统

每次复习时手动整理习题的繁琐过程，相信是许多学生的共同痛点。面对雨课堂平台上分散的练习题，传统复制粘贴方式不仅耗时耗力，还容易出错。本文将带你从零构建一个自动化习题备份系统，将选择题高效导出至MySQL数据库，实现一键式复习资料归档。

1. 系统架构设计与技术选型

一个完整的习题备份系统需要兼顾数据采集、存储和后续查询三个核心环节。我们选择Python作为开发语言，搭配Requests库处理网络请求，lxml解析HTML内容，PyMySQL操作MySQL数据库。

技术栈对比分析：

技术组件	替代方案	选择理由
Requests	urllib3	更简洁的API设计
lxml	BeautifulSoup	解析速度更快
PyMySQL	MySQL Connector	纯Python实现，兼容性好

系统工作流程分为四个阶段：

1. 用户认证与会话保持
2. 习题数据抓取与解析
3. 数据清洗与结构化处理
4. 数据库存储与索引建立

提示：在实际开发前，建议先用浏览器开发者工具分析雨课堂的API请求结构，这对后续爬虫编写至关重要。

2. 核心功能实现详解

2.1 用户认证与会话管理

雨课堂采用Cookie-based认证机制，我们需要先获取有效的会话凭证：

import requests

def get_session(cookie_str):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
        'Cookie': cookie_str
    }
    session = requests.Session()
    session.headers.update(headers)
    return session

获取Cookie的三种方法：

1. 浏览器登录后从开发者工具复制
2. 使用Selenium自动化登录获取
3. 通过OAuth授权流程获取（需平台支持）

2.2 习题数据抓取策略

分析雨课堂接口发现，习题数据主要通过以下API获取：

https://changjiang.yuketang.cn/v2/api/web/cards/detlist/{courseware_id}?classroom_id={class_id}

实现分页抓取的代码示例：

def fetch_exercises(session, course_id):
    base_url = f"https://changjiang.yuketang.cn/v2/api/web/logs/learn/{course_id}"
    params = {
        'actype': -1,
        'page': 0,
        'offset': 20,
        'sort': -1
    }
    response = session.get(base_url, params=params).json()
    return [item['courseware_id'] for item in response['data']['activities']]

2.3 数据结构化处理

原始数据是嵌套的JSON结构，需要多层解析：

from lxml import etree

def parse_question(data):
    try:
        # 提取题目正文
        html = data['slide']['ProblemBodys'][0]['Paragraphs'][0]['Lines'][0]['Html']
        question = etree.HTML(html).xpath('//span/text()')[0]
        
        # 处理选择题选项
        options = []
        for bullet in data['slide']['Problem']['Bullets']:
            option_html = bullet['Contents'][0]['Paragraphs'][0]['Lines'][0]['Html']
            option_text = etree.HTML(option_html).xpath('//span/text()')[0]
            options.append(f"{bullet['Label']}: {option_text}")
            
        return {
            'question': question.strip(),
            'answer': data['answer'],
            'options': '\n'.join(options)
        }
    except KeyError:
        return None

3. 数据库设计与优化

3.1 表结构设计

考虑到后续的查询需求，我们设计了三张关联表：

-- 题目主表
CREATE TABLE `questions` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `content` TEXT NOT NULL,
  `course_id` VARCHAR(32) NOT NULL,
  `create_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  INDEX `idx_course` (`course_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 选择题选项表
CREATE TABLE `choices` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `question_id` INT NOT NULL,
  `option_key` CHAR(1) NOT NULL,
  `option_content` TEXT NOT NULL,
  PRIMARY KEY (`id`),
  FOREIGN KEY (`question_id`) REFERENCES `questions`(`id`),
  INDEX `idx_question` (`question_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 答案表
CREATE TABLE `answers` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `question_id` INT NOT NULL,
  `correct_answer` VARCHAR(32) NOT NULL,
  `explanation` TEXT,
  PRIMARY KEY (`id`),
  FOREIGN KEY (`question_id`) REFERENCES `questions`(`id`),
  UNIQUE KEY `uniq_question` (`question_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

3.2 批量插入优化

使用executemany实现批量插入，提升性能：

def save_to_db(questions, db_config):
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor() as cursor:
            # 批量插入题目
            q_sql = """INSERT INTO questions (content, course_id) VALUES (%s, %s)"""
            cursor.executemany(q_sql, [
                (q['question'], q['course_id']) for q in questions
            ])
            
            # 获取自增ID
            question_ids = cursor.lastrowid - len(questions) + 1
            
            # 批量插入选项和答案
            c_sql = """INSERT INTO choices (question_id, option_key, option_content) VALUES (%s, %s, %s)"""
            a_sql = """INSERT INTO answers (question_id, correct_answer) VALUES (%s, %s)"""
            
            choice_data = []
            answer_data = []
            for idx, q in enumerate(questions):
                for option in q['options'].split('\n'):
                    if ':' in option:
                        key, content = option.split(':', 1)
                        choice_data.append((question_ids + idx, key.strip(), content.strip()))
                
                answer_data.append((question_ids + idx, q['answer']))
            
            cursor.executemany(c_sql, choice_data)
            cursor.executemany(a_sql, answer_data)
        
        conn.commit()
    finally:
        conn.close()

4. 系统增强与错误处理

4.1 健壮性提升策略

• 自动重试机制：对网络请求添加指数退避重试
• Cookie刷新：检测401状态码自动重新登录
• 异常处理：区分可恢复和不可恢复错误

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_fetch(session, url):
    try:
        response = session.get(url)
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as e:
        if e.response.status_code == 401:
            refresh_cookie(session)
            raise
        return None

4.2 日志记录与监控

添加详细的日志记录帮助排查问题：

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('exercise_backup.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def backup_process(course_id, db_config):
    logger.info(f"Starting backup for course {course_id}")
    try:
        session = get_session(get_cookie())
        exercise_ids = fetch_exercises(session, course_id)
        questions = []
        
        for eid in exercise_ids:
            data = safe_fetch(session, f"https://.../{eid}")
            if data:
                question = parse_question(data)
                if question:
                    question['course_id'] = course_id
                    questions.append(question)
        
        save_to_db(questions, db_config)
        logger.info(f"Successfully saved {len(questions)} questions")
    except Exception as e:
        logger.error(f"Backup failed: {str(e)}", exc_info=True)
        raise

5. 应用扩展与进阶功能

5.1 定时自动备份

使用APScheduler实现定时任务：

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('cron', hour=2)  # 每天凌晨2点运行
def daily_backup():
    courses = get_tracking_courses()  # 从配置读取需要备份的课程
    for course in courses:
        backup_process(course['id'], db_config)

if __name__ == '__main__':
    scheduler.start()

5.2 数据导出与可视化

支持多种格式导出：

• Markdown格式的复习笔记
• Anki记忆卡片
• Excel统计报表

def export_to_markdown(course_id, output_file):
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("""
                SELECT q.content, GROUP_CONCAT(c.option_key, ': ', c.option_content SEPARATOR '\n') AS options, 
                       a.correct_answer
                FROM questions q
                LEFT JOIN choices c ON q.id = c.question_id
                LEFT JOIN answers a ON q.id = a.question_id
                WHERE q.course_id = %s
                GROUP BY q.id
            """, (course_id,))
            
            with open(output_file, 'w', encoding='utf-8') as f:
                for row in cursor.fetchall():
                    f.write(f"### {row['content']}\n\n")
                    f.write(f"{row['options']}\n\n")
                    f.write(f"**正确答案**: {row['correct_answer']}\n\n")
                    f.write("---\n\n")
    finally:
        conn.close()

5.3 智能复习功能

基于遗忘曲线设计复习提醒：

-- 添加复习记录表
CREATE TABLE `review_logs` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `question_id` INT NOT NULL,
  `review_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  `correct` TINYINT(1) NOT NULL,
  `next_review_days` INT NOT NULL,
  PRIMARY KEY (`id`),
  FOREIGN KEY (`question_id`) REFERENCES `questions`(`id`),
  INDEX `idx_question` (`question_id`),
  INDEX `idx_review` (`review_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

实现基于SM-2算法的复习调度：

def calculate_next_review(question_id, is_correct):
    # 获取最近的复习记录
    conn = pymysql.connect(**db_config)
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("""
                SELECT next_review_days 
                FROM review_logs 
                WHERE question_id = %s 
                ORDER BY review_time DESC 
                LIMIT 1
            """, (question_id,))
            
            last_review = cursor.fetchone()
            if last_review:
                last_interval = last_review['next_review_days']
                if is_correct:
                    new_interval = min(last_interval * 2, 365)  # 最大间隔1年
                else:
                    new_interval = max(1, last_interval // 2)  # 至少1天后复习
            else:
                new_interval = 1 if is_correct else 1  # 首次回答正确1天后，错误当天复习
            
            # 记录本次复习
            cursor.execute("""
                INSERT INTO review_logs 
                (question_id, correct, next_review_days)
                VALUES (%s, %s, %s)
            """, (question_id, is_correct, new_interval))
            conn.commit()
            
            return datetime.now() + timedelta(days=new_interval)
    finally:
        conn.close()