从目录整理到任务完成，了解小鹅通(PC端)这类内容的下载框架如何组织。这份英文 Python 参考代码结合客户端的目录与资源组织方式，使用统一的教学模型演示选择、调度和文件管理；可在本地运行，示例数据和文件均为演示样本，真实平台接口、鉴权、解密与媒体下载实现不包含在内。
- 分页与机构隔离根据 has_more、结束标记和总数继续翻页，短页仍可能有后续；按机构、资源 ID 与类型去重，重复页和不完整目录明确报错。
- 会员内容与目录编号展开会员和大专栏的子专栏，同时保留直属课节和附件；音视频共享序号，课件另计，多级目录继承完整前缀，选择后不重新编号。
- 可恢复的文件写入使用部分文件和修订记录恢复进度；源数据版本变化时重新开始，检查长度与 SHA-256，通过校验后原子替换正式文件。
- 任务依赖与有界并发执行前检查重名路径、缺失依赖和循环依赖。只安排有限数量的任务，前置资源失败时阻止依赖任务发布不完整结果。
- 进度记录、取消与重试由调度器统一保存任务状态；暂时读取失败时退避重试，磁盘错误直接记录。取消时保留部分文件，再次执行会重新验证实际文件。
- 结果与离线演示准备的示例数据首次运行写入文件，再次运行校验并跳过已有文件；资料索引在资源完成后生成，实际输出展示在代码下方。
xiaoetech_pipeline.py · 541 linesPython 3.5+ · Offline demo
# Author: Xuewuzhi
# Source: https://xuewuzhi.cn/xiaoetech_downloader#source-analysis
# from xuewuzhi.cn
# Python 3.5+; offline teaching example; no real media transport.
PLATFORM = {'client_classes': ['Xiaoetech_App'],
'client_files': ['Mooc/Courses/Xiaoetech/Xiaoetech_App.py',
'Mooc/Courses/Xiaoetech/Xiaoetech_Base.py'],
'family': 'membership',
'resource_lists': ['audio_list',
'clock_list',
'doc_list',
'file_list',
'practice_list',
'text_list',
'video_list'],
'slug': 'xiaoetech'}
"""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'])))
RESOURCE_KINDS = {1: 'article', 2: 'audio', 3: 'video', 4: 'live',
10: 'live', 12: 'live', 20: 'ebook', 34: 'practice', 51: 'document'}
CONTAINER_KINDS = {5: 'member', 6: 'column', 8: 'collection'}
def as_flag(value):
text = str(value).strip().lower()
return True if text in ('1', 'true') else False if text in ('0', 'false') else None
def resource_type(row):
return int(str(row.get('resource_type', -1)).strip())
def native_id(row):
return str(row.get('resource_id') or row.get('chapter_id') or '')
def read_pages(fetch_page, page_size=100, max_pages=100):
"""Prefer explicit pagination metadata; detect repeated pages and overlaps."""
rows, seen_rows, seen_pages = [], set(), set()
loaded = 0
for page_number in range(1, max_pages + 1):
body = fetch_page(page_number, page_size)
items = body.get('list')
if not isinstance(items, list):
raise CatalogError('Invalid page payload')
more = as_flag(body.get('has_more'))
flags = [as_flag(body.get(key)) for key in ('is_finish', 'is_finished', 'is_end', 'is_last')]
if not items:
if more is True or False in flags:
raise CatalogError('Empty page still declares more results')
return rows
signature = tuple((row.get('app_id', ''), native_id(row), resource_type(row))
for row in items)
if signature in seen_pages:
raise CatalogError('Repeated page; refusing an incomplete catalog')
seen_pages.add(signature)
for row, identity in zip(items, signature):
if not identity[1]:
raise CatalogError('Resource identity is missing')
if identity not in seen_rows:
seen_rows.add(identity)
rows.append(dict(row))
loaded += len(items)
if True in flags or more is False:
return rows
if more is True or False in flags:
continue
total = body.get('total', body.get('total_count'))
total = int(total) if total is not None else None
if total is not None and total > 0:
if loaded >= total:
return rows
elif len(items) < page_size:
return rows
raise CatalogError('Pagination limit reached')
def expand_entry(reader, entry, ancestors=()):
"""Members/collections contain both child columns AND direct resources."""
app_id, entry_id = entry['app_id'], native_id(entry)
identity = (app_id, entry_id)
if identity in ancestors or len(ancestors) >= 16:
raise CatalogError('Cyclic or excessively deep catalog')
trail = ancestors + (identity,)
def load(section):
def fetch(number, size):
body = dict(reader(app_id, entry_id, section, number, size))
if isinstance(body.get('list'), list):
body['list'] = [dict(row, app_id=row.get('app_id') or app_id)
for row in body['list']]
return body
return read_pages(fetch)
def chronological(rows):
return sorted(rows, key=lambda row: str(row.get('start_at') or ''))
rows = []
if CONTAINER_KINDS.get(resource_type(entry)) in ('member', 'collection'):
for child in chronological(load('columns')):
children = expand_entry(reader, child, trail)
rows.append(dict(child, resource_type=0, children=children))
rows.extend(chronological(load('resources')))
rows.extend(dict(row, attachment=True) for row in load('files'))
for row in rows:
if resource_type(row) == 0 and 'children' not in row:
row['children'] = expand_entry(reader, row, trail)
return rows
def normalize_catalog(entry, rows):
"""Preserve paths before selection; media and courseware have separate counters."""
chapters = []
def visit(items, index_path=(), folders=()):
media_index, document_index, attachment_index = 0, 0, 0
insertion_index = len(chapters)
lessons = []
for position, row in enumerate(items, 1):
if as_flag(row.get('can_access')) is False or as_flag(row.get('is_expired')) is True:
continue
kind_code = resource_type(row)
title = row.get('resource_title') or row.get('chapter_title') or row.get('file_name') or 'untitled'
if kind_code == 0:
child_path = index_path + (position,)
prefix = '.'.join(map(str, child_path))
visit(row['children'], child_path, folders + ('{' + prefix + '}--' + safe_name(title),))
continue
kind = RESOURCE_KINDS.get(kind_code)
if kind is None:
raise CatalogError('Unsupported resource type: {}'.format(kind_code))
output_root = 'files' if row.get('attachment') else 'course'
if row.get('attachment'):
attachment_index += 1
sequence, brackets = attachment_index, '()'
elif kind in ('video', 'live', 'audio'):
media_index += 1
sequence, brackets = media_index, '[]'
else:
document_index += 1
sequence, brackets = document_index, '##' if kind == 'article' else '()'
prefix = '.'.join(map(str, index_path + (sequence,)))
title = re.sub(r'\.(mp4|mp3|pdf|pptx?|docx?)$', '', title, flags=re.I)
title = re.sub(r'^(?:\[[\d.]+\]|\([\d.]+\)|#[\d.]+#)--', '', title)
name = brackets[0] + prefix + brackets[1] + '--' + safe_name(title)
occurrence = (row['app_id'], native_id(entry), index_path, output_root, kind, native_id(row))
selector_id = sha256(repr(occurrence).encode('utf-8')).hexdigest()
resource = dict(row, id=native_id(row), kind=kind, name=name,
selector_id=selector_id,
output_root=output_root)
lessons.append({'title': title, 'accessible': True, 'resources': [resource]})
if lessons:
chapters.insert(insertion_index, {'title': '/'.join(folders) or 'Direct resources',
'folders': folders, 'lessons': lessons})
visit(rows)
return {'id': native_id(entry), 'app_id': entry['app_id'], 'chapters': chapters}
def demo_catalog():
"""Fictional pages use real catalog field names, with local byte fixtures."""
app_id, blobs = 'demo-shop', {}
def resource(identifier, title, kind):
data = ('Offline sample: ' + identifier + '\n').encode('ascii') * 2
blobs[(app_id, identifier)] = data
return {'resource_id': identifier, 'resource_title': title, 'resource_type': kind,
'revision': 'demo-v1', 'size': len(data), 'sha256': sha256(data).hexdigest()}
video = resource('video-01', 'Introduction.mp4', '3')
ebook = resource('book-01', 'Workbook.pdf', 20)
audio = resource('audio-01', 'Discussion.mp3', 2)
chapter = {'chapter_id': 'chapter-02', 'chapter_title': 'Deep dive', 'resource_type': 0}
pages = {
('member-01', 'columns', 1): {'list': [
{'resource_id': 'column-01', 'resource_title': 'Series one', 'resource_type': 6},
{'resource_id': 'empty-01', 'resource_title': 'Not released', 'resource_type': 6},
], 'has_more': False},
('column-01', 'resources', 1): {'list': [video, ebook], 'has_more': True},
('column-01', 'resources', 2): {'list': [video, audio, chapter], 'has_more': False},
('chapter-02', 'resources', 1): {'list': [resource('video-02', 'Case study', 3)]},
('member-01', 'resources', 1): {'list': [resource('welcome-01', 'Welcome', 1)]},
('member-01', 'files', 1): {'list': [resource('notes-01', 'Handout.pdf', 51)]},
}
def reader(store, entry_id, section, page_number, page_size):
if store != app_id:
raise CatalogError('Unknown demo store')
return pages.get((entry_id, section, page_number), {'list': []})
entry = {'app_id': app_id, 'resource_id': 'member-01', 'resource_type': 5}
return entry, reader, MemorySource(blobs)
def demo_course():
entry, reader, source = demo_catalog()
rows = expand_entry(reader, entry)
return normalize_catalog(entry, rows), source
if __name__ == '__main__':
main()
阅读顺序：从 demo_course 观察目录输入，查看 build_plan 如何形成任务，再阅读 validate_plan、run_plan 与 transfer；Journal 记录运行状态，append_library_index 演示任务依赖。
示例运行结果
course/#1#--Welcome.demo
files/(1)--Handout.demo
course/{1}--Series one/[1.1]--Introduction.demo
course/{1}--Series one/(1.1)--Workbook.demo
course/{1}--Series one/[1.2]--Discussion.demo
course/{1}--Series one/{1.4}--Deep dive/[1.4.1]--Case study.demo
library-index.demo
Run 1: saved=7, skipped=0, failed=0, blocked=0, cancelled=0
Run 2: saved=0, skipped=7, failed=0, blocked=0, cancelled=0
Checkpoint: 7 tracked tasks
作者：学无止 · 来源：xuewuzhi.cn
