gandalf/ticket_text.py

"""Markdown for the tickets Gandalf opens in Tinker Tickets, and the notes it adds to them.

Pure formatting only: monitor.py gathers the facts and passes them in.
"""
from datetime import datetime, timezone
from typing import List, Optional, Sequence, Tuple
from urllib.parse import quote

Section = Tuple[str, List[str]]


def duration(seconds) -> str:
    seconds = max(0, int(seconds))
    if seconds < 60:
        return f'{seconds} s'
    minutes = seconds // 60
    if minutes < 60:
        return f'{minutes} min'
    hours, minutes = divmod(minutes, 60)
    if hours < 48:
        return f'{hours} h {minutes} min' if minutes else f'{hours} h'
    days, hours = divmod(hours, 24)
    return f'{days} d {hours} h' if hours else f'{days} d'


def utc(dt: Optional[datetime]) -> str:
    """Database times are naive UTC."""
    if dt is None:
        return 'unknown'
    if dt.tzinfo is not None:
        dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
    return dt.strftime('%Y-%m-%d %H:%M UTC')


def speed(mbps) -> str:
    if not mbps:
        return ''
    mbps = int(mbps)
    if mbps >= 1000:
        gbps = mbps / 1000
        return f'{gbps:g} Gbps'
    return f'{mbps} Mbps'


def device_url(base: str, name: str, port_idx=None) -> str:
    if not base or not name:
        return ''
    url = f'{base}/device/{quote(name, safe="")}'
    return f'{url}#port-{port_idx}' if port_idx not in (None, '') else url


def history_url(base: str, target: str) -> str:
    return f'{base}/events?target={quote(target, safe="")}&range=90d' if base and target else ''


def cutsheet_url(base: str, cable_id) -> str:
    return f'{base}/cutsheet#cable-{cable_id}' if base else ''


def link(text: str, url: str) -> str:
    return f'[{text}]({url})' if url else text


def port_text(switch: str, port_name: str, port: dict) -> str:
    """'8SFP-AGG port 7 "Large1": up at 10 Gbps, DAC'"""
    idx = port.get('port_idx')
    text = f'{switch} port {idx}' if idx is not None else f'{switch} {port_name}'
    if port_name and idx is not None and port_name != f'Port {idx}':
        text += f' "{port_name}"'
    state = []
    if port.get('up'):
        rate = speed(port.get('speed_mbps'))
        state.append(f'up at {rate}' if rate else 'up')
    else:
        state.append('down')
    if port.get('poe_power'):
        state.append(f'PoE {float(port["poe_power"]):.1f} W')
    sfp = port.get('sfp') or {}
    if sfp.get('found'):
        kind = 'DAC' if sfp.get('kind') == 'dac' else sfp.get('kind', '')
        module = ' '.join(p for p in (kind, sfp.get('part', '')) if p)
        if module:
            state.append(f'module: {module}')
        if sfp.get('rx_fault') or sfp.get('tx_fault'):
            state.append('**module reports a fault**')
    return f'{text}: {", ".join(state)}'


def description(heading: str, facts: Sequence[Tuple[str, str]], summary: str,
                checks: Sequence[str], sections: Sequence[Section] = (), recovers: bool = True) -> str:
    lines = [f'## {heading}', '']
    lines += [f'- **{key}:** {value}' for key, value in facts if value not in (None, '')]
    lines += ['', summary.strip()]
    if checks:
        lines += ['', '### What to check']
        lines += [f'{n}. {check}' for n, check in enumerate(checks, 1)]
    for title, body in sections:
        if body:
            lines += ['', f'### {title}'] + list(body)
    if recovers:
        lines += ['', '_Opened automatically by Gandalf. It adds a note here when the problem clears, '
                      'but leaves closing the ticket to you._']
    else:
        lines += ['', '_Reported automatically by Gandalf. This already happened, so there will be no '
                      'recovery note._']
    return '\n'.join(lines)


def history_lines(stats: dict, now: datetime) -> List[str]:
    week, month = stats.get('last_7d', 0), stats.get('last_30d', 0)
    if not month:
        lines = ['- First time in the last 30 days.']
    else:
        lines = [f'- Happened {week} time{"s" if week != 1 else ""} in the last 7 days '
                 f'and {month} in the last 30 (not counting this one).']
        if week >= 3:
            lines.append('- **This keeps coming back**: look for a cause that comes and goes '
                         '(a loose cable, a failing optic, power, or a scheduled job).')
    first, resolved = stats.get('last_first_seen'), stats.get('last_resolved_at')
    if resolved:
        took = duration((resolved - first).total_seconds()) if first else 'unknown'
        ago = duration((now - resolved).total_seconds())
        lines.append(f'- Last cleared {utc(resolved)} ({ago} ago), after {took}.')
    return lines


ISSUE_FIXED = {
    'cable_mismatch': 'The cable now matches the cutsheet (or the cutsheet was updated).',
    'monitor_source_unavailable': 'Prometheus is returning data again; host and interface alerting has resumed.',
}


def recovery_comment(event: dict, now: datetime, url: str = '') -> str:
    """The note Gandalf adds to a ticket when its alert clears."""
    first, resolved = event.get('first_seen'), event.get('resolved_at') or now
    target = event['target_name'] + (f' {event["target_detail"]}' if event.get('target_detail') else '')
    lines = [f'**✅ Cleared**: {event["event_type"].replace("_", " ")} on {target}.', '']
    lines.append(f'- **Cleared:** {utc(resolved)}')
    if first:
        lines.append(f'- **Lasted:** {duration((resolved - first).total_seconds())} '
                     f'(first seen {utc(first)})')
    if event.get('consecutive_failures'):
        lines.append(f'- **Failed checks:** {event["consecutive_failures"]}')
    if url:
        lines.append(f'- **Gandalf:** {url}')
    lines += ['', ISSUE_FIXED.get(event['event_type'], 'Gandalf no longer sees the problem.')]
    lines += ['', '_Gandalf leaves this ticket open. Close it once you know why it happened, '
                  'or leave it open if it needs follow-up._']
    return '\n'.join(lines)