"""shot_validator 일괄 재실행 스크립트 (v0.5.8+ v2 프롬프트 적용). 기존 프로젝트들의 `shot_validator`를 force로 재실행해 최신 프롬프트 반영. prereq: shot_extract 완료. 사용: python backend/scripts/migrate_shot_validator.py PID1 PID2 ... python backend/scripts/migrate_shot_validator.py --dry-run PID1 python backend/scripts/migrate_shot_validator.py --base-url http://localhost:8000 PID1 원칙: - 프로젝트 간 순차 (rate-limit) - prereq 미충족 에피소드 자동 스킵 - HTTP 오류 3회 재시도 (일시 네트워크 vs 실 미충족 구분) - 실패 시 해당 에피소드/프로젝트 스킵, 나머지 계속 """ from __future__ import annotations import argparse import os import sys import time from typing import List, Optional import requests DEFAULT_BASE_URL = os.environ.get("THEROAD_BASE_URL", "http://localhost:8000") DEFAULT_USERNAME = os.environ.get("THEROAD_USERNAME", "admin") DEFAULT_PASSWORD = os.environ.get("THEROAD_PASSWORD", "admin123") FORCE_STEPS = ["shot_validator"] REQUIRED_PREREQ_STEPS = ["shot_extract"] POLL_INTERVAL_SEC = 20 POLL_TIMEOUT_SEC = 20 * 60 # 20분 class Migration: def __init__(self, base_url: str, session: requests.Session): self.base_url = base_url.rstrip("/") self.sess = session self._creds: Optional[tuple] = None def login(self, username: str, password: str): resp = self.sess.post( f"{self.base_url}/api/v1/auth/login", json={"username": username, "password": password}, timeout=10, ) resp.raise_for_status() self._creds = (username, password) print(f"[login] OK — user={resp.json().get('username')}") def _relogin(self, resp): if resp.status_code != 401 or not self._creds: return False print(" [session expired] re-logging in...") try: self.login(*self._creds) return True except requests.RequestException as exc: print(f" [relogin failed] {exc}") return False def list_episodes(self, pid: str) -> list: resp = self.sess.get(f"{self.base_url}/api/v1/projects/{pid}/episodes/", timeout=10) resp.raise_for_status() data = resp.json() if isinstance(data, dict) and "items" in data: return data["items"] return data if isinstance(data, list) else [] def check_prereq(self, pid: str, eid: str) -> List[str]: """HTTP 오류 3회 재시도 후 미충족 판정.""" url = f"{self.base_url}/api/v1/projects/{pid}/episodes/{eid}/steps" last_exc: Optional[Exception] = None for attempt in range(1, 4): try: resp = self.sess.get(url, timeout=10) # 401 재로그인은 attempt에 포함 안 함 (v0.5.22 사후 리뷰 수용) if resp.status_code == 401 and self._relogin(resp): resp = self.sess.get(url, timeout=10) resp.raise_for_status() states = {s["step_id"]: s.get("status", "") for s in resp.json().get("steps", [])} missing = [] for sid in REQUIRED_PREREQ_STEPS: status = states.get(sid, "") if status not in ("completed", "partial", "stale"): missing.append(f"{sid}({status or 'missing'})") return missing except requests.RequestException as exc: last_exc = exc if attempt < 3: wait = 5 * attempt print(f" [prereq http error {attempt}/3] {exc} — retry in {wait}s") time.sleep(wait) return [f""] def force_step(self, pid: str, eid: str, step_id: str) -> bool: url = f"{self.base_url}/api/v1/projects/{pid}/episodes/{eid}/steps/{step_id}" resp = self.sess.post(url, params={"mode": "force"}, timeout=30) if resp.status_code >= 400: print(f"[force] FAIL {step_id} {resp.status_code} {resp.text[:120]}") return False return True def wait_step_done(self, pid: str, eid: str, step_id: str) -> str: url = f"{self.base_url}/api/v1/projects/{pid}/episodes/{eid}/steps" t0 = time.time() last_status = "" while time.time() - t0 < POLL_TIMEOUT_SEC: try: resp = self.sess.get(url, timeout=10) except requests.RequestException as exc: print(f" [poll error] {exc} — retrying") time.sleep(POLL_INTERVAL_SEC) continue if resp.status_code == 401: if self._relogin(resp): continue return "error" if resp.status_code >= 400: print(f" [poll HTTP {resp.status_code}] {resp.text[:120]} — retrying") time.sleep(POLL_INTERVAL_SEC) continue for s in resp.json().get("steps", []): if s["step_id"] != step_id: continue status = s.get("status", "") if status != last_status: cc = s.get("completed_count") or 0 ac = s.get("applicable_count") or 0 print(f" {step_id}: {status} {cc}/{ac}") last_status = status if status in ("completed", "partial", "failed"): return status break time.sleep(POLL_INTERVAL_SEC) return "timeout" def run_project(self, pid: str, dry_run: bool) -> bool: print(f"\n=== Project {pid} ===") try: episodes = self.list_episodes(pid) except requests.RequestException as exc: print(f"[project] FAIL list_episodes: {exc}") return False if not episodes: print("[project] no episodes — skip") return True any_failed = False for ep in episodes: eid = ep.get("id") title = ep.get("title", "") print(f"\n -- Episode {eid} ({title}) --") missing = self.check_prereq(pid, eid) if missing: print(f" [prereq] SKIP: {', '.join(missing)}") any_failed = True continue if dry_run: print(f" [dry-run] prereq OK — would force {FORCE_STEPS}") continue for step_id in FORCE_STEPS: print(f" → force {step_id}") if not self.force_step(pid, eid, step_id): any_failed = True print(f" stop (force failed on {step_id})") break status = self.wait_step_done(pid, eid, step_id) if status not in ("completed", "partial"): any_failed = True print(f" stop ({step_id} -> {status})") break return not any_failed def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("pids", nargs="+") ap.add_argument("--base-url", default=DEFAULT_BASE_URL) ap.add_argument("--username", default=DEFAULT_USERNAME) ap.add_argument("--password", default=DEFAULT_PASSWORD) ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() sess = requests.Session() mig = Migration(args.base_url, sess) try: mig.login(args.username, args.password) except requests.RequestException as exc: print(f"[login] FAIL {exc}") return 1 overall_ok = True for pid in args.pids: ok = mig.run_project(pid, args.dry_run) overall_ok = overall_ok and ok if not args.dry_run: print(f"\n[project {pid}] {'OK' if ok else 'PARTIAL FAIL'}") return 0 if overall_ok else 2 if __name__ == "__main__": sys.exit(main())