diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml index 5ce36e3..e14ae85 100644 --- a/.github/workflows/python-ci.yml +++ b/.github/workflows/python-ci.yml @@ -24,6 +24,10 @@ jobs: uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} + - name: Install ffmpeg + run: | + sudo apt-get update + sudo apt-get install -y ffmpeg - name: Install dependencies run: | pip install --no-cache-dir --upgrade pip setuptools wheel diff --git a/Dockerfile b/Dockerfile index ad6335c..74fc795 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ FROM python:3.13-alpine -RUN apk add make +RUN apk add make ffmpeg RUN pip install --no-cache-dir --upgrade pip setuptools wheel diff --git a/examples/compose_multistream.py b/examples/compose_multistream.py new file mode 100644 index 0000000..5c99136 --- /dev/null +++ b/examples/compose_multistream.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +""" +Combine several video files into a single side-by-side video using ffmpeg. + +The input videos are laid out side by side, from left to right and top to bottom. +Up to 6 inputs are supported. + +This module is used by ``mass_import.py`` (through :func:`compose_streams`) to combine the +streams of a multi-stream media on the fly, and can also be used as a standalone script to +combine a given list of source files into a destination file:: + + ./examples/compose_multistream.py --inputs a.mp4 a_2.mp4 --output combined/a.mp4 +""" +from __future__ import annotations + +import argparse +import json +import logging +from pathlib import Path +import subprocess +import sys + +try: + from nudgisclient.lib.utils import configure_logging +except ModuleNotFoundError: # pragma: no cover + sys.path.append(str(Path(__file__).resolve().parent.parent)) + from nudgisclient.lib.utils import configure_logging + +logger = logging.getLogger(__name__) + +# Maximum number of streams that can be combined into a multi-stream media. +MAX_INPUTS = 6 + + +def _grid_columns(count: int) -> int: + """Return the number of columns of the side-by-side grid for ``count`` streams.""" + if count <= 4: + return 2 + return 3 + + +def _xstack_layout(count: int, columns: int) -> str: + """ + Build the ``xstack`` layout string placing ``count`` inputs on a grid. + + The inputs are laid out from left to right and top to bottom (specifications ยง3.4). + """ + parts = [] + for index in range(count): + column = index % columns + row = index // columns + if column == 0: + x = '0' + else: + x = '+'.join(f'w{row * columns + col}' for col in range(column)) + if row == 0: + y = '0' + else: + y = '+'.join(f'h{prev_row * columns}' for prev_row in range(row)) + parts.append(f'{x}_{y}') + return '|'.join(parts) + + +def _build_ffmpeg_command( + inputs: list[Path], + output: Path, + audio_indices: list[int], + ffmpeg: str = 'ffmpeg', +) -> list[str]: + """ + Build the ffmpeg command combining ``inputs`` into a single ``output`` video. + + ``audio_indices`` lists the inputs that carry an audio stream: their audio is mixed into + a single track, or mapped as-is when there is only one. The output has no audio when the + list is empty. + """ + count = len(inputs) + columns = _grid_columns(count) + layout = _xstack_layout(count, columns) + video_inputs = ''.join(f'[{index}:v]' for index in range(count)) + filter_complex = f'{video_inputs}xstack=inputs={count}:layout={layout}[v]' + maps = ['-map', '[v]'] + if len(audio_indices) >= 2: + audio_inputs = ''.join(f'[{index}:a]' for index in audio_indices) + filter_complex += f';{audio_inputs}amix=inputs={len(audio_indices)}[a]' + maps += ['-map', '[a]'] + elif len(audio_indices) == 1: + maps += ['-map', f'{audio_indices[0]}:a'] + command = [ffmpeg, '-y'] + for input_path in inputs: + command += ['-i', str(input_path)] + command += ['-filter_complex', filter_complex, *maps, str(output)] + return command + + +def _probe_video(path: Path, ffprobe: str = 'ffprobe') -> tuple[int, int, bool]: + """ + Probe a media file and return its ``(width, height, has_audio)``. + + ``width`` and ``height`` are those of the first video stream (``0`` if there is none). + """ + try: + result = subprocess.run( + [ + ffprobe, '-v', 'error', '-show_entries', 'stream=codec_type,width,height', + '-of', 'json', str(path), + ], + capture_output=True, + text=True, + ) + except FileNotFoundError as err: + raise RuntimeError(f'"{ffprobe}" command not found.') from err + streams = json.loads(result.stdout or '{}').get('streams', []) + width = height = 0 + has_audio = False + for stream in streams: + if stream['codec_type'] == 'video' and not width: + width, height = stream['width'], stream['height'] + if stream['codec_type'] == 'audio': + has_audio = True + return width, height, has_audio + + +def _build_composition_layout( + sizes: list[tuple[int, int]], + area_width: int, + area_height: int, +) -> dict: + """ + Describe how the inputs are placed in the combined video (one layer per input). + + ``sizes`` is the ``(width, height)`` of each input, in display order. The positions + reproduce the side-by-side grid built by ffmpeg's ``xstack`` (see :func:`_xstack_layout`). + """ + columns = _grid_columns(len(sizes)) + layers = [] + for index, (width, height) in enumerate(sizes): + column = index % columns + row = index // columns + x = sum(sizes[row * columns + col][0] for col in range(column)) + y = sum(sizes[prev_row * columns][1] for prev_row in range(row)) + layer_id = index + 1 + layers.append({ + 'id': layer_id, + 'label': f'element-{layer_id}', + 'enabled': True, + 'source': { + 'type': 'video', + 'roi': {'x': x, 'y': y, 'w': width, 'h': height}, + 'native_resolution': {'w': area_width, 'h': area_height}, + }, + 'x': x, + 'y': y, + 'w': width, + 'h': height, + 'z': layer_id, + }) + return { + 'composition_area': {'w': area_width, 'h': area_height}, + 'composition_data': [{'time': 0, 'layers': layers}], + } + + +def compose_streams( + inputs: list[Path], + output: Path, + ffmpeg: str = 'ffmpeg', + ffprobe: str = 'ffprobe', +) -> None: + """ + Combine the given ordered video inputs into a single side-by-side ``output`` video. + + The main stream must come first, followed by the secondary streams in display order. + Inputs without an audio stream are supported: only the inputs that have audio are mixed + into the resulting track. The parent directory of ``output`` is created if needed. + + A companion JSON file describing the placement of each input (the "layout preset") is + written next to ``output``, with the same name and a ``.json`` extension. + + Raise ``ValueError`` if the number of inputs is out of range, or ``RuntimeError`` if + ffmpeg/ffprobe is missing or ffmpeg fails. + """ + if not 2 <= len(inputs) <= MAX_INPUTS: + raise ValueError( + f'Expected between 2 and {MAX_INPUTS} inputs to combine, got {len(inputs)}.' + ) + output.parent.mkdir(parents=True, exist_ok=True) + probes = [_probe_video(path, ffprobe=ffprobe) for path in inputs] + audio_indices = [index for index, (_w, _h, has_audio) in enumerate(probes) if has_audio] + command = _build_ffmpeg_command(inputs, output, audio_indices, ffmpeg=ffmpeg) + logger.info('Combining %s inputs into "%s".', len(inputs), output) + try: + result = subprocess.run(command, capture_output=True, text=True) + except FileNotFoundError as err: + raise RuntimeError(f'"{ffmpeg}" command not found.') from err + if result.returncode != 0: + raise RuntimeError(f'ffmpeg failed: {result.stderr.strip()}') + logger.info('Combined video written to "%s".', output) + + # Write the layout preset describing the placement of each input next to the output. + sizes = [(width, height) for width, height, _has_audio in probes] + area_width, area_height, _ = _probe_video(output, ffprobe=ffprobe) + layout = _build_composition_layout(sizes, area_width, area_height) + layout_path = output.with_suffix('.json') + layout_path.write_text(json.dumps(layout, indent=2)) + logger.info('Composition layout written to "%s".', layout_path) + + +def compose_multistream(sys_args: list[str]) -> int: + parser = argparse.ArgumentParser( + 'compose_multistream', + description=__doc__.strip(), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + '--inputs', + help='Ordered list of source video files (main stream first).', + required=True, + nargs='+', + type=Path, + ) + parser.add_argument( + '--output', + help='Path of the combined video file to produce.', + required=True, + type=Path, + ) + parser.add_argument( + '--ffmpeg', + help='Path to the ffmpeg executable.', + default='ffmpeg', + ) + parser.add_argument( + '--ffprobe', + help='Path to the ffprobe executable (used to detect audio streams).', + default='ffprobe', + ) + parser.add_argument( + '--log-level', + help='Log level.', + default='info', + choices=['critical', 'error', 'warn', 'info', 'debug'], + ) + args = parser.parse_args(sys_args) + + configure_logging(args.log_level.upper()) + + for input_path in args.inputs: + if not input_path.is_file(): + logger.error('Source video "%s" does not exist.', input_path) + return 1 + + try: + compose_streams(args.inputs, args.output, ffmpeg=args.ffmpeg, ffprobe=args.ffprobe) + except (ValueError, RuntimeError) as err: + logger.error('%s', err) + return 1 + return 0 + + +if __name__ == '__main__': # pragma: no cover + sys.exit(compose_multistream(sys.argv[1:])) diff --git a/examples/mass_import.py b/examples/mass_import.py new file mode 100644 index 0000000..87fd3e5 --- /dev/null +++ b/examples/mass_import.py @@ -0,0 +1,1793 @@ +#!/usr/bin/env python3 +""" +Mass media import script for Nudgis migrations. + +This script implements the "Nudgis Standard Migration" +(see https://docs.google.com/document/d/1ZzrQJ_50vnyeotfXd-0dgHO5I68LhAuUlVsvjCsNKy0/edit). +It scans a source directory laid out according to the migration standard and either: + +- audits it (default behaviour, no ``--apply``): it checks that the file tree is + correctly structured, that the media files are valid (using ``ffprobe``), that the + ``metadata.csv``, ``annotations.csv`` and ``channels.csv`` files are well formed and + consistent with the files on disk, and it runs a few server-side checks (slug uniqueness, + annotation types, speaker resolution, explicit channels); +- imports it (with ``--apply``): it uploads each media (reusing the folder tree as a + channel tree), attaches subtitles and additional audio tracks, applies the metadata and + the annotations, applies the channel metadata and writes a ``source_id`` -> ``oid`` + mapping file. + +Problems found during the audit never abort the whole run: recoverable issues are cleaned +up in place (invalid optional fields are dropped, over-long values are truncated) and the +individual media, metadata entries or annotations that cannot be processed are skipped so +that the rest is still imported. Both modes end with a report summarising the content that +was correctly processed and the content that could not be. Only unrecoverable problems (an +unreachable server, a structurally broken CSV, a missing ``ffprobe`` or an empty source +tree) still abort the run. + +Multi-stream (side-by-side) media are combined into a single video with ``ffmpeg`` during +the import (see ``compose_multistream.py``); the temporary combined file is written under +``--temp-dir`` and removed once the upload succeeds. + +A media can be sent to the personal channel of its first speaker by setting its ``channel`` +to ``mscspeaker``, or to a sub-channel of that personal channel with ``mscspeaker-`` (for +example ``mscspeaker-Courses/2026``). The sub-channels that do not exist yet are created during +the import. + +The channel metadata listed in ``channels.csv`` are applied at the very end of the import, +once every media (and therefore every channel of the folder tree) has been created. The +``path`` of a channel is resolved from the root of the Nudgis catalog, so it must include the +title of the main migration channel; the channels that do not exist yet are created. + +Example: + + ./examples/mass_import.py --conf myconf.json --source-dir ./migration --channel "Migration 2026" + ./examples/mass_import.py --conf myconf.json --source-dir ./migration --channel "Migration 2026" --apply +""" +from __future__ import annotations + +import argparse +import csv +from dataclasses import dataclass, field +from datetime import date, datetime +from html import escape +import json +import logging +from pathlib import Path +import re +import subprocess +import sys + +try: + from examples.compose_multistream import compose_streams + from nudgisclient import NudgisClient, NudgisRequestError + from nudgisclient.lib.utils import configure_logging +except ModuleNotFoundError: # pragma: no cover + sys.path.append(str(Path(__file__).resolve().parent.parent)) + from examples.compose_multistream import compose_streams + from nudgisclient import NudgisClient, NudgisRequestError + from nudgisclient.lib.utils import configure_logging + + +logger = logging.getLogger(__name__) + +# Recognized file extensions (lower case, with leading dot). +VIDEO_EXTENSIONS = { + '.mp4', '.mov', '.mkv', '.avi', '.webm', '.m4v', '.mpg', '.mpeg', '.ts', '.wmv', '.flv', +} +AUDIO_EXTENSIONS = {'.mp3', '.m4a', '.aac', '.wav', '.flac', '.ogg', '.opus'} +SUBTITLE_EXTENSIONS = {'.srt', '.vtt'} + +# Name of the metadata file (expected at the root of the source directory). +METADATA_FILENAME = 'metadata.csv' +# Name of the channels metadata file (expected at the root of the source directory). +CHANNELS_FILENAME = 'channels.csv' +# Name of the annotations directory and file (expected at the root of the source directory). +ANNOTATIONS_DIRNAME = 'annotations' +ANNOTATIONS_FILENAME = 'annotations.csv' + +# Columns of metadata.csv. The mandatory ones are flagged with "*" in the specifications. +METADATA_MANDATORY_FIELDS = {'source_id', 'title'} +METADATA_KNOWN_FIELDS = { + 'source_id', 'title', 'slug', 'description', 'keywords', 'categories', 'language', + 'creation', 'speaker_name', 'speaker_email', 'company_name', 'company_url', + 'license_name', 'license_url', 'channel', 'validated', 'unlisted', 'detect_slides', +} +# Columns of channels.csv. Only the channel path is mandatory. +CHANNELS_MANDATORY_FIELDS = {'path'} +CHANNELS_KNOWN_FIELDS = {'path', 'description', 'reference'} +# Columns of annotations.csv. +ANNOTATIONS_MANDATORY_FIELDS = {'source_id', 'type'} +ANNOTATIONS_KNOWN_FIELDS = { + 'source_id', 'type', 'time', 'title', 'content', 'keywords', 'attachment', +} +# Annotation type slugs available by default on Nudgis. +INTERNAL_ANNOTATION_TYPES = {'chapter', 'slide', 'attachment', 'activity'} + +# Allowed values for the yes/no metadata fields. +YESNO_VALUES = {'yes', 'no'} + +# Validation patterns. +SLUG_RE = re.compile(r'^[a-z0-9_-]+$') +# Language code pattern (ISO 639-2). +LANGUAGE_RE = re.compile(r'^[a-z]{3}$') +# A "_stream" suffix designates a secondary stream of a multi-stream media. +STREAM_SUFFIX_RE = re.compile(r'^(.+)_stream(\d+)$') +# A "_<3 letters>" suffix designates a linked subtitle or audio track. +LANG_SUFFIX_RE = re.compile(r'^(.+)_([a-z]{3})$') +# Single field max length. +MAX_FIELD_LENGTH = 200 +# Maximum number of streams that can be combined into a multi-stream media. +MAX_STREAMS = 6 +# Channel target designating the personal channel of the first speaker of a media. It can be +# suffixed with the path of one of its sub-channels ("mscspeaker-Top channel/Sub channel"). +SPEAKER_TARGET = 'mscspeaker' + + +@dataclass +class MediaGroup: + """ + All the data that makes up a single media, grouped by their common source id. + + Besides the files found on disk, a group also carries its validated ``metadata.csv`` + row (``metadata``) and its ``annotations.csv`` rows (``annotations``) so that everything + related to a media is accessible from a single object. + """ + + source_id: str + object_id: str | None + main_file: Path + rel_dir: Path + subtitles: dict[str, Path] = field(default_factory=dict) + audio_tracks: dict[str, Path] = field(default_factory=dict) + extra_streams: dict[int, Path] = field(default_factory=dict) + metadata: dict[str, str] | None = None + annotations: list[dict[str, str]] = field(default_factory=list) + existing_elements: list[str] = field(default_factory=list) + + @property + def external_ref(self) -> str: + return f'migration:{self.source_id}' + + +@dataclass +class ChannelUpdate: + """ + The metadata of a single channel, as described by a ``channels.csv`` row. + + ``path`` holds the titles of all the channels making up the path of the channel, from the + root of the Nudgis catalog down to it. + """ + + path: list[str] + description: str = '' + reference: str = '' + object_id: str | None = None + + @property + def display_path(self) -> str: + return '/'.join(self.path) + + +@dataclass +class Report: + """ + Fatal errors and warnings collected during the audit phase. + + Only *unrecoverable* problems (an unreachable server, a structurally broken CSV, an + empty source tree) are recorded as ``errors`` and abort the whole run. Everything that + can be cleaned up or skipped without blocking the other media is either fixed in place + (with a ``warning``) or accounted for in the :class:`Summary`. + """ + + errors: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + + def error(self, message: str) -> None: + self.errors.append(message) + + def warning(self, message: str) -> None: + self.warnings.append(message) + + @property + def ok(self) -> bool: + return not self.errors + + +# Reasons why a media cannot be imported at all (used as keys of ``Summary.unimportable_media``). +UNIMPORTABLE_CORRUPTED = 'corrupted files (ffprobe)' +UNIMPORTABLE_FILENAME = 'invalid file name' +UNIMPORTABLE_DUPLICATE = 'duplicates' +UNIMPORTABLE_NO_SPEAKER = 'personal channel target without recipient' + + +@dataclass +class Summary: + """Counters aggregated across the audit and import phases for the final report.""" + + # Correctly processed content. + merged_videos: int = 0 + medias_imported: int = 0 + medias_existing: int = 0 + metadata_applied: int = 0 + # Linked elements, broken down into already existing / imported / failed. + audio_tracks_existing: int = 0 + audio_tracks_imported: int = 0 + audio_tracks_failed: int = 0 + subtitles_existing: int = 0 + subtitles_imported: int = 0 + subtitles_failed: int = 0 + annotations_existing: int = 0 + annotations_imported: int = 0 + annotations_failed: int = 0 + # Channels whose metadata have been applied / could not be applied. + channels_updated: int = 0 + channels_failed: int = 0 + # Unprocessable content. + unimportable_media: dict[str, list[str]] = field(default_factory=dict) + invalid_metadata: int = 0 + invalid_channels: int = 0 + unimportable_annotations: int = 0 + import_failures: int = 0 + + def drop_media(self, source_id: str, reason: str) -> None: + """Record a media that cannot be imported under the given ``reason``.""" + self.unimportable_media.setdefault(reason, []).append(source_id) + + +def run_audit( + source_dir: Path, + ngc: NudgisClient, + ids_to_process: list[str] | None = None, + ffprobe: str = 'ffprobe', + apply: bool = False, +) -> tuple[Report, Summary, dict[str, MediaGroup], list[ChannelUpdate]]: + """ + Run the whole audit phase and return the report, the summary, the media groups and the + channel metadata. + + Instead of aborting on the first problem, the audit cleans up recoverable issues in + place and drops the individual media, metadata entries or annotations that cannot be + processed (accounting for them in ``summary``) so that the other content can still be + imported. Only unrecoverable problems end up in ``report.errors``. + + When ``apply`` is true the server-side checks are allowed to fix the catalog (create + missing speaker users and grant them the personal-channel permission); otherwise they + only report what would be done. + """ + report = Report() + summary = Summary() + groups = scan_source_tree(source_dir, ids_to_process, report, summary) + validate_metadata_csv(source_dir / METADATA_FILENAME, groups, ids_to_process, report, summary) + validate_annotations_csv(source_dir, groups, ids_to_process, report, summary) + channels = validate_channels_csv(source_dir / CHANNELS_FILENAME, report, summary) + validate_media_integrity(groups, report, summary, ffprobe=ffprobe) + server_checks(ngc, groups, report, summary, channels=channels, apply=apply) + return report, summary, groups, channels + + +def scan_source_tree( + source_dir: Path, + ids_to_process: list[str] | None, + report: Report, + summary: Summary, +) -> dict[str, MediaGroup]: + """ + Walk ``source_dir`` and group the files by media (source id). + + The ``annotations`` directory and the ``metadata.csv`` and ``channels.csv`` files are + ignored here; they are validated separately. Media that cannot be imported (a video with + a too-long name or a duplicate source id) are dropped and counted in ``summary``; + auxiliary files that cannot be attached are simply ignored with a warning. + """ + logger.info('Scanning source tree "%s".', source_dir) + annotations_dir = source_dir / ANNOTATIONS_DIRNAME + + media_files: list[Path] = [] + other_files: list[Path] = [] + for path in sorted(source_dir.rglob('*')): + if not path.is_file(): + continue + if path in (source_dir / METADATA_FILENAME, source_dir / CHANNELS_FILENAME): + continue + if annotations_dir in path.parents or path == annotations_dir: + continue + if ids_to_process is not None and not any(path.stem.startswith(source_id) for source_id in ids_to_process): + continue + suffix = path.suffix.lower() + if len(path.name) > MAX_FIELD_LENGTH: + # A video with a too-long name is a media we cannot import; anything else is + # an auxiliary file we can simply ignore. + if suffix in VIDEO_EXTENSIONS: + report.warning(f'Media file name too long, media skipped: "{path.name}".') + summary.drop_media(path.stem, UNIMPORTABLE_FILENAME) + else: + report.warning(f'Ignoring file with a too-long name: "{path.name}".') + continue + if suffix in VIDEO_EXTENSIONS: + media_files.append(path) + else: + other_files.append(path) + + # First pass: determine the main media files and detect multi-stream secondary files. + video_stems = {path.stem for path in media_files} + groups: dict[str, MediaGroup] = {} + secondary_files: list[Path] = [] + for path in media_files: + match = STREAM_SUFFIX_RE.match(path.stem) + if match and match.group(1) in video_stems: + secondary_files.append(path) + continue + source_id = path.stem + if source_id in groups: + report.warning( + f'Duplicate source id "{source_id}", media skipped: "{path}" (kept ' + f'"{groups[source_id].main_file}").' + ) + summary.drop_media(source_id, UNIMPORTABLE_DUPLICATE) + continue + if ids_to_process is not None and source_id not in ids_to_process: + continue + groups[source_id] = MediaGroup( + source_id=source_id, + object_id=None, + main_file=path, + rel_dir=path.parent.relative_to(source_dir), + ) + + # Second pass: attach the multi-stream secondary files to their main media. + for path in secondary_files: + source_id, index_str = STREAM_SUFFIX_RE.match(path.stem).groups() + if ids_to_process is not None and source_id not in ids_to_process: + continue + index = int(index_str) + group = groups.get(source_id) + if group is None: + report.warning(f'Ignoring multi-stream file "{path}" with no main media "{source_id}".') + continue + if index < 2 or index > MAX_STREAMS: + report.warning( + f'Ignoring multi-stream file "{path}" with an out-of-range index ' + f'(must be between 2 and {MAX_STREAMS}).' + ) + continue + group.extra_streams[index] = path + + # Third pass: attach the linked subtitle and audio files to their main media. + for path in other_files: + suffix = path.suffix.lower() + if suffix not in SUBTITLE_EXTENSIONS and suffix not in AUDIO_EXTENSIONS: + report.warning(f'Ignoring unexpected file "{path}".') + continue + match = LANG_SUFFIX_RE.match(path.stem) + if not match: + report.warning( + f'Ignoring file "{path}" because it does not follow the "_" naming rule.' + ) + continue + source_id, lang = match.groups() + if ids_to_process is not None and source_id not in ids_to_process: + continue + group = groups.get(source_id) + if group is None: + report.warning(f'Ignoring file "{path}" with no main media "{source_id}".') + continue + target = group.subtitles if suffix in SUBTITLE_EXTENSIONS else group.audio_tracks + if lang in target: + file_type = 'subtitle' if suffix in SUBTITLE_EXTENSIONS else 'audio track' + report.warning( + f'Ignoring duplicate {file_type} file for "{source_id}" and language "{lang}": "{path}".' + ) + continue + target[lang] = path + + # Report multi-stream media (handled by the dedicated compose_multistream.py script). + for group in groups.values(): + if group.extra_streams: + indexes = sorted(group.extra_streams) + if indexes != list(range(2, 2 + len(indexes))): + report.warning( + f'Multi-stream media "{group.source_id}" has non-contiguous stream ' + f'indexes: {indexes}.' + ) + report.warning( + f'Multi-stream media "{group.source_id}" ({1 + len(indexes)} streams) will ' + 'be combined with ffmpeg during import.' + ) + + if not groups: + report.error(f'No media file to import found in "{source_dir}".') + return groups + + +def _split_pipe(value: str, drop_empty: bool = True) -> list[str]: + """Split a pipe-separated metadata value, optionally dropping empty items.""" + return [item.strip() for item in value.split('|') if not drop_empty or item.strip()] + + +def validate_metadata_csv( + csv_path: Path, + groups: dict[str, MediaGroup], + ids_to_process: list[str] | None, + report: Report, + summary: Summary, +) -> None: + """ + Validate ``metadata.csv``, clean up its rows and attach each usable row to its media. + + The file is optional; if it is missing, nothing is attached. A row that cannot be tied + to a media (empty, duplicate or unknown ``source_id``) is dropped and counted as an + invalid metadata entry. Individual invalid fields are cleaned up in place (dropped or + truncated) so that the media can still be imported with the rest of its metadata. + """ + logger.info('Validating metadata CSV "%s".', csv_path) + if not csv_path.is_file(): + report.warning(f'No "{METADATA_FILENAME}" file found at "{csv_path}".') + return + + with csv_path.open('r', encoding='utf-8', newline='') as csvfile: + reader = csv.DictReader(csvfile, delimiter=',', quotechar='"') + + header = reader.fieldnames or [] + missing = METADATA_MANDATORY_FIELDS - set(header) + if missing: + report.error( + f'"{METADATA_FILENAME}" is missing mandatory columns: ' + f'{", ".join(sorted(missing))}.' + ) + return + + for unknown in sorted(set(header) - METADATA_KNOWN_FIELDS): + report.warning(f'"{METADATA_FILENAME}" has an unknown column "{unknown}".') + + slugs: dict[str, str] = {} + seen_ids: set[str] = set() + for line, row in enumerate(reader, start=2): + source_id = (row.get('source_id') or '').strip() + if not source_id: + report.warning(f'{METADATA_FILENAME}:{line}: empty "source_id", row ignored.') + summary.invalid_metadata += 1 + continue + if ids_to_process is not None and source_id not in ids_to_process: + continue + if source_id in seen_ids: + report.warning( + f'{METADATA_FILENAME}:{line}: duplicate "source_id" "{source_id}", row ignored.' + ) + summary.invalid_metadata += 1 + continue + seen_ids.add(source_id) + + group = groups.get(source_id) + if group is None: + report.warning( + f'{METADATA_FILENAME}:{line}: "source_id" "{source_id}" has no matching ' + 'media file, row ignored.' + ) + summary.invalid_metadata += 1 + continue + + if not (row.get('title') or '').strip(): + # The title is mandatory; fall back to the source id at import time. + report.warning( + f'{METADATA_FILENAME}:{line}: empty "title", the source id will be used.' + ) + row['title'] = '' + + slug = (row.get('slug') or '').strip().lower() + if slug and not SLUG_RE.match(slug): + report.warning( + f'{METADATA_FILENAME}:{line}: invalid "slug" "{slug}" ignored (allowed ' + 'characters: a-z, 0-9, "-", "_").' + ) + slug = '' + elif slug and slug in slugs: + report.warning( + f'{METADATA_FILENAME}:{line}: "slug" "{slug}" already used by ' + f'"{slugs[slug]}", ignored.' + ) + slug = '' + elif slug: + slugs[slug] = source_id + row['slug'] = slug + + language = (row.get('language') or '').strip() + if language and not LANGUAGE_RE.match(language): + report.warning( + f'{METADATA_FILENAME}:{line}: invalid "language" "{language}" ignored ' + '(expected a 3-letter ISO 639-2 code).' + ) + row['language'] = '' + + creation = (row.get('creation') or '').strip() + if creation: + try: + datetime.strptime(creation, '%Y-%m-%dT%H:%M:%S') + except ValueError: + report.warning( + f'{METADATA_FILENAME}:{line}: invalid "creation" "{creation}" ignored ' + '(expected format YYYY-MM-DDTHH:MM:SS).' + ) + row['creation'] = '' + + for field_name in ( + 'title', 'slug', 'keywords', 'company_name', 'company_url', 'license_name', 'license_url', + ): + value = (row.get(field_name) or '').strip() + if len(value) > MAX_FIELD_LENGTH: + report.warning( + f'{METADATA_FILENAME}:{line}: "{field_name}" value too long, truncated.' + ) + row[field_name] = value[:MAX_FIELD_LENGTH] + + channel = (row.get('channel') or '').strip() + truncated = truncate_channel_titles(channel) + if truncated != channel: + report.warning( + f'{METADATA_FILENAME}:{line}: a channel title of "{channel}" is longer ' + f'than {MAX_FIELD_LENGTH} characters, truncated.' + ) + row['channel'] = truncated + + names = _split_pipe((row.get('speaker_name') or ''), drop_empty=False) + emails = _split_pipe((row.get('speaker_email') or ''), drop_empty=False) + if len(names) != len(emails): + report.warning( + f'{METADATA_FILENAME}:{line}: "speaker_name" ({len(names)}) and ' + f'"speaker_email" ({len(emails)}) have a different number of values; ' + 'speakers ignored.' + ) + row['speaker_name'] = '' + row['speaker_email'] = '' + + for field_name in ('validated', 'unlisted', 'detect_slides'): + value = (row.get(field_name) or '').strip() + if value and value not in YESNO_VALUES: + report.warning( + f'{METADATA_FILENAME}:{line}: invalid "{field_name}" "{value}" ignored ' + '(expected "yes" or "no").' + ) + row[field_name] = '' + + group.metadata = row + + for source_id, group in groups.items(): + if group.metadata is None: + report.warning( + f'Media "{source_id}" has no row in "{METADATA_FILENAME}".' + ) + + +def validate_channels_csv( + csv_path: Path, + report: Report, + summary: Summary, +) -> list[ChannelUpdate]: + """ + Validate ``channels.csv`` and return the channel metadata to apply after the import. + + The file is optional; if it is missing, no channel metadata is applied. A row that cannot + be applied at all (empty, malformed or duplicate ``path``, no metadata to apply) is + dropped and counted as an invalid channel entry; an over-long channel title is truncated + and an unusable ``reference`` (too long or already used) is dropped on its own so that the + description can still be applied. + """ + logger.info('Validating channels CSV "%s".', csv_path) + if not csv_path.is_file(): + report.warning(f'No "{CHANNELS_FILENAME}" file found at "{csv_path}".') + return [] + + channels: list[ChannelUpdate] = [] + with csv_path.open('r', encoding='utf-8', newline='') as csvfile: + reader = csv.DictReader(csvfile, delimiter=',', quotechar='"') + + header = reader.fieldnames or [] + missing = CHANNELS_MANDATORY_FIELDS - set(header) + if missing: + report.error( + f'"{CHANNELS_FILENAME}" is missing mandatory columns: ' + f'{", ".join(sorted(missing))}.' + ) + return [] + + for unknown in sorted(set(header) - CHANNELS_KNOWN_FIELDS): + report.warning(f'"{CHANNELS_FILENAME}" has an unknown column "{unknown}".') + + seen_paths: set[str] = set() + references: dict[str, str] = {} + for line, row in enumerate(reader, start=2): + path = (row.get('path') or '').strip().strip('/') + titles = [title.strip() for title in path.split('/')] if path else [] + if not titles or not all(titles): + report.warning( + f'{CHANNELS_FILENAME}:{line}: empty or invalid "path", row ignored.' + ) + summary.invalid_channels += 1 + continue + if any(len(title) > MAX_FIELD_LENGTH for title in titles): + # The titles are truncated the same way as the channel targets of + # "metadata.csv" so that the metadata are applied to the channel the media + # have been imported into. + report.warning( + f'{CHANNELS_FILENAME}:{line}: a channel title of "{path}" is longer than ' + f'{MAX_FIELD_LENGTH} characters, truncated.' + ) + titles = [title[:MAX_FIELD_LENGTH] for title in titles] + display_path = '/'.join(titles) + if display_path in seen_paths: + report.warning( + f'{CHANNELS_FILENAME}:{line}: duplicate "path" "{display_path}", row ignored.' + ) + summary.invalid_channels += 1 + continue + seen_paths.add(display_path) + + reference = (row.get('reference') or '').strip() + if reference and len(reference) > MAX_FIELD_LENGTH: + # A reference identifies an external element (for example a course of an LMS), + # so a truncated one would point to the wrong element: it is dropped instead. + report.warning( + f'{CHANNELS_FILENAME}:{line}: "reference" value too long, ignored (at most ' + f'{MAX_FIELD_LENGTH} characters).' + ) + reference = '' + elif reference and reference in references: + report.warning( + f'{CHANNELS_FILENAME}:{line}: "reference" "{reference}" already used by ' + f'"{references[reference]}", ignored.' + ) + reference = '' + elif reference: + references[reference] = display_path + + description = (row.get('description') or '').strip() + if not description and not reference: + report.warning( + f'{CHANNELS_FILENAME}:{line}: no metadata to apply to "{display_path}", ' + 'row ignored.' + ) + summary.invalid_channels += 1 + continue + + channels.append( + ChannelUpdate(path=titles, description=description, reference=reference) + ) + return channels + + +def validate_annotations_csv( + source_dir: Path, + groups: dict[str, MediaGroup], + ids_to_process: list[str] | None, + report: Report, + summary: Summary, +) -> None: + """ + Validate ``annotations/annotations.csv``, clean up its rows and attach the usable ones. + + The annotations directory is optional; if it is missing, nothing is attached. An + annotation that cannot be imported (no matching media, empty type, ``chapter`` without a + title, ``slide`` without an attachment, missing attachment file) is dropped and counted; + recoverable issues (too-long fields, invalid time, an attachment on a ``chapter``) are + cleaned up in place. + """ + annotations_dir = source_dir / ANNOTATIONS_DIRNAME + csv_path = annotations_dir / ANNOTATIONS_FILENAME + logger.info('Validating annotations CSV "%s".', csv_path) + if not annotations_dir.is_dir(): + return + if not csv_path.is_file(): + report.error(f'No "{ANNOTATIONS_FILENAME}" file found in "{annotations_dir}".') + return + + with csv_path.open('r', encoding='utf-8', newline='') as csvfile: + reader = csv.DictReader(csvfile, delimiter=',', quotechar='"') + + header = reader.fieldnames or [] + missing = ANNOTATIONS_MANDATORY_FIELDS - set(header) + if missing: + report.error( + f'"{ANNOTATIONS_FILENAME}" is missing mandatory columns: ' + f'{", ".join(sorted(missing))}.' + ) + return + + for unknown in sorted(set(header) - ANNOTATIONS_KNOWN_FIELDS): + report.warning(f'"{ANNOTATIONS_FILENAME}" has an unknown column "{unknown}".') + + for line, row in enumerate(reader, start=2): + source_id = (row.get('source_id') or '').strip() + if not source_id: + report.warning(f'{ANNOTATIONS_FILENAME}:{line}: empty "source_id", annotation ignored.') + summary.unimportable_annotations += 1 + continue + if ids_to_process is not None and source_id not in ids_to_process: + continue + + group = groups.get(source_id) + if group is None: + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: "source_id" "{source_id}" has no matching ' + 'media file, annotation ignored.' + ) + summary.unimportable_annotations += 1 + continue + + for field_name in ('type', 'time', 'title', 'keywords', 'attachment'): + value = (row.get(field_name) or '').strip() + # The title and keywords are escaped in Nudgis, which increases their length. + escaped = field_name in ('title', 'keywords') + if len(escape(value) if escaped else value) > MAX_FIELD_LENGTH: + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: "{field_name}" value too long, truncated.' + ) + while len(escape(value) if escaped else value) > MAX_FIELD_LENGTH: + value = value[:-1] + row[field_name] = value + + ann_type = (row.get('type') or '').strip().lower() + row['type'] = ann_type + if not ann_type: + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: empty "type", annotation ignored.' + ) + summary.unimportable_annotations += 1 + continue + if ann_type == 'chapter' and not (row.get('title') or '').strip(): + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: a "chapter" annotation requires a ' + '"title", annotation ignored.' + ) + summary.unimportable_annotations += 1 + continue + + time_value = (row.get('time') or '').strip() or '0' + if not time_value.isdigit(): + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: invalid "time" "{time_value}" reset to 0 ' + '(expected an integer number of milliseconds).' + ) + time_value = '0' + row['time'] = time_value + + attachment = (row.get('attachment') or '').strip().strip('/') + if attachment: + attachment = attachment.removeprefix(ANNOTATIONS_DIRNAME + '/') + if ann_type == 'chapter' and attachment: + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: attachment cannot be linked to a "chapter" ' + 'annotation, attachment ignored.' + ) + attachment = '' + elif ann_type == 'slide' and not attachment: + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: attachment is required for "slide" ' + 'annotations, annotation ignored.' + ) + summary.unimportable_annotations += 1 + continue + elif attachment and not (annotations_dir / attachment).is_file(): + report.warning( + f'{ANNOTATIONS_FILENAME}:{line}: attachment "{attachment}" does not exist in ' + 'the annotations directory, annotation ignored.' + ) + summary.unimportable_annotations += 1 + continue + row['attachment'] = attachment + + group.annotations.append(row) + + +def validate_media_integrity( + groups: dict[str, MediaGroup], + report: Report, + summary: Summary, + ffprobe: str = 'ffprobe', +) -> None: + """ + Run ``ffprobe`` on every video and audio file of every media group. + + A media with at least one corrupted file is dropped and counted; a missing ``ffprobe`` + executable is a fatal error since no media could be checked. + """ + corrupted: list[str] = [] + for source_id, group in groups.items(): + files = [group.main_file, *group.extra_streams.values(), *group.audio_tracks.values()] + for path in files: + error = probe_media(path, ffprobe=ffprobe) + if error: + if 'command not found' in error: + report.error(error) + return + report.warning(f'Invalid media file "{path}", media skipped: {error}') + corrupted.append(source_id) + break + for source_id in corrupted: + del groups[source_id] + summary.drop_media(source_id, UNIMPORTABLE_CORRUPTED) + + +def probe_media(path: Path, ffprobe: str = 'ffprobe') -> str | None: + """ + Check the integrity of a media file using ``ffprobe``. + + Return ``None`` if the file is a valid media, or an error message otherwise. + """ + try: + result = subprocess.run( + [ + ffprobe, '-v', 'error', '-show_entries', 'stream=codec_type', + '-of', 'json', str(path), + ], + capture_output=True, + text=True, + ) + except FileNotFoundError: + return f'"{ffprobe}" command not found; cannot check media integrity.' + if result.returncode != 0: + return f'ffprobe failed: {result.stderr.strip()}' + try: + streams = json.loads(result.stdout).get('streams', []) + except json.JSONDecodeError: + return 'ffprobe returned an invalid output.' + if not streams: + return 'no media stream found.' + return None + + +def server_checks( + ngc: NudgisClient, + groups: dict[str, MediaGroup], + report: Report, + summary: Summary, + channels: list[ChannelUpdate] | None = None, + apply: bool = False, +) -> None: + """ + Run the server-side audit checks (slugs, annotation types, speakers, channels). + + Annotations referencing an unknown type are dropped, media targeting a personal channel + without a ``speaker_email`` are dropped, and media with an unresolvable explicit channel + fall back to the folder-based channel. The ``channels.csv`` entries are resolved against + the catalog to report the channels that do not exist yet and the references already used. + When ``apply`` is true, missing speaker users are created and granted the personal-channel + permission; otherwise the audit only reports what would be done. + """ + try: + ngc.check_server() + except Exception as err: + report.error(f'Cannot reach the Nudgis server: {err}') + return + + # Check permissions of the API key. + try: + response = ngc.api('users/me/') + except NudgisRequestError as err: + report.error(f'Unexpected response from the server when testing the API: {err}') + return + else: + if not response.get('user', {}).get('permissions', {}).get('can_change_users'): + report.error('The user account of the given API key does not have the required permissions.') + return + + # Slug and external ref uniqueness against the existing catalog, and resolution of the + # channels listed in "channels.csv". + if groups or channels: + try: + catalog = ngc.get_catalog(fmt='flat') + except NudgisRequestError as err: + report.warning(f'Could not fetch the catalog to check slugs and channels: {err}') + else: + existing_slugs = { + obj['slug']: obj['oid'] + for key in ('channels', 'videos', 'lives', 'photos') + for obj in catalog.get(key, []) + if obj.get('slug') + } + existing_ext_refs = { + obj['external_ref']: obj['oid'] + for obj in catalog.get('videos', []) + if obj.get('external_ref') + } + for group in groups.values(): + if group.external_ref and group.external_ref in existing_ext_refs: + group.object_id = existing_ext_refs[group.external_ref] + logger.info( + 'Found existing video for external_ref "%s": %s', + group.external_ref, group.object_id + ) + else: + slug = (group.metadata.get('slug') or '').strip() if group.metadata else '' + if slug and slug in existing_slugs: + report.warning( + f'Slug "{slug}" (media "{group.source_id}") already exists on the server' + f' (used by "{existing_slugs[slug]}").' + ) + channel_paths = build_channel_paths(catalog) + channel_refs = { + obj['external_ref']: obj['oid'] + for obj in catalog.get('channels', []) + if obj.get('external_ref') + } + for update in channels or []: + update.object_id = channel_paths.get(update.display_path) + if update.object_id is None: + report.warning( + f'Channel "{update.display_path}" does not exist yet; it would be ' + 'created on import.' + ) + owner = channel_refs.get(update.reference) + if update.reference and owner and owner != update.object_id: + report.warning( + f'Reference "{update.reference}" (channel "{update.display_path}") is ' + f'already used by another channel ("{owner}").' + ) + + # Annotation types existence: drop the annotations referencing an unknown type. + used_types = { + ann_type + for group in groups.values() + for row in group.annotations + if (ann_type := (row.get('type') or '').strip()) + and ann_type not in INTERNAL_ANNOTATION_TYPES + } + if used_types: + try: + response = ngc.api('annotations/types/list/') + except NudgisRequestError as err: + report.warning(f'Could not fetch annotation types: {err}') + else: + server_types = { + value['slug'] + for value in response.get('types', []) + if value.get('slug') + } + missing_types = used_types - server_types + for ann_type in sorted(missing_types): + report.warning( + f'Annotation type "{ann_type}" does not exist on the server; related ' + 'annotations were dropped.' + ) + if missing_types: + for group in groups.values(): + kept = [ + row for row in group.annotations + if (row.get('type') or '').strip() not in missing_types + ] + summary.unimportable_annotations += len(group.annotations) - len(kept) + group.annotations = kept + + # Speaker resolution for the personal channel destinations: a media without any speaker + # cannot be imported; missing users and permissions are reconciled (or reported) below. + speaker_emails: set[str] = set() + no_speaker: list[str] = [] + for source_id, group in groups.items(): + row = group.metadata + channel = (row.get('channel') or '').strip() if row else '' + if parse_speaker_path(channel) is not None: + emails = _split_pipe(row.get('speaker_email') or '') + if not emails: + report.warning( + f'Media "{source_id}" targets "{channel}" but has no "speaker_email", ' + 'media skipped.' + ) + no_speaker.append(source_id) + else: + speaker_emails.update(emails) + for source_id in no_speaker: + del groups[source_id] + summary.drop_media(source_id, UNIMPORTABLE_NO_SPEAKER) + + for email in sorted(speaker_emails): + try: + response = ngc.api('users/', params={'search': email, 'limit': 1}) + except NudgisRequestError as err: + report.warning(f'Could not check speaker "{email}": {err}') + continue + users = response.get('users') or [] + user_id = users[0].get('id') if users else None + if user_id is None: + if not apply: + report.warning( + f'Speaker "{email}" does not exist yet; it would be created on import.' + ) + continue + try: + created = ngc.api( + 'users/add/', method='post', data={'username': email, 'email': email} + ) + except NudgisRequestError as err: + report.warning(f'Could not create speaker "{email}": {err}') + continue + user_id = created.get('id') + logger.info('Created Nudgis user "%s" (id %s).', email, user_id) + if user_id is None: + continue + ensure_personal_channel(ngc, user_id, email, report, apply) + + # Existence of explicitly referenced channels (by oid or slug). + for group in groups.values(): + row = group.metadata + channel = (row.get('channel') or '').strip() if row else '' + if not channel: + continue + params = None + if parse_speaker_path(channel) is not None: + # The personal channels and their sub-channels are resolved during the import. + continue + elif channel.startswith('mscid-'): + params = {'oid': channel[len('mscid-'):]} + elif not channel.startswith('mscpath-'): + params = {'slug': channel} + if params is None: + continue + try: + ngc.api('channels/get/', params=params) + except NudgisRequestError as err: + report.warning( + f'Channel "{channel}" for media "{group.source_id}" could not be resolved ' + f'({err}); falling back to the folder-based channel.' + ) + group.metadata['channel'] = '' + + # Existence of audio tracks (to skip already imported audio tracks). + for group in groups.values(): + if not group.object_id: + continue + try: + response = ngc.api('medias/audio/tracks/list/', params={'oid': group.object_id}) + except NudgisRequestError as err: + if err.status_code == 404: + continue + report.warning( + f'Could not list audio tracks of media "{group.source_id}" ({err}); already ' + 'imported audio tracks may be re-added.' + ) + else: + for track in response.get('audio_tracks', []): + if track.get('is_original'): + continue + group.existing_elements.append(f'audio:{track["language"]}') + + # Existence of subtitles (to skip already imported subtitles). + for group in groups.values(): + if not group.object_id: + continue + try: + response = ngc.api('subtitles/', params={'oid': group.object_id}) + except NudgisRequestError as err: + if err.status_code == 404: + continue + report.warning( + f'Could not list subtitles of media "{group.source_id}" ({err}); already ' + 'imported subtitles may be re-added.' + ) + else: + for subtitle in response.get('subtitles', []): + if subtitle.get('auto_transcripted') or subtitle.get('auto_translated'): + continue + group.existing_elements.append(f'subtitle:{subtitle["lang_code"]}') + + # Existence of annotations (to skip already imported annotations). + for group in groups.values(): + if not group.object_id: + continue + try: + response = ngc.api('annotations/list/', params={'oid': group.object_id}) + except NudgisRequestError as err: + if err.status_code == 404: + continue + report.warning( + f'Could not list annotations of media "{group.source_id}" ({err}); already ' + 'imported annotations may be re-added.' + ) + else: + type_by_id = {t['id']: t['slug'] for t in response.get('types', {}).values()} + for annotation in response.get('annotations', []): + type_slug = type_by_id[annotation['type_id']] + group.existing_elements.append(f'annotation:{type_slug}:{annotation["time"]}') + + +def ensure_personal_channel( + ngc: NudgisClient, + user_id: str, + email: str, + report: Report, + apply: bool, +) -> None: + """ + Make sure a speaker user is allowed to own a personal channel. + + When ``apply`` is true the permission is granted if needed; otherwise the audit only + reports that it would be granted. + """ + try: + response = ngc.api('perms/get/', params={'type': 'user', 'id': user_id}) + except NudgisRequestError as err: + report.warning(f'Could not check permissions of speaker "{email}": {err}') + return + perm = (response.get('global_permissions') or {}).get('can_have_personal_channel') or {} + if perm.get('val') or perm.get('inherit_val'): + return + if not apply: + report.warning( + f'Speaker "{email}" cannot own a personal channel; the permission would be ' + 'granted on import.' + ) + return + try: + ngc.api( + 'perms/edit/', + method='post', + data={'type': 'user', 'id': user_id, 'can_have_personal_channel': 'True'}, + ) + logger.info('Granted the personal-channel permission to speaker "%s".', email) + except NudgisRequestError as err: + report.warning(f'Could not grant the personal-channel permission to "{email}": {err}') + + +def import_media( + ngc: NudgisClient, + groups: dict[str, MediaGroup], + main_channel: str, + source_dir: Path, + mapping_file: Path, + temp_dir: Path, + summary: Summary, + ffmpeg: str = 'ffmpeg', + ffprobe: str = 'ffprobe', +) -> dict[str, str]: + """ + Import every media group, fill in ``summary`` and return the ``source_id`` -> ``oid`` map. + + A media is counted as imported (and recorded in the mapping) as soon as its ``add_media`` + step succeeds; a media that cannot be created is counted as an import failure and skipped + (a media targeting the sub-channel of a personal channel that cannot be resolved included). + Each linked element (audio track, subtitle, annotation) is then attached independently and + accounted for as already existing, imported or failed, so a single element failure does + not prevent the others from being attached. + """ + failures: dict[str, str] = {} + mapping: dict[str, str] = {} + # Oids of the personal channels and of their sub-channels, resolved on demand. + speaker_channels: dict[str, str] = {} + for source_id, group in groups.items(): + merged = False + # Media creation: a failure here means the media itself could not be imported. + try: + channel = resolve_channel(main_channel, group.rel_dir, group.metadata) + if titles := parse_speaker_path(channel): + # A sub-channel of a personal channel can only be targeted by its object id. + channel = ensure_speaker_channel( + ngc, group.metadata, titles, speaker_channels + ) + metadata = build_media_metadata(group.metadata) + metadata['external_ref'] = group.external_ref + existing = bool(group.object_id) + if existing: + oid = group.object_id + logger.info('Media "%s" already exists with oid "%s".', source_id, oid) + else: + file_path = group.main_file + if group.extra_streams: + # Combine the multi-stream files into a single side-by-side video. + merged = True + streams = [group.main_file] + [ + group.extra_streams[index] for index in sorted(group.extra_streams) + ] + file_path = temp_dir / group.main_file.name + logger.info( + 'Composing %s streams of multi-stream media "%s".', + len(streams), source_id, + ) + compose_streams(streams, file_path, ffmpeg=ffmpeg, ffprobe=ffprobe) + # The composition layout preset is written next to the composed video. + metadata['layout_preset'] = file_path.with_suffix('.json').read_text() + logger.info('Uploading media "%s" into channel "%s".', source_id, channel) + response = ngc.add_media( + title=metadata.pop('title', source_id), + file_path=file_path, + channel=channel, + origin=metadata['external_ref'], + skip_automatic_subtitles='yes', + skip_automatic_enrichments='yes', + **metadata, + ) + oid = response['oid'] + group.object_id = oid + logger.info('Media "%s" created with oid "%s".', source_id, oid) + if metadata.get('slug') and response['slug'] != metadata['slug']: + logger.warning( + 'Media "%s" "%s" did not receive the requested slug "%s", it got "%s".', + source_id, oid, metadata['slug'], response['slug'] + ) + if file_path != group.main_file: + # The composed video and its layout preset have been uploaded; clean up. + file_path.unlink(missing_ok=True) + file_path.with_suffix('.json').unlink(missing_ok=True) + logger.debug('Removed temporary composed files for "%s".', source_id) + except (NudgisRequestError, RuntimeError) as err: + logger.error('Failed to import media "%s": %s', source_id, err) + failures[source_id] = str(err) + summary.import_failures += 1 + continue + + # The media now exists on the server: record it as imported (or already existing). + mapping[source_id] = oid + if existing: + summary.medias_existing += 1 + else: + summary.medias_imported += 1 + if merged: + summary.merged_videos += 1 + if group.metadata is not None: + summary.metadata_applied += 1 + + for lang, path in group.audio_tracks.items(): + if f'audio:{lang}' in group.existing_elements: + logger.info('The "%s" audio track already exists in "%s".', lang, oid) + summary.audio_tracks_existing += 1 + continue + try: + logger.info('Adding "%s" audio track to "%s".', lang, oid) + with path.open('rb') as fileobj: + # Known limitation: the audio track must have a duration close to the video duration + ngc.api( + 'medias/audio/tracks/add/', + method='post', + data={'oid': oid, 'lang': lang}, + files={'file': (path.name, fileobj)}, + ) + summary.audio_tracks_imported += 1 + except (NudgisRequestError, RuntimeError) as err: + logger.error('Failed to add "%s" audio track to "%s": %s', lang, oid, err) + summary.audio_tracks_failed += 1 + + for lang, path in group.subtitles.items(): + if f'subtitle:{lang}' in group.existing_elements: + logger.info('The "%s" subtitles already exists in "%s".', lang, oid) + summary.subtitles_existing += 1 + continue + try: + logger.info('Adding "%s" subtitles to "%s".', lang, oid) + with path.open('rb') as fileobj: + ngc.api( + 'subtitles/add/', + method='post', + data={'oid': oid, 'lang': lang}, + files={'file': (path.name, fileobj)}, + ) + summary.subtitles_imported += 1 + except (NudgisRequestError, RuntimeError) as err: + logger.error('Failed to add "%s" subtitles to "%s": %s', lang, oid, err) + summary.subtitles_failed += 1 + + for row in group.annotations: + if 'time' not in row or 'type' not in row: + continue + if f'annotation:{row["type"]}:{row["time"]}' in group.existing_elements: + logger.info( + 'The "%s" annotation at time "%s" already exists in "%s".', + row['type'], row['time'], oid, + ) + summary.annotations_existing += 1 + continue + try: + post_annotation(ngc, oid, row, source_dir) + summary.annotations_imported += 1 + except (NudgisRequestError, RuntimeError) as err: + logger.error( + 'Failed to add "%s" annotation at time "%s" to "%s": %s', + row['type'], row['time'], oid, err, + ) + summary.annotations_failed += 1 + + mapping_file.write_text( + 'source_id,oid\n' + ''.join(f'{src},{oid}\n' for src, oid in mapping.items()), + ) + logger.info('Wrote mapping of %s media to "%s".', len(mapping), mapping_file) + + if failures: + logger.warning( + 'List of media that could not be created:\n - %s', + '\n - '.join(f'{src}: {msg}' for src, msg in failures.items()), + ) + logger.info('Media import complete: %s created, %s failed.', len(mapping), len(failures)) + return mapping + + +def build_channel_paths(catalog: dict) -> dict[str, str]: + """ + Build a "channel path" -> ``oid`` map from a flat catalog. + + The path of a channel is made of the titles of all the channels from the root of the + catalog down to it, separated by "/" (the format used in ``channels.csv``). + """ + by_oid = {obj['oid']: obj for obj in catalog.get('channels', []) if obj.get('oid')} + paths: dict[str, str] = {} + for oid, obj in by_oid.items(): + titles: list[str] = [] + seen: set[str] = set() + current = obj + while current is not None and current['oid'] not in seen: + seen.add(current['oid']) + titles.append(current.get('title') or '') + current = by_oid.get(current.get('parent_oid')) + paths.setdefault('/'.join(reversed(titles)), oid) + return paths + + +def ensure_channel(ngc: NudgisClient, titles: list[str], paths: dict[str, str]) -> str: + """ + Return the oid of the channel at the given path, creating the missing levels. + + ``paths`` is the "channel path" -> ``oid`` map of the existing channels (see + :func:`build_channel_paths`); it is updated with the channels created along the way. + """ + parent_oid = '' + for index, title in enumerate(titles): + path = '/'.join(titles[:index + 1]) + oid = paths.get(path) + if oid is None: + data = {'title': title} + if parent_oid: + data['parent'] = parent_oid + oid = ngc.api('channels/add/', method='post', data=data)['oid'] + paths[path] = oid + logger.info('Created channel "%s" with oid "%s".', path, oid) + parent_oid = oid + return parent_oid + + +def get_or_create_channel(ngc: NudgisClient, title: str, parent_oid: str) -> str: + """ + Return the oid of the sub-channel of ``parent_oid`` titled ``title``, creating it if needed. + + Unlike :func:`ensure_channel`, the lookup is made on the server because the personal + channels are not part of the catalog. + """ + try: + response = ngc.api('channels/get/', params={'title': title, 'parent': parent_oid}) + except NudgisRequestError as err: + if err.status_code != 404: + raise + else: + oid = (response.get('info') or {}).get('oid') + if not oid: + raise RuntimeError(f'unexpected response when getting the channel "{title}"') + return oid + oid = ngc.api( + 'channels/add/', method='post', data={'title': title, 'parent': parent_oid} + )['oid'] + logger.info('Created channel "%s" under "%s" with oid "%s".', title, parent_oid, oid) + return oid + + +def ensure_speaker_channel( + ngc: NudgisClient, + row: dict[str, str] | None, + titles: list[str], + cache: dict[str, str], +) -> str: + """ + Return the "mscid-..." target of a sub-channel of the personal channel of a speaker. + + The path is resolved (and its missing levels created) below the personal channel of the + first speaker of the media because the API can only target a personal sub-channel by its + object id. ``cache`` holds the oids already resolved, keyed by the email of the speaker + followed by the titles of the channels, so that each channel is resolved only once. + """ + emails = _split_pipe((row or {}).get('speaker_email') or '') + if not emails: + raise RuntimeError('no "speaker_email" to resolve the personal channel') + email = emails[0] + oid = cache.get(email) + if oid is None: + response = ngc.api('users/', params={'search': email, 'limit': 1}) + users = response.get('users') or [] + if not users: + raise RuntimeError(f'the speaker "{email}" does not exist on the server') + oid = ngc.api('channels/personal/', params={'id': users[0]['id']})['oid'] + cache[email] = oid + logger.info('The personal channel of "%s" is "%s".', email, oid) + for index, title in enumerate(titles): + key = '/'.join([email, *titles[:index + 1]]) + sub_oid = cache.get(key) + if sub_oid is None: + sub_oid = get_or_create_channel(ngc, title, oid) + cache[key] = sub_oid + oid = sub_oid + return f'mscid-{oid}' + + +def apply_channels_metadata( + ngc: NudgisClient, + channels: list[ChannelUpdate], + summary: Summary, +) -> None: + """ + Apply the ``channels.csv`` metadata, creating the channels that do not exist yet. + + This is the last step of the import so that the channels created while uploading the media + (from the source folder tree) can be targeted by their path. The values of the CSV file + always take precedence over the values already set on the server. A channel that cannot be + resolved or updated is counted as a failure and does not prevent the next ones from being + updated. + """ + if not channels: + return + try: + catalog = ngc.get_catalog(fmt='flat') + except NudgisRequestError as err: + logger.error('Could not fetch the catalog to resolve the channel paths: %s', err) + summary.channels_failed += len(channels) + return + paths = build_channel_paths(catalog) + for update in channels: + try: + oid = ensure_channel(ngc, update.path, paths) + data = {'oid': oid} + if update.description: + data['description'] = update.description + if update.reference: + data['external_ref'] = update.reference + logger.info('Applying metadata to channel "%s" ("%s").', update.display_path, oid) + ngc.api('channels/edit/', method='post', data=data) + update.object_id = oid + summary.channels_updated += 1 + except (NudgisRequestError, RuntimeError) as err: + logger.error( + 'Failed to apply metadata to channel "%s": %s', update.display_path, err + ) + summary.channels_failed += 1 + logger.info( + 'Channel metadata import complete: %s applied, %s failed.', + summary.channels_updated, summary.channels_failed, + ) + + +def count_planned( + groups: dict[str, MediaGroup], + summary: Summary, + channels: list[ChannelUpdate] | None = None, +) -> None: + """ + Fill in the "correctly processed" counters of ``summary`` for a dry-run. + + Nothing is uploaded; the counters reflect what a subsequent ``--apply`` run would do + with the media that survived the audit and the channels listed in ``channels.csv``. + """ + for group in groups.values(): + if group.object_id: + summary.medias_existing += 1 + else: + summary.medias_imported += 1 + if group.extra_streams: + summary.merged_videos += 1 + if group.metadata is not None: + summary.metadata_applied += 1 + for lang in group.audio_tracks: + if f'audio:{lang}' in group.existing_elements: + summary.audio_tracks_existing += 1 + else: + summary.audio_tracks_imported += 1 + for lang in group.subtitles: + if f'subtitle:{lang}' in group.existing_elements: + summary.subtitles_existing += 1 + else: + summary.subtitles_imported += 1 + for row in group.annotations: + key = f'annotation:{row.get("type")}:{row.get("time")}' + if key in group.existing_elements: + summary.annotations_existing += 1 + else: + summary.annotations_imported += 1 + summary.channels_updated += len(channels or []) + + +def print_summary(summary: Summary, applied: bool) -> None: + """Log the final import report built from ``summary``.""" + done = 'imported' if applied else 'to import' + applied_label = 'applied' if applied else 'to apply' + total_unimportable = sum(len(ids) for ids in summary.unimportable_media.values()) + lines = [ + 'Import report:', + '', + 'Correctly processed content:', + f' - Merged video files: {summary.merged_videos}', + f' - Media {done}: {summary.medias_imported}', + f' - Media already existing: {summary.medias_existing}', + f' - Metadata entries {applied_label}: {summary.metadata_applied}', + ] + for label, existing, imported, failed in ( + ('Audio tracks', summary.audio_tracks_existing, summary.audio_tracks_imported, + summary.audio_tracks_failed), + ('Subtitles', summary.subtitles_existing, summary.subtitles_imported, + summary.subtitles_failed), + ('Annotations', summary.annotations_existing, summary.annotations_imported, + summary.annotations_failed), + ): + lines.append(f' - {label} {done}: {imported}') + lines.append(f' - {label} already existing: {existing}') + if applied: + lines.append(f' - {label} failed: {failed}') + lines.append(f' - Channel metadata entries {applied_label}: {summary.channels_updated}') + if applied: + lines.append(f' - Channel metadata entries failed: {summary.channels_failed}') + lines += [ + '', + 'Unprocessable content:', + f' - Unimportable media: {total_unimportable}', + ] + for reason in sorted(summary.unimportable_media): + lines.append(f' - {reason}: {len(summary.unimportable_media[reason])}') + lines.append(f' - Invalid metadata entries: {summary.invalid_metadata}') + lines.append(f' - Invalid channel entries: {summary.invalid_channels}') + lines.append(f' - Unimportable annotations: {summary.unimportable_annotations}') + if applied: + lines.append(f' - Media that could not be created: {summary.import_failures}') + logger.info('\n'.join(lines)) + + +def truncate_channel_titles(channel: str) -> str: + """ + Truncate the channel titles of a path based target ("mscpath-..." or "mscspeaker-..."). + + A title longer than what Nudgis can store cannot be matched against the catalog, so the + target would end up in a completely different channel; truncating each title keeps the + media close to where it belongs. + """ + for prefix in ('mscpath-', f'{SPEAKER_TARGET}-'): + if channel.startswith(prefix): + titles = channel[len(prefix):].split('/') + return prefix + '/'.join(title[:MAX_FIELD_LENGTH] for title in titles) + return channel + + +def parse_speaker_path(channel: str) -> list[str] | None: + """ + Split the sub-channel path of a personal channel target. + + Return the titles of the channels making up the path below the personal channel of the + speaker (an empty list for the personal channel itself, which is resolved by the server) + or ``None`` when the given channel does not target a personal channel at all. + """ + if channel == SPEAKER_TARGET: + return [] + if channel.startswith(f'{SPEAKER_TARGET}-'): + return [ + title for part in channel[len(SPEAKER_TARGET) + 1:].split('/') + if (title := part.strip()) + ] + return None + + +def resolve_channel( + main_channel: str, + rel_dir: Path, + row: dict[str, str] | None, +) -> str: + """ + Determine the destination channel of a media. + + The channel column of ``metadata.csv`` takes precedence; otherwise the folder tree is + mirrored below the main migration channel using an "mscpath" identifier. A personal + sub-channel target is left as is, it is resolved by :func:`ensure_speaker_channel`. + """ + if row: + channel = (row.get('channel') or '').strip() + if channel: + return channel + base = main_channel[len('mscpath-'):] if main_channel.startswith('mscpath-') else main_channel + parts = [base, *rel_dir.parts] + return 'mscpath-' + '/'.join(parts) + + +def build_media_metadata(row: dict[str, str] | None) -> dict[str, str]: + """ + Build the ``medias/add/`` metadata payload from a ``metadata.csv`` row. + + The source id is always copied to the "external reference" field. + Pipe-separated lists are converted to the format expected by the API. + """ + metadata: dict[str, str] = {} + if not row: + return metadata + # Simple pass-through string fields (spec column -> API parameter name). + passthrough = { + 'title': 'title', + 'slug': 'slug', + 'description': 'description', + 'language': 'language', + 'creation': 'creation', + 'company_name': 'company', + 'company_url': 'company_url', + 'license_name': 'license', + 'license_url': 'license_url', + 'validated': 'validated', + 'unlisted': 'unlisted', + 'detect_slides': 'detect_slides', + } + for column, param in passthrough.items(): + value = (row.get(column) or '').strip() + if value: + metadata[param] = value + # Keywords are stored as a space-separated list on Nudgis. + keywords = _split_pipe(row.get('keywords') or '') + if keywords: + metadata['keywords'] = ','.join(keywords) + # Categories are stored as a newline-separated list on Nudgis. + categories = _split_pipe(row.get('categories') or '') + if categories: + metadata['category'] = '\n'.join(categories) + # Speakers use a pipe-separated list (same convention as the rest of the API). + names = _split_pipe((row.get('speaker_name') or ''), drop_empty=False) + emails = _split_pipe((row.get('speaker_email') or ''), drop_empty=False) + if names: + # Send both "speaker" and "speaker_name" to Nudgis because the API + # may change and we want to be future-proof. + metadata['speaker'] = '|'.join(names) + metadata['speaker_name'] = '|'.join(names) + if emails: + metadata['speaker_email'] = '|'.join(emails) + return metadata + + +def post_annotation( + ngc: NudgisClient, + oid: str, + row: dict[str, str], + source_dir: Path, +) -> None: + """Post a single annotation (optionally with an attachment) to a media.""" + data: dict[str, str] = {'oid': oid} + ann_type = row['type'] + data['type_slug'] = ann_type + for column in ('time', 'title', 'content'): + value = (row.get(column) or '').strip() + if value: + data[column] = value + keywords = _split_pipe(row.get('keywords') or '') + if keywords: + data['keywords'] = ','.join(keywords) + attachment = (row.get('attachment') or '').strip() + if ann_type not in INTERNAL_ANNOTATION_TYPES and not data.get('content'): + data['content'] = '-' + logger.info('Adding "%s" annotation at "%s" to "%s".', ann_type, data.get('time'), oid) + if attachment: + path = source_dir / ANNOTATIONS_DIRNAME / attachment + with path.open('rb') as fileobj: + ngc.api( + 'annotations/post/', + method='post', + data=data, + files={'attachment': (path.name, fileobj)}, + ) + else: + ngc.api('annotations/post/', method='post', data=data) + + +def mass_import(sys_args: list[str]) -> int: + parser = argparse.ArgumentParser( + 'mass_import', + description=__doc__.strip(), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + '--conf', + help='Path to the configuration file (e.g. myconfig.json).', + required=True, + ) + parser.add_argument( + '--source-dir', + help='Path to the directory containing the media to import.', + required=True, + type=Path, + ) + parser.add_argument( + '--ids-to-process', + help='Comma-separated list of source file IDs to import. ' + 'By default, all source files are imported.', + default=None, + type=str, + ) + parser.add_argument( + '--channel', + help='Main migration channel (a title or an "mscpath-..." identifier). ' + 'Important: The main channel cannot be targeted with an object id.', + required=True, + ) + parser.add_argument( + '--apply', + help='Actually perform the import. Without this flag, the script only audits the ' + 'source directory (file tree, CSV files, media integrity and server checks).', + action='store_true', + ) + parser.add_argument( + '--mapping-file', + help='Path of the produced "source_id,oid" mapping CSV file.', + default=Path(f'./mapping_{date.today().strftime("%Y-%m-%d")}.csv'), + type=Path, + ) + parser.add_argument( + '--ffprobe', + help='Path to the ffprobe executable used to check media integrity.', + default='ffprobe', + ) + parser.add_argument( + '--ffmpeg', + help='Path to the ffmpeg executable used to combine multi-stream media.', + default='ffmpeg', + ) + parser.add_argument( + '--temp-dir', + help='Directory for the temporary videos produced when combining multi-stream media.', + default=Path('./temp'), + type=Path, + ) + parser.add_argument( + '--log-level', + help='Log level.', + default='info', + choices=['critical', 'error', 'warn', 'info', 'debug'], + ) + args = parser.parse_args(sys_args) + + configure_logging(args.log_level.upper()) + + if args.channel.startswith('mscid-') or re.match(r'^c[A-Za-z0-9]{19}$', args.channel): + logger.error('The channel cannot be targeted with an object id.') + return 1 + + source_dir = args.source_dir + if not source_dir.is_dir(): + logger.error('Source directory "%s" does not exist.', source_dir) + return 1 + ids_to_process = args.ids_to_process.split(',') if args.ids_to_process else None + + ngc = NudgisClient(args.conf, setup_logging=False) + ngc.conf['TIMEOUT'] = max(600, ngc.conf['TIMEOUT']) + + report, summary, groups, channels = run_audit( + source_dir, ngc, ids_to_process, ffprobe=args.ffprobe, apply=args.apply + ) + + logger.info('Audit completed.') + for warning in report.warnings: + logger.warning(warning) + for error in report.errors: + logger.error(error) + + if not report.ok: + logger.error( + 'Audit failed with %s fatal error(s); aborting.', len(report.errors), + ) + return 1 + + if args.apply: + import_media( + ngc, + groups, + args.channel, + source_dir, + args.mapping_file, + args.temp_dir, + summary, + ffmpeg=args.ffmpeg, + ffprobe=args.ffprobe, + ) + # The channel metadata are applied last so that the channels created while importing + # the media can be targeted by their path. + apply_channels_metadata(ngc, channels, summary) + else: + logger.info( + 'Audit succeeded for %s media (dry-run). Re-run with "--apply" to import.', + len(groups), + ) + count_planned(groups, summary, channels=channels) + + print_summary(summary, applied=args.apply) + return 0 + + +if __name__ == '__main__': # pragma: no cover + sys.exit(mass_import(sys.argv[1:])) diff --git a/examples/nudgisclient b/examples/nudgisclient new file mode 120000 index 0000000..fee09c8 --- /dev/null +++ b/examples/nudgisclient @@ -0,0 +1 @@ +../nudgisclient \ No newline at end of file diff --git a/nudgisclient/client.py b/nudgisclient/client.py index 2427a5e..00954d3 100644 --- a/nudgisclient/client.py +++ b/nudgisclient/client.py @@ -14,6 +14,7 @@ download as download_lib, upload as upload_lib, users_csv as users_csv_lib, + utils as utils_lib, ) logger = logging.getLogger(__name__) @@ -41,11 +42,7 @@ def __init__(self, local_conf: Path | str | dict | None = None, setup_logging: b # `local_conf` can be either a dict, a path (`str` object) or a unix user (`unix:msuser` for example) # Setup logging if setup_logging: - logging.basicConfig( - format='%(asctime)s.%(msecs)03d pid:%(process)d %(name)s %(levelname)s %(message)s', - datefmt='%Y-%m-%d %H:%M:%S', - level=logging.INFO, - ) + utils_lib.configure_logging() # Read conf file self.load_conf(local_conf) # Configure logging diff --git a/nudgisclient/lib/utils.py b/nudgisclient/lib/utils.py index 273b420..f43f158 100644 --- a/nudgisclient/lib/utils.py +++ b/nudgisclient/lib/utils.py @@ -1,4 +1,5 @@ from datetime import timedelta +import logging import re import sys @@ -21,6 +22,14 @@ class TTYColors: GRAY = RED = GREEN = YELLOW = BLUE = PURPLE = TEAL = RESET = '' +def configure_logging(level: str = 'INFO') -> None: + logging.basicConfig( + format='%(asctime)s.%(msecs)03d pid:%(process)d %(name)s %(levelname)s %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', + level=getattr(logging, level), + ) + + def _size_repr(value: int, unit: str, short: bool = True) -> str: # https://en.wikipedia.org/wiki/Template:Quantities_of_bytes if short: diff --git a/tests/examples/test_compose_multistream.py b/tests/examples/test_compose_multistream.py new file mode 100644 index 0000000..87b14cb --- /dev/null +++ b/tests/examples/test_compose_multistream.py @@ -0,0 +1,206 @@ +import json +from pathlib import Path +import subprocess + +import pytest + +import examples.compose_multistream as cm + +# Real sample videos used to exercise ffmpeg for real. +SAMPLES_DIR = Path(__file__).resolve().parent.parent / 'samples' +SAMPLE_VIDEOS = [ + SAMPLES_DIR / 'ball_1920x1080_h264_aac.mp4', + SAMPLES_DIR / 'mire_1280x720_h264_aac.mp4', + SAMPLES_DIR / 'ball_720x540_av1_opus.webm', +] +# A video-only sample (no audio stream). +SAMPLE_NO_AUDIO = SAMPLES_DIR / 'ball_640x480_h265.mp4' + + +def probe_codec_types(path: Path) -> list[str]: + """Return the codec types (e.g. "video", "audio") of the streams of a media file.""" + result = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', 'stream=codec_type', '-of', 'json', str(path)], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + return [stream['codec_type'] for stream in json.loads(result.stdout)['streams']] + + +def _video_size(path: Path) -> tuple[int, int]: + """Return the ``(width, height)`` of the first video stream of a media file.""" + result = subprocess.run( + [ + 'ffprobe', '-v', 'error', '-select_streams', 'v:0', + '-show_entries', 'stream=width,height', '-of', 'json', str(path), + ], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + stream = json.loads(result.stdout)['streams'][0] + return stream['width'], stream['height'] + + +# -------- ffmpeg command building + + +@pytest.mark.parametrize('count, columns', [(2, 2), (4, 2), (5, 3), (6, 3)]) +def test_grid_columns(count, columns): + assert cm._grid_columns(count) == columns + + +def test_xstack_layout(): + assert cm._xstack_layout(2, 2) == '0_0|w0_0' + assert cm._xstack_layout(3, 2) == '0_0|w0_0|0_h0' + assert cm._xstack_layout(5, 3) == '0_0|w0_0|w0+w1_0|0_h0|w3_h0' + + +def test_build_ffmpeg_command_audio_mapping(tmp_path): + inputs = [tmp_path / 'a.mp4', tmp_path / 'a_2.mp4'] + output = tmp_path / 'out' / 'a.mp4' + prefix = ['ffmpeg', '-y', '-i', str(inputs[0]), '-i', str(inputs[1]), '-filter_complex'] + video = '[0:v][1:v]xstack=inputs=2:layout=0_0|w0_0[v]' + + # Both inputs have audio: the tracks are mixed together. + assert cm._build_ffmpeg_command(inputs, output, [0, 1]) == [ + *prefix, f'{video};[0:a][1:a]amix=inputs=2[a]', + '-map', '[v]', '-map', '[a]', str(output), + ] + # Only one input has audio: it is mapped as-is. + assert cm._build_ffmpeg_command(inputs, output, [1]) == [ + *prefix, video, '-map', '[v]', '-map', '1:a', str(output), + ] + # No input has audio: the output has no audio. + assert cm._build_ffmpeg_command(inputs, output, []) == [ + *prefix, video, '-map', '[v]', str(output), + ] + + +def test_build_composition_layout_two_inputs(): + layout = cm._build_composition_layout([(1920, 1080), (1280, 720)], 3200, 1080) + assert layout == { + 'composition_area': {'w': 3200, 'h': 1080}, + 'composition_data': [{ + 'time': 0, + 'layers': [ + { + 'id': 1, 'label': 'element-1', 'enabled': True, + 'source': { + 'type': 'video', + 'roi': {'x': 0, 'y': 0, 'w': 1920, 'h': 1080}, + 'native_resolution': {'w': 3200, 'h': 1080}, + }, + 'x': 0, 'y': 0, 'w': 1920, 'h': 1080, 'z': 1, + }, + { + 'id': 2, 'label': 'element-2', 'enabled': True, + 'source': { + 'type': 'video', + 'roi': {'x': 1920, 'y': 0, 'w': 1280, 'h': 720}, + 'native_resolution': {'w': 3200, 'h': 1080}, + }, + 'x': 1920, 'y': 0, 'w': 1280, 'h': 720, 'z': 2, + }, + ], + }], + } + + +def test_build_composition_layout_grid_positions(): + # 4 inputs -> 2x2 grid: the second row is placed below the first. + sizes = [(100, 50), (200, 50), (100, 60), (200, 60)] + layout = cm._build_composition_layout(sizes, 300, 110) + coords = [(layer['x'], layer['y']) for layer in layout['composition_data'][0]['layers']] + assert coords == [(0, 0), (100, 0), (0, 50), (100, 50)] + + +# -------- compose_streams (real ffmpeg) + + +@pytest.mark.parametrize('count', [2, 3]) +def test_compose_streams_creates_video(tmp_path, count): + output = tmp_path / 'out' / 'combined.mp4' + cm.compose_streams(SAMPLE_VIDEOS[:count], output) + assert output.is_file() and output.stat().st_size > 0 + codec_types = probe_codec_types(output) + assert 'video' in codec_types # the streams were stacked into a single video stream + assert 'audio' in codec_types # the audio tracks were mixed into a single audio stream + + # The layout preset JSON is written next to the output and matches the real video size. + layout = json.loads(output.with_suffix('.json').read_text()) + layers = layout['composition_data'][0]['layers'] + assert len(layers) == count + assert layers[0]['source']['roi'] == {'x': 0, 'y': 0, 'w': 1920, 'h': 1080} + assert [layer['id'] for layer in layers] == list(range(1, count + 1)) + assert (layout['composition_area']['w'], layout['composition_area']['h']) == _video_size(output) + + +def test_compose_streams_partial_audio(tmp_path): + # One input has no audio: the audio of the other input is still kept. + output = tmp_path / 'combined.mp4' + cm.compose_streams([SAMPLE_NO_AUDIO, SAMPLE_VIDEOS[0]], output) + codec_types = probe_codec_types(output) + assert 'video' in codec_types + assert 'audio' in codec_types + + +def test_compose_streams_without_audio(tmp_path): + # No input has audio: the resulting video has no audio stream. + output = tmp_path / 'combined.mp4' + cm.compose_streams([SAMPLE_NO_AUDIO, SAMPLE_NO_AUDIO], output) + assert probe_codec_types(output) == ['video'] + + +def test_compose_streams_invalid_count(tmp_path): + with pytest.raises(ValueError): + cm.compose_streams(SAMPLE_VIDEOS[:1], tmp_path / 'o.mp4') + with pytest.raises(ValueError): + cm.compose_streams(SAMPLE_VIDEOS * 3, tmp_path / 'o.mp4') # 9 inputs > maximum + + +def test_compose_streams_ffmpeg_failure(tmp_path): + # A non-video input makes ffmpeg exit with a non-zero status. + bogus = tmp_path / 'bogus.mp4' + bogus.write_bytes(b'not a video') + with pytest.raises(RuntimeError, match='ffmpeg failed'): + cm.compose_streams([bogus, SAMPLE_VIDEOS[0]], tmp_path / 'o.mp4') + + +def test_compose_streams_ffmpeg_not_found(tmp_path): + with pytest.raises(RuntimeError, match='not found'): + cm.compose_streams(SAMPLE_VIDEOS[:2], tmp_path / 'o.mp4', ffmpeg='ffmpeg-does-not-exist') + + +def test_compose_streams_ffprobe_not_found(tmp_path): + with pytest.raises(RuntimeError, match='not found'): + cm.compose_streams(SAMPLE_VIDEOS[:2], tmp_path / 'o.mp4', ffprobe='ffprobe-does-not-exist') + + +# -------- CLI (real ffmpeg) + + +def test_cli_success(tmp_path): + output = tmp_path / 'combined' / 'a.mp4' + assert cm.compose_multistream([ + '--inputs', str(SAMPLE_VIDEOS[0]), str(SAMPLE_VIDEOS[1]), + '--output', str(output), '--log-level', 'debug', + ]) == 0 + assert output.is_file() + assert 'video' in probe_codec_types(output) + + +def test_cli_missing_source(tmp_path): + assert cm.compose_multistream([ + '--inputs', str(SAMPLE_VIDEOS[0]), str(tmp_path / 'missing.mp4'), + '--output', str(tmp_path / 'o.mp4'), + ]) == 1 + + +def test_cli_compose_error(tmp_path): + bogus = tmp_path / 'bogus.mp4' + bogus.write_bytes(b'not a video') + assert cm.compose_multistream([ + '--inputs', str(bogus), str(SAMPLE_VIDEOS[0]), '--output', str(tmp_path / 'o.mp4'), + ]) == 1 diff --git a/tests/examples/test_mass_import.py b/tests/examples/test_mass_import.py new file mode 100644 index 0000000..83c3fbd --- /dev/null +++ b/tests/examples/test_mass_import.py @@ -0,0 +1,1915 @@ +import logging +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +import pytest + +import examples.mass_import as mi + +# -------- helpers + + +def write(path: Path, content: bytes = b'data') -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return path + + +def group(source_id: str, rel_dir: str = '.', object_id=None) -> mi.MediaGroup: + return mi.MediaGroup(source_id, object_id, Path(f'{source_id}.mp4'), Path(rel_dir)) + + +def group_with( + source_id, *, metadata=None, annotations=None, rel_dir='.', object_id=None, + existing_elements=None, +) -> mi.MediaGroup: + g = group(source_id, rel_dir, object_id) + g.metadata = metadata + g.annotations = annotations or [] + g.existing_elements = existing_elements or [] + return g + + +def make_client( + *, + check_server_exc=None, + catalog=None, + catalog_exc=None, + api_map=None, + api_exc=None, +): + client = mi.NudgisClient() + client._server_version = (12, 3, 0) + client.check_server = mock.MagicMock( + side_effect=check_server_exc, return_value={} + ) + client.get_catalog = mock.MagicMock( + side_effect=catalog_exc, return_value=catalog if catalog is not None else {} + ) + client.add_media = mock.MagicMock( + side_effect=lambda **kwargs: { + 'oid': 'v_' + kwargs.get('external_ref', 'x'), + 'slug': kwargs.get('slug', ''), + } + ) + + def api(url, **kwargs): + if api_exc and url in api_exc: + raise api_exc[url] + if api_map and url in api_map: + return api_map[url] + # By default the API key has the permissions required by server_checks. + if url == 'users/me/': + return {'user': {'permissions': {'can_change_users': True}}} + # Channel creation happens when applying the "channels.csv" metadata. + if url == 'channels/add/': + return {'oid': 'c_new'} + return {} + + client.api = mock.MagicMock(side_effect=api) + return client + + +METADATA_HEADER = ( + 'source_id,title,slug,description,keywords,categories,language,creation,' + 'speaker_name,speaker_email,company_name,company_url,license_name,license_url,' + 'channel,validated,unlisted,detect_slides' +) +CHANNELS_HEADER = 'path,description,reference' + + +# -------- scan + + +def test_scan_basic(tmp_path): + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm1_fre.srt') + write(tmp_path / 'm1_eng.mp3') + write(tmp_path / 'sub' / 'm2.mp4') + write(tmp_path / 'metadata.csv', b'ignored') + write(tmp_path / 'channels.csv', b'ignored') + write(tmp_path / 'annotations' / 'annotations.csv', b'ignored') + write(tmp_path / 'readme.txt') # unexpected file -> warning + + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + + assert set(groups) == {'m1', 'm2'} + assert set(groups['m1'].subtitles) == {'fre'} + assert set(groups['m1'].audio_tracks) == {'eng'} + assert groups['m2'].rel_dir == Path('sub') + assert not report.errors + # The CSV files of the migration standard are validated separately, not scanned here. + assert sum('Ignoring unexpected file' in w for w in report.warnings) == 1 + assert any('readme.txt' in w for w in report.warnings) + + +def test_scan_filename_too_long(tmp_path): + write(tmp_path / 'ok.mp4') + # 201 chars: allowed by the filesystem but above MAX_FIELD_LENGTH (200). + write(tmp_path / ('x' * 197 + '.mp4')) + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + # The too-long video is skipped as unimportable, the valid one is kept. + assert set(groups) == {'ok'} + assert len(summary.unimportable_media[mi.UNIMPORTABLE_FILENAME]) == 1 + assert any('name too long' in w for w in report.warnings) + assert not report.errors + + +def test_scan_filename_too_long_non_video(tmp_path): + write(tmp_path / 'ok.mp4') + # A non-video file with a too-long name is simply ignored, not counted as a media. + write(tmp_path / ('x' * 247 + '.srt')) + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + assert set(groups) == {'ok'} + assert not summary.unimportable_media + assert any('too-long name' in w for w in report.warnings) + assert not report.errors + + +def test_scan_multistream_warnings(tmp_path): + write(tmp_path / 'multi.mp4') + write(tmp_path / 'multi_stream2.mp4') + write(tmp_path / 'multi_stream4.mp4') # gap -> non-contiguous warning + write(tmp_path / 'noLang.srt') # no "_lang" suffix + + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + + assert set(groups['multi'].extra_streams) == {2, 4} + assert any('non-contiguous' in w for w in report.warnings) + assert any('will be combined with ffmpeg' in w for w in report.warnings) + assert any('naming rule' in w for w in report.warnings) + assert not report.errors + + +def test_scan_duplicate_source_id(tmp_path): + write(tmp_path / 'a' / 'dup.mp4') + write(tmp_path / 'b' / 'dup.mp4') + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + # The first file is kept, the duplicate is skipped as unimportable. + assert set(groups) == {'dup'} + assert summary.unimportable_media[mi.UNIMPORTABLE_DUPLICATE] == ['dup'] + assert any('Duplicate source id' in w for w in report.warnings) + assert not report.errors + + +def test_scan_multistream_without_main(tmp_path): + write(tmp_path / 'clip.mp4') + write(tmp_path / 'clip_stream2.mp4') + write(tmp_path / 'clip_stream2_stream2.mp4') # secondary of "clip_stream2", not a main media + report = mi.Report() + summary = mi.Summary() + mi.scan_source_tree(tmp_path, None, report, summary) + assert any('no main media' in w for w in report.warnings) + assert not report.errors + + +def test_scan_multistream_out_of_range(tmp_path): + write(tmp_path / 'range.mp4') + write(tmp_path / 'range_stream1.mp4') # index < 2 + write(tmp_path / 'range_stream7.mp4') # index > 6 + report = mi.Report() + summary = mi.Summary() + mi.scan_source_tree(tmp_path, None, report, summary) + assert sum('out-of-range' in w for w in report.warnings) == 2 + assert not report.errors + + +def test_scan_linked_orphan(tmp_path): + write(tmp_path / 'vid.mp4') + write(tmp_path / 'orphan_fre.srt') # base "orphan" has no media + report = mi.Report() + summary = mi.Summary() + mi.scan_source_tree(tmp_path, None, report, summary) + assert any('no main media' in w for w in report.warnings) + assert not report.errors + + +def test_scan_linked_duplicate(tmp_path): + write(tmp_path / 'vid.mp4') + write(tmp_path / 'vid_fre.srt') + write(tmp_path / 'vid_fre.vtt') # same base + language + report = mi.Report() + summary = mi.Summary() + mi.scan_source_tree(tmp_path, None, report, summary) + assert any('duplicate subtitle' in w.lower() for w in report.warnings) + assert not report.errors + + +def test_scan_empty(tmp_path): + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, None, report, summary) + assert groups == {} + assert any('No media file to import found' in err for err in report.errors) + + +def test_scan_ids_to_process(tmp_path): + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm1_fre.srt') + write(tmp_path / 'm2.mp4') + write(tmp_path / 'm2_fre.srt') + write(tmp_path / 'm2_stream2.mp4') + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, ['m1'], report, summary) + assert set(groups) == {'m1'} + assert set(groups['m1'].subtitles) == {'fre'} + assert not report.errors + + +def test_scan_ids_to_process_prefix_collisions(tmp_path): + # Files sharing a prefix with an id_to_process but not exact matches exercise + # the exact-match guards for main files, secondary streams and linked files. + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm1extra.mp4') + write(tmp_path / 'm1extra_stream2.mp4') + write(tmp_path / 'm1extra_fre.srt') + report = mi.Report() + summary = mi.Summary() + groups = mi.scan_source_tree(tmp_path, ['m1'], report, summary) + assert set(groups) == {'m1'} + assert not groups['m1'].extra_streams + assert not report.errors + + +# -------- metadata + + +def test_metadata_missing_file(tmp_path): + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_metadata_csv(tmp_path / 'metadata.csv', groups, None, report, summary) + assert groups['m1'].metadata is None + assert any('No "metadata.csv"' in w for w in report.warnings) + + +def test_metadata_missing_columns(tmp_path): + path = write(tmp_path / 'metadata.csv', b'foo,bar\n1,2\n') + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_metadata_csv(path, groups, None, report, summary) + assert groups['m1'].metadata is None + # A structurally broken CSV is a fatal error. + assert any('missing mandatory columns' in err for err in report.errors) + + +def test_metadata_unknown_column(tmp_path): + path = write(tmp_path / 'metadata.csv', b'source_id,title,weird\nm1,Title,x\n') + report = mi.Report() + summary = mi.Summary() + mi.validate_metadata_csv(path, {'m1': group('m1')}, None, report, summary) + assert any('unknown column "weird"' in w for w in report.warnings) + + +def test_metadata_valid(tmp_path): + content = ( + METADATA_HEADER + '\n' + 'm1,"Title 1",my-slug,desc,a|b,c1|c2,fre,2026-02-26T17:10:00,' + 'Jane|John,jane@x|john@x,Comp,http://c,Lic,http://l,,yes,no,no\n' + ) + path = write(tmp_path / 'metadata.csv', content.encode('utf-8')) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_metadata_csv(path, groups, None, report, summary) + assert groups['m1'].metadata is not None + assert not report.errors + assert summary.invalid_metadata == 0 + + +def test_metadata_invalid_rows_dropped(tmp_path): + # Rows that cannot be tied to a media are dropped and counted as invalid entries. + content = ( + 'source_id,title,slug,language,creation,speaker_name,speaker_email,validated\n' + ',Title,,,,,,\n' # empty source_id + 'good,Title,goodslug,fre,2026-01-01T00:00:00,,,yes\n' # valid row (slug recorded) + 'good,Title,other,fre,,,,\n' # duplicate source_id + 'ghost,Title,,,,,,\n' # source_id not in groups + ) + path = write(tmp_path / 'metadata.csv', content.encode('utf-8')) + groups = {'good': group('good')} + report = mi.Report() + summary = mi.Summary() + mi.validate_metadata_csv(path, groups, None, report, summary) + assert not report.errors + assert summary.invalid_metadata == 3 + assert groups['good'].metadata['slug'] == 'goodslug' + + +def test_metadata_invalid_fields_cleaned(tmp_path): + # Invalid optional fields are cleaned in place so the media is still importable. + content = ( + 'source_id,title,slug,language,creation,speaker_name,speaker_email,validated\n' + 'm2,,Bad Slug,xx,not-a-date,A|B,a@x,maybe\n' # title/slug/lang/date/speaker/validated + 'm3,Title,goodslug,,,,,\n' # slug recorded on m3 + 'm4,Title,goodslug,,,,,\n' # duplicate slug -> cleaned + ) + path = write(tmp_path / 'metadata.csv', content.encode('utf-8')) + groups = {'m2': group('m2'), 'm3': group('m3'), 'm4': group('m4')} + report = mi.Report() + summary = mi.Summary() + mi.validate_metadata_csv(path, groups, None, report, summary) + assert not report.errors + assert summary.invalid_metadata == 0 + m2 = groups['m2'].metadata + assert m2['title'] == '' + assert m2['slug'] == '' + assert m2['language'] == '' + assert m2['creation'] == '' + assert m2['speaker_name'] == '' and m2['speaker_email'] == '' + assert m2['validated'] == '' + assert groups['m3'].metadata['slug'] == 'goodslug' + assert groups['m4'].metadata['slug'] == '' # duplicate slug dropped + warnings = '\n'.join(report.warnings) + assert 'invalid "slug"' in warnings + assert 'invalid "language"' in warnings + assert 'invalid "creation"' in warnings + assert 'different number of values' in warnings + assert 'invalid "validated"' in warnings + assert 'already used' in warnings + + +def test_metadata_too_long_field(tmp_path): + long_title = 'T' * 300 + path = write( + tmp_path / 'metadata.csv', + (METADATA_HEADER + f'\nm1,{long_title},,,,,,,,,,,,,,,,\n').encode('utf-8'), + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_metadata_csv(path, groups, None, report, summary) + assert any('value too long' in w for w in report.warnings) + assert len(groups['m1'].metadata['title']) == mi.MAX_FIELD_LENGTH + assert not report.errors + + +@pytest.mark.parametrize('prefix', ['mscspeaker-', 'mscpath-']) +def test_metadata_too_long_channel_title(tmp_path, prefix): + # An over long channel title is truncated so that the media stays close to where it + # belongs, instead of being sent to a completely different channel. + row = ['m1', 'Title'] + [''] * 12 + [f'{prefix}Courses/' + 'T' * 300] + [''] * 3 + path = write( + tmp_path / 'metadata.csv', + (METADATA_HEADER + '\n' + ','.join(row) + '\n').encode('utf-8'), + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_metadata_csv(path, groups, None, report, summary) + assert any('channel title' in w and 'truncated' in w for w in report.warnings) + assert groups['m1'].metadata['channel'] == f'{prefix}Courses/' + 'T' * mi.MAX_FIELD_LENGTH + assert not report.errors + + +def test_metadata_media_without_row(tmp_path): + path = write( + tmp_path / 'metadata.csv', (METADATA_HEADER + '\nm1,Title,,,,,,,,,,,,,,,,\n').encode() + ) + report = mi.Report() + summary = mi.Summary() + mi.validate_metadata_csv(path, {'m1': group('m1'), 'm2': group('m2')}, None, report, summary) + assert any('Media "m2" has no row' in w for w in report.warnings) + + +def test_metadata_csv_ids_to_process(tmp_path): + content = METADATA_HEADER + '\nm1,Title 1,,,,,,,,,,,,,,,,\nm2,Title 2,,,,,,,,,,,,,,,,\n' + path = write(tmp_path / 'metadata.csv', content.encode('utf-8')) + groups = {'m1': group('m1')} + report = mi.Report() + summary = mi.Summary() + mi.validate_metadata_csv(path, groups, ['m1'], report, summary) + assert groups['m1'].metadata is not None + assert not report.errors + assert summary.invalid_metadata == 0 + + +# -------- channels + + +def test_channels_missing_file(tmp_path): + report = mi.Report() + summary = mi.Summary() + assert mi.validate_channels_csv(tmp_path / 'channels.csv', report, summary) == [] + assert any('No "channels.csv"' in w for w in report.warnings) + assert not report.errors + + +def test_channels_missing_columns(tmp_path): + path = write(tmp_path / 'channels.csv', b'foo,bar\n1,2\n') + report = mi.Report() + summary = mi.Summary() + assert mi.validate_channels_csv(path, report, summary) == [] + # A structurally broken CSV is a fatal error. + assert any('missing mandatory columns' in err for err in report.errors) + + +def test_channels_unknown_column(tmp_path): + path = write(tmp_path / 'channels.csv', b'path,description,weird\nCourse A,Desc,x\n') + report = mi.Report() + summary = mi.Summary() + channels = mi.validate_channels_csv(path, report, summary) + assert len(channels) == 1 + assert any('unknown column "weird"' in w for w in report.warnings) + + +def test_channels_valid(tmp_path): + content = ( + CHANNELS_HEADER + '\n' + '"Course A/Year 1","

