80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
"""
|
|
Media-family extractor: mp4.
|
|
|
|
doocus does not transcribe here. It probes container metadata via ffprobe, grabs
|
|
a thumbnail via ffmpeg, and records a pointer delegating full transcription to
|
|
the meetus pipeline (process_meeting.py). Meeting videos are thus discoverable
|
|
and previewable in doocus while their transcript lives in meetus.
|
|
"""
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from .base import ExtractionResult
|
|
|
|
EXTRACTOR_ID = "media/ffprobe@1"
|
|
|
|
|
|
def _ffprobe(source: Path) -> dict:
|
|
if not shutil.which("ffprobe"):
|
|
return {}
|
|
cmd = ["ffprobe", "-v", "quiet", "-print_format", "json",
|
|
"-show_format", "-show_streams", str(source)]
|
|
try:
|
|
out = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout
|
|
return json.loads(out)
|
|
except (subprocess.CalledProcessError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def _thumbnail(source: Path, at_seconds: float = 1.0) -> bytes:
|
|
if not shutil.which("ffmpeg"):
|
|
return b""
|
|
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
|
|
tmp_path = Path(tmp.name)
|
|
try:
|
|
cmd = ["ffmpeg", "-v", "quiet", "-y", "-ss", str(at_seconds),
|
|
"-i", str(source), "-frames:v", "1", "-vf", "scale=640:-1",
|
|
str(tmp_path)]
|
|
subprocess.run(cmd, check=True)
|
|
return tmp_path.read_bytes() if tmp_path.exists() else b""
|
|
except subprocess.CalledProcessError:
|
|
return b""
|
|
finally:
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
|
|
def extract(source: Path, options: dict) -> ExtractionResult:
|
|
warnings = []
|
|
probe = _ffprobe(source)
|
|
metadata = {"delegate": "meetus", "delegate_hint": "run process_meeting.py for transcript"}
|
|
|
|
fmt = probe.get("format", {})
|
|
if fmt:
|
|
if "duration" in fmt:
|
|
metadata["duration_seconds"] = float(fmt["duration"])
|
|
tags = fmt.get("tags", {})
|
|
if "creation_time" in tags:
|
|
metadata["created"] = tags["creation_time"]
|
|
video_streams = [s for s in probe.get("streams", []) if s.get("codec_type") == "video"]
|
|
if video_streams:
|
|
v = video_streams[0]
|
|
metadata["width"], metadata["height"] = v.get("width"), v.get("height")
|
|
if not probe:
|
|
warnings.append("ffprobe unavailable or failed; metadata limited")
|
|
|
|
thumb = _thumbnail(source) or None
|
|
if thumb is None:
|
|
warnings.append("thumbnail not generated (ffmpeg unavailable or failed)")
|
|
|
|
dur = metadata.get("duration_seconds")
|
|
content = (f"_Video {source.name}"
|
|
+ (f", {dur:.0f}s" if dur else "")
|
|
+ ". Transcript is produced by the meetus pipeline "
|
|
"(`process_meeting.py`), not doocus._\n")
|
|
|
|
return ExtractionResult(content=content, metadata=metadata, thumb=thumb,
|
|
warnings=warnings, extractor=EXTRACTOR_ID)
|