Universal Media Downloader

MIT License

Copyright (c) 2025 杨卫薪律师（微信ywxlaw）

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""抖音图文笔记下载器

下载抖音图文笔记中的所有图片。

使用方法:
    python download_douyin_note.py "<url>" [--out-dir <directory>]

示例:
    python download_douyin_note.py "https://v.douyin.com/xxx/"
    python download_douyin_note.py "https://www.douyin.com/note/xxx/" --out-dir ~/Downloads/douyin
"""

import argparse
import os
import re
import json
import requests
from pathlib import Path
from urllib.parse import urlparse, unquote

SCRIPT_DIR = Path(__file__).parent.parent
DEFAULT_OUT_DIR = SCRIPT_DIR / "downloads" / "douyin_notes"

# 请求头
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.38',
    'Referer': 'https://www.douyin.com/',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
}


def expand_short_url(url: str) -> str:
    """展开短链接"""
    try:
        resp = requests.head(url, headers=HEADERS, allow_redirects=True, timeout=10)
        return resp.url
    except:
        return url


def extract_note_id(url: str) -> str:
    """从 URL 中提取笔记 ID"""
    # 展开短链接
    full_url = expand_short_url(url)
    
    # 匹配各种格式的笔记 ID
    patterns = [
        r'/note/(\d+)',
        r'/share/note/(\d+)',
        r'modal_id=(\d+)',
        r'/video/(\d+)',  # 有些图文笔记也可能是 video 格式
    ]
    
    for pattern in patterns:
        match = re.search(pattern, full_url)
        if match:
            return match.group(1)
    
    return None


def get_note_data(note_id: str) -> dict:
    """获取笔记数据（通过模拟 API 请求）"""
    # 尝试通过移动端 API 获取数据
    api_url = f"https://www.iesdouyin.com/web/api/v2/aweme/iteminfo/?item_ids={note_id}"
    
    try:
        resp = requests.get(api_url, headers=HEADERS, timeout=10)
        data = resp.json()
        
        if data.get('status_code') == 0:
            return data.get('item_list', [{}])[0]
    except Exception as e:
        print(f"[warning] API 请求失败: {e}")
    
    return None


def extract_images_from_html(url: str) -> list:
    """从 HTML 页面提取图片 URL"""
    try:
        resp = requests.get(url, headers=HEADERS, timeout=15)
        html = resp.text
        
        # 尝试从页面中提取图片数据
        # 抖音页面中通常有 <script>__RENDER_DATA__=xxx</script>
        match = re.search(r'__RENDER_DATA__\s*=\s*([^<]+)</script>', html)
        if match:
            try:
                render_data = json.loads(unquote(match.group(1)))
                # 解析 render_data 获取图片列表
                # 这里需要根据实际数据结构调整
                images = []
                
                # 尝试不同的数据路径
                paths = [
                    ('app', 'video', 'images'),
                    ('app', 'aweme', 'detail', 'aweme', 'images'),
                    ('aweme_detail', 'images'),
                ]
                
                for path in paths:
                    data = render_data
                    for key in path:
                        if isinstance(data, dict) and key in data:
                            data = data[key]
                        else:
                            data = None
                            break
                    
                    if data and isinstance(data, list):
                        for img in data:
                            if isinstance(img, dict):
                                url_list = img.get('url_list', [])
                                if url_list:
                                    images.append(url_list[0])
                            elif isinstance(img, str):
                                images.append(img)
                
                if images:
                    return images
            except json.JSONDecodeError:
                pass
        
        # 备用：直接匹配图片 URL
        img_pattern = r'https?://[^"\'>\s]+\.(?:jpg|jpeg|png|webp|heic)'
        images = list(set(re.findall(img_pattern, html)))
        
        # 过滤掉头像等非内容图片
        images = [img for img in images if 'p26-' in img or 'p3-' in img or 'p9-' in img]
        
        return images[:20]  # 最多返回 20 张
        
    except Exception as e:
        print(f"[error] 提取图片失败: {e}")
        return []


def download_images(images: list, out_dir: Path, title: str = "douyin_note") -> list:
    """下载所有图片"""
    downloaded = []
    
    # 清理标题
    safe_title = re.sub(r'[^\w\s-]', '', title)[:50]
    safe_title = re.sub(r'[-\s]+', '_', safe_title)
    
    for i, img_url in enumerate(images, 1):
        try:
            # 确定扩展名
            ext = '.jpg'
            if '.png' in img_url:
                ext = '.png'
            elif '.webp' in img_url:
                ext = '.webp'
            elif '.heic' in img_url:
                ext = '.heic'
            
            # 下载图片
            resp = requests.get(img_url, headers=HEADERS, timeout=30)
            
            if resp.status_code == 200:
                filename = f"{safe_title}_{i:02d}{ext}"
                filepath = out_dir / filename
                
                with open(filepath, 'wb') as f:
                    f.write(resp.content)
                
                downloaded.append(str(filepath))
                print(f"[downloaded] {filename}")
            else:
                print(f"[warning] 图片 {i} 下载失败: HTTP {resp.status_code}")
                
        except Exception as e:
            print(f"[warning] 图片 {i} 下载失败: {e}")
    
    return downloaded


def main():
    parser = argparse.ArgumentParser(description='抖音图文笔记下载器')
    parser.add_argument('url', help='抖音图文笔记链接')
    parser.add_argument('--out-dir', default=str(DEFAULT_OUT_DIR), help='输出目录')
    parser.add_argument('--title', default='', help='自定义标题')
    
    args = parser.parse_args()
    
    # 创建输出目录
    out_dir = Path(args.out_dir).expanduser().resolve()
    out_dir.mkdir(parents=True, exist_ok=True)
    
    url = args.url
    print(f"[info] 处理链接: {url}")
    
    # 展开短链接
    full_url = expand_short_url(url)
    print(f"[info] 完整链接: {full_url}")
    
    # 检查是否是图文笔记
    if '/note/' not in full_url and '/share/note/' not in full_url:
        print("[warning] 这可能不是图文笔记链接，可能需要使用 universal-media-downloader 处理视频")
    
    # 提取图片
    print("[info] 正在提取图片...")
    images = extract_images_from_html(full_url)
    
    if not images:
        print("[error] 未能提取到任何图片")
        print("[提示] 可以尝试：")
        print("  1. 手动打开链接截图保存")
        print("  2. 使用浏览器开发者工具查看图片 URL")
        return 1
    
    print(f"[info] 找到 {len(images)} 张图片")
    
    # 下载图片
    print("[info] 开始下载...")
    title = args.title or "douyin_note"
    downloaded = download_images(images, out_dir, title)
    
    print(f"\n[完成] 成功下载 {len(downloaded)} 张图片到: {out_dir}")
    
    # 输出文件路径（供 AI 读取）
    for path in downloaded:
        print(f"SAVED_FILEPATH={path}")
    
    return 0


if __name__ == '__main__':
    exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Universal media downloader for video sites and podcasts.

This script is designed to be called by AnyGen as a deterministic helper.
It relies on yt-dlp, which supports many video sites (YouTube/Bilibili/Douyin, etc.)
AND can often download podcast audio when the page exposes a direct audio URL.

Examples:
  # Download best video+audio (mp4)
  python scripts/download_media.py "https://www.youtube.com/watch?v=xxxx"

  # Download audio only (mp3)
  python scripts/download_media.py --audio-only --audio-format mp3 "https://www.xiaoyuzhoufm.com/episode/xxxx"

  # Download with subtitles (auto-detect all available)
  python scripts/download_media.py --subtitles "https://www.youtube.com/watch?v=xxxx"

  # Download with Chinese subtitles only
  python scripts/download_media.py --subtitles --sub-lang "zh" "https://www.youtube.com/watch?v=xxxx"

  # Use cookies when needed (e.g., Douyin/Bilibili/YouTube age-gated)
  python scripts/download_media.py --cookies "/path/to/cookies.txt" "<URL>"

Output:
  - Prints the final saved filepath (absolute) as the last line: SAVED_FILEPATH=<path>
"""