My description

",lti:moodle.example.local:245\n' + ' Course B / Year 2 ,,lti:moodle.example.local:246\n' # surrounding spaces are stripped + '/Course C/,Only a description,\n' # leading and trailing separators are stripped + ) + path = write(tmp_path / 'channels.csv', content.encode('utf-8')) + report = mi.Report() + summary = mi.Summary() + channels = mi.validate_channels_csv(path, report, summary) + assert [channel.path for channel in channels] == [ + ['Course A', 'Year 1'], ['Course B', 'Year 2'], ['Course C'], + ] + assert channels[0].description == '

My description

' + assert channels[0].reference == 'lti:moodle.example.local:245' + assert channels[0].display_path == 'Course A/Year 1' + assert channels[1].description == '' + assert channels[2].reference == '' + assert not report.errors + assert summary.invalid_channels == 0 + + +def test_channels_invalid_rows_dropped(tmp_path): + content = ( + CHANNELS_HEADER + '\n' + ',Desc,\n' # empty path + 'Course A//Year 1,Desc,\n' # empty channel title in the path + 'Course A,Desc,\n' # valid row + 'Course A,Other desc,\n' # duplicate path + 'Course B,,\n' # nothing to apply + ) + path = write(tmp_path / 'channels.csv', content.encode('utf-8')) + report = mi.Report() + summary = mi.Summary() + channels = mi.validate_channels_csv(path, report, summary) + assert [channel.path for channel in channels] == [['Course A']] + assert channels[0].description == 'Desc' + assert summary.invalid_channels == 4 + assert not report.errors + warnings = '\n'.join(report.warnings) + assert 'empty or invalid "path"' in warnings + assert 'duplicate "path"' in warnings + assert 'no metadata to apply' in warnings + + +def test_channels_too_long_title_truncated(tmp_path): + # An over long channel title is truncated (as in the "channel" column of "metadata.csv") + # so that the metadata are applied to the channel the media have been imported into. + long_title = 'T' * (mi.MAX_FIELD_LENGTH + 1) + content = CHANNELS_HEADER + '\n' + f'Course A/{long_title},Desc,\n' + path = write(tmp_path / 'channels.csv', content.encode('utf-8')) + report = mi.Report() + summary = mi.Summary() + channels = mi.validate_channels_csv(path, report, summary) + assert [channel.path for channel in channels] == [ + ['Course A', 'T' * mi.MAX_FIELD_LENGTH], + ] + warnings = '\n'.join(report.warnings) + assert f'longer than {mi.MAX_FIELD_LENGTH} characters, truncated' in warnings + assert summary.invalid_channels == 0 + assert not report.errors + + +def test_channels_unusable_reference_dropped(tmp_path): + long_reference = 'lti:moodle.example.local:' + '1' * mi.MAX_FIELD_LENGTH + content = ( + CHANNELS_HEADER + '\n' + f'Course A,Desc,{long_reference}\n' # too long to be stored + 'Course B,Desc,any free-form value\n' # the reference format is not constrained + 'Course C,Desc,lti:moodle.example.local:245\n' + 'Course D,Desc,lti:moodle.example.local:245\n' # reference already used + ) + path = write(tmp_path / 'channels.csv', content.encode('utf-8')) + report = mi.Report() + summary = mi.Summary() + channels = mi.validate_channels_csv(path, report, summary) + # The rows are kept (their description can be applied), only the reference is dropped. + assert [channel.reference for channel in channels] == [ + '', 'any free-form value', 'lti:moodle.example.local:245', '', + ] + assert summary.invalid_channels == 0 + assert not report.errors + warnings = '\n'.join(report.warnings) + assert '"reference" value too long' in warnings + assert f'at most {mi.MAX_FIELD_LENGTH} characters' in warnings + assert 'already used by "Course C"' in warnings + + +# -------- annotations + + +def test_annotations_no_dir(tmp_path): + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_annotations_csv(tmp_path, groups, None, report, summary) + assert groups['m1'].annotations == [] + assert not report.errors + + +def test_annotations_dir_without_csv(tmp_path): + (tmp_path / 'annotations').mkdir() + report = mi.Report() + summary = mi.Summary() + mi.validate_annotations_csv(tmp_path, {'m1': group('m1')}, None, report, summary) + assert any('No "annotations.csv"' in err for err in report.errors) + + +def test_annotations_missing_columns(tmp_path): + write(tmp_path / 'annotations' / 'annotations.csv', b'source_id,time\nm1,1\n') + report = mi.Report() + summary = mi.Summary() + mi.validate_annotations_csv(tmp_path, {'m1': group('m1')}, None, report, summary) + assert any('missing mandatory columns' in err for err in report.errors) + + +def test_annotations_unknown_column(tmp_path): + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,weird\nm1,slide,1,x\n', + ) + report = mi.Report() + summary = mi.Summary() + mi.validate_annotations_csv(tmp_path, {'m1': group('m1')}, None, report, summary) + assert any('unknown column "weird"' in w for w in report.warnings) + + +def test_annotations_valid(tmp_path): + write(tmp_path / 'annotations' / 'doc.pdf') + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,title,content,keywords,attachment\n' + b'm1,chapter,1000,Intro,Hello,a|b,\n' + b'm1,slide,2000,Slide,,,doc.pdf\n', + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_annotations_csv(tmp_path, groups, None, report, summary) + assert len(groups['m1'].annotations) == 2 + assert not report.errors + assert summary.unimportable_annotations == 0 + + +def test_annotations_csv_ids_to_process(tmp_path): + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,title\nm1,chapter,1000,Intro\nm2,chapter,2000,Other\n', + ) + groups = {'m1': group('m1')} + report = mi.Report() + summary = mi.Summary() + mi.validate_annotations_csv(tmp_path, groups, ['m1'], report, summary) + assert len(groups['m1'].annotations) == 1 + assert not report.errors + + +def test_annotations_unimportable_dropped(tmp_path): + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,title,attachment\n' + b',chapter,1,,\n' # empty source_id -> dropped + b'ghost,slide,1,Title,a.pdf\n' # source_id not in groups -> dropped + b'm1,,1,Title,\n' # empty type -> dropped + b'm1,chapter,1,,\n' # chapter without title -> dropped + b'm1,chapter,1,Title,doc.pdf\n' # chapter with attachment -> cleaned + kept + b'm1,slide,abc,Title,\n' # slide without attachment -> dropped + b'm1,slide,1,Title,missing.pdf\n', # attachment does not exist -> dropped + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_annotations_csv(tmp_path, groups, None, report, summary) + assert not report.errors + assert summary.unimportable_annotations == 6 + # Only the cleaned chapter survives, without its (forbidden) attachment. + assert len(groups['m1'].annotations) == 1 + assert groups['m1'].annotations[0]['type'] == 'chapter' + assert groups['m1'].annotations[0]['attachment'] == '' + + +def test_annotations_invalid_time_reset(tmp_path): + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,title\nm1,chapter,abc,Intro\n', + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_annotations_csv(tmp_path, groups, None, report, summary) + assert not report.errors + assert groups['m1'].annotations[0]['time'] == '0' + assert any('reset to 0' in w for w in report.warnings) + + +def test_annotations_too_long_field(tmp_path): + long_title = 'T' * 300 + write( + tmp_path / 'annotations' / 'annotations.csv', + (f'source_id,type,time,title\nm1,chapter,1000,{long_title}\n').encode('utf-8'), + ) + report = mi.Report() + summary = mi.Summary() + groups = {'m1': group('m1')} + mi.validate_annotations_csv(tmp_path, groups, None, report, summary) + assert any('value too long' in w for w in report.warnings) + assert len(groups['m1'].annotations[0]['title']) == mi.MAX_FIELD_LENGTH + assert not report.errors + + +# -------- ffprobe + + +def _completed(returncode=0, stdout='', stderr=''): + return SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr) + + +def test_probe_not_found(tmp_path): + with mock.patch.object(mi.subprocess, 'run', side_effect=FileNotFoundError): + assert 'not found' in mi.probe_media(tmp_path / 'x.mp4') + + +def test_probe_failure(tmp_path): + with mock.patch.object(mi.subprocess, 'run', return_value=_completed(1, stderr='boom')): + assert 'ffprobe failed' in mi.probe_media(tmp_path / 'x.mp4') + + +def test_probe_bad_json(tmp_path): + with mock.patch.object(mi.subprocess, 'run', return_value=_completed(0, stdout='nope')): + assert 'invalid output' in mi.probe_media(tmp_path / 'x.mp4') + + +def test_probe_no_streams(tmp_path): + with mock.patch.object( + mi.subprocess, 'run', return_value=_completed(0, stdout='{"streams": []}') + ): + assert 'no media stream' in mi.probe_media(tmp_path / 'x.mp4') + + +def test_probe_valid(tmp_path): + with mock.patch.object( + mi.subprocess, + 'run', + return_value=_completed(0, stdout='{"streams": [{"codec_type": "video"}]}'), + ): + assert mi.probe_media(tmp_path / 'x.mp4') is None + + +def test_validate_media_integrity_drops_corrupted(): + g = group('m1') + g.extra_streams[2] = Path('m1_2.mp4') + g.audio_tracks['eng'] = Path('m1_eng.mp3') + report = mi.Report() + summary = mi.Summary() + groups = {'m1': g} + with mock.patch.object(mi, 'probe_media', return_value='bad'): + mi.validate_media_integrity(groups, report, summary) + # The whole media is dropped as soon as one of its files is corrupted. + assert 'm1' not in groups + assert summary.unimportable_media[mi.UNIMPORTABLE_CORRUPTED] == ['m1'] + assert not report.errors + + +def test_validate_media_integrity_valid(): + groups = {'m1': group('m1')} + report = mi.Report() + summary = mi.Summary() + with mock.patch.object(mi, 'probe_media', return_value=None): + mi.validate_media_integrity(groups, report, summary) + assert set(groups) == {'m1'} + assert not report.errors + assert not summary.unimportable_media + + +def test_validate_media_integrity_ffprobe_missing(): + groups = {'m1': group('m1')} + report = mi.Report() + summary = mi.Summary() + with mock.patch.object(mi, 'probe_media', return_value='"ffprobe" command not found; boom'): + mi.validate_media_integrity(groups, report, summary) + # A missing ffprobe is fatal, media are not dropped. + assert set(groups) == {'m1'} + assert any('command not found' in err for err in report.errors) + assert not summary.unimportable_media + + +# -------- server checks + + +def request_error(status_code=500): + return mi.NudgisRequestError('boom', status_code=status_code) + + +def test_server_unreachable(): + report = mi.Report() + summary = mi.Summary() + client = make_client(check_server_exc=RuntimeError('down')) + mi.server_checks(client, {}, report, summary) + assert any('Cannot reach' in err for err in report.errors) + client.get_catalog.assert_not_called() + + +def test_server_api_permission_missing(): + # An API key whose account lacks the required permission aborts the run. + client = make_client( + api_map={'users/me/': {'user': {'permissions': {'can_change_users': False}}}}, + ) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': group_with('a')}, report, summary) + assert any('does not have the required permissions' in err for err in report.errors) + client.get_catalog.assert_not_called() + + +def test_server_api_check_error(): + # A failure while testing the API key aborts the run. + client = make_client(api_exc={'users/me/': request_error()}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': group_with('a')}, report, summary) + assert any('testing the API' in err for err in report.errors) + client.get_catalog.assert_not_called() + + +def test_server_full(): + groups = { + 'a': group_with( + 'a', + metadata={'slug': 'taken', 'channel': ''}, + annotations=[{'type': 'custom'}, {'type': 'chapter'}], + ), + 'b': group_with('b', metadata={'slug': 'free', 'channel': 'mscspeaker', + 'speaker_email': 'known@x'}), + 'c': group_with('c', metadata={'channel': 'mscspeaker', 'speaker_email': ''}), + 'd': group_with('d', metadata={'channel': 'mscid-CID'}), # resolved by oid + 'g': group_with('g', metadata={'channel': 'my-channel'}), # resolved by slug + 'f': group_with('f', metadata={'channel': 'mscpath-Foo'}), # auto-created, skipped + } + client = make_client( + catalog={'videos': [{'slug': 'taken', 'oid': 'vexist'}], 'channels': [{}]}, + api_map={ + 'annotations/types/list/': {'types': [{'id': 1, 'slug': 'other'}]}, + 'users/': {'users': [{'email': 'known@x', 'id': '25508'}]}, + 'perms/get/': {'global_permissions': {'can_have_personal_channel': {'val': True}}}, + }, + ) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + warnings = '\n'.join(report.warnings) + assert not report.errors + # An already used slug is only a warning (the media may be re-imported). + assert 'Slug "taken"' in warnings + assert 'Slug "free"' not in warnings + # The annotation using an unknown type is dropped, the internal one is kept. + assert [row['type'] for row in groups['a'].annotations] == ['chapter'] + assert summary.unimportable_annotations == 1 + # A media targeting "mscspeaker" without a speaker email is dropped. + assert 'c' not in groups + assert summary.unimportable_media[mi.UNIMPORTABLE_NO_SPEAKER] == ['c'] + # The explicit channels (oid and slug) were looked up on the server. + channel_calls = [c for c in client.api.call_args_list if c.args[0] == 'channels/get/'] + assert {tuple(c.kwargs['params'].items()) for c in channel_calls} == { + (('oid', 'CID'),), (('slug', 'my-channel'),), + } + + +def test_server_existing_elements(): + # A media already on the server (matched by external ref) has its existing audio + # tracks, subtitles and annotations collected so they can be skipped on import. + g = group_with('a') + client = make_client( + catalog={'videos': [{'external_ref': 'migration:a', 'oid': 'vexist'}]}, + api_map={ + 'medias/audio/tracks/list/': {'audio_tracks': [ + {'language': 'fre', 'is_original': False}, + {'language': 'eng', 'is_original': True}, # original -> skipped + ]}, + 'subtitles/': {'subtitles': [ + {'lang_code': 'fre', 'auto_transcripted': False, 'auto_translated': False}, + {'lang_code': 'eng', 'auto_translated': True}, # auto -> skipped + ]}, + 'annotations/list/': { + 'types': {10: {'id': 10, 'slug': 'chapter'}}, + 'annotations': [{'type_id': 10, 'time': 1000}], + }, + }, + ) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': g}, report, summary) + assert g.object_id == 'vexist' + assert g.existing_elements == ['audio:fre', 'subtitle:fre', 'annotation:chapter:1000'] + assert not report.errors + + +def test_server_existing_elements_not_found(): + g = group_with('a', object_id='vexist') + err = request_error(status_code=404) + client = make_client(api_exc={ + 'medias/audio/tracks/list/': err, + 'subtitles/': err, + 'annotations/list/': err, + }) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': g}, report, summary) + assert g.existing_elements == [] + assert not report.errors + + +def test_server_existing_elements_error(): + # A listing failure no longer aborts the run: it is a warning and the import proceeds. + g = group_with('a', object_id='vexist') + err = request_error(status_code=500) + client = make_client(api_exc={ + 'medias/audio/tracks/list/': err, + 'subtitles/': err, + 'annotations/list/': err, + }) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': g}, report, summary) + assert not report.errors + warnings = '\n'.join(report.warnings) + assert 'Could not list audio tracks' in warnings + assert 'Could not list subtitles' in warnings + assert 'Could not list annotations' in warnings + + +def test_server_catalog_error(): + client = make_client(catalog_exc=request_error()) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {'a': group_with('a', metadata={'slug': 's'})}, report, summary) + assert any('check slugs' in w for w in report.warnings) + + +def test_server_annotation_types_error(): + client = make_client(api_exc={'annotations/types/list/': request_error()}) + report = mi.Report() + summary = mi.Summary() + groups = {'a': group_with('a', annotations=[{'type': 'custom'}])} + mi.server_checks(client, groups, report, summary) + assert any('annotation types' in w for w in report.warnings) + # The annotations are kept when the types cannot be checked. + assert len(groups['a'].annotations) == 1 + + +def test_server_speaker_user_missing(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'ghost@x'})} + client = make_client(api_map={'users/': {'users': []}}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + # In audit mode the missing user is only reported, not created. + assert any('does not exist yet' in w for w in report.warnings) + assert not report.errors + assert not any(c.args[0] == 'users/add/' for c in client.api.call_args_list) + + +def test_server_speaker_user_created(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'ghost@x'})} + client = make_client(api_map={ + 'users/': {'users': []}, + 'users/add/': {'id': '77'}, + 'perms/get/': {'global_permissions': {'can_have_personal_channel': {'val': False}}}, + }) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary, apply=True) + urls = [c.args[0] for c in client.api.call_args_list] + # The missing user is created and granted the personal-channel permission. + assert 'users/add/' in urls + assert 'perms/edit/' in urls + assert not report.errors + + +def test_server_speaker_permission_already_granted(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'known@x'})} + client = make_client(api_map={ + 'users/': {'users': [{'email': 'known@x', 'id': '25508'}]}, + 'perms/get/': {'global_permissions': {'can_have_personal_channel': {'inherit_val': True}}}, + }) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary, apply=True) + urls = [c.args[0] for c in client.api.call_args_list] + assert 'perms/edit/' not in urls + assert not report.errors + + +def test_server_speaker_create_error(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'ghost@x'})} + client = make_client( + api_map={'users/': {'users': []}}, + api_exc={'users/add/': request_error()}, + ) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary, apply=True) + assert any('Could not create speaker' in w for w in report.warnings) + # No permission is checked when the user could not be created. + assert not any(c.args[0] == 'perms/get/' for c in client.api.call_args_list) + + +def test_server_speaker_create_without_id(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'ghost@x'})} + client = make_client(api_map={'users/': {'users': []}, 'users/add/': {}}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary, apply=True) + # The created user has no id, so no permission is checked. + assert not any(c.args[0] == 'perms/get/' for c in client.api.call_args_list) + assert not report.errors + + +def test_server_speaker_user_error(): + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker', + 'speaker_email': 'ghost@x'})} + client = make_client(api_exc={'users/': request_error()}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + assert any('Could not check speaker' in w for w in report.warnings) + + +def test_server_speaker_subchannel_target(): + # A personal sub-channel target resolves the speaker like "mscspeaker" does, and is not + # looked up as an explicit channel (it is resolved during the import). + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker-Courses/2026', + 'speaker_email': 'known@x'})} + client = make_client(api_map={ + 'users/': {'users': [{'email': 'known@x', 'id': '25508'}]}, + 'perms/get/': {'global_permissions': {'can_have_personal_channel': {'val': True}}}, + }) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + urls = [c.args[0] for c in client.api.call_args_list] + assert 'users/' in urls + assert 'channels/get/' not in urls + assert groups['b'].metadata['channel'] == 'mscspeaker-Courses/2026' + assert not report.errors + + +def test_server_speaker_subchannel_without_email(): + # As for "mscspeaker", a personal sub-channel target without recipient is not importable. + groups = {'b': group_with('b', metadata={'channel': 'mscspeaker-Courses', + 'speaker_email': ''})} + client = make_client() + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + assert 'b' not in groups + assert summary.unimportable_media[mi.UNIMPORTABLE_NO_SPEAKER] == ['b'] + + +def test_server_channel_missing(): + groups = {'d': group_with('d', metadata={'channel': 'mscid-CID'})} + client = make_client(api_exc={'channels/get/': request_error()}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, groups, report, summary) + # An unresolvable channel is cleaned so the media falls back to the folder channel. + assert any('could not be resolved' in w for w in report.warnings) + assert groups['d'].metadata['channel'] == '' + assert not report.errors + + +def test_server_checks_channels(): + # The "channels.csv" entries are resolved against the catalog to report the channels that + # do not exist yet and the references already used by another channel. + existing = mi.ChannelUpdate(path=['Migration', 'Course A'], description='D') + missing = mi.ChannelUpdate(path=['Migration', 'Course B'], reference='lti:moodle:1') + taken = mi.ChannelUpdate(path=['Migration'], reference='lti:moodle:2') + client = make_client(catalog={'channels': [ + {'oid': 'c1', 'title': 'Migration', 'parent_oid': None}, + {'oid': 'c2', 'title': 'Course A', 'parent_oid': 'c1'}, + {'oid': 'c3', 'title': 'Other', 'parent_oid': None, 'external_ref': 'lti:moodle:2'}, + ]}) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {}, report, summary, channels=[existing, missing, taken]) + assert existing.object_id == 'c2' + assert missing.object_id is None + assert taken.object_id == 'c1' + assert not report.errors + warnings = '\n'.join(report.warnings) + assert 'Channel "Migration/Course B" does not exist yet' in warnings + assert 'Channel "Migration/Course A" does not exist yet' not in warnings + assert 'Reference "lti:moodle:2" (channel "Migration") is already used' in warnings + assert 'Reference "lti:moodle:1"' not in warnings + + +def test_server_checks_channels_catalog_error(): + # Without the catalog the channels cannot be resolved, but the audit still succeeds. + update = mi.ChannelUpdate(path=['Migration'], description='D') + client = make_client(catalog_exc=request_error()) + report = mi.Report() + summary = mi.Summary() + mi.server_checks(client, {}, report, summary, channels=[update]) + assert update.object_id is None + assert not report.errors + assert any('check slugs and channels' in w for w in report.warnings) + + +# -------- ensure_personal_channel + + +def test_ensure_personal_channel_perms_error(): + client = make_client(api_exc={'perms/get/': request_error()}) + report = mi.Report() + mi.ensure_personal_channel(client, '25508', 'x@x', report, apply=True) + assert any('Could not check permissions' in w for w in report.warnings) + + +def test_ensure_personal_channel_audit_reports(): + client = make_client(api_map={ + 'perms/get/': {'global_permissions': {'can_have_personal_channel': {'val': False}}}, + }) + report = mi.Report() + mi.ensure_personal_channel(client, '25508', 'x@x', report, apply=False) + # In audit mode the missing permission is only reported, not granted. + assert any('cannot own a personal channel' in w for w in report.warnings) + assert not any(c.args[0] == 'perms/edit/' for c in client.api.call_args_list) + + +def test_ensure_personal_channel_grant_error(): + client = make_client( + api_map={'perms/get/': {'global_permissions': {'can_have_personal_channel': {'val': False}}}}, + api_exc={'perms/edit/': request_error()}, + ) + report = mi.Report() + mi.ensure_personal_channel(client, '25508', 'x@x', report, apply=True) + assert any('Could not grant' in w for w in report.warnings) + + +# -------- mapping helpers + + +def test_build_media_metadata_full(): + row = { + 'title': 'T', 'slug': 's', 'description': 'd', 'language': 'fre', + 'creation': '2026-01-01T00:00:00', 'company_name': 'C', 'company_url': 'cu', + 'license_name': 'L', 'license_url': 'lu', 'validated': 'yes', 'unlisted': 'no', + 'detect_slides': 'no', 'keywords': 'a|b', 'categories': 'c1|c2', + 'speaker_name': 'N1|N2', 'speaker_email': 'e1|e2', + } + metadata = mi.build_media_metadata(row) + assert metadata == { + 'title': 'T', 'slug': 's', 'description': 'd', + 'language': 'fre', 'creation': '2026-01-01T00:00:00', 'company': 'C', + 'company_url': 'cu', 'license': 'L', 'license_url': 'lu', 'validated': 'yes', + 'unlisted': 'no', 'detect_slides': 'no', 'keywords': 'a,b', 'category': 'c1\nc2', + 'speaker': 'N1|N2', 'speaker_name': 'N1|N2', 'speaker_email': 'e1|e2', + } + + +def test_build_media_metadata_no_row(): + assert mi.build_media_metadata(None) == {} + + +@pytest.mark.parametrize('channel, expected', [ + ('mscpath-A/B', 'mscpath-A/B'), + ('mscpath-A/' + 'B' * 300, 'mscpath-A/' + 'B' * mi.MAX_FIELD_LENGTH), + ('mscspeaker-' + 'C' * 201 + '/D', 'mscspeaker-' + 'C' * mi.MAX_FIELD_LENGTH + '/D'), + # Targets without a path are left untouched. + ('mscspeaker', 'mscspeaker'), + ('mscid-' + 'x' * 300, 'mscid-' + 'x' * 300), + ('a-slug', 'a-slug'), +]) +def test_truncate_channel_titles(channel, expected): + assert mi.truncate_channel_titles(channel) == expected + + +@pytest.mark.parametrize('channel, expected', [ + ('mscspeaker', []), + ('mscspeaker-Top', ['Top']), + ('mscspeaker-Top channel/Mid channel/Sub channel', + ['Top channel', 'Mid channel', 'Sub channel']), + ('mscspeaker- Top / Sub /', ['Top', 'Sub']), # empty and padded titles are cleaned up + ('mscspeaker-', []), # an empty path targets the personal channel itself + ('mscspeakers', None), + ('mscpath-Top/Sub', None), + ('', None), +]) +def test_parse_speaker_path(channel, expected): + assert mi.parse_speaker_path(channel) == expected + + +@pytest.mark.parametrize('main, rel_dir, row, expected', [ + ('Migration', '.', {'channel': 'mscid-x'}, 'mscid-x'), + # A personal channel target is kept as is, it is resolved during the import. + ('Migration', '.', {'channel': 'mscspeaker-Sub'}, 'mscspeaker-Sub'), + ('Migration', 'sub/deep', {'channel': ''}, 'mscpath-Migration/sub/deep'), + ('mscpath-Root', 'a', None, 'mscpath-Root/a'), + ('Migration', '.', None, 'mscpath-Migration'), +]) +def test_resolve_channel(main, rel_dir, row, expected): + assert mi.resolve_channel(main, Path(rel_dir), row) == expected + + +def test_post_annotation_default_content(tmp_path): + # A non-internal annotation type with no content gets a placeholder content. + client = make_client() + row = {'type': 'custom', 'time': '1000', 'title': '', 'content': '', + 'keywords': '', 'attachment': ''} + mi.post_annotation(client, 'oid1', row, tmp_path) + _, kwargs = client.api.call_args + assert kwargs['data']['type_slug'] == 'custom' + assert kwargs['data']['content'] == '-' + + +# -------- import + + +def test_import_media(tmp_path): + write(tmp_path / 'm1.mp4') + sub_path = write(tmp_path / 'm1_fre.srt') + audio_path = write(tmp_path / 'm1_eng.mp3') + write(tmp_path / 'annotations' / 'doc.pdf') + + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.subtitles['fre'] = sub_path + g1.audio_tracks['eng'] = audio_path + g1.metadata = {'title': 'Title 1', 'channel': ''} + g1.annotations = [ + {'type': 'chapter', 'time': '1000', 'title': 'Intro', 'content': 'Hi', + 'keywords': 'a|b', 'attachment': ''}, + {'type': 'slide', 'time': '2000', 'title': 'Slide', 'content': '', + 'keywords': '', 'attachment': 'doc.pdf'}, + ] + groups = {'m1': g1} + + mapping_file = tmp_path / 'mapping.csv' + client = make_client() + summary = mi.Summary() + + mapping = mi.import_media( + client, groups, 'Migration', tmp_path, mapping_file, tmp_path / 'temp', summary + ) + + assert mapping == {'m1': 'v_migration:m1'} + client.add_media.assert_called_once() + _, kwargs = client.add_media.call_args + assert kwargs['title'] == 'Title 1' + assert kwargs['channel'] == 'mscpath-Migration' + assert kwargs['external_ref'] == 'migration:m1' + assert kwargs['origin'] == 'migration:m1' + called_urls = [call.args[0] for call in client.api.call_args_list] + assert 'subtitles/add/' in called_urls + assert 'medias/audio/tracks/add/' in called_urls + assert called_urls.count('annotations/post/') == 2 + assert mapping_file.read_text() == 'source_id,oid\nm1,v_migration:m1\n' + assert summary.medias_imported == 1 + assert summary.medias_existing == 0 + assert summary.metadata_applied == 1 + assert summary.audio_tracks_imported == 1 + assert summary.subtitles_imported == 1 + assert summary.annotations_imported == 2 + assert summary.annotations_existing == 0 + assert summary.annotations_failed == 0 + + +def test_import_media_slug_mismatch(tmp_path, caplog): + # The server may assign a different slug than requested; this is only warned about. + write(tmp_path / 'm1.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.metadata = {'title': 'T', 'slug': 'wanted', 'channel': ''} + client = make_client() + client.add_media = mock.MagicMock(side_effect=lambda **_kw: {'oid': 'v1', 'slug': 'other'}) + summary = mi.Summary() + with caplog.at_level(logging.WARNING): + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, tmp_path / 'map.csv', tmp_path / 'temp', summary + ) + assert mapping == {'m1': 'v1'} + assert 'did not receive the requested slug' in caplog.text + + +def test_import_media_already_exists(tmp_path): + # A media already present on the server (object_id set) is not re-uploaded, and its + # already-imported elements (listed in existing_elements) are skipped. + sub_path = write(tmp_path / 'm1_fre.srt') + audio_path = write(tmp_path / 'm1_eng.mp3') + + g1 = mi.MediaGroup('m1', 'vexist', tmp_path / 'm1.mp4', Path('.')) + g1.subtitles['fre'] = sub_path + g1.audio_tracks['eng'] = audio_path + g1.annotations = [{'type': 'chapter', 'time': '1000', 'title': 'Intro'}] + g1.existing_elements = ['audio:eng', 'subtitle:fre', 'annotation:chapter:1000'] + + mapping_file = tmp_path / 'mapping.csv' + client = make_client() + summary = mi.Summary() + + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, mapping_file, tmp_path / 'temp', summary + ) + + assert mapping == {'m1': 'vexist'} + client.add_media.assert_not_called() + called_urls = [call.args[0] for call in client.api.call_args_list] + assert 'subtitles/add/' not in called_urls + assert 'medias/audio/tracks/add/' not in called_urls + assert 'annotations/post/' not in called_urls + assert summary.medias_existing == 1 + assert summary.medias_imported == 0 + assert summary.metadata_applied == 0 + assert summary.audio_tracks_existing == 1 + assert summary.audio_tracks_imported == 0 + assert summary.subtitles_existing == 1 + assert summary.subtitles_imported == 0 + assert summary.annotations_existing == 1 + assert summary.annotations_imported == 0 + + +def test_import_media_without_metadata_row(tmp_path): + write(tmp_path / 'm1.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('sub')) + client = make_client() + summary = mi.Summary() + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, tmp_path / 'map.csv', tmp_path / 'temp', summary + ) + assert mapping == {'m1': 'v_migration:m1'} + _, kwargs = client.add_media.call_args + assert kwargs['title'] == 'm1' + assert kwargs['channel'] == 'mscpath-Migration/sub' + assert summary.medias_imported == 1 + assert summary.metadata_applied == 0 + + +def speaker_channel_client(created): + # A client resolving the personal channel of "known@x" and creating the missing levels. + def api(url, **kwargs): + if url == 'users/': + return {'users': [{'email': 'known@x', 'id': '25508'}]} + if url == 'channels/personal/': + return {'oid': 'cPerso'} + if url == 'channels/get/': + raise request_error(404) + if url == 'channels/add/': + created.append((kwargs['data']['title'], kwargs['data']['parent'])) + return {'oid': f'c{len(created)}'} + return {} + + client = make_client() + client.api = mock.MagicMock(side_effect=api) + return client + + +def test_import_media_speaker_subchannel(tmp_path): + # Media targeting a sub-channel of a personal channel are uploaded into the resolved oid, + # and the resolution is shared by every media of the same speaker. + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm2.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.metadata = {'title': 'T1', 'channel': 'mscspeaker-Courses/2026', + 'speaker_email': 'known@x'} + g2 = mi.MediaGroup('m2', None, tmp_path / 'm2.mp4', Path('.')) + g2.metadata = {'title': 'T2', 'channel': 'mscspeaker-Courses/2026', + 'speaker_email': 'known@x'} + created = [] + client = speaker_channel_client(created) + summary = mi.Summary() + + mapping = mi.import_media( + client, {'m1': g1, 'm2': g2}, 'Migration', tmp_path, tmp_path / 'map.csv', + tmp_path / 'temp', summary, + ) + + assert mapping == {'m1': 'v_migration:m1', 'm2': 'v_migration:m2'} + channels = [kwargs['channel'] for _, kwargs in client.add_media.call_args_list] + assert channels == ['mscid-c2', 'mscid-c2'] + # The channels of the path are created once, not once per media. + assert created == [('Courses', 'cPerso'), ('2026', 'c1')] + assert summary.medias_imported == 2 + assert summary.import_failures == 0 + + +def test_import_media_speaker_subchannel_error(tmp_path): + # A personal sub-channel that cannot be resolved makes its media an import failure only. + write(tmp_path / 'm1.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.metadata = {'title': 'T1', 'channel': 'mscspeaker-Courses', 'speaker_email': 'ghost@x'} + client = make_client(api_map={'users/': {'users': []}}) + summary = mi.Summary() + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, tmp_path / 'map.csv', tmp_path / 'temp', + summary, + ) + assert mapping == {} + client.add_media.assert_not_called() + assert summary.import_failures == 1 + + +def test_import_media_continues_on_upload_error(tmp_path): + # A media whose upload fails is reported but does not stop the following ones. + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm2.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g2 = mi.MediaGroup('m2', None, tmp_path / 'm2.mp4', Path('.')) + mapping_file = tmp_path / 'mapping.csv' + + client = make_client() + summary = mi.Summary() + + def add_media(**kwargs): + if kwargs['external_ref'] == 'migration:m1': + raise mi.NudgisRequestError('upload failed', status_code=500) + return {'oid': 'v_' + kwargs['external_ref'], 'slug': ''} + + client.add_media = mock.MagicMock(side_effect=add_media) + + mapping = mi.import_media( + client, {'m1': g1, 'm2': g2}, 'Migration', tmp_path, mapping_file, tmp_path / 'temp', summary + ) + + # The failed media is excluded from the mapping; the next one is still imported. + assert mapping == {'m2': 'v_migration:m2'} + assert client.add_media.call_count == 2 + assert mapping_file.read_text() == 'source_id,oid\nm2,v_migration:m2\n' + assert summary.import_failures == 1 + assert summary.medias_imported == 1 + + +def test_import_media_continues_on_linked_element_error(tmp_path): + # A linked-element failure does not prevent the media (and the next ones) from importing. + write(tmp_path / 'm1.mp4') + sub_path = write(tmp_path / 'm1_fre.srt') + write(tmp_path / 'm2.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.subtitles['fre'] = sub_path + g2 = mi.MediaGroup('m2', None, tmp_path / 'm2.mp4', Path('.')) + + client = make_client() + summary = mi.Summary() + + def api(url, **kwargs): + if url == 'subtitles/add/': + raise mi.NudgisRequestError('subtitle failed', status_code=500) + return {} + + client.api = mock.MagicMock(side_effect=api) + + mapping = mi.import_media( + client, {'m1': g1, 'm2': g2}, 'Migration', tmp_path, tmp_path / 'map.csv', + tmp_path / 'temp', summary, + ) + + # m1's upload succeeded, so it is counted as imported even though its subtitle failed. + assert set(mapping) == {'m1', 'm2'} + assert client.add_media.call_count == 2 + assert summary.import_failures == 0 + assert summary.medias_imported == 2 + assert summary.subtitles_failed == 1 + assert summary.subtitles_imported == 0 + + +def test_import_media_element_failures(tmp_path): + # Each linked element failure is isolated and counted; the media is still imported. + write(tmp_path / 'm1.mp4') + sub_path = write(tmp_path / 'm1_fre.srt') + audio_path = write(tmp_path / 'm1_eng.mp3') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.subtitles['fre'] = sub_path + g1.audio_tracks['eng'] = audio_path + g1.annotations = [{'type': 'chapter', 'time': '1000', 'title': 'Intro'}] + + client = make_client() + summary = mi.Summary() + + def api(url, **kwargs): + if url in ('medias/audio/tracks/add/', 'subtitles/add/', 'annotations/post/'): + raise mi.NudgisRequestError('boom', status_code=500) + return {} + + client.api = mock.MagicMock(side_effect=api) + + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, tmp_path / 'map.csv', tmp_path / 'temp', summary + ) + + # The media itself was created, so it is imported despite every element failing. + assert mapping == {'m1': 'v_migration:m1'} + assert summary.medias_imported == 1 + assert summary.import_failures == 0 + assert summary.audio_tracks_failed == 1 + assert summary.subtitles_failed == 1 + assert summary.annotations_failed == 1 + + +def test_import_media_skips_malformed_annotation(tmp_path): + # A row missing its type/time is defensively skipped and not counted. + write(tmp_path / 'm1.mp4') + g1 = mi.MediaGroup('m1', None, tmp_path / 'm1.mp4', Path('.')) + g1.annotations = [{'title': 'no type nor time'}] + client = make_client() + summary = mi.Summary() + mapping = mi.import_media( + client, {'m1': g1}, 'Migration', tmp_path, tmp_path / 'map.csv', tmp_path / 'temp', summary + ) + assert mapping == {'m1': 'v_migration:m1'} + called_urls = [call.args[0] for call in client.api.call_args_list] + assert 'annotations/post/' not in called_urls + assert summary.annotations_imported == 0 + assert summary.annotations_failed == 0 + + +def test_import_media_multistream(tmp_path): + # A multi-stream media is combined into a temporary file, uploaded, then cleaned up. + main = write(tmp_path / 'multi.mp4') + second = write(tmp_path / 'multi_2.mp4') + temp_dir = tmp_path / 'temp' + g = mi.MediaGroup('multi', None, main, Path('.')) + g.extra_streams[2] = second + + client = make_client() + summary = mi.Summary() + captured = {} + + def fake_compose(inputs, output, ffmpeg='ffmpeg', ffprobe='ffprobe'): + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(b'composed') + # compose_streams writes the layout preset next to the output (see compose_multistream). + output.with_suffix('.json').write_text('{"composition_area": {"w": 1280, "h": 640}}') + captured['call'] = (inputs, output, ffmpeg, ffprobe) + + with mock.patch.object(mi, 'compose_streams', side_effect=fake_compose): + mapping = mi.import_media( + client, {'multi': g}, 'Migration', tmp_path, tmp_path / 'map.csv', temp_dir, summary, + ffmpeg='/usr/bin/ffmpeg', ffprobe='/usr/bin/ffprobe', + ) + + temp_file = temp_dir / 'multi.mp4' + assert captured['call'] == ([main, second], temp_file, '/usr/bin/ffmpeg', '/usr/bin/ffprobe') + _, kwargs = client.add_media.call_args + assert kwargs['file_path'] == temp_file + # The layout preset is forwarded to the server as the JSON content written by compose_streams. + assert kwargs['layout_preset'] == '{"composition_area": {"w": 1280, "h": 640}}' + assert not temp_file.exists() # the temporary file is removed after a successful upload + assert not temp_file.with_suffix('.json').exists() # the layout preset is cleaned up too + assert mapping == {'multi': 'v_migration:multi'} + assert summary.merged_videos == 1 + assert summary.medias_imported == 1 + + +def test_import_media_multistream_skipped_when_exists(tmp_path): + # An already imported multi-stream media is neither combined nor re-uploaded. + g = mi.MediaGroup('multi', 'vexist', tmp_path / 'multi.mp4', Path('.')) + g.extra_streams[2] = tmp_path / 'multi_2.mp4' + client = make_client() + summary = mi.Summary() + with mock.patch.object(mi, 'compose_streams') as compose: + mapping = mi.import_media( + client, {'multi': g}, 'Migration', tmp_path, tmp_path / 'map.csv', + tmp_path / 'temp', summary, + ) + compose.assert_not_called() + client.add_media.assert_not_called() + assert mapping == {'multi': 'vexist'} + assert summary.medias_existing == 1 + assert summary.merged_videos == 0 + + +def test_import_media_multistream_compose_error(tmp_path): + # A composition failure is reported and does not stop the process. + main = write(tmp_path / 'multi.mp4') + g = mi.MediaGroup('multi', None, main, Path('.')) + g.extra_streams[2] = write(tmp_path / 'multi_2.mp4') + client = make_client() + summary = mi.Summary() + with mock.patch.object(mi, 'compose_streams', side_effect=RuntimeError('ffmpeg failed')): + mapping = mi.import_media( + client, {'multi': g}, 'Migration', tmp_path, tmp_path / 'map.csv', + tmp_path / 'temp', summary, + ) + assert mapping == {} + client.add_media.assert_not_called() + assert summary.import_failures == 1 + assert summary.merged_videos == 0 + + +# -------- channels import + + +def test_build_channel_paths(): + catalog = {'channels': [ + {'oid': 'c1', 'title': 'Migration', 'parent_oid': None}, + {'oid': 'c2', 'title': 'Course A', 'parent_oid': 'c1'}, + {'oid': 'c3', 'title': 'Year 1', 'parent_oid': 'c2'}, + {'oid': 'c4', 'title': 'Course A', 'parent_oid': None}, # same title at the root + {'title': 'No oid', 'parent_oid': None}, # ignored + ]} + assert mi.build_channel_paths(catalog) == { + 'Migration': 'c1', + 'Migration/Course A': 'c2', + 'Migration/Course A/Year 1': 'c3', + 'Course A': 'c4', + } + + +def test_build_channel_paths_empty(): + assert mi.build_channel_paths({}) == {} + + +def test_build_channel_paths_loop(): + # A channel that is its own ancestor must not loop forever. + catalog = {'channels': [{'oid': 'c1', 'title': 'Loop', 'parent_oid': 'c1'}]} + assert mi.build_channel_paths(catalog) == {'Loop': 'c1'} + + +def test_ensure_channel_existing(): + client = make_client() + paths = {'Migration': 'c1', 'Migration/Course A': 'c2'} + assert mi.ensure_channel(client, ['Migration', 'Course A'], paths) == 'c2' + client.api.assert_not_called() + + +def test_ensure_channel_creates_root(): + client = make_client(api_map={'channels/add/': {'oid': 'cRoot'}}) + paths = {} + assert mi.ensure_channel(client, ['Root'], paths) == 'cRoot' + assert paths == {'Root': 'cRoot'} + _, kwargs = client.api.call_args + # A channel created at the root of the catalog has no parent. + assert kwargs['data'] == {'title': 'Root'} + + +def test_get_or_create_channel_existing(): + client = make_client(api_map={'channels/get/': {'info': {'oid': 'cSub'}}}) + assert mi.get_or_create_channel(client, 'Sub', 'cParent') == 'cSub' + urls = [c.args[0] for c in client.api.call_args_list] + assert 'channels/add/' not in urls + + +def test_get_or_create_channel_created(): + client = make_client( + api_exc={'channels/get/': request_error(404)}, + api_map={'channels/add/': {'oid': 'cSub'}}, + ) + assert mi.get_or_create_channel(client, 'Sub', 'cParent') == 'cSub' + _, kwargs = client.api.call_args + assert kwargs['data'] == {'title': 'Sub', 'parent': 'cParent'} + + +def test_get_or_create_channel_error(): + # Any error other than a 404 is propagated (the media is counted as an import failure). + client = make_client(api_exc={'channels/get/': request_error()}) + with pytest.raises(mi.NudgisRequestError): + mi.get_or_create_channel(client, 'Sub', 'cParent') + + +def test_get_or_create_channel_unexpected_response(): + client = make_client(api_map={'channels/get/': {}}) + with pytest.raises(RuntimeError): + mi.get_or_create_channel(client, 'Sub', 'cParent') + + +def test_ensure_speaker_channel(): + created = [] + client = speaker_channel_client(created) + cache = {} + row = {'speaker_email': 'known@x|other@x'} # only the first speaker is used + assert mi.ensure_speaker_channel(client, row, ['Courses', '2026'], cache) == 'mscid-c2' + # Each level is created below the previous one, starting at the personal channel. + assert created == [('Courses', 'cPerso'), ('2026', 'c1')] + assert cache == { + 'known@x': 'cPerso', 'known@x/Courses': 'c1', 'known@x/Courses/2026': 'c2', + } + # A second media targeting the same channel is resolved from the cache, without any call. + client.api.reset_mock() + assert mi.ensure_speaker_channel(client, row, ['Courses', '2026'], cache) == 'mscid-c2' + client.api.assert_not_called() + # A sibling channel of the same speaker only creates the missing level. + assert mi.ensure_speaker_channel(client, row, ['Courses', '2025'], cache) == 'mscid-c3' + assert created[-1] == ('2025', 'c1') + assert [c.args[0] for c in client.api.call_args_list] == ['channels/get/', 'channels/add/'] + + +def test_ensure_speaker_channel_personal_channel_only(): + # An empty path targets the personal channel itself (no sub-channel is created). + created = [] + client = speaker_channel_client(created) + assert mi.ensure_speaker_channel(client, {'speaker_email': 'known@x'}, [], {}) == 'mscid-cPerso' + assert not created + + +def test_ensure_speaker_channel_without_email(): + client = make_client() + with pytest.raises(RuntimeError): + mi.ensure_speaker_channel(client, {'speaker_email': ''}, ['Courses'], {}) + with pytest.raises(RuntimeError): + mi.ensure_speaker_channel(client, None, ['Courses'], {}) + client.api.assert_not_called() + + +def test_ensure_speaker_channel_unknown_speaker(): + client = make_client(api_map={'users/': {'users': []}}) + with pytest.raises(RuntimeError): + mi.ensure_speaker_channel(client, {'speaker_email': 'ghost@x'}, ['Courses'], {}) + + +def test_apply_channels_metadata_existing_channel(): + update = mi.ChannelUpdate( + path=['Migration', 'Course A'], + description='

