user_avatars

class UserAvatarCache(QtCore.QObject):
    """Cache of round avatar pixmaps keyed by user name and size.

    :meth:`pixmap` never blocks: it returns an initials avatar right away
    and schedules the real image, emitting :attr:`avatar_updated` when the
    download lands so views can repaint.
    """

    #: Emitted with the login name once a downloaded avatar replaced the
    #: initials placeholder for that user.
    avatar_updated = QtCore.Signal(str)

    def __init__(self, parent: QtCore.QObject | None = None) -> None:
        super().__init__(parent)
        self._pixmaps: dict[tuple[str, int], QtGui.QPixmap] = {}
        self._sources: dict[str, str] = {}
        self._pending: set[str] = set()

    def clear(self) -> None:
        """Drop every cached pixmap and downloaded source path."""
        self._pixmaps.clear()
        self._sources.clear()

    def pixmap(
        self,
        user_name: str,
        full_name: str,
        size: int,
    ) -> QtGui.QPixmap | None:
        """Return the avatar for *user_name*, scheduling a download once.

        Args:
            user_name: Login name, used as the cache key and for initials.
            full_name: Display name the initials are derived from.
            size: Avatar diameter in logical pixels.

        Returns:
            A rendered pixmap, or ``None`` when there is nothing to draw.
        """
        if not user_name:
            return None
        key = (user_name, int(size))
        pixmap = self._pixmaps.get(key)
        if pixmap is None:
            pixmap = self._render(user_name, full_name, int(size))
            self._pixmaps[key] = pixmap
        self._schedule_fetch(user_name)
        return pixmap

    def _render(
        self, user_name: str, full_name: str, size: int
    ) -> QtGui.QPixmap:
        """Render one avatar through :class:`AYUserImage`.

        The widget is only a renderer here - it is never shown - so the
        avatars stay identical to the ones the rest of the UI draws.
        """
        src = self._sources.get(user_name, "")
        if src and not os.path.exists(src):
            # The image cache is shared by all AYON processes on the
            #   machine, any of them may have evicted the file meanwhile.
            #   Fall back to initials and let it be downloaded again.
            del self._sources[user_name]
            src = ""
        widget = AYUserImage(
            src=src,
            name=user_name,
            full_name=full_name or user_name,
            size=size,
            outline=False,
        )
        pixmap = widget.pxm
        widget.deleteLater()
        return pixmap

    def _schedule_fetch(self, user_name: str) -> None:
        """Queue a one-off background download of *user_name*'s avatar."""
        if user_name in self._sources or user_name in self._pending:
            return
        self._pending.add(user_name)

        def _work() -> str:
            cache = ImageCache.get_instance()

            def _download() -> io.BytesIO:
                stream = _fetch_avatar(user_name)
                if stream is None:
                    raise _NoAvatarError(user_name)
                return stream

            try:
                file_path = cache.get(f"user-avatar/{user_name}", _download)
                # Generated initials may be cached from before they
                #   were skipped on download.
                if file_path and _is_generated_initials(
                    _read_file_head(file_path)
                ):
                    return ""
                return file_path
            except _NoAvatarError:
                return ""
            except Exception:  # noqa: BLE001 - avatars are optional
                log.debug(
                    "Could not fetch avatar for %r", user_name, exc_info=True
                )
                return ""

        def _done(file_path: str) -> None:
            # The cache outlives no view, but a queued download can land
            # after the owning widget was torn down.
            if shiboken.isValid(self):
                self._on_avatar_ready(user_name, file_path)

        get_task_queue().enqueue(
            AsyncTask(
                name=f"fetch_avatar:{user_name}",
                function=_work,
                callback=_done,
                priority=_FETCH_PRIORITY,
                context_id="user-avatars",
                cancellable=False,
            )
        )

    def _on_avatar_ready(self, user_name: str, file_path: str) -> None:
        """Swap the initials placeholder for the downloaded avatar."""
        self._pending.discard(user_name)
        if not file_path:
            # Remember the miss so the download is not retried per repaint.
            self._sources[user_name] = ""
            return
        self._sources[user_name] = file_path
        for key in [k for k in self._pixmaps if k[0] == user_name]:
            del self._pixmaps[key]
        self.avatar_updated.emit(user_name)