Extractors Module¶

Video source extraction from Photos library and folders.

Photos Extractor¶

photos_extractor ¶

Photos library integration module.

This module provides read-only access to the macOS Photos library for querying video assets using the osxphotos library.

SDS Reference: SDS-P01-004 SRS Reference: SRS-301 (Photos Library Integration)

Example

library = PhotosLibrary() if library.check_permissions(): ... print(f"Found {library.get_video_count()} videos") ... else: ... print("Access denied")

PhotosLibraryError ¶

Bases: Exception

Base exception for Photos library operations.

PhotosAccessDeniedError ¶

PhotosAccessDeniedError(message=None)

Bases: PhotosLibraryError

Raised when access to Photos library is denied.

This typically occurs when the application doesn't have Full Disk Access permission in System Settings.

Initialize with optional custom message.

Parameters:

Name	Type	Description	Default
`message`	`str \| None`	Custom error message. If None, uses default.	`None`

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, message: str | None = None) -> None:
    """Initialize with optional custom message.

    Args:
        message: Custom error message. If None, uses default.
    """
    if message is None:
        message = (
            "Photos library access denied. "
            "Grant Full Disk Access permission in System Settings."
        )
    super().__init__(message)

PhotosLibraryNotFoundError ¶

PhotosLibraryNotFoundError(path=None)

Bases: PhotosLibraryError

Raised when Photos library cannot be found.

Initialize with optional library path.

Parameters:

Name	Type	Description	Default
`path`	`Path \| None`	Path that was attempted if custom path was specified.	`None`

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, path: Path | None = None) -> None:
    """Initialize with optional library path.

    Args:
        path: Path that was attempted if custom path was specified.
    """
    if path:
        message = f"Photos library not found at: {path}"
    else:
        message = "Default Photos library not found"
    super().__init__(message)

VideoNotAvailableError ¶

VideoNotAvailableError(filename)

Bases: PhotosLibraryError

Raised when video is not available locally (iCloud only).

This exception indicates that the video file exists only in iCloud and must be downloaded first using the Photos app.

Initialize with video filename.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Name of the unavailable video file.	required

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, filename: str) -> None:
    """Initialize with video filename.

    Args:
        filename: Name of the unavailable video file.
    """
    message = (
        f"'{filename}' is stored in iCloud only. "
        "Download it first in Photos app before exporting."
    )
    super().__init__(message)

ExportError ¶

ExportError(message, filename=None)

Bases: PhotosLibraryError

Raised when video export fails.

This exception covers various export failures including file copy errors, permission issues, and insufficient disk space.

Initialize with error message and optional filename.

Parameters:

Name	Type	Description	Default
`message`	`str`	Description of what failed.	required
`filename`	`str \| None`	Name of the video file if applicable.	`None`

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, message: str, filename: str | None = None) -> None:
    """Initialize with error message and optional filename.

    Args:
        message: Description of what failed.
        filename: Name of the video file if applicable.
    """
    if filename:
        full_message = f"Failed to export '{filename}': {message}"
    else:
        full_message = f"Export failed: {message}"
    super().__init__(full_message)

MediaType ¶

Bases: Enum

Media types available in Photos library.

Attributes:

Name	Type	Description
`VIDEO`		Video files (mov, mp4, etc.).
`PHOTO`		Photo files (jpg, heic, etc.).
`ALL`		All media types.

PhotosVideoInfo `dataclass` ¶

PhotosVideoInfo(uuid, filename, path, date, date_modified, duration, favorite=False, hidden=False, in_cloud=False, location=None, albums=list(), codec=None, size=0)

Information about a video in the Photos library.

Attributes:

Name	Type	Description
`uuid`	`str`	Unique identifier for the asset in Photos.
`filename`	`str`	Original filename of the video.
`path`	`Path \| None`	Path to the video file (may be None if in iCloud).
`date`	`datetime \| None`	Creation date of the video.
`date_modified`	`datetime \| None`	Last modification date.
`duration`	`float`	Video duration in seconds.
`favorite`	`bool`	Whether the video is marked as favorite.
`hidden`	`bool`	Whether the video is hidden.
`in_cloud`	`bool`	Whether the video is stored in iCloud.
`location`	`tuple[float, float] \| None`	GPS coordinates (latitude, longitude) if available.
`albums`	`list[str]`	List of album names containing this video.
`codec`	`str \| None`	Video codec name (e.g., "h264", "hevc").
`size`	`int`	File size in bytes.

is_available_locally `property` ¶

is_available_locally

Check if the video file is available locally.

Returns:

Type	Description
`bool`	True if the file exists locally, False if in iCloud only.

is_h264 `property` ¶

is_h264

Check if video codec is H.264/AVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.264.

is_hevc `property` ¶

is_hevc

Check if video codec is H.265/HEVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.265/HEVC.

needs_conversion `property` ¶

needs_conversion

Check if video needs H.265 conversion.

A video needs conversion if: - It is locally available (not iCloud-only) - Its codec is H.264

Returns:

Type	Description
`bool`	True if the video should be converted to H.265.

PhotosLibrary ¶

PhotosLibrary(library_path=None)

Interface to macOS Photos library via osxphotos.

This class provides a high-level interface for querying videos from the macOS Photos library. It handles library initialization, permission checking, and provides caching for performance.

SDS Reference: SDS-P01-004

Example

library = PhotosLibrary() if library.check_permissions(): ... videos = library.get_videos() ... for video in videos: ... print(f"{video.filename}: {video.duration}s")

Attributes:

Name	Type	Description
`library_path`	`Path`	Path to the Photos library (None for default).

Initialize Photos library connection.

Parameters:

Name	Type	Description	Default
`library_path`	`Path \| None`	Custom library path. None uses the default system Photos library.	`None`

Raises:

Type	Description
`PhotosLibraryNotFoundError`	If specified library path doesn't exist.

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, library_path: Path | None = None) -> None:
    """Initialize Photos library connection.

    Args:
        library_path: Custom library path. None uses the default
            system Photos library.

    Raises:
        PhotosLibraryNotFoundError: If specified library path doesn't exist.
    """
    self._library_path = library_path
    self._db: osxphotos.PhotosDB | None = None
    self._initialized: bool = False
    self._permission_checked: bool = False
    self._has_permission: bool = False

    # Validate custom path if provided
    if library_path is not None and not library_path.exists():
        raise PhotosLibraryNotFoundError(library_path)

library_path `property` ¶

library_path

Get the Photos library path.

Returns:

Type	Description
`Path`	Path to the Photos library (default or custom).

db `property` ¶

db

Get the PhotosDB instance (lazy-loaded).

Returns:

Type	Description
`PhotosDB`	osxphotos.PhotosDB instance.

Raises:

Type	Description
`PhotosAccessDeniedError`	If access is denied.
`PhotosLibraryNotFoundError`	If library cannot be found.

check_permissions ¶

check_permissions()

Check if we have access to the Photos library.

This method attempts to open the Photos library to verify permissions. The result is cached for subsequent calls.

Returns:

Type	Description
`bool`	True if access is granted, False otherwise.

Source code in src/video_converter/extractors/photos_extractor.py

def check_permissions(self) -> bool:
    """Check if we have access to the Photos library.

    This method attempts to open the Photos library to verify
    permissions. The result is cached for subsequent calls.

    Returns:
        True if access is granted, False otherwise.
    """
    if self._permission_checked:
        return self._has_permission

    try:
        # Try to access the library
        _ = self.db.library_path
        self._has_permission = True
        logger.info("Photos library permission check passed")
    except PhotosAccessDeniedError:
        self._has_permission = False
        logger.warning("Photos library permission denied")
    except PhotosLibraryNotFoundError:
        self._has_permission = False
        logger.warning("Photos library not found")
    except Exception as e:
        self._has_permission = False
        logger.warning(f"Photos library permission check failed: {e}")

    self._permission_checked = True
    return self._has_permission

get_library_info ¶

get_library_info()

Get information about the Photos library.

Returns:

Type	Description
`dict[str, str \| int]`	Dictionary with library information: - path: Library path - photo_count: Total number of photos - video_count: Total number of videos

Raises:

Type	Description
`PhotosAccessDeniedError`	If access is denied.

Source code in src/video_converter/extractors/photos_extractor.py

def get_library_info(self) -> dict[str, str | int]:
    """Get information about the Photos library.

    Returns:
        Dictionary with library information:
            - path: Library path
            - photo_count: Total number of photos
            - video_count: Total number of videos

    Raises:
        PhotosAccessDeniedError: If access is denied.
    """
    return {
        "path": str(self.db.library_path),
        "photo_count": len(self.db.photos(images=True, movies=False)),
        "video_count": len(self.db.photos(movies=True, images=False)),
    }

get_video_count ¶

get_video_count()

Get the total number of videos in the library.

Returns:

Type	Description
`int`	Number of videos in the Photos library.

Raises:

