8e4e96dcbb
Video thumbnails are now generated too (recursively, along with images) on the `thumb` command. The poster frame is extracted at 25% of the duration (instead of a fixed 10s seek). Thumbnails are now kept in cache indefinitely.
123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia: an XMPP client
|
|
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
"""Methods to manipulate videos"""
|
|
|
|
from typing import Union
|
|
from pathlib import Path
|
|
from twisted.python.procutils import which
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core import exceptions
|
|
from libervia.backend.core.log import getLogger
|
|
from .common import async_process
|
|
|
|
log = getLogger(__name__)
|
|
|
|
try:
|
|
ffmpeg_path = which("ffmpeg")[0]
|
|
except IndexError:
|
|
log.warning(_("ffmpeg executable not found, video thumbnails won't be available"))
|
|
ffmpeg_path = None
|
|
|
|
try:
|
|
ffprobe_path = which("ffprobe")[0]
|
|
except IndexError:
|
|
log.warning(
|
|
_("ffprobe executable not found, video thumbnails will use the first frame")
|
|
)
|
|
ffprobe_path = None
|
|
|
|
|
|
async def _get_duration(video_path: Path | str) -> float | None:
|
|
"""Get the duration of a video in seconds
|
|
|
|
@return: duration in seconds, or None if ffprobe is unavailable or fails
|
|
"""
|
|
if ffprobe_path is None:
|
|
return None
|
|
try:
|
|
result = await async_process.run(
|
|
ffprobe_path,
|
|
"-v",
|
|
"error",
|
|
"-show_entries",
|
|
"format=duration",
|
|
"-of",
|
|
"default=noprint_wrappers=1:nokey=1",
|
|
str(video_path),
|
|
)
|
|
except exceptions.CommandException:
|
|
return None
|
|
try:
|
|
return float(result)
|
|
except ValueError:
|
|
log.error(f"Can't parse ffprobe duration for {video_path}: {result!r}")
|
|
return None
|
|
|
|
|
|
async def get_thumbnail(video_path: Union[Path, str], dest_path: Path) -> Path:
|
|
"""Extract thumbnail from video
|
|
|
|
The frame is extracted at 25% of the video duration, falling back to the first frame
|
|
if the duration is unknown or the extraction fails.
|
|
|
|
@param video_path: source of the video
|
|
@param dest_path: path where the file must be saved
|
|
@return: path of the generated thumbnail
|
|
image is created in temporary directory but is not delete automatically
|
|
it should be deleted after use.
|
|
Image will be in JPEG format.
|
|
@raise exceptions.NotFound: ffmpeg is missing
|
|
@raise exceptions.DataError: can't extract a frame from the video
|
|
"""
|
|
if ffmpeg_path is None:
|
|
raise exceptions.NotFound(
|
|
_("ffmpeg executable is not available, can't generate video thumbnail")
|
|
)
|
|
ffmpeg = ffmpeg_path
|
|
|
|
async def extract(seek_time: str) -> None:
|
|
await async_process.run(
|
|
ffmpeg,
|
|
"-ss",
|
|
seek_time,
|
|
"-i",
|
|
str(video_path),
|
|
"-frames:v",
|
|
"1",
|
|
"-y",
|
|
str(dest_path),
|
|
)
|
|
|
|
duration = await _get_duration(video_path)
|
|
data_error = exceptions.DataError(
|
|
_("Can't extract a frame from the video {video_path}").format(
|
|
video_path=video_path
|
|
)
|
|
)
|
|
seek_times = [f"{duration * 0.25:.2f}", "0"] if duration else ["0"]
|
|
last_error = None
|
|
for seek_time in seek_times:
|
|
try:
|
|
await extract(seek_time)
|
|
break
|
|
except exceptions.CommandException as e:
|
|
last_error = e
|
|
if not dest_path.exists() or dest_path.stat().st_size == 0:
|
|
raise data_error from last_error
|
|
return dest_path
|