27555111584

Committed 15 Jun 2026 02:55PM UTC coverage: 69.164% (+0.1%) from 69.062%

Build # 27555111584

Build Type

push

github

Committed by

web-flow

Commit Message

Security Hardening and Priority-User Support for Encoding/Transcript workflows (#1454)

## Summary
Improves security across multiple entry points (forms, views, JS handlers, and encoding utilities) and introduces priority-user support for video encoding/transcript processing.

## What changed
- Hardened input validation/sanitization in backend forms and views.
- Improved client-side input handling in several JavaScript modules.
- Strengthened encoding/transcript execution paths and task-processing safeguards.
- Added/updated admin customizations for encoding-related management screens.
- Extended automated test coverage for security and queue behavior.
- Updated i18n strings and related project dependencies.
- Allow the v3-to-v4 migration scripts to run on any Pod 4.x version instead of only Pod 4.0.x.

## Database impact
Includes a **database schema change related to priority users** (PriorityUser support for encoding/transcript workflows).  
A migration must be applied in target environments before/with deployment.

## Operational notes
- Run migrations during deployment.
- No functional rollback should be attempted without rolling back the related migration as well.

Coverage Stats

424 of 600 new or added lines in 15 files covered. (70.67%)

15 existing lines in 4 files now uncovered.

13628 of 19704 relevant lines covered (69.16%)

0.69 hits per line

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

90.44

/pod/video_encode_transcript/models.py

"""Models for Esup-Pod video_encode."""

import os

from django.conf import settings
from django.contrib.sites.models import Site
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator
from django.db import models
from django.db.models.signals import post_save, pre_save
from django.dispatch import receiver
from django.utils import timezone
from django.utils.translation import gettext_lazy as _

from pod.recorder.models import Recording
from pod.video.models import Video, get_storage_path_video

ENCODING_CHOICES = getattr(
    settings,
    "ENCODING_CHOICES",
    (
        ("audio", "audio"),
        ("360p", "360p"),
        ("480p", "480p"),
        ("720p", "720p"),
        ("1080p", "1080p"),
        ("playlist", "playlist"),
    ),
)

FORMAT_CHOICES = getattr(
    settings,
    "FORMAT_CHOICES",
    (
        ("video/mp4", "video/mp4"),
        ("video/mp2t", "video/mp2t"),
        ("video/webm", "video/webm"),
        ("audio/mp3", "audio/mp3"),
        ("audio/wav", "audio/wav"),
        ("application/x-mpegURL", "application/x-mpegURL"),
    ),
)

SITE_ID = getattr(settings, "SITE_ID", 1)


class VideoRendition(models.Model):
    """Model representing the rendition video."""

    resolution = models.CharField(
        _("resolution"),
        max_length=50,
        unique=True,
        help_text=_(
            "Please use the only format x. i.e.: "
            + "<em>640x360</em> or <em>1280x720</em>"
            + " or <em>1920x1080</em>."
        ),
    )
    minrate = models.CharField(
        _("minrate"),
        max_length=50,
        help_text=_(
            "Please use the only format k. i.e.: "
            + "<em>300k</em> or <em>600k</em>"
            + " or <em>1000k</em>."
        ),
    )
    video_bitrate = models.CharField(
        _("bitrate video"),
        max_length=50,
        help_text=_(
            "Please use the only format k. i.e.: "
            + "<em>300k</em> or <em>600k</em>"
            + " or <em>1000k</em>."
        ),
    )
    maxrate = models.CharField(
        _("maxrate"),
        max_length=50,
        help_text=_(
            "Please use the only format k. i.e.: "
            + "<em>300k</em> or <em>600k</em>"
            + " or <em>1000k</em>."
        ),
    )
    encoding_resolution_threshold = models.PositiveIntegerField(
        _("encoding resolution threshold"),
        default=0,
        validators=[MaxValueValidator(100)],
    )
    audio_bitrate = models.CharField(
        _("bitrate audio"),
        max_length=50,
        help_text=_(
            "Please use the only format k. i.e.: "
            + "<em>300k</em> or <em>600k</em>"
            + " or <em>1000k</em>."
        ),
    )
    encode_mp4 = models.BooleanField(_("Make a MP4 version"), default=False)
    sites = models.ManyToManyField(Site)

    @property
    def height(self) -> int:
        """The height of the video rendition based on the resolution."""
        return int(self.resolution.split("x")[1])

    @property
    def width(self) -> int:
        """The width of the video rendition based on the resolution."""
        return int(self.resolution.split("x")[0])

    class Meta:
        # ordering = ['height'] # Not work
        verbose_name = _("rendition")
        verbose_name_plural = _("renditions")

    def __str__(self) -> str:
        return "VideoRendition num %s with resolution %s" % (
            "%04d" % self.id,
            self.resolution,
        )

    def bitrate(self, field_value, field_name, name=None) -> None:
        """Validate the bitrate field value."""
        if name is None:
            name = field_name
        if field_value and "k" not in field_value:
            msg = "Error in %s: " % _(name)
            raise ValidationError(
                "%s %s" % (msg, VideoRendition._meta.get_field(field_name).help_text)
            )
        else:
            vb = field_value.replace("k", "")
            if not vb.isdigit():
                msg = "Error in %s: " % _(name)
                raise ValidationError(
                    "%s %s" % (msg, VideoRendition._meta.get_field(field_name).help_text)
                )

    def clean_bitrate(self) -> None:
        """Clean the bitrate-related fields."""
        self.bitrate(self.video_bitrate, "video_bitrate", "bitrate video")
        self.bitrate(self.maxrate, "maxrate")
        self.bitrate(self.minrate, "minrate")

    def clean(self) -> None:
        """Clean the fields of the VideoRendition model."""
        if self.resolution and "x" not in self.resolution:
            raise ValidationError(VideoRendition._meta.get_field("resolution").help_text)
        else:
            res = self.resolution.replace("x", "")
            if not res.isdigit():
                raise ValidationError(
                    VideoRendition._meta.get_field("resolution").help_text
                )

        self.clean_bitrate()
        self.bitrate(self.audio_bitrate, "audio_bitrate", "bitrate audio")


@receiver(post_save, sender=VideoRendition)
def default_site_videorendition(sender, instance, created: bool, **kwargs) -> None:
    """Add the current site as a default site."""
    if instance.sites.count() == 0:
        instance.sites.add(Site.objects.get_current())


class EncodingVideo(models.Model):
    """Model representing the encoding video for a video."""

    name = models.CharField(
        _("Name"),
        max_length=10,
        choices=ENCODING_CHOICES,
        default="360p",
        help_text=_("Please use the only format in encoding choices:")
        + " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
    )
    video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)
    rendition = models.ForeignKey(
        VideoRendition, verbose_name=_("rendition"), on_delete=models.CASCADE
    )
    encoding_format = models.CharField(
        _("Format"),
        max_length=22,
        choices=FORMAT_CHOICES,
        default="video/mp4",
        help_text=_("Please use the only format in format choices:")
        + " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
    )
    source_file = models.FileField(
        _("encoding source file"),
        upload_to=get_storage_path_video,
        max_length=255,
    )

    @property
    def sites(self):
        """Property representing the sites associated with the video."""
        return self.video.sites

    @property
    def sites_all(self):
        """Property representing all the sites associated with the video."""
        return self.video.sites_set.all()

    def clean(self) -> None:
        """Validate the encoding video model."""
        if self.name:
            if self.name not in dict(ENCODING_CHOICES):
                raise ValidationError(EncodingVideo._meta.get_field("name").help_text)
        if self.encoding_format:
            if self.encoding_format not in dict(FORMAT_CHOICES):
                raise ValidationError(
                    EncodingVideo._meta.get_field("encoding_format").help_text
                )

    class Meta:
        ordering = ["name"]
        verbose_name = _("Encoding video")
        verbose_name_plural = _("Encoding videos")

    def __str__(self) -> str:
        return "EncodingVideo num: %s with resolution %s for video %s in %s" % (
            "%04d" % self.id,
            self.name,
            self.video.id,
            self.encoding_format,
        )

    @property
    def owner(self):
        """Property representing the owner of the video."""
        return self.video.owner

    @property
    def height(self) -> int:
        """Property representing the height of the video rendition."""
        return int(self.rendition.resolution.split("x")[1])

    @property
    def width(self) -> int:
        """Property representing the width of the video rendition."""
        return int(self.rendition.resolution.split("x")[0])

    def delete(self) -> None:
        """Delete the encoding video."""
        if self.source_file:
            if os.path.isfile(self.source_file.path):
                os.remove(self.source_file.path)
        super(EncodingVideo, self).delete()