Type	Description
`PhotosAccessDeniedError`	If access is denied.

Source code in src/video_converter/extractors/photos_extractor.py

def get_video_count(self) -> int:
    """Get the total number of videos in the library.

    Returns:
        Number of videos in the Photos library.

    Raises:
        PhotosAccessDeniedError: If access is denied.
    """
    return len(self.db.photos(movies=True, images=False))

get_videos ¶

get_videos(*, from_date=None, to_date=None, albums=None, favorites_only=False, include_hidden=False)

Get videos from the Photos library.

Parameters:

Name	Type	Description	Default
`from_date`	`datetime \| None`	Only include videos created on or after this date.	`None`
`to_date`	`datetime \| None`	Only include videos created on or before this date.	`None`
`albums`	`list[str] \| None`	Filter by album names (None for all albums).	`None`
`favorites_only`	`bool`	Only include favorite videos.	`False`
`include_hidden`	`bool`	Include hidden videos.	`False`

Returns:

Type	Description
`list[PhotosVideoInfo]`	List of PhotosVideoInfo objects.

Raises:

Type	Description
`PhotosAccessDeniedError`	If access is denied.

Source code in src/video_converter/extractors/photos_extractor.py

def get_videos(
    self,
    *,
    from_date: datetime | None = None,
    to_date: datetime | None = None,
    albums: list[str] | None = None,
    favorites_only: bool = False,
    include_hidden: bool = False,
) -> list[PhotosVideoInfo]:
    """Get videos from the Photos library.

    Args:
        from_date: Only include videos created on or after this date.
        to_date: Only include videos created on or before this date.
        albums: Filter by album names (None for all albums).
        favorites_only: Only include favorite videos.
        include_hidden: Include hidden videos.

    Returns:
        List of PhotosVideoInfo objects.

    Raises:
        PhotosAccessDeniedError: If access is denied.
    """
    logger.debug(
        f"Querying videos: from_date={from_date}, to_date={to_date}, "
        f"albums={albums}, favorites={favorites_only}, hidden={include_hidden}"
    )

    # Build query parameters
    # Use movies=True, images=False per osxphotos API
    # Ref: https://github.com/RhetTbull/osxphotos
    query_params: dict[str, datetime | bool] = {
        "movies": True,
        "images": False,
    }

    if from_date is not None:
        query_params["from_date"] = from_date

    if to_date is not None:
        query_params["to_date"] = to_date

    # Query videos
    photos = self.db.photos(**query_params)

    # Apply additional filters
    videos: list[PhotosVideoInfo] = []
    for photo in photos:
        # Skip hidden if not requested
        if photo.hidden and not include_hidden:
            continue

        # Skip non-favorites if favorites_only
        if favorites_only and not photo.favorite:
            continue

        # Filter by albums if specified
        if albums is not None:
            photo_albums = [a.title for a in photo.albums]
            if not any(album in photo_albums for album in albums):
                continue

        video_info = self._convert_to_video_info(photo)
        videos.append(video_info)

    logger.info(f"Found {len(videos)} videos matching criteria")
    return videos

get_video_by_uuid ¶

get_video_by_uuid(uuid)

Get a specific video by its UUID.

Parameters:

Name	Type	Description	Default
`uuid`	`str`	The UUID of the video asset.	required

Returns:

Type	Description
`PhotosVideoInfo \| None`	PhotosVideoInfo if found, None otherwise.

Raises:

Type	Description
`PhotosAccessDeniedError`	If access is denied.

Source code in src/video_converter/extractors/photos_extractor.py

def get_video_by_uuid(self, uuid: str) -> PhotosVideoInfo | None:
    """Get a specific video by its UUID.

    Args:
        uuid: The UUID of the video asset.

    Returns:
        PhotosVideoInfo if found, None otherwise.

    Raises:
        PhotosAccessDeniedError: If access is denied.
    """
    photos = self.db.photos(uuid=[uuid])
    if not photos:
        return None

    return self._convert_to_video_info(photos[0])

close ¶

close()

Close the Photos library connection.

This releases any resources held by the connection. The library can be reopened by accessing the db property.

Source code in src/video_converter/extractors/photos_extractor.py

def close(self) -> None:
    """Close the Photos library connection.

    This releases any resources held by the connection.
    The library can be reopened by accessing the db property.
    """
    self._db = None
    self._initialized = False
    self._permission_checked = False
    logger.debug("Photos library connection closed")

enter ¶

__enter__()

Context manager entry.

Returns:

Type	Description
`PhotosLibrary`	Self for context manager usage.

Source code in src/video_converter/extractors/photos_extractor.py

def __enter__(self) -> PhotosLibrary:
    """Context manager entry.

    Returns:
        Self for context manager usage.
    """
    return self

exit ¶

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Parameters:

Name	Type	Description	Default
`exc_type`	`type[BaseException] \| None`	Exception type if an exception occurred.	required
`exc_val`	`BaseException \| None`	Exception value if an exception occurred.	required
`exc_tb`	`object \| None`	Exception traceback if an exception occurred.	required

Source code in src/video_converter/extractors/photos_extractor.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object | None,
) -> None:
    """Context manager exit.

    Args:
        exc_type: Exception type if an exception occurred.
        exc_val: Exception value if an exception occurred.
        exc_tb: Exception traceback if an exception occurred.
    """
    self.close()

LibraryStats `dataclass` ¶

LibraryStats(total=0, h264=0, hevc=0, other=0, in_cloud=0, total_size_h264=0)

Statistics about videos in the Photos library.

Attributes:

Name	Type	Description
`total`	`int`	Total number of videos.
`h264`	`int`	Number of H.264 encoded videos.
`hevc`	`int`	Number of H.265/HEVC encoded videos.
`other`	`int`	Number of videos with other codecs.
`in_cloud`	`int`	Number of videos stored in iCloud only.
`total_size_h264`	`int`	Total size of H.264 videos in bytes.
`estimated_savings`	`int`	Estimated storage savings after conversion in bytes.

estimated_savings `property` ¶

estimated_savings

Estimate storage savings after H.265 conversion.

Assumes approximately 50% size reduction with H.265.

Returns:

Type	Description
`int`	Estimated bytes that could be saved.

estimated_savings_gb `property` ¶

estimated_savings_gb

Get estimated savings in gigabytes.

Returns:

Type	Description
`float`	Estimated savings in GB.

PhotosVideoFilter ¶

PhotosVideoFilter(library, include_albums=None, exclude_albums=None)

Filter Photos library videos for conversion candidates.

This class provides filtering capabilities to identify H.264 videos that need conversion to H.265/HEVC, with support for album-based and date-based filtering.

SDS Reference: SDS-P01-005 SRS Reference: SRS-302 (Video Filtering)

Example

library = PhotosLibrary() filter = PhotosVideoFilter( ... library, ... exclude_albums=["Screenshots", "Bursts"] ... ) candidates = filter.get_conversion_candidates(limit=100) print(f"Found {len(candidates)} videos to convert")

Initialize PhotosVideoFilter.

Parameters:

Name	Type	Description	Default
`library`	`PhotosLibrary`	PhotosLibrary instance to filter.	required
`include_albums`	`list[str] \| None`	Only include videos from these albums. If None, includes all albums.	`None`
`exclude_albums`	`list[str] \| None`	Exclude videos from these albums. If None, uses DEFAULT_EXCLUDE_ALBUMS.	`None`

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(
    self,
    library: PhotosLibrary,
    include_albums: list[str] | None = None,
    exclude_albums: list[str] | None = None,
) -> None:
    """Initialize PhotosVideoFilter.

    Args:
        library: PhotosLibrary instance to filter.
        include_albums: Only include videos from these albums.
            If None, includes all albums.
        exclude_albums: Exclude videos from these albums.
            If None, uses DEFAULT_EXCLUDE_ALBUMS.
    """
    self._library = library
    self._include_albums = set(include_albums) if include_albums else None
    self._exclude_albums = (
        set(exclude_albums) if exclude_albums is not None else set(self.DEFAULT_EXCLUDE_ALBUMS)
    )
    self._codec_detector: CodecDetector | None = None

codec_detector `property` ¶

codec_detector

Get or create codec detector instance.

Returns:

Type	Description
`CodecDetector`	CodecDetector for analyzing video codecs.

get_conversion_candidates ¶

get_conversion_candidates(from_date=None, to_date=None, limit=None)

Get videos that need conversion from H.264 to H.265.

This method filters Photos library videos to find candidates for conversion, applying album and date filters.

Parameters:

Name	Type	Description	Default
`from_date`	`datetime \| None`	Only include videos created after this date.	`None`
`to_date`	`datetime \| None`	Only include videos created before this date.	`None`
`limit`	`int \| None`	Maximum number of candidates to return.	`None`

Returns:

Type	Description
`list[PhotosVideoInfo]`	List of PhotosVideoInfo for videos that need conversion.

Source code in src/video_converter/extractors/photos_extractor.py

