From 8ea6ce36b36951a77a5e52dd54834a86dce10843 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Wed, 13 Aug 2025 11:31:42 +0200 Subject: [PATCH 1/3] feature: add chapters and task for chapter frames --- cds/modules/deposit/api.py | 78 ++++- cds/modules/deposit/receivers.py | 2 + cds/modules/flows/api.py | 2 + cds/modules/flows/tasks.py | 264 ++++++++++++++++- .../cds_previewer/macros/player.html | 166 ++++++++--- .../records/serializers/schemas/video.py | 15 +- .../templates/cds_records/video/detail.html | 199 +++++++++---- cds/modules/records/utils.py | 74 +++++ .../bootstrap3/js/cds_records/cdsRecord.js | 247 ++++++++++++++-- .../theme/assets/bootstrap3/scss/cds/cds.scss | 269 ++++++++++++++++++ tests/unit/test_flows_tasks.py | 88 ++++++ 11 files changed, 1284 insertions(+), 120 deletions(-) diff --git a/cds/modules/deposit/api.py b/cds/modules/deposit/api.py index 5cc9bed37..b7ae89048 100644 --- a/cds/modules/deposit/api.py +++ b/cds/modules/deposit/api.py @@ -65,6 +65,7 @@ get_tasks_status_grouped_by_task_name, merge_tasks_status, ) +from ..flows.tasks import ExtractChapterFramesTask from ..flows.models import FlowMetadata from ..invenio_deposit.api import Deposit, has_status, preserve from ..invenio_deposit.utils import mark_as_action @@ -76,7 +77,7 @@ ) from ..records.minters import cds_doi_generator, is_local_doi, report_number_minter from ..records.resolver import record_resolver -from ..records.utils import is_record, lowercase_value +from ..records.utils import is_record, lowercase_value, parse_video_chapters from ..records.validators import PartialDraft4Validator from ..records.permissions import is_public from .errors import DiscardConflict @@ -504,7 +505,7 @@ def create(cls, data, id_=None, **kwargs): data.setdefault("_access", {}) access_update = data["_access"].setdefault("update", []) try: - if current_user.email not in access_update: + if current_user.email not in access_update: # Add the current user to the ``_access.update`` list access_update.append(current_user.email) except AttributeError: @@ -905,11 +906,74 @@ def _publish_edited(self): return super(Video, self)._publish_edited() + def _has_chapters_changed(self, old_record=None): + """Check if chapters in description have changed.""" + current_description = self.get("description", "") + current_chapters = parse_video_chapters(current_description) + + if old_record is None: + # First publish - trigger if chapters exist + return len(current_chapters) > 0 + + old_description = old_record.get("description", "") + old_chapters = parse_video_chapters(old_description) + + # Compare chapter timestamps and titles + if len(current_chapters) != len(old_chapters): + return True + + for curr, old in zip(current_chapters, old_chapters): + if curr["seconds"] != old["seconds"] or curr["title"] != old["title"]: + return True + + return False + + def _trigger_chapter_frame_extraction(self): + """Trigger chapter frame extraction asynchronously for existing video files.""" + try: + # Get the current flow for this deposit + current_flow = FlowMetadata.get_by_deposit(self["_deposit"]["id"]) + + if current_flow is None: + current_app.logger.warning( + f"No current flow found for video {self.id}. Cannot trigger chapter frame extraction." + ) + return + + current_app.logger.info( + f"Triggering asynchronous ExtractChapterFramesTask for video {self.id} with flow {current_flow.id}" + ) + + payload = current_flow.payload.copy() + + current_app.logger.info(f"Submitting ExtractChapterFramesTask with payload: {payload}") + + ExtractChapterFramesTask().s(**payload).apply_async() + + current_app.logger.info( + f"ExtractChapterFramesTask submitted asynchronously for video {self.id}, flow_id: {current_flow.id}" + ) + except Exception as e: + current_app.logger.error( + f"Failed to trigger async chapter frame extraction for video {self.id}: {e}" + ) + import traceback + + current_app.logger.error(f"Traceback: {traceback.format_exc()}") + @mark_as_action def publish(self, pid=None, id_=None, **kwargs): """Publish a video and update the related project.""" # save a copy of the old PID video_old_id = self["_deposit"]["id"] + + # Check if this is a republish and get the old record + old_record = None + try: + _, old_record = self.fetch_published() + except KeyError as e: # First publish (no pid key) + pass + try: self["category"] = self.project["category"] self["type"] = self.project["type"] @@ -930,6 +994,13 @@ def publish(self, pid=None, id_=None, **kwargs): video_published = super(Video, self).publish(pid=pid, id_=id_, **kwargs) _, record_new = self.fetch_published() + # Check if chapters have changed and trigger frame extraction + if self._has_chapters_changed(old_record): + current_app.logger.info( + f"Chapters changed for video {self.id}, triggering frame extraction" + ) + self._trigger_chapter_frame_extraction() + # update associated project video_published.project._update_videos( [video_build_url(video_old_id)], @@ -1088,7 +1159,6 @@ def _create_tags(self): except IndexError: return - def mint_doi(self): """Mint DOI.""" assert self.has_record() @@ -1109,7 +1179,7 @@ def mint_doi(self): status=PIDStatus.RESERVED, ) return self - + project_resolver = Resolver( pid_type="depid", diff --git a/cds/modules/deposit/receivers.py b/cds/modules/deposit/receivers.py index 18727fec7..d5949a39d 100644 --- a/cds/modules/deposit/receivers.py +++ b/cds/modules/deposit/receivers.py @@ -33,6 +33,7 @@ from cds.modules.flows.tasks import ( DownloadTask, ExtractFramesTask, + ExtractChapterFramesTask, ExtractMetadataTask, TranscodeVideoTask, ) @@ -87,4 +88,5 @@ def register_celery_class_based_tasks(sender, app=None): celery.register_task(ExtractMetadataTask()) celery.register_task(DownloadTask()) celery.register_task(ExtractFramesTask()) + celery.register_task(ExtractChapterFramesTask()) celery.register_task(TranscodeVideoTask()) diff --git a/cds/modules/flows/api.py b/cds/modules/flows/api.py index 136bef012..09acbaadf 100644 --- a/cds/modules/flows/api.py +++ b/cds/modules/flows/api.py @@ -39,6 +39,7 @@ from .tasks import ( CeleryTask, DownloadTask, + ExtractChapterFramesTask, ExtractFramesTask, ExtractMetadataTask, TranscodeVideoTask, @@ -245,6 +246,7 @@ def _find_celery_task_by_name(name): ExtractMetadataTask, ExtractFramesTask, TranscodeVideoTask, + ExtractChapterFramesTask, ]: if celery_task.name == name: return celery_task diff --git a/cds/modules/flows/tasks.py b/cds/modules/flows/tasks.py index eaa97952a..8dc0068d5 100644 --- a/cds/modules/flows/tasks.py +++ b/cds/modules/flows/tasks.py @@ -62,7 +62,7 @@ from ..opencast.api import OpenCast from ..opencast.error import RequestError from ..opencast.utils import get_qualities -from ..records.utils import to_string +from ..records.utils import to_string, parse_video_chapters from ..xrootd.utils import file_opener_xrootd from .deposit import index_deposit_project from .files import dispose_object_version, move_file_into_local @@ -197,7 +197,9 @@ def _meta_exception_envelope(self, exc): NOTE: workaround to be able to save the payload in celery in case of exceptions. """ - meta = dict(message=str(exc), payload=self._base_payload) + # Safety check in case base payload is not set yet + payload = getattr(self, '_base_payload', {}) + meta = dict(message=str(exc), payload=payload) return dict(exc_message=meta, exc_type=exc.__class__.__name__) def on_failure(self, exc, task_id, args, kwargs, einfo): @@ -223,7 +225,16 @@ def on_success(self, exc, task_id, args, kwargs): def _reindex_video_project(self): """Reindex video and project.""" with celery_app.flask_app.app_context(): - deposit_id = self._base_payload["deposit_id"] + # Safety check in case base payload is not set yet + if not hasattr(self, '_base_payload') or not self._base_payload or 'deposit_id' not in self._base_payload: + if hasattr(self, 'deposit_id') and self.deposit_id: + deposit_id = self.deposit_id + else: + self.log("Cannot reindex: deposit_id not available") + return + else: + deposit_id = self._base_payload["deposit_id"] + try: index_deposit_project(deposit_id) except PIDDeletedError: @@ -590,10 +601,10 @@ def progress_updater(current_frame): object_=self.object_version, output_dir=output_folder, progress_updater=progress_updater, - **options + **options, ), object_=self.object_version, - **options + **options, ) except Exception: db.session.rollback() @@ -601,6 +612,8 @@ def progress_updater(current_frame): self.clean(version_id=self.object_version_id) raise + total_frames = len(frames) + # Generate GIF images self._create_gif( bucket=str(self.object_version.bucket.id), @@ -618,7 +631,7 @@ def progress_updater(current_frame): db.session.commit() self.log("Finished task {0}".format(kwargs["task_id"])) - return "Created {0} frames.".format(len(frames)) + return "Created {0} frames.".format(total_frames) @classmethod def _time_position(cls, duration, frames_start=5, frames_end=95, frames_gap=10): @@ -648,7 +661,7 @@ def _create_tmp_frames( duration, output_dir, progress_updater=None, - **kwargs + **kwargs, ): """Create frames in temporary files.""" # Generate frames @@ -727,6 +740,239 @@ def _create_object( [ObjectVersionTag.create(obj, k, to_string(tags[k])) for k in tags] +class ExtractChapterFramesTask(AVCTask): + """Extract chapter frames task - dedicated task for chapter frame extraction only.""" + + name = "file_video_extract_chapter_frames" + + @staticmethod + def clean(version_id, *args, **kwargs): + """Delete generated chapter frame ObjectVersion slaves.""" + # remove all objects version "slave" with type "frame" that are chapter frames + tag_alias_1 = aliased(ObjectVersionTag) + tag_alias_2 = aliased(ObjectVersionTag) + tag_alias_3 = aliased(ObjectVersionTag) + + slaves = ( + ObjectVersion.query.join(tag_alias_1, ObjectVersion.tags) + .join(tag_alias_2, ObjectVersion.tags) + .join(tag_alias_3, ObjectVersion.tags) + .filter(tag_alias_1.key == "master", tag_alias_1.value == version_id) + .filter(tag_alias_2.key == "context_type", tag_alias_2.value == "frame") + .filter(tag_alias_3.key == "is_chapter_frame", tag_alias_3.value == "true") + .all() + ) + + for slave in slaves: + dispose_object_version(slave) + + def run(self, *args, **kwargs): + """Extract frames only at chapter timestamps from video description. + + This task is specifically designed to extract frames for chapters only, + without affecting other frame extraction processes. + + The task receives parameters through the standard AVCTask initialization: + - self.deposit_id: The deposit ID containing the video description + - self.object_version: The ObjectVersion of the master video file + - self.flow_id: The current flow ID for task metadata integration + """ + + # Create or update the TaskMetadata + flow_task_metadata = self.get_or_create_flow_task() + kwargs["celery_task_id"] = str(self.request.id) + kwargs["task_id"] = str(flow_task_metadata.id) + flow_task_metadata.payload = self.get_full_payload(**kwargs) + flow_task_metadata.status = FlowTaskStatus.STARTED + flow_task_metadata.message = "" + db.session.commit() + + self.log("Started task {0}".format(kwargs["task_id"])) + + output_folder = tempfile.mkdtemp() + + bucket_was_locked = False + if self.object_version.bucket.locked: + # If record was published we need to unlock the bucket + bucket_was_locked = True + self.object_version.bucket.locked = False + + try: + # Get the deposit to access the description + from cds.modules.deposit.api import deposit_video_resolver + db.session.refresh(self.object_version) + deposit_video = deposit_video_resolver(self.deposit_id) + description = deposit_video.get("description", "") + + self.log("Found description with {0} characters".format(len(description))) + + # Parse chapters from description + chapters = parse_video_chapters(description) + + if not chapters: + self.log("No chapters found in description - task completed") + return {"chapter_frames_extracted": 0, "status": "no_chapters"} + + self.log("Found {0} chapters in description".format(len(chapters))) + + # Get video duration from metadata + duration = float(self._base_payload.get("tags", {}).get("duration", 0)) + + if duration == 0: + raise ValueError("Video duration is 0 - cannot extract frames") + + # Check which timestamps already have frames + existing_timestamps = self._get_existing_frame_timestamps() + + def progress_updater(current_chapter): + """Progress reporter.""" + percentage = current_chapter / len(chapters) * 100 + meta = dict( + payload=dict(size=len(chapters), percentage=percentage), + message="Extracting chapter frames [{0} out of {1}]".format( + current_chapter, len(chapters) + ), + ) + self.log(meta["message"]) + + frames = self._create_chapter_frames( + chapters=chapters, + duration=duration, + object_=self.object_version, + output_dir=output_folder, + existing_timestamps=existing_timestamps, + progress_updater=progress_updater, + ) + + # Sync deposit and record files + sync_records_with_deposit_files(self.deposit_id) + + except Exception: + db.session.rollback() + shutil.rmtree(output_folder, ignore_errors=True) + self.clean(version_id=self.object_version_id) + raise + + total_frames = len(frames) + + if bucket_was_locked: + # Lock the bucket again + self.object_version.bucket.locked = True + + # Cleanup + shutil.rmtree(output_folder) + + self.log("Finished task {0}".format(kwargs["task_id"])) + return "Created {0} chapter frames.".format(total_frames) + + def _get_existing_frame_timestamps(self): + """Get set of existing frame timestamps to avoid duplicates.""" + tag_alias_1 = aliased(ObjectVersionTag) + tag_alias_2 = aliased(ObjectVersionTag) + tag_alias_3 = aliased(ObjectVersionTag) + + existing = ( + ObjectVersion.query.join(tag_alias_1, ObjectVersion.tags) + .join(tag_alias_2, ObjectVersion.tags) + .join(tag_alias_3, ObjectVersion.tags) + .filter(tag_alias_1.key == "master", tag_alias_1.value == self.object_version_id) + .filter(tag_alias_2.key == "context_type", tag_alias_2.value == "frame") + .filter(tag_alias_3.key == "timestamp") + .all() + ) + + existing_timestamps = set() + for obj in existing: + for tag in obj.tags: + if tag.key == "timestamp": + try: + existing_timestamps.add(float(tag.value)) + except ValueError: + continue + + return existing_timestamps + + @classmethod + def _create_chapter_frames( + cls, + chapters, + duration, + object_, + output_dir, + existing_timestamps, + progress_updater=None, + ): + """Create frames for chapters that don't already exist at those timestamps.""" + created_frames = [] + current_chapter = 0 + + with move_file_into_local(object_, delete=True) as url: + for chapter in chapters: + current_chapter += 1 + + if progress_updater: + progress_updater(current_chapter) + + chapter_seconds = chapter["seconds"] + chapter_title = chapter["title"] + + # Skip chapters that are beyond video duration + if chapter_seconds > duration: + continue + + # Skip if frame already exists at this timestamp (with some tolerance) + timestamp_exists = any( + abs(existing_ts - chapter_seconds) < 0.1 + for existing_ts in existing_timestamps + ) + if timestamp_exists: + continue + + frame_filename = "chapter-{0}.jpg".format(int(chapter_seconds)) + frame_path = os.path.join(output_dir, frame_filename) + + # For 0:00 chapters, use a small offset to avoid extraction issues + chapter_seconds = max(chapter_seconds, 0.1) if chapter_seconds == 0 else chapter_seconds + + try: + # Extract single frame at chapter timestamp using ff_frames + ff_frames( + input_file=url, + start=chapter_seconds, + end=chapter_seconds + 0.01, # Extract just one frame + step=1, + duration=duration, + output=frame_path, + ) + + if os.path.exists(frame_path) and os.path.getsize(frame_path) > 0: + # Create ObjectVersion for chapter frame (as normal frame) + ExtractFramesTask._create_object( + bucket=object_.bucket, + key=frame_filename, + stream=file_opener_xrootd(frame_path, "rb"), + size=os.path.getsize(frame_path), + media_type="image", + context_type="frame", + master_id=object_.version_id, + is_chapter_frame=True, + timestamp=chapter_seconds, + ) + + created_frames.append(frame_path) + + except Exception as e: + # Log error but continue with other chapters + current_app.logger.error( + "Failed to extract frame for chapter at {0}s: {1}".format( + chapter_seconds, str(e) + ) + ) + continue + + return created_frames + + class TranscodeVideoTask(AVCTask): """Transcode video task. @@ -793,7 +1039,7 @@ def _update_flow_tasks(self, flow_tasks, status, message, **kwargs): opencast_publication_tag=current_app.config["CDS_OPENCAST_QUALITIES"][ quality ]["opencast_publication_tag"], - **kwargs # may contain `opencast_event_id` + **kwargs, # may contain `opencast_event_id` ) # JSONb cols needs to be assigned (not updated) to be persisted flow_task_metadata.payload = new_payload @@ -848,7 +1094,7 @@ def _start_transcodable_flow_tasks_or_cancel(self, wanted_qualities=None): new_payload.update( task_id=str(t.id), celery_task_id=str(self.request.id), - **self._base_payload + **self._base_payload, ) # JSONb cols needs to be assigned (not updated) to be persisted t.payload = new_payload diff --git a/cds/modules/previewer/templates/cds_previewer/macros/player.html b/cds/modules/previewer/templates/cds_previewer/macros/player.html index a7301dd58..a198cc557 100644 --- a/cds/modules/previewer/templates/cds_previewer/macros/player.html +++ b/cds/modules/previewer/templates/cds_previewer/macros/player.html @@ -53,44 +53,134 @@ }); window.top.player = player; - // Preload - player.source = { - sources: [ - { - {% if video_source %} - src: "{{ video_source }}", - type: 'application/x-mpegURL' - {% elif obj.m3u8_uri and obj.subformats|length > 0 %} - src: '{{ obj.m3u8_uri }}', - type: 'application/x-mpegURL' - {% else %} - src: '{{ obj.uri }}', - type: 'video/mp4' - {% endif %} - }, - ], - {% if not embed_config.subtitlesOff %} - textTracks: [ - { - kind: 'metadata', - src: '{{ obj.thumbnails_uri }}', - label: 'thumbnails', - default: true, - }, - {% for uri, lang in obj.subtitles %} - { - kind: 'subtitles', - src: '{{ uri }}', - label: '{{ lang }}', - srclang: '{{ lang }}', - {% if embed_config.subtitles and embed_config.subtitles == lang %} - default: true, - {% endif %} - }, - {% endfor %} - ], - {% endif %} - poster: '{{ obj.poster_uri }}', + + // --- Chapters helpers --- + function durationToSeconds(durationStr) { + if (!durationStr) return null; + const parts = durationStr.split(':').map(Number); // [HH, MM, SS] or [MM, SS] + if (parts.length === 3) { + return parts[0] * 3600 + parts[1] * 60 + parts[2]; + } + if (parts.length === 2) { + return parts[0] * 60 + parts[1]; + } + return null; + } + function cleanHtmlFromTitle(title) { + if (!title) return title; + + // Decode HTML entities by using a temporary DOM element + const temp = document.createElement("textarea"); + temp.innerHTML = title; + let decoded = temp.value; + decoded = decoded.replace(/<[^>]+>/g, " "); + decoded = decoded.replace(/\s+/g, " ").trim(); + + return decoded; + } + function parseChapters(description) { + if (!description) return []; + + const pattern = /(?:^|\n)\s*(\d{1,2}:(?:\d{1,2}:)?\d{1,2})\s*[-\s]*(.+?)(?=\n|$)/gm; + const chapters = []; + let match; + + while ((match = pattern.exec(description)) !== null) { + const [, timestampStr, title] = match; + + const timeParts = timestampStr.split(":"); + let totalSeconds; + + if (timeParts.length === 2) { + const [minutes, seconds] = timeParts.map(Number); + totalSeconds = minutes * 60 + seconds; + } else if (timeParts.length === 3) { + const [hours, minutes, seconds] = timeParts.map(Number); + totalSeconds = hours * 3600 + minutes * 60 + seconds; + } else { + continue; + } + + const cleanTitle = cleanHtmlFromTitle(title); + if (cleanTitle) { + chapters.push({ + startTime: totalSeconds, + text: cleanTitle + }); + } + } + + chapters.sort((a, b) => a.startTime - b.startTime); + + // Add endTime for VTT + const videoDuration = durationToSeconds({{ (record.duration if record and record.duration else "") | tojson }}); + for (let i = 0; i < chapters.length; i++) { + if (i + 1 < chapters.length) { + chapters[i].endTime = chapters[i + 1].startTime; + } else { + chapters[i].endTime = videoDuration || (chapters[i].startTime + 1); + } + } + + return chapters; + } + function formatVttTime(total) { + const h = Math.floor(total / 3600); + const m = Math.floor((total % 3600) / 60); + const s = Math.floor(total % 60); + return String(h).padStart(2,'0') + ':' + String(m).padStart(2,'0') + ':' + String(s).padStart(2,'0') + '.000'; + } + function buildChaptersVtt(description) { + const chapters = parseChapters(description); + if (!chapters.length) return null; + let vtt = 'WEBVTT\n\n'; + for (let i = 0; i < chapters.length; i++) { + const c = chapters[i]; + vtt += `${i+1}\n${formatVttTime(c.startTime)} --> ${formatVttTime(c.endTime)}\n${c.text}\n\n`; + } + return vtt; + } + + // Build textTracks + var textTracksArr = []; + // Chapters + var descStr = {{ (record.description if record and record.description else "") | tojson }}; + var vttStr = buildChaptersVtt(descStr); + if (vttStr) { + var chapUrl = URL.createObjectURL(new Blob([vttStr], { type: 'text/vtt' })); + textTracksArr.push({ kind: 'chapters', src: chapUrl, label: 'Chapters' }); + } + {% if not embed_config.subtitlesOff %} + textTracksArr.push({ kind: 'metadata', src: '{{ obj.thumbnails_uri }}', label: 'thumbnails', default: true }); + {% for uri, lang in obj.subtitles %} + textTracksArr.push({ + kind: 'subtitles', + src: '{{ uri }}', + label: '{{ lang }}', + srclang: '{{ lang }}' + {% if embed_config.subtitles and embed_config.subtitles == lang %}, default: true{% endif %} + }); + {% endfor %} + {% endif %} + + // Set source with textTracks + player.source = { + sources: [ + { + {% if video_source %} + src: "{{ video_source }}", + type: 'application/x-mpegURL' + {% elif obj.m3u8_uri and obj.subformats|length > 0 %} + src: '{{ obj.m3u8_uri }}', + type: 'application/x-mpegURL' + {% else %} + src: '{{ obj.uri }}', + type: 'video/mp4' + {% endif %} + } + ], + textTracks: textTracksArr, + poster: '{{ obj.poster_uri }}', {% if obj.vr %} vr: { 360: true, diff --git a/cds/modules/records/serializers/schemas/video.py b/cds/modules/records/serializers/schemas/video.py index 88fcb3d8c..13b34ac89 100644 --- a/cds/modules/records/serializers/schemas/video.py +++ b/cds/modules/records/serializers/schemas/video.py @@ -19,7 +19,7 @@ """Video JSON schema.""" from invenio_jsonschemas import current_jsonschemas -from marshmallow import Schema, fields, pre_load, post_load +from marshmallow import Schema, fields, pre_load, post_load, post_dump from ....deposit.api import Video from ..fields.datetime import DateString @@ -43,6 +43,7 @@ TranslationsSchema, ) from .doi import DOI +from ...utils import parse_video_chapters class _CDSSSchema(Schema): @@ -166,6 +167,7 @@ class VideoSchema(StrictKeysSchema): ) collections = fields.List(fields.Str, many=True) additional_languages = fields.List(fields.Str, many=True) + chapters = fields.List(fields.Dict, dump_only=True) # Preservation fields location = fields.Str() @@ -177,3 +179,14 @@ def post_load(self, data, **kwargs): """Post load.""" data["$schema"] = current_jsonschemas.path_to_url(Video._schema) return data + + @post_dump(pass_many=False) + def post_dump(self, data, **kwargs): + """Post dump - add parsed chapters.""" + description = data.get('description', '') + if description: + data['chapters'] = parse_video_chapters(description) + else: + data['chapters'] = [] + + return data diff --git a/cds/modules/records/static/templates/cds_records/video/detail.html b/cds/modules/records/static/templates/cds_records/video/detail.html index a8444c7e1..f28126fec 100644 --- a/cds/modules/records/static/templates/cds_records/video/detail.html +++ b/cds/modules/records/static/templates/cds_records/video/detail.html @@ -4,9 +4,9 @@
-
-
-
+
+
+