My description

', + reference='lti:moodle.example.local:245', + ) + client = make_client(catalog={'channels': [ + {'oid': 'c1', 'title': 'Migration', 'parent_oid': None}, + {'oid': 'c2', 'title': 'Course A', 'parent_oid': 'c1'}, + ]}) + summary = mi.Summary() + mi.apply_channels_metadata(client, [update], summary) + called_urls = [call.args[0] for call in client.api.call_args_list] + assert 'channels/add/' not in called_urls + _, kwargs = client.api.call_args + assert kwargs['data'] == { + 'oid': 'c2', + 'description': '

My description

', + 'external_ref': 'lti:moodle.example.local:245', + } + assert update.object_id == 'c2' + assert summary.channels_updated == 1 + assert summary.channels_failed == 0 + + +def test_apply_channels_metadata_creates_missing_channels(): + update = mi.ChannelUpdate(path=['Migration', 'Course A', 'Year 1'], description='D') + created: list[dict] = [] + + def api(url, **kwargs): + if url == 'channels/add/': + created.append(kwargs['data']) + return {'oid': f'c{len(created) + 1}'} + return {} + + client = make_client( + catalog={'channels': [{'oid': 'c1', 'title': 'Migration', 'parent_oid': None}]}, + ) + client.api = mock.MagicMock(side_effect=api) + summary = mi.Summary() + mi.apply_channels_metadata(client, [update], summary) + # Only the missing levels are created, each one below the previous one. + assert created == [ + {'title': 'Course A', 'parent': 'c1'}, + {'title': 'Year 1', 'parent': 'c2'}, + ] + edits = [c for c in client.api.call_args_list if c.args[0] == 'channels/edit/'] + assert [c.kwargs['data'] for c in edits] == [{'oid': 'c3', 'description': 'D'}] + assert summary.channels_updated == 1 + + +def test_apply_channels_metadata_no_channel(): + client = make_client() + summary = mi.Summary() + mi.apply_channels_metadata(client, [], summary) + client.get_catalog.assert_not_called() + client.api.assert_not_called() + assert summary.channels_updated == 0 + + +def test_apply_channels_metadata_catalog_error(): + # Without the catalog no channel can be resolved, so all of them are counted as failed. + client = make_client(catalog_exc=request_error()) + summary = mi.Summary() + mi.apply_channels_metadata( + client, + [mi.ChannelUpdate(path=['A'], description='D'), + mi.ChannelUpdate(path=['B'], description='D')], + summary, + ) + client.api.assert_not_called() + assert summary.channels_updated == 0 + assert summary.channels_failed == 2 + + +def test_apply_channels_metadata_continues_on_error(): + # A channel that cannot be updated does not prevent the next ones from being updated. + first = mi.ChannelUpdate(path=['A'], description='D') + second = mi.ChannelUpdate(path=['B'], description='D') + client = make_client(catalog={'channels': [ + {'oid': 'c1', 'title': 'A', 'parent_oid': None}, + {'oid': 'c2', 'title': 'B', 'parent_oid': None}, + ]}) + + def api(url, **kwargs): + if url == 'channels/edit/' and kwargs['data']['oid'] == 'c1': + raise mi.NudgisRequestError('boom', status_code=500) + return {} + + client.api = mock.MagicMock(side_effect=api) + summary = mi.Summary() + mi.apply_channels_metadata(client, [first, second], summary) + assert first.object_id is None + assert second.object_id == 'c2' + assert summary.channels_updated == 1 + assert summary.channels_failed == 1 + + +# -------- planning / report + + +def test_count_planned(): + groups = { + 'new': group_with('new', metadata={'title': 'T'}, + annotations=[{'type': 'chapter', 'time': '0'}]), + 'existing': group_with( + 'existing', object_id='vexist', + annotations=[{'type': 'chapter', 'time': '1000'}], + existing_elements=['audio:eng', 'subtitle:fre', 'annotation:chapter:1000'], + ), + } + groups['new'].extra_streams[2] = Path('new_2.mp4') + groups['new'].audio_tracks['eng'] = Path('new_eng.mp3') + groups['new'].subtitles['fre'] = Path('new_fre.srt') + groups['existing'].audio_tracks['eng'] = Path('existing_eng.mp3') + groups['existing'].subtitles['fre'] = Path('existing_fre.srt') + summary = mi.Summary() + mi.count_planned(groups, summary) + assert summary.medias_imported == 1 + assert summary.medias_existing == 1 + assert summary.merged_videos == 1 + assert summary.metadata_applied == 1 + assert summary.audio_tracks_imported == 1 + assert summary.audio_tracks_existing == 1 + assert summary.subtitles_imported == 1 + assert summary.subtitles_existing == 1 + assert summary.annotations_imported == 1 + assert summary.annotations_existing == 1 + + +def test_count_planned_channels(): + # Every valid "channels.csv" entry would be applied by an "--apply" run. + summary = mi.Summary() + mi.count_planned({}, summary, channels=[ + mi.ChannelUpdate(path=['A'], description='D'), + mi.ChannelUpdate(path=['B'], reference='lti:moodle:1'), + ]) + assert summary.channels_updated == 2 + + +def test_print_summary_smoke(): + summary = mi.Summary( + medias_imported=2, import_failures=1, channels_updated=1, channels_failed=1, + invalid_channels=1, + ) + summary.drop_media('x', mi.UNIMPORTABLE_CORRUPTED) + # Should not raise in either mode. + mi.print_summary(summary, applied=True) + mi.print_summary(summary, applied=False) + + +# -------- orchestrator + + +def build_valid_tree(tmp_path): + write(tmp_path / 'm1.mp4') + write(tmp_path / 'm1_fre.srt') + write(tmp_path / 'm1_eng.mp3') + write(tmp_path / 'sub' / 'm2.mp4') + write(tmp_path / 'readme.txt') # produces a warning + write(tmp_path / 'annotations' / 'doc.pdf') + write( + tmp_path / 'metadata.csv', + ( + METADATA_HEADER + '\n' + 'm1,"Title 1",slug-1,,,,,2026-01-01T00:00:00,,,,,,,,yes,no,no\n' + 'm2,"Title 2",slug-2,,,,,,,,,,,,,,,\n' + ).encode('utf-8'), + ) + write( + tmp_path / 'annotations' / 'annotations.csv', + b'source_id,type,time,title,content,keywords,attachment\n' + b'm1,chapter,1000,Intro,Hi,,\n' + b'm1,slide,2000,Slide,,,doc.pdf\n' + b'm2,chapter,0,Start,,,\n', + ) + write( + tmp_path / 'channels.csv', + ( + CHANNELS_HEADER + '\n' + 'Migration,"

