src/notifications.py

"""Hugging Face submission notifications.

Notifies a submitter by ``@``-mentioning them in a discussion comment on the
leaderboard Space, which triggers Hugging Face's native notification.

Configuration (see :mod:`src.envs`):
    * ``HF_TOKEN`` — token used to post the comment / discussion.
    * ``REPO_ID`` — the Space repo id (default notification target).
        * ``NOTIFICATION_DISCUSSION_NUM`` — comments are posted on this existing
            discussion. If unset, notification sending is skipped.
"""

import logging
from datetime import datetime, timezone

from huggingface_hub import HfApi

from src.envs import HF_TOKEN, NOTIFICATION_DISCUSSION_NUM, REPO_ID

logger = logging.getLogger(__name__)


def send_hf_notification(
    username: str,
    message: str,
    *,
    repo_id: str | None = None,
    discussion_num: int | None = None,
) -> str | None:
    """Post a notification to ``username`` on the leaderboard Space.

    Returns the discussion URL on success, or ``None`` if the notification
    could not be sent (missing token / username, or an API error).  All
    failures are swallowed and logged so the caller's loop is never broken.
    """
    if not HF_TOKEN:
        logger.warning("[notify] HF_TOKEN not set — skipping notification for %r", username)
        return None

    username = (username or "").strip().lstrip("@")
    if not username:
        logger.warning("[notify] empty username — skipping notification")
        return None

    repo_id = repo_id or REPO_ID
    if discussion_num is None:
        discussion_num = NOTIFICATION_DISCUSSION_NUM
    if not discussion_num:
        logger.warning(
            "[notify] NOTIFICATION_DISCUSSION_NUM not set — skipping notification for %r",
            username,
        )
        return None

    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    comment = f"@{username}\n\n{message}\n\nTime: {now}"

    api = HfApi(token=HF_TOKEN)
    try:
        api.comment_discussion(
            repo_id=repo_id,
            repo_type="space",
            discussion_num=int(discussion_num),
            comment=comment,
        )
        url = f"https://huggingface.co/spaces/{repo_id}/discussions/{discussion_num}"
    except Exception:
        logger.error("[notify] failed to notify @%s", username, exc_info=True)
        return None

    logger.info("[notify] notified @%s (%s)", username, url)
    return url