Spaces:
Sleeping
Sleeping
File size: 4,591 Bytes
eda316b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """Plain stdin interactive gates for the pipeline."""
from __future__ import annotations
from pathlib import Path
from humeo_core.schemas import ApprovalResult, Clip, RatingFeedback
_ISSUE_MAP = {
"a": "wrong_moments",
"b": "bad_cuts",
"c": "boring",
"d": "confusing",
"e": "wrong_layout",
"f": "length_off",
"g": "other",
}
def _preview(text: str, limit: int = 100) -> str:
compact = " ".join(text.split())
if len(compact) <= limit:
return compact
return compact[: limit - 3].rstrip() + "..."
def approve_clips(clips: list[Clip]) -> ApprovalResult:
"""Prompt the user to approve or refine the selected clips."""
clip_ids = [clip.clip_id for clip in clips]
for clip in clips:
print(
f'[{clip.clip_id}] score={clip.virality_score:.2f} '
f'duration={clip.duration_sec:.1f}s "{clip.topic}"'
)
print(f' "{_preview(clip.transcript)}"')
print()
print("Actions:")
print(" numbers in order (e.g. '3,1,5') — select these clips to proceed")
print(" 'all' — accept all clips as-is")
print(" 'refine <note>' — re-run selection with steering")
print(" 'quit' — abort pipeline")
print()
while True:
raw = input("> ").strip()
lowered = raw.lower()
if lowered == "all":
return ApprovalResult(action="accept_all", selected_ids=list(clip_ids))
if lowered == "quit":
return ApprovalResult(action="quit")
if lowered.startswith("refine"):
note = raw[6:].strip()
if not note:
print("Refine requires a note. Try: refine more emotional clips")
continue
return ApprovalResult(action="refine", steering_note=note)
tokens = [token.strip() for token in raw.split(",") if token.strip()]
if not tokens:
print("Enter clip numbers, 'all', 'refine <note>', or 'quit'.")
continue
selected_ids: list[str] = []
seen_ids: set[str] = set()
invalid = False
for token in tokens:
clip_id: str | None = None
if token in clip_ids:
clip_id = token
elif token.isdigit():
idx = int(token)
if 1 <= idx <= len(clips):
clip_id = clips[idx - 1].clip_id
if clip_id is None:
print(f"Unknown clip selection: {token}")
invalid = True
break
if clip_id in seen_ids:
print(f"Duplicate clip selection: {token}")
invalid = True
break
seen_ids.add(clip_id)
selected_ids.append(clip_id)
if invalid:
continue
return ApprovalResult(action="proceed", selected_ids=selected_ids)
def rate_output(outputs: list[Path]) -> RatingFeedback:
"""Prompt the user to rate the rendered outputs."""
print("Outputs:")
for path in outputs:
print(f" {path}")
print()
print("Watch them, then rate:")
print(" 1. slop 2. good 3. great")
while True:
rating_raw = input("> ").strip()
if rating_raw in {"1", "2", "3"}:
rating = int(rating_raw)
break
print("Enter 1, 2, or 3.")
if rating == 3:
return RatingFeedback(rating=3)
print("What's wrong? (space-separated letters, or empty for skip):")
print(" [a] wrong_moments [b] bad_cuts [c] boring [d] confusing")
print(" [e] wrong_layout [f] length_off [g] other (free text)")
while True:
issues_raw = input("> ").strip()
if not issues_raw:
return RatingFeedback(rating=rating)
tokens = issues_raw.lower().split()
issues: list[str] = []
invalid = [token for token in tokens if token not in _ISSUE_MAP]
if invalid:
print(f"Unknown issue selection: {' '.join(invalid)}")
continue
for token in tokens:
issue = _ISSUE_MAP[token]
if issue not in issues:
issues.append(issue)
free_text = None
if "other" in issues:
other_text = input("> ").strip()
free_text = other_text or None
return RatingFeedback(rating=rating, issues=issues, free_text=free_text)
|