Huashu Douyin Script

#!/bin/bash
# douyin-viral-script skill 安装脚本
# 用法：bash <(curl -fsSL https://raw.githubusercontent.com/alchaincyf/Write-Prompt/master/.claude/skills/douyin-viral-script/install.sh)

set -e

SKILL_NAME="douyin-viral-script"
REPO="alchaincyf/Write-Prompt"
BRANCH="master"
SRC_PATH=".claude/skills/$SKILL_NAME"
TARGET_DIR="$HOME/.claude/skills/$SKILL_NAME"

echo "📦 安装 $SKILL_NAME skill..."
echo ""

# 检查依赖
check_dep() {
    if ! command -v "$1" &>/dev/null; then
        echo "❌ 需要 $1，请先安装：$2"
        exit 1
    fi
}

check_dep git "https://git-scm.com"
check_dep uv "curl -LsSf https://astral.sh/uv/install.sh | sh"

if ! command -v yt-dlp &>/dev/null; then
    echo "⚠️  未检测到 yt-dlp（下载抖音视频需要）"
    echo "   安装方式：brew install yt-dlp 或 pip install yt-dlp"
    echo ""
fi

# 创建目标目录
mkdir -p "$TARGET_DIR"

# 用临时目录做 sparse checkout
TMPDIR=$(mktemp -d)
trap "rm -rf $TMPDIR" EXIT

echo "⬇️  从 GitHub 下载 skill 文件..."
cd "$TMPDIR"
git init -q
git remote add origin "https://github.com/$REPO.git"
git config core.sparseCheckout true
echo "$SRC_PATH/" > .git/info/sparse-checkout
git pull -q origin "$BRANCH" --depth=1

# 复制到目标位置
if [ -d "$TMPDIR/$SRC_PATH" ]; then
    cp -R "$TMPDIR/$SRC_PATH/"* "$TARGET_DIR/"
    echo ""
    echo "✅ 安装成功！"
    echo ""
    echo "📁 安装位置：$TARGET_DIR"
    echo ""
    echo "📋 文件清单："
    find "$TARGET_DIR" -type f | while read f; do
        echo "   $(echo "$f" | sed "s|$TARGET_DIR/||")"
    done
    echo ""
    echo "🔧 使用前请确保："
    echo "   1. 设置 GEMINI_API_KEY 环境变量（用于视频分析）"
    echo "   2. Chrome 浏览器已登录抖音（用于视频下载）"
    echo ""
    echo "🚀 在 Claude Code 中输入 /douyin-viral-script 即可使用"
else
    echo "❌ 下载失败，请检查网络连接"
    exit 1
fi

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "google-genai>=1.0.0",
# ]
# ///
"""
Gemini 3 Pro/Flash video analysis tool.

Usage:
    uv run analyze_video.py --video "video.mp4" --prompt "summarize this video"
    uv run analyze_video.py --video "https://youtube.com/watch?v=ID" --prompt "describe key moments"
    uv run analyze_video.py --video "video.mp4" --prompt "generate SRT subtitles" --output subs.srt
"""

import argparse
import os
import sys
import time
from pathlib import Path


MODELS = {
    "pro": "gemini-3-pro-preview",
    "flash": "gemini-3-flash-preview",
}

RESOLUTIONS = {
    "low": "media_resolution_low",
    "medium": "media_resolution_medium",
    "high": "media_resolution_high",
}


def get_api_key(provided_key: str | None) -> str | None:
    if provided_key:
        return provided_key
    return os.environ.get("GEMINI_API_KEY")


def is_youtube_url(video_path: str) -> bool:
    return any(domain in video_path for domain in [
        "youtube.com", "youtu.be", "youtube-nocookie.com"
    ])


def is_url(video_path: str) -> bool:
    return video_path.startswith("http://") or video_path.startswith("https://")


def get_file_size_mb(file_path: str) -> float:
    return Path(file_path).stat().st_size / (1024 * 1024)


def upload_and_wait(client, file_path: str) -> object:
    """Upload video via File API and wait for processing."""
    print(f"Uploading: {file_path}", file=sys.stderr)
    video_file = client.files.upload(file=file_path)
    print(f"Upload complete: {video_file.name}", file=sys.stderr)

    while not video_file.state or video_file.state.name != "ACTIVE":
        print(f"  Processing... ({video_file.state})", file=sys.stderr)
        time.sleep(5)
        video_file = client.files.get(name=video_file.name)

    print("Video ready for analysis", file=sys.stderr)
    return video_file