Root channel

",\n' + 'Migration/sub,,lti:moodle.example.local:245\n' + ).encode('utf-8'), + ) + + +@pytest.fixture() +def patched_client(): + client = make_client() + with mock.patch('examples.mass_import.NudgisClient', return_value=client), \ + mock.patch('examples.mass_import.probe_media', return_value=None): + yield client + + +def test_mass_import_missing_source_dir(tmp_path, patched_client): + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path / 'nope'), + '--channel', 'Migration', + ]) == 1 + + +@pytest.mark.parametrize('channel, expected', [ + ('Migration', 0), + ('mscpath-Migration/sub', 0), + ('c123456789012345678', 0), # only 19 chars, too short to be an object id + ('c-1234567890123456789', 0), # right length but not alphanumeric + ('mscid-c1234567890123456789', 1), + ('c1234567890123456789', 1), + ('cAbCdEfGhIjKlMnOpQrS', 1), +]) +def test_mass_import_channel_object_id(tmp_path, patched_client, channel, expected): + # The main channel must be a title or an "mscpath-...", an object id is rejected. + build_valid_tree(tmp_path) + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path), '--channel', channel, + ]) == expected + + +def test_mass_import_audit_failure(tmp_path, patched_client): + # A structurally broken metadata.csv (missing mandatory columns) aborts the run. + write(tmp_path / 'm1.mp4') + write(tmp_path / 'metadata.csv', b'foo,bar\n1,2\n') + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path), '--channel', 'Migration', + ]) == 1 + + +def test_mass_import_dry_run(tmp_path, patched_client): + build_valid_tree(tmp_path) + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path), '--channel', 'Migration', + '--log-level', 'debug', + ]) == 0 + patched_client.add_media.assert_not_called() + called_urls = [call.args[0] for call in patched_client.api.call_args_list] + assert 'channels/edit/' not in called_urls + assert 'channels/add/' not in called_urls + + +def test_mass_import_apply(tmp_path, patched_client): + build_valid_tree(tmp_path) + mapping_file = tmp_path / 'mapping.csv' + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path), '--channel', 'Migration', + '--mapping-file', str(mapping_file), '--apply', + ]) == 0 + assert patched_client.add_media.call_count == 2 + content = mapping_file.read_text() + assert 'm1,v_migration:m1' in content + assert 'm2,v_migration:m2' in content + # The channel metadata are applied once every media has been imported. + edits = [c for c in patched_client.api.call_args_list if c.args[0] == 'channels/edit/'] + assert [c.kwargs['data'].get('description') for c in edits] == ['

