Eram 882 linhas de seis papéis sem relação, sob um nome que só dizia
"compartilhado" — o depósito onde tudo que servia a mais de um handler
acabava caindo.
media 316 transcrição em cache, corte por fala, relatório
paths 206 sandbox, limites, caminho de saída
project 116 abrir projeto, preparar modifier/generator
captions 112 SRT, VTT, listas com timestamp
detection 99 flash frames, buracos, duplicados
formatting 86 tabelas e relatórios dos handlers
O __init__ reexporta os 46 nomes, então os treze pontos que importam daqui
não mudaram.
_transcript_cut_report saiu de formatting para media: ele precisa do hint de
instalação e do _text_result, ou seja, é relatório de transcrição e não
formatação genérica — mover foi mais honesto que cruzar imports entre os
dois módulos.
Quatro testes patchavam `server_tools._shared.transcribe`; o nome agora é
ligado por _shared/media.py, então o patch passou a apontar para lá — mesmo
padrão da experiência #23.
Lint zerado, 1454 testes passando.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
"""Detecção para QC: flash frames, buracos e clipes duplicados.
|
|
|
|
Extraído de _shared.py — ver server_tools/_shared/__init__.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fcpxml.models import (
|
|
DuplicateGroup,
|
|
FlashFrame,
|
|
FlashFrameSeverity,
|
|
GapInfo,
|
|
Timecode,
|
|
)
|
|
|
|
from .formatting import format_timecode
|
|
|
|
|
|
def _detect_flash_frames(
|
|
tl: Any, *, critical_threshold: int = 2, warning_threshold: int = 6,
|
|
) -> list:
|
|
"""Find clips shorter than *warning_threshold* frames.
|
|
|
|
Returns a list of ``FlashFrame`` objects sorted by severity. Shared by
|
|
``handle_detect_flash_frames`` and ``handle_validate_timeline`` so the
|
|
detection logic lives in exactly one place.
|
|
"""
|
|
fps = tl.frame_rate
|
|
flash_frames: list[FlashFrame] = []
|
|
for clip in tl.clips:
|
|
duration_frames = int(clip.duration_seconds * fps)
|
|
if duration_frames < warning_threshold:
|
|
severity = (
|
|
FlashFrameSeverity.CRITICAL
|
|
if duration_frames < critical_threshold
|
|
else FlashFrameSeverity.WARNING
|
|
)
|
|
flash_frames.append(FlashFrame(
|
|
clip_name=clip.name, clip_id=clip.name,
|
|
start=clip.start, duration_frames=duration_frames,
|
|
duration_seconds=clip.duration_seconds, severity=severity,
|
|
))
|
|
return flash_frames
|
|
|
|
def _detect_gaps(tl: Any, *, min_gap_frames: int = 1) -> list:
|
|
"""Find inter-clip gaps of at least *min_gap_frames* length.
|
|
|
|
Returns a list of ``GapInfo`` objects. Shared by ``handle_detect_gaps``
|
|
and ``handle_validate_timeline``.
|
|
"""
|
|
fps = tl.frame_rate
|
|
min_gap_seconds = min_gap_frames / fps
|
|
gaps: list[GapInfo] = []
|
|
sorted_clips = sorted(tl.clips, key=lambda c: c.start.seconds)
|
|
for i in range(len(sorted_clips) - 1):
|
|
current_end = sorted_clips[i].end.seconds
|
|
next_start = sorted_clips[i + 1].start.seconds
|
|
gap_duration = next_start - current_end
|
|
if gap_duration >= min_gap_seconds:
|
|
gaps.append(GapInfo(
|
|
start=Timecode(frames=int(current_end * fps), frame_rate=fps),
|
|
duration_frames=int(gap_duration * fps),
|
|
duration_seconds=gap_duration,
|
|
previous_clip=sorted_clips[i].name,
|
|
next_clip=sorted_clips[i + 1].name,
|
|
))
|
|
return gaps
|
|
|
|
def _detect_duplicate_groups(tl: Any, *, mode: str = "same_source") -> list:
|
|
"""Group clips that share a source media reference.
|
|
|
|
Returns a list of ``DuplicateGroup`` objects. Shared by
|
|
``handle_detect_duplicates`` and ``handle_validate_timeline``.
|
|
"""
|
|
source_groups: dict[str, list[dict]] = {}
|
|
for clip in tl.clips:
|
|
source_key = clip.media_path or clip.name
|
|
if source_key not in source_groups:
|
|
source_groups[source_key] = []
|
|
source_groups[source_key].append({
|
|
'name': clip.name,
|
|
'start': clip.start.seconds,
|
|
'duration': clip.duration_seconds,
|
|
'source_start': clip.source_start.seconds if clip.source_start else 0,
|
|
'source_duration': clip.duration_seconds,
|
|
'timecode': format_timecode(clip.start),
|
|
})
|
|
|
|
duplicates: list[DuplicateGroup] = []
|
|
for source_key, clips in source_groups.items():
|
|
if len(clips) <= 1:
|
|
continue
|
|
group = DuplicateGroup(
|
|
source_ref=source_key,
|
|
source_name=source_key.split('/')[-1] if '/' in source_key else source_key,
|
|
clips=clips,
|
|
)
|
|
if mode == "same_source":
|
|
duplicates.append(group)
|
|
elif mode == "overlapping_ranges" and group.has_overlapping_ranges:
|
|
duplicates.append(group)
|
|
elif mode == "identical":
|
|
seen_ranges: set[tuple] = set()
|
|
identical_clips = []
|
|
for c in clips:
|
|
range_key = (c['source_start'], c['source_duration'])
|
|
if range_key in seen_ranges:
|
|
identical_clips.append(c)
|
|
seen_ranges.add(range_key)
|
|
if identical_clips:
|
|
group.clips = identical_clips
|
|
duplicates.append(group)
|
|
return duplicates
|