def get_conversion_candidates(
    self,
    from_date: datetime | None = None,
    to_date: datetime | None = None,
    limit: int | None = None,
) -> list[PhotosVideoInfo]:
    """Get videos that need conversion from H.264 to H.265.

    This method filters Photos library videos to find candidates
    for conversion, applying album and date filters.

    Args:
        from_date: Only include videos created after this date.
        to_date: Only include videos created before this date.
        limit: Maximum number of candidates to return.

    Returns:
        List of PhotosVideoInfo for videos that need conversion.
    """
    logger.info("Searching for H.264 conversion candidates...")

    candidates: list[PhotosVideoInfo] = []

    # Get all videos with date filtering
    videos = self._library.get_videos(
        from_date=from_date,
        to_date=to_date,
    )

    # Progress tracking
    total_videos = len(videos)
    processed = 0
    skipped_icloud = 0
    skipped_album = 0
    skipped_hevc = 0
    log_interval = max(100, total_videos // 20)  # Log every 5% or 100 videos

    logger.info(f"Analyzing {total_videos} videos for H.264 codec...")

    for video in videos:
        processed += 1

        # Log progress periodically
        if processed % log_interval == 0 or processed == total_videos:
            progress_pct = (processed / total_videos) * 100
            logger.info(
                f"Progress: {processed}/{total_videos} ({progress_pct:.1f}%) - "
                f"Found {len(candidates)} H.264, "
                f"Skipped: {skipped_icloud} iCloud, {skipped_album} album filter, "
                f"{skipped_hevc} already HEVC"
            )

        # Skip if doesn't pass album filter
        if not self._passes_album_filter(video):
            skipped_album += 1
            continue

        # Skip if not available locally
        if not video.is_available_locally:
            skipped_icloud += 1
            logger.debug(f"Skipping iCloud-only video: {video.filename}")
            continue

        # Enrich with codec info
        enriched = self._enrich_with_codec(video)

        # Check if needs conversion
        if enriched.needs_conversion:
            candidates.append(enriched)
            logger.debug(f"Found candidate: {enriched.filename} ({enriched.codec})")

            if limit and len(candidates) >= limit:
                logger.info(f"Reached limit of {limit} candidates")
                break
        elif enriched.is_hevc:
            skipped_hevc += 1

    logger.info(
        f"Scan complete: {len(candidates)} H.264 videos found for conversion "
        f"(Skipped: {skipped_icloud} iCloud, {skipped_album} album filter, "
        f"{skipped_hevc} already HEVC)"
    )
    return candidates

get_stats ¶

get_stats()

Get statistics about videos in the library.

Analyzes all videos in the library to provide statistics about codec distribution and potential storage savings.

Returns:

Type	Description
`LibraryStats`	LibraryStats with codec distribution and size information.

Source code in src/video_converter/extractors/photos_extractor.py

def get_stats(self) -> LibraryStats:
    """Get statistics about videos in the library.

    Analyzes all videos in the library to provide statistics
    about codec distribution and potential storage savings.

    Returns:
        LibraryStats with codec distribution and size information.
    """
    logger.info("Analyzing library statistics...")

    stats = LibraryStats()

    videos = self._library.get_videos()
    stats.total = len(videos)

    for video in videos:
        # Track iCloud-only videos
        if not video.is_available_locally:
            stats.in_cloud += 1
            continue

        # Detect codec
        codec = self._detect_codec(video)
        if codec is None:
            stats.other += 1
            continue

        codec_lower = codec.lower()
        if codec_lower in PhotosVideoInfo.H264_CODECS:
            stats.h264 += 1
            try:
                if video.path:
                    stats.total_size_h264 += video.path.stat().st_size
            except OSError:
                pass
        elif codec_lower in PhotosVideoInfo.HEVC_CODECS:
            stats.hevc += 1
        else:
            stats.other += 1

    logger.info(
        f"Library stats: {stats.total} total, {stats.h264} H.264, "
        f"{stats.hevc} HEVC, {stats.in_cloud} in cloud"
    )
    return stats

VideoExporter ¶

VideoExporter(temp_dir=None)

Export videos from Photos library to temporary directory.

This class provides functionality to export original quality videos from the Photos library to a temporary directory for conversion, with support for progress tracking and cleanup.

SDS Reference: SDS-P01-006 SRS Reference: SRS-303 (Video Export)

Example

exporter = VideoExporter() video = library.get_video_by_uuid("some-uuid") if video and video.is_available_locally: ... exported_path = exporter.export(video) ... # ... do conversion ... ... exporter.cleanup(exported_path) exporter.cleanup_all()

Attributes:

Name	Type	Description
`temp_dir`	`Path`	Directory where exported files are stored.

Initialize video exporter.

Parameters:

Name	Type	Description	Default
`temp_dir`	`Path \| None`	Custom temporary directory for exports. If None, creates a system temporary directory.	`None`

Source code in src/video_converter/extractors/photos_extractor.py

def __init__(self, temp_dir: Path | None = None) -> None:
    """Initialize video exporter.

    Args:
        temp_dir: Custom temporary directory for exports.
            If None, creates a system temporary directory.
    """
    if temp_dir is not None:
        self._temp_dir = temp_dir
        self._temp_dir.mkdir(parents=True, exist_ok=True)
        self._owns_temp_dir = False
    else:
        self._temp_dir = Path(tempfile.mkdtemp(prefix="video_converter_"))
        self._owns_temp_dir = True

    self._exported_files: set[Path] = set()
    logger.debug(f"VideoExporter initialized with temp_dir: {self._temp_dir}")

temp_dir `property` ¶

temp_dir

Get the temporary directory path.

Returns:

Type	Description
`Path`	Path to the temporary directory.

export ¶

export(video, on_progress=None)

Export video to temporary directory.

Copies the original video file to the temporary directory, preserving file metadata (modification times).

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video information from Photos library.	required
`on_progress`	`Callable[[float], None] \| None`	Optional callback for progress updates. Called with progress value from 0.0 to 1.0.	`None`

Returns:

Type	Description
`Path`	Path to the exported video file.

Raises:

Type	Description
`VideoNotAvailableError`	Video is in iCloud and not downloaded.
`ExportError`	Export failed due to file system error.

Source code in src/video_converter/extractors/photos_extractor.py

def export(
    self,
    video: PhotosVideoInfo,
    on_progress: Callable[[float], None] | None = None,
) -> Path:
    """Export video to temporary directory.

    Copies the original video file to the temporary directory,
    preserving file metadata (modification times).

    Args:
        video: Video information from Photos library.
        on_progress: Optional callback for progress updates.
            Called with progress value from 0.0 to 1.0.

    Returns:
        Path to the exported video file.

    Raises:
        VideoNotAvailableError: Video is in iCloud and not downloaded.
        ExportError: Export failed due to file system error.
    """
    # Validate video availability
    if video.in_cloud and not video.is_available_locally:
        raise VideoNotAvailableError(video.filename)

    if video.path is None:
        raise ExportError("No path available", video.filename)

    if not video.path.exists():
        raise ExportError("Source file does not exist", video.filename)

    # Generate unique output path to avoid collisions
    output_path = self._temp_dir / f"{video.uuid}_{video.filename}"

    logger.info(f"Exporting video: {video.filename}")
    logger.debug(f"Source: {video.path}")
    logger.debug(f"Destination: {output_path}")

    try:
        if on_progress is not None:
            self._copy_with_progress(video.path, output_path, on_progress)
        else:
            # Use shutil.copy2 to preserve metadata
            shutil.copy2(video.path, output_path)

        self._exported_files.add(output_path)
        logger.info(f"Exported: {video.filename} -> {output_path.name}")
        return output_path

    except PermissionError as e:
        raise ExportError(f"Permission denied: {e}", video.filename) from e
    except OSError as e:
        # Clean up partial file if it exists
        if output_path.exists():
            try:
                output_path.unlink()
            except OSError:
                pass
        raise ExportError(str(e), video.filename) from e

cleanup ¶

cleanup(path)

Remove a single exported file.

Only removes files that are within the temporary directory and were exported by this exporter.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the exported file to remove.	required

Returns:

Type	Description
`bool`	True if file was removed, False otherwise.

Source code in src/video_converter/extractors/photos_extractor.py

def cleanup(self, path: Path) -> bool:
    """Remove a single exported file.

    Only removes files that are within the temporary directory
    and were exported by this exporter.

    Args:
        path: Path to the exported file to remove.

    Returns:
        True if file was removed, False otherwise.
    """
    # Safety check: only remove files in our temp directory
    try:
        path.relative_to(self._temp_dir)
    except ValueError:
        logger.warning(f"Refusing to cleanup file outside temp_dir: {path}")
        return False

    if path.exists():
        try:
            path.unlink()
            self._exported_files.discard(path)
            logger.debug(f"Cleaned up: {path.name}")
            return True
        except OSError as e:
            logger.warning(f"Failed to cleanup {path.name}: {e}")
            return False
    return False

cleanup_all ¶

cleanup_all()

Remove all exported files and the temporary directory.

Only removes the temporary directory if it was created by this exporter (not a custom directory).

Returns:

Type	Description
`int`	Number of files removed.

Source code in src/video_converter/extractors/photos_extractor.py

def cleanup_all(self) -> int:
    """Remove all exported files and the temporary directory.

    Only removes the temporary directory if it was created by
    this exporter (not a custom directory).

    Returns:
        Number of files removed.
    """
    removed_count = 0

    # Remove tracked files
    for path in list(self._exported_files):
        if self.cleanup(path):
            removed_count += 1

    # Remove temp directory if we own it
    if self._owns_temp_dir and self._temp_dir.exists():
        try:
            shutil.rmtree(self._temp_dir, ignore_errors=True)
            logger.info(f"Removed temporary directory: {self._temp_dir}")
        except OSError as e:
            logger.warning(f"Failed to remove temp directory: {e}")

    self._exported_files.clear()
    return removed_count

get_exported_count ¶

get_exported_count()

Get the number of currently exported files.

Returns:

Type	Description
`int`	Number of files in the exported files set.

Source code in src/video_converter/extractors/photos_extractor.py

def get_exported_count(self) -> int:
    """Get the number of currently exported files.

    Returns:
        Number of files in the exported files set.
    """
    return len(self._exported_files)

get_temp_dir_size ¶

get_temp_dir_size()

Get total size of files in the temporary directory.

Returns:

Type	Description
`int`	Total size in bytes.

Source code in src/video_converter/extractors/photos_extractor.py

def get_temp_dir_size(self) -> int:
    """Get total size of files in the temporary directory.

    Returns:
        Total size in bytes.
    """
    total = 0
    if self._temp_dir.exists():
        for path in self._temp_dir.iterdir():
            if path.is_file():
                try:
                    total += path.stat().st_size
                except OSError:
                    pass
    return total

enter ¶

__enter__()

Context manager entry.

Returns:

Type	Description
`VideoExporter`	Self for context manager usage.

Source code in src/video_converter/extractors/photos_extractor.py

def __enter__(self) -> VideoExporter:
    """Context manager entry.

    Returns:
        Self for context manager usage.
    """
    return self

exit ¶

__exit__(exc_type, exc_val, exc_tb)

Context manager exit - cleanup all exported files.

Parameters:

Name	Type	Description	Default
`exc_type`	`type[BaseException] \| None`	Exception type if an exception occurred.	required
`exc_val`	`BaseException \| None`	Exception value if an exception occurred.	required
`exc_tb`	`object \| None`	Exception traceback if an exception occurred.	required

Source code in src/video_converter/extractors/photos_extractor.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object | None,
) -> None:
    """Context manager exit - cleanup all exported files.

    Args:
        exc_type: Exception type if an exception occurred.
        exc_val: Exception value if an exception occurred.
        exc_tb: Exception traceback if an exception occurred.
    """
    self.cleanup_all()