from __future__ import annotations

import argparse
import os
import shlex
import subprocess
import sys
from pathlib import Path

# Default output directory: relative to skill root
SCRIPT_DIR = Path(__file__).parent.parent
DEFAULT_OUT_DIR = SCRIPT_DIR / "downloads"


def run(cmd: list[str]) -> tuple[int, str]:
    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    return p.returncode, p.stdout


def main() -> int:
    parser = argparse.ArgumentParser(description="Download video/audio by URL using yt-dlp")
    parser.add_argument("url", help="Video or podcast episode URL")

    parser.add_argument(
        "--out-dir",
        default=str(DEFAULT_OUT_DIR),
        help=f"Output directory (default: {DEFAULT_OUT_DIR})",
    )
    parser.add_argument(
        "--cookies",
        default="",
        help="Path to cookies.txt (Netscape format). Useful when 403/login required.",
    )
    parser.add_argument(
        "--proxy",
        default="",
        help="Proxy URL, e.g. socks5://127.0.0.1:7890 or http://127.0.0.1:7890",
    )

    parser.add_argument(
        "--audio-only",
        action="store_true",
        help="Extract audio only (no video). For podcasts, this is usually what you want.",
    )
    parser.add_argument(
        "--audio-format",
        default="mp3",
        help="Audio format when --audio-only is set (default: mp3)",
    )

    parser.add_argument(
        "--subtitles",
        action="store_true",
        help="Download subtitles if available.",
    )
    parser.add_argument(
        "--sub-lang",
        default="all",
        help="Subtitle language(s) to download (default: all, e.g., 'zh,en')",
    )

    args = parser.parse_args()

    out_dir = Path(args.out_dir).expanduser().resolve()
    out_dir.mkdir(parents=True, exist_ok=True)

    # Output template: keep title short + id to avoid collisions
    outtmpl = str(out_dir / "%(title).120B_%(id)s.%(ext)s")

    cmd = [
        "yt-dlp",
        "--no-playlist",
        "--js-runtimes",
        "node",
        "-o",
        outtmpl,
        "--print",
        "after_move:filepath",
    ]

    if args.cookies:
        cookies_path = Path(args.cookies).expanduser().resolve()
        if not cookies_path.exists():
            raise FileNotFoundError(f"cookies file not found: {cookies_path}")
        cmd.extend(["--cookies", str(cookies_path)])

    if args.proxy:
        cmd.extend(["--proxy", args.proxy])

    if args.audio_only:
        cmd.extend([
            "--extract-audio",
            "--audio-format",
            args.audio_format,
            "--audio-quality",
            "0",
        ])
    else:
        # Prefer mp4 for convenience (may still be webm depending on source)
        cmd.extend(["--merge-output-format", "mp4"])

    if args.subtitles:
        cmd.extend([
            "--write-sub",
            "--sub-langs",
            args.sub_lang,
        ])

    cmd.append(args.url)

    rc, out = run(cmd)

    # Echo full yt-dlp output for debugging.
    sys.stdout.write(out)

    if rc != 0:
        sys.stderr.write("\n[download_media] yt-dlp failed. Command was:\n")
        sys.stderr.write("  " + " ".join(shlex.quote(c) for c in cmd) + "\n")
        return rc

    # yt-dlp prints the moved file path via --print after_move:filepath.
    # Usually the last non-empty line.
    lines = [ln.strip() for ln in out.splitlines() if ln.strip()]
    saved = ""
    if lines:
        cand = lines[-1]
        if os.path.exists(cand):
            saved = cand

    if saved:
        sys.stdout.write(f"\nSAVED_FILEPATH={saved}\n")
    else:
        # Fallback: pick newest file in out_dir
        files = sorted([p for p in out_dir.glob('*') if p.is_file()], key=lambda p: p.stat().st_mtime, reverse=True)
        if files:
            saved = str(files[0])
            sys.stdout.write(f"\nSAVED_FILEPATH={saved}\n")
        else:
            sys.stdout.write("\nSAVED_FILEPATH=\n")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())