fix pipelines

This commit is contained in:
2026-06-02 22:43:12 -07:00
parent 40a4274fd3
commit b2c0c6ccdb
4 changed files with 92 additions and 1 deletions
+6
View File
@@ -232,13 +232,16 @@ def run_pipeline(session: Session, video_id: int, run_id: int | None = None) ->
video.processing_status = "running" video.processing_status = "running"
update_run(run, "media", "running") update_run(run, "media", "running")
prepare_media(session, video) prepare_media(session, video)
session.commit()
update_run(run, "transcribe", "running") update_run(run, "transcribe", "running")
transcribe_video(session, video) transcribe_video(session, video)
session.commit()
if settings.suggest_enabled: if settings.suggest_enabled:
update_run(run, "suggest", "running") update_run(run, "suggest", "running")
suggest_clips(session, video) suggest_clips(session, video)
session.commit()
if settings.render_enabled: if settings.render_enabled:
clips = session.scalars( clips = session.scalars(
@@ -250,12 +253,15 @@ def run_pipeline(session: Session, video_id: int, run_id: int | None = None) ->
render_clip_by_id(session, clip.id) render_clip_by_id(session, clip.id)
if settings.upload_enabled: if settings.upload_enabled:
upload_clip_by_id(session, clip.id) upload_clip_by_id(session, clip.id)
session.commit()
video.processing_status = "done" video.processing_status = "done"
update_run(run, "done", "done") update_run(run, "done", "done")
session.commit()
logger.info("pipeline done video_id=%s run_id=%s", video_id, run_id) logger.info("pipeline done video_id=%s run_id=%s", video_id, run_id)
except Exception as exc: except Exception as exc:
video.processing_status = "failed" video.processing_status = "failed"
update_run(run, "failed", "failed", str(exc)) update_run(run, "failed", "failed", str(exc))
session.commit()
logger.exception("pipeline failed video_id=%s run_id=%s", video_id, run_id) logger.exception("pipeline failed video_id=%s run_id=%s", video_id, run_id)
raise raise
+38
View File
@@ -190,6 +190,7 @@ def chunk_transcript(
def parse_clip_response(content: str) -> list[ClipCandidate]: def parse_clip_response(content: str) -> list[ClipCandidate]:
try: try:
payload = json.loads(content) payload = json.loads(content)
payload = normalize_clip_payload(payload)
parsed = ClipCandidateResponse.model_validate(payload) parsed = ClipCandidateResponse.model_validate(payload)
except (json.JSONDecodeError, ValidationError) as exc: except (json.JSONDecodeError, ValidationError) as exc:
logger.error("invalid clip json content_prefix=%s", content[:1000]) logger.error("invalid clip json content_prefix=%s", content[:1000])
@@ -197,6 +198,43 @@ def parse_clip_response(content: str) -> list[ClipCandidate]:
return parsed.clips return parsed.clips
def normalize_clip_payload(payload: object) -> dict:
if not isinstance(payload, dict):
return {"clips": []}
if "clips" not in payload and "candidates" in payload:
payload = payload | {"clips": payload["candidates"]}
clips = payload.get("clips")
if not isinstance(clips, list):
return payload
normalized_clips = []
for clip in clips:
if not isinstance(clip, dict):
normalized_clips.append(clip)
continue
normalized = dict(clip)
normalized["score"] = normalize_score(normalized.get("score", 0))
normalized_clips.append(normalized)
return payload | {"clips": normalized_clips}
def normalize_score(value: object) -> float:
try:
score = float(value)
except (TypeError, ValueError):
logger.warning("invalid llm score=%r; using 0", value)
return 0.0
if score > 1:
original = score
if score <= 10:
score = score / 10
elif score <= 100:
score = score / 100
else:
score = 1.0
logger.debug("normalized llm score original=%s normalized=%s", original, score)
return max(0.0, min(1.0, score))
def overlap_ratio(left: ClipCandidate, right: ClipCandidate) -> float: def overlap_ratio(left: ClipCandidate, right: ClipCandidate) -> float:
overlap = max(0, min(left.end_sec, right.end_sec) - max(left.start_sec, right.start_sec)) overlap = max(0, min(left.end_sec, right.end_sec) - max(left.start_sec, right.start_sec))
shortest = min(left.end_sec - left.start_sec, right.end_sec - right.start_sec) shortest = min(left.end_sec - left.start_sec, right.end_sec - right.start_sec)
+4 -1
View File
@@ -182,7 +182,10 @@ def bootstrap_existing(client: WebDavClient, session: Session) -> BootstrapResul
session.add(video) session.add(video)
inserted += 1 inserted += 1
logger.debug("bootstrap existing file=%s size=%s", file.filename, file.size_bytes) logger.debug("bootstrap existing file=%s size=%s", file.filename, file.size_bytes)
elif video.ingest_status == "observing" and video.processing_status == "pending": elif (
video.ingest_status in {"observing", "stable", "queued"}
and video.processing_status in {"pending", "queued"}
):
video.size_bytes = file.size_bytes video.size_bytes = file.size_bytes
video.ingest_status = "existing_done" video.ingest_status = "existing_done"
video.processing_status = "done" video.processing_status = "done"
+44
View File
@@ -30,6 +30,50 @@ def test_parse_clip_response():
assert candidates[0].title_zh == "标题" assert candidates[0].title_zh == "标题"
def test_parse_clip_response_accepts_candidates_key_and_score_out_of_ten():
candidates = parse_clip_response(
"""
{
"candidates": [
{
"start_sec": 10,
"end_sec": 60,
"title_zh": "标题",
"summary_zh": "摘要",
"reason": "有趣",
"score": 8.5,
"tags": ["反应"],
"subtitle_priority": "high"
}
]
}
"""
)
assert candidates[0].score == 0.85
def test_parse_clip_response_normalizes_score_out_of_hundred():
candidates = parse_clip_response(
"""
{
"clips": [
{
"start_sec": 10,
"end_sec": 60,
"title_zh": "标题",
"summary_zh": "摘要",
"reason": "有趣",
"score": 92,
"tags": ["反应"],
"subtitle_priority": "high"
}
]
}
"""
)
assert candidates[0].score == 0.92
def test_dedupe_and_rank_candidates_filters_short_and_overlapping(): def test_dedupe_and_rank_candidates_filters_short_and_overlapping():
candidates = [ candidates = [
ClipCandidate( ClipCandidate(