get_permission_instructions ¶

get_permission_instructions()

Get instructions for granting Photos library access.

Returns:

Type	Description
`str`	Multi-line string with detailed instructions for granting
`str`	Full Disk Access permission.

Source code in src/video_converter/extractors/photos_extractor.py

def get_permission_instructions() -> str:
    """Get instructions for granting Photos library access.

    Returns:
        Multi-line string with detailed instructions for granting
        Full Disk Access permission.
    """
    return """
Photos Library Access Denied

Video Converter needs access to your Photos library.

To grant access:
1. Open System Settings → Privacy & Security → Full Disk Access
2. Click the + button
3. Navigate to: /Applications/Utilities/Terminal.app
   (or the application running this script)
4. Add it to the list and enable the toggle
5. Restart the application and try again

Quick access command:
  open "x-apple.systempreferences:com.apple.preference.security?Privacy_AllFiles"
""".strip()

Folder Extractor¶

folder_extractor ¶

Folder-based video extractor module.

This module provides video extraction from filesystem directories, allowing users to convert videos stored in regular folders instead of the macOS Photos library.

SDS Reference: SDS-E01-003 SRS Reference: SRS-304 (Folder-based Video Extraction)

Example

extractor = FolderExtractor(Path("~/Videos/iPhone")) for video in extractor.scan(): ... print(f"Found: {video}") ... candidates = extractor.get_conversion_candidates() print(f"Found {len(candidates)} H.264 videos to convert")

FolderExtractorError ¶

Bases: Exception

Base exception for folder extractor operations.

FolderNotFoundError ¶

FolderNotFoundError(path)

Bases: FolderExtractorError

Raised when the specified folder does not exist.

Attributes:

Name	Type	Description
`path`		The path that was not found.

Initialize with the missing folder path.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the folder that was not found.	required

Source code in src/video_converter/extractors/folder_extractor.py

def __init__(self, path: Path) -> None:
    """Initialize with the missing folder path.

    Args:
        path: Path to the folder that was not found.
    """
    self.path = path
    super().__init__(f"Folder not found: {path}")

FolderAccessDeniedError ¶

FolderAccessDeniedError(path)

Bases: FolderExtractorError

Raised when access to a folder is denied.

Attributes:

Name	Type	Description
`path`		The path that could not be accessed.

Initialize with the inaccessible folder path.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the folder that could not be accessed.	required

Source code in src/video_converter/extractors/folder_extractor.py

def __init__(self, path: Path) -> None:
    """Initialize with the inaccessible folder path.

    Args:
        path: Path to the folder that could not be accessed.
    """
    self.path = path
    super().__init__(f"Access denied to folder: {path}")

InvalidVideoFileError ¶

InvalidVideoFileError(path, reason='Invalid or corrupted video file')

Bases: FolderExtractorError

Raised when a video file cannot be read or analyzed.

Attributes:

Name	Type	Description
`path`		Path to the invalid video file.
`reason`		Description of why the file is invalid.

Initialize with the invalid file path and reason.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the invalid video file.	required
`reason`	`str`	Description of why the file is invalid.	`'Invalid or corrupted video file'`

Source code in src/video_converter/extractors/folder_extractor.py

def __init__(self, path: Path, reason: str = "Invalid or corrupted video file") -> None:
    """Initialize with the invalid file path and reason.

    Args:
        path: Path to the invalid video file.
        reason: Description of why the file is invalid.
    """
    self.path = path
    self.reason = reason
    super().__init__(f"Invalid video file '{path}': {reason}")

FolderVideoInfo `dataclass` ¶

FolderVideoInfo(path, filename, size, modified_time, created_time=None, codec=None, duration=0.0, width=0, height=0, fps=0.0, bitrate=0, container='', in_cloud=False, stub_path=None)

Information about a video file in a folder.

Attributes:

Name	Type	Description
`path`	`Path`	Absolute path to the video file (original path, not stub).
`filename`	`str`	Name of the video file.
`size`	`int`	File size in bytes.
`modified_time`	`datetime`	Last modification time.
`created_time`	`datetime \| None`	Creation time (if available).
`codec`	`str \| None`	Video codec name (e.g., "h264", "hevc").
`duration`	`float`	Video duration in seconds.
`width`	`int`	Video width in pixels.
`height`	`int`	Video height in pixels.
`fps`	`float`	Frames per second.
`bitrate`	`int`	Video bitrate in bits per second.
`container`	`str`	Container format (e.g., "mp4", "mov").
`in_cloud`	`bool`	Whether file is stored in iCloud (stub file exists locally).
`stub_path`	`Path \| None`	Path to iCloud stub file if in_cloud is True.

is_h264 `property` ¶

is_h264

Check if video codec is H.264/AVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.264.

is_hevc `property` ¶

is_hevc

Check if video codec is H.265/HEVC.

Returns:

Type	Description
`bool`	True if the codec is any variant of H.265/HEVC.

needs_conversion `property` ¶

needs_conversion

Check if video needs H.265 conversion.

A video needs conversion if its codec is H.264.

Returns:

Type	Description
`bool`	True if the video should be converted to H.265.

resolution_label `property` ¶

resolution_label

Get human-readable resolution label.

Returns:

Type	Description
`str`	Resolution label like "4K", "1080p", "720p", etc.

size_mb `property` ¶

size_mb

Get file size in megabytes.

Returns:

Type	Description
`float`	File size in MB.

size_gb `property` ¶

size_gb

Get file size in gigabytes.

Returns:

Type	Description
`float`	File size in GB.

str ¶

__str__()

Return human-readable summary.

Source code in src/video_converter/extractors/folder_extractor.py

def __str__(self) -> str:
    """Return human-readable summary."""
    codec_str = self.codec.upper() if self.codec else "unknown"
    return f"{self.filename}: {codec_str} {self.resolution_label} ({self.size_mb:.1f} MB)"

FolderStats `dataclass` ¶

FolderStats(total=0, h264=0, hevc=0, other=0, errors=0, total_size=0, h264_size=0, in_cloud=0)

