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

1. 同一课节的多种资源将讲师画面、音频和课件时间信息放回同一课节，保留各自的资源类型与编号，为后续任务安排建立一致的目录。
2. 时间线整理示例按毫秒时间排序课件事件、去除重复事件并确定显示区间。这里展示准备阶段的时间线模型，不包含真实平台的渲染、解密或媒体合成。
3. 可恢复的文件写入使用部分文件和修订记录恢复进度；源数据版本变化时重新开始，检查长度与 SHA-256，通过校验后原子替换正式文件。
4. 任务依赖与有界并发执行前检查重名路径、缺失依赖和循环依赖。只安排有限数量的任务，前置资源失败时阻止依赖任务发布不完整结果。
5. 进度记录、取消与重试由调度器统一保存任务状态；暂时读取失败时退避重试，磁盘错误直接记录。取消时保留部分文件，再次执行会重新验证实际文件。
6. 结果与离线演示准备的示例数据首次运行写入文件，再次运行校验并跳过已有文件；资料索引在资源完成后生成，实际输出展示在代码下方。

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

PLATFORM = {'client_classes': ['Yuyou_Course'],
 'client_files': ['Mooc/Courses/Plaso/Plaso_Base.py',
                  'Mooc/Courses/Plaso/Plaso_Course.py',
                  'Mooc/Courses/Yuyou/Yuyou_Base.py',
                  'Mooc/Courses/Yuyou/Yuyou_Course.py'],
 'family': 'board',
 'resource_lists': ['file_list'],
 'slug': 'yuyou'}

"""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 timeline_windows(events, duration_ms):
    """Order prepared slide events on one clock without rendering any media."""
    if type(duration_ms) is not int or duration_ms <= 0:
        raise CatalogError('Invalid timeline duration')
    ordered, seen = [], set()
    for event in sorted(events, key=lambda item: item['at_ms']):
        at = event['at_ms']
        if type(at) is not int or at < 0:
            raise CatalogError('Invalid event timestamp')
        identity = (event['id'], at)
        if at >= duration_ms or identity in seen:
            continue
        seen.add(identity)
        ordered.append(dict(event))
    for index, event in enumerate(ordered):
        event['until_ms'] = ordered[index + 1]['at_ms'] if index + 1 < len(ordered) else duration_ms
    return ordered


def demo_course():
    """Keep lecture tracks and slide timing together before downstream rendering."""
    blobs = {}
    windows = timeline_windows([{'id': 'slide-2', 'at_ms': 15000},
                                {'id': 'slide-1', 'at_ms': 0},
                                {'id': 'slide-1', 'at_ms': 0}], 60000)
    timeline = json.dumps(windows, sort_keys=True).encode('ascii')
    lesson = {'id': 'session-01', 'title': 'Recorded lesson', 'resources': [
        fixture(blobs, 'teacher-track', 'Teacher track', 'video'),
        fixture(blobs, 'audio-track', 'Audio track', 'audio'),
        fixture(blobs, 'slide-timing', 'Slide timeline', 'document', payload=timeline),
    ]}
    # Timeline normalization is demonstrated; platform rendering is omitted.
    return normalize_tree(root_group([lesson])), MemorySource(blobs)

if __name__ == '__main__':
    main()

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

course/{1}--Recorded lesson/[1.1]--Teacher track.demo
course/{1}--Recorded lesson/[1.2]--Audio track.demo
course/{1}--Recorded lesson/(1.1)--Slide timeline.demo
library-index.demo
Run 1: saved=4, skipped=0, failed=0, blocked=0, cancelled=0
Run 2: saved=0, skipped=4, failed=0, blocked=0, cancelled=0
Checkpoint: 4 tracked tasks

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