102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""썰박스 생성 잡 매니저 (subprocess → 내장 generator 엔진).
|
||
|
||
편당 2~4분·과금이 크므로 동시 실행 제한. stdout 진행 마커([1/4]…, "완성: ...mp4")를 파싱.
|
||
DB 작업은 하지 않는다 — 완료 후 처리(content 삽입·크레딧 차감)는 잡 상태 조회 엔드포인트가 1회 수행.
|
||
"""
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import uuid
|
||
from collections import deque
|
||
from datetime import datetime
|
||
|
||
from .config import GENERATOR_DIR, SCENARIO_ENGINE, GEMINI_API_KEY, settings
|
||
|
||
_jobs: dict[str, dict] = {}
|
||
_queue: deque[str] = deque()
|
||
_lock = threading.Lock()
|
||
_running = 0
|
||
|
||
_STEP_RE = re.compile(r"\[(\d)\s*/\s*4\]")
|
||
_DONE_RE = re.compile(r"완성[:：]\s*(.+\.mp4)")
|
||
STEP_NAMES = ["대본 생성", "스토리보드", "이미지·음성", "영상 합성"]
|
||
|
||
|
||
def create_job(scenario: str, input_text: str, scenes: int, seconds: int, owner_uuid: str | None) -> dict:
|
||
jid = uuid.uuid4().hex[:12]
|
||
_jobs[jid] = {
|
||
"id": jid, "scenario": scenario, "input": input_text,
|
||
"scenes": scenes, "seconds": seconds, "owner_uuid": owner_uuid,
|
||
"status": "queued", "step": 0, "log": [], "output": None, "error": None,
|
||
"finalized": False, "created": datetime.now().isoformat(timespec="seconds"),
|
||
}
|
||
with _lock:
|
||
_queue.append(jid)
|
||
_pump()
|
||
return _jobs[jid]
|
||
|
||
|
||
def get_job(jid: str) -> dict | None:
|
||
return _jobs.get(jid)
|
||
|
||
|
||
def _pump():
|
||
global _running
|
||
with _lock:
|
||
while _queue and _running < settings.MAX_CONCURRENT_JOBS:
|
||
jid = _queue.popleft()
|
||
_running += 1
|
||
threading.Thread(target=_run, args=(jid,), daemon=True).start()
|
||
|
||
|
||
def _run(jid: str):
|
||
global _running
|
||
job = _jobs[jid]
|
||
try:
|
||
engine = SCENARIO_ENGINE[job["scenario"]]
|
||
engine_dir = GENERATOR_DIR / engine
|
||
main_py = engine_dir / "main.py"
|
||
if not main_py.exists():
|
||
raise FileNotFoundError(f"generator not found: {main_py}")
|
||
|
||
cmd = [sys.executable, "-u", str(main_py), job["input"],
|
||
"--scenes", str(job["scenes"]), "--seconds", str(job["seconds"])]
|
||
env = dict(os.environ)
|
||
env["PYTHONIOENCODING"] = "utf-8"
|
||
if GEMINI_API_KEY:
|
||
env["GEMINI_API_KEY"] = GEMINI_API_KEY
|
||
|
||
job["status"] = "running"
|
||
p = subprocess.Popen(
|
||
cmd, cwd=str(engine_dir), env=env,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||
text=True, encoding="utf-8", errors="replace", bufsize=1,
|
||
)
|
||
for line in p.stdout:
|
||
line = line.rstrip()
|
||
if not line:
|
||
continue
|
||
job["log"].append(line)
|
||
if len(job["log"]) > 300:
|
||
del job["log"][:-300]
|
||
m = _STEP_RE.search(line)
|
||
if m:
|
||
job["step"] = int(m.group(1))
|
||
d = _DONE_RE.search(line)
|
||
if d:
|
||
job["output"] = d.group(1).strip()
|
||
code = p.wait()
|
||
if code == 0:
|
||
job["status"] = "done"; job["step"] = 4
|
||
else:
|
||
job["status"] = "error"; job["error"] = f"생성 프로세스 종료코드 {code}"
|
||
except Exception as e:
|
||
job["status"] = "error"; job["error"] = f"{type(e).__name__}: {e}"
|
||
finally:
|
||
with _lock:
|
||
_running -= 1
|
||
_pump()
|