Blackboard课程附件批量下载：认证、遍历与增量同步实践

简介：BlackboardDownloader 是一款面向高校学生与教师的 Java 小工具，用于批量抓取 Blackboard 平台各门课程中的全部文档，减少逐门课程、逐文件夹手动下载的重复操作。程序运行后提示输入 Blackboard 用户名和密码，并按平台原有组织方式在本地重建目录，教学大纲、作业、课程内容等文件会归入对应文件夹，便于后续归档与复习；下载深度限定为进入一层子文件夹，边界明确。压缩包约 51.56MB，上游文件总数显示为 0、类型明细暂无统计，从内容构成看以 Java 源码、可执行 JAR 和说明文本为主，使用前可解压核对实际文件。已有 234 人浏览或学习。它适合需要快速备份课程资料、梳理课程目录结构，或参考 Java 登录与批量下载实现思路的读者，使用时也应注意账号信息安全与平台访问规范。

1. 几百个附件不该靠手点：BlackboardDownloader 要替掉的是哪一段手工活

学期中段，十几门课的课件、习题、参考 PDF 散在 Blackboard 的各个目录里，靠鼠标一个个点开、另存为，半小时就没了，还容易漏掉藏在二级文件夹里的那一份。BlackboardDownloader 这类工具要替掉的正是这段纯体力的链路：用一次登录换到有效会话，枚举出账号选修的全部课程，递归展开每门课的内容树，把附件按“课程 / 章节 / 文件”的结构落到本地磁盘。它适合需要长期归档自己有权访问的课程资料的人——带多门课的助教、跨院系选课的研究生、要做离线复习的工程实践者。

边界得先说在前面：只下载账号本身有权限看到的内容，遵守平台的使用条款和资料版权，别做全站扫描，也别把并发开到把别人的服务拖垮。工具的价值在于“可重跑、可增量”，而不是“一次抓干净”。

2. 先把登录做稳：会话 cookie 和 OAuth2 令牌分别用在什么场景

批量下载的所有失败里，登录问题占一半以上，而且表现极具迷惑性——返回码 200，内容却是登录页。这一章把认证链路拆开，先解决“我是谁”，再谈怎么下。

2.1 两条认证链路：网页会话和 REST 令牌不是一回事

同一套课程管理系统通常同时暴露两个面：面向浏览器的网页界面，和面向程序的 REST 接口。网页界面靠会话 cookie 维持状态，表单里还嵌一个一次性的 CSRF nonce；REST 接口靠Authorization: Bearer <token>。两者的权限模型完全不同，混用就会得到那种“明明登录成功了，接口全 401”的玄学现场。

选型结论很直白：能注册应用、能拿到 key/secret，就走令牌；拿不到，就老老实实用会话。现实里绝大多数人是后者，所以两条链路都得会。

2.2 用 requests.Session 走完一次表单登录

会话模式的核心是“同一个 Session 对象贯穿始终”。先 GET 登录页拿 nonce 和初始 cookie，再用同一个 Session POST 表单，cookie 会自动带上；顺序错了，nonce 对不上，服务端只会安静地把你重定向回登录页。

import re, pickle, pathlib, requests BASE = "https://lms.example.edu" # 换成自己的平台域名 HOME_PATH = "/webapps/portal/execute/tabs/tabAction?tab_tab_group_id=_1_1" # 换成"我的课程"页地址 COOKIE_JAR = pathlib.Path("state/cookies.pkl") def login(user: str, pwd: str) -> requests.Session: s = requests.Session() # UA 贴近真实浏览器，部分网关会按 UA 做第一层拦截 s.headers.update({"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) BdDownloader/1.0"}) # 1) 先取登录页，拿到一次性 nonce 和初始 cookie page = s.get(f"{BASE}/webapps/login/", timeout=20) page.raise_for_status() m = re.search(r'name="[^"]*NonceUtil\.nonce"\s+value="([^"]+)"', page.text) nonce = m.group(1) if m else "" # 2) 同一个 Session 提交表单，cookie 自动续上 r = s.post(f"{BASE}/webapps/login/", data={ "user_id": user, "password": pwd, "action": "login", "blackboard.platform.security.NonceUtil.nonce": nonce, # 字段名按实际表单改 }, timeout=20, allow_redirects=True) r.raise_for_status() # 3) 用"登录后才存在"的页面判断成败，别看状态码 probe = s.get(BASE + HOME_PATH, timeout=20, allow_redirects=True) if probe.status_code != 200 or "login" in probe.url.lower(): raise RuntimeError(f"登录失败：最终 URL={probe.url}，状态={probe.status_code}") COOKIE_JAR.parent.mkdir(parents=True, exist_ok=True) pickle.dump(s.cookies, COOKIE_JAR.open("wb")) return s

