| """Post-processing review-risk flags for accepted classifications.""" |
|
|
| from __future__ import annotations |
|
|
| from gcmd_classifier.models import ( |
| ClassificationFinalStatus, |
| ClassificationRecord, |
| OutputWarning, |
| SupportType, |
| ) |
|
|
| REVIEW_RECOMMENDED_WEAK_SUPPORT = "REVIEW_RECOMMENDED_WEAK_SUPPORT" |
|
|
|
|
| def flag_review_risk( |
| classifications: tuple[ClassificationRecord, ...], |
| ) -> tuple[ClassificationRecord, ...]: |
| """Mark accepted classifications that should receive manual scientific review.""" |
| return tuple(_flag_record(record) for record in classifications) |
|
|
|
|
| def _flag_record(record: ClassificationRecord) -> ClassificationRecord: |
| if not _requires_review(record): |
| return record |
| warning = OutputWarning( |
| code=REVIEW_RECOMMENDED_WEAK_SUPPORT, |
| message="Manual scientific review is recommended because support is weak or inferred.", |
| stage="review_flagging", |
| details={ |
| "level": record.level, |
| "support_type": None if record.support_type is None else record.support_type.value, |
| "confidence_final": _confidence_final(record), |
| }, |
| ) |
| warnings = record.warnings |
| if not any(existing.code == REVIEW_RECOMMENDED_WEAK_SUPPORT for existing in warnings): |
| warnings = (*warnings, warning) |
| return record.model_copy(update={"review_required": True, "warnings": warnings}) |
|
|
|
|
| def _requires_review(record: ClassificationRecord) -> bool: |
| if record.final_status is not ClassificationFinalStatus.ACCEPTED: |
| return False |
| if record.support_type is None: |
| return False |
| confidence_final = _confidence_final(record) |
| if record.level == "Topic" and record.support_type in { |
| SupportType.INFERRED, |
| SupportType.MIXED, |
| }: |
| return True |
| if record.support_type is SupportType.INFERRED and record.level in { |
| "Variable_Level_2", |
| "Variable_Level_3", |
| }: |
| return True |
| if ( |
| record.support_type is SupportType.INFERRED |
| and confidence_final is not None |
| and confidence_final < 0.75 |
| ): |
| return True |
| return ( |
| record.support_type is SupportType.MIXED |
| and confidence_final is not None |
| and confidence_final < 0.70 |
| ) |
|
|
|
|
| def _confidence_final(record: ClassificationRecord) -> float | None: |
| return None if record.confidence is None else record.confidence.final |
|
|
