from __future__ import annotations AUTO_ATTACHMENT_SIZE_LIMIT_BYTES = 50 * 1024 * 1024 def _attachment_size_summary(attachment_refs: list[dict]) -> dict[str, object]: total_known_bytes = 0 unknown_size_count = 0 considered_count = 0 for item in attachment_refs or []: considered_count += 1 size = item.get('size') try: total_known_bytes += max(0, int(size)) except (TypeError, ValueError): unknown_size_count += 1 return { 'considered_count': considered_count, 'total_known_bytes': total_known_bytes, 'unknown_size_count': unknown_size_count, 'auto_download_limit_bytes': AUTO_ATTACHMENT_SIZE_LIMIT_BYTES, } def _should_attempt_automatic_attachment_analysis( *, args, context: dict, attachment_refs: list[dict], clinic_results: list[dict], context_requires_attachment_analysis, analyze_missing_evidence_reasons, ) -> tuple[bool, dict[str, object]]: if not attachment_refs: return False, { 'mode': 'none', 'reason': 'no_attachment_references', } if bool(getattr(args, 'skip_attachments', False)): return False, { 'mode': 'skipped', 'reason': 'attachments_skipped_by_flag', } size_summary = _attachment_size_summary(attachment_refs) total_known_bytes = int(size_summary.get('total_known_bytes') or 0) unknown_size_count = int(size_summary.get('unknown_size_count') or 0) if bool(getattr(args, 'download_attachments', False)): return True, { 'mode': 'forced', 'reason': 'attachments_forced_by_flag', **size_summary, } missing_evidence_reasons = analyze_missing_evidence_reasons(context, clinic_results) if not missing_evidence_reasons: return False, { 'mode': 'skipped', 'reason': 'clinic_evidence_sufficient', 'missing_evidence_reasons': [], **size_summary, } if not context_requires_attachment_analysis(context.get('routing_context_text') or '', attachment_refs): return False, { 'mode': 'skipped', 'reason': 'attachments_not_indicated_for_analysis', 'missing_evidence_reasons': missing_evidence_reasons, **size_summary, } if unknown_size_count > 0: return False, { 'mode': 'blocked', 'reason': 'attachment_size_unknown', 'missing_evidence_reasons': missing_evidence_reasons, **size_summary, } if total_known_bytes > AUTO_ATTACHMENT_SIZE_LIMIT_BYTES: return False, { 'mode': 'blocked', 'reason': 'attachment_size_exceeds_auto_limit', 'missing_evidence_reasons': missing_evidence_reasons, **size_summary, } return True, { 'mode': 'auto', 'reason': 'clinic_evidence_insufficient_but_attachment_size_within_limit', 'missing_evidence_reasons': missing_evidence_reasons, **size_summary, } def resolve_analyze_context( args, *, is_pingkai_issue, is_pingkai_report_issue, build_pingkai_context, build_jira_context, build_grafana_context, build_clinic_context, apply_window_to_clinic_link, is_grafana_link, prepend_reference, ): issue_ref = str(args.issue or '').strip() grafana_link = str(getattr(args, 'grafana_link', '') or '').strip() clinic_link = str(getattr(args, 'clinic_link', '') or '').strip() if clinic_link: clinic_link = apply_window_to_clinic_link(clinic_link, getattr(args, 'from_ts', ''), getattr(args, 'to_ts', '')) input_ref = issue_ref or grafana_link or clinic_link if not input_ref: raise ValueError('issue or an explicit monitoring link is required') if issue_ref: if is_pingkai_issue(issue_ref): if not is_pingkai_report_issue(issue_ref): raise ValueError(f'unsupported Pingkai analyze reference: {issue_ref}. Use ticket-xxxx or TICKET-xxxx.') context = build_pingkai_context(issue_ref) else: context = build_jira_context(issue_ref) if clinic_link: context['clinic_references'] = prepend_reference(context.get('clinic_references') or [], clinic_link, 'cli-flag') if grafana_link: context['grafana_references'] = prepend_reference(context.get('grafana_references') or [], grafana_link, 'cli-flag') return context if grafana_link or is_grafana_link(issue_ref): return build_grafana_context(grafana_link or issue_ref, args) if clinic_link: return build_clinic_context(clinic_link, args) if is_pingkai_issue(issue_ref): if not is_pingkai_report_issue(issue_ref): raise ValueError(f'unsupported Pingkai analyze reference: {issue_ref}. Use ticket-xxxx or TICKET-xxxx.') return build_pingkai_context(issue_ref) return build_jira_context(issue_ref) def execute_analyze_context( context: dict, args, *, run_grafana_analyze, run_clinic_analyze, run_clinic_bundle_download, context_requires_attachment_analysis, analyze_missing_evidence_reasons, run_attachment_downloads, build_analyze_payload, ): issue_ref = str(context.get('issue_ref') or '').strip() clinic_refs = context.get('clinic_references') or [] clinic_download_refs = context.get('clinic_download_references') or [] attachment_refs = context.get('attachment_references') or [] max_clinic_links = max(0, int(getattr(args, 'max_clinic_links', 1) or 0)) selected_refs = clinic_refs[:max_clinic_links] if max_clinic_links else [] selected_download_refs = clinic_download_refs[:max_clinic_links] if max_clinic_links else [] clinic_results = [] grafana_results = [] bundle_results = [] attachment_results = [] if not getattr(args, 'skip_clinic', False): for ref in selected_refs: result = run_clinic_analyze(ref['link'], args) result['source'] = ref.get('source') or '' result['author'] = ref.get('author') or '' result['created'] = ref.get('created') or '' clinic_results.append(result) if not getattr(args, 'skip_clinic_download', False): preferred_status_link = '' if selected_refs: preferred_status_link = str((selected_refs[0] or {}).get('link') or '').strip() for ref in selected_download_refs: result = run_clinic_bundle_download( ref['link'], issue_ref, args, status_link=preferred_status_link, ) result['source'] = ref.get('source') or '' result['author'] = ref.get('author') or '' result['created'] = ref.get('created') or '' bundle_results.append(result) grafana_refs = context.get('grafana_references') or [] for ref in grafana_refs[:1]: result = run_grafana_analyze(ref['link'], args) result['source'] = ref.get('source') or '' result['author'] = ref.get('author') or '' result['created'] = ref.get('created') or '' grafana_results.append(result) should_download_attachments, attachment_policy = _should_attempt_automatic_attachment_analysis( args=args, context=context, attachment_refs=attachment_refs, clinic_results=clinic_results, context_requires_attachment_analysis=context_requires_attachment_analysis, analyze_missing_evidence_reasons=analyze_missing_evidence_reasons, ) context['attachment_policy'] = attachment_policy if should_download_attachments and attachment_refs: attachment_results = run_attachment_downloads(context, args) payload = build_analyze_payload( context, clinic_refs, selected_refs, clinic_results, grafana_results, clinic_download_refs, selected_download_refs, bundle_results, attachment_refs, attachment_results, ) attachment_policy = context.get('attachment_policy') or {} attachment_error = '' if attachment_policy.get('mode') == 'blocked': total_known_bytes = int(attachment_policy.get('total_known_bytes') or 0) total_known_mb = total_known_bytes / (1024 * 1024) limit_mb = int((attachment_policy.get('auto_download_limit_bytes') or AUTO_ATTACHMENT_SIZE_LIMIT_BYTES) / (1024 * 1024)) reason = str(attachment_policy.get('reason') or '').strip() if reason == 'attachment_size_exceeds_auto_limit': attachment_error = ( f'Attachment-backed analysis was not run automatically because the attachment set is {total_known_mb:.1f} MB, ' f'which exceeds the {limit_mb} MB auto-download limit. Re-run with --download-attachments to force attachment analysis.' ) elif reason == 'attachment_size_unknown': attachment_error = ( 'Attachment-backed analysis was not run automatically because one or more attachment sizes are unknown. ' 'Re-run with --download-attachments to force attachment analysis.' ) if attachment_error and not payload.get('fatal_error'): payload['fatal_error'] = attachment_error fatal_error = str(payload.get('fatal_error') or '').strip() return payload, (1 if fatal_error else 0)