class EncodingAudio(models.Model):
    """Model representing the encoding audio for a video."""

    name = models.CharField(
        _("Name"),
        max_length=10,
        choices=ENCODING_CHOICES,
        default="audio",
        help_text=_("Please use the only format in encoding choices:")
        + " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
    )
    video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)
    encoding_format = models.CharField(
        _("Format"),
        max_length=22,
        choices=FORMAT_CHOICES,
        default="audio/mp3",
        help_text=_("Please use the only format in format choices:")
        + " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
    )
    source_file = models.FileField(
        _("encoding source file"),
        upload_to=get_storage_path_video,
        max_length=255,
    )

    @property
    def sites(self):
        """Property representing the sites associated with the video."""
        return self.video.sites

    @property
    def sites_all(self):
        """Property representing all the sites associated with the video."""
        return self.video.sites_set.all()

    class Meta:
        ordering = ["name"]
        verbose_name = _("Encoding audio")
        verbose_name_plural = _("Encoding audios")

    def clean(self) -> None:
        """Validate the encoding audio model."""
        if self.name:
            if self.name not in dict(ENCODING_CHOICES):
                raise ValidationError(EncodingAudio._meta.get_field("name").help_text)
        if self.encoding_format:
            if self.encoding_format not in dict(FORMAT_CHOICES):
                raise ValidationError(
                    EncodingAudio._meta.get_field("encoding_format").help_text
                )

    def __str__(self) -> str:
        return "EncodingAudio num: %s for video %s in %s" % (
            "%04d" % self.id,
            self.video.id,
            self.encoding_format,
        )

    @property
    def owner(self):
        """Property representing the owner of the video."""
        return self.video.owner

    def delete(self) -> None:
        """Delete the encoding audio, including the source file if it exists."""
        if self.source_file:
            if os.path.isfile(self.source_file.path):
                os.remove(self.source_file.path)
        super(EncodingAudio, self).delete()


