영상 생성하고 파일 받기

import getpass
import json
import os
import sys
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen
 
BASE = "https://gw.letsur.ai"
RECORD = Path("video-job.json")
key = os.environ.get("LETSUR_API_KEY") or getpass.getpass("AI 게이트웨이 API 키: ")
if not key.strip():
    raise SystemExit("API 키가 필요합니다.")
 
 
def api(method, path, body=None):
    headers = {"Authorization": "Bearer " + key}
    data = None
    if body is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body).encode("utf-8")
    request = Request(BASE + path, data=data, headers=headers, method=method)
    try:
        with urlopen(request, timeout=60) as response:
            return json.load(response)
    except HTTPError as error:
        try:
            payload = json.load(error)
        except (ValueError, OSError):
            payload = {}
        # source_url 등 민감한 값을 통째로 출력하지 않습니다.
        problem = payload.get("error", payload)
        code = problem.get("code", problem.get("type", "unknown"))
        retry_after = error.headers.get("Retry-After")
        if retry_after:
            print("다음 조회 또는 링크 요청까지 기다릴 시간:", retry_after)
        raise SystemExit(f"HTTP {error.code}, 오류: {code}. 새 생성을 자동 재시도하지 않습니다.")
 
 
def current_job():
    record = json.loads(RECORD.read_text())
    job_id = record["id"]
    return api("GET", "/v1/jobs/" + quote(job_id, safe=""))["data"]["job"]
 
 
command = sys.argv[1] if len(sys.argv) == 2 else ""
if command == "submit":
    # 기존 작업 ID가 있으면 덮어쓰거나 새 요청을 보내지 않습니다.
    with RECORD.open("x") as saved:
        saved.write('{}')
    accepted = api("POST", "/v1/video/generations", {
        "model": "byteplus-seedance-2.5",
        "prompt": "바람에 천천히 흔들리는 초원의 풀을 고정 카메라로 촬영한 영상",
        "duration": 5,
        "resolution": "480p",
        "aspect_ratio": "16:9",
        "generate_audio": True,
    })
    RECORD.write_text(json.dumps({"id": accepted["id"]}))
    print("접수한 작업 ID:", accepted["id"])
elif command == "status":
    job = current_job()
    print("작업:", job["id"], "상태:", job["status"])
    for media in job.get("result", {}).get("media", []):
        print("파일:", media["media_id"], media["kind"], media["state"])
        if media.get("url_unavailable_reason"):
            print("파일 링크를 사용할 수 없는 이유:", media["url_unavailable_reason"])
    if "usage" in job:
        print("작업 비용:", json.dumps(job["usage"], ensure_ascii=False))
    else:
        print("작업 비용이 아직 응답에 없습니다.")
    if "error" in job:
        print("작업 오류 유형:", job["error"]["type"])
elif command == "download":
    job = current_job()
    if job["status"] == "failed":
        raise SystemExit("작업이 실패했습니다. status로 같은 작업 ID의 오류 유형을 확인하세요.")
    if job["status"] != "succeeded":
        raise SystemExit("작업이 대기 또는 처리 중입니다. 잠시 뒤 status로 같은 작업 ID를 다시 확인하세요.")
    videos = [m for m in job.get("result", {}).get("media", []) if m["kind"] == "video"]
    if not videos:
        raise SystemExit("영상 결과가 없습니다. status로 작업 상태를 확인하세요.")
    if any(media.get("state") == "unavailable" for media in videos):
        raise SystemExit("영상 파일을 지금 사용할 수 없습니다. status로 같은 작업 ID의 파일 상태와 이유를 확인하고 필요하면 문의하세요.")
    if any(media.get("state") != "ready" for media in videos):
        raise SystemExit("영상 파일이 준비 중입니다. 잠시 뒤 status로 같은 작업 ID의 파일 상태를 다시 확인하세요.")
    for index, media in enumerate(videos, 1):
        link = api("POST", "/v1/media/" + quote(media["media_id"], safe="") + "/link")
        output = Path(f"video-{index}.mp4")
        # 다운로드 URL에는 API 키 헤더를 보내지 않습니다.
        with urlopen(link["url"], timeout=60) as response, output.open("xb") as target:
            while chunk := response.read(1024 * 1024):
                target.write(chunk)
        print("다운로드 완료:", output, output.stat().st_size, "bytes")
else:
    raise SystemExit("사용법: python3 video_job.py submit|status|download")