Statistics about videos in a folder.

Attributes:

Name	Type	Description
`total`	`int`	Total number of video files.
`h264`	`int`	Number of H.264 encoded videos.
`hevc`	`int`	Number of H.265/HEVC encoded videos.
`other`	`int`	Number of videos with other codecs.
`errors`	`int`	Number of files that could not be analyzed.
`total_size`	`int`	Total size of all videos in bytes.
`h264_size`	`int`	Total size of H.264 videos in bytes.
`in_cloud`	`int`	Number of videos stored in iCloud (stub files only).

estimated_savings `property` ¶

estimated_savings

Estimate storage savings after H.265 conversion.

Assumes approximately 50% size reduction with H.265.

Returns:

Type	Description
`int`	Estimated bytes that could be saved.

estimated_savings_gb `property` ¶

estimated_savings_gb

Get estimated savings in gigabytes.

Returns:

Type	Description
`float`	Estimated savings in GB.

total_size_gb `property` ¶

total_size_gb

Get total size in gigabytes.

Returns:

Type	Description
`float`	Total size in GB.

FolderExtractor ¶

FolderExtractor(root_path, *, recursive=True, include_patterns=None, exclude_patterns=None, video_extensions=None)

Extract videos from filesystem folders.

This class provides video extraction from regular filesystem directories, with support for recursive scanning, file filtering, codec detection, and iCloud stub file detection.

SDS Reference: SDS-E01-003 SRS Reference: SRS-304 (Folder-based Video Extraction)

Example

extractor = FolderExtractor( ... root_path=Path("~/Videos/iPhone"), ... recursive=True, ... exclude_patterns=[".tmp", "._"] ... ) candidates = extractor.get_conversion_candidates() for video in candidates: ... print(f"{video.filename}: {video.codec} (in_cloud={video.in_cloud})")

Attributes:

Name	Type	Description
`root_path`	`Path`	Root directory for scanning.
`recursive`	`bool`	Whether to scan subdirectories.
`include_patterns`	`bool`	Glob patterns to include.
`exclude_patterns`	`bool`	Glob patterns to exclude.

Initialize folder extractor.

Parameters:

Name	Type	Description	Default
`root_path`	`Path \| str`	Root directory to scan for videos.	required
`recursive`	`bool`	Scan subdirectories if True.	`True`
`include_patterns`	`list[str] \| None`	Glob patterns to include (e.g., ["vacation*"]). If None, includes all files matching video extensions.	`None`
`exclude_patterns`	`list[str] \| None`	Glob patterns to exclude (e.g., ["*.tmp"]). If None, uses DEFAULT_EXCLUDE_PATTERNS.	`None`
`video_extensions`	`set[str] \| None`	Custom set of video file extensions. If None, uses VIDEO_EXTENSIONS.	`None`

Raises:

Type	Description
`FolderNotFoundError`	If root_path does not exist.
`FolderAccessDeniedError`	If root_path is not readable.

Source code in src/video_converter/extractors/folder_extractor.py

def __init__(
    self,
    root_path: Path | str,
    *,
    recursive: bool = True,
    include_patterns: list[str] | None = None,
    exclude_patterns: list[str] | None = None,
    video_extensions: set[str] | None = None,
) -> None:
    """Initialize folder extractor.

    Args:
        root_path: Root directory to scan for videos.
        recursive: Scan subdirectories if True.
        include_patterns: Glob patterns to include (e.g., ["vacation*"]).
            If None, includes all files matching video extensions.
        exclude_patterns: Glob patterns to exclude (e.g., ["*.tmp"]).
            If None, uses DEFAULT_EXCLUDE_PATTERNS.
        video_extensions: Custom set of video file extensions.
            If None, uses VIDEO_EXTENSIONS.

    Raises:
        FolderNotFoundError: If root_path does not exist.
        FolderAccessDeniedError: If root_path is not readable.
    """
    self._root_path = Path(root_path).expanduser().resolve()
    self._recursive = recursive
    self._include_patterns = list(include_patterns) if include_patterns else []
    self._exclude_patterns = (
        set(exclude_patterns)
        if exclude_patterns is not None
        else set(self.DEFAULT_EXCLUDE_PATTERNS)
    )
    self._video_extensions = video_extensions or self.VIDEO_EXTENSIONS
    self._codec_detector: CodecDetector | None = None

    # Validate root path
    if not self._root_path.exists():
        raise FolderNotFoundError(self._root_path)

    if not self._root_path.is_dir():
        raise FolderNotFoundError(self._root_path)

    # Check read access
    try:
        list(self._root_path.iterdir())
    except PermissionError as e:
        raise FolderAccessDeniedError(self._root_path) from e

    logger.info(
        f"FolderExtractor initialized: {self._root_path} "
        f"(recursive={recursive}, extensions={len(self._video_extensions)})"
    )

root_path `property` ¶

root_path

Get the root directory path.

Returns:

Type	Description
`Path`	Path to the root directory.

recursive `property` ¶

recursive

Check if recursive scanning is enabled.

Returns:

Type	Description
`bool`	True if scanning subdirectories.

codec_detector `property` ¶

codec_detector

Get or create codec detector instance.

Returns:

Type	Description
`CodecDetector`	CodecDetector for analyzing video codecs.

scan ¶

scan(*, include_icloud=True)

Scan for video files in the root directory.

Yields video file paths that match the configured filters. When iCloud stub files are found, the original file path is yielded instead of the stub path.

Parameters:

Name	Type	Description	Default
`include_icloud`	`bool`	Whether to include iCloud stub files in scan. Default is True.	`True`

Yields:

Type	Description
`Path`	Path to each video file found (original path for iCloud files).

Example

extractor = FolderExtractor(Path("~/Videos")) for video_path in extractor.scan(): ... print(video_path)

Source code in src/video_converter/extractors/folder_extractor.py

def scan(self, *, include_icloud: bool = True) -> Iterator[Path]:
    """Scan for video files in the root directory.

    Yields video file paths that match the configured filters.
    When iCloud stub files are found, the original file path is yielded
    instead of the stub path.

    Args:
        include_icloud: Whether to include iCloud stub files in scan.
                       Default is True.

    Yields:
        Path to each video file found (original path for iCloud files).

    Example:
        >>> extractor = FolderExtractor(Path("~/Videos"))
        >>> for video_path in extractor.scan():
        ...     print(video_path)
    """
    logger.info(f"Scanning for videos in: {self._root_path}")

    pattern = "**/*" if self._recursive else "*"

    count = 0
    icloud_count = 0
    seen_paths: set[Path] = set()

    for path in self._root_path.glob(pattern):
        # Skip directories
        if not path.is_file():
            continue

        # Check for iCloud stub files
        if include_icloud and self._is_video_stub(path):
            original_path = self._get_original_path_from_stub(path)

            # Skip if we've already seen this file (as local file)
            if original_path in seen_paths:
                continue

            # Apply filters to original filename
            if not self._passes_filters(original_path):
                continue

            seen_paths.add(original_path)
            count += 1
            icloud_count += 1
            logger.debug(f"Found iCloud stub: {path.name} -> {original_path.name}")
            yield original_path
            continue

        # Skip non-video files
        if not self._is_video_file(path):
            continue

        # Skip if we've already seen this file (as stub)
        if path in seen_paths:
            continue

        # Apply filters
        if not self._passes_filters(path):
            continue

        seen_paths.add(path)
        count += 1
        yield path

    logger.info(f"Scan complete: found {count} video files ({icloud_count} in iCloud)")

get_video_info ¶

get_video_info(path)

Get video information for a single file.

Handles both local files and iCloud stub files. For iCloud files, the path parameter should be the original file path (not stub path).

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the video file (original path, not stub).	required

Returns:

Type	Description
`FolderVideoInfo`	FolderVideoInfo with file and video properties.
`FolderVideoInfo`	For iCloud files, in_cloud=True and stub_path is set.

Raises:

Type	Description
`FileNotFoundError`	If neither the file nor its stub exists.
`InvalidVideoFileError`	If the file cannot be analyzed.

Source code in src/video_converter/extractors/folder_extractor.py