def main():
    parser = argparse.ArgumentParser(
        description="Analyze video using Gemini 3 Pro/Flash"
    )
    parser.add_argument(
        "--video", "-v", required=True,
        help="Video file path or YouTube URL"
    )
    parser.add_argument(
        "--prompt", "-p", required=True,
        help="Analysis prompt"
    )
    parser.add_argument(
        "--model", "-m", choices=["pro", "flash"], default="pro",
        help="Model: pro (default, better) or flash (cheaper, has free tier)"
    )
    parser.add_argument(
        "--resolution", "-r", choices=["low", "medium", "high"], default="medium",
        help="Video resolution: low (save tokens), medium (default), high (read text)"
    )
    parser.add_argument(
        "--start", type=int, default=None,
        help="Start time in seconds (for clipping)"
    )
    parser.add_argument(
        "--end", type=int, default=None,
        help="End time in seconds (for clipping)"
    )
    parser.add_argument(
        "--fps", type=float, default=None,
        help="Custom FPS (default 1.0, use 0.5 for long videos)"
    )
    parser.add_argument(
        "--output", "-o", default=None,
        help="Save output to file (e.g., subtitles.srt)"
    )
    parser.add_argument(
        "--api-key", "-k", default=None,
        help="Gemini API key (overrides GEMINI_API_KEY env var)"
    )

    args = parser.parse_args()

    api_key = get_api_key(args.api_key)
    if not api_key:
        print("Error: No API key. Set GEMINI_API_KEY or use --api-key", file=sys.stderr)
        sys.exit(1)

    from google import genai
    from google.genai import types

    client = genai.Client(api_key=api_key)
    model_id = MODELS[args.model]

    # Build config
    config_kwargs = {}
    if args.resolution:
        config_kwargs["media_resolution"] = RESOLUTIONS[args.resolution]

    config = types.GenerateContentConfig(**config_kwargs) if config_kwargs else None

    # Build content parts
    parts = []

    if is_youtube_url(args.video):
        # YouTube URL - direct input
        print(f"Analyzing YouTube video: {args.video}", file=sys.stderr)
        part_kwargs = {
            "file_data": types.FileData(file_uri=args.video)
        }
        if args.start is not None or args.end is not None or args.fps is not None:
            video_meta = {}
            if args.start is not None:
                video_meta["start_offset"] = f"{args.start}s"
            if args.end is not None:
                video_meta["end_offset"] = f"{args.end}s"
            if args.fps is not None:
                video_meta["fps"] = args.fps
            part_kwargs["video_metadata"] = types.VideoMetadata(**video_meta)
        parts.append(types.Part(**part_kwargs))

    elif is_url(args.video):
        # Other URL
        print(f"Analyzing video URL: {args.video}", file=sys.stderr)
        parts.append(types.Part(
            file_data=types.FileData(file_uri=args.video)
        ))

    else:
        # Local file
        file_path = args.video
        if not Path(file_path).exists():
            print(f"Error: File not found: {file_path}", file=sys.stderr)
            sys.exit(1)

        file_size = get_file_size_mb(file_path)
        print(f"Video file: {file_path} ({file_size:.1f} MB)", file=sys.stderr)

        if file_size <= 20:
            # Inline upload for small files
            print("Using inline upload (< 20MB)", file=sys.stderr)
            video_bytes = open(file_path, "rb").read()

            # Detect MIME type
            ext = Path(file_path).suffix.lower()
            mime_map = {
                ".mp4": "video/mp4", ".mov": "video/mov",
                ".avi": "video/avi", ".webm": "video/webm",
                ".wmv": "video/wmv", ".mpg": "video/mpg",
                ".mpeg": "video/mpeg", ".flv": "video/x-flv",
                ".3gp": "video/3gpp",
            }
            mime_type = mime_map.get(ext, "video/mp4")

            parts.append(types.Part(
                inline_data=types.Blob(data=video_bytes, mime_type=mime_type)
            ))
        else:
            # File API for large files
            video_file = upload_and_wait(client, file_path)
            part_kwargs = {
                "file_data": types.FileData(file_uri=video_file.uri)
            }
            if args.start is not None or args.end is not None or args.fps is not None:
                video_meta = {}
                if args.start is not None:
                    video_meta["start_offset"] = f"{args.start}s"
                if args.end is not None:
                    video_meta["end_offset"] = f"{args.end}s"
                if args.fps is not None:
                    video_meta["fps"] = args.fps
                part_kwargs["video_metadata"] = types.VideoMetadata(**video_meta)
            parts.append(types.Part(**part_kwargs))

    # Add text prompt
    parts.append(types.Part(text=args.prompt))

    # Call API
    print(f"Analyzing with {model_id}...", file=sys.stderr)
    try:
        generate_kwargs = {
            "model": model_id,
            "contents": types.Content(parts=parts),
        }
        if config:
            generate_kwargs["config"] = config

        response = client.models.generate_content(**generate_kwargs)
        result = response.text

        # Output
        if args.output:
            output_path = Path(args.output)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(result, encoding="utf-8")
            print(f"\nResult saved to: {output_path.resolve()}", file=sys.stderr)
        else:
            print(result)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""
