高校数据采集教学实战：静态/动态/登录态三类场景闭环方案

简介本资源是一套面向高校数据科学与大数据技术课程的教学实践源码专为数据采集技术作业设计适用于本科生课程实训与自学提升。包内共69个文件含8个Python脚本实现网页抓取、HTML解析、数据清洗与存储等核心逻辑、9个HTML页面涵盖表单交互、结果展示与模板渲染、48张PNG/JPG图片含流程图、界面截图与可视化示例以及readme.txt说明文档和.keep占位文件整体压缩包38.08MB结构按“第一次作业”至“第四次作业”分层组织便于循序渐进学习。已有328人下载学习配套代码完整可运行覆盖requestsBeautifulSoup基础爬虫、静态页面解析、本地数据持久化等典型教学场景且每个作业模块均含独立main.py与templates目录利于学生理解前后端协同逻辑并开展二次开发。1. 这不是爬虫作业模板而是数据科学课里「能交、能跑、能讲清楚」的数据采集实战闭环“基于数据科学与大数据技术课程的数据采集技术作业设计源码”——这个标题背后不是一份交完就扔的 Python 脚本合集而是一套紧扣高校《数据科学与大数据技术》课程教学目标、覆盖真实数据采集全链路、且能支撑学生答辩陈述的技术落地方案。它解决的痛点非常具体学生常把 requests BeautifulSoup 当成万能解结果在遇到反爬策略、动态渲染、登录态维持、数据清洗一致性时集体翻车教师则苦于作业千篇一律、无法评估学生对「采集逻辑设计—异常鲁棒性—数据质量校验」这一完整能力维度的真实掌握程度。本方案聚焦「教学可验证性」所有源码均基于 Python 3.9、requests、selenium无头模式、pandas 和 logging 构建不依赖任何云服务或第三方平台账号本地即可完成从目标分析、请求构造、页面解析、字段映射、存储验证到日志回溯的全流程。适合高校教师布置课程作业、助教批改评分也适合学生作为毕业设计前期数据获取模块直接复用——重点不在“爬得多”而在“采得稳、理得清、说得明”。2. 从教学目标出发为什么选这三类典型采集场景而不是写个通用爬虫框架课程作业不是工程交付核心是考察学生对「数据来源特性—采集技术匹配度—质量控制意识」三者的综合判断能力。我们不堆砌技术而是按教学递进逻辑锁定三个最具代表性的采集类型静态 HTML 页面如政府公开数据门户、JavaScript 渲染页面如雪球股吧评论、带登录态的 API 接口如高校教务系统成绩查询页。每类都对应课程大纲中明确要求的能力点HTML 解析考察 DOM 结构理解与 XPath/CSS 选择器熟练度动态渲染考察浏览器自动化与等待策略设计登录态采集则直击会话管理、Token 提取与请求头构造等真实业务难点。这种设计让教师能通过学生代码中的 selector 写法、显式等待超时设置、cookies 序列化方式等细节精准识别其是否真正理解底层机制而非调包糊弄。2.1 静态页面采集以「中国气象数据网历史天气」为例的最小可行实现这类页面结构稳定、无 JS 渲染、无登录门槛是教学起点。关键不是“能不能拿到”而是“怎么拿得规范、可复现、可追溯”。我们采用 requests lxml 组合强制要求学生封装get_page()函数并注入 User-Agent 与 Referer同时记录 HTTP 状态码与响应耗时——这是培养工程习惯的第一步。import requests from lxml import etree import time import logging def get_page(url: str, timeout: int 10) - tuple[bool, str, dict]: 返回 (是否成功, 响应文本, 元信息字典) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://data.cma.cn/ } start_time time.time() try: resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() elapsed time.time() - start_time return True, resp.text, { status_code: resp.status_code, elapsed_sec: round(elapsed, 3), content_length: len(resp.content) } except Exception as e: elapsed time.time() - start_time logging.error(fGET {url} failed: {e}, elapsed{round(elapsed,3)}s) return False, , {error: str(e), elapsed_sec: round(elapsed, 3)} # 使用示例 success, html, meta get_page(https://data.cma.cn/data/detail/dataCode/A.0012.html) if success: tree etree.HTML(html) title tree.xpath(//title/text())[0].strip() if tree.xpath(//title/text()) else N/A print(f✅ 获取成功 | 标题: {title} | 耗时: {meta[elapsed_sec]}s)参数说明timeout10是教学硬约束——避免学生写死time.sleep(5)导致批量采集时长不可控meta字典强制记录elapsed_sec和content_length为后续性能分析和异常归因留证据logging.error不仅打日志还必须包含 URL 和耗时方便助教快速定位是网络问题还是 selector 错误。2.2 动态渲染页面采集用 Selenium 无头模式抓取雪球股吧实时评论雪球股吧是典型教学案例评论区由 AJAX 加载URL 不变但内容随滚动刷新且存在防自动化检测如 navigator.webdriver 检测。这里不教“绕过反爬”而教“如何设计合理等待与容错”。我们禁用图片加载、设置 window.navigator.webdriver 为 false并采用WebDriverWaitexpected_conditions.presence_of_element_located等待关键元素出现而非time.sleep()。from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import time def init_driver() - webdriver.Chrome: options Options() options.add_argument(--headless) # 无头模式 options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) options.add_argument(--disable-gpu) options.add_argument(--disable-extensions) options.add_argument(--disable-images) # 节省带宽与内存 # 关键欺骗 navigator.webdriver 检测 options.add_experimental_option(excludeSwitches, [enable-automation]) options.add_experimental_option(useAutomationExtension, False) driver webdriver.Chrome(optionsoptions) driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }) }) return driver def scrape_xueqiu_comments(stock_code: str, max_pages: int 3) - list[dict]: driver init_driver() wait WebDriverWait(driver, 10) # 显式等待最长10秒 comments [] try: url fhttps://xueqiu.com/S/{stock_code}/posts driver.get(url) # 等待帖子列表容器出现雪球页面结构 wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, div.timeline__item))) for page in range(max_pages): # 滚动到底部触发懒加载 driver.execute_script(window.scrollTo(0, document.body.scrollHeight);) time.sleep(1.5) # 给JS留出渲染时间非固定等待 # 提取当前页所有评论块 post_elements driver.find_elements(By.CSS_SELECTOR, div.timeline__item) for elem in post_elements[-10:]: # 取最后10条避免重复 try: author elem.find_element(By.CSS_SELECTOR, a.user-name).text.strip() content elem.find_element(By.CSS_SELECTOR, div.post__content).text.strip() time_str elem.find_element(By.CSS_SELECTOR, span.time).text.strip() comments.append({ author: author, content: content[:200] ... if len(content) 200 else content, post_time: time_str, source_url: driver.current_url }) except Exception as e: continue # 单条评论解析失败不影响整体 # 尝试点击下一页雪球分页按钮 try: next_btn driver.find_element(By.CSS_SELECTOR, button.next) if disabled in next_btn.get_attribute(class): break next_btn.click() wait.until(EC.staleness_of(post_elements[0])) # 等待旧元素失效 except: break # 无下一页按钮则退出 finally: driver.quit() return comments # 调用示例注意实际教学中需提供 stock_code 白名单如 SH600519 comments scrape_xueqiu_comments(SH600519, max_pages2) print(f✅ 抓取到 {len(comments)} 条评论)逻辑说明execute_cdp_cmd注入脚本是绕过navigator.webdriver检测的教学标准解法比修改 chromedriver 二进制更安全、更易理解wait.until(EC.staleness_of(...))是判断页面是否刷新的关键——它比time.sleep()更精准且能暴露学生是否理解 DOM 元素生命周期post_elements[-10:]的取法是为了应对懒加载重复抓取这是学生最容易忽略的细节。2.3 登录态 API 采集模拟高校教务系统成绩查询的会话保持这是课程作业的“高阶关卡”。学生必须理解 Cookie、Session、Token 三者区别并能手动构造符合服务端校验逻辑的请求链。我们以某高校教务系统模拟为例先 GET 登录页获取隐藏字段如lt,execution再 POST 表单提交最后用返回的 cookies 访问成绩接口。全程禁用requests.Session()的自动 cookie 管理要求学生显式提取、序列化、复用——因为考试时可能被要求手写 curl 命令验证。import requests from bs4 import BeautifulSoup import re import json def login_and_fetch_grades(login_url: str, grade_api_url: str, username: str, password: str) - dict: 手动管理会话1. GET登录页提取隐藏字段 2. POST登录 3. GET成绩页 返回 { status: success/fail, data: [...] 或 error: xxx } s requests.Session() s.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) # Step 1: GET 登录页提取隐藏字段 try: resp1 s.get(login_url, timeout10) resp1.raise_for_status() soup BeautifulSoup(resp1.text, html.parser) lt soup.find(input, {name: lt})[value] if soup.find(input, {name: lt}) else execution soup.find(input, {name: execution})[value] if soup.find(input, {name: execution}) else # Step 2: POST 登录表单 login_data { username: username, password: password, lt: lt, execution: execution, _eventId: submit } resp2 s.post(login_url, datalogin_data, timeout10) # 检查是否登录成功根据重定向或页面标题 if resp2.url login_url or 用户名或密码错误 in resp2.text: return {status: fail, error: 登录失败} # Step 3: GET 成绩接口教务系统通常返回 JSON resp3 s.get(grade_api_url, timeout10) resp3.raise_for_status() # 强制解析为 JSON失败则返回原始文本供调试 try: data resp3.json() return {status: success, data: data} except json.JSONDecodeError: return {status: fail, error: 成绩接口返回非JSON格式, raw_text: resp3.text[:500]} except Exception as e: return {status: fail, error: f请求异常: {str(e)}} # 使用示例教学中提供 mock 登录页与成绩接口地址 result login_and_fetch_grades( login_urlhttps://jwxt.example.edu.cn/login, grade_api_urlhttps://jwxt.example.edu.cn/api/student/grades, username20230001, password****** ) if result[status] success: print(f✅ 成功获取 {len(result[data])} 门课程成绩) else: print(f❌ {result[error]})参数说明s.headers.update(...)确保 User-Agent 一致避免服务端因 header 差异拒绝会话lt和execution字段的提取是 CSRF 防护的典型体现学生必须理解其作用resp2.url login_url是判断登录失败的教学级判据——比检查状态码更贴近真实场景很多教务系统登录失败仍返回 200json.JSONDecodeError捕获强制要求学生处理接口格式变更这是生产环境必备意识。3. 教学级数据质量校验不是存成 CSV 就算完成而是用 pandas 做字段级可信度审计作业提交物不只是.py文件还必须包含output/目录下的结构化数据文件CSV/JSON及配套的quality_report.md。我们要求学生用 pandas 对采集结果做三类校验完整性校验空值率 5%、一致性校验日期字段符合 ISO 格式、数值字段无乱码、逻辑校验如天气温度在 -100~60℃ 之间、股吧评论长度 5 字符。这些不是附加题而是评分项。3.1 完整性与一致性校验脚本validate_weather_data.pyimport pandas as pd import numpy as np import re from datetime import datetime def validate_weather_csv(file_path: str) - dict: 对气象数据 CSV 做字段级校验返回 {字段名: [问题列表]} try: df pd.read_csv(file_path, encodingutf-8) except UnicodeDecodeError: df pd.read_csv(file_path, encodinggbk) # 兼容中文 Windows 默认编码 report {} # 1. 完整性校验空值率 for col in df.columns: null_ratio df[col].isnull().mean() if null_ratio 0.05: # 超过5%即告警 report[col] report.get(col, []) [f空值率 {null_ratio:.1%} 5%] # 2. 一致性校验日期格式 if date in df.columns: date_pattern r^\d{4}-\d{2}-\d{2}$ invalid_dates df[date].dropna().apply(lambda x: not re.match(date_pattern, str(x))) if invalid_dates.any(): report[date] report.get(date, []) [f发现 {invalid_dates.sum()} 个非法日期格式] # 3. 逻辑校验温度范围 for temp_col in [max_temp, min_temp, avg_temp]: if temp_col in df.columns: # 尝试转为数值失败则标记 try: temps pd.to_numeric(df[temp_col], errorsraise) out_of_range ((temps -100) | (temps 60)).sum() if out_of_range 0: report[temp_col] report.get(temp_col, []) [f发现 {out_of_range} 个超限温度值] except: report[temp_col] report.get(temp_col, []) [f无法转为数值请检查字段内容] return report # 调用示例 report validate_weather_csv(output/weather_2023.csv) if report: print(⚠️ 数据质量问题:) for field, issues in report.items(): print(f {field}: {; .join(issues)}) else: print(✅ 数据质量校验通过)逻辑说明encodinggbk是针对国内气象网站导出 CSV 的常见编码兼容处理属于血泪经验pd.to_numeric(..., errorsraise)强制类型转换而非静默忽略让学生直面脏数据re.match(date_pattern, str(x))用正则而非pd.to_datetime()因为后者会自动纠错如把2023-13-01转成2024-01-01掩盖真实问题——教学目的就是暴露问题。3.2 自动生成质量报告generate_quality_report.pyimport pandas as pd from datetime import datetime def generate_report(csv_files: list[str], output_md: str quality_report.md): 汇总所有 CSV 的校验结果生成 Markdown 报告 all_reports {} total_issues 0 for csv_file in csv_files: try: report validate_weather_csv(csv_file) # 复用上节函数 all_reports[csv_file] report total_issues sum(len(v) for v in report.values()) except Exception as e: all_reports[csv_file] {ERROR: [f校验脚本执行失败: {e}]} total_issues 1 # 生成 Markdown with open(output_md, w, encodingutf-8) as f: f.write(f# 数据质量校验报告\n\n生成时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n\n) f.write(f## 总体概览\n- 检查文件数{len(csv_files)}\n- 发现问题总数{total_issues}\n\n) for csv_file, report in all_reports.items(): f.write(f## {csv_file}\n) if not report: f.write(- ✅ 无问题\n\n) else: for field, issues in report.items(): f.write(f- {field}:\n) for issue in issues: f.write(f - {issue}\n) f.write(\n) print(f✅ 质量报告已生成{output_md}) # 使用示例 generate_report([output/weather_2023.csv, output/grades_2023.csv])参数说明output_md参数允许学生指定报告路径便于集成到作业提交目录结构报告中明确区分✅ 无问题和-{field} 问题列表助教可快速扫描时间戳记录是教学过程可追溯性的基础要求。4. 避坑指南学生在数据采集作业中最常踩的 4 个坑以及助教一眼就能识别的信号这些不是“报错怎么办”的泛泛而谈而是从三年课程助教批改 1276 份作业中提炼出的教学级高频翻车点。每个坑都对应一个评分扣分项且助教能在 30 秒内通过代码/日志/输出文件定位。4.1 坑一requests 不设 timeout导致作业运行卡死在实验室机房现象学生代码在本地能跑通但提交到学校统一实验环境后长时间无响应最终超时被杀进程。原因未设置timeout参数当目标服务器响应慢或网络抖动时requests 会无限等待默认 timeout 为 None。解决所有requests.get()/requests.post()必须显式传入timeout参数且值不超过 15 秒。教学要求timeout值需写在配置字典中如CONFIG {timeout: 10}不得硬编码在请求行内便于统一调整。4.2 坑二BeautifulSoup 解析时用find_all()无条件遍历却没处理空列表现象程序运行不报错但 CSV 中大量字段为空学生自己都看不出哪里错了。原因soup.find_all(div, class_price)返回空列表[]后续price_elem[0].text直接 IndexError但学生用try-except吞掉异常或用if price_elem:判断后直接跳过导致数据缺失。解决强制要求使用find()get_text(stripTrue)组合并对返回None做默认值填充。例如price soup.find(span, class_price).get_text(stripTrue) if soup.find(span, class_price) else N/A。助教检查点代码中是否存在裸find_all()[0]或未判空的索引访问。4.3 坑三Selenium 等待策略混乱混用time.sleep()和WebDriverWait现象采集结果时好时坏同一份代码今天成功、明天失败学生归因为“网络不好”。原因在WebDriverWait等待后又加time.sleep(2)造成双重等待或对动态加载元素用presence_of_element_located但元素实际需要visibility_of_element_located如被 CSSdisplay:none隐藏。解决教学明确规定——禁用time.sleep()所有等待必须用WebDriverWait 明确的expected_conditions。助教检查点代码中import time且存在time.sleep(...)调用即扣分WebDriverWait的timeout值必须 ≤ 15 秒。4.4 坑四CSV 输出不指定 encoding中文在 Excel 里显示为乱码现象学生声称“数据已保存”但助教用 Excel 打开全是方块用记事本打开却是正常的。原因df.to_csv(output.csv)默认用系统编码Windows 是 gbk但 Excel for Windows 默认读取 UTF-8 BOM 编码。解决强制要求to_csv(..., encodingutf_8_sig)。utf_8_sig会在文件开头写入 BOM确保 Excel 正确识别 UTF-8。助教检查点to_csv调用中缺失encoding参数即判定为不合格。提示以上四坑均已在课程 PPT 的「作业雷区清单」页列出学生提交前需自查打钩。这不是刁难而是训练生产环境必备的防御性编程意识。5. 教学落地技巧用 logging 模块构建「可回溯、可评分、可归因」的采集过程证据链很多学生以为打印几行print(正在抓取...)就是日志但教学需要的是能支撑评分、归因、复现的结构化过程证据。我们要求所有采集脚本必须使用 Python 标准logging模块且配置必须满足三个硬性条件1日志级别为 INFO2输出到logs/目录下的唯一文件文件名含时间戳3每条日志必须包含采集目标、HTTP 状态码、耗时、数据条数四个字段。这不是炫技而是让助教在 2 分钟内完成一份作业的「过程可信度」评估。5.1 标准化日志配置setup_logging.pyimport logging import os from datetime import datetime def setup_logger(name: str data_collection) - logging.Logger: 创建带时间戳的独立日志器确保每次运行日志隔离 # 创建 logs 目录 os.makedirs(logs, exist_okTrue) # 生成带时间戳的日志文件名 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) log_file flogs/{name}_{timestamp}.log # 配置 logger logger logging.getLogger(name) logger.setLevel(logging.INFO) # 避免重复添加 handler if not logger.handlers: # 文件 handler file_handler logging.FileHandler(log_file, encodingutf-8) file_handler.setLevel(logging.INFO) # 控制台 handler仅 INFO 级别 console_handler logging.StreamHandler() console_handler.setLevel(logging.INFO) # 格式化器强制包含采集目标、状态码、耗时、条数 formatter logging.Formatter( %(asctime)s | %(levelname)-8s | %(name)s | target%(target)s | status%(status_code)s | elapsed%(elapsed_sec).3fs | count%(count)d | %(message)s, datefmt%Y-%m-%d %H:%M:%S ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 使用示例在主采集函数中 logger setup_logger(weather_crawler) logger.info(开始采集中国气象网数据, extra{target: https://data.cma.cn/, status_code: 200, elapsed_sec: 1.234, count: 0})参数说明extra字典是传递结构化字段的关键target/status_code/elapsed_sec/count四个 key 是硬性要求%(asctime)s自动注入时间encodingutf-8确保中文日志不乱码os.makedirs(logs, exist_okTrue)避免因目录不存在导致日志写入失败。5.2 日志驱动的采集流程main.py示例import logging from setup_logging import setup_logger from scraper import get_page, scrape_xueqiu_comments, login_and_fetch_grades logger setup_logger(course_assignment) def run_weather_task(): logger.info(启动气象数据采集任务, extra{target: https://data.cma.cn/data/detail/dataCode/A.0012.html, status_code: 0, elapsed_sec: 0, count: 0}) success, html, meta get_page(https://data.cma.cn/data/detail/dataCode/A.0012.html) if not success: logger.error(气象页面获取失败, extra{target: https://data.cma.cn/..., status_code: meta.get(status_code, 0), elapsed_sec: meta.get(elapsed_sec, 0), count: 0}) return # 解析逻辑... parsed_count 127 # 假设解析出127条记录 logger.info(气象数据解析完成, extra{target: https://data.cma.cn/..., status_code: 200, elapsed_sec: meta[elapsed_sec], count: parsed_count}) def run_xueqiu_task(): logger.info(启动雪球股吧采集任务, extra{target: https://xueqiu.com/S/SH600519/posts, status_code: 0, elapsed_sec: 0, count: 0}) comments scrape_xueqiu_comments(SH600519, max_pages2) logger.info(雪球评论采集完成, extra{target: https://xueqiu.com/S/SH600519/posts, status_code: 200, elapsed_sec: 0, # 实际应传入总耗时 count: len(comments)}) if __name__ __main__: run_weather_task() run_xueqiu_task() logger.info(全部采集任务结束, extra{target: N/A, status_code: 0, elapsed_sec: 0, count: 0})逻辑说明run_weather_task()中logger.info()的extra字段必须与get_page()返回的meta字典对齐确保elapsed_sec和count真实反映过程status_code在失败时取meta.get(status_code, 0)成功时硬写200形成可比基准target字段始终是原始 URL不因重定向改变——这是归因分析的基础。5.3 助教快速评分技巧用 grep 三分钟完成 50 份作业的过程审计助教不需要逐行读代码而是用 shell 命令快速扫描日志证据# 1. 检查是否所有作业都生成了 logs/ 目录 find . -name logs -type d | wc -l # 2. 检查日志中是否有 ERROR 级别记录表明采集失败 grep -r ERROR logs/ | wc -l # 3. 统计各作业采集到的有效数据条数提取 count 字段 grep -o count[0-9]* logs/*.log | sort | uniq -c | sort -nr # 4. 检查是否存在未设置 timeout 的 requests 调用代码层面 grep -r requests\.get . --include*.py | grep -v timeout | wc -l教学价值这些命令本身就被写入助教手册学生提交前也可自行运行。它把主观的“代码质量”转化为客观的“日志完备性”和“过程可量化性”让评分有据可依也让学生明白在数据科学领域过程证据和结果数据同等重要。我带过 7 届数据科学课最深的体会是学生不是不会写爬虫而是不知道“写给谁看”。写给机器看只要语法正确就行写给老师看要体现设计思路写给未来自己看得有日志和注释而写给真实业务看就必须有质量校验和过程证据。这份源码设计就是把这四重“读者意识”揉进每一行代码、每一个参数、每一份日志里。希望帮到你。本文还有配套的精品资源点击获取