def get_video_info(self, path: Path) -> FolderVideoInfo:
    """Get video information for a single file.

    Handles both local files and iCloud stub files. For iCloud files,
    the path parameter should be the original file path (not stub path).

    Args:
        path: Path to the video file (original path, not stub).

    Returns:
        FolderVideoInfo with file and video properties.
        For iCloud files, in_cloud=True and stub_path is set.

    Raises:
        FileNotFoundError: If neither the file nor its stub exists.
        InvalidVideoFileError: If the file cannot be analyzed.
    """
    # Check for iCloud stub
    stub_path = self._get_stub_path(path)
    in_cloud = stub_path.exists() and not path.exists()

    if in_cloud:
        # File is in iCloud, get info from stub
        logger.debug(f"File is in iCloud: {path.name}")
        try:
            stat = stub_path.stat()
            # Stub file size is not the actual video size, set to 0
            size = 0
            modified_time = datetime.fromtimestamp(stat.st_mtime)
            try:
                created_time = datetime.fromtimestamp(stat.st_birthtime)
            except AttributeError:
                created_time = None
        except OSError as e:
            logger.warning(f"Failed to get stub file stats for {path}: {e}")
            size = 0
            modified_time = datetime.now()
            created_time = None

        # Cannot analyze codec for iCloud-only files
        return FolderVideoInfo(
            path=path,
            filename=path.name,
            size=size,
            modified_time=modified_time,
            created_time=created_time,
            codec=None,
            duration=0.0,
            width=0,
            height=0,
            fps=0.0,
            bitrate=0,
            container=path.suffix.lstrip(".").lower(),
            in_cloud=True,
            stub_path=stub_path,
        )

    # File exists locally
    if not path.exists():
        raise FileNotFoundError(f"Video file not found: {path}")

    # Get file stats
    try:
        stat = path.stat()
        size = stat.st_size
        modified_time = datetime.fromtimestamp(stat.st_mtime)
        created_time = datetime.fromtimestamp(stat.st_birthtime)
    except (OSError, AttributeError) as e:
        logger.warning(f"Failed to get file stats for {path}: {e}")
        size = 0
        modified_time = datetime.now()
        created_time = None

    # Analyze codec
    codec_info: CodecInfo | None = None
    # Import exceptions first to ensure they're available in except clause
    from video_converter.processors.codec_detector import (
        CorruptedVideoError,
        InvalidVideoError,
    )

    try:
        codec_info = self.codec_detector.analyze(path)
    except (InvalidVideoError, CorruptedVideoError) as e:
        logger.warning(f"Failed to analyze video codec for {path}: {e}")
    except Exception as e:
        logger.warning(f"Unexpected error analyzing {path}: {e}")

    return FolderVideoInfo(
        path=path,
        filename=path.name,
        size=size,
        modified_time=modified_time,
        created_time=created_time,
        codec=codec_info.codec if codec_info else None,
        duration=codec_info.duration if codec_info else 0.0,
        width=codec_info.width if codec_info else 0,
        height=codec_info.height if codec_info else 0,
        fps=codec_info.fps if codec_info else 0.0,
        bitrate=codec_info.bitrate if codec_info else 0,
        container=codec_info.container if codec_info else "",
        in_cloud=False,
        stub_path=None,
    )

get_videos ¶

get_videos()

Get all videos with their information.

Scans the folder and retrieves detailed information for each video.

Returns:

Type	Description
`list[FolderVideoInfo]`	List of FolderVideoInfo for all videos found.

Example

extractor = FolderExtractor(Path("~/Videos")) videos = extractor.get_videos() for video in videos: ... print(f"{video.filename}: {video.codec}")

Source code in src/video_converter/extractors/folder_extractor.py

def get_videos(self) -> list[FolderVideoInfo]:
    """Get all videos with their information.

    Scans the folder and retrieves detailed information for each video.

    Returns:
        List of FolderVideoInfo for all videos found.

    Example:
        >>> extractor = FolderExtractor(Path("~/Videos"))
        >>> videos = extractor.get_videos()
        >>> for video in videos:
        ...     print(f"{video.filename}: {video.codec}")
    """
    logger.info("Getting video information for all files...")
    videos: list[FolderVideoInfo] = []

    for path in self.scan():
        try:
            video_info = self.get_video_info(path)
            videos.append(video_info)
        except (FileNotFoundError, InvalidVideoFileError) as e:
            logger.warning(f"Skipping {path}: {e}")

    logger.info(f"Retrieved information for {len(videos)} videos")
    return videos

get_conversion_candidates ¶

get_conversion_candidates(*, limit=None)

Get videos that need conversion from H.264 to H.265.

This method scans the folder and identifies H.264 videos that are candidates for conversion.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	Maximum number of candidates to return.	`None`

Returns:

Type	Description
`list[FolderVideoInfo]`	List of FolderVideoInfo for H.264 videos.

Example

extractor = FolderExtractor(Path("~/Videos")) candidates = extractor.get_conversion_candidates(limit=10) print(f"Found {len(candidates)} videos to convert")

Source code in src/video_converter/extractors/folder_extractor.py

def get_conversion_candidates(
    self,
    *,
    limit: int | None = None,
) -> list[FolderVideoInfo]:
    """Get videos that need conversion from H.264 to H.265.

    This method scans the folder and identifies H.264 videos
    that are candidates for conversion.

    Args:
        limit: Maximum number of candidates to return.

    Returns:
        List of FolderVideoInfo for H.264 videos.

    Example:
        >>> extractor = FolderExtractor(Path("~/Videos"))
        >>> candidates = extractor.get_conversion_candidates(limit=10)
        >>> print(f"Found {len(candidates)} videos to convert")
    """
    logger.info("Searching for H.264 conversion candidates...")
    candidates: list[FolderVideoInfo] = []

    for path in self.scan():
        try:
            video_info = self.get_video_info(path)

            if video_info.needs_conversion:
                candidates.append(video_info)
                logger.debug(
                    f"Found candidate: {video_info.filename} "
                    f"({video_info.codec}, {video_info.size_mb:.1f} MB)"
                )

                if limit is not None and len(candidates) >= limit:
                    break

        except (FileNotFoundError, InvalidVideoFileError) as e:
            logger.warning(f"Skipping {path}: {e}")

    logger.info(f"Found {len(candidates)} H.264 videos for conversion")
    return candidates

get_stats ¶

get_stats()

Get statistics about videos in the folder.

Analyzes all videos to provide statistics about codec distribution, potential storage savings, and iCloud status.

Returns:

Type	Description
`FolderStats`	FolderStats with codec distribution, size, and iCloud information.

Example

extractor = FolderExtractor(Path("~/Videos")) stats = extractor.get_stats() print(f"Found {stats.h264} H.264 videos") print(f"In iCloud: {stats.in_cloud}") print(f"Estimated savings: {stats.estimated_savings_gb:.1f} GB")

Source code in src/video_converter/extractors/folder_extractor.py

def get_stats(self) -> FolderStats:
    """Get statistics about videos in the folder.

    Analyzes all videos to provide statistics about codec distribution,
    potential storage savings, and iCloud status.

    Returns:
        FolderStats with codec distribution, size, and iCloud information.

    Example:
        >>> extractor = FolderExtractor(Path("~/Videos"))
        >>> stats = extractor.get_stats()
        >>> print(f"Found {stats.h264} H.264 videos")
        >>> print(f"In iCloud: {stats.in_cloud}")
        >>> print(f"Estimated savings: {stats.estimated_savings_gb:.1f} GB")
    """
    logger.info("Analyzing folder statistics...")
    stats = FolderStats()

    for path in self.scan():
        stats.total += 1

        try:
            video_info = self.get_video_info(path)

            # Track iCloud status
            if video_info.in_cloud:
                stats.in_cloud += 1
                # Cannot get size for iCloud-only files
                size = 0
            else:
                try:
                    stat = path.stat()
                    size = stat.st_size
                    stats.total_size += size
                except OSError:
                    size = 0

            if video_info.is_h264:
                stats.h264 += 1
                stats.h264_size += size
            elif video_info.is_hevc:
                stats.hevc += 1
            else:
                stats.other += 1

        except (FileNotFoundError, InvalidVideoFileError):
            stats.errors += 1
        except Exception as e:
            logger.warning(f"Error analyzing {path}: {e}")
            stats.errors += 1

    logger.info(
        f"Folder stats: {stats.total} total, {stats.h264} H.264, "
        f"{stats.hevc} HEVC, {stats.other} other, "
        f"{stats.in_cloud} in iCloud, {stats.errors} errors"
    )
    return stats

get_video_count ¶

get_video_count()

Get the total number of video files in the folder.

Returns:

Type	Description
`int`	Number of video files matching the configured filters.

Source code in src/video_converter/extractors/folder_extractor.py

def get_video_count(self) -> int:
    """Get the total number of video files in the folder.

    Returns:
        Number of video files matching the configured filters.
    """
    return sum(1 for _ in self.scan())

enter ¶

__enter__()

Context manager entry.

Returns:

Type	Description
`FolderExtractor`	Self for context manager usage.

Source code in src/video_converter/extractors/folder_extractor.py

def __enter__(self) -> FolderExtractor:
    """Context manager entry.

    Returns:
        Self for context manager usage.
    """
    return self

exit ¶

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Parameters:

Name	Type	Description	Default
`exc_type`	`type[BaseException] \| None`	Exception type if an exception occurred.	required
`exc_val`	`BaseException \| None`	Exception value if an exception occurred.	required
`exc_tb`	`object \| None`	Exception traceback if an exception occurred.	required

