好友
阅读权限10
听众
最后登录1970-1-1
|
本帖最后由 threettiger 于 2026-3-26 08:02 编辑
在参考大佬 某鹅通最新方法 的基础上写的代码，代码分两部分，一部分是抓取课程链接，另一部分是批量下载视频文件。
获取视频下载链接地址代码，具体使用时请替换下你自己的网页版课程链接、Cookie、课程id,以及相关请求参数。
课程使用的某鹅通网页版，请求网页也需要替换你们课程的网页版地址。
我是为了下载小鹅通自己的学习视频，才搞的，虽然发了全部源码，发现很多朋友，还是不会用。教大家个简单的方法，下载个traecn，然后把代码复制进去，让AI帮你去调试，trae免费版也很强，2026了，不需要手搓代码了，你甚至可以把本帖子的链接直接丢给AI,让它去帮你阅读，然后帮你去写代码，最后成功跑通。
这里面小白，可能还是需要点网页调试的能力，这些都是基础的能力，现在AI大行其道，很多小朋友可能都不熟悉，我简单讲一下：
1.打开你的课程电脑版网址，记得要登录（至少你有看视频的权限），到课程页面，然后按F12,调出开发者工具，找到网络，
2.输入比如我们获取视频详细信息的接口是：https://appvvutermf4498.h5.xiaoecloud.com/xe.course.business.video.detail_info.get/2.0.0
你在筛选器里面输入筛选下内容detail_info，筛选一下，接口很多，开始的时候需要自己去分析的。
看到接口后，就可以看到请求头信息，负载信息，响应信息了，你可以右键复制这些丢给AI，让AI帮你去替换网页版课程链接、Cookie、课程id,以及相关请求参数。
这个不是小白的无脑使用工具，本来就是我自己下载视频用的，借鉴的也是论坛大佬的破解方法和思路， 鼓励大家动手试试。网络爬虫，API调试，web逆向，这些都是最最基础的。我看评论区还是有些人不会，所以补充啰嗦了下。大佬跳过。
[Python] 纯文本查看 复制代码 import requests, json, re, os, base64, ssl, urllib3, concurrent.futures
urllib3.disable_warnings()
# ssl._create_default_https_context = ssl._create_unverified_context
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36',
'Accept-Encoding': 'gzip, deflate, br',
'Cookie': 'xenbyfpfUnhLsdkZbX=0; ko_token=ffcfaf82d99e32af1695286910e7510c; newuserdays=90; olduserdays=180; regtime=1689246133; shop_version_type=4; sensorsdata2015jssdkcross=%7B%22%24device_id%22%3A%2219d08adcd9f8de-04410a3b780ae48-5704343e-400760-19d08adcda03e2%22%7D; sajssdk_2015_new_user_appvvutermf4498_h5_xiaoecloud_com=1; sa_jssdk_2015_appvvutermf4498_h5_xiaoecloud_com=%7B%22distinct_id%22%3A%22u_64afd9b51ff0c_jEaSdgUTpb%22%2C%22first_id%22%3A%2219d08adcd9f8de-04410a3b780ae48-5704343e-400760-19d08adcda03e2%22%2C%22props%22%3A%7B%7D%7D; logintime=1773967151',
'Referer': 'https://appvvutermf4498.h5.xiaoecloud.com/',
'Origin': 'https://appvvutermf4498.h5.xiaoecloud.com',
'Host': 'appvvutermf4498.h5.xiaoecloud.com',
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json, text/plain, */*',
'sec-ch-ua': '"Chromium";v="146", "Not-A.Brand";v="24", "Microsoft Edge";v="146"',
'sec-ch-ua-mobile': '?1',
'sec-ch-ua-platform': '"iOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'req-uuid': '20260320083913000905730',
'retry': '1',
}
# 解密小鹅通url播放地址
s = "W$siZGVmaW5pdGlvbl9uYW@lIjoiXHU5YWQ%XHU#ZTA@IiwiZGVmaW5pdGlvbl9wIjoiNzIwUCIsInVybCI6Imh0dHBzOlwvXC9jLXZvZC@ody@rLnhpYW9la#5vdy5jb#@cL#Fzc#V0XC9jM#Q5YzBhZDM#ZWQ5OGZiZTQzMGMwYTQ%MGYwZjdjNlwvYzkxNzBjN#Q%NjkyNjBhMzI0OGYzZWQ0ZGUwNTQxZWUubTN@OD9zaWduPTBmOTdjOTFhMzM5NDIzZWMwMDM%NTc0N#E@MmJmYjYyJnQ9NjVhNTdlZDgmdXM9ZkxLdGN%VEFtUyIsImlzX$N@cHBvcnQiOmZhbHNlLCJleHQiOnsiaG9zdCI6Imh0dHBzOlwvXC9jLXZvZC@ody@rLnhpYW9la#5vdy5jb#0iLCJwYXRoIjoiYXNzZXRcL#MzZDljMGFkMzZlZDk%ZmJlNDMwYzBhNDgwZjBmN#M#IiwicGFyYW0iOiJzaWduPTBmOTdjOTFhMzM5NDIzZWMwMDM%NTc0N#E@MmJmYjYyJnQ9NjVhNTdlZDgmdXM9ZkxLdGN%VEFtUyJ9fV0=__ba"
# def decode_url(s):
# s = s.replace("@", "1").replace("#", "2").replace("$", "3").replace("%", "4").replace("__ba", "")
# json_str = base64.b64decode(s.encode("utf-8")).decode("utf-8")[1:-1]
# jjson = json.loads(json_str)
# video_url = jjson.get("url")
# return video_url
def decode_url(s):
s = s.replace("@", "1").replace("#", "2").replace("$", "3").replace("%", "4").replace("__ba", "")
try:
# 先解码base64
decoded_bytes = base64.b64decode(s.encode("utf-8"))
json_str = decoded_bytes.decode("utf-8")
# 移除可能存在的前后引号
if json_str.startswith('"') and json_str.endswith('"'):
json_str = json_str[1:-1]
# 解析JSON
jjson = json.loads(json_str)
if isinstance(jjson, list):
# 如果是数组，返回第一个视频URL（通常是高清）
if jjson and isinstance(jjson[0], dict):
video_url = jjson[0].get("url")
return video_url
else:
print("视频URL数组格式不正确")
elif isinstance(jjson, dict):
# 如果是单个字典
video_url = jjson.get("url")
return video_url
else:
print("无法确定数据结构")
except Exception as e:
print(f"解码URL时出错：{e}")
return None
def get_video_url(resource_id):
global headers
# 使用浏览器中的正确API端点
url = "https://appvvutermf4498.h5.xiaoecloud.com/xe.course.business.video.detail_info.get/2.0.0"
data = {
'bizData[resource_id]': resource_id,
'bizData[product_id]': "p_5ef1f421a523f_xHcyCVLp",
'bizData[opr_sys]': "MacIntel" # 使用浏览器中的操作系统
}
# 创建专用的请求头，确保包含所有必要信息
video_headers = headers.copy()
video_headers['Referer'] = f"https://appvvutermf4498.h5.xiaoecloud.com/p/course/video/{resource_id}?product_id=p_5ef1f421a523f_xHcyCVLp"
try:
# 添加超时设置
response = requests.post(url, headers=video_headers, data=data, verify=False, timeout=30)
if not response.text.strip():
print(f"警告: 获取视频URL时收到空响应，resource_id: {resource_id}")
return None
r = json.loads(response.text)
# 只打印状态码和必要信息，减少内存使用
print(f"视频详情响应状态: {r.get('code')}")
if r.get('code') == 0 and 'data' in r:
video_urls = decode_url(r['data']['video_urls'])
return video_urls
else:
print(f"视频详情请求失败: {r.get('msg')}")
return None
except requests.exceptions.Timeout:
print(f"请求超时，resource_id: {resource_id}")
return None
except requests.exceptions.RequestException as e:
print(f"网络请求错误: {e}, resource_id: {resource_id}")
return None
except (json.JSONDecodeError, KeyError) as e:
print(f"解析视频URL时出错: {e}, resource_id: {resource_id}")
return None
except Exception as e:
print(f"获取视频URL时发生未知错误: {e}, resource_id: {resource_id}")
return None
# 测试解码功能
url = decode_url(s)
print(f"解码测试结果: {url}")
# 获取课程列表功能（分页获取全部视频）
url = "https://appvvutermf4498.h5.xiaoecloud.com/xe.course.business.column.items.get/2.0.0"
list_headers = headers.copy()
list_headers['Referer'] = 'https://appvvutermf4498.h5.xiaoecloud.com/p/course/column/p_5ef1f421a523f_xHcyCVLp'
# 先获取课程总数量和总页数
page_size = 20 # 每页数量
page_index = 1
total_count = 0
all_video_courses = []
# 第一次请求获取总数量
first_data = {
'bizData[column_id]': 'p_5ef1f421a523f_xHcyCVLp',
'bizData[page_index]': str(page_index),
'bizData[page_size]': str(page_size),
'bizData[sort]': 'desc'
}
first_response = requests.post(url, headers=list_headers, data=first_data, verify=False)
if first_response.status_code == 200 and first_response.text.strip():
try:
first_r = json.loads(first_response.text)
if first_r.get('code') == 0 and 'data' in first_r:
data_content = first_r['data']
if isinstance(data_content, dict):
# 获取总数量和总页数
if 'total' in data_content:
total_count = data_content['total']
print(f"课程总数量: {total_count}")
elif 'total_num' in data_content:
total_count = data_content['total_num']
print(f"课程总数量: {total_count}")
else:
print("未找到总数量字段，将使用列表长度作为总数量")
if 'list' in data_content:
total_count = len(data_content['list'])
print(f"假设总数量: {total_count}")
if total_count > 0:
# 计算总页数
import math
total_pages = math.ceil(total_count / page_size)
print(f"总页数: {total_pages}")
# 获取所有页的视频课程信息
for page in range(1, total_pages + 1):
print(f"\n正在获取第 {page}/{total_pages} 页...")
page_data = {
'bizData[column_id]': 'p_5ef1f421a523f_xHcyCVLp',
'bizData[page_index]': str(page),
'bizData[page_size]': str(page_size),
'bizData[sort]': 'desc'
}
page_response = requests.post(url, headers=list_headers, data=page_data, verify=False, timeout=30)
if page_response.status_code == 200 and page_response.text.strip():
try:
page_r = json.loads(page_response.text)
if page_r.get('code') == 0 and 'data' in page_r and 'list' in page_r['data']:
current_list = page_r['data']['list']
print(f"第 {page} 页获取到 {len(current_list)} 个课程")
for i in current_list:
if i.get('resource_type') == 3:
# 收集视频课程信息，稍后使用多线程获取URL
all_video_courses.append({
'title': i.get('resource_title', '未知'),
'resource_id': i['resource_id']
})
else:
print(f"第 {page} 页请求失败: {page_r.get('msg')}")
except json.JSONDecodeError as e:
print(f"第 {page} 页JSON解析错误: {e}")
else:
print(f"第 {page} 页请求失败，状态码: {page_response.status_code}")
# 使用多线程获取视频URL
if all_video_courses:
print(f"\n开始使用多线程获取 {len(all_video_courses)} 个视频URL...")
all_video_info = []
# 定义一个线程安全的回调函数
def process_video(course):
title = course['title']
resource_id = course['resource_id']
print(f"正在处理视频: {title}")
video_url = get_video_url(resource_id)
if video_url:
return {
'chapterName': title,
'filePath': video_url
}
return None
# 创建线程池，最大线程数设置为10
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
# 提交所有任务
future_to_course = {
executor.submit(process_video, course): course
for course in all_video_courses
}
# 收集结果
for future in concurrent.futures.as_completed(future_to_course):
result = future.result()
if result:
all_video_info.append(result)
# 保存所有视频信息
if all_video_info:
file_name = 'D:/code/python/test2/xiaoetong/小鹅通高项课程2026.json'
with open(file_name, 'w', encoding='utf-8') as f:
json.dump(all_video_info, f, ensure_ascii=False, indent=4)
print(f"\n已保存{len(all_video_info)}个视频信息到{file_name}")
else:
print("\n没有获取到任何视频URL")
except Exception as e:
print(f"获取课程列表时出错: {e}")
import traceback
traceback.print_exc()
else:
print(f"请求失败，状态码：{first_response.status_code}")
print(f"响应内容: {first_response.text}")
批量下载代码：
[Python] 纯文本查看 复制代码 import requests
import imageio_ffmpeg
import os
import json
import time
import re
import shutil
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin, urlparse
from Crypto.Cipher import AES # 需要安装pycryptodome
import binascii
# 设置日志级别
DEBUG = True
# 创建video目录和temp目录（如果不存在）
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "video")
temp_dir = os.path.join(output_dir, "temp")
os.makedirs(output_dir, exist_ok=True)
os.makedirs(temp_dir, exist_ok=True)
print(f"Output directory: {output_dir}")
print(f"Temp directory: {temp_dir}")
# 处理文件名，确保合法
def sanitize_filename(filename):
# 替换或移除非法字符
return re.sub(r'[<>"/\\|?*]', '', filename)
# 创建全局会话，用于复用连接
session = requests.Session()
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
})
# 下载单个视频片段
def download_segment(segment_url, segment_file, encryption_key=None):
try:
if DEBUG:
print(f"Downloading segment: {segment_url}")
response = session.get(segment_url, timeout=60) # 使用会话，增加超时
response.raise_for_status() # 检查HTTP错误
content = response.content
# 如果有加密密钥，解密TS文件
if encryption_key:
try:
# AES-128解密，IV通常是文件编号（从文件名提取）
iv_match = re.search(r'segment_\d+_(\d+)\.ts', segment_file)
if iv_match:
segment_number = int(iv_match.group(1))
# 使用段号作为IV（16字节）
iv = segment_number.to_bytes(16, byteorder='big')
# 创建AES解密器
cipher = AES.new(encryption_key, AES.MODE_CBC, iv=iv)
# 解密内容（需要是16字节的倍数）
if len(content) % 16 == 0:
content = cipher.decrypt(content)
# 去除PKCS7填充
padding_len = content[-1]
content = content[:-padding_len]
else:
print(f"Warning: Segment {segment_file} length not multiple of 16, skipping decryption")
else:
print(f"Warning: Could not extract segment number from {segment_file}, skipping decryption")
except Exception as e:
print(f"Error decrypting segment {segment_file}: {e}")
# 保留原始内容，不中断下载
with open(segment_file, 'wb') as f:
f.write(content)
if DEBUG:
print(f"Segment saved: {segment_file}")
return segment_file
except Exception as e:
print(f"Error downloading segment {segment_url}: {e}")
raise
# 下载单个视频
def download_video_from_url(video_url, output_file, video_index):
try:
print(f"\nStarting download for: {output_file}")
# 解析URL，获取基础URL和m3u8路径
parsed_url = urlparse(video_url)
base_url = f"{parsed_url.scheme}://{parsed_url.netloc}{os.path.dirname(parsed_url.path)}/"
m3u8_path = os.path.basename(parsed_url.path) + "?" + parsed_url.query if parsed_url.query else os.path.basename(parsed_url.path)
# 下载m3u8文件（使用会话提高速度）
print(f"Downloading m3u8 file from {video_url}")
response = session.get(video_url, timeout=30)
response.raise_for_status() # 检查HTTP错误
m3u8_content = response.text
# 不保存m3u8文件到本地，直接处理内容
# 检查m3u8内容是否包含加密信息
encryption_key = None
if "#EXT-X-KEY" in m3u8_content:
print("Encrypted HLS stream detected (AES-128)")
# 提取加密信息
import re
key_matches = re.findall(r'#EXT-X-KEY:([^\n]+)', m3u8_content)
if key_matches:
key_info = key_matches[0]
print(f"Encryption details: {key_info}")
# 提取密钥URL
key_url_match = re.search(r'URI="([^"]+)"', key_info)
if key_url_match:
key_url = key_url_match.group(1)
print(f"Downloading encryption key from: {key_url}")
# 下载密钥
try:
key_response = session.get(key_url, timeout=30)
key_response.raise_for_status()
encryption_key = key_response.content
print(f"Successfully downloaded encryption key ({len(encryption_key)} bytes)")
except Exception as e:
print(f"Error downloading encryption key: {e}")
encryption_key = None
if DEBUG:
print(f"M3U8 content length: {len(m3u8_content)} bytes")
print(f"M3U8 content preview: {m3u8_content[:200]}...")
# 提取所有视频片段URL
segments = [line.strip() for line in m3u8_content.split('\n') if line and not line.startswith("#")]
print(f"Found {len(segments)} segments")
if not segments:
print(f"No segments found in m3u8 file")
return
segment_files = []
# 生成固定的时间戳，用于所有segment文件名
timestamp = int(time.time() * 1000)
# 为当前视频创建独立的临时目录
video_temp_dir = os.path.join(temp_dir, f"video_{video_index}")
os.makedirs(video_temp_dir, exist_ok=True)
print(f"Using temporary directory: {video_temp_dir}")
# 创建一个字典来存储片段URL和对应的文件路径
segment_info = {}
# 多线程下载视频片段（增加到16个线程以提高下载速度）
with ThreadPoolExecutor(max_workers=16) as executor:
futures = []
for i, segment_url in enumerate(segments):
absolute_url = urljoin(base_url, segment_url)
segment_file = os.path.join(video_temp_dir, f"segment_{timestamp}_{i}.ts") # 保存到视频专用临时目录
segment_info[absolute_url] = segment_file
futures.append(executor.submit(download_segment, absolute_url, segment_file, encryption_key))
# 等待所有片段下载完成
downloaded_count = 0
failed_count = 0
for future in as_completed(futures):
try:
result = future.result()
downloaded_count += 1
segment_files.append(result) # 只添加成功下载的文件
print(f"Downloaded segment {downloaded_count}/{len(segments)}")
except Exception as e:
failed_count += 1
print(f"Error in segment download: {e}")
print(f"Download complete: {downloaded_count} segments downloaded successfully, {failed_count} failed")
# 检查是否有足够的片段下载成功
if not segment_files:
print("No segments were downloaded successfully")
return
# 确保片段按顺序排列（按数字排序）
segment_files.sort(key=lambda x: int(x.split('_')[-1].split('.')[0])) # 提取数字部分进行排序
# 合并视频片段
print(f"Merging {len(segment_files)} segments into {output_file}")
# 使用Python直接合并TS文件为TS格式
merge_success = False
try:
# 过滤掉不存在的文件
existing_segment_files = []
for seg_file in segment_files:
if os.path.exists(seg_file):
existing_segment_files.append(seg_file)
else:
print(f"Warning: Segment file not found: {seg_file}")
if not existing_segment_files:
print("Error: No existing segment files found for merging")
return
print(f"Found {len(existing_segment_files)} existing segment files for merging")
# 调试：显示前几个文件
print("First 5 segment files:")
for i, seg_file in enumerate(existing_segment_files[:5]):
print(f" {i+1}: {seg_file} (Size: {os.path.getsize(seg_file)} bytes)")
# 创建临时TS格式文件
ts_output_file = output_file.replace('.mp4', '_temp.ts')
print(f"\nMerging all TS segments directly into {ts_output_file}...")
# 合并文件
with open(ts_output_file, 'wb') as outfile:
total_size = 0
for i, seg_file in enumerate(existing_segment_files):
with open(seg_file, 'rb') as infile:
content = infile.read()
outfile.write(content)
total_size += len(content)
if (i + 1) % 100 == 0 or i + 1 == len(existing_segment_files):
print(f" Merged {i + 1}/{len(existing_segment_files)} segments ({total_size/1024/1024:.1f} MB)")
# 检查输出文件
if os.path.exists(ts_output_file):
file_size = os.path.getsize(ts_output_file)
print(f"Successfully merged TS segments into temporary file {ts_output_file} (Size: {file_size} bytes)")
# 使用FFmpeg将TS转换为MP4
print(f"Converting TS to MP4: {output_file}...")
try:
# 构建FFmpeg命令
import shutil
ffmpeg_path = shutil.which('ffmpeg')
if not ffmpeg_path:
print("Error: FFmpeg not found in PATH")
print("You can manually convert TS to MP4 using:")
print(f" ffmpeg -i '{ts_output_file}' -c copy '{output_file}'")
merge_success = False # FFmpeg不存在，无法转换，合并失败
return
ffmpeg_cmd = [
ffmpeg_path,
'-i', ts_output_file,
'-c', 'copy', # 直接复制流，不重新编码
'-y', # 覆盖输出文件
output_file
]
# 使用UTF-8编码处理输出，避免UnicodeDecodeError
result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
if result.returncode == 0:
# 检查MP4文件是否创建成功且大小合理
if os.path.exists(output_file) and os.path.getsize(output_file) > 0:
mp4_size = os.path.getsize(output_file)
print(f"Successfully converted to MP4: {output_file} (Size: {mp4_size} bytes)")
merge_success = True
# 删除临时TS文件
if os.path.exists(ts_output_file):
os.remove(ts_output_file)
else:
print(f"Error: MP4 output file {output_file} was not created or is empty")
merge_success = False
else:
print(f"FFmpeg conversion failed with return code: {result.returncode}")
print(f"FFmpeg stderr: {result.stderr}")
print("You can manually convert TS to MP4 using:")
print(f" ffmpeg -i '{ts_output_file}' -c copy '{output_file}'")
merge_success = False # 转换失败，合并失败
except Exception as e:
print(f"Error converting TS to MP4: {e}")
print("You can manually convert TS to MP4 using:")
print(f" ffmpeg -i '{ts_output_file}' -c copy '{output_file}'")
merge_success = False # 转换异常，合并失败
else:
print(f"Error: Temporary TS file {ts_output_file} was not created")
merge_success = False
except Exception as e:
print(f"Error merging segments: {e}")
import traceback
traceback.print_exc()
# 不要清理临时文件，以便调试
print("Keeping temporary files for debugging due to merge failure")
return
# 清理临时文件（只有合并成功后才清理）
if merge_success:
try:
# 使用批量删除提高效率，不显示详细信息
if os.path.exists(video_temp_dir):
start_time = time.time()
shutil.rmtree(video_temp_dir)
cleanup_time = time.time() - start_time
if cleanup_time > 5: # 只有清理时间超过5秒才显示信息
print(f"Cleanup took {cleanup_time:.2f} seconds")
except Exception as e:
print(f"Error cleaning up temporary files: {e}")
print(f"Successfully downloaded and merged: {output_file}")
else:
print(f"Merge failed, not cleaning temporary files for debugging")
except Exception as e:
print(f"Error downloading video {video_url}: {e}")
# 清理临时文件
try:
if os.path.exists(video_temp_dir):
shutil.rmtree(video_temp_dir)
except:
pass
# 重新抛出异常，让调用者知道下载失败
raise
# 从JSON文件下载所有视频
def download_all_videos(json_file):
try:
with open(json_file, 'r', encoding='utf-8') as f:
videos = json.load(f)
total_videos = len(videos)
print(f"Found {total_videos} videos in total")
print(f"Output directory: {output_dir}")
# 准备所有视频下载任务
video_tasks = []
for i, video in enumerate(videos):
video_name = sanitize_filename(video["chapterName"])
output_file = os.path.join(output_dir, f"{i+1:03d}_{video_name}.mp4")
video_url = video["filePath"]
# 检查文件是否已存在
if os.path.exists(output_file):
print(f"Video {i+1}/{total_videos} already exists, skipping: {output_file}")
continue
video_tasks.append((video_url, output_file, i+1))
remaining_videos = len(video_tasks)
print(f"Will download {remaining_videos} videos (skipping {total_videos - remaining_videos} already downloaded)")
if remaining_videos == 0:
print("\nAll videos are already downloaded!")
return
# 使用多线程同时下载5个视频任务
max_workers = 11 # 同时下载的视频数量
print(f"\nStarting download with {max_workers} concurrent video tasks...")
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 创建任务
futures = {}
for i, (video_url, output_file, video_index) in enumerate(video_tasks):
print(f"Scheduling video {video_index}/{total_videos}: {output_file}")
future = executor.submit(download_video_from_url, video_url, output_file, video_index)
futures[future] = (video_index, output_file)
# 处理结果
completed = 0
failed = 0
for future in as_completed(futures):
video_index, output_file = futures[future]
try:
future.result()
completed += 1
print(f"✓ Video {video_index} completed: {output_file}")
except Exception as e:
failed += 1
print(f"✗ Video {video_index} failed: {e}")
print(f"Progress: {completed + failed}/{remaining_videos} (Completed: {completed}, Failed: {failed})")
print("\nAll videos processed!")
print(f"Summary: {completed} videos downloaded successfully, {failed} failed")
except Exception as e:
print(f"Error in download_all_videos: {e}")
if __name__ == "__main__":
json_file = "d:\\code\\python\\test2\\xiaoetong\\小鹅通高项课程2026.json"
download_all_videos(json_file)
|
免费评分
-
查看全部评分
|
[复制链接]
发表于 2026-3-21 18:48
分享到朋友圈