Root channel

', None] + assert edits[1].kwargs['data']['external_ref'] == 'lti:moodle.example.local:245' + add_media_index = min( + index for index, call in enumerate(patched_client.api.call_args_list) + if call.args[0] == 'subtitles/add/' + ) + edit_index = min( + index for index, call in enumerate(patched_client.api.call_args_list) + if call.args[0] == 'channels/edit/' + ) + assert add_media_index < edit_index + + +def test_mass_import_ids_to_process(tmp_path, patched_client): + build_valid_tree(tmp_path) + mapping_file = tmp_path / 'mapping.csv' + assert mi.mass_import([ + '--conf', 'conf.json', '--source-dir', str(tmp_path), '--channel', 'Migration', + '--mapping-file', str(mapping_file), '--apply', '--ids-to-process', 'm1', + ]) == 0 + assert patched_client.add_media.call_count == 1 + content = mapping_file.read_text() + assert 'm1,v_migration:m1' in content + assert 'm2' not in content diff --git a/tests/samples/ball_1920x1080_h264_aac.mp4 b/tests/samples/ball_1920x1080_h264_aac.mp4 new file mode 100644 index 0000000..ac88a21 Binary files /dev/null and b/tests/samples/ball_1920x1080_h264_aac.mp4 differ diff --git a/tests/samples/ball_640x480_h265.mp4 b/tests/samples/ball_640x480_h265.mp4 new file mode 100644 index 0000000..a43cb9c Binary files /dev/null and b/tests/samples/ball_640x480_h265.mp4 differ diff --git a/tests/samples/ball_720x540_av1_opus.webm b/tests/samples/ball_720x540_av1_opus.webm new file mode 100644 index 0000000..d272bee Binary files /dev/null and b/tests/samples/ball_720x540_av1_opus.webm differ diff --git a/tests/samples/mire_1280x720_h264_aac.mp4 b/tests/samples/mire_1280x720_h264_aac.mp4 new file mode 100644 index 0000000..72199cc Binary files /dev/null and b/tests/samples/mire_1280x720_h264_aac.mp4 differ