class EncodingLog(models.Model):
    """Model representing the encoding log for a video."""

    video = models.OneToOneField(
        Video, verbose_name=_("Video"), editable=False, on_delete=models.CASCADE
    )
    log = models.TextField(null=True, blank=True, editable=False)
    logfile = models.FileField(max_length=255, blank=True, null=True)

    @property
    def sites(self):
        """Property representing the sites associated with the video."""
        return self.video.sites

    @property
    def sites_all(self):
        """Property representing all the sites associated with the video."""
        return self.video.sites_set.all()

    class Meta:
        ordering = ["video"]
        verbose_name = _("Encoding log")
        verbose_name_plural = _("Encoding logs")

    def __str__(self):
        return "Log for encoding video %s" % (self.video.id)


class EncodingStep(models.Model):
    """Model representing an encoding step for a video."""

    video = models.OneToOneField(
        Video, verbose_name=_("Video"), editable=False, on_delete=models.CASCADE
    )
    num_step = models.IntegerField(default=0, editable=False)
    desc_step = models.CharField(null=True, max_length=255, blank=True, editable=False)

    @property
    def sites(self):
        """Property representing the sites associated with the video."""
        return self.video.sites

    @property
    def sites_all(self):
        """Property representing all the sites associated with the video."""
        return self.video.sites_set.all()

    class Meta:
        ordering = ["video"]
        verbose_name = _("Encoding step")
        verbose_name_plural = _("Encoding steps")

    def __str__(self):
        return "Step for encoding video %s" % (self.video.id)


class PlaylistVideo(models.Model):
    name = models.CharField(
        _("Name"),
        max_length=10,
        choices=ENCODING_CHOICES,
        default="360p",
        help_text=_("Please use the only format in encoding choices:")
        + " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
    )
    video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)
    encoding_format = models.CharField(
        _("Format"),
        max_length=22,
        choices=FORMAT_CHOICES,
        default="application/x-mpegURL",
        help_text=_("Please use the only format in format choices:")
        + " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
    )
    source_file = models.FileField(
        _("encoding source file"),
        upload_to=get_storage_path_video,
        max_length=255,
    )

    class Meta:
        verbose_name = _("Video Playlist")
        verbose_name_plural = _("Video Playlists")

    @property
    def sites(self):
        return self.video.sites

    @property
    def sites_all(self):
        return self.video.sites_set.all()

    def clean(self) -> None:
        """Validate some PlaylistVideomodels fields."""
        if self.name:
            if self.name not in dict(ENCODING_CHOICES):
                raise ValidationError(
                    PlaylistVideo._meta.get_field("name").help_text, code="invalid_name"
                )
        if self.encoding_format:
            if self.encoding_format not in dict(FORMAT_CHOICES):
                raise ValidationError(
                    PlaylistVideo._meta.get_field("encoding_format").help_text,
                    code="invalid_encoding",
                )

    def __str__(self) -> str:
        return "Playlist num: %s for video %s in %s" % (
            "%04d" % self.id,
            self.video.id,
            self.encoding_format,
        )

    @property
    def owner(self):
        return self.video.owner

    def delete(self) -> None:
        if self.source_file:
            if os.path.isfile(self.source_file.path):
                os.remove(self.source_file.path)
        super(PlaylistVideo, self).delete()


class RunnerManager(models.Model):
    """Hold information about runner manager."""

    # Runner manager name
    name = models.CharField(
        max_length=250,
        verbose_name=_("Runner manager name"),
        help_text=_("Runner manager name"),
    )

    # Priority
    priority = models.IntegerField(
        verbose_name=_("Priority"),
        help_text=_(
            "Priority of the runner manager. Lower values indicate higher priority."
        ),
        default=1,
    )

    # Activation status
    is_active = models.BooleanField(
        default=True,
        verbose_name=_("Active"),
        help_text=_("If checked, this runner manager can be used to process tasks."),
    )

    # Runner manager URL
    # Format: https://manager.univ.fr:port/
    url = models.CharField(
        max_length=250,
        verbose_name=_("URL of the runner manager"),
        help_text=_("Example format: %(url)s") % {"url": "https://manager.univ.fr:port/"},
    )

    # Bearer token for the runner manager server (e.g. `6YqG_73xt-9s8v5aBz`)
    token = models.CharField(
        max_length=50,
        verbose_name=_("Bearer token for the runner manager."),
        help_text=_("Example format: 6YqG_73xt-9s8v5aBz"),
    )

    # Site
    site = models.ForeignKey(
        Site,
        verbose_name=_("Site"),
        on_delete=models.CASCADE,
        default=1,
    )

    def __unicode__(self):
        return "%s (%s)" % (self.name, self.site.id)

    def __str__(self):
        return "%s (%s)" % (self.name, self.site.id)

    def save(self, *args, **kwargs):
        super(RunnerManager, self).save(*args, **kwargs)

    class Meta:
        db_table = "runner_manager"
        verbose_name = _("Runner manager")
        verbose_name_plural = _("Runner managers")
        constraints = [
            models.UniqueConstraint(
                fields=["url", "site"],
                name="runner_manager_unique_url_site",
            ),
        ]


@receiver(pre_save, sender=RunnerManager)
def default_site_runner_manager(sender, instance, **kwargs):
    """Save default site for this runner manager."""
    if not hasattr(instance, "site"):
        instance.site = Site.objects.get_current()


