28868391225

Committed 07 Jul 2026 01:05PM UTC coverage: 69.231%. First build

Build # 28868391225

Build Type

Pull #1427

github

Committed by

web-flow

Commit Message

Merge c4df79857 into 82a38ac96

Pull Request Pull Request #1427: [Release] 4.3.0

Coverage Stats

921 of 1273 new or added lines in 31 files covered. (72.35%)

13743 of 19851 relevant lines covered (69.23%)

0.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.52

/pod/video_encode_transcript/rest_views.py

import json
import logging
import os

import webvtt
from django.conf import settings
from django.core.exceptions import SuspiciousFileOperation, SuspiciousOperation
from django.shortcuts import get_object_or_404
from django.utils._os import safe_join
from django.utils.translation import gettext_lazy as _
from django.views.decorators.csrf import csrf_exempt
from rest_framework import serializers, viewsets
from rest_framework.decorators import action, api_view
from rest_framework.response import Response

from pod.recorder.models import Recording
from pod.video.models import Video
from pod.video.rest_views import VideoSerializer

from .models import EncodingAudio, EncodingVideo, PlaylistVideo, VideoRendition

USE_TRANSCRIPTION = getattr(settings, "USE_TRANSCRIPTION", False)
if USE_TRANSCRIPTION:
    from pod.video_encode_transcript.transcript import start_transcript

MEDIA_ROOT = getattr(settings, "MEDIA_ROOT", "")
DEFAULT_RECORDER_PATH = getattr(settings, "DEFAULT_RECORDER_PATH", "")

DEBUG = getattr(settings, "DEBUG", True)
logger = logging.getLogger(__name__)
if DEBUG:
    logger.setLevel(logging.DEBUG)


def _safe_temp_media_path(filename: str) -> str:
    """Resolve a temporary media file path and keep it under MEDIA_ROOT/temp."""
    media_temp_dir = os.path.realpath(os.path.join(MEDIA_ROOT, "temp"))
    try:
        filepath = os.path.realpath(safe_join(media_temp_dir, filename))
        if os.path.commonpath([filepath, media_temp_dir]) != media_temp_dir:
            raise SuspiciousOperation(_("Unsafe temporary path."))
    except (SuspiciousFileOperation, ValueError) as exc:
        raise SuspiciousOperation(_("Unsafe temporary path.")) from exc
    return filepath


class VideoRenditionSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = VideoRendition
        fields = (
            "id",
            "url",
            "resolution",
            "video_bitrate",
            "audio_bitrate",
            "encode_mp4",
            "sites",
        )


class EncodingVideoSerializer(serializers.HyperlinkedModelSerializer):
    """Serializer for the EncodingVideo model."""

    class Meta:
        model = EncodingVideo
        fields = (
            "id",
            "url",
            "name",
            "video",
            "rendition",
            "encoding_format",
            "source_file",
            "sites_all",
        )


class EncodingAudioSerializer(serializers.HyperlinkedModelSerializer):
    """Serializer for the EncodingAudio model."""

    class Meta:
        model = EncodingAudio
        fields = (
            "id",
            "url",
            "name",
            "video",
            "encoding_format",
            "source_file",
            "sites_all",
        )


class EncodingVideoViewSet(viewsets.ModelViewSet):
    """Viewset for EncodingVideo model."""

    queryset = EncodingVideo.objects.all()
    serializer_class = EncodingVideoSerializer
    filterset_fields = [
        "video",
    ]

    @action(detail=False, methods=["get"])
    def video_encodedfiles(self, request):
        """Retrieve encoded video files."""
        encoded_videos = EncodingVideoViewSet.filter_encoded_medias(
            self.queryset, request
        )
        encoded_videos = sorted(encoded_videos, key=lambda x: x.height)
        serializer = EncodingVideoSerializer(
            encoded_videos, many=True, context={"request": request}
        )
        return Response(serializer.data)

    @staticmethod
    def filter_encoded_medias(queryset, request):
        """Filter encoded media files."""
        encoded_audios = queryset
        if request.GET.get("video"):
            encoded_audios = encoded_audios.filter(video__id=request.GET.get("video"))
        if request.GET.get("extension"):
            encoded_audios = encoded_audios.filter(
                source_file__iendswith=request.GET.get("extension")
            )
        return encoded_audios


