25740886384

Committed 12 May 2026 02:25PM UTC coverage: 69.122%. First build

Build # 25740886384

Build Type

Pull #1427

github

Committed by

web-flow

Commit Message

Bump urllib3 from 2.6.3 to 2.7.0 (#1444)

Bumps [urllib3](https://github.com/urllib3/urllib3) from 2.6.3 to 2.7.0.
- [Release notes](https://github.com/urllib3/urllib3/releases)
- [Changelog](https://github.com/urllib3/urllib3/blob/main/CHANGES.rst)
- [Commits](https://github.com/urllib3/urllib3/compare/2.6.3...2.7.0)

---
updated-dependencies:
- dependency-name: urllib3
  dependency-version: 2.7.0
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>

Pull Request Pull Request #1427: [Release] 4.3.0

Coverage Stats

173 of 199 new or added lines in 6 files covered. (86.93%)

13203 of 19101 relevant lines covered (69.12%)

0.69 hits per line

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

99.03

/pod/video_encode_transcript/runner_manager.py

"""Runner Manager orchestration helpers for encoding and transcription tasks in Esup-Pod.

This module builds task payloads, dispatches them to available runner managers,
and keeps local task rows synchronized with runner-side task status.
"""

import json
import logging
import os
from typing import Any, Literal, Optional, TypeAlias, TypedDict, Union, cast

import requests
from django.conf import settings
from django.contrib.sites.models import Site
from django.shortcuts import get_object_or_404
from django.utils.translation import gettext_lazy as _
from pod.cut.models import CutVideo
from pod.recorder.models import Recording
from pod.video.models import Video
from pod.video_encode_transcript.models import RunnerManager, Task
from pod.video_encode_transcript.runner_manager_utils import (
    store_before_remote_encoding_recording,
    store_before_remote_encoding_video,
)

from .utils import change_encoding_step

if __name__ == "__main__":
    from encoding_utils import get_list_rendition
else:
    from .encoding_utils import get_list_rendition

log = logging.getLogger(__name__)

DEBUG = getattr(settings, "DEBUG", True)

# Settings for template customization
TEMPLATE_VISIBLE_SETTINGS = getattr(
    settings,
    "TEMPLATE_VISIBLE_SETTINGS",
    {
        "TITLE_SITE": "Pod",
        "TITLE_ETB": "University name",
        "LOGO_SITE": "img/logoPod.svg",
        "LOGO_ETB": "img/esup-pod.svg",
        "LOGO_PLAYER": "img/pod_favicon.svg",
        "LINK_PLAYER": "",
        "LINK_PLAYER_NAME": _("Home"),
        "FOOTER_TEXT": ("",),
        "FAVICON": "img/pod_favicon.svg",
        "CSS_OVERRIDE": "",
        "PRE_HEADER_TEMPLATE": "",
        "POST_FOOTER_TEMPLATE": "",
        "TRACKING_TEMPLATE": "",
    },
)
__TITLE_SITE__ = (
    TEMPLATE_VISIBLE_SETTINGS["TITLE_SITE"]
    if (TEMPLATE_VISIBLE_SETTINGS.get("TITLE_SITE"))
    else "Pod"
)
__TITLE_ETB__ = (
    TEMPLATE_VISIBLE_SETTINGS["TITLE_ETB"]
    if (TEMPLATE_VISIBLE_SETTINGS.get("TITLE_ETB"))
    else "University name"
)

VERSION = getattr(settings, "VERSION", "4.X")

SECURE_SSL_REDIRECT = getattr(settings, "SECURE_SSL_REDIRECT", False)


SourceType = Literal["video", "recording"]
TaskType = Literal["encoding", "studio", "transcription"]
ParametersDict: TypeAlias = dict[str, Any]
HeadersDict: TypeAlias = dict[str, str]


class RunnerManagerTaskPayload(TypedDict):
    """Task payload expected by the Runner Manager API."""

    etab_name: str
    app_name: str
    app_version: str
    task_type: TaskType
    source_url: str
    notify_url: str
    parameters: ParametersDict


class RunnerManagerResponse(TypedDict, total=False):
    """Relevant fields returned by the Runner Manager API."""

    task_id: str
    status: str


def _build_rendition_parameters() -> ParametersDict:
    """Return rendition parameters serialized for the runner payload."""
    list_rendition = get_list_rendition()
    str_resolution: dict[str, dict[str, Any]] = {
        str(k): {
            "resolution": v["resolution"],
            "video_bitrate": v["video_bitrate"],
            "audio_bitrate": v["audio_bitrate"],
            "encode_mp4": v["encode_mp4"],
        }
        for k, v in list_rendition.items()
    }
    return {"rendition": json.dumps(str_resolution)}


def _attach_cut_info(parameters: ParametersDict, video: Video) -> None:
    """Attach cut information to parameters if it exists for the given video."""
    try:
        cut_video = CutVideo.objects.get(video=video)
        str_cut_info = {
            "start": str(cut_video.start),
            "end": str(cut_video.end),
            "initial_duration": str(cut_video.duration),
        }
        parameters["cut"] = json.dumps(str_cut_info)
    except CutVideo.DoesNotExist:
        pass


def _build_base_url(site: Site) -> str:
    """Build the public base URL for a given Django site."""
    url_scheme = "https" if SECURE_SSL_REDIRECT else "http"
    return f"{url_scheme}://{site.domain}"


def _build_video_source_url(video: Video, base_url: str) -> str:
    """Build the public source URL for a video file."""
    return f"{base_url}/media/{video.video}"


def _build_transcription_source_url(video: Video, base_url: str) -> str:
    """Build the preferred source URL for transcription tasks."""
    mp3 = video.get_video_mp3() if hasattr(video, "get_video_mp3") else None
    mp3file = getattr(mp3, "source_file", None) if mp3 else None
    if mp3file is not None and getattr(mp3file, "url", None):
        return f"{base_url}{mp3file.url}"
    return _build_video_source_url(video, base_url)


def _build_studio_source_url(recording: Recording, base_url: str) -> str:
    """Build the public source URL for a studio XML file."""
    source_file = recording.source_file
    media_url = getattr(settings, "MEDIA_URL", "/media/").rstrip("/")
    try:
        rel_path = os.path.relpath(
            str(source_file), str(getattr(settings, "MEDIA_ROOT", ""))
        )
    except Exception:
        rel_path = str(source_file)
    rel_path = rel_path.lstrip("/")
    return f"{base_url}{media_url}/{rel_path}"


def _build_media_asset_url(base_url: str, asset_name: str) -> str:
    """Build a public media URL for a stored asset name."""
    return f"{base_url}/media/{asset_name}"


def _extend_with_watermark_info(
    dressing_info: ParametersDict, dressing: Any, base_url: str
) -> None:
    """Append watermark-related dressing metadata when available."""
    if not dressing.watermark:
        return

    log.info("Dressing watermark found")
    dressing_info["watermark"] = _build_media_asset_url(
        base_url, str(dressing.watermark.file.name)
    )
    dressing_info["watermark_position"] = dressing.position
    dressing_info["watermark_opacity"] = str(dressing.opacity)


def _extend_with_credit_info(
    dressing_info: ParametersDict, credit_name: str, credit, base_url: str
) -> None:
    """Append opening/ending credit metadata when available."""
    if not credit:
        return

    log.info("Dressing %s found", credit_name.replace("_", " "))
    dressing_info[credit_name] = credit.slug
    dressing_info[f"{credit_name}_video"] = _build_media_asset_url(
        base_url, str(credit.video.name)
    )
    dressing_info[f"{credit_name}_video_duration"] = str(credit.duration)


def _attach_dressing_info(
    parameters: ParametersDict, video: Video, base_url: Optional[str] = None
) -> None:
    """Attach dressing information to parameters if available for the given video."""
    try:
        from pod.dressing.models import Dressing

        if base_url is None:
            site = Site.objects.get_current()
            base_url = _build_base_url(site)

        if not Dressing.objects.filter(videos=video).exists():
            return

        log.info("Dressing found for video id: %s", video.id)
        dressing = Dressing.objects.get(videos=video)
        if not dressing:
            return

        dressing_info: ParametersDict = {}
        _extend_with_watermark_info(dressing_info, dressing, base_url)
        _extend_with_credit_info(
            dressing_info, "opening_credits", dressing.opening_credits, base_url
        )
        _extend_with_credit_info(
            dressing_info, "ending_credits", dressing.ending_credits, base_url
        )

        if dressing_info:
            parameters["dressing"] = json.dumps(dressing_info)
    except Exception as exc:
        log.error(f"Error obtaining dressing for video {video.id}: {str(exc)}")


def _attach_video_metadata(parameters: ParametersDict, video: Video) -> None:
    """Attach lightweight video metadata when available."""
    video_id = getattr(video, "id", None)
    if video_id is not None:
        parameters["video_id"] = int(video_id)

    video_slug = getattr(video, "slug", None)
    if video_slug:
        parameters["video_slug"] = str(video_slug)

    video_title = getattr(video, "title", None)
    if video_title:
        parameters["video_title"] = str(video_title)


def _prepare_encoding_parameters(
    video: Optional[Video] = None,
    base_url: Optional[str] = None,
) -> ParametersDict:
    """Prepare encoding parameters for video or recording.

    Args:
        video: Video object (for video encoding).
               For studio recordings, pass None as video-specific metadata,
               cut and dressing info do not apply.
        base_url: Explicit base URL to use for dressing asset URLs.
                  Defaults to the current site URL when omitted.

    Returns:
        Dictionary with rendition and, when a video is provided, optional
        cut info, optional dressing info, and video metadata
        (video_id, video_slug, video_title).
    """
    parameters = _build_rendition_parameters()

    if video:
        _attach_cut_info(parameters, video)
        if base_url is None:
            _attach_dressing_info(parameters, video)
        else:
            _attach_dressing_info(parameters, video, base_url=base_url)
        _attach_video_metadata(parameters, video)

    return parameters


def _prepare_task_data(
    source_url: str,
    base_url: str,
    parameters: ParametersDict,
    task_type: TaskType,
) -> RunnerManagerTaskPayload:
    """Prepare task payload for runner manager.

    Args:
        source_url: URL to the source file (video or XML)
        base_url: Base URL of the site
        parameters: Encoding parameters

    Returns:
        Dictionary with task data
    """
    return {
        "etab_name": f"{__TITLE_ETB__} / {__TITLE_SITE__}",
        "app_name": "Esup-Pod",
        "app_version": VERSION,
        "task_type": task_type,
        "source_url": source_url,
        "notify_url": f"{base_url}/runner/notify_task_end/",
        "parameters": parameters,
    }


# ---- Runner manager helpers (module-level to keep complexity low) ----
def _rotate_same_priority_runner_managers(
    runner_managers: list[RunnerManager],
) -> list[RunnerManager]:
    """Rotate a same-priority runner manager list using last assigned task."""
    if len(runner_managers) <= 1:
        return runner_managers

    runner_manager_ids = [rm.id for rm in runner_managers]
    last_runner_manager_id = (
        Task.objects.filter(runner_manager_id__in=runner_manager_ids)
        .order_by("-date_added", "-id")
        .values_list("runner_manager_id", flat=True)
        .first()
    )
    if not last_runner_manager_id or last_runner_manager_id not in runner_manager_ids:
        return runner_managers

    last_index = runner_manager_ids.index(last_runner_manager_id)
    return runner_managers[last_index + 1 :] + runner_managers[: last_index + 1]


def _get_runner_managers(site: Site) -> list[RunnerManager]:
    """Return active site runner managers ordered by priority with round-robin per priority."""
    ordered_runner_managers = list(
        RunnerManager.objects.filter(site=site, is_active=True).order_by("priority", "id")
    )
    if len(ordered_runner_managers) <= 1:
        return ordered_runner_managers

    runner_managers: list[RunnerManager] = []
    current_priority: Optional[int] = None
    current_group: list[RunnerManager] = []
    # Apply round-robin only inside groups that share the same priority level.
    for runner_manager in ordered_runner_managers:
        if current_priority is None or runner_manager.priority == current_priority:
            current_group.append(runner_manager)
            current_priority = runner_manager.priority
            continue

        runner_managers.extend(_rotate_same_priority_runner_managers(current_group))
        current_group = [runner_manager]
        current_priority = runner_manager.priority

    if current_group:
        runner_managers.extend(_rotate_same_priority_runner_managers(current_group))

    return runner_managers


def _ids_for(
    source_type: SourceType, source_id: Union[int, str]
) -> tuple[Optional[int], Optional[int]]:
    """Return (video_id, recording_id) tuple based on source type."""
    return (int(source_id), None) if source_type == "video" else (None, int(source_id))


def _execute_url(rm: RunnerManager) -> str:
    """Build the execute endpoint URL for the given runner manager."""
    base = rm.url if rm.url.endswith("/") else rm.url + "/"
    return base + "task/execute"


def _headers(rm: RunnerManager) -> HeadersDict:
    """Build authentication and content headers for the runner manager API."""
    return {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {rm.token}",
    }


def _try_send_to_rm(
    rm: RunnerManager, payload: RunnerManagerTaskPayload
) -> Optional[requests.Response]:
    """Try to POST the payload to a runner manager; log and return None on failure."""
    try:
        return requests.post(
            _execute_url(rm), data=json.dumps(payload), headers=_headers(rm), timeout=30
        )
    except requests.RequestException as exc:
        log.warning(
            f"Cannot reach runner manager {rm.name}: {str(exc)}. Trying next one."
        )
        return None


def _parse_runner_response(
    rm: RunnerManager, response: requests.Response
) -> Optional[RunnerManagerResponse]:
    """Parse and validate a runner manager HTTP response.

    Returns None when the response body is not a valid JSON payload expected from
    the runner manager (for example an HTML error page returned by a reverse proxy
    with HTTP 200).
    """
    if not response.content:
        return {}

    try:
        payload = response.json()
    except ValueError:
        content_type = response.headers.get("Content-Type", "")
        log.warning(
            "Runner manager %s returned HTTP 200 with non-JSON body "
            "(Content-Type=%s). Trying next one.",
            rm.name,
            content_type or "unknown",
        )
        return None

    if not isinstance(payload, dict):
        log.warning(
            "Runner manager %s returned an unexpected JSON payload type (%s). "
            "Trying next one.",
            rm.name,
            type(payload).__name__,
        )
        return None

    return cast(RunnerManagerResponse, payload)


def _prestore_encoding_if_needed(
    *,
    task_type: TaskType,
    source_type: SourceType,
    video_id: Optional[int],
    recording_id: Optional[int],
    rm: RunnerManager,
    data: RunnerManagerTaskPayload,
) -> None:
    """Run pre-store steps for encoding/studio tasks.

    Does nothing for transcription tasks.
    """
    if task_type not in ("encoding", "studio"):
        return
    execute_url = _execute_url(rm)
    if source_type == "video":
        if video_id is not None:
            store_before_remote_encoding_video(video_id, execute_url, data)
        else:
            log.warning(
                "Unexpected None video_id for source_type 'video' while preparing store_before_remote_encoding_video."
            )
    elif source_type == "recording":
        if recording_id is not None:
            store_before_remote_encoding_recording(recording_id, execute_url, data)
        else:
            log.warning(
                "Unexpected None recording_id for source_type 'recording' while preparing store_before_remote_encoding_recording."
            )


def _submit_to_runner_manager(
    rm: RunnerManager,
    data: RunnerManagerTaskPayload,
    task_type: TaskType,
    source_type: SourceType,
    video_id: Optional[int],
    recording_id: Optional[int],
) -> bool:
    """Submit payload to one runner manager and handle response and pre-store."""
    response = _try_send_to_rm(rm, data)
    if response is None:
        return False
    if response.status_code != 200:
        log.warning(
            f"Runner manager {rm.name} returned status code {response.status_code}. Trying next one."
        )
        return False
    log.info(
        f"Runner manager {rm.name} is available to process {task_type} for {source_type} {video_id or recording_id}."
    )
    payload = _parse_runner_response(rm, response)
    if payload is None:
        return False

    _update_task_from_response(video_id, recording_id, task_type, rm, payload)
    _prestore_encoding_if_needed(
        task_type=task_type,
        source_type=source_type,
        video_id=video_id,
        recording_id=recording_id,
        rm=rm,
        data=data,
    )
    return True


def _submit_to_runner_managers(
    *,
    runner_managers: list[RunnerManager],
    data: RunnerManagerTaskPayload,
    task_type: TaskType,
    source_type: SourceType,
    source_id: Union[int, str],
) -> bool:
    """Try runner managers in order and stop on the first successful submission."""
    video_id, recording_id = _ids_for(source_type, source_id)

    for rm in runner_managers:
        if _submit_to_runner_manager(
            rm,
            data=data,
            task_type=task_type,
            source_type=source_type,
            video_id=video_id,
            recording_id=recording_id,
        ):
            return True

    log.warning(
        f"No runner manager available to process {task_type} for {source_type} {source_id}. "
        f"Task will remain pending and will be retried by the process_tasks command."
    )
    return False


def _update_task_pending(
    source_type: SourceType, source_id: Union[int, str], task_type: TaskType
) -> tuple[Optional[int], Optional[int]]:
    """Create or set a pending task for the given source and return (video_id, recording_id)."""
    video_id, recording_id = _ids_for(source_type, source_id)
    log.info(
        "Update task to pending for video_id: %s, recording_id: %s",
        video_id,
        recording_id,
    )
    _edit_task(
        video_id=video_id,
        recording_id=recording_id,
        type=task_type,
        status="pending",
        runner_manager_id=None,
        task_id=None,
    )
    return video_id, recording_id


def _update_task_from_response(
    video_id: Optional[int],
    recording_id: Optional[int],
    task_type: TaskType,
    rm: RunnerManager,
    response_json: RunnerManagerResponse,
) -> None:
    """Update the task row using the response payload returned by the runner manager."""
    task_id = response_json.get("task_id")
    status = str(response_json.get("status", "pending"))
    log.info(
        "Update task for video_id=%s, recording_id=%s, task_type=%s with response_json=%s",
        video_id,
        recording_id,
        task_type,
        response_json,
    )
    _edit_task(
        video_id=video_id,
        recording_id=recording_id,
        type=task_type,
        status=status,
        runner_manager_id=rm.id,
        task_id=task_id,
    )


def _send_task_to_runner_manager(
    *,
    task_type: TaskType,
    source_id: Union[int, str],
    source_type: SourceType,
    source_url: str,
    base_url: str,
    parameters: ParametersDict,
) -> bool:
    """Submit a task to the Runner Manager and update the DB task row.

    - task_type: one of "encoding", "studio", "transcription"
    - source_type: "video" or "recording" (used to resolve ids and pre-store behavior)
    """

    try:
        # Keep a local pending row even when no runner is currently available.
        # This allows process_tasks to retry submission later.
        _update_task_pending(source_type, source_id, task_type)

        site = Site.objects.get_current()
        runner_managers_list = _get_runner_managers(site)
        if not runner_managers_list:
            log.warning(
                f"No active runner manager defined for site {site.domain}. Cannot process {task_type} for {source_type} {source_id}."
            )
            return False

        # Build payload and try immediate submission
        data = _prepare_task_data(source_url, base_url, parameters, task_type)

        return _submit_to_runner_managers(
            runner_managers=runner_managers_list,
            data=data,
            task_type=task_type,
            source_type=source_type,
            source_id=source_id,
        )

    except Exception as exc:
        log.error(
            f"Error to process {task_type} for {source_type} {source_id}: {str(exc)}"
        )
        return False


def encode_video(video_id: int) -> None:
    """Start video encoding with runner manager."""
    log.info("Start encoding, with runner manager, for id: %s" % video_id)
    try:
        site = Site.objects.get_current()
        # Get video info
        video = get_object_or_404(Video, id=video_id)
        # Build content URL
        base_url = _build_base_url(site)
        content_url = _build_video_source_url(video, base_url)

        # Prepare encoding parameters
        parameters = _prepare_encoding_parameters(video=video)

        # Send encoding task to runner manager
        _send_task_to_runner_manager(
            task_type="encoding",
            source_id=video_id,
            source_type="video",
            source_url=content_url,
            base_url=base_url,
            parameters=parameters,
        )

    except Exception as exc:
        log.error(
            'Error to encode video "%(id)s": %(exc)s' % {"id": video_id, "exc": str(exc)}
        )


def encode_studio_recording(recording_id: int) -> None:
    """Start encoding studio recording with runner manager.

    This function handles encoding of studio recordings by passing the XML
    source file URL to the runner manager.
    """
    log.info(
        "Start encoding, with runner manager, for studio recording id %s" % recording_id
    )
    try:
        site = Site.objects.get_current()
        # Get studio recording
        recording = Recording.objects.get(id=recording_id)
        base_url = _build_base_url(site)
        source_url = _build_studio_source_url(recording, base_url)

        # Prepare encoding parameters (no specific cut info for studio recordings)
        parameters = _prepare_encoding_parameters(video=None)

        # Send studio task to runner manager
        _send_task_to_runner_manager(
            task_type="studio",
            source_id=recording_id,
            source_type="recording",
            source_url=source_url,
            base_url=base_url,
            parameters=parameters,
        )

    except Recording.DoesNotExist:
        log.error(f"Recording {recording_id} not found.")
    except Exception as exc:
        log.error(f"Error to encode recording {recording_id}: {str(exc)}")


def transcript_video(video_id: int) -> None:
    """Start video transcription with runner manager."""
    log.info("Start transcription, with runner manager, for id: %s" % video_id)
    try:
        site = Site.objects.get_current()
        # Get video info
        video = get_object_or_404(Video, id=video_id)
        base_url = _build_base_url(site)
        content_url = _build_transcription_source_url(video, base_url)

        # Prepare transcript parameters
        parameters = _prepare_transcription_parameters(video=video)

        # Mark video as encoding in progress
        Video.objects.filter(id=video_id).update(encoding_in_progress=True)

        # Update encoding step to transcripting audio
        change_encoding_step(video_id, 5, "transcripting audio")

        # Send transcription task to runner manager
        _send_task_to_runner_manager(
            task_type="transcription",
            source_id=video_id,
            source_type="video",
            source_url=content_url,
            base_url=base_url,
            parameters=parameters,
        )

    except Exception as exc:
        log.error(
            'Error to transcribe video "%(id)s": %(exc)s'
            % {"id": video_id, "exc": str(exc)}
        )


def _prepare_transcription_parameters(video: Video) -> ParametersDict:
    """Prepare parameters for a transcription task.

    Args:
        video: `Video` instance to transcribe.

    Returns:
        Parameter dictionary for the Runner Manager.
    """
    try:
        from .transcript import resolve_transcription_language

        # Requested language (video `transcript` field)
        lang = resolve_transcription_language(video)

        # Options from settings (optional on runner side)
        transcription_type = getattr(settings, "TRANSCRIPTION_TYPE", None)
        normalize = bool(getattr(settings, "TRANSCRIPTION_NORMALIZE", False))

        params: ParametersDict = {
            "language": lang,
            # Duration may help runner to tune/optimize
            "duration": float(getattr(video, "duration", 0) or 0),
            # Text normalization (punctuation/casing) on runner side if supported
            "normalize": normalize,
        }
        _attach_video_metadata(params, video)
        # If needed in future, we can add model size or other options here
        if transcription_type:
            params["model_type"] = transcription_type
            # Possibility to add model size if needed in future
            # params["model"] = "medium"

        return params
    except Exception:
        # Keep legacy key name for backward compatibility with older runners.
        params: ParametersDict = {"lang": getattr(video, "transcript", "") or ""}
        _attach_video_metadata(params, video)
        return params


def _edit_task(
    video_id: Optional[int],
    type: str,
    status: str,
    runner_manager_id: Optional[int] = None,
    task_id: Optional[str] = None,
    recording_id: Optional[int] = None,
) -> None:
    """Edit or create a task for a video or studio recording."""
    try:
        from .task_queue import refresh_pending_task_ranks

        log.info(
            f"Edit or create a task: {video_id} {type} {runner_manager_id} {status} {task_id}"
        )
        # Check if a task already exists for this video and type with pending status
        # Build base queryset depending on source type
        if type == "studio":
            tasks_list = list(
                Task.objects.filter(
                    recording_id=recording_id,
                    type=type,
                    status="pending",
                )
            )
        else:
            tasks_list = list(
                Task.objects.filter(
                    video_id=video_id,
                    type=type,
                    status="pending",
                )
            )
        if not tasks_list:
            # Create new task
            task = Task(
                video_id=video_id if type != "studio" else None,
                recording_id=recording_id if type == "studio" else None,
                type=type,
                runner_manager_id=runner_manager_id,
                status=status,
                task_id=task_id,
            )
            task.save()
        else:
            # Edit existing task
            task = tasks_list[0]
            task.status = status
            if runner_manager_id is not None:
                task.runner_manager_id = runner_manager_id
            if task_id is not None:
                task.task_id = task_id
            # Keep association fields as-is
            task.save()

        refresh_pending_task_ranks()

    except Exception as exc:
        log.error(
            f"Unable to edit a task (video_id={video_id}, recording_id={recording_id}): {str(exc)}"
        )

EsupPortail / Esup-Pod / 25740886384

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