抖音视频下载工具。

使用 yt-dlp + cookies 下载无水印视频。
支持短链、分享链接、页面链接、从分享文本中提取URL。

Usage:
    uv run download_douyin.py --urls "https://v.douyin.com/xxx" "https://www.douyin.com/video/123"
    uv run download_douyin.py --urls "7.56 复制打开抖音，看看【某某的作品】 https://v.douyin.com/xxx"
    uv run download_douyin.py --output-dir /path/to/output
"""

import argparse
import json
import os
import re
import subprocess
import sys
from pathlib import Path


def clear_proxy_for_douyin():
    """清除代理环境变量，避免国内站点走代理导致失败。"""
    proxy_vars = ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY',
                  'all_proxy', 'ALL_PROXY']
    cleared = []
    for var in proxy_vars:
        if var in os.environ:
            del os.environ[var]
            cleared.append(var)
    if cleared:
        print(f"  已清除代理环境变量: {', '.join(cleared)}", file=sys.stderr)


def extract_douyin_urls(texts: list[str]) -> list[str]:
    """从文本列表中提取抖音URL。支持短链、页面链接、分享文本中的URL。"""
    url_pattern = re.compile(
        r'https?://(?:v\.douyin\.com|www\.douyin\.com|www\.iesdouyin\.com)[^\s\'"<>]*'
    )
    urls = []
    for text in texts:
        text = text.strip()
        # 如果本身就是URL
        if re.match(r'^https?://', text) and 'douyin.com' in text:
            urls.append(text.split('?')[0] if 'www.douyin.com/video/' in text else text)
        else:
            # 从分享文本中提取URL
            found = url_pattern.findall(text)
            urls.extend(found)
    return urls


def extract_video_id(url: str) -> str:
    """尝试从URL中提取视频ID。"""
    # 长链接格式：www.douyin.com/video/7xxxxxxxxx
    match = re.search(r'douyin\.com/video/(\d+)', url)
    if match:
        return match.group(1)
    # 短链接格式：v.douyin.com/xxx - 无法直接提取，用URL hash
    return re.sub(r'[^\w]', '', url)[-12:]


def download_video(url: str, output_dir: Path, index: int, cookies_browser: str) -> dict:
    """用 yt-dlp 下载单个抖音视频。"""
    video_id = extract_video_id(url)
    output_template = str(output_dir / f"douyin-{index}-{video_id}.%(ext)s")

    cmd = [
        "yt-dlp",
        "--cookies-from-browser", cookies_browser,
        "--output", output_template,
        "--no-warnings",
        "--proxy", "",  # 强制不使用代理（国内站点）
        "--no-check-certificates",
        url
    ]

    print(f"  [{index}] 下载中: {url}", file=sys.stderr)

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=180
        )

        if result.returncode != 0:
            error_msg = result.stderr.strip() or "未知错误"
            # 检查常见错误
            if "cookies" in error_msg.lower() or "login" in error_msg.lower():
                error_msg = f"需要登录Cookie: {error_msg[:200]}"
            elif "not found" in error_msg.lower() or "unavailable" in error_msg.lower():
                error_msg = f"视频不存在或已删除: {error_msg[:200]}"
            print(f"  [{index}] 下载失败: {error_msg}", file=sys.stderr)
            return {"url": url, "success": False, "error": error_msg}

        # 查找下载的文件
        downloaded = list(output_dir.glob(f"douyin-{index}-{video_id}.*"))
        video_files = [f for f in downloaded if f.suffix.lower() in ('.mp4', '.webm', '.mkv', '.flv')]

        if video_files:
            file_path = str(video_files[0])
            size_mb = video_files[0].stat().st_size / (1024 * 1024)
            print(f"  [{index}] 下载成功: {video_files[0].name} ({size_mb:.1f}MB)", file=sys.stderr)
            return {"url": url, "success": True, "path": file_path, "size_mb": round(size_mb, 1)}
        else:
            print(f"  [{index}] 下载完成但未找到视频文件", file=sys.stderr)
            return {"url": url, "success": False, "error": "下载完成但未找到视频文件"}

    except subprocess.TimeoutExpired:
        print(f"  [{index}] 下载超时（180秒）", file=sys.stderr)
        return {"url": url, "success": False, "error": "下载超时（180秒）"}
    except FileNotFoundError:
        print(f"  [{index}] 未找到yt-dlp，请安装: pip install yt-dlp 或 brew install yt-dlp", file=sys.stderr)
        return {"url": url, "success": False, "error": "未安装yt-dlp"}


def main():
    parser = argparse.ArgumentParser(description="抖音视频下载工具")
    parser.add_argument(
        "--urls", "-u", nargs="+", required=True,
        help="抖音视频URL（1-5个），支持短链、页面链接、分享文本"
    )
    parser.add_argument(
        "--output-dir", "-o",
        default=str(Path.cwd() / "_temp" / "douyin-downloads"),
        help="输出目录（默认：当前目录/_temp/douyin-downloads/）"
    )
    parser.add_argument(
        "--cookies-browser", "-c", default="chrome",
        help="从哪个浏览器提取cookies（默认：chrome）"
    )

    args = parser.parse_args()

    # 提取URL
    urls = extract_douyin_urls(args.urls)
    if not urls:
        print("错误：未从输入中提取到有效的抖音URL", file=sys.stderr)
        print("支持的格式：", file=sys.stderr)
        print("  - https://v.douyin.com/xxx（短链）", file=sys.stderr)
        print("  - https://www.douyin.com/video/123（页面链接）", file=sys.stderr)
        print("  - 分享文本（包含URL的文本）", file=sys.stderr)
        sys.exit(1)

    if len(urls) > 5:
        print(f"警告：输入了{len(urls)}个URL，仅处理前5个", file=sys.stderr)
        urls = urls[:5]

    # 清除代理（国内站点不走代理）
    clear_proxy_for_douyin()

    # 创建输出目录
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    print(f"准备下载 {len(urls)} 个视频到 {output_dir}", file=sys.stderr)
    print(f"使用 {args.cookies_browser} 浏览器Cookie", file=sys.stderr)
    print("", file=sys.stderr)

    # 逐个下载
    results = []
    for i, url in enumerate(urls, 1):
        result = download_video(url, output_dir, i, args.cookies_browser)
        results.append(result)

    # 汇总结果
    success_count = sum(1 for r in results if r["success"])
    fail_count = len(results) - success_count

    print("", file=sys.stderr)
    print(f"下载完成：{success_count} 成功，{fail_count} 失败", file=sys.stderr)

    if fail_count > 0:
        print("", file=sys.stderr)
        print("下载失败的视频可以尝试：", file=sys.stderr)
        print("  1. 确认Chrome浏览器已登录抖音", file=sys.stderr)
        print("  2. 使用 --cookies-browser edge/firefox 切换浏览器", file=sys.stderr)
        print("  3. 手动下载视频后放入输出目录", file=sys.stderr)

    # JSON输出（供调用方解析）
    output = {
        "total": len(results),
        "success": success_count,
        "failed": fail_count,
        "output_dir": str(output_dir),
        "results": results
    }
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()