deploy_space.py

import argparse
from pathlib import Path

from huggingface_hub import HfApi
from huggingface_hub.errors import HfHubHTTPError


DEFAULT_SPACE_REPO_ID = "bykhan/wan22-i2v-endpoint"
DEFAULT_HARDWARE = "t4-small"
DEFAULT_SLEEP_TIME = 60


def read_env_file(file_path: Path) -> dict[str, str]:
    values: dict[str, str] = {}

    for raw_line in file_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue

        key, value = line.split("=", 1)
        values[key.strip()] = value.strip()

    return values


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Create or update a Hugging Face Docker Space for the Wan backend.")
    parser.add_argument("--repo-id", default=DEFAULT_SPACE_REPO_ID)
    parser.add_argument("--hardware", default=DEFAULT_HARDWARE)
    parser.add_argument("--sleep-time", type=int, default=DEFAULT_SLEEP_TIME)
    parser.add_argument("--private", action="store_true")
    parser.add_argument("--skip-hardware-request", action="store_true")
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()


def build_space_url(repo_id: str) -> str:
    owner, name = repo_id.split("/", 1)
    return f"https://{owner}-{name}.hf.space"


def main() -> int:
    args = parse_args()
    project_root = Path(__file__).resolve().parents[1]
    env_values = read_env_file(project_root / ".env")
    token = env_values.get("HF_TOKEN", "")

    if not token:
        raise SystemExit(".env icinde HF_TOKEN bulunamadi.")

    space_variables = [
        {"key": "WAN_BACKEND", "value": "wan-vace-gguf"},
        {"key": "WAN_VACE_CONFIG_REPO", "value": "Wan-AI/Wan2.1-VACE-1.3B-diffusers"},
        {"key": "WAN_VACE_GGUF_REPO", "value": "bykhan/wan-1.3b-gguf"},
        {"key": "WAN_VACE_GGUF_FILE", "value": "wan2.1-vace-1.3b-q4_0.gguf"},
        {"key": "WAN_TEXT_ENCODER_REPO", "value": "bykhan/wan-1.3b-gguf"},
        {"key": "WAN_TEXT_ENCODER_GGUF_FILE", "value": "umt5-xxl-encoder-q4_k_m.gguf"},
        {"key": "WAN_ENABLE_CPU_OFFLOAD", "value": "1"},
        {"key": "WAN_ENABLE_VAE_TILING", "value": "1"},
        {"key": "WAN_FLOW_SHIFT", "value": "3.0"},
        {"key": "WAN_DEFAULT_HEIGHT", "value": "480"},
        {"key": "WAN_DEFAULT_WIDTH", "value": "832"},
        {"key": "WAN_DEFAULT_NUM_FRAMES", "value": "81"},
        {"key": "WAN_DEFAULT_NUM_STEPS", "value": "30"},
        {"key": "WAN_DEFAULT_GUIDANCE", "value": "5.0"},
        {"key": "WAN_DEFAULT_CONDITIONING_SCALE", "value": "1.0"},
        {"key": "WAN_DEFAULT_FPS", "value": "16"},
        {"key": "WAN_RESULTS_REPO_ID", "value": "bykhan/generated-videos"},
        {"key": "WAN_RESULTS_REPO_TYPE", "value": "dataset"},
        {"key": "WAN_RESULTS_PATH_PREFIX", "value": "generated-videos"},
    ]
    space_secrets = [{"key": "HF_TOKEN", "value": token}]

    if args.dry_run:
        print(f"repo_id={args.repo_id}")
        print(f"hardware={args.hardware}")
        print(f"sleep_time={args.sleep_time}")
        print(f"private={args.private}")
        print(f"skip_hardware_request={args.skip_hardware_request}")
        print(f"space_url={build_space_url(args.repo_id)}")
        return 0

    api = HfApi(token=token)
    api.create_repo(
        repo_id=args.repo_id,
        repo_type="space",
        space_sdk="docker",
        private=args.private,
        exist_ok=True,
        space_secrets=space_secrets,
        space_variables=space_variables,
        space_sleep_time=args.sleep_time,
    )

    api.upload_folder(
        repo_id=args.repo_id,
        repo_type="space",
        folder_path=Path(__file__).resolve().parent,
        commit_message="Deploy Wan custom backend",
        ignore_patterns=[".venv/*", "__pycache__/*", "*.pyc", ".env", ".env.*"],
    )

    runtime = api.get_space_runtime(args.repo_id)

    if not args.skip_hardware_request:
        try:
            runtime = api.request_space_hardware(
                repo_id=args.repo_id,
                hardware=args.hardware,
                sleep_time=args.sleep_time,
            )
        except HfHubHTTPError as error:
            print(f"space_url={build_space_url(args.repo_id)}")
            print(f"stage={runtime.stage}")
            if "402" in str(error):
                print("hardware_request=blocked")
                print("reason=paid_hardware_credits_required")
                print("billing_url=https://huggingface.co/settings/billing")
                return 2
            raise

    print(f"space_url={build_space_url(args.repo_id)}")
    print(f"stage={runtime.stage}")
    print(f"hardware={getattr(runtime, 'hardware', None)}")
    return 0


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