class EncodingAudioViewSet(viewsets.ModelViewSet):
    """Viewset for EncodingAudio model."""

    queryset = EncodingAudio.objects.all()
    serializer_class = EncodingAudioSerializer
    filterset_fields = [
        "video",
    ]

    @action(detail=False, methods=["get"])
    def audio_encodedfiles(self, request):
        """Retrieve encoded audio files."""
        encoded_audios = EncodingVideoViewSet.filter_encoded_medias(
            self.queryset, request
        )
        serializer = EncodingAudioSerializer(
            encoded_audios, many=True, context={"request": request}
        )
        return Response(serializer.data)


class VideoRenditionViewSet(viewsets.ModelViewSet):
    queryset = VideoRendition.objects.all()
    serializer_class = VideoRenditionSerializer


class PlaylistVideoSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = PlaylistVideo
        fields = (
            "id",
            "url",
            "name",
            "video",
            "encoding_format",
            "source_file",
            "sites_all",
        )


class PlaylistVideoViewSet(viewsets.ModelViewSet):
    queryset = PlaylistVideo.objects.all()
    serializer_class = PlaylistVideoSerializer


@api_view(["GET"])
def launch_encode_view(request):
    """View API for launching video encoding."""
    video = get_object_or_404(Video, slug=request.GET.get("slug"))
    if (
        video is not None
        and (
            not hasattr(video, "launch_encode") or getattr(video, "launch_encode") is True
        )
        and video.encoding_in_progress is False
    ):
        video.launch_encode = True
        video.save()
    return Response(VideoSerializer(instance=video, context={"request": request}).data)


@api_view(["GET"])
def launch_transcript_view(request):
    """View API for launching transcript."""
    video = get_object_or_404(Video, slug=request.GET.get("slug"))
    if video is not None and video.get_video_mp3():
        start_transcript(video.id, threaded=True)
    return Response(VideoSerializer(instance=video, context={"request": request}).data)


@csrf_exempt
@api_view(["POST"])
def store_remote_encoded_video(request):
    """View API for storing remote encoded videos."""
    from .encode import end_of_encoding, store_encoding_info
    from .Encoding_video_model import Encoding_video_model

    video_id = request.GET.get("id", 0)
    logger.info("Start importing encoding data for video: %s" % video_id)
    video = get_object_or_404(Video, id=video_id)
    # start_store_remote_encoding_video(video_id)
    # check if video is encoding!!!
    data = json.loads(request.body.decode("utf-8"))
    if video.encoding_in_progress is False:
        raise SuspiciousOperation(_("Video not being encoded."))
    if str(video_id) != str(data["video_id"]):
        raise SuspiciousOperation(
            _("Different video id: %(expected)s - %(received)s")
            % {"expected": video_id, "received": data["video_id"]}
        )
    encoding_video = Encoding_video_model(
        video_id, video.video.path, data["cut_start"], data["cut_end"]
    )
    encoding_video.start = data["start"]
    encoding_video.stop = data["stop"]
    final_video = store_encoding_info(video_id, encoding_video)
    end_of_encoding(final_video)
    return Response(VideoSerializer(instance=video, context={"request": request}).data)


@csrf_exempt
@api_view(["POST"])
def store_remote_encoded_video_studio(request):
    from .encode import store_encoding_studio_info

    recording_id = request.GET.get("recording_id", 0)
    get_object_or_404(Recording, id=recording_id)
    data = json.loads(request.body.decode("utf-8"))
    safe_video_output = _get_safe_video_output(data)
    store_encoding_studio_info(recording_id, safe_video_output, data["msg"])
    return Response("ok")


def _get_safe_video_output(data):
    """Return a safe absolute video output path within allowed base dirs."""
    raw_video_output = str(data.get("video_output", ""))
    if not raw_video_output:
        raise SuspiciousOperation(_("Missing video output path."))
    normalized_video_output = os.path.normpath(raw_video_output)
    if os.path.isabs(normalized_video_output) or normalized_video_output.startswith(".."):
        raise SuspiciousFileOperation(_("Invalid video output path."))
    base_output_dir = DEFAULT_RECORDER_PATH or MEDIA_ROOT
    if not base_output_dir:
        raise SuspiciousOperation(_("No safe base directory configured."))
    try:
        safe_video_output = safe_join(base_output_dir, normalized_video_output)
    except SuspiciousFileOperation:
        raise SuspiciousFileOperation(_("Invalid video output path."))
    if not safe_video_output:
        raise SuspiciousFileOperation(_("Invalid video output path."))
    return safe_video_output