class PriorityUser(models.Model):
    """Hold users that should always be dequeued first (priority 0)."""

    user = models.OneToOneField(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        verbose_name=_("User"),
        help_text=_("User with absolute task queue priority (priority 0)."),
    )
    date_added = models.DateTimeField(
        verbose_name=_("Date added"), default=timezone.now, editable=False
    )

    def __str__(self):
        return str(self.user)

    class Meta:
        verbose_name = _("Priority user")
        verbose_name_plural = _("Priority users")
        ordering = ["user__username", "id"]


class Task(models.Model):
    """Hold information about tasks managed by the runner managers."""

    # Task type
    TYPE = (
        ("encoding", _("Encoding task")),
        ("studio", _("Studio task")),
        ("transcription", _("Transcription task")),
    )
    type = models.CharField(
        max_length=30,
        verbose_name=_("Task type"),
        choices=TYPE,
        default=TYPE[0][0],
    )

    # Task status
    STATUS = (
        ("pending", _("Task pending")),
        ("running", _("Task in progress")),
        ("completed", _("Task completed")),
        ("failed", _("Task failed")),
        ("timeout", _("Task timeouted")),
    )
    status = models.CharField(
        max_length=30,
        verbose_name=_("Task status"),
        choices=STATUS,
        default=STATUS[0][0],
    )
    # Task identifier from runner manager
    task_id = models.CharField(
        max_length=100,
        verbose_name=_("Task identifier from runner manager"),
        help_text=_("Identifier of the task provided by the runner manager"),
        null=True,
        blank=True,
    )

    # Video associated to the task
    video = models.ForeignKey(
        Video,
        on_delete=models.CASCADE,
        verbose_name=_("Video"),
        help_text=_("Video associated to the task"),
        null=True,
        blank=True,
    )

    # Recording associated to the task (for Studio tasks)
    recording = models.ForeignKey(
        Recording,
        on_delete=models.CASCADE,
        verbose_name=_("Recording"),
        help_text=_("Studio recording associated to the task"),
        null=True,
        blank=True,
    )

    # Runner manager that manages this task
    runner_manager = models.ForeignKey(
        RunnerManager,
        on_delete=models.CASCADE,
        verbose_name=_("Runner manager that manages this task"),
        help_text=_("Runner manager that achieves this task"),
        null=True,
        blank=True,
    )

    # Date task added
    date_added = models.DateTimeField(
        verbose_name=_("Date added"), default=timezone.now, editable=False
    )

    # Queue rank for pending tasks
    rank = models.IntegerField(
        verbose_name=_("Queue rank"),
        help_text=_("Rank of the task in the pending queue"),
        null=True,
        blank=True,
        default=None,
    )

    # Script output
    script_output = models.TextField(
        verbose_name=_("Script output"),
        help_text=_("Output from the runner manager script"),
        null=True,
        blank=True,
    )

    def __unicode__(self):
        ref = (
            self.video.id
            if self.video_id
            else (self.recording.id if self.recording_id else "-")
        )
        return "%s - %s - %s" % (ref, self.type, self.status)

    def __str__(self):
        ref = (
            self.video.id
            if self.video_id
            else (self.recording.id if self.recording_id else "-")
        )
        return "%s - %s - %s" % (ref, self.type, self.status)

    def save(self, *args, **kwargs):
        super(Task, self).save(*args, **kwargs)

    class Meta:
        db_table = "runner_manager_task"
        verbose_name = _("Task")
        verbose_name_plural = _("Tasks")
        ordering = ["id"]