几个参数值得单独说。allow_redirects=True必须开，因为登录成功后服务端会 302 到首页，你需要看的是“最终落在哪”。timeout=20是防守，不是优化，没超时的请求在校园网出口抖动时会挂死整个线程。把 cookie 序列化到磁盘有两个用处：一是第二天重跑不用再登录，二是出问题时能把这串 cookie 拿出来，用 curl 手工复现一次请求，快速区分“程序写错了”还是“会话过期了”。

失败时的排查顺序是固定的：先打印r.url，被重定向回登录域就说明表单没被接受；再把r.text的前 2KB 落盘，看是不是弹了二次验证或密码错误提示；最后看响应头里有没有下发会话 cookie，没有的话多半是 nonce 没取到。这条顺序我踩过坑，乱猜会浪费一整晚。

2.3 令牌模式下的取 token 与自动续期

令牌模式看起来更“正规”，但有个必须提前知道的坑：client_credentials拿到的是应用身份，“我的课程”这类以用户为主语的接口大概率返回空的results列表。这不是接口坏了，是权限模型如此。要拿到用户身份，必须走三方授权（authorization_code + 回调），把 refresh_token 持久化下来。

import os, time, requests _TOKEN_CACHE = {} GUARD = 120 # 提前 120 秒换新，避免卡在过期边界上 def get_token(base: str, scope: str = "read") -> str: now = time.time() hit = _TOKEN_CACHE.get(base) if hit and hit["exp"] - GUARD > now: return hit["value"] resp = requests.post( f"{base}/learn/api/public/v1/oauth2/token", data={"grant_type": "client_credentials", "scope": scope}, auth=(os.environ["LMS_KEY"], os.environ["LMS_SECRET"]), # 密钥不写进代码 timeout=20, ) resp.raise_for_status() j = resp.json() _TOKEN_CACHE[base] = { "value": j["access_token"], "exp": now + int(j.get("expires_in", 3600)), } return _TOKEN_CACHE[base]["value"] def auth_headers(base: str) -> dict: return {"Authorization": f"Bearer {get_token(base)}"}

GUARD=120这个数字不是拍脑袋：一次下载批次可能持续十几分钟，如果 token 在第 5 分钟过期，中间那几十个请求会集体 401 并把重试队列打满。提前换新比“失败了再换”省事得多。scope按最小权限申请，read够用就别加write。

2.4 登录成功的判定标准

把判定逻辑写成独立函数，所有下游代码只认这个函数的结果，不要在业务代码里散落if status_code == 200。判据三条：最终 URL 不在登录域、页面里能找到只有登录后才有的标识、调用一次课程列表接口不是 401。

def assert_logged_in(session, base, home_path): r = session.get(base + home_path, timeout=20, allow_redirects=True) if "login" in r.url.lower(): raise RuntimeError("被重定向回登录页，会话已失效") if r.status_code != 200: raise RuntimeError(f"首页返回 {r.status_code}，可能是网关或权限问题") return True

提示：会话过期在长任务里是必然事件，不是异常。给 Session 挂一个 401 拦截器，命中后重新登录一次再重放原请求，比重跑整个任务划算。

3. 拉全课程与内容树：分页兜底、递归遍历和按 handler 分派

认证过了，下一步是把“有哪些课、每门课里有哪些文件”这件事描述成结构化数据。这里的坑不在难，而在静默——接口不报错，只是少给你几条。

3.1 课程清单：limit/offset 分页必须自己兜底

分页接口的默认limit可能是 25 也可能是 100，取决于平台版本和租户配置。你以为拿到的是全部课程，实际只有第一页。兜底要做三件事：显式传limit；用“本页返回条数小于 limit”判断终止；加max_pages防死循环。

def iter_paged(session, url, params=None, page_size=100, max_pages=200): """通用分页迭代器：显式 limit + 短页终止 + 页数上限""" params = dict(params or {}) offset = 0 for _ in range(max_pages): params.update({"limit": page_size, "offset": offset}) r = session.get(url, params=params, timeout=30) r.raise_for_status() results = r.json().get("results", []) if not results: return yield from results if len(results) < page_size: # 短页即末页，比看总数可靠 return offset += len(results)