@csrf_exempt
@api_view(["POST"])
def store_remote_transcripted_video(request):
    """View API for storing remote transcripted videos."""
    from .transcript import save_vtt_and_notify

    video_id = request.GET.get("id", 0)
    video = get_object_or_404(Video, id=video_id)
    # check if video is encoding!!!
    data = json.loads(request.body.decode("utf-8"))
    if str(video_id) != str(data["video_id"]):
        raise SuspiciousOperation(
            _("Different video id: %(expected)s - %(received)s")
            % {"expected": video_id, "received": data["video_id"]}
        )
    logger.info("Start importing transcription data for video: %s" % video_id)
    filename = os.path.basename(data["temp_vtt_file"])
    filepath = _safe_temp_media_path(filename)
    media_temp_dir = os.path.realpath(os.path.join(MEDIA_ROOT, "temp"))
    if os.path.commonpath([filepath, media_temp_dir]) != media_temp_dir:
        raise SuspiciousOperation(_("Unsafe temporary path."))
    new_vtt = webvtt.read(filepath)
    save_vtt_and_notify(video, data["msg"], new_vtt)
    os.remove(filepath)
    return Response(VideoSerializer(instance=video, context={"request": request}).data)

1	import json	1✔
2	import logging	1✔
3	import os	1✔
4
5	import webvtt	1✔
6	from django.conf import settings	1✔
7	from django.core.exceptions import SuspiciousFileOperation, SuspiciousOperation	1✔
8	from django.shortcuts import get_object_or_404	1✔
9	from django.utils._os import safe_join	1✔
10	from django.utils.translation import gettext_lazy as _	1✔
11	from django.views.decorators.csrf import csrf_exempt	1✔
12	from rest_framework import serializers, viewsets	1✔
13	from rest_framework.decorators import action, api_view	1✔
14	from rest_framework.response import Response	1✔
15
16	from pod.recorder.models import Recording	1✔
17	from pod.video.models import Video	1✔
18	from pod.video.rest_views import VideoSerializer	1✔
19
20	from .models import EncodingAudio, EncodingVideo, PlaylistVideo, VideoRendition	1✔
21
22	USE_TRANSCRIPTION = getattr(settings, "USE_TRANSCRIPTION", False)	1✔
23	if USE_TRANSCRIPTION:	1✔
24	from pod.video_encode_transcript.transcript import start_transcript	1✔
25
26	MEDIA_ROOT = getattr(settings, "MEDIA_ROOT", "")	1✔
27	DEFAULT_RECORDER_PATH = getattr(settings, "DEFAULT_RECORDER_PATH", "")	1✔
28
29	DEBUG = getattr(settings, "DEBUG", True)	1✔
30	logger = logging.getLogger(__name__)	1✔
31	if DEBUG:	1✔
32	logger.setLevel(logging.DEBUG)	1✔
33
34
35	def _safe_temp_media_path(filename: str) -> str:	1✔
36	"""Resolve a temporary media file path and keep it under MEDIA_ROOT/temp."""
37	media_temp_dir = os.path.realpath(os.path.join(MEDIA_ROOT, "temp"))	1✔
38	try:	1✔
39	filepath = os.path.realpath(safe_join(media_temp_dir, filename))	1✔
40	if os.path.commonpath([filepath, media_temp_dir]) != media_temp_dir:	1✔
NEW 41	raise SuspiciousOperation(_("Unsafe temporary path."))	×
42	except (SuspiciousFileOperation, ValueError) as exc:	1✔
43	raise SuspiciousOperation(_("Unsafe temporary path.")) from exc	1✔
44	return filepath	1✔
45
46
47	class VideoRenditionSerializer(serializers.HyperlinkedModelSerializer):	1✔
48	class Meta:	1✔
49	model = VideoRendition	1✔
50	fields = (	1✔
51	"id",
52	"url",
53	"resolution",
54	"video_bitrate",
55	"audio_bitrate",
56	"encode_mp4",
57	"sites",
58	)
59
60
61	class EncodingVideoSerializer(serializers.HyperlinkedModelSerializer):	1✔
62	"""Serializer for the EncodingVideo model."""
63
64	class Meta:	1✔
65	model = EncodingVideo	1✔
66	fields = (	1✔
67	"id",
68	"url",
69	"name",
70	"video",
71	"rendition",
72	"encoding_format",
73	"source_file",
74	"sites_all",
75	)
76
77
78	class EncodingAudioSerializer(serializers.HyperlinkedModelSerializer):	1✔
79	"""Serializer for the EncodingAudio model."""
80
81	class Meta:	1✔
82	model = EncodingAudio	1✔
83	fields = (	1✔
84	"id",
85	"url",
86	"name",
87	"video",
88	"encoding_format",
89	"source_file",
90	"sites_all",
91	)
92
93
94	class EncodingVideoViewSet(viewsets.ModelViewSet):	1✔
95	"""Viewset for EncodingVideo model."""
96
97	queryset = EncodingVideo.objects.all()	1✔
98	serializer_class = EncodingVideoSerializer	1✔
99	filterset_fields = [	1✔
100	"video",
101	]
102
103	@action(detail=False, methods=["get"])	1✔
104	def video_encodedfiles(self, request):	1✔
105	"""Retrieve encoded video files."""
106	encoded_videos = EncodingVideoViewSet.filter_encoded_medias(	×
107	self.queryset, request
108	)
109	encoded_videos = sorted(encoded_videos, key=lambda x: x.height)	×
110	serializer = EncodingVideoSerializer(	×
111	encoded_videos, many=True, context={"request": request}
112	)
113	return Response(serializer.data)	×
114
115	@staticmethod	1✔
116	def filter_encoded_medias(queryset, request):	1✔
117	"""Filter encoded media files."""
118	encoded_audios = queryset	×
119	if request.GET.get("video"):	×
120	encoded_audios = encoded_audios.filter(video__id=request.GET.get("video"))	×
121	if request.GET.get("extension"):	×
122	encoded_audios = encoded_audios.filter(	×
123	source_file__iendswith=request.GET.get("extension")
124	)
125	return encoded_audios	×
126
127
128	class EncodingAudioViewSet(viewsets.ModelViewSet):	1✔
129	"""Viewset for EncodingAudio model."""
130
131	queryset = EncodingAudio.objects.all()	1✔
132	serializer_class = EncodingAudioSerializer	1✔
133	filterset_fields = [	1✔
134	"video",
135	]
136
137	@action(detail=False, methods=["get"])	1✔
138	def audio_encodedfiles(self, request):	1✔
139	"""Retrieve encoded audio files."""
140	encoded_audios = EncodingVideoViewSet.filter_encoded_medias(	×
141	self.queryset, request
142	)
143	serializer = EncodingAudioSerializer(	×
144	encoded_audios, many=True, context={"request": request}
145	)
146	return Response(serializer.data)	×
147
148
149	class VideoRenditionViewSet(viewsets.ModelViewSet):	1✔
150	queryset = VideoRendition.objects.all()	1✔
151	serializer_class = VideoRenditionSerializer	1✔
152
153
154	class PlaylistVideoSerializer(serializers.HyperlinkedModelSerializer):	1✔
155	class Meta:	1✔
156	model = PlaylistVideo	1✔
157	fields = (	1✔
158	"id",
159	"url",
160	"name",
161	"video",
162	"encoding_format",
163	"source_file",
164	"sites_all",
165	)
166
167
168	class PlaylistVideoViewSet(viewsets.ModelViewSet):	1✔
169	queryset = PlaylistVideo.objects.all()	1✔
170	serializer_class = PlaylistVideoSerializer	1✔
171
172
173	@api_view(["GET"])	1✔
174	def launch_encode_view(request):	1✔
175	"""View API for launching video encoding."""
176	video = get_object_or_404(Video, slug=request.GET.get("slug"))	1✔
177	if (	1✔
178	video is not None
179	and (
180	not hasattr(video, "launch_encode") or getattr(video, "launch_encode") is True
181	)
182	and video.encoding_in_progress is False
183	):
184	video.launch_encode = True	1✔
185	video.save()	1✔
186	return Response(VideoSerializer(instance=video, context={"request": request}).data)	1✔
187
188
189	@api_view(["GET"])	1✔
190	def launch_transcript_view(request):	1✔
191	"""View API for launching transcript."""
192	video = get_object_or_404(Video, slug=request.GET.get("slug"))	1✔
193	if video is not None and video.get_video_mp3():	1✔
194	start_transcript(video.id, threaded=True)	1✔
195	return Response(VideoSerializer(instance=video, context={"request": request}).data)	1✔
196
197
198	@csrf_exempt	1✔
199	@api_view(["POST"])	1✔
200	def store_remote_encoded_video(request):	1✔
201	"""View API for storing remote encoded videos."""
202	from .encode import end_of_encoding, store_encoding_info	×
203	from .Encoding_video_model import Encoding_video_model	×
204
205	video_id = request.GET.get("id", 0)	×
206	logger.info("Start importing encoding data for video: %s" % video_id)	×
207	video = get_object_or_404(Video, id=video_id)	×
208	# start_store_remote_encoding_video(video_id)
209	# check if video is encoding!!!
210	data = json.loads(request.body.decode("utf-8"))	×
211	if video.encoding_in_progress is False:	×
NEW 212	raise SuspiciousOperation(_("Video not being encoded."))	×
213	if str(video_id) != str(data["video_id"]):	×
214	raise SuspiciousOperation(	×
215	_("Different video id: %(expected)s - %(received)s")
216	% {"expected": video_id, "received": data["video_id"]}
217	)
218	encoding_video = Encoding_video_model(	×
219	video_id, video.video.path, data["cut_start"], data["cut_end"]
220	)
221	encoding_video.start = data["start"]	×
222	encoding_video.stop = data["stop"]	×
223	final_video = store_encoding_info(video_id, encoding_video)	×
224	end_of_encoding(final_video)	×
225	return Response(VideoSerializer(instance=video, context={"request": request}).data)	×
226
227
228	@csrf_exempt	1✔
229	@api_view(["POST"])	1✔
230	def store_remote_encoded_video_studio(request):	1✔
231	from .encode import store_encoding_studio_info	1✔
232
233	recording_id = request.GET.get("recording_id", 0)	1✔
234	get_object_or_404(Recording, id=recording_id)	1✔
235	data = json.loads(request.body.decode("utf-8"))	1✔
236	safe_video_output = _get_safe_video_output(data)	1✔
237	store_encoding_studio_info(recording_id, safe_video_output, data["msg"])	1✔
238	return Response("ok")	1✔
239
240
241	def _get_safe_video_output(data):	1✔
242	"""Return a safe absolute video output path within allowed base dirs."""
243	raw_video_output = str(data.get("video_output", ""))	1✔
244	if not raw_video_output:	1✔
NEW 245	raise SuspiciousOperation(_("Missing video output path."))	×
246	normalized_video_output = os.path.normpath(raw_video_output)	1✔
247	if os.path.isabs(normalized_video_output) or normalized_video_output.startswith(".."):	1✔
248	raise SuspiciousFileOperation(_("Invalid video output path."))	1✔
249	base_output_dir = DEFAULT_RECORDER_PATH or MEDIA_ROOT	1✔
250	if not base_output_dir:	1✔
NEW 251	raise SuspiciousOperation(_("No safe base directory configured."))	×
252	try:	1✔
253	safe_video_output = safe_join(base_output_dir, normalized_video_output)	1✔
NEW 254	except SuspiciousFileOperation:	×
NEW 255	raise SuspiciousFileOperation(_("Invalid video output path."))	×
256	if not safe_video_output:	1✔
NEW 257	raise SuspiciousFileOperation(_("Invalid video output path."))	×
258	return safe_video_output	1✔
259
260
261	@csrf_exempt	1✔
262	@api_view(["POST"])	1✔
263	def store_remote_transcripted_video(request):	1✔
264	"""View API for storing remote transcripted videos."""
265	from .transcript import save_vtt_and_notify	×
266
267	video_id = request.GET.get("id", 0)	×
268	video = get_object_or_404(Video, id=video_id)	×
269	# check if video is encoding!!!
270	data = json.loads(request.body.decode("utf-8"))	×
271	if str(video_id) != str(data["video_id"]):	×
272	raise SuspiciousOperation(	×
273	_("Different video id: %(expected)s - %(received)s")
274	% {"expected": video_id, "received": data["video_id"]}
275	)
276	logger.info("Start importing transcription data for video: %s" % video_id)	×
277	filename = os.path.basename(data["temp_vtt_file"])	×
NEW 278	filepath = _safe_temp_media_path(filename)	×
NEW 279	media_temp_dir = os.path.realpath(os.path.join(MEDIA_ROOT, "temp"))	×
NEW 280	if os.path.commonpath([filepath, media_temp_dir]) != media_temp_dir:	×
NEW 281	raise SuspiciousOperation(_("Unsafe temporary path."))	×
282	new_vtt = webvtt.read(filepath)	×
283	save_vtt_and_notify(video, data["msg"], new_vtt)	×
284	os.remove(filepath)	×
285	return Response(VideoSerializer(instance=video, context={"request": request}).data)	×

EsupPortail / Esup-Pod / 28868391225

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous