sgl-project--sglang/python/sglang/multimodal_gen/test/server/common/slack.py

"""
This file upload the media generated in diffusion-nightly-test to a slack channel of SGLang
"""

import logging
import os
import tempfile
from datetime import datetime
from typing import List, Union
from urllib.parse import urlparse
from urllib.request import urlopen

from sglang.multimodal_gen.runtime.utils.perf_logger import get_git_commit_hash

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

import inspect

try:
    import sglang.multimodal_gen.test.server.testcase_configs as configs
    from sglang.multimodal_gen.test.server.testcase_configs import DiffusionTestCase

    ALL_CASES = []
    for name, value in inspect.getmembers(configs):
        if name.endswith("_CASES") or "_CASES_" in name:
            if (
                isinstance(value, list)
                and len(value) > 0
                and isinstance(value[0], DiffusionTestCase)
            ):
                ALL_CASES.extend(value)
            elif isinstance(value, list) and len(value) == 0:
                # Assume empty list with matching name is a valid case list container
                pass

    # Deduplicate cases by ID
    seen_ids = set()
    unique_cases = []
    for c in ALL_CASES:
        if c.id not in seen_ids:
            seen_ids.add(c.id)
            unique_cases.append(c)
    ALL_CASES = unique_cases

except Exception as e:
    logger.warning(f"Failed to import test cases: {e}")
    ALL_CASES = []


def _get_status_message(run_id, current_case_id, thread_messages=None):
    date_str = datetime.now().strftime("%d/%m")
    base_header = f"""🧵 for nightly test of {date_str}
*Git Revision:* {get_git_commit_hash()}
*GitHub Run ID:* {run_id}
*Total Tasks:* {len(ALL_CASES)}
"""

    if not ALL_CASES:
        return base_header

    default_emoji_for_case_in_progress = "⏳"
    status_map = {c.id: default_emoji_for_case_in_progress for c in ALL_CASES}

    if thread_messages:
        for msg in thread_messages:
            text = msg.get("text", "")
            # Look for case_id in the message (format: *Case ID:* `case_id`)
            for c in ALL_CASES:
                if f"*Case ID:* `{c.id}`" in text:
                    status_map[c.id] = "✅"

    if current_case_id:
        status_map[current_case_id] = "✅"

    lines = [base_header, "", "*Tasks Status:*"]

    # Calculate padding
    max_len = max(len(c.id) for c in ALL_CASES) if ALL_CASES else 10
    max_len = max(max_len, len("Case ID"))

    # Build markdown table inside a code block
    table_lines = ["```"]
    table_lines.append(f"| {'Case ID'.ljust(max_len)} | Status |")
    table_lines.append(f"| {'-' * max_len} | :----: |")

    for c in ALL_CASES:
        mark = status_map.get(c.id, default_emoji_for_case_in_progress)
        table_lines.append(f"| {c.id.ljust(max_len)} |   {mark}   |")

    table_lines.append("```")

    lines.extend(table_lines)

    return "\n".join(lines)


def upload_file_to_slack(
    case_id: str = None,
    model: str = None,
    prompt: str = None,
    file_path: str = None,
    origin_file_path: Union[str, List[str]] = None,
) -> bool:
    temp_paths = []
    try:
        from slack_sdk import WebClient

        run_id = os.getenv("GITHUB_RUN_ID", "local")

        token = os.environ.get("SGLANG_DIFFUSION_SLACK_TOKEN")
        if not token:
            logger.info("Slack upload failed: no token")
            return False

        if not file_path or not os.path.exists(file_path):
            logger.info("Slack upload failed: no file path")
            return False

        origin_paths = []
        if isinstance(origin_file_path, str):
            if origin_file_path:
                origin_paths.append(origin_file_path)
        elif isinstance(origin_file_path, list):
            origin_paths = [p for p in origin_file_path if p]

        final_origin_paths = []
        for path in origin_paths:
            if path.startswith(("http", "https")):
                try:
                    suffix = os.path.splitext(urlparse(path).path)[1] or ".tmp"
                    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tf:
                        with urlopen(path, timeout=30) as response:
                            tf.write(response.read())
                    temp_paths.append(tf.name)
                    final_origin_paths.append(tf.name)
                except Exception as e:
                    logger.warning(f"Failed to download {path}: {e}")
            else:
                final_origin_paths.append(path)

        uploads = []
        for i, path in enumerate(final_origin_paths):
            if os.path.exists(path):
                title = (
                    "Original Image"
                    if len(final_origin_paths) == 1
                    else f"Original Image {i+1}"
                )
                uploads.append({"file": path, "title": title})

        uploads.append({"file": file_path, "title": "Generated Image"})

        message = (
            f"*Case ID:* `{case_id}`\n" f"*Model:* `{model}`\n" f"*Prompt:* {prompt}"
        )

        client = WebClient(token=token, timeout=60)
        channel_id = "C0A02NDF7UY"
        thread_ts = None

        parent_msg_text = None
        try:
            history = client.conversations_history(channel=channel_id, limit=100)
            for msg in history.get("messages", []):
                if f"*GitHub Run ID:* {run_id}" in msg.get("text", ""):
                    # Use thread_ts if it exists (msg is a reply), otherwise use ts (msg is a parent)
                    thread_ts = msg.get("thread_ts") or msg.get("ts")
                    parent_msg_text = msg.get("text", "")
                    logger.info(f"Found thread_ts: {thread_ts}")
                    break
        except Exception as e:
            logger.warning(f"Failed to search slack history: {e}")

        if not thread_ts:
            try:
                text = _get_status_message(run_id, case_id)
                response = client.chat_postMessage(channel=channel_id, text=text)
                thread_ts = response["ts"]
            except Exception as e:
                logger.warning(f"Failed to create parent thread: {e}")

        # Upload first to ensure it's in history
        client.files_upload_v2(
            channel=channel_id,
            file_uploads=uploads,
            initial_comment=message,
            thread_ts=thread_ts,
        )

        # Then update status based on thread replies
        if thread_ts:
            try:
                replies = client.conversations_replies(
                    channel=channel_id, ts=thread_ts, limit=200
                )
                messages = replies.get("messages", [])
                new_text = _get_status_message(run_id, case_id, messages)

                # Only update if changed significantly (ignoring timestamp diffs if any)
                # But here we just check text content
                if new_text != parent_msg_text:
                    client.chat_update(channel=channel_id, ts=thread_ts, text=new_text)
            except Exception as e:
                logger.warning(f"Failed to update parent message: {e}")

        logger.info(f"File uploaded successfully: {os.path.basename(file_path)}")
        return True

    except Exception as e:
        logger.info(f"Slack upload failed: {e}")
        return False
    finally:
        for p in temp_paths:
            if os.path.exists(p):
                os.remove(p)