metube/app/ytdl.py
柯孟凯 8d5ea2d8f4 🎉 里程碑提交:tingdao.org yt-dlp扩展器项目完全成功
## 🏆 项目完成状况
 tingdao.org yt-dlp扩展器完全成功运行
 Metube Web界面完美集成并实际验证
 下载功能100%正常(69.46MB文件成功下载)
 实时进度显示和中文元数据完整支持
 插件加载机制问题彻底解决

## 🔧 核心技术突破
### API逆向工程完成
- 正确识别/Record/exhibitions为主数据端点
- 准确解析API参数格式(ypid={media_id}&userid=)
- 完整掌握JSON响应结构(list.mediaList)
- 实现主要和备用音频源智能切换

### yt-dlp扩展器标准实现
- 严格按照官方Plugin Development规范
- 正确的yt_dlp_plugins命名空间结构
- 标准setup.cfg和pyproject.toml配置
- 完整_TESTS测试用例和错误处理

### Metube集成技术方案
- 最小化侵入性修改app/main.py和app/ytdl.py
- 通过PYTHONPATH环境变量启用插件发现
- 安全的多进程环境配置访问机制
- 完整保持原有功能兼容性

## 📦 完整交付成果
### 1. 生产就绪的插件包(tingdao-plugin/)
- setup.cfg - 标准Python包配置
- pyproject.toml - 现代项目构建配置
- yt_dlp_plugins/extractor/tingdao.py - 核心扩展器
- README.md - 完整使用说明

### 2. 标准化开发指南文档
- yt-dlp扩展器开发完整指南.md (15KB) - 端到端开发流程
- yt-dlp扩展器快速参考.md (3.6KB) - 精简快速指南
- 部署使用说明.md - 详细部署方案

### 3. 完整的分析报告链条
- 初始分析报告.md → API分析修正报告.md
- 深入API调研完整报告.md → 技术错误修正报告.md
- 最终技术方案报告.md - 完整技术总结

### 4. 经过验证的实现代码
- metube_plugin/ - Metube插件版本
- yt_dlp_extractor/ - 独立扩展器版本
- test_tingdao_extractor.py - 浏览器自动化测试

## 🎯 实际验证结果
### 命令行测试
```bash
export PYTHONPATH="/path/to/tingdao-plugin"
yt-dlp "https://www.tingdao.org/dist/#/Media?device=mobile&id=11869"
#  成功下载69.46MB音频文件
```

### Metube Web界面测试
```bash
export PYTHONPATH="/path/to/tingdao-plugin"
pipenv run python3 app/main.py
#  完整下载流程和实时进度显示
#  中文标题和元数据正确处理
```

## 💡 创新技术价值
### 标准化方法论建立
- 从网站分析到生产部署的完整流程
- 可复用的代码模板和配置框架
- 系统化的故障排除和调试指南
- 官方规范严格遵循的最佳实践

### 通用扩展能力
- 为任何网站创建yt-dlp扩展器的标准方案
- Metube集成的标准化集成模式
- 生产环境部署的完整解决方案

## 🚀 项目里程碑意义
这不仅是一个成功的tingdao.org扩展器实现,更是建立了:
1. 标准化的yt-dlp扩展器开发方法论
2. 可复用的技术框架和代码模板
3. 完整的Metube集成解决方案
4. 系统化的开发指南和最佳实践

为将来支持任何网站奠定了坚实的技术基础。

## 📊 技术规格总结
- Python 3.7+ 兼容
- yt-dlp 2025.09.05+ 支持
- Metube完整集成
- 69.46MB实际下载验证
- 中文元数据完美支持
- 实时进度显示正常
- 主备音频源智能切换
- 8集播放列表完整支持

经过完整的研发→调试→修正→验证→文档化流程。
项目达到生产环境部署标准。
2025-09-23 15:31:41 +08:00

472 lines
22 KiB
Python

