79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
"""Higgsfield CLI 래퍼.
|
|
|
|
인증이 `~/.config/higgsfield/credentials.json`의 CLI 토큰이라 서브프로세스를 피할 수 없고,
|
|
머신마다 사람이 한 번 로그인해야 한다:
|
|
`npm i -g @higgsfield/cli` → `higgsfield auth login` → `higgsfield workspace set <id>`
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import re
|
|
|
|
CLI = "higgsfield"
|
|
CLI_TIMEOUT = 2100
|
|
|
|
# Higgsfield가 제공하는 모델 식별자. 어느 단계가 무엇을 쓸지는 그 단계가 정한다.
|
|
KLING_3_0 = "kling3_0"
|
|
VEO_3_1 = "veo3_1"
|
|
TOPAZ_IMAGE = "topaz_image"
|
|
TOPAZ_TEXT_REFINE = "Text Refine"
|
|
|
|
RESULT_URL_PATTERN = re.compile(r'"result_url"\s*:\s*"([^"]+)"')
|
|
CREDITS_PATTERN = re.compile(r'"credits(?:_exact)?"\s*:\s*([0-9.]+)')
|
|
PLAIN_CREDITS_PATTERN = re.compile(r"([0-9.]+)\s*credits")
|
|
|
|
|
|
async def run_cli(*args: str, timeout: int = CLI_TIMEOUT) -> str:
|
|
try:
|
|
process = await asyncio.create_subprocess_exec(
|
|
CLI, *args,
|
|
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
|
except FileNotFoundError:
|
|
raise RuntimeError(
|
|
"higgsfield CLI가 없다 — `npm i -g @higgsfield/cli` 후 `higgsfield auth login`")
|
|
try:
|
|
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout)
|
|
except TimeoutError:
|
|
process.kill()
|
|
raise RuntimeError(f"higgsfield {args[0]} 타임아웃 {timeout}초")
|
|
if process.returncode != 0:
|
|
detail = (stderr or stdout).decode(errors="ignore")[-600:]
|
|
raise RuntimeError(f"higgsfield {args[0]} 실패:\n{detail}")
|
|
return stdout.decode(errors="ignore")
|
|
|
|
|
|
async def account_status() -> dict:
|
|
output = await run_cli("account", "status", "--json", timeout=60)
|
|
try:
|
|
return json.loads(output)
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError(f"account status 응답을 파싱할 수 없다:\n{output[-400:]}")
|
|
|
|
|
|
async def assert_account(expected: str) -> dict:
|
|
"""다른 계정으로 로그인하면 토큰이 조용히 바뀌어 그 계정 크레딧이 빠지므로,
|
|
편당 22크레딧을 쓰기 직전에 로그인 계정을 확인하고 어긋나면 죽는다."""
|
|
status = await account_status()
|
|
email = (status.get("email") or "").strip().lower()
|
|
if email != expected.lower():
|
|
raise RuntimeError(
|
|
f"[가드] Higgsfield 계정 불일치 — 로그인: {email or '(없음)'} / 기대: {expected}\n"
|
|
f"크레딧이 다른 계정에서 빠진다. logout 후 {expected}로 다시 로그인할 것. "
|
|
f"의도한 전환이면 HIGGSFIELD_ACCOUNT={email} 로 명시할 것.")
|
|
return status
|
|
|
|
|
|
def parse_credits(output: str) -> float:
|
|
"""못 찾으면 죽는다. 0.0으로 때우면 CLI 출력 포맷이 바뀌어도 "0 크레딧 썼다"로 통과한다."""
|
|
match = CREDITS_PATTERN.search(output) or PLAIN_CREDITS_PATTERN.search(output)
|
|
if not match:
|
|
raise RuntimeError(f"CLI 출력에서 크레딧을 못 읽었다 — 포맷이 바뀌었는지 확인할 것:\n"
|
|
f"{output[-600:]}")
|
|
return float(match.group(1))
|
|
|
|
|
|
def parse_result_url(output: str) -> str:
|
|
match = RESULT_URL_PATTERN.search(output)
|
|
if not match:
|
|
raise RuntimeError(f"result_url 없음 — 생성 실패:\n{output[-1200:]}")
|
|
return match.group(1)
|