Source code in src/video_converter/extractors/folder_extractor.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object | None,
) -> None:
    """Context manager exit.

    Args:
        exc_type: Exception type if an exception occurred.
        exc_val: Exception value if an exception occurred.
        exc_tb: Exception traceback if an exception occurred.
    """
    # No resources to clean up currently
    pass

repr ¶

__repr__()

Return string representation.

Source code in src/video_converter/extractors/folder_extractor.py

def __repr__(self) -> str:
    """Return string representation."""
    return f"FolderExtractor(root_path={self._root_path!r}, recursive={self._recursive})"

iCloud Handler¶

icloud_handler ¶

iCloud video download handling module.

This module provides functionality for detecting and downloading videos stored in iCloud, including download triggering, progress tracking, and wait-for-download functionality.

SDS Reference: SDS-P01-007 SRS Reference: SRS-304 (iCloud Download Handling)

Example

from video_converter.extractors.icloud_handler import iCloudHandler from video_converter.extractors.photos_extractor import PhotosVideoInfo handler = iCloudHandler(timeout=1800) status = handler.get_status(video) if status == CloudStatus.CLOUD_ONLY: ... success = handler.download_and_wait(video)

iCloudError ¶

Bases: Exception

Base exception for iCloud operations.

iCloudDownloadError ¶

iCloudDownloadError(filename, reason='Unknown error')

Bases: iCloudError

Raised when iCloud download fails.

This exception indicates that a video file could not be downloaded from iCloud, either due to network issues, timeout, or other errors.

Initialize with video filename and failure reason.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Name of the video file that failed to download.	required
`reason`	`str`	Description of why the download failed.	`'Unknown error'`

Source code in src/video_converter/extractors/icloud_handler.py

def __init__(
    self,
    filename: str,
    reason: str = "Unknown error",
) -> None:
    """Initialize with video filename and failure reason.

    Args:
        filename: Name of the video file that failed to download.
        reason: Description of why the download failed.
    """
    message = f"Failed to download '{filename}' from iCloud: {reason}"
    super().__init__(message)
    self.filename = filename
    self.reason = reason

iCloudTimeoutError ¶

iCloudTimeoutError(filename, timeout)

Bases: iCloudError

Raised when iCloud download times out.

This exception indicates that the download did not complete within the specified timeout period.

Initialize with video filename and timeout duration.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Name of the video file.	required
`timeout`	`int`	Timeout duration in seconds.	required

Source code in src/video_converter/extractors/icloud_handler.py

def __init__(self, filename: str, timeout: int) -> None:
    """Initialize with video filename and timeout duration.

    Args:
        filename: Name of the video file.
        timeout: Timeout duration in seconds.
    """
    message = f"Download of '{filename}' timed out after {timeout} seconds"
    super().__init__(message)
    self.filename = filename
    self.timeout = timeout

CloudStatus ¶

Bases: Enum

iCloud storage status for a video file.

Attributes:

Name	Type	Description
`LOCAL`		Full file is available locally.
`CLOUD_ONLY`		File is stored in iCloud only (stub file locally).
`DOWNLOADING`		File is currently being downloaded from iCloud.
`FAILED`		Download attempt failed.
`UNKNOWN`		Status cannot be determined.

DownloadProgress `dataclass` ¶

DownloadProgress(filename, status, progress=-1.0, bytes_downloaded=0, bytes_total=0, elapsed_seconds=0.0)

Progress information for iCloud download.

Attributes:

Name	Type	Description
`filename`	`str`	Name of the file being downloaded.
`status`	`CloudStatus`	Current download status.
`progress`	`float`	Download progress percentage (0-100), -1 if indeterminate.
`bytes_downloaded`	`int`	Bytes downloaded so far.
`bytes_total`	`int`	Total file size in bytes.
`elapsed_seconds`	`float`	Time elapsed since download started.

is_complete `property` ¶

is_complete

Check if download is complete.

is_failed `property` ¶

is_failed

Check if download failed.

iCloudHandler ¶

iCloudHandler(timeout=ICLOUD_DOWNLOAD_TIMEOUT, poll_interval=ICLOUD_POLL_INTERVAL)

Handle iCloud video downloads.

This class provides functionality to detect, trigger, and monitor downloads of videos stored in iCloud. It uses the macOS brctl command-line tool to interact with the Bird (iCloud daemon).

SDS Reference: SDS-P01-007 SRS Reference: SRS-304 (iCloud Download Handling)

Example

handler = iCloudHandler(timeout=1800) # 30 min timeout for video in candidates: ... status = handler.get_status(video) ... if status == CloudStatus.CLOUD_ONLY: ... print(f"Downloading {video.filename} from iCloud...") ... success = handler.download_and_wait( ... video, ... on_progress=lambda p: print(f" {p.progress:.1f}%") ... ) ... if not success: ... print(f" Failed to download, skipping") ... continue ... # Proceed with conversion ... convert(video)

Attributes:

Name	Type	Description
`timeout`		Maximum time to wait for download in seconds.
`poll_interval`		Time between status checks in seconds.

Initialize iCloud handler.

Parameters:

Name	Type	Description	Default
`timeout`	`int`	Maximum time to wait for downloads in seconds.	`ICLOUD_DOWNLOAD_TIMEOUT`
`poll_interval`	`float`	Time between status checks in seconds.	`ICLOUD_POLL_INTERVAL`

Source code in src/video_converter/extractors/icloud_handler.py

def __init__(
    self,
    timeout: int = ICLOUD_DOWNLOAD_TIMEOUT,
    poll_interval: float = ICLOUD_POLL_INTERVAL,
) -> None:
    """Initialize iCloud handler.

    Args:
        timeout: Maximum time to wait for downloads in seconds.
        poll_interval: Time between status checks in seconds.
    """
    self.timeout = timeout
    self.poll_interval = poll_interval

get_status ¶

get_status(video)

Check iCloud status of a video.

Determines whether the video file is available locally, stored only in iCloud, or currently downloading.

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video information from Photos library.	required

Returns:

Type	Description
`CloudStatus`	CloudStatus indicating the current state of the file.

Source code in src/video_converter/extractors/icloud_handler.py

def get_status(self, video: PhotosVideoInfo) -> CloudStatus:
    """Check iCloud status of a video.

    Determines whether the video file is available locally,
    stored only in iCloud, or currently downloading.

    Args:
        video: Video information from Photos library.

    Returns:
        CloudStatus indicating the current state of the file.
    """
    # If not marked as in cloud, assume local
    if not video.in_cloud:
        return CloudStatus.LOCAL

    # Check for iCloud stub file first
    if video.path is not None:
        stub_path = self._get_stub_path(video.path)
        if stub_path.exists():
            # Stub exists - check if download is in progress
            if self._is_downloading(video.path):
                return CloudStatus.DOWNLOADING
            return CloudStatus.CLOUD_ONLY

    # If path is available and file exists (no stub), it's local
    if video.path is not None and video.path.exists():
        # Double-check it's not a stub file
        if not self._is_stub_file(video.path):
            return CloudStatus.LOCAL

    # If in_cloud is True but no path, it's cloud only
    if video.in_cloud and video.path is None:
        return CloudStatus.CLOUD_ONLY

    return CloudStatus.UNKNOWN

trigger_download ¶

trigger_download(video)

Trigger iCloud download for a video.

Uses the macOS brctl command to request file download from iCloud. This is available on macOS 12 (Monterey) and later.

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video information with path to download.	required

Returns:

Type	Description
`bool`	True if download was triggered successfully, False otherwise.

Source code in src/video_converter/extractors/icloud_handler.py