1	"""Models for Esup-Pod video_encode."""
2
3	import os	1✔
4
5	from django.conf import settings	1✔
6	from django.contrib.sites.models import Site	1✔
7	from django.core.exceptions import ValidationError	1✔
8	from django.core.validators import MaxValueValidator	1✔
9	from django.db import models	1✔
10	from django.db.models.signals import post_save, pre_save	1✔
11	from django.dispatch import receiver	1✔
12	from django.utils import timezone	1✔
13	from django.utils.translation import gettext_lazy as _	1✔
14
15	from pod.recorder.models import Recording	1✔
16	from pod.video.models import Video, get_storage_path_video	1✔
17
18	ENCODING_CHOICES = getattr(	1✔
19	settings,
20	"ENCODING_CHOICES",
21	(
22	("audio", "audio"),
23	("360p", "360p"),
24	("480p", "480p"),
25	("720p", "720p"),
26	("1080p", "1080p"),
27	("playlist", "playlist"),
28	),
29	)
30
31	FORMAT_CHOICES = getattr(	1✔
32	settings,
33	"FORMAT_CHOICES",
34	(
35	("video/mp4", "video/mp4"),
36	("video/mp2t", "video/mp2t"),
37	("video/webm", "video/webm"),
38	("audio/mp3", "audio/mp3"),
39	("audio/wav", "audio/wav"),
40	("application/x-mpegURL", "application/x-mpegURL"),
41	),
42	)
43
44	SITE_ID = getattr(settings, "SITE_ID", 1)	1✔
45
46
47	class VideoRendition(models.Model):	1✔
48	"""Model representing the rendition video."""
49
50	resolution = models.CharField(	1✔
51	_("resolution"),
52	max_length=50,
53	unique=True,
54	help_text=_(
55	"Please use the only format x. i.e.: "
56	+ "<em>640x360</em> or <em>1280x720</em>"
57	+ " or <em>1920x1080</em>."
58	),
59	)
60	minrate = models.CharField(	1✔
61	_("minrate"),
62	max_length=50,
63	help_text=_(
64	"Please use the only format k. i.e.: "
65	+ "<em>300k</em> or <em>600k</em>"
66	+ " or <em>1000k</em>."
67	),
68	)
69	video_bitrate = models.CharField(	1✔
70	_("bitrate video"),
71	max_length=50,
72	help_text=_(
73	"Please use the only format k. i.e.: "
74	+ "<em>300k</em> or <em>600k</em>"
75	+ " or <em>1000k</em>."
76	),
77	)
78	maxrate = models.CharField(	1✔
79	_("maxrate"),
80	max_length=50,
81	help_text=_(
82	"Please use the only format k. i.e.: "
83	+ "<em>300k</em> or <em>600k</em>"
84	+ " or <em>1000k</em>."
85	),
86	)
87	encoding_resolution_threshold = models.PositiveIntegerField(	1✔
88	_("encoding resolution threshold"),
89	default=0,
90	validators=[MaxValueValidator(100)],
91	)
92	audio_bitrate = models.CharField(	1✔
93	_("bitrate audio"),
94	max_length=50,
95	help_text=_(
96	"Please use the only format k. i.e.: "
97	+ "<em>300k</em> or <em>600k</em>"
98	+ " or <em>1000k</em>."
99	),
100	)
101	encode_mp4 = models.BooleanField(_("Make a MP4 version"), default=False)	1✔
102	sites = models.ManyToManyField(Site)	1✔
103
104	@property	1✔
105	def height(self) -> int:	1✔
106	"""The height of the video rendition based on the resolution."""
107	return int(self.resolution.split("x")[1])	×
108
109	@property	1✔
110	def width(self) -> int:	1✔
111	"""The width of the video rendition based on the resolution."""
112	return int(self.resolution.split("x")[0])	×
113
114	class Meta:	1✔
115	# ordering = ['height'] # Not work
116	verbose_name = _("rendition")	1✔
117	verbose_name_plural = _("renditions")	1✔
118
119	def __str__(self) -> str:	1✔
120	return "VideoRendition num %s with resolution %s" % (	1✔
121	"%04d" % self.id,
122	self.resolution,
123	)
124
125	def bitrate(self, field_value, field_name, name=None) -> None:	1✔
126	"""Validate the bitrate field value."""
127	if name is None:	1✔
128	name = field_name	1✔
129	if field_value and "k" not in field_value:	1✔
130	msg = "Error in %s: " % _(name)	1✔
131	raise ValidationError(	1✔
132	"%s %s" % (msg, VideoRendition._meta.get_field(field_name).help_text)
133	)
134	else:
135	vb = field_value.replace("k", "")	1✔
136	if not vb.isdigit():	1✔
137	msg = "Error in %s: " % _(name)	1✔
138	raise ValidationError(	1✔
139	"%s %s" % (msg, VideoRendition._meta.get_field(field_name).help_text)
140	)
141
142	def clean_bitrate(self) -> None:	1✔
143	"""Clean the bitrate-related fields."""
144	self.bitrate(self.video_bitrate, "video_bitrate", "bitrate video")	1✔
145	self.bitrate(self.maxrate, "maxrate")	1✔
146	self.bitrate(self.minrate, "minrate")	1✔
147
148	def clean(self) -> None:	1✔
149	"""Clean the fields of the VideoRendition model."""
150	if self.resolution and "x" not in self.resolution:	1✔
151	raise ValidationError(VideoRendition._meta.get_field("resolution").help_text)	1✔
152	else:
153	res = self.resolution.replace("x", "")	1✔
154	if not res.isdigit():	1✔
155	raise ValidationError(	1✔
156	VideoRendition._meta.get_field("resolution").help_text
157	)
158
159	self.clean_bitrate()	1✔
160	self.bitrate(self.audio_bitrate, "audio_bitrate", "bitrate audio")	1✔
161
162
163	@receiver(post_save, sender=VideoRendition)	1✔
164	def default_site_videorendition(sender, instance, created: bool, **kwargs) -> None:	1✔
165	"""Add the current site as a default site."""
166	if instance.sites.count() == 0:	1✔
167	instance.sites.add(Site.objects.get_current())	1✔
168
169
170	class EncodingVideo(models.Model):	1✔
171	"""Model representing the encoding video for a video."""
172
173	name = models.CharField(	1✔
174	_("Name"),
175	max_length=10,
176	choices=ENCODING_CHOICES,
177	default="360p",
178	help_text=_("Please use the only format in encoding choices:")
179	+ " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
180	)
181	video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)	1✔
182	rendition = models.ForeignKey(	1✔
183	VideoRendition, verbose_name=_("rendition"), on_delete=models.CASCADE
184	)
185	encoding_format = models.CharField(	1✔
186	_("Format"),
187	max_length=22,
188	choices=FORMAT_CHOICES,
189	default="video/mp4",
190	help_text=_("Please use the only format in format choices:")
191	+ " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
192	)
193	source_file = models.FileField(	1✔
194	_("encoding source file"),
195	upload_to=get_storage_path_video,
196	max_length=255,
197	)
198
199	@property	1✔
200	def sites(self):	1✔
201	"""Property representing the sites associated with the video."""
202	return self.video.sites	×
203
204	@property	1✔
205	def sites_all(self):	1✔
206	"""Property representing all the sites associated with the video."""
207	return self.video.sites_set.all()	×
208
209	def clean(self) -> None:	1✔
210	"""Validate the encoding video model."""
211	if self.name:	1✔
212	if self.name not in dict(ENCODING_CHOICES):	1✔
213	raise ValidationError(EncodingVideo._meta.get_field("name").help_text)	1✔
214	if self.encoding_format:	1✔
215	if self.encoding_format not in dict(FORMAT_CHOICES):	1✔
216	raise ValidationError(	×
217	EncodingVideo._meta.get_field("encoding_format").help_text
218	)
219
220	class Meta:	1✔
221	ordering = ["name"]	1✔
222	verbose_name = _("Encoding video")	1✔
223	verbose_name_plural = _("Encoding videos")	1✔
224
225	def __str__(self) -> str:	1✔
226	return "EncodingVideo num: %s with resolution %s for video %s in %s" % (	1✔
227	"%04d" % self.id,
228	self.name,
229	self.video.id,
230	self.encoding_format,
231	)
232
233	@property	1✔
234	def owner(self):	1✔
235	"""Property representing the owner of the video."""
236	return self.video.owner	1✔
237
238	@property	1✔
239	def height(self) -> int:	1✔
240	"""Property representing the height of the video rendition."""
241	return int(self.rendition.resolution.split("x")[1])	×
242
243	@property	1✔
244	def width(self) -> int:	1✔
245	"""Property representing the width of the video rendition."""
246	return int(self.rendition.resolution.split("x")[0])	×
247
248	def delete(self) -> None:	1✔
249	"""Delete the encoding video."""
250	if self.source_file:	1✔
251	if os.path.isfile(self.source_file.path):	1✔
252	os.remove(self.source_file.path)	1✔
253	super(EncodingVideo, self).delete()	1✔
254
255
256	class EncodingAudio(models.Model):	1✔
257	"""Model representing the encoding audio for a video."""
258
259	name = models.CharField(	1✔
260	_("Name"),
261	max_length=10,
262	choices=ENCODING_CHOICES,
263	default="audio",
264	help_text=_("Please use the only format in encoding choices:")
265	+ " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
266	)
267	video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)	1✔
268	encoding_format = models.CharField(	1✔
269	_("Format"),
270	max_length=22,
271	choices=FORMAT_CHOICES,
272	default="audio/mp3",
273	help_text=_("Please use the only format in format choices:")
274	+ " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
275	)
276	source_file = models.FileField(	1✔
277	_("encoding source file"),
278	upload_to=get_storage_path_video,
279	max_length=255,
280	)
281
282	@property	1✔
283	def sites(self):	1✔
284	"""Property representing the sites associated with the video."""
285	return self.video.sites	×
286
287	@property	1✔
288	def sites_all(self):	1✔
289	"""Property representing all the sites associated with the video."""
290	return self.video.sites_set.all()	×
291
292	class Meta:	1✔
293	ordering = ["name"]	1✔
294	verbose_name = _("Encoding audio")	1✔
295	verbose_name_plural = _("Encoding audios")	1✔
296
297	def clean(self) -> None:	1✔
298	"""Validate the encoding audio model."""
299	if self.name:	1✔
300	if self.name not in dict(ENCODING_CHOICES):	1✔
301	raise ValidationError(EncodingAudio._meta.get_field("name").help_text)	1✔
302	if self.encoding_format:	1✔
303	if self.encoding_format not in dict(FORMAT_CHOICES):	1✔
304	raise ValidationError(	×
305	EncodingAudio._meta.get_field("encoding_format").help_text
306	)
307
308	def __str__(self) -> str:	1✔
309	return "EncodingAudio num: %s for video %s in %s" % (	1✔
310	"%04d" % self.id,
311	self.video.id,
312	self.encoding_format,
313	)
314
315	@property	1✔
316	def owner(self):	1✔
317	"""Property representing the owner of the video."""
318	return self.video.owner	1✔
319
320	def delete(self) -> None:	1✔
321	"""Delete the encoding audio, including the source file if it exists."""
322	if self.source_file:	1✔
323	if os.path.isfile(self.source_file.path):	1✔
324	os.remove(self.source_file.path)	1✔
325	super(EncodingAudio, self).delete()	1✔
326
327
328	class EncodingLog(models.Model):	1✔
329	"""Model representing the encoding log for a video."""
330
331	video = models.OneToOneField(	1✔
332	Video, verbose_name=_("Video"), editable=False, on_delete=models.CASCADE
333	)
334	log = models.TextField(null=True, blank=True, editable=False)	1✔
335	logfile = models.FileField(max_length=255, blank=True, null=True)	1✔
336
337	@property	1✔
338	def sites(self):	1✔
339	"""Property representing the sites associated with the video."""
340	return self.video.sites	×
341
342	@property	1✔
343	def sites_all(self):	1✔
344	"""Property representing all the sites associated with the video."""
345	return self.video.sites_set.all()	×
346
347	class Meta:	1✔
348	ordering = ["video"]	1✔
349	verbose_name = _("Encoding log")	1✔
350	verbose_name_plural = _("Encoding logs")	1✔
351
352	def __str__(self):	1✔
353	return "Log for encoding video %s" % (self.video.id)	1✔
354
355
356	class EncodingStep(models.Model):	1✔
357	"""Model representing an encoding step for a video."""
358
359	video = models.OneToOneField(	1✔
360	Video, verbose_name=_("Video"), editable=False, on_delete=models.CASCADE
361	)
362	num_step = models.IntegerField(default=0, editable=False)	1✔
363	desc_step = models.CharField(null=True, max_length=255, blank=True, editable=False)	1✔
364
365	@property	1✔
366	def sites(self):	1✔
367	"""Property representing the sites associated with the video."""
368	return self.video.sites	×
369
370	@property	1✔
371	def sites_all(self):	1✔
372	"""Property representing all the sites associated with the video."""
373	return self.video.sites_set.all()	×
374
375	class Meta:	1✔
376	ordering = ["video"]	1✔
377	verbose_name = _("Encoding step")	1✔
378	verbose_name_plural = _("Encoding steps")	1✔
379
380	def __str__(self):	1✔
381	return "Step for encoding video %s" % (self.video.id)	1✔
382
383
384	class PlaylistVideo(models.Model):	1✔
385	name = models.CharField(	1✔
386	_("Name"),
387	max_length=10,
388	choices=ENCODING_CHOICES,
389	default="360p",
390	help_text=_("Please use the only format in encoding choices:")
391	+ " %s" % " ".join(str(key) for key, value in ENCODING_CHOICES),
392	)
393	video = models.ForeignKey(Video, verbose_name=_("Video"), on_delete=models.CASCADE)	1✔
394	encoding_format = models.CharField(	1✔
395	_("Format"),
396	max_length=22,
397	choices=FORMAT_CHOICES,
398	default="application/x-mpegURL",
399	help_text=_("Please use the only format in format choices:")
400	+ " %s" % " ".join(str(key) for key, value in FORMAT_CHOICES),
401	)
402	source_file = models.FileField(	1✔
403	_("encoding source file"),
404	upload_to=get_storage_path_video,
405	max_length=255,
406	)
407
408	class Meta:	1✔
409	verbose_name = _("Video Playlist")	1✔
410	verbose_name_plural = _("Video Playlists")	1✔
411
412	@property	1✔
413	def sites(self):	1✔
414	return self.video.sites	×
415
416	@property	1✔
417	def sites_all(self):	1✔
418	return self.video.sites_set.all()	×
419
420	def clean(self) -> None:	1✔
421	"""Validate some PlaylistVideomodels fields."""
422	if self.name:	1✔
423	if self.name not in dict(ENCODING_CHOICES):	1✔
424	raise ValidationError(	1✔
425	PlaylistVideo._meta.get_field("name").help_text, code="invalid_name"
426	)
427	if self.encoding_format:	1✔
428	if self.encoding_format not in dict(FORMAT_CHOICES):	1✔
429	raise ValidationError(	×
430	PlaylistVideo._meta.get_field("encoding_format").help_text,
431	code="invalid_encoding",
432	)
433
434	def __str__(self) -> str:	1✔
435	return "Playlist num: %s for video %s in %s" % (	1✔
436	"%04d" % self.id,
437	self.video.id,
438	self.encoding_format,
439	)
440
441	@property	1✔
442	def owner(self):	1✔
443	return self.video.owner	1✔
444
445	def delete(self) -> None:	1✔
446	if self.source_file:	1✔
447	if os.path.isfile(self.source_file.path):	1✔
448	os.remove(self.source_file.path)	1✔
449	super(PlaylistVideo, self).delete()	1✔
450
451
452	class RunnerManager(models.Model):	1✔
453	"""Hold information about runner manager."""
454
455	# Runner manager name
456	name = models.CharField(	1✔
457	max_length=250,
458	verbose_name=_("Runner manager name"),
459	help_text=_("Runner manager name"),
460	)
461
462	# Priority
463	priority = models.IntegerField(	1✔
464	verbose_name=_("Priority"),
465	help_text=_(
466	"Priority of the runner manager. Lower values indicate higher priority."
467	),
468	default=1,
469	)
470
471	# Activation status
472	is_active = models.BooleanField(	1✔
473	default=True,
474	verbose_name=_("Active"),
475	help_text=_("If checked, this runner manager can be used to process tasks."),
476	)
477
478	# Runner manager URL
479	# Format: https://manager.univ.fr:port/
480	url = models.CharField(	1✔
481	max_length=250,
482	verbose_name=_("URL of the runner manager"),
483	help_text=_("Example format: %(url)s") % {"url": "https://manager.univ.fr:port/"},
484	)
485
486	# Bearer token for the runner manager server (e.g. `6YqG_73xt-9s8v5aBz`)
487	token = models.CharField(	1✔
488	max_length=50,
489	verbose_name=_("Bearer token for the runner manager."),
490	help_text=_("Example format: 6YqG_73xt-9s8v5aBz"),
491	)
492
493	# Site
494	site = models.ForeignKey(	1✔
495	Site,
496	verbose_name=_("Site"),
497	on_delete=models.CASCADE,
498	default=1,
499	)
500
501	def __unicode__(self):	1✔
502	return "%s (%s)" % (self.name, self.site.id)	×
503
504	def __str__(self):	1✔
505	return "%s (%s)" % (self.name, self.site.id)	1✔
506
507	def save(self, args, *kwargs):	1✔
508	super(RunnerManager, self).save(args, *kwargs)	1✔
509
510	class Meta:	1✔
511	db_table = "runner_manager"	1✔
512	verbose_name = _("Runner manager")	1✔
513	verbose_name_plural = _("Runner managers")	1✔
514	constraints = [	1✔
515	models.UniqueConstraint(
516	fields=["url", "site"],
517	name="runner_manager_unique_url_site",
518	),
519	]
520
521
522	@receiver(pre_save, sender=RunnerManager)	1✔
523	def default_site_runner_manager(sender, instance, **kwargs):	1✔
524	"""Save default site for this runner manager."""
525	if not hasattr(instance, "site"):	1✔
526	instance.site = Site.objects.get_current()	×
527
528
529	class PriorityUser(models.Model):	1✔
530	"""Hold users that should always be dequeued first (priority 0)."""
531
532	user = models.OneToOneField(	1✔
533	settings.AUTH_USER_MODEL,
534	on_delete=models.CASCADE,
535	verbose_name=_("User"),
536	help_text=_("User with absolute task queue priority (priority 0)."),
537	)
538	date_added = models.DateTimeField(	1✔
539	verbose_name=_("Date added"), default=timezone.now, editable=False
540	)
541
542	def __str__(self):	1✔
NEW 543	return str(self.user)	×
544
545	class Meta:	1✔
546	verbose_name = _("Priority user")	1✔
547	verbose_name_plural = _("Priority users")	1✔
548	ordering = ["user__username", "id"]	1✔
549
550
551	class Task(models.Model):	1✔
552	"""Hold information about tasks managed by the runner managers."""
553
554	# Task type
555	TYPE = (	1✔
556	("encoding", _("Encoding task")),
557	("studio", _("Studio task")),
558	("transcription", _("Transcription task")),
559	)
560	type = models.CharField(	1✔
561	max_length=30,
562	verbose_name=_("Task type"),
563	choices=TYPE,
564	default=TYPE[0][0],
565	)
566
567	# Task status
568	STATUS = (	1✔
569	("pending", _("Task pending")),
570	("running", _("Task in progress")),
571	("completed", _("Task completed")),
572	("failed", _("Task failed")),
573	("timeout", _("Task timeouted")),
574	)
575	status = models.CharField(	1✔
576	max_length=30,
577	verbose_name=_("Task status"),
578	choices=STATUS,
579	default=STATUS[0][0],
580	)
581	# Task identifier from runner manager
582	task_id = models.CharField(	1✔
583	max_length=100,
584	verbose_name=_("Task identifier from runner manager"),
585	help_text=_("Identifier of the task provided by the runner manager"),
586	null=True,
587	blank=True,
588	)
589
590	# Video associated to the task
591	video = models.ForeignKey(	1✔
592	Video,
593	on_delete=models.CASCADE,
594	verbose_name=_("Video"),
595	help_text=_("Video associated to the task"),
596	null=True,
597	blank=True,
598	)
599
600	# Recording associated to the task (for Studio tasks)
601	recording = models.ForeignKey(	1✔
602	Recording,
603	on_delete=models.CASCADE,
604	verbose_name=_("Recording"),
605	help_text=_("Studio recording associated to the task"),
606	null=True,
607	blank=True,
608	)
609
610	# Runner manager that manages this task
611	runner_manager = models.ForeignKey(	1✔
612	RunnerManager,
613	on_delete=models.CASCADE,
614	verbose_name=_("Runner manager that manages this task"),
615	help_text=_("Runner manager that achieves this task"),
616	null=True,
617	blank=True,
618	)
619
620	# Date task added
621	date_added = models.DateTimeField(	1✔
622	verbose_name=_("Date added"), default=timezone.now, editable=False
623	)
624
625	# Queue rank for pending tasks
626	rank = models.IntegerField(	1✔
627	verbose_name=_("Queue rank"),
628	help_text=_("Rank of the task in the pending queue"),
629	null=True,
630	blank=True,
631	default=None,
632	)
633
634	# Script output
635	script_output = models.TextField(	1✔
636	verbose_name=_("Script output"),
637	help_text=_("Output from the runner manager script"),
638	null=True,
639	blank=True,
640	)
641
642	def __unicode__(self):	1✔
643	ref = (	×
644	self.video.id
645	if self.video_id
646	else (self.recording.id if self.recording_id else "-")
647	)
648	return "%s - %s - %s" % (ref, self.type, self.status)	×
649
650	def __str__(self):	1✔
651	ref = (	×
652	self.video.id
653	if self.video_id
654	else (self.recording.id if self.recording_id else "-")
655	)
656	return "%s - %s - %s" % (ref, self.type, self.status)	×
657
658	def save(self, args, *kwargs):	1✔
659	super(Task, self).save(args, *kwargs)	1✔
660
661	class Meta:	1✔
662	db_table = "runner_manager_task"	1✔
663	verbose_name = _("Task")	1✔
664	verbose_name_plural = _("Tasks")	1✔
665	ordering = ["id"]	1✔

EsupPortail / Esup-Pod / 27555111584

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