从目录整理到任务完成，了解中国大学这类内容的下载框架如何组织。这份英文 Python 参考代码结合客户端的目录与资源组织方式，使用统一的教学模型演示选择、调度和文件管理；可在本地运行，示例数据和文件均为演示样本，真实平台接口、鉴权、解密与媒体下载实现不包含在内。
- 批量入口去重将多个课程入口整理为批次，重复课程只进入一次；每个课程保留自己的章节层级、输出目录和选择标识。
- 课程之间互不混淆不同课程可能使用相同课节 ID，示例用平台和课程范围共同标识资源，避免把两个课程的第一节课误判为同一文件。
- 可恢复的文件写入使用部分文件和修订记录恢复进度；源数据版本变化时重新开始，检查长度与 SHA-256，通过校验后原子替换正式文件。
- 任务依赖与有界并发执行前检查重名路径、缺失依赖和循环依赖。只安排有限数量的任务，前置资源失败时阻止依赖任务发布不完整结果。
- 进度记录、取消与重试由调度器统一保存任务状态；暂时读取失败时退避重试，磁盘错误直接记录。取消时保留部分文件，再次执行会重新验证实际文件。
- 结果与离线演示准备的示例数据首次运行写入文件，再次运行校验并跳过已有文件；资料索引在资源完成后生成，实际输出展示在代码下方。
icourse163_pipeline.py · 464 linesPython 3.5+ · Offline demo
# Author: Xuewuzhi
# Source: https://xuewuzhi.cn/icourse163_downloader#source-analysis
# from xuewuzhi.cn
# Python 3.5+; offline teaching example; no real media transport.
PLATFORM = {'client_classes': ['Icourse163_Batch'],
'client_files': ['Mooc/Courses/Course_Batch.py',
'Mooc/Courses/Mooc163/Icourse163/Icourse163_Base.py',
'Mooc/Courses/Mooc163/Icourse163/Icourse163_Batch.py',
'Mooc/Courses/Mooc163/Mooc163_Base.py'],
'family': 'batch',
'resource_lists': [],
'slug': 'icourse163'}
"""Offline task framework; adapters provide immutable, already-prepared bytes.
This module contains no network transport, platform authorization or decryption.
The data contract below is a teaching model, not a platform API response.
"""
from collections import Counter, deque, namedtuple
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from hashlib import sha256
from pathlib import Path
from tempfile import TemporaryDirectory
import json
import os
import re
import threading
import time
Task = namedtuple('Task', 'key source_id path size digest after')
Task.__new__.__defaults__ = ((),)
Result = namedtuple('Result', 'key state detail')
SUCCESS = frozenset(('saved', 'skipped'))
class CatalogError(ValueError):
"""A catalog is incomplete, cyclic or internally inconsistent."""
class RetryableReadError(IOError):
"""A temporary interruption; the next attempt may resume saved bytes."""
class IntegrityError(ValueError):
"""The source does not match its declared immutable revision."""
class Cancelled(Exception):
"""Cooperative cancellation preserves unfinished work for a later run."""
def stable_key(*parts):
data = json.dumps(parts, ensure_ascii=True, separators=(',', ':'))
return sha256(data.encode('ascii')).hexdigest()
def safe_name(value):
name = re.sub(r'[<>:"/\\|?*\x00-\x1f]+', '_', str(value))
name = name.strip(' .')[:64].rstrip(' .')
if not name or name in ('.', '..'):
return 'untitled'
if name.split('.')[0].upper() in ('CON', 'PRN', 'AUX', 'NUL') or re.match(r'^(COM|LPT)[1-9](\.|$)', name, re.I):
name = '_' + name
return name
def atomic_json(path, data):
"""Publish state only after the replacement file has reached the disk."""
temporary = path.with_name(path.name + '.tmp')
with temporary.open('w', encoding='utf-8') as stream:
json.dump(data, stream, ensure_ascii=True, sort_keys=True, indent=2)
stream.flush()
os.fsync(stream.fileno())
os.replace(str(temporary), str(path))
class Journal:
"""One coordinator owns this journal; task workers never write it.
Completion records are observations, not proof that an output still exists.
Every run revalidates files, even when the journal says they were saved.
"""
def __init__(self, path):
self.path = Path(path)
os.makedirs(str(self.path.parent), exist_ok=True)
self.data = {'schema': 1, 'tasks': {}}
if self.path.exists():
with self.path.open(encoding='utf-8') as stream:
data = json.load(stream)
if not isinstance(data, dict) or data.get('schema') != 1 or not isinstance(data.get('tasks'), dict):
raise ValueError('Invalid checkpoint journal')
self.data = data
def record(self, task, state, detail=''):
previous = self.data['tasks'].get(task.key, {})
self.data['tasks'][task.key] = {
'state': state, 'detail': detail, 'size': task.size,
'sha256': task.digest,
'runs': previous.get('runs', 0) + (1 if state == 'running' else 0),
}
atomic_json(self.path, self.data)
def build_plan(course, destination, selected_ids=None):
"""Assign paths before filtering; selection never changes lesson numbers."""
root = Path(os.path.realpath(str(destination)))
tasks, seen = [], {}
for chapter in course['chapters']:
for lesson in chapter['lessons']:
if not lesson['accessible']:
continue
for resource in lesson['resources']:
if selected_ids is not None and resource['selector_id'] not in selected_ids:
continue
identity = (course['app_id'], course['id'], resource['selector_id'], resource['revision'])
size, digest = resource['size'], resource['sha256']
if type(size) is not int or size < 0:
raise ValueError('Invalid resource size')
if not isinstance(digest, str) or not re.fullmatch(r'[0-9a-f]{64}', digest):
raise ValueError('Invalid resource checksum')
if identity in seen:
if seen[identity] != (size, digest):
raise ValueError('Conflicting resource metadata')
continue
seen[identity] = (size, digest)
folder = root / resource['output_root']
for name in chapter['folders']:
folder = folder / name
target = Path(os.path.realpath(str(folder / (resource['name'] + '.demo'))))
if root not in target.parents:
raise ValueError('Destination escapes the output directory')
tasks.append(Task(stable_key(*identity), (resource['app_id'], resource['id']), target, size, digest))
return tasks
def is_complete(path, task):
if not path.is_file() or path.stat().st_size != task.size:
return False
checksum = sha256()
with path.open('rb') as stream:
for chunk in iter(lambda: stream.read(64 * 1024), b''):
checksum.update(chunk)
return checksum.hexdigest() == task.digest
def remove_partial(path):
if path.exists():
path.unlink()
def check_cancel(stop):
if stop is not None and stop.is_set():
raise Cancelled('Run was cancelled')
def transfer(task, source, attempts=3, pause=time.sleep, stop=None):
"""Resume a revision, verify the whole file, then replace the destination.
Source.chunks(task, offset) must begin at exactly offset in the manifest's
immutable resource revision. A real transport would need to validate range
and revision responses. This example deliberately has no such transport.
"""
if attempts < 1:
raise ValueError('At least one attempt is required')
check_cancel(stop)
if is_complete(task.path, task):
return Result(task.key, 'skipped', str(task.path))
os.makedirs(str(task.path.parent), exist_ok=True)
partial = task.path.with_name(task.path.name + '.part')
metadata = partial.with_name(partial.name + '.json')
identity = {'key': task.key, 'size': task.size, 'sha256': task.digest}
if metadata.exists():
try:
with metadata.open(encoding='utf-8') as stream:
previous = json.load(stream)
except ValueError:
previous = None
if previous != identity:
remove_partial(partial)
atomic_json(metadata, identity)
for attempt in range(attempts):
try:
check_cancel(stop)
offset = partial.stat().st_size if partial.exists() else 0
if offset > task.size:
remove_partial(partial)
offset = 0
if offset < task.size:
with partial.open('ab') as output:
for chunk in source.chunks(task, offset):
check_cancel(stop)
if not chunk:
continue
if offset + len(chunk) > task.size:
raise IntegrityError('Source exceeded the expected size')
output.write(chunk)
offset += len(chunk)
output.flush()
os.fsync(output.fileno())
if offset != task.size:
raise RetryableReadError('Source ended before the expected size')
elif not partial.exists():
partial.touch()
check_cancel(stop)
if not is_complete(partial, task):
raise IntegrityError('Checksum mismatch; restart from byte zero')
partial.replace(task.path)
remove_partial(metadata)
return Result(task.key, 'saved', str(task.path))
except (RetryableReadError, IntegrityError) as error:
if isinstance(error, IntegrityError):
remove_partial(partial)
if attempt + 1 == attempts:
raise
delay = min(0.25 * (2 ** attempt), 2.0)
if stop is None:
pause(delay)
elif stop.wait(delay):
raise Cancelled('Cancelled during retry delay')
def validate_plan(tasks):
"""Reject collisions, missing prerequisites and cycles before writing files."""
by_key = {task.key: task for task in tasks}
paths = [os.path.realpath(str(task.path)).casefold() for task in tasks]
if len(by_key) != len(tasks) or len(set(paths)) != len(paths):
raise ValueError('Duplicate task identity or destination')
children = {key: [] for key in by_key}
degree = {}
for task in tasks:
if len(set(task.after)) != len(task.after):
raise ValueError('Duplicate prerequisite')
degree[task.key] = len(task.after)
for dependency in task.after:
if dependency not in by_key:
raise ValueError('Missing prerequisite')
children[dependency].append(task.key)
queue = deque(key for key in by_key if degree[key] == 0)
count = 0
while queue:
key = queue.popleft()
count += 1
for child in children[key]:
degree[child] -= 1
if degree[child] == 0:
queue.append(child)
if count != len(tasks):
raise ValueError('Cyclic task dependencies')
return by_key, children
def run_plan(tasks, source, workers=3, journal=None, stop=None):
"""Bound queued work, isolate failures and run dependents only after success."""
by_key, children = validate_plan(tasks)
degree = {task.key: len(task.after) for task in tasks}
ready = deque(task.key for task in tasks if not task.after)
results, pending = {}, {}
limit = max(1, min(workers, 4))
def finish(task, result):
results[task.key] = result
if journal is not None:
journal.record(task, result.state, result.detail)
for child in children[task.key]:
degree[child] -= 1
if degree[child] == 0:
ready.append(child)
with ThreadPoolExecutor(max_workers=limit) as pool:
while ready or pending:
while ready and len(pending) < limit:
task = by_key[ready.popleft()]
if stop is not None and stop.is_set():
finish(task, Result(task.key, 'cancelled', 'Run was cancelled'))
elif any(results[key].state not in SUCCESS for key in task.after):
finish(task, Result(task.key, 'blocked', 'A prerequisite did not finish'))
else:
if journal is not None:
journal.record(task, 'running')
future = pool.submit(transfer, task, source, stop=stop)
pending[future] = task
if pending:
done, _ = wait(pending, return_when=FIRST_COMPLETED)
for future in done:
task = pending.pop(future)
try:
result = future.result()
except Cancelled:
result = Result(task.key, 'cancelled', 'Run was cancelled')
except Exception as error:
result = Result(task.key, 'failed', type(error).__name__)
finish(task, result)
return [results[task.key] for task in tasks]
class MemorySource:
"""Only local fixture bytes; never resolves URLs or reads platform sessions."""
def __init__(self, blobs, chunk_size=8):
if chunk_size < 1:
raise ValueError('Chunk size must be positive')
self.blobs = blobs
self.chunk_size = chunk_size
def chunks(self, task, offset):
data = self.blobs[task.source_id]
for start in range(offset, len(data), self.chunk_size):
yield data[start:start + self.chunk_size]
def append_library_index(tasks, source, destination):
"""Publish a library index only after every selected resource has succeeded."""
root = Path(destination)
records = [{'path': task.path.relative_to(root).as_posix(), 'sha256': task.digest}
for task in tasks]
payload = json.dumps(records, sort_keys=True, indent=2).encode('ascii')
digest = sha256(payload).hexdigest()
source_id = ('demo-library', digest)
source.blobs[source_id] = payload
index = Task(stable_key('library-index', digest), source_id, root / 'library-index.demo',
len(payload), digest, tuple(task.key for task in tasks))
return tasks + [index]
def main():
course, source = demo_course()
with TemporaryDirectory(prefix='xuewuzhi-demo-') as destination:
tasks = build_plan(course, destination)
tasks = append_library_index(tasks, source, destination)
for task in tasks:
print(task.path.relative_to(destination).as_posix())
first = tasks[0]
os.makedirs(str(first.path.parent), exist_ok=True)
first.path.with_name(first.path.name + '.part').write_bytes(source.blobs[first.source_id][:7])
for run_number in (1, 2):
journal = Journal(Path(destination) / 'checkpoint.json')
report = run_plan(tasks, source, journal=journal)
counts = Counter(item.state for item in report)
print('Run {}: saved={}, skipped={}, failed={}, blocked={}, cancelled={}'.format(
run_number, counts['saved'], counts['skipped'], counts['failed'],
counts['blocked'], counts['cancelled']))
print('Checkpoint: {} tracked tasks'.format(len(journal.data['tasks'])))
LIST_KINDS = {'video_list': 'video', 'audio_list': 'audio', 'pdf_list': 'document',
'ppt_list': 'document', 'doc_list': 'document', 'file_list': 'document',
'attach_list': 'document', 'html_list': 'article', 'text_list': 'article',
'sub_list': 'subtitle', 'practice_list': 'practice', 'clock_list': 'practice'}
def collect_pages(fetch, limit=100):
"""Finish explicit pagination; a short page may still have a successor."""
result, identities, signatures = [], {}, set()
for number in range(1, limit + 1):
page = fetch(number)
rows = page.get('items')
if not isinstance(rows, list) or type(page.get('has_more')) is not bool:
raise CatalogError('Page requires items and an explicit continuation flag')
signature = tuple((row['scope'], row['id']) for row in rows)
if not rows and page['has_more']:
raise CatalogError('Empty page declares more results')
if rows and signature in signatures:
raise CatalogError('Repeated page')
signatures.add(signature)
for row in rows:
identity = (row['scope'], row['id'])
if identity in identities:
if identities[identity] != row:
raise CatalogError('Conflicting metadata for one catalog identity')
else:
identities[identity] = row
result.append(dict(row))
if not page['has_more']:
return result
raise CatalogError('Pagination limit reached')
def fixture(blobs, identifier, title, kind, scope=None, payload=None):
scope = scope or PLATFORM['slug']
data = payload if payload is not None else ('Offline sample: ' + scope + '/' + identifier + '\n').encode('ascii') * 2
blobs[(scope, identifier)] = data
return {'id': identifier, 'scope': scope, 'title': title, 'kind': kind,
'revision': 'demo-v1', 'size': len(data), 'sha256': sha256(data).hexdigest()}
def normalize_tree(tree):
"""Support resource-list dictionaries and typed rows through one model.
Group position belongs to the complete catalog. Filtering inaccessible rows
or empty groups must not renumber the remaining chapters or lesson names.
"""
chapters = []
def walk(group, indexes=(), folders=(), trail=()):
identity = (group.get('scope', PLATFORM['slug']), group['id'])
if identity in trail or len(trail) >= 16:
raise CatalogError('Cyclic or excessively deep catalog')
resources = list(group.get('resources', []))
for key, kind in sorted(LIST_KINDS.items()):
resources.extend(dict(row, kind=kind, attachment=key in ('file_list', 'attach_list'))
for row in group.get(key, []))
lessons, seen = [], {}
counters = Counter()
for row in resources:
kind = row['kind']
if kind not in set(LIST_KINDS.values()) | {'live'}:
raise CatalogError('Unsupported normalized resource kind')
category = 'files' if row.get('attachment') else 'course'
counter = 'attachment' if category == 'files' else 'media' if kind in ('video', 'audio', 'live') else 'document'
resource_id = (row['scope'], kind, row['id'], category)
if resource_id in seen:
if seen[resource_id] != row:
raise CatalogError('Conflicting catalog occurrence')
continue
seen[resource_id] = row
counters[counter] += 1
if row.get('accessible') is False:
continue
sequence = '.'.join(map(str, indexes + (counters[counter],)))
brackets = '[]' if counter == 'media' else '##' if kind == 'article' else '()'
name = brackets[0] + sequence + brackets[1] + '--' + safe_name(row['title'])
selected = stable_key(tree['id'], identity, indexes, resource_id)
resource = dict(row, app_id=row['scope'], selector_id=selected,
output_root=category, name=name)
lessons.append({'title': row['title'], 'accessible': True, 'resources': [resource]})
if lessons:
chapters.append({'title': group['title'], 'folders': folders, 'lessons': lessons})
for position, child in enumerate(group.get('children', []), 1):
child_indexes = indexes + (position,)
prefix = '.'.join(map(str, child_indexes))
folder = '{' + prefix + '}--' + safe_name(child['title'])
walk(child, child_indexes, folders + (folder,), trail + (identity,))
walk(tree)
return {'id': tree['id'], 'app_id': PLATFORM['slug'], 'chapters': chapters}
def root_group(children):
return {'id': PLATFORM['slug'] + '-selection', 'title': 'Selected content', 'children': children}
def demo_course():
"""Different courses may reuse a lesson ID; scope every resource by course."""
blobs, groups, seen = {}, [], set()
entries = [('course-a', 'Course A'), ('course-b', 'Course B'), ('course-a', 'Course A')]
for course_id, title in entries:
identity = (PLATFORM['slug'], course_id)
if identity in seen:
continue
seen.add(identity)
scope = PLATFORM['slug'] + '/' + course_id
chapter = {'id': course_id + '-chapter', 'title': 'Chapter one', 'resources': [
fixture(blobs, 'lesson-01', 'Lecture', 'video', scope),
fixture(blobs, 'handout-01', 'Handout', 'document', scope),
]}
groups.append({'id': course_id, 'scope': scope, 'title': title, 'children': [chapter]})
return normalize_tree(root_group(groups)), MemorySource(blobs)
if __name__ == '__main__':
main()
阅读顺序：从 demo_course 观察目录输入，查看 build_plan 如何形成任务，再阅读 validate_plan、run_plan 与 transfer；Journal 记录运行状态，append_library_index 演示任务依赖。
示例运行结果
course/{1}--Course A/{1.1}--Chapter one/[1.1.1]--Lecture.demo
course/{1}--Course A/{1.1}--Chapter one/(1.1.1)--Handout.demo
course/{2}--Course B/{2.1}--Chapter one/[2.1.1]--Lecture.demo
course/{2}--Course B/{2.1}--Chapter one/(2.1.1)--Handout.demo
library-index.demo
Run 1: saved=5, skipped=0, failed=0, blocked=0, cancelled=0
Run 2: saved=0, skipped=5, failed=0, blocked=0, cancelled=0
Checkpoint: 5 tracked tasks
作者：学无止 · 来源：xuewuzhi.cn
