28237579315

Committed 26 Jun 2026 12:19PM UTC coverage: 69.176%. First build

Build # 28237579315

Build Type

Pull #1427

github

Committed by

web-flow

Commit Message

Fix FFMPEG commands by adding spaces (#1472)

Avoid errors like :

```
Error opening input: No such file or directory
Error opening input file /data/www/poduser/media/videos/xxx/yyy/360p.mp4-vf.
```

(here, the `-vf` corrrespond to options given to FFMPEG command)

Pull Request Pull Request #1427: [Release] 4.3.0

Coverage Stats

798 of 1112 new or added lines in 24 files covered. (71.76%)

13676 of 19770 relevant lines covered (69.18%)

0.69 hits per line

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

85.19

/pod/video_encode_transcript/task_queue.py

"""Queue ranking helpers for encoding/transcription task dispatch in Esup-Pod.

The queue is intentionally simple:
- only pending tasks receive a rank,
- rank is recomputed from scratch to keep ordering deterministic,
- non-pending tasks must not retain stale rank values.
"""

import logging
from typing import TypeAlias

from pod.video.models import Video

from .models import PriorityUser, Task

log = logging.getLogger(__name__)

QueuePriority: TypeAlias = int
HIGH_PRIORITY: QueuePriority = 0
NORMAL_PRIORITY: QueuePriority = 1
LOW_PRIORITY: QueuePriority = 2


def get_priority_user_ids() -> set[int]:
    """Return the set of users configured with absolute queue priority."""
    return set(PriorityUser.objects.values_list("user_id", flat=True))


def get_user_priority(
    video: Video,
    priority_user_ids: set[int] | None = None,
) -> QueuePriority:
    """Return queue priority for a video owner."""
    try:
        owner_id = getattr(video, "owner_id", None)
        if owner_id:
            if priority_user_ids is None:
                if PriorityUser.objects.filter(user_id=owner_id).exists():
                    return HIGH_PRIORITY
            elif owner_id in priority_user_ids:
                return HIGH_PRIORITY

        affiliation = video.owner.owner.affiliation
        if affiliation == "student":
            return LOW_PRIORITY
        return NORMAL_PRIORITY
    except AttributeError:
        log.warning(
            "Cannot determine affiliation for video %s owner. Defaulting to normal priority.",
            video.id,
        )
        return NORMAL_PRIORITY
    except Exception as exc:
        log.warning(
            "Error getting affiliation for video %s: %s. Defaulting to normal priority.",
            video.id,
            str(exc),
        )
        return NORMAL_PRIORITY


def _task_priority(
    task: Task,
    priority_user_ids: set[int] | None = None,
) -> QueuePriority:
    """Return priority for a pending task."""
    if not task.video_id or not task.video:
        return NORMAL_PRIORITY
    return get_user_priority(task.video, priority_user_ids=priority_user_ids)


def get_sorted_pending_tasks() -> list[Task]:
    """Return all pending tasks sorted by queue priority and creation date."""
    priority_user_ids = get_priority_user_ids()
    pending_tasks = list(
        Task.objects.filter(status="pending")
        .select_related("video", "video__owner", "video__owner__owner")
        .order_by("date_added", "id")
    )
    pending_tasks.sort(
        key=lambda task: (
            _task_priority(task, priority_user_ids=priority_user_ids),
            task.date_added,
            task.id,
        ),
    )
    return pending_tasks


def refresh_pending_task_ranks() -> None:
    """Recalculate rank for all pending tasks and clear rank for non-pending ones."""
    sorted_tasks = get_sorted_pending_tasks()
    updates: list[Task] = []
    for index, task in enumerate(sorted_tasks, start=1):
        if task.rank != index:
            task.rank = index
            updates.append(task)

    if updates:
        # Batch write only changed rows to avoid unnecessary UPDATE queries.
        Task.objects.bulk_update(updates, ["rank"])

    # Keep DB consistent: only pending tasks are expected to have a queue rank.
    Task.objects.exclude(status="pending").exclude(rank__isnull=True).update(rank=None)


def get_video_pending_encoding_queue_info(video: Video) -> tuple[int | None, int]:
    """Return (rank, total_pending) for the encoding task linked to a video."""
    queue_total = Task.objects.filter(status="pending").count()
    queue_task = (
        Task.objects.filter(video_id=video.id, type="encoding", status="pending")
        .order_by("date_added", "id")
        .first()
    )
    queue_rank = queue_task.rank if queue_task else None
    return queue_rank, queue_total

1	"""Queue ranking helpers for encoding/transcription task dispatch in Esup-Pod.
2
3	The queue is intentionally simple:
4	- only pending tasks receive a rank,
5	- rank is recomputed from scratch to keep ordering deterministic,
6	- non-pending tasks must not retain stale rank values.
7	"""
8
9	import logging	1✔
10	from typing import TypeAlias	1✔
11
12	from pod.video.models import Video	1✔
13
14	from .models import PriorityUser, Task	1✔
15
16	log = logging.getLogger(__name__)	1✔
17
18	QueuePriority: TypeAlias = int	1✔
19	HIGH_PRIORITY: QueuePriority = 0	1✔
20	NORMAL_PRIORITY: QueuePriority = 1	1✔
21	LOW_PRIORITY: QueuePriority = 2	1✔
22
23
24	def get_priority_user_ids() -> set[int]:	1✔
25	"""Return the set of users configured with absolute queue priority."""
26	return set(PriorityUser.objects.values_list("user_id", flat=True))	1✔
27
28
29	def get_user_priority(	1✔
30	video: Video,
31	priority_user_ids: set[int] \| None = None,
32	) -> QueuePriority:
33	"""Return queue priority for a video owner."""
34	try:	1✔
35	owner_id = getattr(video, "owner_id", None)	1✔
36	if owner_id:	1✔
37	if priority_user_ids is None:	1✔
NEW 38	if PriorityUser.objects.filter(user_id=owner_id).exists():	×
NEW 39	return HIGH_PRIORITY	×
40	elif owner_id in priority_user_ids:	1✔
41	return HIGH_PRIORITY	1✔
42
43	affiliation = video.owner.owner.affiliation	1✔
44	if affiliation == "student":	1✔
45	return LOW_PRIORITY	1✔
46	return NORMAL_PRIORITY	1✔
47	except AttributeError:	×
48	log.warning(	×
49	"Cannot determine affiliation for video %s owner. Defaulting to normal priority.",
50	video.id,
51	)
NEW 52	return NORMAL_PRIORITY	×
53	except Exception as exc:	×
54	log.warning(	×
55	"Error getting affiliation for video %s: %s. Defaulting to normal priority.",
56	video.id,
57	str(exc),
58	)
NEW 59	return NORMAL_PRIORITY	×
60
61
62	def _task_priority(	1✔
63	task: Task,
64	priority_user_ids: set[int] \| None = None,
65	) -> QueuePriority:
66	"""Return priority for a pending task."""
67	if not task.video_id or not task.video:	1✔
68	return NORMAL_PRIORITY	1✔
69	return get_user_priority(task.video, priority_user_ids=priority_user_ids)	1✔
70
71
72	def get_sorted_pending_tasks() -> list[Task]:	1✔
73	"""Return all pending tasks sorted by queue priority and creation date."""
74	priority_user_ids = get_priority_user_ids()	1✔
75	pending_tasks = list(	1✔
76	Task.objects.filter(status="pending")
77	.select_related("video", "video__owner", "video__owner__owner")
78	.order_by("date_added", "id")
79	)
80	pending_tasks.sort(	1✔
81	key=lambda task: (
82	_task_priority(task, priority_user_ids=priority_user_ids),
83	task.date_added,
84	task.id,
85	),
86	)
87	return pending_tasks	1✔
88
89
90	def refresh_pending_task_ranks() -> None:	1✔
91	"""Recalculate rank for all pending tasks and clear rank for non-pending ones."""
92	sorted_tasks = get_sorted_pending_tasks()	1✔
93	updates: list[Task] = []	1✔
94	for index, task in enumerate(sorted_tasks, start=1):	1✔
95	if task.rank != index:	1✔
96	task.rank = index	1✔
97	updates.append(task)	1✔
98
99	if updates:	1✔
100	# Batch write only changed rows to avoid unnecessary UPDATE queries.
101	Task.objects.bulk_update(updates, ["rank"])	1✔
102
103	# Keep DB consistent: only pending tasks are expected to have a queue rank.
104	Task.objects.exclude(status="pending").exclude(rank__isnull=True).update(rank=None)	1✔
105
106
107	def get_video_pending_encoding_queue_info(video: Video) -> tuple[int \| None, int]:	1✔
108	"""Return (rank, total_pending) for the encoding task linked to a video."""
109	queue_total = Task.objects.filter(status="pending").count()	1✔
110	queue_task = (	1✔
111	Task.objects.filter(video_id=video.id, type="encoding", status="pending")
112	.order_by("date_added", "id")
113	.first()
114	)
115	queue_rank = queue_task.rank if queue_task else None	1✔
116	return queue_rank, queue_total	1✔

EsupPortail / Esup-Pod / 28237579315

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