shaiwatcher/modules/common/boot_notice.py

# modules/common/boot_notice.py
import os
import re
import html
import discord
import aiohttp
import xml.etree.ElementTree as ET
from modules.common.settings import cfg

# ---------------- RSS helpers ----------------

def _strip_html_keep_text(s: str) -> str:
    """Remove HTML tags, unescape entities, collapse excessive blank lines."""
    if not s:
        return ""
    # Replace <br> and <p> with newlines before stripping tags
    s = re.sub(r'(?i)<\s*br\s*/?\s*>', '\n', s)
    s = re.sub(r'(?i)</\s*p\s*>', '\n', s)
    s = re.sub(r'(?i)<\s*p\s*>', '', s)
    # Strip all remaining tags
    s = re.sub(r'<[^>]+>', '', s)
    # Unescape HTML entities
    s = html.unescape(s)
    # Trim trailing spaces on each line
    s = '\n'.join(line.rstrip() for line in s.splitlines())
    # Collapse 3+ blank lines to max 2
    s = re.sub(r'\n{3,}', '\n\n', s).strip()
    return s

async def _fetch_latest_commit_from_rss(rss_url: str):
    """
    Return (subject:str|None, body:str|None) from the newest item in a Gitea/Git RSS feed.
    Best-effort. We avoid posting links/usernames and keep content human-friendly.
    """
    try:
        timeout = aiohttp.ClientTimeout(total=8)
        async with aiohttp.ClientSession(timeout=timeout) as sess:
            async with sess.get(rss_url) as resp:
                if resp.status != 200:
                    return None, None
                text = await resp.text()
        root = ET.fromstring(text)
        item = root.find('./channel/item')
        if item is None:
            return None, None

        title = (item.findtext('title') or '').strip()
        # Gitea typically puts commit message (possibly HTML-wrapped) in <description>
        desc_raw = (item.findtext('description') or '').strip()
        body = _strip_html_keep_text(desc_raw)

        # Some feeds stuff noise/usernames into title; keep it short & human:
        # If title contains " pushed " etc., try to fall back to first line of body.
        if title and re.search(r'\b(pushed|commit|committed)\b', title, re.I):
            # If body has a first line that looks like a summary, use it.
            first_line = body.splitlines()[0].strip() if body else ""
            if first_line:
                title = first_line

        # Clean again in case any entities lingered in title
        title = _strip_html_keep_text(title)

        # If title empty but body present, use the first non-empty line of body as title.
        if not title and body:
            for line in body.splitlines():
                if line.strip():
                    title = line.strip()
                    break

        return (title or None), (body or None)
    except Exception:
        return None, None

# ---------------- Status helpers ----------------

def _build_status_line(status: str, old_v: str, new_v: str, desc: str) -> str | None:
    """
    Return a short human-readable boot status line, or None if nothing to post.
    Known statuses:
      - fetched_new            -> updated & booted
      - cached_no_update       -> booted cached, no update
      - cache_only_error       -> booted cached, repo unavailable
      - scheduled_restart      -> a scheduled restart was initiated
    """
    status = (status or "").strip()
    old_v = (old_v or "").strip()
    new_v = (new_v or "").strip()

    if status == "fetched_new":
        line = f"✅ Booted new version: v{old_v or '0.0.0.0'} → **v{new_v}**"
    elif status == "cached_no_update":
        line = f"🟢 Booted cached version: **v{new_v}** — no new update found"
    elif status == "cache_only_error":
        line = f"🟡 Booted cached version: **v{new_v}** — repository not accessible"
    elif status == "scheduled_restart":
        line = "🕒 Scheduled restart executed"
    else:
        return None

    return f"{line}\n_{desc.strip()}_" if desc else line

def _only_version_and_details(subject: str | None, body: str | None) -> str | None:
    """
    Format to 'Version number' (bold) + 'Version details' (md).
    We try to extract a version-like token from subject; otherwise we use subject as-is.
    """
    if not subject and not body:
        return None

    version = None
    if subject:
        # Try to find a version-like token (v1.2.3 or 1.2.3.4 etc.)
        m = re.search(r'\bv?(\d+\.\d+(?:\.\d+){0,2})\b', subject)
        if m:
            version = m.group(0)
        else:
            # Fall back to the subject line itself as "version-ish" title
            version = subject.strip()

    if version and body:
        return f"**{version}**\n{body.strip()}"
    if version:
        return f"**{version}**"
    # No subject/version, only body
    return body.strip() if body else None

# ---------------- Main entry ----------------

async def post_boot_notice(bot):
    """
    Posts concise boot status to the modlog channel.
    - Always: one status line (if SHAI_BOOT_STATUS is set to a known value).
    - If SHAI_BOOT_STATUS == 'fetched_new': fetch latest commit from SHAI_REPO_RSS
      and post ONLY the commit message (version number + markdown details).
    """
    status = os.getenv("SHAI_BOOT_STATUS", "").strip()    # fetched_new | cached_no_update | cache_only_error | scheduled_restart | ''
    desc   = os.getenv("SHAI_BOOT_DESC", "").strip()
    old_v  = os.getenv("SHAI_BOOT_OLD", "").strip()
    new_v  = os.getenv("SHAI_BOOT_NEW", "").strip()
    rss    = os.getenv("SHAI_REPO_RSS", "").strip()

    # Build status line
    status_line = _build_status_line(status, old_v, new_v, desc)
    if not status_line:
        return  # nothing to say

    # Resolve modlog channel
    modlog_channel_id = cfg(bot).int('modlog_channel_id', 0)
    if not modlog_channel_id:
        return

    ch = None
    for g in bot.guilds:
        ch = g.get_channel(modlog_channel_id)
        if ch:
            break
    if not ch:
        return

    # Post the status
    try:
        await ch.send(status_line, allowed_mentions=discord.AllowedMentions.none())
    except Exception:
        return

    # If we fetched & booted a new version, follow up with the commit message from RSS (no env details assumed).
    if status == "fetched_new" and rss:
        subj, body = await _fetch_latest_commit_from_rss(rss)
        commit_msg = _only_version_and_details(subj, body)
        if commit_msg:
            try:
                await ch.send(commit_msg, allowed_mentions=discord.AllowedMentions.none())
            except Exception:
                pass