368 lines
15 KiB
Python
Executable File
368 lines
15 KiB
Python
Executable File
"""
|
|
FCPXML Parser - Reads Final Cut Pro XML files into Python objects.
|
|
"""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
from .models import (
|
|
MARKER_XML_TAGS,
|
|
Clip,
|
|
ConnectedClip,
|
|
Keyword,
|
|
Marker,
|
|
MarkerType,
|
|
Project,
|
|
Timecode,
|
|
Timeline,
|
|
TimeValue,
|
|
Transition,
|
|
)
|
|
from .safe_xml import safe_fromstring, safe_parse
|
|
|
|
# Maximum FCPXML file size (50 MB) — prevents memory exhaustion from crafted files
|
|
_MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024
|
|
|
|
# Tags that represent connected clip elements (includes 'title' for text overlays)
|
|
_CONNECTED_CLIP_TAGS = ('asset-clip', 'clip', 'video', 'audio', 'title', 'ref-clip')
|
|
|
|
|
|
class FCPXMLParser:
|
|
"""Parser for Final Cut Pro FCPXML files. Supports versions 1.8 - 1.14.
|
|
|
|
Unknown elements introduced by newer FCPXML versions (e.g. 1.13's
|
|
``adjust-stereo-3D`` / ``hidden-clip-marker``, 1.14's smart-collection
|
|
search rules) are ignored on read and preserved untouched by the
|
|
modify path, which operates on the raw ElementTree.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.resources: Dict[str, Dict[str, Any]] = {}
|
|
self.formats: Dict[str, Dict[str, Any]] = {}
|
|
self.frame_rate: float = 24.0
|
|
|
|
def _tc(self, elem: ET.Element, attr: str, default: str = '0s') -> Timecode:
|
|
"""Parse a rational time attribute from an XML element.
|
|
|
|
Centralises the ``Timecode.from_rational(elem.get(attr), frame_rate)``
|
|
pattern that repeats across every clip/marker/transition parser.
|
|
"""
|
|
return Timecode.from_rational(elem.get(attr, default), self.frame_rate)
|
|
|
|
def parse_file(self, filepath: str) -> Project:
|
|
"""Parse an FCPXML file and return a Project object.
|
|
|
|
Enforces a file size limit to prevent memory exhaustion from
|
|
maliciously large XML files.
|
|
"""
|
|
path = Path(filepath)
|
|
if path.suffix == '.fcpxmld':
|
|
fcpxml_path = path / 'Info.fcpxml'
|
|
if not fcpxml_path.exists():
|
|
raise FileNotFoundError(f"Info.fcpxml not found in bundle: {filepath}")
|
|
filepath = str(fcpxml_path)
|
|
path = Path(filepath)
|
|
file_size = path.stat().st_size
|
|
if file_size > _MAX_FILE_SIZE_BYTES:
|
|
raise ValueError(
|
|
f"FCPXML file exceeds maximum size "
|
|
f"({file_size / 1024 / 1024:.1f} MB > "
|
|
f"{_MAX_FILE_SIZE_BYTES / 1024 / 1024:.0f} MB limit)"
|
|
)
|
|
tree = safe_parse(filepath)
|
|
return self._parse_fcpxml(tree.getroot())
|
|
|
|
def parse_string(self, xml_string: str) -> Project:
|
|
"""Parse FCPXML from a string."""
|
|
return self._parse_fcpxml(safe_fromstring(xml_string))
|
|
|
|
def _parse_fcpxml(self, root: ET.Element) -> Project:
|
|
"""Parse the root fcpxml element."""
|
|
version = root.get('version', '1.11')
|
|
resources_elem = root.find('resources')
|
|
if resources_elem is not None:
|
|
self._parse_resources(resources_elem)
|
|
|
|
timelines = []
|
|
for library in root.findall('.//library'):
|
|
for event in library.findall('event'):
|
|
for project in event.findall('project'):
|
|
timeline = self._parse_project(project)
|
|
if timeline:
|
|
timelines.append(timeline)
|
|
|
|
if not timelines:
|
|
for project in root.findall('.//project'):
|
|
timeline = self._parse_project(project)
|
|
if timeline:
|
|
timelines.append(timeline)
|
|
|
|
project_name = timelines[0].name if timelines else "Untitled"
|
|
return Project(name=project_name, timelines=timelines, fcpxml_version=version)
|
|
|
|
def _parse_resources(self, resources: ET.Element):
|
|
"""Parse the resources section."""
|
|
for fmt in resources.findall('format'):
|
|
fmt_id = fmt.get('id', '')
|
|
self.formats[fmt_id] = {
|
|
'id': fmt_id, 'name': fmt.get('name', ''),
|
|
'width': int(fmt.get('width', 1920)),
|
|
'height': int(fmt.get('height', 1080)),
|
|
'frameDuration': fmt.get('frameDuration', '1/24s')
|
|
}
|
|
frame_dur = fmt.get('frameDuration', '1/24s')
|
|
if '/' in frame_dur:
|
|
parts = frame_dur.rstrip('s').split('/', 1)
|
|
num, denom = int(parts[0]), int(parts[1])
|
|
if num <= 0:
|
|
raise ValueError(f"Invalid frameDuration numerator: {frame_dur}")
|
|
if denom <= 0:
|
|
raise ValueError(f"Invalid frameDuration denominator: {frame_dur}")
|
|
self.frame_rate = denom / num
|
|
|
|
for asset in resources.findall('asset'):
|
|
asset_id = asset.get('id', '')
|
|
self.resources[asset_id] = {
|
|
'id': asset_id, 'name': asset.get('name', ''),
|
|
'src': asset.get('src', '') or (media_rep.get('src', '') if (media_rep := asset.find('media-rep')) is not None else ''),
|
|
'start': asset.get('start', '0s'),
|
|
'duration': asset.get('duration', '0s'),
|
|
'hasVideo': asset.get('hasVideo', '1') == '1',
|
|
'hasAudio': asset.get('hasAudio', '1') == '1',
|
|
}
|
|
|
|
def _parse_project(self, project: ET.Element) -> Optional[Timeline]:
|
|
"""Parse a project element into a Timeline."""
|
|
name = project.get('name', 'Untitled')
|
|
sequence = project.find('sequence')
|
|
if sequence is None:
|
|
return None
|
|
|
|
format_ref = sequence.get('format', '')
|
|
fmt = self.formats.get(format_ref, {})
|
|
|
|
timeline = Timeline(
|
|
name=name,
|
|
duration=self._tc(sequence, 'duration'),
|
|
frame_rate=self.frame_rate,
|
|
width=fmt.get('width', 1920),
|
|
height=fmt.get('height', 1080)
|
|
)
|
|
|
|
spine = sequence.find('spine')
|
|
if spine is not None:
|
|
self._parse_spine(spine, timeline)
|
|
|
|
timeline.markers.extend(self._collect_markers(sequence))
|
|
|
|
return timeline
|
|
|
|
def _parse_spine(self, spine: ET.Element, timeline: Timeline):
|
|
"""Parse the spine (primary storyline) including connected clips."""
|
|
current_offset = 0
|
|
for elem in spine:
|
|
tag = elem.tag
|
|
if tag in ('asset-clip', 'clip', 'video', 'mc-clip', 'sync-clip', 'ref-clip'):
|
|
clip = self._parse_clip(elem, current_offset)
|
|
if clip:
|
|
timeline.clips.append(clip)
|
|
self._parse_connected_clips(elem, clip, timeline)
|
|
current_offset += clip.duration.frames
|
|
elif tag == 'gap':
|
|
gap_frames = self._tc(elem, 'duration').frames
|
|
self._parse_gap_connected_clips(elem, current_offset, timeline)
|
|
current_offset += gap_frames
|
|
elif tag == 'transition':
|
|
transition = self._parse_transition(elem, current_offset)
|
|
if transition:
|
|
timeline.transitions.append(transition)
|
|
|
|
def _parse_clip(self, elem: ET.Element, offset: int) -> Optional[Clip]:
|
|
"""Parse a clip element."""
|
|
name = elem.get('name', 'Untitled Clip')
|
|
duration = self._tc(elem, 'duration')
|
|
source_start = self._tc(elem, 'start')
|
|
ref = elem.get('ref', '')
|
|
media_path = self.resources.get(ref, {}).get('src', '')
|
|
|
|
clip = Clip(
|
|
name=name,
|
|
start=Timecode(frames=offset, frame_rate=self.frame_rate),
|
|
duration=duration,
|
|
source_start=source_start,
|
|
media_path=media_path,
|
|
audio_role=elem.get('audioRole', ''),
|
|
video_role=elem.get('videoRole', ''),
|
|
)
|
|
|
|
clip.markers.extend(self._collect_markers(elem))
|
|
|
|
for keyword_elem in elem.findall('keyword'):
|
|
keyword = self._parse_keyword(keyword_elem)
|
|
if keyword:
|
|
clip.keywords.append(keyword)
|
|
|
|
return clip
|
|
|
|
def _parse_marker_element(self, elem: ET.Element) -> Optional[Marker]:
|
|
"""Parse any marker element (<marker> or <chapter-marker>).
|
|
|
|
Type detection is delegated to MarkerType.from_xml_element which
|
|
owns the completed-attribute semantics. This means the parser
|
|
doesn't need separate methods for each tag.
|
|
"""
|
|
return Marker(
|
|
name=elem.get('value', ''),
|
|
start=self._tc(elem, 'start'),
|
|
duration=self._tc(elem, 'duration', '1/24s'),
|
|
marker_type=MarkerType.from_xml_element(elem),
|
|
note=elem.get('note', '')
|
|
)
|
|
|
|
def _collect_markers(self, elem: ET.Element) -> list:
|
|
"""Collect all markers (standard + chapter) from an element in a single pass.
|
|
|
|
Iterates children once, selecting recognised marker tags via
|
|
MARKER_XML_TAGS rather than making a separate findall per tag.
|
|
"""
|
|
return [
|
|
marker
|
|
for child in elem
|
|
if child.tag in MARKER_XML_TAGS
|
|
for marker in [self._parse_marker_element(child)]
|
|
if marker is not None
|
|
]
|
|
|
|
def _parse_keyword(self, elem: ET.Element) -> Optional[Keyword]:
|
|
"""Parse a keyword element."""
|
|
return Keyword(
|
|
value=elem.get('value', ''),
|
|
start=self._tc(elem, 'start') if elem.get('start') else None,
|
|
duration=self._tc(elem, 'duration') if elem.get('duration') else None,
|
|
)
|
|
|
|
def _parse_transition(self, elem: ET.Element, offset: int) -> Optional[Transition]:
|
|
"""Parse a transition element."""
|
|
return Transition(
|
|
name=elem.get('name', 'Cross Dissolve'),
|
|
duration=self._tc(elem, 'duration', '1s'),
|
|
start=Timecode(frames=offset, frame_rate=self.frame_rate)
|
|
)
|
|
|
|
def get_library_clips(self, keywords: Optional[list] = None) -> list:
|
|
"""
|
|
Get all available clips from the library (assets in resources section).
|
|
|
|
Args:
|
|
keywords: Optional list of keywords to filter by
|
|
|
|
Returns:
|
|
List of dicts with asset metadata: name, asset_id, duration_seconds, src
|
|
"""
|
|
result = []
|
|
for asset_id, asset_data in self.resources.items():
|
|
# Parse duration to seconds
|
|
duration_str = asset_data.get('duration', '0s')
|
|
duration_seconds = self._parse_duration_to_seconds(duration_str)
|
|
|
|
clip_info = {
|
|
'asset_id': asset_id,
|
|
'name': asset_data.get('name', ''),
|
|
'duration_seconds': duration_seconds,
|
|
'src': asset_data.get('src', ''),
|
|
'has_video': asset_data.get('hasVideo', True),
|
|
'has_audio': asset_data.get('hasAudio', True),
|
|
}
|
|
result.append(clip_info)
|
|
|
|
# Filter by keywords if provided
|
|
if keywords:
|
|
# For now, assets don't have keywords directly - return empty if filtering
|
|
# In real FCPXML, keywords are typically on clips in events, not assets
|
|
return []
|
|
|
|
return result
|
|
|
|
def _iter_connected_elements(self, parent_elem: ET.Element, parent_name: str):
|
|
"""Yield ``(element, lane, parent_name)`` tuples for connected clips.
|
|
|
|
Shared iteration logic for both spine-clip and gap-attached connected
|
|
clips — walks direct children with a ``lane`` attribute and
|
|
``<storyline>`` wrappers, yielding parsed :class:`ConnectedClip`
|
|
objects without prescribing where they get stored.
|
|
"""
|
|
for child in parent_elem:
|
|
lane = child.get('lane')
|
|
if lane is not None and child.tag in _CONNECTED_CLIP_TAGS:
|
|
connected = self._parse_one_connected_clip(
|
|
child, int(lane), parent_name)
|
|
if connected:
|
|
yield connected
|
|
elif child.tag == 'storyline':
|
|
lane_val = int(child.get('lane', '1'))
|
|
for sub_elem in child:
|
|
if sub_elem.tag in _CONNECTED_CLIP_TAGS:
|
|
connected = self._parse_one_connected_clip(
|
|
sub_elem, lane_val, parent_name)
|
|
if connected:
|
|
yield connected
|
|
|
|
def _parse_connected_clips(self, parent_elem: ET.Element,
|
|
parent_clip: Clip, timeline: Timeline):
|
|
"""Parse connected clips attached to a primary storyline clip."""
|
|
for connected in self._iter_connected_elements(parent_elem, parent_clip.name):
|
|
parent_clip.connected_clips.append(connected)
|
|
timeline.connected_clips.append(connected)
|
|
|
|
def _parse_gap_connected_clips(self, gap_elem: ET.Element,
|
|
gap_offset: int, timeline: Timeline):
|
|
"""Parse connected clips attached to gap elements."""
|
|
for connected in self._iter_connected_elements(gap_elem, f"gap@{gap_offset}"):
|
|
timeline.connected_clips.append(connected)
|
|
|
|
def _parse_one_connected_clip(self, elem: ET.Element, lane: int,
|
|
parent_name: str) -> Optional[ConnectedClip]:
|
|
"""Parse a single connected clip element."""
|
|
name = elem.get('name', 'Untitled')
|
|
duration = self._tc(elem, 'duration')
|
|
start = self._tc(elem, 'start')
|
|
offset = self._tc(elem, 'offset')
|
|
ref = elem.get('ref', '')
|
|
media_path = self.resources.get(ref, {}).get('src', '')
|
|
role = elem.get('audioRole', '') or elem.get('videoRole', '')
|
|
|
|
connected = ConnectedClip(
|
|
name=name, start=start, duration=duration,
|
|
lane=lane, offset=offset, source_start=start,
|
|
media_path=media_path, clip_type=elem.tag, role=role,
|
|
ref_id=ref, parent_clip_name=parent_name,
|
|
)
|
|
|
|
connected.markers.extend(self._collect_markers(elem))
|
|
|
|
for keyword_elem in elem.findall('keyword'):
|
|
keyword = self._parse_keyword(keyword_elem)
|
|
if keyword:
|
|
connected.keywords.append(keyword)
|
|
|
|
return connected
|
|
|
|
def _parse_duration_to_seconds(self, duration_str: str) -> float:
|
|
"""Convert FCPXML duration string to seconds.
|
|
|
|
Delegates to TimeValue.from_timecode() which handles rational
|
|
format (``"150/30s"``), plain seconds (``"10s"``), timecode
|
|
(``HH:MM:SS:FF``), and frame counts (``"15f"``).
|
|
"""
|
|
try:
|
|
return TimeValue.from_timecode(duration_str).to_seconds()
|
|
except (ValueError, ZeroDivisionError):
|
|
# Zero-denominator or unparseable → 0.0 (matches prior behaviour)
|
|
return 0.0
|
|
|
|
|
|
def parse_fcpxml(filepath: str) -> Project:
|
|
"""Convenience function to parse an FCPXML file."""
|
|
return FCPXMLParser().parse_file(filepath)
|