ComfyUI用の、WANVideoやCogVideoXなどのローカル動画生成AIのモデルの一括ダウンロード法

ComfyUI用の、WANVideoやCogVideoXなどのローカル動画生成AIのモデルの一括ダウンロード法

ComfyUIのフレームワークで動作するローカル動画生成AIを実験する場合、多くのモデルをダウンロードする必要があり、そのための待ち時間が長いので、出かける前や寝る前にしかけて一気にまとめてダウンロードするのが吉。
それを一気に行うpython用のコードをClaudeで生成した。

import os
import requests
import re
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

def get_files_list(repo_url):
    """
    HuggingFaceリポジトリから.safetensorsファイルのURLリストを取得します
    """
    # APIエンドポイントのURLを作成
    api_url = repo_url.replace('https://huggingface.co/', 'https://huggingface.co/api/repos/').rstrip('/') + '/refs/main/tree'
    
    # APIリクエスト
    response = requests.get(api_url)
    if response.status_code != 200:
        raise Exception(f"APIリクエストエラー: {response.status_code}")
    
    repo_files = response.json()
    
    # .safetensorsファイルのみをフィルタリング
    safetensor_files = []
    
    def process_item(item):
        # ディレクトリの場合は再帰的に取得
        if item['type'] == 'directory':
            subdir_url = f"{api_url}/{item['path']}"
            subdir_response = requests.get(subdir_url)
            if subdir_response.status_code == 200:
                return process_items(subdir_response.json())
        # .safetensorsファイルの場合はリストに追加
        elif item['path'].endswith('.safetensors'):
            return [{
                'filename': os.path.basename(item['path']),
                'path': item['path'],
                'download_url': f"https://huggingface.co/{repo_url.split('https://huggingface.co/')[1]}/resolve/main/{item['path']}"
            }]
        return []
    
    def process_items(items):
        result = []
        for item in items:
            result.extend(process_item(item))
        return result
    
    safetensor_files = process_items(repo_files)
    
    return safetensor_files

def download_file(file_info, output_dir, max_retries=3):
    """
    ファイルをダウンロードする関数
    """
    filename = file_info['filename']
    download_url = file_info['download_url']
    output_path = os.path.join(output_dir, filename)
    
    # 既に存在するファイルはスキップ
    if os.path.exists(output_path):
        print(f"ファイルはすでに存在します: {filename}")
        return True
    
    for attempt in range(max_retries):
        try:
            response = requests.get(download_url, stream=True)
            response.raise_for_status()
            
            total_size = int(response.headers.get('content-length', 0))
            
            with open(output_path, 'wb') as file, tqdm(
                    desc=filename,
                    total=total_size,
                    unit='B',
                    unit_scale=True,
                    unit_divisor=1024,
            ) as progress_bar:
                for chunk in response.iter_content(chunk_size=1024*1024):
                    file.write(chunk)
                    progress_bar.update(len(chunk))
            
            return True
            
        except Exception as e:
            print(f"ダウンロード中にエラーが発生しました ({attempt+1}/{max_retries}): {filename} - {str(e)}")
            if attempt == max_retries - 1:
                print(f"ファイルのダウンロードに失敗しました: {filename}")
                return False

def main():
    # ダウンロード対象のリポジトリURL
    repo_url = "https://huggingface.co/Kijai/WanVideo_comfy/tree/main"
    
    # 出力ディレクトリ
    output_dir = "downloaded_safetensors"
    os.makedirs(output_dir, exist_ok=True)
    
    print(f"リポジトリからファイルリストを取得中: {repo_url}")
    files_to_download = get_files_list(repo_url)
    
    print(f"ダウンロードするファイル数: {len(files_to_download)}")
    
    # 複数スレッドでダウンロードを実行
    successful_downloads = 0
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(
            lambda file_info: download_file(file_info, output_dir),
            files_to_download
        ))
        successful_downloads = sum(results)
    
    print(f"ダウンロード完了: {successful_downloads}/{len(files_to_download)} ファイル")

if __name__ == "__main__":
    main()

WANVideoの場合は、上のコードの以下の部分がWANVideoのダウンロードもとを指定している。

   # ダウンロード対象のリポジトリURL
    repo_url = "https://huggingface.co/Kijai/WanVideo_comfy/tree/main"

ComfyUIの動作する環境で、上のコードをコピペしたファイルがgetmodels.pyとすると

python getmodels.py

でダウンロードが始まる。
モデルデータはでかいので、ダウンロードする場所の要領に注意

このコードはよくできていて、CogVideoX1.5-5B-SAT 関連のモデルをダウンロードしようと
以下のようのしたが、これは間違いでエラーがでるが代替の方法が示唆される

代替ダウンロード方法でエラーが発生しました: Command '['huggingface-cli', 'download', 'THUDM/CogVideoX1.5-5B-SAT/t5-v1_1-xxl', '--include=*.safetensors', '--local-dir', 'downloaded_safetensors']' returned non-zero exit status 1.
すべてのダウンロード方法が失敗しました。

代替方法2: 以下のコマンドを直接実行してみてください:
pip install huggingface_hub
python -c "from huggingface_hub import snapshot_download; snapshot_download('Kijai/WanVideo_comfy', local_dir='downloaded_safetensors', allow_patterns=['*.safetensors'])"

それにしたがって、snapshot_download('Kijai/WanVideo_comfy' のクォーテーション内をHuggingFaceで検索したレポジトリ名に入れ替えて

python -c "from huggingface_hub import snapshot_download; snapshot_download('THUDM/CogVideoX1.5-5B-SAT', local_dir='downloaded_safetensors', allow_patterns=['*.safetensors'])"

と実行することで、CogVideoX1.5-5B-SAT　を無事ダウンロードできた。