offset分页在数据频繁变动的场景下会漏也会重——翻页过程中如果前面插入了新课程，后面的整体位移一格。所以拿到课程列表后先按课程 id 去重，再落库，不要假设顺序稳定。max_pages=200是保险丝，防止某个接口在参数被忽略时无限返回同一页。

3.2 内容树：递归遍历与深度保护

内容树是典型的树结构：文件夹套文件夹，节点通过hasChildren标记还能不能往下。用栈做广度优先，配一个seen集合防环，再配一个max_depth防某些平台把软链接当子节点导致无限下钻。

def walk_contents(session, base, course_id, max_depth=6): """遍历一门课的内容树，产出 (depth, node)""" stack = [("", 0)] seen = set() while stack: parent, depth = stack.pop() if depth > max_depth: continue url = f"{base}/learn/api/public/v1/courses/{course_id}/contents" # 子节点的查询参数名以平台文档为准，这里用 parentId 占位 params = {"parentId": parent} if parent else {} for node in iter_paged(session, url, params): nid = node.get("id") if not nid or nid in seen: continue seen.add(nid) yield depth, node if node.get("hasChildren"): stack.append((nid, depth + 1))

max_depth=6是经验值。正常的课程结构是“课程 → 单元 → 章节 → 文件”，四层以内；超过六层基本是数据异常或者遍历逻辑有环。与其相信远端数据的整洁度，不如在本地设一道闸。

3.3 按 handler 分派：哪些节点有文件、哪些只是壳

内容节点的类型由contentHandler.id决定。判断“这个节点是不是文件”，别去看标题后缀，标题可以随便写；要看contentHandler.id和attachments字段。

这张表的意义在于控制下载范围。把所有 handler 一视同仁地当文件抓，会得到一堆只有标题没有内容的空壳目录，白白拉长任务时间。

3.4 附件与内嵌文件：一个节点可能挂多个文件

一个 document 节点可能挂 0 到 n 个附件，每个附件有自己的 id、名字、MIME、大小。去重键用(contentId, attachmentId)，不要用文件名——同一门课里两个老师都把文件叫lecture1.pdf是常态。落盘时用附件的fileName而不是节点标题，因为标题常常是“第一次作业”这种没有扩展名的中文短语。

for depth, node in walk_contents(session, BASE, course_id): for att in node.get("attachments", []): key = (node["id"], att["id"]) if key in done: continue queue.put({ "course": course_id, "rel_dir": "/".join(path_stack[:depth]), "name": att.get("fileName") or node.get("title", "unnamed"), "url": f"{BASE}/learn/api/public/v1/courses/{course_id}" f"/contents/{node['id']}/attachments/{att['id']}/downloads", })

正文里内嵌的图片走的是另一套地址，一般不纳入批量下载范围，除非目标是做完整离线镜像。区别在于前者是“资料”，后者是“页面快照”，混在一起会让目录结构变得没法看。

4. 下载主干怎么搭：并发限速、断点续传与落盘命名规则

前两章解决“能不能拿到地址”，这一章解决“拿得稳、拿得全、拿完还能找得到”。批量下载器写崩的地方，八成在这个环节。

4.1 并发模型：为什么 4 个线程比 32 个快

直觉上并发越高越快，实际上服务端的限流阈值才是真瓶颈。把并发开到 32，前 30 秒确实飞起，然后开始大面积 429；所有线程一起退避，总吞吐反而低于稳定跑 4 个线程。合理的组合是：工作线程 4～8，全局 QPS 2～3，桶容量 4。

import threading, time class Bucket: """令牌桶：控制全局请求速率，而不是单线程速率""" def __init__(self, rate=2.0, burst=4): self.rate, self.burst = rate, burst self.tokens, self.ts = burst, time.monotonic() self.lock = threading.Lock() def take(self): with self.lock: now = time.monotonic() self.tokens = min(self.burst, self.tokens + (now - self.ts) * self.rate) self.ts = now if self.tokens < 1: time.sleep((1 - self.tokens) / self.rate) self.tokens = 0.0 else: self.tokens -= 1.0

桶要放在共享层，所有线程共用一个实例。每个线程各持一个桶等于没有限速，这一点在第一次写的时候特别容易搞错。

4.2 流式下载与 .part 原子落盘

课件里出现 200MB 的录屏压缩包不稀奇，stream=True是必须的，否则内存会被撑爆。落盘先写.part，写完再os.replace原子改名，这样即使进程被杀，目录里也不会留下半截文件被下一次同步误判成“已下载”。

