#!/usr/bin/env python # -*- coding: utf-8 -*- """ pip install playwright 抖音下载器 - 统一增强版 支持视频、图文、用户主页、合集等多种内容的批量下载 """ import asyncio import json import logging import os import re import sys import time from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from urllib.parse import urlparse import argparse import yaml # 第三方库 try: import aiohttp import requests from rich.console import Console from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeRemainingColumn from rich.table import Table from rich.panel import Panel from rich.live import Live from rich import print as rprint except ImportError as e: print(f"请安装必要的依赖: pip install aiohttp requests rich pyyaml") sys.exit(1) # 添加项目路径 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # 导入项目模块 from apiproxy.douyin import douyin_headers from apiproxy.douyin.urls import Urls from apiproxy.douyin.result import Result from apiproxy.common.utils import Utils from apiproxy.douyin.auth.cookie_manager import AutoCookieManager from apiproxy.douyin.database import DataBase # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('downloader.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Rich console console = Console() class ContentType: """内容类型枚举""" VIDEO = "video" IMAGE = "image" USER = "user" MIX = "mix" MUSIC = "music" LIVE = "live" class DownloadStats: """下载统计""" def __init__(self): self.total = 0 self.success = 0 self.failed = 0 self.skipped = 0 self.start_time = time.time() @property def success_rate(self): return (self.success / self.total * 100) if self.total > 0 else 0 @property def elapsed_time(self): return time.time() - self.start_time def to_dict(self): return { 'total': self.total, 'success': self.success, 'failed': self.failed, 'skipped': self.skipped, 'success_rate': f"{self.success_rate:.1f}%", 'elapsed_time': f"{self.elapsed_time:.1f}s" } class RateLimiter: """速率限制器""" def __init__(self, max_per_second: float = 2): self.max_per_second = max_per_second self.min_interval = 1.0 / max_per_second self.last_request = 0 async def acquire(self): """获取许可""" current = time.time() time_since_last = current - self.last_request if time_since_last < self.min_interval: await asyncio.sleep(self.min_interval - time_since_last) self.last_request = time.time() class RetryManager: """重试管理器""" def __init__(self, max_retries: int = 3): self.max_retries = max_retries self.retry_delays = [1, 2, 5] # 重试延迟 async def execute_with_retry(self, func, *args, **kwargs): """执行函数并自动重试""" last_error = None for attempt in range(self.max_retries): try: return await func(*args, **kwargs) except Exception as e: last_error = e if attempt < self.max_retries - 1: delay = self.retry_delays[min(attempt, len(self.retry_delays) - 1)] logger.warning(f"第 {attempt + 1} 次尝试失败: {e}, {delay}秒后重试...") await asyncio.sleep(delay) raise last_error class UnifiedDownloader: """统一下载器""" def __init__(self, config_path: str = "config.yml"): self.config = self._load_config(config_path) self.urls_helper = Urls() self.result_helper = Result() self.utils = Utils() # 组件初始化 self.stats = DownloadStats() self.rate_limiter = RateLimiter(max_per_second=2) self.retry_manager = RetryManager(max_retries=self.config.get('retry_times', 3)) # Cookie与请求头（延迟初始化，支持自动获取） self.cookies = self.config.get('cookies') if 'cookies' in self.config else self.config.get('cookie') self.auto_cookie = bool(self.config.get('auto_cookie')) or (isinstance(self.config.get('cookie'), str) and self.config.get('cookie') == 'auto') or (isinstance(self.config.get('cookies'), str) and self.config.get('cookies') == 'auto') self.headers = {**douyin_headers} # 避免服务端使用brotli导致aiohttp无法解压（未安装brotli库时会出现空响应） self.headers['accept-encoding'] = 'gzip, deflate' # 增量下载与数据库 self.increase_cfg: Dict[str, Any] = self.config.get('increase', {}) or {} self.enable_database: bool = bool(self.config.get('database', True)) self.db: Optional[DataBase] = DataBase() if self.enable_database else None # 保存路径 self.save_path = Path(self.config.get('path', './Downloaded')) self.save_path.mkdir(parents=True, exist_ok=True) def _load_config(self, config_path: str) -> Dict: """加载配置文件""" if not os.path.exists(config_path): # 兼容配置文件命名：优先 config.yml，其次 config_simple.yml alt_path = 'config_simple.yml' if os.path.exists(alt_path): config_path = alt_path else: # 返回一个空配置，由命令行参数决定 return {} with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) # 简化配置兼容：links/link, output_dir/path, cookie/cookies if 'links' in config and 'link' not in config: config['link'] = config['links'] if 'output_dir' in config and 'path' not in config: config['path'] = config['output_dir'] if 'cookie' in config and 'cookies' not in config: config['cookies'] = config['cookie'] if isinstance(config.get('cookies'), str) and config.get('cookies') == 'auto': config['auto_cookie'] = True # 允许无 link（通过命令行传入） # 如果两者都没有，后续会在运行时提示 return config def _build_cookie_string(self) -> str: """构建Cookie字符串""" if isinstance(self.cookies, str): return self.cookies elif isinstance(self.cookies, dict): return '; '.join([f'{k}={v}' for k, v in self.cookies.items()]) elif isinstance(self.cookies, list): # 支持来自AutoCookieManager的cookies列表 try: kv = {c.get('name'): c.get('value') for c in self.cookies if c.get('name') and c.get('value')} return '; '.join([f'{k}={v}' for k, v in kv.items()]) except Exception: return '' return '' async def _initialize_cookies_and_headers(self): """初始化Cookie与请求头（支持自动获取）""" # 若配置为字符串 'auto'，视为未提供，触发自动获取 if isinstance(self.cookies, str) and self.cookies.strip().lower() == 'auto': self.cookies = None # 若已显式提供cookies，则直接使用 cookie_str = self._build_cookie_string() if cookie_str: self.headers['Cookie'] = cookie_str # 同时设置到全局 douyin_headers，确保所有 API 请求都能使用 from apiproxy.douyin import douyin_headers douyin_headers['Cookie'] = cookie_str return # 自动获取Cookie if self.auto_cookie: try: console.print("[cyan]🔐 正在自动获取Cookie...[/cyan]") async with AutoCookieManager(cookie_file='cookies.pkl', headless=False) as cm: cookies_list = await cm.get_cookies() if cookies_list: self.cookies = cookies_list cookie_str = self._build_cookie_string() if cookie_str: self.headers['Cookie'] = cookie_str # 同时设置到全局 douyin_headers，确保所有 API 请求都能使用 from apiproxy.douyin import douyin_headers douyin_headers['Cookie'] = cookie_str console.print("[green]✅ Cookie获取成功[/green]") return console.print("[yellow]⚠️ 自动获取Cookie失败或为空，继续尝试无Cookie模式[/yellow]") except Exception as e: logger.warning(f"自动获取Cookie失败: {e}") console.print("[yellow]⚠️ 自动获取Cookie失败，继续尝试无Cookie模式[/yellow]") # 未能获取Cookie则不设置，使用默认headers def detect_content_type(self, url: str) -> ContentType: """检测URL内容类型""" if '/user/' in url: return ContentType.USER elif '/video/' in url or 'v.douyin.com' in url: return ContentType.VIDEO elif '/note/' in url: return ContentType.IMAGE elif '/collection/' in url or '/mix/' in url: return ContentType.MIX elif '/music/' in url: return ContentType.MUSIC elif 'live.douyin.com' in url: return ContentType.LIVE else: return ContentType.VIDEO # 默认当作视频 async def resolve_short_url(self, url: str) -> str: """解析短链接""" if 'v.douyin.com' in url: try: # 使用同步请求获取重定向 response = requests.get(url, headers=self.headers, allow_redirects=True, timeout=10) final_url = response.url logger.info(f"解析短链接: {url} -> {final_url}") return final_url except Exception as e: logger.warning(f"解析短链接失败: {e}") return url def extract_id_from_url(self, url: str, content_type: ContentType = None) -> Optional[str]: """从URL提取ID Args: url: 要解析的URL content_type: 内容类型（可选，用于指导提取） """ # 如果已知是用户页面，直接提取用户ID if content_type == ContentType.USER or '/user/' in url: user_patterns = [ r'/user/([\w-]+)', r'sec_uid=([\w-]+)' ] for pattern in user_patterns: match = re.search(pattern, url) if match: user_id = match.group(1) logger.info(f"提取到用户ID: {user_id}") return user_id # 视频ID模式（优先） video_patterns = [ r'/video/(\d+)', r'/note/(\d+)', r'modal_id=(\d+)', r'aweme_id=(\d+)', r'item_id=(\d+)' ] for pattern in video_patterns: match = re.search(pattern, url) if match: video_id = match.group(1) logger.info(f"提取到视频ID: {video_id}") return video_id # 其他模式 other_patterns = [ r'/collection/(\d+)', r'/music/(\d+)' ] for pattern in other_patterns: match = re.search(pattern, url) if match: return match.group(1) # 尝试从URL中提取数字ID number_match = re.search(r'(\d{15,20})', url) if number_match: video_id = number_match.group(1) logger.info(f"从URL提取到数字ID: {video_id}") return video_id logger.error(f"无法从URL提取ID: {url}") return None def _get_aweme_id_from_info(self, info: Dict) -> Optional[str]: """从 aweme 信息中提取 aweme_id""" try: if 'aweme_id' in info: return str(info.get('aweme_id')) # aweme_detail 结构 return str(info.get('aweme', {}).get('aweme_id') or info.get('aweme_id')) except Exception: return None def _get_sec_uid_from_info(self, info: Dict) -> Optional[str]: """从 aweme 信息中提取作者 sec_uid""" try: return info.get('author', {}).get('sec_uid') except Exception: return None def _should_skip_increment(self, context: str, info: Dict, mix_id: Optional[str] = None, music_id: Optional[str] = None, sec_uid: Optional[str] = None) -> bool: """根据增量配置与数据库记录判断是否跳过下载""" if not self.db: return False aweme_id = self._get_aweme_id_from_info(info) if not aweme_id: return False try: if context == 'post' and self.increase_cfg.get('post', False): sec = sec_uid or self._get_sec_uid_from_info(info) or '' return bool(self.db.get_user_post(sec, int(aweme_id)) if aweme_id.isdigit() else None) if context == 'like' and self.increase_cfg.get('like', False): sec = sec_uid or self._get_sec_uid_from_info(info) or '' return bool(self.db.get_user_like(sec, int(aweme_id)) if aweme_id.isdigit() else None) if context == 'mix' and self.increase_cfg.get('mix', False): sec = sec_uid or self._get_sec_uid_from_info(info) or '' mid = mix_id or '' return bool(self.db.get_mix(sec, mid, int(aweme_id)) if aweme_id.isdigit() else None) if context == 'music' and self.increase_cfg.get('music', False): mid = music_id or '' return bool(self.db.get_music(mid, int(aweme_id)) if aweme_id.isdigit() else None) except Exception: return False return False def _record_increment(self, context: str, info: Dict, mix_id: Optional[str] = None, music_id: Optional[str] = None, sec_uid: Optional[str] = None): """下载成功后写入数据库记录""" if not self.db: return aweme_id = self._get_aweme_id_from_info(info) if not aweme_id or not aweme_id.isdigit(): return try: if context == 'post': sec = sec_uid or self._get_sec_uid_from_info(info) or '' self.db.insert_user_post(sec, int(aweme_id), info) elif context == 'like': sec = sec_uid or self._get_sec_uid_from_info(info) or '' self.db.insert_user_like(sec, int(aweme_id), info) elif context == 'mix': sec = sec_uid or self._get_sec_uid_from_info(info) or '' mid = mix_id or '' self.db.insert_mix(sec, mid, int(aweme_id), info) elif context == 'music': mid = music_id or '' self.db.insert_music(mid, int(aweme_id), info) except Exception: pass async def download_single_video(self, url: str, progress=None) -> bool: """下载单个视频/图文""" try: # 解析短链接 url = await self.resolve_short_url(url) # 提取ID video_id = self.extract_id_from_url(url, ContentType.VIDEO) if not video_id: logger.error(f"无法从URL提取ID: {url}") return False # 如果没有提取到视频ID，尝试作为视频ID直接使用 if not video_id and '/user/' not in url: # 可能短链接直接包含了视频ID video_id = url.split('/')[-2] if url.endswith('/') else url.split('/')[-1] logger.info(f"尝试从短链接路径提取ID: {video_id}") if not video_id: logger.error(f"无法从URL提取视频ID: {url}") return False # 限速 await self.rate_limiter.acquire() # 获取视频信息 if progress: progress.update(task_id=progress.task_ids[-1], description="获取视频信息...") video_info = await self.retry_manager.execute_with_retry( self._fetch_video_info, video_id ) if not video_info: logger.error(f"无法获取视频信息: {video_id}") self.stats.failed += 1 return False # 下载视频文件 if progress: progress.update(task_id=progress.task_ids[-1], description="下载视频文件...") success = await self._download_media_files(video_info, progress) if success: self.stats.success += 1 logger.info(f"✅ 下载成功: {url}") else: self.stats.failed += 1 logger.error(f"❌ 下载失败: {url}") return success except Exception as e: logger.error(f"下载视频异常 {url}: {e}") self.stats.failed += 1 return False finally: self.stats.total += 1 async def _fetch_video_info(self, video_id: str) -> Optional[Dict]: """获取视频信息""" try: # 直接使用 DouYinCommand.py 中成功的 Douyin 类 from apiproxy.douyin.douyin import Douyin # 创建 Douyin 实例 dy = Douyin(database=False) # 设置我们的 cookies 到 douyin_headers if hasattr(self, 'cookies') and self.cookies: cookie_str = self._build_cookie_string() if cookie_str: from apiproxy.douyin import douyin_headers douyin_headers['Cookie'] = cookie_str logger.info(f"设置 Cookie 到 Douyin 类: {cookie_str[:100]}...") try: # 使用现有的成功实现 result = dy.getAwemeInfo(video_id) if result: logger.info(f"Douyin 类成功获取视频信息: {result.get('desc', '')[:30]}") return result else: logger.error("Douyin 类返回空结果") except Exception as e: logger.error(f"Douyin 类获取视频信息失败: {e}") except Exception as e: logger.error(f"导入或使用 Douyin 类失败: {e}") import traceback traceback.print_exc() # 如果 Douyin 类失败，尝试备用接口（iesdouyin，无需X-Bogus） try: fallback_url = f"https://www.iesdouyin.com/web/api/v2/aweme/iteminfo/?item_ids={video_id}" logger.info(f"尝试备用接口获取视频信息: {fallback_url}") # 设置更通用的请求头 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Referer': 'https://www.douyin.com/', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive' } async with aiohttp.ClientSession() as session: async with session.get(fallback_url, headers=headers, timeout=15) as response: logger.info(f"备用接口响应状态: {response.status}") if response.status != 200: logger.error(f"备用接口请求失败，状态码: {response.status}") return None text = await response.text() logger.info(f"备用接口响应内容长度: {len(text)}") if not text: logger.error("备用接口响应为空") return None try: data = json.loads(text) logger.info(f"备用接口返回数据: {data}") item_list = (data or {}).get('item_list') or [] if item_list: aweme_detail = item_list[0] logger.info("备用接口成功获取视频信息") return aweme_detail else: logger.error("备用接口返回的数据中没有 item_list") except json.JSONDecodeError as e: logger.error(f"备用接口JSON解析失败: {e}") logger.error(f"原始响应内容: {text}") return None except Exception as e: logger.error(f"备用接口获取视频信息失败: {e}") return None def _build_detail_params(self, aweme_id: str) -> str: """构建详情API参数""" # 使用与现有 douyinapi.py 相同的参数格式 params = [ f'aweme_id={aweme_id}', 'device_platform=webapp', 'aid=6383' ] return '&'.join(params) async def _download_media_files(self, video_info: Dict, progress=None) -> bool: """下载媒体文件""" try: # 判断类型 is_image = bool(video_info.get('images')) # 构建保存路径 author_name = video_info.get('author', {}).get('nickname', 'unknown') desc = video_info.get('desc', '')[:50].replace('/', '_') # 兼容 create_time 为时间戳或格式化字符串 raw_create_time = video_info.get('create_time') dt_obj = None if isinstance(raw_create_time, (int, float)): dt_obj = datetime.fromtimestamp(raw_create_time) elif isinstance(raw_create_time, str) and raw_create_time: for fmt in ('%Y-%m-%d %H.%M.%S', '%Y-%m-%d_%H-%M-%S', '%Y-%m-%d %H:%M:%S'): try: dt_obj = datetime.strptime(raw_create_time, fmt) break except Exception: pass if dt_obj is None: dt_obj = datetime.fromtimestamp(time.time()) create_time = dt_obj.strftime('%Y-%m-%d_%H-%M-%S') folder_name = f"{create_time}_{desc}" if desc else create_time save_dir = self.save_path / author_name / folder_name save_dir.mkdir(parents=True, exist_ok=True) success = True if is_image: # 下载图文（无水印） images = video_info.get('images', []) for i, img in enumerate(images): img_url = self._get_best_quality_url(img.get('url_list', [])) if img_url: file_path = save_dir / f"image_{i+1}.jpg" if await self._download_file(img_url, file_path): logger.info(f"下载图片 {i+1}/{len(images)}: {file_path.name}") else: success = False else: # 下载视频（无水印） video_url = self._get_no_watermark_url(video_info) if video_url: file_path = save_dir / f"{folder_name}.mp4" if await self._download_file(video_url, file_path): logger.info(f"下载视频: {file_path.name}") else: success = False # 下载音频 if self.config.get('music', True): music_url = self._get_music_url(video_info) if music_url: file_path = save_dir / f"{folder_name}_music.mp3" await self._download_file(music_url, file_path) # 下载封面 if self.config.get('cover', True): cover_url = self._get_cover_url(video_info) if cover_url: file_path = save_dir / f"{folder_name}_cover.jpg" await self._download_file(cover_url, file_path) # 保存JSON数据 if self.config.get('json', True): json_path = save_dir / f"{folder_name}_data.json" with open(json_path, 'w', encoding='utf-8') as f: json.dump(video_info, f, ensure_ascii=False, indent=2) return success except Exception as e: logger.error(f"下载媒体文件失败: {e}") return False def _get_no_watermark_url(self, video_info: Dict) -> Optional[str]: """获取无水印视频URL""" try: # 优先使用play_addr_h264 play_addr = video_info.get('video', {}).get('play_addr_h264') or \ video_info.get('video', {}).get('play_addr') if play_addr: url_list = play_addr.get('url_list', []) if url_list: # 替换URL以获取无水印版本 url = url_list[0] url = url.replace('playwm', 'play') url = url.replace('720p', '1080p') return url # 备用：download_addr download_addr = video_info.get('video', {}).get('download_addr') if download_addr: url_list = download_addr.get('url_list', []) if url_list: return url_list[0] except Exception as e: logger.error(f"获取无水印URL失败: {e}") return None def _get_best_quality_url(self, url_list: List[str]) -> Optional[str]: """获取最高质量的URL""" if not url_list: return None # 优先选择包含特定关键词的URL for keyword in ['1080', 'origin', 'high']: for url in url_list: if keyword in url: return url # 返回第一个 return url_list[0] def _get_music_url(self, video_info: Dict) -> Optional[str]: """获取音乐URL""" try: music = video_info.get('music', {}) play_url = music.get('play_url', {}) url_list = play_url.get('url_list', []) return url_list[0] if url_list else None except: return None def _get_cover_url(self, video_info: Dict) -> Optional[str]: """获取封面URL""" try: cover = video_info.get('video', {}).get('cover', {}) url_list = cover.get('url_list', []) return self._get_best_quality_url(url_list) except: return None async def _download_file(self, url: str, save_path: Path) -> bool: """下载文件""" try: if save_path.exists(): logger.info(f"文件已存在，跳过: {save_path.name}") return True async with aiohttp.ClientSession() as session: async with session.get(url, headers=self.headers) as response: if response.status == 200: content = await response.read() with open(save_path, 'wb') as f: f.write(content) return True else: logger.error(f"下载失败，状态码: {response.status}") return False except Exception as e: logger.error(f"下载文件失败 {url}: {e}") return False async def download_user_page(self, url: str) -> bool: """下载用户主页内容""" try: # 提取用户ID user_id = self.extract_id_from_url(url, ContentType.USER) if not user_id: logger.error(f"无法从URL提取用户ID: {url}") return False console.print(f"\n[cyan]正在获取用户 {user_id} 的作品列表...[/cyan]") # 根据配置下载不同类型的内容 mode = self.config.get('mode', ['post']) if isinstance(mode, str): mode = [mode] # 增加总任务数统计 total_posts = 0 if 'post' in mode: total_posts += self.config.get('number', {}).get('post', 0) or 1 if 'like' in mode: total_posts += self.config.get('number', {}).get('like', 0) or 1 if 'mix' in mode: total_posts += self.config.get('number', {}).get('allmix', 0) or 1 self.stats.total += total_posts for m in mode: if m == 'post': await self._download_user_posts(user_id) elif m == 'like': await self._download_user_likes(user_id) elif m == 'mix': await self._download_user_mixes(user_id) return True except Exception as e: logger.error(f"下载用户主页失败: {e}") return False async def _download_user_posts(self, user_id: str): """下载用户发布的作品""" max_count = self.config.get('number', {}).get('post', 0) cursor = 0 downloaded = 0 console.print(f"\n[green]开始下载用户发布的作品...[/green]") with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeRemainingColumn(), console=console ) as progress: while True: # 限速 await self.rate_limiter.acquire() # 获取作品列表 posts_data = await self._fetch_user_posts(user_id, cursor) if not posts_data: break aweme_list = posts_data.get('aweme_list', []) if not aweme_list: break # 下载作品 for aweme in aweme_list: if max_count > 0 and downloaded >= max_count: console.print(f"[yellow]已达到下载数量限制: {max_count}[/yellow]") return # 时间过滤 if not self._check_time_filter(aweme): continue # 创建下载任务 task_id = progress.add_task( f"下载作品 {downloaded + 1}", total=100 ) # 增量判断 if self._should_skip_increment('post', aweme, sec_uid=user_id): continue # 下载 success = await self._download_media_files(aweme, progress) if success: downloaded += 1 self.stats.success += 1 # 增加成功计数 progress.update(task_id, completed=100) self._record_increment('post', aweme, sec_uid=user_id) else: self.stats.failed += 1 # 增加失败计数 progress.update(task_id, description="[red]下载失败[/red]") # 检查是否有更多 if not posts_data.get('has_more'): break cursor = posts_data.get('max_cursor', 0) console.print(f"[green]✅ 用户作品下载完成，共下载 {downloaded} 个[/green]") async def _fetch_user_posts(self, user_id: str, cursor: int = 0) -> Optional[Dict]: """获取用户作品列表""" try: # 直接使用 Douyin 类的 getUserInfo 方法，就像 DouYinCommand.py 那样 from apiproxy.douyin.douyin import Douyin # 创建 Douyin 实例 dy = Douyin(database=False) # 获取用户作品列表 result = dy.getUserInfo( user_id, "post", 35, 0, # 不限制数量 False, # 不启用增量 "", # start_time "" # end_time ) if result: logger.info(f"Douyin 类成功获取用户作品列表，共 {len(result)} 个作品") # 转换为期望的格式 return { 'status_code': 0, 'aweme_list': result, 'max_cursor': cursor, 'has_more': False } else: logger.error("Douyin 类返回空结果") return None except Exception as e: logger.error(f"获取用户作品列表失败: {e}") import traceback traceback.print_exc() return None async def _download_user_likes(self, user_id: str): """下载用户喜欢的作品""" max_count = 0 try: max_count = int(self.config.get('number', {}).get('like', 0)) except Exception: max_count = 0 cursor = 0 downloaded = 0 console.print(f"\n[green]开始下载用户喜欢的作品...[/green]") with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeRemainingColumn(), console=console ) as progress: while True: # 限速 await self.rate_limiter.acquire() # 获取喜欢列表 likes_data = await self._fetch_user_likes(user_id, cursor) if not likes_data: break aweme_list = likes_data.get('aweme_list', []) if not aweme_list: break # 下载作品 for aweme in aweme_list: if max_count > 0 and downloaded >= max_count: console.print(f"[yellow]已达到下载数量限制: {max_count}[/yellow]") return if not self._check_time_filter(aweme): continue task_id = progress.add_task( f"下载喜欢 {downloaded + 1}", total=100 ) # 增量判断 if self._should_skip_increment('like', aweme, sec_uid=user_id): continue success = await self._download_media_files(aweme, progress) if success: downloaded += 1 progress.update(task_id, completed=100) self._record_increment('like', aweme, sec_uid=user_id) else: progress.update(task_id, description="[red]下载失败[/red]") # 翻页 if not likes_data.get('has_more'): break cursor = likes_data.get('max_cursor', 0) console.print(f"[green]✅ 喜欢作品下载完成，共下载 {downloaded} 个[/green]") async def _fetch_user_likes(self, user_id: str, cursor: int = 0) -> Optional[Dict]: """获取用户喜欢的作品列表""" try: params_list = [ f'sec_user_id={user_id}', f'max_cursor={cursor}', 'count=35', 'aid=6383', 'device_platform=webapp', 'channel=channel_pc_web', 'pc_client_type=1', 'version_code=170400', 'version_name=17.4.0', 'cookie_enabled=true', 'screen_width=1920', 'screen_height=1080', 'browser_language=zh-CN', 'browser_platform=MacIntel', 'browser_name=Chrome', 'browser_version=122.0.0.0', 'browser_online=true' ] params = '&'.join(params_list) api_url = self.urls_helper.USER_FAVORITE_A try: xbogus = self.utils.getXbogus(params) full_url = f"{api_url}{params}&X-Bogus={xbogus}" except Exception as e: logger.warning(f"获取X-Bogus失败: {e}, 尝试不带X-Bogus") full_url = f"{api_url}{params}" logger.info(f"请求用户喜欢列表: {full_url[:100]}...") async with aiohttp.ClientSession() as session: async with session.get(full_url, headers=self.headers, timeout=10) as response: if response.status != 200: logger.error(f"请求失败，状态码: {response.status}") return None text = await response.text() if not text: logger.error("响应内容为空") return None data = json.loads(text) if data.get('status_code') == 0: return data else: logger.error(f"API返回错误: {data.get('status_msg', '未知错误')}") return None except Exception as e: logger.error(f"获取用户喜欢列表失败: {e}") return None async def _download_user_mixes(self, user_id: str): """下载用户的所有合集（按配置可限制数量）""" max_allmix = 0 try: # 兼容旧键名 allmix 或 mix number_cfg = self.config.get('number', {}) or {} max_allmix = int(number_cfg.get('allmix', number_cfg.get('mix', 0)) or 0) except Exception: max_allmix = 0 cursor = 0 fetched = 0 console.print(f"\n[green]开始获取用户合集列表...[/green]") while True: await self.rate_limiter.acquire() mix_list_data = await self._fetch_user_mix_list(user_id, cursor) if not mix_list_data: break mix_infos = mix_list_data.get('mix_infos') or [] if not mix_infos: break for mix in mix_infos: if max_allmix > 0 and fetched >= max_allmix: console.print(f"[yellow]已达到合集数量限制: {max_allmix}[/yellow]") return mix_id = mix.get('mix_id') mix_name = mix.get('mix_name', '') console.print(f"[cyan]下载合集[/cyan]: {mix_name} ({mix_id})") await self._download_mix_by_id(mix_id) fetched += 1 if not mix_list_data.get('has_more'): break cursor = mix_list_data.get('cursor', 0) console.print(f"[green]✅ 用户合集下载完成，共处理 {fetched} 个[/green]") async def _fetch_user_mix_list(self, user_id: str, cursor: int = 0) -> Optional[Dict]: """获取用户合集列表""" try: params_list = [ f'sec_user_id={user_id}', f'cursor={cursor}', 'count=35', 'aid=6383', 'device_platform=webapp', 'channel=channel_pc_web', 'pc_client_type=1', 'version_code=170400', 'version_name=17.4.0', 'cookie_enabled=true', 'screen_width=1920', 'screen_height=1080', 'browser_language=zh-CN', 'browser_platform=MacIntel', 'browser_name=Chrome', 'browser_version=122.0.0.0', 'browser_online=true' ] params = '&'.join(params_list) api_url = self.urls_helper.USER_MIX_LIST try: xbogus = self.utils.getXbogus(params) full_url = f"{api_url}{params}&X-Bogus={xbogus}" except Exception as e: logger.warning(f"获取X-Bogus失败: {e}, 尝试不带X-Bogus") full_url = f"{api_url}{params}" logger.info(f"请求用户合集列表: {full_url[:100]}...") async with aiohttp.ClientSession() as session: async with session.get(full_url, headers=self.headers, timeout=10) as response: if response.status != 200: logger.error(f"请求失败，状态码: {response.status}") return None text = await response.text() if not text: logger.error("响应内容为空") return None data = json.loads(text) if data.get('status_code') == 0: return data else: logger.error(f"API返回错误: {data.get('status_msg', '未知错误')}") return None except Exception as e: logger.error(f"获取用户合集列表失败: {e}") return None async def download_mix(self, url: str) -> bool: """根据合集链接下载合集内所有作品""" try: mix_id = None for pattern in [r'/collection/(\d+)', r'/mix/detail/(\d+)']: m = re.search(pattern, url) if m: mix_id = m.group(1) break if not mix_id: logger.error(f"无法从合集链接提取ID: {url}") return False await self._download_mix_by_id(mix_id) return True except Exception as e: logger.error(f"下载合集失败: {e}") return False async def _download_mix_by_id(self, mix_id: str): """按合集ID下载全部作品""" cursor = 0 downloaded = 0 console.print(f"\n[green]开始下载合集 {mix_id} ...[/green]") while True: await self.rate_limiter.acquire() data = await self._fetch_mix_awemes(mix_id, cursor) if not data: break aweme_list = data.get('aweme_list') or [] if not aweme_list: break for aweme in aweme_list: success = await self._download_media_files(aweme) if success: downloaded += 1 if not data.get('has_more'): break cursor = data.get('cursor', 0) console.print(f"[green]✅ 合集下载完成，共下载 {downloaded} 个[/green]") async def _fetch_mix_awemes(self, mix_id: str, cursor: int = 0) -> Optional[Dict]: """获取合集下作品列表""" try: params_list = [ f'mix_id={mix_id}', f'cursor={cursor}', 'count=35', 'aid=6383', 'device_platform=webapp', 'channel=channel_pc_web', 'pc_client_type=1', 'version_code=170400', 'version_name=17.4.0', 'cookie_enabled=true', 'screen_width=1920', 'screen_height=1080', 'browser_language=zh-CN', 'browser_platform=MacIntel', 'browser_name=Chrome', 'browser_version=122.0.0.0', 'browser_online=true' ] params = '&'.join(params_list) api_url = self.urls_helper.USER_MIX try: xbogus = self.utils.getXbogus(params) full_url = f"{api_url}{params}&X-Bogus={xbogus}" except Exception as e: logger.warning(f"获取X-Bogus失败: {e}, 尝试不带X-Bogus") full_url = f"{api_url}{params}" logger.info(f"请求合集作品列表: {full_url[:100]}...") async with aiohttp.ClientSession() as session: async with session.get(full_url, headers=self.headers, timeout=10) as response: if response.status != 200: logger.error(f"请求失败，状态码: {response.status}") return None text = await response.text() if not text: logger.error("响应内容为空") return None data = json.loads(text) # USER_MIX 返回没有统一的 status_code，这里直接返回 return data except Exception as e: logger.error(f"获取合集作品失败: {e}") return None async def download_music(self, url: str) -> bool: """根据音乐页链接下载音乐下的所有作品（支持增量）""" try: # 提取 music_id music_id = None m = re.search(r'/music/(\d+)', url) if m: music_id = m.group(1) if not music_id: logger.error(f"无法从音乐链接提取ID: {url}") return False cursor = 0 downloaded = 0 limit_num = 0 try: limit_num = int((self.config.get('number', {}) or {}).get('music', 0)) except Exception: limit_num = 0 console.print(f"\n[green]开始下载音乐 {music_id} 下的作品...[/green]") while True: await self.rate_limiter.acquire() data = await self._fetch_music_awemes(music_id, cursor) if not data: break aweme_list = data.get('aweme_list') or [] if not aweme_list: break for aweme in aweme_list: if limit_num > 0 and downloaded >= limit_num: console.print(f"[yellow]已达到音乐下载数量限制: {limit_num}[/yellow]") return True if self._should_skip_increment('music', aweme, music_id=music_id): continue success = await self._download_media_files(aweme) if success: downloaded += 1 self._record_increment('music', aweme, music_id=music_id) if not data.get('has_more'): break cursor = data.get('cursor', 0) console.print(f"[green]✅ 音乐作品下载完成，共下载 {downloaded} 个[/green]") return True except Exception as e: logger.error(f"下载音乐页失败: {e}") return False async def _fetch_music_awemes(self, music_id: str, cursor: int = 0) -> Optional[Dict]: """获取音乐下作品列表""" try: params_list = [ f'music_id={music_id}', f'cursor={cursor}', 'count=35', 'aid=6383', 'device_platform=webapp', 'channel=channel_pc_web', 'pc_client_type=1', 'version_code=170400', 'version_name=17.4.0', 'cookie_enabled=true', 'screen_width=1920', 'screen_height=1080', 'browser_language=zh-CN', 'browser_platform=MacIntel', 'browser_name=Chrome', 'browser_version=122.0.0.0', 'browser_online=true' ] params = '&'.join(params_list) api_url = self.urls_helper.MUSIC try: xbogus = self.utils.getXbogus(params) full_url = f"{api_url}{params}&X-Bogus={xbogus}" except Exception as e: logger.warning(f"获取X-Bogus失败: {e}, 尝试不带X-Bogus") full_url = f"{api_url}{params}" logger.info(f"请求音乐作品列表: {full_url[:100]}...") async with aiohttp.ClientSession() as session: async with session.get(full_url, headers=self.headers, timeout=10) as response: if response.status != 200: logger.error(f"请求失败，状态码: {response.status}") return None text = await response.text() if not text: logger.error("响应内容为空") return None data = json.loads(text) return data except Exception as e: logger.error(f"获取音乐作品失败: {e}") return None def _check_time_filter(self, aweme: Dict) -> bool: """检查时间过滤""" start_time = self.config.get('start_time') end_time = self.config.get('end_time') if not start_time and not end_time: return True raw_create_time = aweme.get('create_time') if not raw_create_time: return True create_date = None if isinstance(raw_create_time, (int, float)): try: create_date = datetime.fromtimestamp(raw_create_time) except Exception: create_date = None elif isinstance(raw_create_time, str): for fmt in ('%Y-%m-%d %H.%M.%S', '%Y-%m-%d_%H-%M-%S', '%Y-%m-%d %H:%M:%S'): try: create_date = datetime.strptime(raw_create_time, fmt) break except Exception: pass if create_date is None: return True if start_time: start_date = datetime.strptime(start_time, '%Y-%m-%d') if create_date < start_date: return False if end_time: end_date = datetime.strptime(end_time, '%Y-%m-%d') if create_date > end_date: return False return True async def run(self): """运行下载器""" # 显示启动信息 console.print(Panel.fit( "[bold cyan]抖音下载器 v3.0 - 统一增强版[/bold cyan]\n" "[dim]支持视频、图文、用户主页、合集批量下载[/dim]", border_style="cyan" )) # 初始化Cookie与请求头 await self._initialize_cookies_and_headers() # 获取URL列表 urls = self.config.get('link', []) # 兼容：单条字符串 if isinstance(urls, str): urls = [urls] if not urls: console.print("[red]没有找到要下载的链接！[/red]") return # 分析URL类型 console.print(f"\n[cyan]📊 链接分析[/cyan]") url_types = {} for url in urls: content_type = self.detect_content_type(url) url_types[url] = content_type console.print(f" • {content_type.upper()}: {url[:50]}...") # 开始下载 console.print(f"\n[green]⏳ 开始下载 {len(urls)} 个链接...[/green]\n") for i, url in enumerate(urls, 1): content_type = url_types[url] console.print(f"[{i}/{len(urls)}] 处理: {url}") if content_type == ContentType.VIDEO or content_type == ContentType.IMAGE: await self.download_single_video(url) elif content_type == ContentType.USER: await self.download_user_page(url) # 若配置包含 like 或 mix，顺带处理 modes = self.config.get('mode', ['post']) if 'like' in modes: user_id = self.extract_id_from_url(url, ContentType.USER) if user_id: await self._download_user_likes(user_id) if 'mix' in modes: user_id = self.extract_id_from_url(url, ContentType.USER) if user_id: await self._download_user_mixes(user_id) elif content_type == ContentType.MIX: await self.download_mix(url) elif content_type == ContentType.MUSIC: await self.download_music(url) else: console.print(f"[yellow]不支持的内容类型: {content_type}[/yellow]") # 显示进度 console.print(f"进度: {i}/{len(urls)} | 成功: {self.stats.success} | 失败: {self.stats.failed}") console.print("-" * 60) # 显示统计 self._show_stats() def _show_stats(self): """显示下载统计""" console.print("\n" + "=" * 60) # 创建统计表格 table = Table(title="📊 下载统计", show_header=True, header_style="bold magenta") table.add_column("项目", style="cyan", width=12) table.add_column("数值", style="green") stats = self.stats.to_dict() table.add_row("总任务数", str(stats['total'])) table.add_row("成功", str(stats['success'])) table.add_row("失败", str(stats['failed'])) table.add_row("跳过", str(stats['skipped'])) table.add_row("成功率", stats['success_rate']) table.add_row("用时", stats['elapsed_time']) console.print(table) console.print("\n[bold green]✅ 下载任务完成！[/bold green]") def main(): """主函数""" parser = argparse.ArgumentParser( description='抖音下载器 - 统一增强版', formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument( '-c', '--config', default='config.yml', help='配置文件路径 (默认: config.yml，自动兼容 config_simple.yml)' ) parser.add_argument( '-u', '--url', nargs='+', help='直接指定要下载的URL' ) parser.add_argument( '-p', '--path', default=None, help='保存路径 (覆盖配置文件)' ) parser.add_argument( '--auto-cookie', action='store_true', help='自动获取Cookie（需要已安装Playwright）' ) parser.add_argument( '--cookie', help='手动指定Cookie字符串，例如 "msToken=xxx; ttwid=yyy"' ) args = parser.parse_args() # 组合配置来源：优先命令行 temp_config = {} if args.url: temp_config['link'] = args.url # 覆盖保存路径 if args.path: temp_config['path'] = args.path # Cookie配置 if args.auto_cookie: temp_config['auto_cookie'] = True temp_config['cookies'] = 'auto' if args.cookie: temp_config['cookies'] = args.cookie temp_config['auto_cookie'] = False # 如果存在临时配置，则生成一个临时文件供现有构造函数使用 if temp_config: # 合并文件配置（如存在） file_config = {} if os.path.exists(args.config): try: with open(args.config, 'r', encoding='utf-8') as f: file_config = yaml.safe_load(f) or {} except Exception: file_config = {} # 兼容简化键名 if 'links' in file_config and 'link' not in file_config: file_config['link'] = file_config['links'] if 'output_dir' in file_config and 'path' not in file_config: file_config['path'] = file_config['output_dir'] if 'cookie' in file_config and 'cookies' not in file_config: file_config['cookies'] = file_config['cookie'] merged = {**(file_config or {}), **temp_config} with open('temp_config.yml', 'w', encoding='utf-8') as f: yaml.dump(merged, f, allow_unicode=True) config_path = 'temp_config.yml' else: config_path = args.config # 运行下载器 try: downloader = UnifiedDownloader(config_path) asyncio.run(downloader.run()) except KeyboardInterrupt: console.print("\n[yellow]⚠️ 用户中断下载[/yellow]") except Exception as e: console.print(f"\n[red]❌ 程序异常: {e}[/red]") logger.exception("程序异常") finally: # 清理临时配置 if args.url and os.path.exists('temp_config.yml'): os.remove('temp_config.yml') if __name__ == '__main__': main()