management/commands/publish_epicprod_live.py

0001 """Publish epicprod action-stream live events to a Mattermost channel.
0002
0003 The publisher is a polling tailer over the action stream: every cycle it
0004 selects new records passing the live filter (`live_stream_q`), posts one
0005 compact Mattermost message per event, and advances a high-water mark in
0006 PersistentState. It posts as the dedicated 'epicprod' bot account
0007 (EPICPROD_LIVE_TOKEN; falls back to the DISpatcher token) — plain channel
0008 posts never wake DISpatcher, so events flow without bot involvement; people
0009 follow up by @mentioning DISpatcher in a thread under an event post, and
0010 the post carries what the bot needs to drill in (action, subject, outcome,
0011 reason, record link).
0012
0013 Operator knobs live in SysConfig and are re-read every cycle, so channel
0014 rename, importance threshold, and cadence changes need no deploy:
0015   epicprod_live_channel        (default 'epicprod-live')
0016   epicprod_live_min_sublevel   (default 'normal')
0017   epicprod_live_poll_seconds   (default 30)
0018
0019 Run: manage.py publish_epicprod_live   (systemd unit swf-epicprod-live)
0020 """
0021 import logging
0022 import os
0023 import time
0024 from datetime import timezone as dt_timezone
0025
0026 import requests
0027 from django.core.management.base import BaseCommand
0028 from django.utils import timezone
0029
0030 logger = logging.getLogger(__name__)
0031
0032 MM_URL = os.environ.get('MATTERMOST_URL', 'chat.epic-eic.org')
0033 # Post as the dedicated 'epicprod' identity; fall back to the DISpatcher bot
0034 # token (posts then carry the DISpatcher face) until EPICPROD_LIVE_TOKEN is set.
0035 MM_TOKEN = (os.environ.get('EPICPROD_LIVE_TOKEN')
0036             or os.environ.get('MATTERMOST_TOKEN', ''))
0037 MM_TEAM = os.environ.get('MATTERMOST_TEAM', 'main')
0038
0039
0040 def _link_base():
0041     """Open-face link base so event links work for the whole
0042     collaboration: env override, else the external-face configuration
0043     point plus the production path."""
0044     env = os.environ.get('EPICPROD_LIVE_LINK_BASE')
0045     if env:
0046         return env.rstrip('/')
0047     from monitor_app.models import external_face_base_url
0048     return f"{external_face_base_url()}/prod"
0049
0050 STATE_KEY = 'epicprod_live_last_id'
0051 DEFAULT_CHANNEL = 'epicprod-live'
0052 DEFAULT_MIN_SUBLEVEL = 'normal'
0053 DEFAULT_POLL_SECONDS = 30
0054 BATCH_MAX = 20          # per-cycle post cap; overflow is summarized, not dropped
0055 HTTP_TIMEOUT = 15
0056
0057
0058 class Command(BaseCommand):
0059     help = "Tail the epicprod action stream and post live events to Mattermost."
0060
0061     def handle(self, *args, **options):
0062         if not MM_TOKEN:
0063             self.stderr.write("EPICPROD_LIVE_TOKEN/MATTERMOST_TOKEN not set; cannot publish")
0064             raise SystemExit(2)
0065         self.session = requests.Session()
0066         self.session.headers['Authorization'] = f'Bearer {MM_TOKEN}'
0067         self.base = f'https://{MM_URL}/api/v4'
0068         me = self._get('/users/me')
0069         self.user_id = me['id']
0070         logger.info("posting as @%s", me.get('username'))
0071         self.team_id = self._get(f'/teams/name/{MM_TEAM}')['id']
0072         self.channel_name = ''
0073         self.channel_id = ''
0074         self._init_high_water()
0075         logger.info("epicprod-live publisher started (team %s)", MM_TEAM)
0076         while True:
0077             poll = DEFAULT_POLL_SECONDS
0078             try:
0079                 poll = self._cycle()
0080             except Exception:
0081                 logger.exception("publish cycle failed; continuing")
0082             time.sleep(max(int(poll), 5))
0083
0084     # -- one cycle -----------------------------------------------------------
0085
0086     def _cycle(self):
0087         from monitor_app.epicprod_logging import (
0088             SUBLEVEL_VALUES, live_stream_q)
0089         from monitor_app.models import AppLog, SysConfig
0090
0091         # get_setting seeds a missing key into the SysConfig document — the
0092         # knobs in use are always visible on the System page, never hidden
0093         # behind a code default. An explicitly set but unusable value is a
0094         # failure and is logged, never silently replaced.
0095         channel = str(SysConfig.get_setting(
0096             'epicprod_live_channel', DEFAULT_CHANNEL) or '')
0097         if not channel:
0098             logger.warning("epicprod_live_channel is blank in SysConfig; "
0099                            "using default %r", DEFAULT_CHANNEL)
0100             channel = DEFAULT_CHANNEL
0101         min_sublevel = str(SysConfig.get_setting(
0102             'epicprod_live_min_sublevel', DEFAULT_MIN_SUBLEVEL) or '')
0103         if min_sublevel not in SUBLEVEL_VALUES:
0104             logger.warning("epicprod_live_min_sublevel %r is not one of %s; "
0105                            "using default %r", min_sublevel,
0106                            list(SUBLEVEL_VALUES), DEFAULT_MIN_SUBLEVEL)
0107             min_sublevel = DEFAULT_MIN_SUBLEVEL
0108         poll = SysConfig.get_setting(
0109             'epicprod_live_poll_seconds', DEFAULT_POLL_SECONDS)
0110         try:
0111             poll = int(poll)
0112         except (TypeError, ValueError):
0113             logger.warning("epicprod_live_poll_seconds %r is not an integer; "
0114                            "using default %r", poll, DEFAULT_POLL_SECONDS)
0115             poll = DEFAULT_POLL_SECONDS
0116
0117         if channel != self.channel_name:      # first cycle or painless rename
0118             self.channel_id = self._get(
0119                 f'/teams/{self.team_id}/channels/name/{channel}')['id']
0120             self.channel_name = channel
0121             try:                               # self-join (idempotent, public channel)
0122                 self.session.post(
0123                     f'{self.base}/channels/{self.channel_id}/members',
0124                     json={'user_id': self.user_id},
0125                     timeout=HTTP_TIMEOUT).raise_for_status()
0126             except Exception:
0127                 logger.warning("could not self-join #%s", channel)
0128             logger.info("publishing to #%s (%s)", channel, self.channel_id)
0129
0130         last_id = self._get_high_water()
0131         rows = list(
0132             AppLog.objects.filter(live_stream_q(min_sublevel), id__gt=last_id)
0133             .order_by('id')[:BATCH_MAX + 1]
0134         )
0135         overflow = len(rows) > BATCH_MAX
0136         for row in rows[:BATCH_MAX]:
0137             self._post(self._format(row))
0138             self._set_high_water(row.id)       # per-record: no replay, no skip
0139         if overflow:
0140             newest = (AppLog.objects.filter(live_stream_q(min_sublevel))
0141                       .order_by('-id').values_list('id', flat=True).first())
0142             skipped = AppLog.objects.filter(
0143                 live_stream_q(min_sublevel), id__gt=self._get_high_water(),
0144                 id__lte=newest).count()
0145             self._post(f"… and {skipped} more events this cycle — see the "
0146                        f"[live view]({_link_base()}/logs/?app_name=epicprod&live=1)")
0147             self._set_high_water(newest)
0148         return poll
0149
0150     # -- formatting ----------------------------------------------------------
0151
0152     def _format(self, row):
0153         extra = row.extra_data if isinstance(row.extra_data, dict) else {}
0154         action = extra.get('action') or row.funcname or 'action'
0155         outcome = str(extra.get('outcome') or '')
0156         if action == 'assessment_register' and outcome == 'ok':
0157             notice = self._format_assessment(row, extra)
0158             if notice:
0159                 return notice
0160         subject = ':'.join(x for x in (extra.get('subject_type'),
0161                                        extra.get('subject_key')) if x)
0162         username = str(extra.get('username') or '')
0163         reason = str(extra.get('reason') or '')
0164         dur = extra.get('duration_ms')
0165         stamp = timezone.localtime(row.timestamp).strftime('%H:%M')
0166
0167         parts = [f"`{stamp}`", f"**{action}**"]
0168         if subject:
0169             parts.append(subject)
0170         parts.append(f"{row.instance_name}")
0171         if username:
0172             parts.append(f"by {username}")
0173         if outcome and outcome != 'ok':
0174             parts.append(f"⚠️ **{outcome.upper()}**")
0175         if reason:
0176             parts.append(reason)
0177         summary = str(extra.get('summary') or '')
0178         if summary:
0179             parts.append(summary)
0180         if isinstance(dur, (int, float)):
0181             parts.append(f"{dur / 1000:.1f} s")
0182         parts.append(f"[record]({_link_base()}/logs/{row.id}/)")
0183         return ' · '.join(parts)
0184
0185     def _format_assessment(self, row, extra):
0186         """Linked publication notice; never duplicate the report body."""
0187         title = ' '.join(str(extra.get('report_title') or '').split())
0188         path = str(extra.get('report_path') or '').strip()
0189         if not title or not path.startswith('/ai/assessments/'):
0190             return ''
0191         url = f"{_link_base()}/{path.lstrip('/')}"
0192         stamp = timezone.localtime(row.timestamp).strftime('%H:%M %Z')
0193         subject_type = str(extra.get('subject_type') or '').strip()
0194         subject_key = str(extra.get('subject_key') or '').strip()
0195         subject = ''
0196         if subject_type and subject_key:
0197             subject = f'{subject_type.replace("_", " ").title()} {subject_key}'
0198         elif subject_key:
0199             subject = subject_key
0200         kind = str(extra.get('assessment_kind') or '').strip().lower()
0201         if kind == 'nightly':
0202             kind = 'daily'
0203         kind_label = kind.replace('_', ' ').title()
0204         verdict = str(extra.get('verdict') or '').strip()
0205         publication = (f'{kind_label} AI assessment published'
0206                        if kind_label else 'AI assessment published')
0207         parts = [f'`{stamp}`', f'**{publication}**']
0208         if subject:
0209             parts.append(subject)
0210         if verdict:
0211             parts.append(f'Verdict: **{verdict.capitalize()}**')
0212         parts.append(f'[record]({_link_base()}/logs/{row.id}/)')
0213         return f'### [{title}]({url})\n' + ' · '.join(parts)
0214
0215     # -- plumbing ------------------------------------------------------------
0216
0217     def _get(self, path):
0218         r = self.session.get(self.base + path, timeout=HTTP_TIMEOUT)
0219         r.raise_for_status()
0220         return r.json()
0221
0222     def _post(self, message):
0223         r = self.session.post(self.base + '/posts', timeout=HTTP_TIMEOUT,
0224                               json={'channel_id': self.channel_id,
0225                                     'message': message})
0226         r.raise_for_status()
0227
0228     def _init_high_water(self):
0229         """Start at the current stream head — never replay history."""
0230         from monitor_app.models import AppLog, PersistentState
0231         state = PersistentState.get_state()
0232         if STATE_KEY not in state:
0233             head = (AppLog.objects.filter(app_name='epicprod')
0234                     .order_by('-id').values_list('id', flat=True).first()) or 0
0235             PersistentState.update_state({STATE_KEY: head})
0236             logger.info("initialized high-water mark at %s", head)
0237
0238     def _get_high_water(self):
0239         from monitor_app.models import PersistentState
0240         return int(PersistentState.get_state().get(STATE_KEY) or 0)
0241
0242     def _set_high_water(self, value):
0243         from monitor_app.models import PersistentState
0244         PersistentState.update_state({STATE_KEY: int(value)})