File indexing completed on 2026-07-21 08:46:24
0001 """AI assessment helpers for production objects."""
0002 from datetime import datetime
0003
0004 import bleach
0005 import markdown
0006 from django.db import transaction
0007 from django.utils.safestring import mark_safe
0008
0009 from monitor_app.utils import format_datetime
0010
0011
0012 AI_CONTENT_IDS_KEY = 'ai_content_ids'
0013 CORUN_PAGE_GROUP_IDS_KEY = 'corun_page_group_ids'
0014 CORUN_ASSESSMENT_SECTION = 'epicprod.assessment'
0015 AI_CONTENT_RETRIEVE_TOOL = 'epic_get_ai_content'
0016 AI_CONTENT_QUALITY_KEY = 'quality'
0017
0018
0019 AI_CONTENT_QUALITY_VALUES = ('wrong', 'poor', 'ok', 'good')
0020 AI_CONTENT_COMMENT_KEY = 'comment'
0021 _MARKDOWN_EXTENSIONS = ['extra', 'sane_lists']
0022 _ALLOWED_TAGS = set(bleach.sanitizer.ALLOWED_TAGS) | {
0023 'p', 'pre', 'code', 'br', 'hr',
0024 'h1', 'h2', 'h3', 'h4', 'h5', 'h6',
0025 'table', 'thead', 'tbody', 'tr', 'th', 'td',
0026 'ul', 'ol', 'li',
0027 }
0028 _ALLOWED_ATTRIBUTES = {
0029 **bleach.sanitizer.ALLOWED_ATTRIBUTES,
0030 'a': ['href', 'title', 'rel'],
0031 }
0032
0033
0034 def render_assessment_markdown(text, *, omit_leading_title=''):
0035 """Render assessment Markdown to sanitized HTML."""
0036 source = text or ''
0037 title = ' '.join(str(omit_leading_title or '').split())
0038 if title:
0039 lines = source.splitlines()
0040 first = next((i for i, line in enumerate(lines) if line.strip()), None)
0041 if first is not None and lines[first].lstrip().startswith('# '):
0042 heading = lines[first].lstrip()[2:].strip().rstrip('#').strip()
0043 if ' '.join(heading.split()) == title:
0044 del lines[first]
0045 if first < len(lines) and not lines[first].strip():
0046 del lines[first]
0047 source = '\n'.join(lines)
0048 html = markdown.markdown(
0049 source,
0050 extensions=_MARKDOWN_EXTENSIONS,
0051 output_format='html5',
0052 )
0053 clean = bleach.clean(
0054 html,
0055 tags=_ALLOWED_TAGS,
0056 attributes=_ALLOWED_ATTRIBUTES,
0057 protocols=['http', 'https', 'mailto'],
0058 strip=True,
0059 )
0060 return mark_safe(clean)
0061
0062
0063 def ai_content_ids(data):
0064 """Return ordered AIContent ids from a model JSON field."""
0065 if not isinstance(data, dict):
0066 return []
0067 ids = []
0068 for raw in data.get(AI_CONTENT_IDS_KEY) or []:
0069 try:
0070 value = int(raw)
0071 except (TypeError, ValueError):
0072 continue
0073 if value not in ids:
0074 ids.append(value)
0075 return ids
0076
0077
0078 def corun_page_group_ids(data):
0079 """Return ordered corun Page group UUID strings from a model JSON field."""
0080 if not isinstance(data, dict):
0081 return []
0082 ids = []
0083 for raw in data.get(CORUN_PAGE_GROUP_IDS_KEY) or []:
0084 value = str(raw or '').strip()
0085 if value and value not in ids:
0086 ids.append(value)
0087 return ids
0088
0089
0090 def _display_time(value):
0091 if isinstance(value, datetime):
0092 return format_datetime(value)
0093 if isinstance(value, str):
0094 try:
0095 return format_datetime(datetime.fromisoformat(value))
0096 except ValueError:
0097 return value
0098 return value or ''
0099
0100
0101 def _panda_task_parts(subject_key, subject_label):
0102 display_key = subject_key
0103 display_label = subject_label
0104 if subject_key and not str(subject_key).isdigit():
0105 try:
0106 from pcs.models import PandaTasks
0107 row = (
0108 PandaTasks.objects
0109 .filter(task_name__in=[subject_key, subject_label])
0110 .order_by('-jedi_task_id')
0111 .first()
0112 )
0113 if row:
0114 display_key = str(row.jedi_task_id or subject_key)
0115 display_label = row.task_name or subject_label
0116 except Exception:
0117 pass
0118 return str(display_key or '').strip(), str(display_label or '').strip()
0119
0120
0121 def _subject_parts(subject_type, subject_key, subject_label):
0122 subject_kind = subject_type
0123 subject_id = subject_key
0124 subject_name = subject_label
0125 subject_display = subject_label or subject_key
0126 if subject_type == 'panda_task' and subject_key:
0127 subject_kind = 'PanDA task'
0128 subject_id, subject_name = _panda_task_parts(subject_key, subject_label)
0129 subject_display = f'{subject_kind} {subject_id}'
0130 if subject_name and subject_name != subject_id:
0131 subject_display = f'{subject_display}: {subject_name}'
0132 elif subject_type == 'panda_job' and subject_key:
0133 subject_kind = 'PanDA job'
0134 subject_id = subject_key
0135 subject_name = subject_label
0136 subject_display = f'PanDA job {subject_key}'
0137 if subject_label and subject_key not in subject_label:
0138 subject_display = f'{subject_display}: {subject_label}'
0139 elif subject_type == 'campaign_task':
0140 subject_kind = 'Campaign task'
0141 subject_id = subject_key
0142 subject_name = subject_label
0143 elif subject_type == 'panda_queue':
0144 subject_kind = 'PanDA queue'
0145 subject_id = subject_key
0146 subject_name = subject_label
0147 return subject_kind, subject_id, subject_name, subject_display
0148
0149
0150 def _assessment_item(*, item_id, storage, username, ai, assessment, created_at,
0151 subject_type, subject_key, subject_label, subject_url,
0152 quality='', comment='', corun_page_group_id='',
0153 title='', assessment_kind='', verdict='', quarantined=False,
0154 render_body=True):
0155 subject_kind, subject_id, subject_name, subject_display = _subject_parts(
0156 subject_type, subject_key, subject_label)
0157
0158
0159
0160 kind_display = 'daily' if assessment_kind == 'nightly' else assessment_kind
0161 return {
0162 'id': item_id,
0163 'storage': storage,
0164 'corun_page_group_id': corun_page_group_id,
0165 'username': username,
0166 'ai': ai,
0167 'quality': quality if quality in AI_CONTENT_QUALITY_VALUES else '',
0168 'comment': comment,
0169 'assessment': assessment,
0170 'assessment_html': (render_assessment_markdown(assessment)
0171 if render_body else ''),
0172 'created_at': created_at,
0173 'created_display': _display_time(created_at),
0174 'subject_type': subject_type,
0175 'subject_key': subject_key,
0176 'subject_label': subject_label,
0177 'subject_kind': subject_kind,
0178 'subject_id': subject_id,
0179 'subject_name': subject_name,
0180 'subject_display': subject_display,
0181 'subject_url': subject_url,
0182 'title': title,
0183 'assessment_kind': kind_display,
0184 'verdict': verdict,
0185 'quarantined': quarantined,
0186 }
0187
0188
0189 def ai_content_items(rows, *, render_body=True):
0190 """Normalize legacy AIContent rows for display."""
0191 items = []
0192 for row in rows:
0193 assessment = str(getattr(row, 'assessment', '') or '').strip()
0194 if not assessment:
0195 continue
0196 subject_type = str(getattr(row, 'subject_type', '') or '').strip()
0197 subject_key = str(getattr(row, 'subject_key', '') or '').strip()
0198 subject_label = str(getattr(row, 'subject_label', '') or '').strip()
0199 created_at = getattr(row, 'created_at', '')
0200 data = getattr(row, 'data', None) or {}
0201 quality = str(data.get(AI_CONTENT_QUALITY_KEY) or '').strip()
0202 comment = str(data.get(AI_CONTENT_COMMENT_KEY) or '').strip()
0203 items.append(_assessment_item(
0204 item_id=row.pk,
0205 storage='legacy',
0206 username=str(getattr(row, 'username', '') or '').strip(),
0207 ai=str(getattr(row, 'ai', '') or '').strip(),
0208 quality=quality,
0209 comment=comment,
0210 assessment=assessment,
0211 created_at=created_at,
0212 subject_type=subject_type,
0213 subject_key=subject_key,
0214 subject_label=subject_label,
0215 subject_url=str(getattr(row, 'subject_url', '') or '').strip(),
0216 title=str(data.get('title') or '').strip(),
0217 assessment_kind=str(data.get('assessment_kind') or '').strip(),
0218 verdict=str(data.get('verdict') or '').strip(),
0219 quarantined=data.get('quarantined') is True,
0220 render_body=render_body,
0221 ))
0222 return items
0223
0224
0225 def corun_page_items(pages, *, render_body=True):
0226 """Normalize corun Page API payloads for display."""
0227 items = []
0228 for page in pages or []:
0229 if not isinstance(page, dict):
0230 continue
0231 data = page.get('data') if isinstance(page.get('data'), dict) else {}
0232 assessment = str(page.get('content') or '').strip()
0233 if not assessment:
0234 continue
0235 items.append(_assessment_item(
0236 item_id='',
0237 storage='corun',
0238 corun_page_group_id=str(page.get('group_id') or ''),
0239 username=str(
0240 data.get('created_by_user')
0241 or data.get('username')
0242 or data.get('submitted_by')
0243 or ''
0244 ).strip(),
0245 ai=str(data.get('ai') or data.get('model') or '').strip(),
0246 quality=str(data.get(AI_CONTENT_QUALITY_KEY) or '').strip(),
0247 comment=str(data.get(AI_CONTENT_COMMENT_KEY) or '').strip(),
0248 assessment=assessment,
0249 created_at=str(page.get('created_at') or ''),
0250 subject_type=str(data.get('subject_type') or '').strip(),
0251 subject_key=str(data.get('subject_key') or '').strip(),
0252 subject_label=str(data.get('subject_label') or '').strip(),
0253 subject_url=str(data.get('subject_url') or '').strip(),
0254 title=str(data.get('title') or page.get('title') or '').strip(),
0255 assessment_kind=str(data.get('assessment_kind') or '').strip(),
0256 verdict=str(data.get('verdict') or '').strip(),
0257 quarantined=data.get('quarantined') is True,
0258 render_body=render_body,
0259 ))
0260 return items
0261
0262
0263 def ai_content_summary(data):
0264 """JSON-serializable AIContent summaries for client-rendered pages."""
0265 out = []
0266 for item in ai_content_items(ai_content_for_json(data)) + corun_assessment_items_for_json(data):
0267 out.append({
0268 'id': item['id'],
0269 'storage': item['storage'],
0270 'corun_page_group_id': item['corun_page_group_id'],
0271 'username': item['username'],
0272 'ai': item['ai'],
0273 'quality': item['quality'],
0274 'comment': item['comment'],
0275 'assessment': item['assessment'],
0276 'created_at': item['created_display'] or str(item['created_at'] or ''),
0277 'subject_type': item['subject_type'],
0278 'subject_key': item['subject_key'],
0279 'subject_label': item['subject_label'],
0280 'subject_kind': item['subject_kind'],
0281 'subject_id': item['subject_id'],
0282 'subject_name': item['subject_name'],
0283 'subject_display': item['subject_display'],
0284 'subject_url': item['subject_url'],
0285 })
0286 return out
0287
0288
0289 def ai_content_retrieval_guidance(data):
0290 """Return MCP retrieval guidance for assessment pointers in a JSON field."""
0291 ids = ai_content_ids(data)
0292 page_group_ids = corun_page_group_ids(data)
0293 args = {}
0294 if ids:
0295 args['ids'] = ids
0296 if page_group_ids:
0297 args['corun_page_group_ids'] = page_group_ids
0298 return {
0299 'available': bool(ids or page_group_ids),
0300 'count': len(ids) + len(page_group_ids),
0301 'ids': ids,
0302 'corun_page_group_ids': page_group_ids,
0303 'retrieval': {
0304 'tool': AI_CONTENT_RETRIEVE_TOOL,
0305 'arguments': args,
0306 } if args else None,
0307 }
0308
0309
0310 def ai_content_for_json(data):
0311 """Fetch AIContent rows pointed to by a model JSON field."""
0312 ids = ai_content_ids(data)
0313 if not ids:
0314 return []
0315 from monitor_app.models import AIContent
0316 return list(
0317 AIContent.objects
0318 .filter(pk__in=ids)
0319 .order_by('-created_at', '-id')
0320 )
0321
0322
0323 def corun_assessment_items_for_json(data):
0324 """Fetch corun-backed assessment Pages pointed to by a model JSON field."""
0325 page_group_ids = corun_page_group_ids(data)
0326 if not page_group_ids:
0327 return []
0328 try:
0329 from ai.corun_client import CorunAPIError, CorunClient, corun_configured
0330 if not corun_configured():
0331 return []
0332 client = CorunClient()
0333 pages = []
0334 for page_group_id in page_group_ids:
0335 try:
0336 pages.append(client.get_page(page_group_id))
0337 except CorunAPIError:
0338 continue
0339 return corun_page_items(pages)
0340 except CorunAPIError:
0341 return []
0342
0343
0344 def assessment_presentation(data, *, title='AI Assessments'):
0345 """Return template-ready presentation data for assessment pointer JSON."""
0346 items = ai_content_items(ai_content_for_json(data)) + corun_assessment_items_for_json(data)
0347 return {
0348 'title': title,
0349 'items': items,
0350 'count': len(items),
0351 'has_assessments': bool(items),
0352 'quality_choices': AI_CONTENT_QUALITY_VALUES,
0353 }
0354
0355
0356 def append_ai_content_id(target_obj, json_field, content_id):
0357 """Append an AIContent id to ``target_obj.<json_field>``."""
0358 data = dict(getattr(target_obj, json_field) or {})
0359 ids = ai_content_ids(data)
0360 if int(content_id) not in ids:
0361 ids.append(int(content_id))
0362 data[AI_CONTENT_IDS_KEY] = ids
0363 setattr(target_obj, json_field, data)
0364 target_obj.save(update_fields=[json_field, 'updated_at'])
0365
0366
0367 def append_corun_page_group_id(target_obj, json_field, page_group_id):
0368 """Append a corun Page group id to ``target_obj.<json_field>``."""
0369 data = dict(getattr(target_obj, json_field) or {})
0370 ids = corun_page_group_ids(data)
0371 value = str(page_group_id or '').strip()
0372 if value and value not in ids:
0373 ids.append(value)
0374 data[CORUN_PAGE_GROUP_IDS_KEY] = ids
0375 setattr(target_obj, json_field, data)
0376 target_obj.save(update_fields=[json_field, 'updated_at'])
0377
0378
0379 def create_ai_content(*, subject_type, subject_key, username, ai, assessment,
0380 subject_label='', subject_url='', data=None,
0381 target_obj=None, target_json_field=None):
0382 """Create one AIContent row and optionally link it from object JSON.
0383
0384 The AI content itself is append-only. Followups create additional rows.
0385 If a local object is supplied, its JSON field gets an ``ai_content_ids``
0386 pointer in the same transaction.
0387 """
0388 from monitor_app.models import AIContent
0389 payload_data = dict(data or {})
0390 payload_data[AI_CONTENT_QUALITY_KEY] = ''
0391 payload_data[AI_CONTENT_COMMENT_KEY] = ''
0392 with transaction.atomic():
0393 row = AIContent.objects.create(
0394 subject_type=subject_type,
0395 subject_key=str(subject_key),
0396 subject_label=subject_label or '',
0397 subject_url=subject_url or '',
0398 username=username,
0399 ai=ai,
0400 assessment=assessment,
0401 data=payload_data,
0402 )
0403 if target_obj is not None and target_json_field:
0404 append_ai_content_id(target_obj, target_json_field, row.pk)
0405 return row
