从目录整理到任务完成，了解钉盘文件这类内容的下载框架如何组织。这份英文 Python 参考代码结合客户端的目录与资源组织方式，使用统一的教学模型演示选择、调度和文件管理；可在本地运行，示例数据和文件均为演示样本，真实平台接口、鉴权、解密与媒体下载实现不包含在内。
- 空间和目录边界将空间、文件夹路径与文件标识共同保留；同名文件不跨目录合并。教学模型只使用准备好的文件样本，不连接钉钉。
- 原格式与完成校验文档、视频和压缩包分别保留文件类型，检查实际文件长度与摘要后再记为完成。示例后缀为 demo，不是实际下载文件。
- 可恢复的文件写入使用部分文件和修订记录恢复进度；源数据版本变化时重新开始，检查长度与 SHA-256，通过校验后原子替换正式文件。
- 任务依赖与有界并发执行前检查重名路径、缺失依赖和循环依赖。只安排有限数量的任务，前置资源失败时阻止依赖任务发布不完整结果。
- 进度记录、取消与重试由调度器统一保存任务状态；暂时读取失败时退避重试，磁盘错误直接记录。取消时保留部分文件，再次执行会重新验证实际文件。
- 结果与离线演示准备的示例数据首次运行写入文件，再次运行校验并跳过已有文件；资料索引在资源完成后生成，实际输出展示在代码下方。
dingpan_pipeline.py · 457 linesPython 3.5+ · Offline demo
# Author: Xuewuzhi
# Source: https://xuewuzhi.cn/dingpan_downloader#source-analysis
# Python 3.5+; offline teaching example; no real file transport.
PLATFORM = {'client_classes': ['Dingtalk_Dingpan_Course'],
'client_files': ['Mooc/Courses/Dingtalk/Dingtalk_Group_Course.py'],
'family': 'drive',
'resource_lists': [],
'slug': 'dingpan'}
"""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():
"""Prepared file rows, not real DingTalk API responses."""
blobs = {}
groups = [
{'id': 'space-A', 'title': 'Team space', 'children': [
{'id': 'folder-A', 'title': 'Study materials', 'resources': [
fixture(blobs, 'file-01', 'Handout.pdf', 'document', scope='space-A'),
fixture(blobs, 'file-02', 'Recording.mp4', 'video', scope='space-A')]},
{'id': 'folder-B', 'title': 'Exercises', 'resources': [
fixture(blobs, 'file-03', 'Practice.zip', 'document', scope='space-A')]}
]}
]
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}--Team space/{1.1}--Study materials/(1.1.1)--Handout.pdf.demo
course/{1}--Team space/{1.1}--Study materials/[1.1.1]--Recording.mp4.demo
course/{1}--Team space/{1.2}--Exercises/(1.2.1)--Practice.zip.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
