从目录整理到任务完成，了解哎上课这类内容的下载框架如何组织。这份英文 Python 参考代码结合客户端的目录与资源组织方式，使用统一的教学模型演示选择、调度和文件管理；可在本地运行，示例数据和文件均为演示样本，真实平台接口、鉴权、解密与媒体下载实现不包含在内。

1. 目录与资源适配将准备好的课节、资料和附件整理成统一目录。示例按客户端中可确认的资源列表字段组织数据，其他目录使用教学用的类型化资源模型。
2. 完整分页与稳定选择短页继续翻页，重叠条目去重，冲突数据报错；先确定目录位置和序号，再过滤选择与权限，空章节不生成输出目录。
3. 可恢复的文件写入使用部分文件和修订记录恢复进度；源数据版本变化时重新开始，检查长度与 SHA-256，通过校验后原子替换正式文件。
4. 任务依赖与有界并发执行前检查重名路径、缺失依赖和循环依赖。只安排有限数量的任务，前置资源失败时阻止依赖任务发布不完整结果。
5. 进度记录、取消与重试由调度器统一保存任务状态；暂时读取失败时退避重试，磁盘错误直接记录。取消时保留部分文件，再次执行会重新验证实际文件。
6. 结果与离线演示准备的示例数据首次运行写入文件，再次运行校验并跳过已有文件；资料索引在资源完成后生成，实际输出展示在代码下方。

# Author: Xuewuzhi
# Source: https://xuewuzhi.cn/aishangke_downloader#source-analysis
# from xuewuzhi.cn
# Python 3.5+; offline teaching example; no real media transport.

PLATFORM = {'client_classes': ['Aishangke_Course'],
 'client_files': ['Mooc/Courses/Aishangke/Aishangke_Base.py',
                  'Mooc/Courses/Aishangke/Aishangke_Course.py',
                  'Mooc/Courses/Chaoge/Chaoge_Local.py'],
 'family': 'course',
 'resource_lists': ['file_list', 'video_list'],
 'slug': 'aishangke'}

"""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)
        self.path.parent.mkdir(parents=True, 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))
    task.path.parent.mkdir(parents=True, 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)
    # A partial without metadata is still checked against the final digest.
    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')
        # Disk errors and programming errors are terminal, not read retries.


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]
        first.path.parent.mkdir(parents=True, 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):
            # Reload state as a newly started process would; verify files again.
            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'])))
        # The temporary directory is removed after this offline demonstration.

# Normalize prepared catalog data; these are not platform endpoint schemas.
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():
    """Adapt discovered client list fields, or use a generic prepared row tree."""
    blobs = {}
    keys = PLATFORM['resource_lists'][:3]
    rows = [fixture(blobs, 'lesson-01', 'Introduction', 'video'),
            fixture(blobs, 'lesson-02', 'Workbook', 'document'),
            fixture(blobs, 'lesson-03', 'Summary', 'article')]
    pages = {1: {'items': rows[:1], 'has_more': True},
             2: {'items': rows, 'has_more': False}}
    prepared = collect_pages(lambda number: pages[number])
    chapter = {'id': 'chapter-01', 'title': 'Learning unit'}
    if keys:
        for index, key in enumerate(keys):
            row = dict(prepared[index], kind=LIST_KINDS[key])
            chapter[key] = [row]
    else:
        chapter['resources'] = prepared
    empty = {'id': 'chapter-02', 'title': 'Not released', 'resources': []}
    return normalize_tree(root_group([chapter, empty])), MemorySource(blobs)

if __name__ == '__main__':
    main()

阅读顺序：从 demo_course 观察目录输入，查看 build_plan 如何形成任务，再阅读 validate_plan、run_plan 与 transfer；Journal 记录运行状态，append_library_index 演示任务依赖。

files/{1}--Learning unit/(1.1)--Introduction.demo
course/{1}--Learning unit/[1.1]--Workbook.demo
library-index.demo
Run 1: saved=3, skipped=0, failed=0, blocked=0, cancelled=0
Run 2: saved=0, skipped=3, failed=0, blocked=0, cancelled=0
Checkpoint: 3 tracked tasks

作者：学无止 · 来源：xuewuzhi.cn