当前版本：V2.19.56 · 发布日期： · Windows 电脑版
学而思网校课程的登录、内容选择与下载操作说明。
客户端平台入口：学而思网校
课程链接域名：xueersi.com
课程、回放及资料是否可下载，以当前账号权限和平台实际开放的内容为准。
作者：学无止 · 来源：xuewuzhi.cn · 教程更新：
下方图片用于说明操作位置，旧版界面和快捷字母可能不同，请以上方步骤与当前下载器提示为准。
学无止是品牌简称，学无止科技是发布方，学无止下载器是面向 Windows 电脑的网课视频与课件下载工具。官方网站首页为 https://www.xuewuzhi.cn/。
先确认登录的是购买或报名课程的账号，并检查课程期次、章节范围和内容类型。尚未发布、未获授权或已下架的内容可能不会出现在可下载列表中。
课程结束不一定代表资源已失效。能否下载取决于账号是否仍有访问权限、平台是否保留资源，以及当前版本是否支持该内容。权限到期、资源下架或平台关闭访问时，不保证能够下载。
请使用有课程访问权限的账号登录。可下载内容以原平台授权及开放情况为准，资源下架或权限到期后可能无法下载。
先在原平台确认该账号能够观看或打开内容，再核对下载器版本、登录状态、课程链接是否完整及保存位置是否可写。保留具体报错与课程名称，通过页面下方的客服入口反馈；不要公开账号密码或登录凭证。
在下载器首页输入 M 并回车进入菜单，修改下载路径。开始任务前核对保存位置及剩余磁盘空间。
“播放列表.dpl”可用 PotPlayer 顺序播放已下载的视频。如果移动课程目录后列表失效，且目录内提供了“修复播放列表.bat”，可运行该脚本重新生成路径。
从目录整理到任务完成，了解学而思网校这类内容的下载框架如何组织。这份英文 Python 参考代码结合客户端的目录与资源组织方式，使用统一的教学模型演示选择、调度和文件管理；可在本地运行，示例数据和文件均为演示样本，真实平台接口、鉴权、解密与媒体下载实现不包含在内。
# Author: Xuewuzhi
# Source: https://xuewuzhi.cn/xueersi_downloader#source-analysis
# from xuewuzhi.cn
# Python 3.5+; offline teaching example; no real media transport.
PLATFORM = {'client_classes': ['Xueersi_Course'],
'client_files': ['Mooc/Courses/Xueersi/Xueersi_Base.py',
'Mooc/Courses/Xueersi/Xueersi_Course.py'],
'family': 'course',
'resource_lists': ['video_list'],
'slug': 'xueersi'}
"""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 tolerates concurrent parent creation on Python 3.5 as well.
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)
# 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]
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):
# 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 演示任务依赖。
course/{1}--Learning unit/[1.1]--Introduction.demo
library-index.demo
Run 1: saved=2, skipped=0, failed=0, blocked=0, cancelled=0
Run 2: saved=0, skipped=2, failed=0, blocked=0, cancelled=0
Checkpoint: 2 tracked tasks
作者：学无止来源：xuewuzhi.cn
