main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
唐人街影视网站电视剧下载工具
支持下载不同电视剧的指定集数或批量下载
用法：python main.py <电视剧ID> <集数> [结束集数]
"""
import re
import sys
import subprocess
import requests
import urllib.parse
import os
import time
import logging
from pathlib import Path
from typing import Optional

# 导入配置
from config import (
    BASE_URL_TEMPLATE, REQUEST_TIMEOUT, MAX_RETRIES,
    DOWNLOAD_DIR, STREAMLINK_THREADS, STREAMLINK_QUALITY,
    DEFAULT_HEADERS, FILENAME_INVALID_CHARS, FILENAME_REPLACEMENT_CHAR,
    LOG_LEVEL, LOG_FORMAT
)

# 配置日志
logging.basicConfig(level=getattr(logging, LOG_LEVEL), format=LOG_FORMAT)
logger = logging.getLogger(__name__)

def clean_filename(filename: str) -> str:
    """清理文件名，移除非法字符"""
    for char in FILENAME_INVALID_CHARS:
        filename = filename.replace(char, FILENAME_REPLACEMENT_CHAR)
    return filename.strip()

def make_request_with_retry(url: str, max_retries: int = MAX_RETRIES) -> Optional[str]:
    """带重试机制的HTTP请求"""
    for attempt in range(max_retries):
        try:
            logger.info(f"正在请求: {url} (尝试 {attempt + 1}/{max_retries})")
            response = requests.get(url, headers=DEFAULT_HEADERS, timeout=REQUEST_TIMEOUT)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            logger.warning(f"请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # 指数退避
            else:
                logger.error(f"所有重试都失败了: {e}")
                raise
    return None

def get_m3u8_url(drama_id: str, episode: int) -> str:
    """输入电视剧ID和第几集，返回真实的 m3u8 地址"""
    base_url = BASE_URL_TEMPLATE.format(drama_id)
    play_page = f"{base_url}/sid/2/nid/{episode}.html"
    
    html = make_request_with_retry(play_page)
    
    # 首先尝试原始的正则表达式
    m = re.search(r'url=(https[^&"]+\.m3u8)', html)
    if m:
        return m.group(1)
    
    # 尝试查找URL编码的m3u8链接
    encoded_pattern = r'"url":"([^"]+)"'
    matches = re.findall(encoded_pattern, html)
    for match in matches:
        decoded_url = urllib.parse.unquote(match)
        if 'm3u8' in decoded_url:
            print(f"找到编码的URL，解码后: {decoded_url}")
            return decoded_url
    
    raise RuntimeError("未解析到 m3u8，请检查该集是否存在")

def download(drama_id: str, drama_name: str, episode: int, m3u8_url: str):
    """下载指定集数的视频"""
    # 清理剧名，确保文件名合法
    clean_drama_name = clean_filename(drama_name)
    
    # 确保下载目录存在
    download_path = Path(DOWNLOAD_DIR) / clean_drama_name
    download_path.mkdir(parents=True, exist_ok=True)
    
    temp_file = download_path / f"{clean_drama_name}_第{episode}集.ts"
    outfile = download_path / f"{clean_drama_name}_第{episode}集.mp4"
    
    logger.info(f"开始下载第{episode}集到: {outfile}")
    
    try:
        # 先用streamlink下载为ts文件
        cmd1 = [
            "streamlink",
            "--stream-segment-threads", str(STREAMLINK_THREADS),
            "--loglevel", "info",
            "--http-no-ssl-verify",
            "--output", str(temp_file),
            m3u8_url,
            STREAMLINK_QUALITY
        ]
        logger.info(f"执行streamlink命令: {' '.join(cmd1)}")
        result1 = subprocess.run(cmd1, check=True, capture_output=True, text=True)
        logger.info("streamlink下载完成")
        
        # 用ffmpeg转换为标准MP4格式
        cmd2 = [
            "ffmpeg",
            "-i", str(temp_file),
            "-c", "copy",
            "-y",  # 覆盖输出文件
            str(outfile)
        ]
        logger.info(f"执行ffmpeg命令: {' '.join(cmd2)}")
        result2 = subprocess.run(cmd2, check=True, capture_output=True, text=True)
        logger.info("ffmpeg转换完成")
        
    except subprocess.CalledProcessError as e:
        logger.error(f"命令执行失败: {e}")
        logger.error(f"错误输出: {e.stderr}")
        raise RuntimeError(f"下载或转换失败: {e}") from e
    except FileNotFoundError as e:
        logger.error(f"命令未找到: {e}")
        raise RuntimeError(f"请确保已安装streamlink和ffmpeg: {e}") from e
    
    # 删除临时ts文件
    if temp_file.exists():
        temp_file.unlink()
    
    print(f"\n✅ 已保存：{outfile}")

def download_episodes(drama_id: str, drama_name: str, start_ep: int, end_ep: int):
    """批量下载指定范围的集数"""
    for ep in range(start_ep, end_ep + 1):
        try:
            print(f"\n开始下载第{ep}集...")
            m3u8 = get_m3u8_url(drama_id, ep)
            print(f"第{ep}集 m3u8：{m3u8}")
            download(drama_id, drama_name, ep, m3u8)
        except Exception as e:
            print(f"❌ 第{ep}集下载失败：{e}")
            continue

def get_drama_name(drama_id: str) -> str:
    """根据电视剧ID获取电视剧名称"""
    try:
        base_url = BASE_URL_TEMPLATE.format(drama_id)
        html = make_request_with_retry(base_url)
        
        # 尝试从页面标题中提取剧名
        title_match = re.search(r'<title>([^<]+)</title>', html)
        if title_match:
            title = title_match.group(1).strip()
            # 清理标题，移除网站名称等
            title = re.sub(r'[-_].*?唐人街.*', '', title)
            title = re.sub(r'[-_].*?在线观看.*', '', title)
            title = title.strip('-_ ')
            if title:
                return title
    except Exception as e:
        print(f"获取剧名失败：{e}")
    
    # 如果无法获取剧名，使用默认格式
    return f"电视剧_{drama_id}"

def print_usage():
    """打印使用说明"""
    print("唐人街影视网站电视剧下载工具")
    print("用法：")
    print("  单集下载：python main.py <电视剧ID> <集数> [自定义名称]")
    print("  批量下载：python main.py <电视剧ID> <起始集> <结束集> [自定义名称]")
    print("")
    print("示例：")
    print("  python main.py 163720 1                    # 下载ID为163720的电视剧第1集")
    print("  python main.py 163720 1 5                  # 下载ID为163720的电视剧第1-5集")
    print("  python main.py 163720 1 \"我的电视剧\"         # 下载并使用自定义名称")
    print("  python main.py 163720 1 5 \"我的电视剧\"       # 批量下载并使用自定义名称")
    print("")
    print("说明：")
    print("  - 电视剧ID可以从唐人街影视网站的播放页面URL中获取")
    print("  - 如果不提供自定义名称，将自动从网站获取电视剧名称")
    print("  - 下载的文件将保存在downloads目录下")

if __name__ == "__main__":
    if len(sys.argv) == 3:
        # 单集下载模式（无自定义名称）
        drama_id = sys.argv[1]
        ep = int(sys.argv[2])
        
        print(f"获取电视剧信息...")
        drama_name = get_drama_name(drama_id)
        print(f"电视剧：{drama_name} (ID: {drama_id})")
        
        try:
            m3u8 = get_m3u8_url(drama_id, ep)
            print(f"第{ep}集 m3u8：{m3u8}")
            download(drama_id, drama_name, ep, m3u8)
        except Exception as e:
            print(f"❌ 下载失败：{e}")
            sys.exit(1)
            
    elif len(sys.argv) == 4:
        # 判断是批量下载还是单集下载（带自定义名称）
        drama_id = sys.argv[1]
        try:
            # 尝试将第三个参数转换为整数
            start_ep = int(sys.argv[2])
            end_ep = int(sys.argv[3])
            
            # 批量下载模式（无自定义名称）
            print(f"获取电视剧信息...")
            drama_name = get_drama_name(drama_id)
            print(f"电视剧：{drama_name} (ID: {drama_id})")
            print(f"批量下载第{start_ep}集到第{end_ep}集")
            
            download_episodes(drama_id, drama_name, start_ep, end_ep)
        except ValueError:
            # 单集下载模式（带自定义名称）
            ep = int(sys.argv[2])
            custom_name = sys.argv[3]
            
            print(f"使用自定义名称：{custom_name} (ID: {drama_id})")
            
            try:
                m3u8 = get_m3u8_url(drama_id, ep)
                print(f"第{ep}集 m3u8：{m3u8}")
                download(drama_id, custom_name, ep, m3u8)
            except Exception as e:
                print(f"❌ 下载失败：{e}")
                sys.exit(1)
                
    elif len(sys.argv) == 5:
        # 批量下载模式（带自定义名称）
        drama_id = sys.argv[1]
        start_ep = int(sys.argv[2])
        end_ep = int(sys.argv[3])
        custom_name = sys.argv[4]
        
        print(f"使用自定义名称：{custom_name} (ID: {drama_id})")
        print(f"批量下载第{start_ep}集到第{end_ep}集")
        
        download_episodes(drama_id, custom_name, start_ep, end_ep)
    else:
        print_usage()
        sys.exit(1)