import os from pathlib import Path def download(session, url, dest: Path, chunk=1 << 16): dest.parent.mkdir(parents=True, exist_ok=True) part = dest.with_suffix(dest.suffix + ".part") have = part.stat().st_size if part.exists() else 0 headers = {"Range": f"bytes={have}-"} if have else {} with session.get(url, headers=headers, stream=True, timeout=60) as r: if have and r.status_code == 200: have = 0 # 服务端不支持 Range，从头来 part.unlink(missing_ok=True) elif r.status_code not in (200, 206): r.raise_for_status() with part.open("ab" if have else "wb") as f: for block in r.iter_content(chunk): if block: f.write(block) os.replace(part, dest) # 原子改名，杜绝半截文件 return dest

chunk = 1 << 16即 64KB，是个折中值：太小系统调用多，太大占内存且断点粒度粗。对超大文件可以放到 1MB，但对一堆几十 KB 的 PDF 就没必要了。

4.3 文件名与路径：一套必须写死的清洗规则

文件名是最容易出问题的环节，特别是在 Windows 上跑、文件来自 Linux 服务端的组合。规则必须写死，不能靠“应该没问题”。

import re, unicodedata WIN_RESERVED = {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)), *(f"LPT{i}" for i in range(1, 10))} def safe_name(raw: str, fallback="unnamed", max_bytes=180) -> str: name = unicodedata.normalize("NFC", raw or "") # 统一 Unicode 形式 name = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "_", name) # 非法字符 name = name.strip(" .") # Windows 不允许尾部空格和点 if not name: return fallback if name.split(".")[0].upper() in WIN_RESERVED: name = "_" + name while len(name.encode("utf-8")) > max_bytes: # 按字节截断 name = name[:-1] return name or fallback

max_bytes=180是为了给完整路径留余量。Windows 的路径长度限制算的是整条绝对路径，你按文件名 255 去截，到了深层目录照样超。按字节截断而不是按字符截断，是因为中文一个字占 3 字节，按字符截会低估实际长度。NFC归一化看着多余，实际上 macOS 存的文件名是 NFD 形式，不统一的话同一个文件在两次同步里会被认成两个。

重名冲突的解法是在文件名后加 6 位短 hash，而不是_1、_2——后者在增量同步时顺序一变就全乱套。

4.4 重试与退避：429/5xx 的处理姿势

重试不是“失败了再来一次”这么简单。要区分四类：408/429/5xx 值得重试；401 要去重新登录；403 说明没权限，重试多少次都没用；404 说明资源没了，应该从状态库里标记删除。

import random, time from functools import wraps def retry(max_tries=5, base=1.5, cap=60): def deco(fn): @wraps(fn) def wrapper(*a, **kw): for i in range(max_tries): try: return fn(*a, **kw) except Exception as e: resp = getattr(e, "response", None) status = getattr(resp, "status_code", None) if i == max_tries - 1: raise if status and status not in (408, 429, 500, 502, 503, 504): raise ra = (getattr(resp, "headers", {}) or {}).get("Retry-After") wait = float(ra) if ra else min(cap, base ** i) * (0.5 + random.random()) time.sleep(wait) return wrapper return deco

Retry-After优先于指数退避，因为那是服务端明确告诉你的等待时间，比你猜的准。加 0.5～1.5 的随机抖动，是为了避免所有线程在同一时刻醒来再次撞墙——这一点在 4 个线程同时被 429 时体现得特别明显。

5. 避坑与排查：五个让批量下载半夜翻车的现场

这一章是按“现象 → 原因 → 解决”整理的真实故障模式，每一条都能让你少熬一个晚上。

5.1 登录返回 200，后续接口全是 401

现象：登录函数不报错，probe也过了，但一调课程接口就 401，重试几次还是 401，日志里看不出任何异常。

原因：把网页会话的 cookie 直接拿去调 REST 接口。这两个面用的是不同鉴权机制，cookie 在 REST 层不被识别。另一种可能是 token 拿的是client_credentials，代表应用身份，而接口要求用户上下文。

解决：把两条链路的请求分开封装，页面抓取用Session，接口调用统一走auth_headers()。如果确实只能用令牌又需要用户身份，就走三方授权换 refresh_token。在probe里加一次真实业务接口的探测，比只看首页可靠得多。

5.2 课程列表永远只有第一页

现象：明明选了十几门课，程序只处理了 5 门或 25 门，日志显示“遍历完成”，没有任何报错。

