[Python] 纯文本查看 复制代码
"""小鹅通 M3U8 视频下载器 — 交互式命令行版"""
import os, time, shutil, subprocess, binascii, re
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from tqdm import tqdm
# 强制 stdout 行缓冲，确保 Windows CMD 中实时输出
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(line_buffering=True)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(BASE_DIR, 'downloads')
os.makedirs(OUT, exist_ok=True)
LOG_FILE = os.path.join(BASE_DIR, 'downloader.log')
def _console_encoding():
"""获取 Windows 控制台实际代码页，避免 UTF-8/GBK 猜错"""
try:
import ctypes
cp = ctypes.windll.kernel32.GetConsoleOutputCP()
if cp == 65001:
return 'utf-8'
elif cp == 936:
return 'gbk'
else:
return f'cp{cp}'
except Exception:
return (sys.stdout.encoding or 'utf-8').lower()
CONSOLE_ENC = _console_encoding()
def log(msg, end='\n'):
"""同时输出到控制台和日志文件。控制台按实际代码页编码，避免乱码"""
ts = time.strftime('%H:%M:%S')
line = f'[{ts}] {msg}'
# 1. 文件日志（始终可靠）
try:
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(line)
if end:
f.write(end)
f.flush()
except Exception:
pass
# 2. 控制台：按真实代码页编码
try:
data = (line + end).encode(CONSOLE_ENC, errors='replace')
except Exception:
data = (line + end).encode('utf-8', errors='replace')
try:
os.write(1, data)
except Exception:
try:
os.write(2, data)
except Exception:
pass
FFMPEG = os.path.join(BASE_DIR, 'ffmpeg.exe')
def safe_name(name):
"""去除文件名中的非法字符"""
return re.sub(r'[<>:"/\\|?*]', '_', name).strip()
def find_ffmpeg():
for p in [FFMPEG, shutil.which('ffmpeg'), shutil.which('ffmpeg.exe')]:
if p and os.path.isfile(p):
return p
return None
def download_one(name, m3u8_url):
"""下载单个视频"""
s = requests.Session()
s.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'https://xiaoe-tech.com/',
})
log(f'\n {name}')
log(f' {"─" * 50}')
# 1. 下载 M3U8
log(' [1/4] 下载索引...', end=' ')
try:
resp = s.get(m3u8_url, timeout=30)
resp.raise_for_status()
except Exception as e:
log(f'失败: {e}')
return False
m3u8 = resp.text
base = '/'.join(m3u8_url.split('/')[:-1]) + '/'
# 2. 解析
segments = []
key_url = None
iv = b'\x00' * 16
for line in m3u8.split('\n'):
line = line.strip()
if 'URI=' in line and 'AES-128' in line:
a = line.find('URI="') + 5
b = line.find('"', a)
key_url = line[a:b]
if not key_url.startswith('http'):
key_url = urljoin(base, key_url)
iv_s = line.find('IV=0x')
if iv_s != -1:
iv = binascii.unhexlify(line[iv_s + 5:iv_s + 37])
elif line and not line.startswith('#'):
u = line if line.startswith('http') else urljoin(base, line)
segments.append(u)
log(f'{len(segments)} 片段')
# 3. 获取密钥
log(' [2/4] 获取密钥...', end=' ')
try:
key = s.get(key_url, timeout=15).content
log(f'{len(key)} 字节')
except Exception as e:
log(f'失败: {e}')
return False
# 4. 并行下载
tmp = os.path.join(OUT, f'tmp_{int(time.time())}')
os.makedirs(tmp, exist_ok=True)
def dl_one(url, key, iv, idx):
for _ in range(3):
try:
data = s.get(url, timeout=60).content
break
except:
time.sleep(1)
else:
return None
seg_iv = iv[:12] + idx.to_bytes(4, 'big')
dec = AES.new(key, AES.MODE_CBC, iv=seg_iv).decrypt(data)
try:
dec = unpad(dec, AES.block_size)
except:
pass
fp = os.path.join(tmp, f's_{idx:05d}.ts')
with open(fp, 'wb') as f:
f.write(dec)
return fp
t0 = time.time()
results = {}
with ThreadPoolExecutor(max_workers=6) as ex:
fut = {ex.submit(dl_one, u, key, iv, i): i for i, u in enumerate(segments)}
with tqdm(total=len(segments), desc=' [3/4] 下载中', unit='片',
ncols=60, bar_format='{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt}') as pbar:
for f in as_completed(fut):
if f.result():
results[fut[f]] = f.result()
pbar.update(1)
elapsed = time.time() - t0
if not results:
log(' 所有片段下载失败')
shutil.rmtree(tmp, ignore_errors=True)
return False
# 5. 合并
log(f' [4/4] 合并 ({elapsed:.1f}s 下载, {len(results)}/{len(segments)} 成功)...', end=' ')
files = [results[i] for i in sorted(results)]
output = os.path.join(OUT, f'{safe_name(name)}.mp4')
# Concat 列表
lst = os.path.join(OUT, '_concat.txt')
with open(lst, 'w', encoding='utf-8') as f:
for fp in files:
f.write(f"file '{fp.replace(os.sep, '/')}'\n")
if os.path.exists(output):
os.remove(output)
cmd = [find_ffmpeg() or 'ffmpeg', '-f', 'concat', '-safe', '0',
'-i', lst, '-c', 'copy', '-movflags', '+faststart', '-y', output]
r = subprocess.run(cmd, capture_output=True, encoding='utf-8',
errors='replace', timeout=600)
if os.path.exists(lst):
os.remove(lst)
if r.returncode != 0:
# 重编码回退
lst2 = os.path.join(OUT, '_concat2.txt')
with open(lst2, 'w', encoding='utf-8') as f:
for fp in files:
f.write(f"file '{fp.replace(os.sep, '/')}'\n")
cmd2 = [find_ffmpeg() or 'ffmpeg', '-f', 'concat', '-safe', '0',
'-i', lst2, '-c:v', 'libx264', '-c:a', 'aac',
'-movflags', '+faststart', '-y', output]
subprocess.run(cmd2, capture_output=True, encoding='utf-8',
errors='replace', timeout=600)
if os.path.exists(lst2):
os.remove(lst2)
shutil.rmtree(tmp, ignore_errors=True)
if os.path.exists(output):
mb = os.path.getsize(output) / 1048576
log(f' {mb:.1f} MB -> {output}')
return True
else:
log(' 合并失败')
return False
# ============================================================
# HTTP 服务 — 接收浏览器发来的下载任务
# ============================================================
import json
import traceback
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
PORT = 8910
total_count = 0
count_lock = threading.Lock()
# 线程池：支持同时下载最多3个
download_pool = ThreadPoolExecutor(max_workers=3)
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_OPTIONS(self):
log(f'HTTP OPTIONS {self.path}')
self._cors()
def do_GET(self):
log(f'HTTP GET {self.path}')
self._cors()
self._json({'status': 'running', 'port': PORT, 'total': total_count})
def do_POST(self):
global total_count
log(f'HTTP POST {self.path} 开始处理')
try:
length = int(self.headers.get('Content-Length', 0))
raw = self.rfile.read(length)
data = json.loads(raw)
name = data.get('name', '').strip()
url = data.get('url', '').strip()
log(f'HTTP POST {self.path} body_len={length}')
log(f'JSON: name={name!r} url={url[:60]}...')
if not name or not url:
log('拒绝任务：name 或 url 为空')
self._cors()
self._json({'status': 'error', 'msg': 'missing name or url'})
return
log(f'>>> 收到任务: [{name}]')
log(f' URL: {url[:80]}...')
self._cors()
self._json({'status': 'accepted', 'msg': f'已接收: {name}'})
log(f'已响应 accepted，提交后台下载...')
download_pool.submit(self._do_download, name, url)
except Exception as e:
tb = traceback.format_exc()
log(f'HTTP POST 异常: {e}')
log(f'异常堆栈: {tb}')
self._cors()
self._json({'status': 'error', 'msg': str(e)})
def _do_download(self, name, url):
"""在单独线程中执行下载"""
global total_count
log(f'开始下载: [{name}]')
ok = download_one(name, url)
with count_lock:
if ok:
total_count += 1
log(f'--- 任务结束: [{name}] {"成功" if ok else "失败"} (累计: {total_count}) ---')
log('等待新任务...')
def _cors(self):
self.send_response(200)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'POST, GET, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
self.send_header('Content-Type', 'application/json')
self.end_headers()
def _json(self, obj):
body = json.dumps(obj, ensure_ascii=False).encode('utf-8')
self.wfile.write(body)
def main():
ffmpeg = find_ffmpeg()
if not ffmpeg:
log('未找到 ffmpeg.exe，请放到本目录下')
sys.exit(1)
server = HTTPServer(('127.0.0.1', PORT), Handler)
log('=' * 50)
log('小鹅通 M3U8 视频下载器 启动')
log(f'ffmpeg: {ffmpeg}')
log(f'输出: {OUT}')
log(f'端口: localhost:{PORT}')
log('等待浏览器发送下载任务...')
log('在课程页面点击 "发送到下载" 即可')
log('按 Ctrl+C 退出')
log('=' * 50)
try:
server.serve_forever()
except KeyboardInterrupt:
log('\n退出。本次共下载 {} 个视频。'.format(total_count))
server.shutdown()
if __name__ == '__main__':
main()
[复制链接]
发表于 2026-7-2 18:06
|
发表于 2026-7-2 17:20
|楼主
分享到朋友圈