def trigger_download(self, video: PhotosVideoInfo) -> bool:
    """Trigger iCloud download for a video.

    Uses the macOS brctl command to request file download from iCloud.
    This is available on macOS 12 (Monterey) and later.

    Args:
        video: Video information with path to download.

    Returns:
        True if download was triggered successfully, False otherwise.
    """
    if video.path is None:
        logger.warning(f"Cannot download video without path: {video.filename}")
        return False

    # Get the target path (use stub path if it exists)
    target_path = video.path
    stub_path = self._get_stub_path(video.path)
    if stub_path.exists():
        target_path = stub_path

    logger.info(f"Triggering iCloud download for: {video.filename}")
    logger.debug(f"Target path: {target_path}")

    try:
        # Use brctl to trigger download
        result = subprocess.run(
            ["brctl", "download", str(target_path)],
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode == 0:
            logger.debug(f"Download triggered successfully: {video.filename}")
            return True

        # Log error details
        logger.warning(
            f"brctl download failed for {video.filename}: exit code {result.returncode}"
        )
        if result.stderr:
            logger.debug(f"brctl stderr: {result.stderr}")

        return False

    except subprocess.TimeoutExpired:
        logger.warning(f"brctl command timed out for: {video.filename}")
        return False
    except FileNotFoundError:
        logger.error("brctl command not found. iCloud download requires macOS 12 or later.")
        return False
    except subprocess.SubprocessError as e:
        logger.error(f"Failed to run brctl: {e}")
        return False

wait_for_download ¶

wait_for_download(video, on_progress=None)

Wait for iCloud download to complete.

Polls the file status until the download completes, fails, or the timeout is reached.

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video to wait for.	required
`on_progress`	`Callable[[DownloadProgress], None] \| None`	Optional callback for progress updates.	`None`

Returns:

Type	Description
`bool`	True if download completed successfully, False otherwise.

Source code in src/video_converter/extractors/icloud_handler.py

def wait_for_download(
    self,
    video: PhotosVideoInfo,
    on_progress: Callable[[DownloadProgress], None] | None = None,
) -> bool:
    """Wait for iCloud download to complete.

    Polls the file status until the download completes, fails,
    or the timeout is reached.

    Args:
        video: Video to wait for.
        on_progress: Optional callback for progress updates.

    Returns:
        True if download completed successfully, False otherwise.
    """
    if video.path is None:
        logger.warning(f"Cannot wait for video without path: {video.filename}")
        return False

    start_time = time.time()
    last_status = CloudStatus.UNKNOWN

    logger.debug(f"Waiting for download: {video.filename} (timeout={self.timeout}s)")

    while True:
        elapsed = time.time() - start_time

        # Check timeout
        if elapsed >= self.timeout:
            logger.warning(f"Download timed out: {video.filename}")
            if on_progress:
                on_progress(
                    DownloadProgress(
                        filename=video.filename,
                        status=CloudStatus.FAILED,
                        elapsed_seconds=elapsed,
                    )
                )
            return False

        # Check status
        status = self.get_status(video)

        # Report progress if status changed
        if status != last_status:
            logger.debug(f"Status changed: {last_status.value} -> {status.value}")
            last_status = status

        if status == CloudStatus.LOCAL:
            logger.info(f"Download complete: {video.filename}")
            if on_progress:
                on_progress(
                    DownloadProgress(
                        filename=video.filename,
                        status=CloudStatus.LOCAL,
                        progress=100.0,
                        elapsed_seconds=elapsed,
                    )
                )
            return True

        if status == CloudStatus.FAILED:
            logger.warning(f"Download failed: {video.filename}")
            if on_progress:
                on_progress(
                    DownloadProgress(
                        filename=video.filename,
                        status=CloudStatus.FAILED,
                        elapsed_seconds=elapsed,
                    )
                )
            return False

        # Report intermediate progress
        if on_progress:
            progress = self._get_download_progress(video.path)
            on_progress(
                DownloadProgress(
                    filename=video.filename,
                    status=status,
                    progress=progress,
                    elapsed_seconds=elapsed,
                )
            )

        # Wait before next check
        time.sleep(self.poll_interval)

download_and_wait ¶

download_and_wait(video, on_progress=None)

Trigger download and wait for completion.

Combined convenience method that triggers a download if needed and waits for it to complete.

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video to download and wait for.	required
`on_progress`	`Callable[[DownloadProgress], None] \| None`	Optional callback for progress updates.	`None`

Returns:

Type	Description
`bool`	True if the video is now available locally, False otherwise.

Source code in src/video_converter/extractors/icloud_handler.py

def download_and_wait(
    self,
    video: PhotosVideoInfo,
    on_progress: Callable[[DownloadProgress], None] | None = None,
) -> bool:
    """Trigger download and wait for completion.

    Combined convenience method that triggers a download if needed
    and waits for it to complete.

    Args:
        video: Video to download and wait for.
        on_progress: Optional callback for progress updates.

    Returns:
        True if the video is now available locally, False otherwise.
    """
    status = self.get_status(video)

    # Already local, nothing to do
    if status == CloudStatus.LOCAL:
        logger.debug(f"Video already local: {video.filename}")
        return True

    # Failed status, cannot download
    if status == CloudStatus.FAILED:
        logger.warning(f"Video marked as failed: {video.filename}")
        return False

    # Need to trigger download for cloud-only files
    if status == CloudStatus.CLOUD_ONLY:
        if not self.trigger_download(video):
            logger.warning(f"Failed to trigger download: {video.filename}")
            return False

    # Wait for download to complete
    return self.wait_for_download(video, on_progress)

evict ¶

evict(video)

Evict a video from local storage (keep in iCloud only).

Uses the macOS brctl command to remove the local copy while keeping the file in iCloud. This frees up local disk space.

Parameters:

Name	Type	Description	Default
`video`	`PhotosVideoInfo`	Video to evict from local storage.	required

Returns:

Type	Description
`bool`	True if eviction was successful, False otherwise.

Source code in src/video_converter/extractors/icloud_handler.py

def evict(self, video: PhotosVideoInfo) -> bool:
    """Evict a video from local storage (keep in iCloud only).

    Uses the macOS brctl command to remove the local copy while
    keeping the file in iCloud. This frees up local disk space.

    Args:
        video: Video to evict from local storage.

    Returns:
        True if eviction was successful, False otherwise.
    """
    if video.path is None:
        logger.warning(f"Cannot evict video without path: {video.filename}")
        return False

    if not video.path.exists():
        logger.debug(f"Video not present locally: {video.filename}")
        return True

    logger.info(f"Evicting video from local storage: {video.filename}")

    try:
        result = subprocess.run(
            ["brctl", "evict", str(video.path)],
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode == 0:
            logger.debug(f"Eviction successful: {video.filename}")
            return True

        logger.warning(
            f"brctl evict failed for {video.filename}: exit code {result.returncode}"
        )
        return False

    except subprocess.TimeoutExpired:
        logger.warning(f"brctl evict timed out: {video.filename}")
        return False
    except FileNotFoundError:
        logger.error("brctl command not found")
        return False
    except subprocess.SubprocessError as e:
        logger.error(f"Failed to run brctl evict: {e}")
        return False

Extractors Module¶

Photos Extractor¶

photos_extractor ¶

PhotosLibraryError ¶

PhotosAccessDeniedError ¶

PhotosLibraryNotFoundError ¶

VideoNotAvailableError ¶

ExportError ¶

MediaType ¶

PhotosVideoInfo dataclass ¶

is_available_locally property ¶

is_h264 property ¶

is_hevc property ¶

needs_conversion property ¶

PhotosLibrary ¶

library_path property ¶

db property ¶

check_permissions ¶

get_library_info ¶

get_video_count ¶

get_videos ¶

get_video_by_uuid ¶

close ¶

__enter__ ¶

__exit__ ¶

LibraryStats dataclass ¶

estimated_savings property ¶

estimated_savings_gb property ¶

PhotosVideoFilter ¶

codec_detector property ¶

get_conversion_candidates ¶

get_stats ¶

VideoExporter ¶

temp_dir property ¶

export ¶

cleanup ¶

cleanup_all ¶

get_exported_count ¶

get_temp_dir_size ¶

__enter__ ¶

__exit__ ¶

get_permission_instructions ¶

Folder Extractor¶

folder_extractor ¶

FolderExtractorError ¶

FolderNotFoundError ¶

FolderAccessDeniedError ¶

InvalidVideoFileError ¶

FolderVideoInfo dataclass ¶

is_h264 property ¶

is_hevc property ¶

needs_conversion property ¶

resolution_label property ¶

size_mb property ¶

size_gb property ¶

__str__ ¶

FolderStats dataclass ¶

estimated_savings property ¶

estimated_savings_gb property ¶

total_size_gb property ¶

FolderExtractor ¶

root_path property ¶

recursive property ¶

codec_detector property ¶

scan ¶

get_video_info ¶

get_videos ¶

get_conversion_candidates ¶

get_stats ¶

get_video_count ¶

__enter__ ¶

__exit__ ¶

__repr__ ¶

iCloud Handler¶

icloud_handler ¶

iCloudError ¶

iCloudDownloadError ¶

iCloudTimeoutError ¶

CloudStatus ¶

DownloadProgress dataclass ¶

PhotosVideoInfo `dataclass` ¶

is_available_locally `property` ¶

is_h264 `property` ¶

is_hevc `property` ¶

needs_conversion `property` ¶

library_path `property` ¶

db `property` ¶

enter ¶

exit ¶

LibraryStats `dataclass` ¶

estimated_savings `property` ¶

estimated_savings_gb `property` ¶

codec_detector `property` ¶

temp_dir `property` ¶

enter ¶

exit ¶

FolderVideoInfo `dataclass` ¶

is_h264 `property` ¶

is_hevc `property` ¶

needs_conversion `property` ¶

resolution_label `property` ¶

size_mb `property` ¶

size_gb `property` ¶

str ¶

FolderStats `dataclass` ¶

estimated_savings `property` ¶

estimated_savings_gb `property` ¶

total_size_gb `property` ¶

root_path `property` ¶

recursive `property` ¶

codec_detector `property` ¶

enter ¶

exit ¶

repr ¶

DownloadProgress `dataclass` ¶

is_complete `property` ¶

is_failed `property` ¶