原因：没传limit，用了服务端默认值；或者传了但被网关改写。更隐蔽的是用total字段判断终止，而那个字段在某些版本里恒为 0。

解决：显式传limit=100，用“本页条数 < limit”判终止，同时打印每页的offset和条数，跑一次就能从日志里看出是不是卡在第一页。加一层“最终课程数 < 预期数”的断言，最近一次平台改配置就是靠这个断言发现的。

5.3 文件下来了，名字全是乱码或者互相覆盖

现象：Content-Disposition里的中文名变成%E8%AF%BE%E4%BB%B6.pdf或者一串?，目录里只剩最后一个同名文件。

原因：一是没按 RFC 5987 解析filename*=UTF-8''...，只读了filename=；二是用节点标题当文件名，同一章下多个附件标题相同；三是没做路径清洗，带斜杠的名字被当成子目录。

解决：优先读filename*，解不出再退回filename；去重键用(contentId, attachmentId)；所有落地名过一遍safe_name()；重名时追加 6 位 hash。三条一起做，乱码和覆盖会同时消失。

5.4 跑到一半开始大面积 429 和 502

现象：前 200 个文件下得飞快，之后错误率陡增，重试也救不回来，任务卡在 60% 不动。

原因：并发太高，触发了服务端按用户或按 IP 的限流；或者重试没读Retry-After，所有线程在同一秒集体重试，把限流窗口一直续着。

解决：全局并发降到 4～8，加令牌桶把 QPS 压在 2～3；重试优先用Retry-After，没有就指数退避加抖动；把 429 单独计数，超过阈值就主动降速一半。这套下来，任务时间可能从 20 分钟变 40 分钟，但能跑完。

5.5 增量同步把更新过的文件当成了旧文件

现象：老师换了新版课件，重跑脚本后本地还是旧版本，程序认为“文件已存在，跳过”。

原因：只用文件名或路径判断是否已下载，完全没看大小、修改时间和内容指纹。

解决：状态库里存size、etag、last_modified，判定逻辑是“路径相同且 size 与 etag 都一致才跳过”。如果平台不返回 etag，就退回“size + 修改时间”，再不行就对小文件做内容 hash。宁可多下一次，不要漏掉一版。

6. 增量同步加抽样校验：让这套脚本一个学期不用盯着

第一次全量下完之后，真正决定这套东西好不好用的是“第二次跑”。我的做法是先用 SQLite 存一份状态表，把每次同步变成一个可以对比的差集任务。

CREATE TABLE IF NOT EXISTS files ( course_id TEXT NOT NULL, item_id TEXT NOT NULL, rel_path TEXT NOT NULL, size INTEGER, etag TEXT, sha1 TEXT, updated_at TEXT, PRIMARY KEY (course_id, item_id) ); CREATE INDEX IF NOT EXISTS idx_path ON files(rel_path);

字段不多，但每一项都有用：item_id是稳定主键（路径会变，id 不会），etag用来判断远端是否变过，sha1只在抽查时才算，避免每次全量哈希拖慢同步。同步流程就三步：拉远端清单 → 与表做差集得到新增 / 变更 / 消失三类 → 只对前两类下文件，第三类只标missing不删本地。

抽样校验是我后来加的习惯：每次同步随机抽 5 个文件，重新算一次 sha1 和表里比对。

import hashlib, random def sha1_of(path, chunk=1 << 20): h = hashlib.sha1() with open(path, "rb") as f: for block in iter(lambda: f.read(chunk), b""): h.update(block) return h.hexdigest() def spot_check(conn, root, n=5): rows = conn.execute("SELECT rel_path, sha1 FROM files WHERE sha1 IS NOT NULL").fetchall() bad = [] for rel, expect in random.sample(rows, min(n, len(rows))): p = root / rel if not p.exists() or sha1_of(p) != expect: bad.append(rel) return bad

抽样而不是全量，是因为全量校验在大目录上要跑十几分钟，实际没人会天天跑；抽 5 个只要几秒，能稳定暴露“下载过程被截断”“磁盘写满”“编码转换损坏”这几类问题。返回非空就说明本地副本已经不可信，下一次同步强制全量重下。

最后说个血泪经验：不要把“下载成功”等同于“文件完整”。我曾经因为一个 502 的重试逻辑写错，拿到了 300 多个 0 字节的 PDF，直到复习时打开才发现。从那以后，每个文件落盘后都校验一次Content-Length（有的话）和实际大小，不一致就删掉.part重来。慢一点没关系，后悔药没地方买。希望帮到你。

本文还有配套的精品资源，点击获取