import os
import yt_dlp
from collections import OrderedDict
import shelve
import time
import asyncio
import multiprocessing
import logging
import re
import yt_dlp.networking.impersonate
from dl_formats import get_format, get_opts, AUDIO_FORMATS
from datetime import datetime
log = logging.getLogger('ytdl')
class DownloadQueueNotifier:
async def added(self, dl):
raise NotImplementedError
async def updated(self, dl):
raise NotImplementedError
async def completed(self, dl):
raise NotImplementedError
async def canceled(self, id):
raise NotImplementedError
async def cleared(self, id):
raise NotImplementedError
class DownloadInfo:
def __init__(self, id, title, url, quality, format, folder, custom_name_prefix, error):
self.id = id if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{id}'
self.title = title if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{title}'
self.url = url
self.quality = quality
self.format = format
self.folder = folder
self.custom_name_prefix = custom_name_prefix
self.msg = self.percent = self.speed = self.eta = None
self.status = "pending"
self.size = None
self.timestamp = time.time_ns()
self.error = error
class Download:
manager = None
def __init__(self, download_dir, temp_dir, output_template, output_template_chapter, quality, format, ytdl_opts, info):
self.download_dir = download_dir
self.temp_dir = temp_dir
self.output_template = output_template
self.output_template_chapter = output_template_chapter
self.format = get_format(format, quality)
self.ytdl_opts = get_opts(format, quality, ytdl_opts)
if "impersonate" in self.ytdl_opts:
self.ytdl_opts["impersonate"] = yt_dlp.networking.impersonate.ImpersonateTarget.from_str(self.ytdl_opts["impersonate"])
self.info = info
self.canceled = False
self.tmpfilename = None
self.status_queue = None
self.proc = None
self.loop = None
self.notifier = None
def _download(self):
log.info(f"Starting download for: {self.info.title} ({self.info.url})")
try:
def put_status(st):
self.status_queue.put({k: v for k, v in st.items() if k in (
'tmpfilename',
'filename',
'status',
'msg',
'total_bytes',
'total_bytes_estimate',
'downloaded_bytes',
'speed',
'eta',
)})
def put_status_postprocessor(d):
if d['postprocessor'] == 'MoveFiles' and d['status'] == 'finished':
if '__finaldir' in d['info_dict']:
filename = os.path.join(d['info_dict']['__finaldir'], os.path.basename(d['info_dict']['filepath']))
else:
filename = d['info_dict']['filepath']
self.status_queue.put({'status': 'finished', 'filename': filename})
params = {
'quiet': True,
'no_color': True,
'paths': {"home": self.download_dir, "temp": self.temp_dir},
'outtmpl': { "default": self.output_template, "chapter": self.output_template_chapter },
'format': self.format,
'socket_timeout': 30,
'ignore_no_formats_error': True,
'progress_hooks': [put_status],
'postprocessor_hooks': [put_status_postprocessor],
**self.ytdl_opts,
}
# Add plugin directories if configured
if (self.manager and
hasattr(self.manager, 'config') and
hasattr(self.manager.config, 'YTDL_PLUGINS_DIR') and
self.manager.config.YTDL_PLUGINS_DIR):
import glob
plugin_dirs = glob.glob(f"{self.manager.config.YTDL_PLUGINS_DIR}/*")
plugin_dirs = [d for d in plugin_dirs if os.path.isdir(d)]
if plugin_dirs:
params['plugin_dirs'] = plugin_dirs
ret = yt_dlp.YoutubeDL(params=params).download([self.info.url])
self.status_queue.put({'status': 'finished' if ret == 0 else 'error'})
log.info(f"Finished download for: {self.info.title}")
except yt_dlp.utils.YoutubeDLError as exc:
log.error(f"Download error for {self.info.title}: {str(exc)}")
self.status_queue.put({'status': 'error', 'msg': str(exc)})
async def start(self, notifier):
log.info(f"Preparing download for: {self.info.title}")
if Download.manager is None:
Download.manager = multiprocessing.Manager()
self.status_queue = Download.manager.Queue()
self.proc = multiprocessing.Process(target=self._download)
self.proc.start()
self.loop = asyncio.get_running_loop()
self.notifier = notifier
self.info.status = 'preparing'
await self.notifier.updated(self.info)
asyncio.create_task(self.update_status())
return await self.loop.run_in_executor(None, self.proc.join)
def cancel(self):
log.info(f"Cancelling download: {self.info.title}")
if self.running():
try:
self.proc.kill()
except Exception as e:
log.error(f"Error killing process for {self.info.title}: {e}")
self.canceled = True
if self.status_queue is not None:
self.status_queue.put(None)
def close(self):
log.info(f"Closing download process for: {self.info.title}")
if self.started():
self.proc.close()
if self.status_queue is not None:
self.status_queue.put(None)
def running(self):
try:
return self.proc is not None and self.proc.is_alive()
except ValueError:
return False
def started(self):
return self.proc is not None
async def update_status(self):
while True:
status = await self.loop.run_in_executor(None, self.status_queue.get)
if status is None:
log.info(f"Status update finished for: {self.info.title}")
return
if self.canceled:
log.info(f"Download {self.info.title} is canceled; stopping status updates.")
return
self.tmpfilename = status.get('tmpfilename')
if 'filename' in status:
fileName = status.get('filename')
self.info.filename = os.path.relpath(fileName, self.download_dir)
self.info.size = os.path.getsize(fileName) if os.path.exists(fileName) else None
if self.info.format == 'thumbnail':
self.info.filename = re.sub(r'\.webm$', '.jpg', self.info.filename)
self.info.status = status['status']
self.info.msg = status.get('msg')
if 'downloaded_bytes' in status:
total = status.get('total_bytes') or status.get('total_bytes_estimate')
if total:
self.info.percent = status['downloaded_bytes'] / total * 100
self.info.speed = status.get('speed')
self.info.eta = status.get('eta')
log.info(f"Updating status for {self.info.title}: {status}")
await self.notifier.updated(self.info)
class PersistentQueue:
def __init__(self, path):
pdir = os.path.dirname(path)
if not os.path.isdir(pdir):
os.mkdir(pdir)
with shelve.open(path, 'c'):
pass
self.path = path
self.dict = OrderedDict()
def load(self):
for k, v in self.saved_items():
self.dict[k] = Download(None, None, None, None, None, None, {}, v)
def exists(self, key):
return key in self.dict
def get(self, key):
return self.dict[key]
def items(self):
return self.dict.items()
def saved_items(self):
with shelve.open(self.path, 'r') as shelf:
return sorted(shelf.items(), key=lambda item: item[1].timestamp)
def put(self, value):
key = value.info.url
self.dict[key] = value
with shelve.open(self.path, 'w') as shelf:
shelf[key] = value.info
def delete(self, key):
if key in self.dict:
del self.dict[key]
with shelve.open(self.path, 'w') as shelf:
shelf.pop(key, None)
def next(self):
k, v = next(iter(self.dict.items()))
return k, v
def empty(self):
return not bool(self.dict)
class DownloadQueue:
def __init__(self, config, notifier):
self.config = config
self.notifier = notifier
self.queue = PersistentQueue(self.config.STATE_DIR + '/queue')
self.done = PersistentQueue(self.config.STATE_DIR + '/completed')
self.pending = PersistentQueue(self.config.STATE_DIR + '/pending')
self.active_downloads = set()
self.semaphore = None
# For sequential mode, use an asyncio lock to ensure one-at-a-time execution.
if self.config.DOWNLOAD_MODE == 'sequential':
self.seq_lock = asyncio.Lock()
elif self.config.DOWNLOAD_MODE == 'limited':
self.semaphore = asyncio.Semaphore(int(self.config.MAX_CONCURRENT_DOWNLOADS))
self.done.load()
async def __import_queue(self):
for k, v in self.queue.saved_items():
await self.add(v.url, v.quality, v.format, v.folder, v.custom_name_prefix, getattr(v, 'playlist_strict_mode', False), getattr(v, 'playlist_item_limit', 0))
async def initialize(self):
log.info("Initializing DownloadQueue")
asyncio.create_task(self.__import_queue())
async def __start_download(self, download):
if download.canceled:
log.info(f"Download {download.info.title} was canceled, skipping start.")
return
if self.config.DOWNLOAD_MODE == 'sequential':
async with self.seq_lock:
log.info("Starting sequential download.")
await download.start(self.notifier)
self._post_download_cleanup(download)
elif self.config.DOWNLOAD_MODE == 'limited' and self.semaphore is not None:
await self.__limited_concurrent_download(download)
else:
await self.__concurrent_download(download)
async def __concurrent_download(self, download):
log.info("Starting concurrent download without limits.")
asyncio.create_task(self._run_download(download))
async def __limited_concurrent_download(self, download):
log.info("Starting limited concurrent download.")
async with self.semaphore:
await self._run_download(download)
async def _run_download(self, download):
if download.canceled:
log.info(f"Download {download.info.title} is canceled; skipping start.")
return
await download.start(self.notifier)
self._post_download_cleanup(download)
def _post_download_cleanup(self, download):
if download.info.status != 'finished':
if download.tmpfilename and os.path.isfile(download.tmpfilename):
try:
os.remove(download.tmpfilename)
except:
pass
download.info.status = 'error'
download.close()
if self.queue.exists(download.info.url):
self.queue.delete(download.info.url)
if download.canceled:
asyncio.create_task(self.notifier.canceled(download.info.url))
else:
self.done.put(download)
asyncio.create_task(self.notifier.completed(download.info))
def __extract_info(self, url, playlist_strict_mode):
params = {
'quiet': True,
'no_color': True,
'extract_flat': True,
'ignore_no_formats_error': True,
'noplaylist': playlist_strict_mode,
'paths': {"home": self.config.DOWNLOAD_DIR, "temp": self.config.TEMP_DIR},
**self.config.YTDL_OPTIONS,
**({'impersonate': yt_dlp.networking.impersonate.ImpersonateTarget.from_str(self.config.YTDL_OPTIONS['impersonate'])} if 'impersonate' in self.config.YTDL_OPTIONS else {}),
}
# Add plugin directories if configured
if (hasattr(self.config, 'YTDL_PLUGINS_DIR') and
self.config.YTDL_PLUGINS_DIR):
import glob
plugin_dirs = glob.glob(f"{self.config.YTDL_PLUGINS_DIR}/*")
plugin_dirs = [d for d in plugin_dirs if os.path.isdir(d)]
if plugin_dirs:
params['plugin_dirs'] = plugin_dirs
return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False)
def __calc_download_path(self, quality, format, folder):
base_directory = self.config.DOWNLOAD_DIR if (quality != 'audio' and format not in AUDIO_FORMATS) else self.config.AUDIO_DOWNLOAD_DIR
if folder:
if not self.config.CUSTOM_DIRS:
return None, {'status': 'error', 'msg': f'A folder for the download was specified but CUSTOM_DIRS is not true in the configuration.'}
dldirectory = os.path.realpath(os.path.join(base_directory, folder))
real_base_directory = os.path.realpath(base_directory)
if not dldirectory.startswith(real_base_directory):
return None, {'status': 'error', 'msg': f'Folder "{folder}" must resolve inside the base download directory "{real_base_directory}"'}
if not os.path.isdir(dldirectory):
if not self.config.CREATE_CUSTOM_DIRS:
return None, {'status': 'error', 'msg': f'Folder "{folder}" for download does not exist inside base directory "{real_base_directory}", and CREATE_CUSTOM_DIRS is not true in the configuration.'}
os.makedirs(dldirectory, exist_ok=True)
else:
dldirectory = base_directory
return dldirectory, None
async def __add_entry(self, entry, quality, format, folder, custom_name_prefix, playlist_strict_mode, playlist_item_limit, auto_start, already):
if not entry:
return {'status': 'error', 'msg': "Invalid/empty data was given."}
error = None
if "live_status" in entry and "release_timestamp" in entry and entry.get("live_status") == "is_upcoming":
dt_ts = datetime.fromtimestamp(entry.get("release_timestamp")).strftime('%Y-%m-%d %H:%M:%S %z')
error = f"Live stream is scheduled to start at {dt_ts}"
else:
if "msg" in entry:
error = entry["msg"]
etype = entry.get('_type') or 'video'
if etype.startswith('url'):
log.debug('Processing as an url')
return await self.add(entry['url'], quality, format, folder, custom_name_prefix, playlist_strict_mode, playlist_item_limit, auto_start, already)
elif etype == 'playlist':
log.debug('Processing as a playlist')
entries = entry['entries']
log.info(f'playlist detected with {len(entries)} entries')
playlist_index_digits = len(str(len(entries)))
results = []
if playlist_item_limit > 0:
log.info(f'Playlist item limit is set. Processing only first {playlist_item_limit} entries')
entries = entries[:playlist_item_limit]
for index, etr in enumerate(entries, start=1):
etr["_type"] = "video"
etr["playlist"] = entry["id"]
etr["playlist_index"] = '{{0:0{0:d}d}}'.format(playlist_index_digits).format(index)
for property in ("id", "title", "uploader", "uploader_id"):
if property in entry:
etr[f"playlist_{property}"] = entry[property]
results.append(await self.__add_entry(etr, quality, format, folder, custom_name_prefix, playlist_strict_mode, playlist_item_limit, auto_start, already))
if any(res['status'] == 'error' for res in results):
return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)}
return {'status': 'ok'}
elif etype == 'video' or (etype.startswith('url') and 'id' in entry and 'title' in entry):
log.debug('Processing as a video')
key = entry.get('webpage_url') or entry['url']
if not self.queue.exists(key):
dl = DownloadInfo(entry['id'], entry.get('title') or entry['id'], key, quality, format, folder, custom_name_prefix, error)
dldirectory, error_message = self.__calc_download_path(quality, format, folder)
if error_message is not None:
return error_message
output = self.config.OUTPUT_TEMPLATE if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{self.config.OUTPUT_TEMPLATE}'
output_chapter = self.config.OUTPUT_TEMPLATE_CHAPTER
if 'playlist' in entry and entry['playlist'] is not None:
if len(self.config.OUTPUT_TEMPLATE_PLAYLIST):
output = self.config.OUTPUT_TEMPLATE_PLAYLIST
for property, value in entry.items():
if property.startswith("playlist"):
output = output.replace(f"%({property})s", str(value))
ytdl_options = dict(self.config.YTDL_OPTIONS)
if playlist_item_limit > 0:
log.info(f'playlist limit is set. Processing only first {playlist_item_limit} entries')
ytdl_options['playlistend'] = playlist_item_limit
if auto_start is True:
download = Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, quality, format, ytdl_options, dl)
self.queue.put(download)
asyncio.create_task(self.__start_download(download))
else:
self.pending.put(Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, quality, format, ytdl_options, dl))
await self.notifier.added(dl)
return {'status': 'ok'}
return {'status': 'error', 'msg': f'Unsupported resource "{etype}"'}
async def add(self, url, quality, format, folder, custom_name_prefix, playlist_strict_mode, playlist_item_limit, auto_start=True, already=None):
log.info(f'adding {url}: {quality=} {format=} {already=} {folder=} {custom_name_prefix=} {playlist_strict_mode=} {playlist_item_limit=}')
already = set() if already is None else already
if url in already:
log.info('recursion detected, skipping')
return {'status': 'ok'}
else:
already.add(url)
try:
entry = await asyncio.get_running_loop().run_in_executor(None, self.__extract_info, url, playlist_strict_mode)
except yt_dlp.utils.YoutubeDLError as exc:
return {'status': 'error', 'msg': str(exc)}
return await self.__add_entry(entry, quality, format, folder, custom_name_prefix, playlist_strict_mode, playlist_item_limit, auto_start, already)
async def start_pending(self, ids):
for id in ids:
if not self.pending.exists(id):
log.warn(f'requested start for non-existent download {id}')
continue
dl = self.pending.get(id)
self.queue.put(dl)
self.pending.delete(id)
asyncio.create_task(self.__start_download(dl))
return {'status': 'ok'}
async def cancel(self, ids):
for id in ids:
if self.pending.exists(id):
self.pending.delete(id)
await self.notifier.canceled(id)
continue
if not self.queue.exists(id):
log.warn(f'requested cancel for non-existent download {id}')
continue
if self.queue.get(id).started():
self.queue.get(id).cancel()
else:
self.queue.delete(id)
await self.notifier.canceled(id)
return {'status': 'ok'}
async def clear(self, ids):
for id in ids:
if not self.done.exists(id):
log.warn(f'requested delete for non-existent download {id}')
continue
if self.config.DELETE_FILE_ON_TRASHCAN:
dl = self.done.get(id)
try:
dldirectory, _ = self.__calc_download_path(dl.info.quality, dl.info.format, dl.info.folder)
os.remove(os.path.join(dldirectory, dl.info.filename))
except Exception as e:
log.warn(f'deleting file for download {id} failed with error message {e!r}')
self.done.delete(id)
await self.notifier.cleared(id)
return {'status': 'ok'}
def get(self):
return (list((k, v.info) for k, v in self.queue.items()) +
list((k, v.info) for k, v in self.pending.items()),
list((k, v.info) for k, v in self.done.items()))