""" Domain models for Criteria (Inclusion & Exclusion). Ported from AI_Screening_UI criteria_workspace/models.py. Converted from dataclasses to Pydantic BaseModel. """ from __future__ import annotations from datetime import datetime from enum import Enum from typing import Optional from pydantic import BaseModel, Field, field_validator class CriterionSource(str, Enum): """Source of a criterion (human or AI-generated).""" human = "human" ai_generated = "ai_generated" class CriterionType(str, Enum): """Type of criterion (inclusion or exclusion).""" include = "include" exclude = "exclude" # PICO-style clusters for categorizing eligibility criteria. DEFAULT_CATEGORIES = [ "Population", "Intervention/Exposure", "Comparator", "Outcome", "Study Design", "Setting/Geography", "Language", "Date", "Other", ] # Valid PICO(S) categories for structured output schemas PICO_CATEGORIES = [ "Population", "Intervention/Exposure", "Comparator", "Outcome", "Study Design", "Publication Type", "Language", "Geography/Setting", "Time Period", "Other", ] # RQ link relevance types RQ_RELEVANCE_TYPES = ["direct", "indirect", "contextual"] class ExclusionCriterion(BaseModel): """ A criterion for systematic review screening (inclusion or exclusion). Note: Class name kept as ExclusionCriterion for backward compatibility. The criterion_type field distinguishes between 'include' and 'exclude'. is_active: True = active (used for screening), False = inactive (on the shelf). """ id: int project_id: int category: str text: str description: Optional[str] = None source: CriterionSource = CriterionSource.human is_active: bool = True criterion_type: CriterionType = CriterionType.exclude ai_confidence: Optional[float] = None ai_rationale: Optional[str] = None created_at: datetime = Field(default_factory=datetime.now) created_by: str = "system" modified_at: Optional[datetime] = None notes: Optional[str] = None title_abstract_assessable: bool = True research_question_links: list[dict] = Field(default_factory=list) model_config = {"from_attributes": True} @field_validator("ai_confidence") @classmethod def validate_confidence(cls, v: Optional[float]) -> Optional[float]: if v is not None and not (0.0 <= v <= 1.0): raise ValueError(f"ai_confidence must be between 0.0 and 1.0, got {v}") return v @property def confidence_percentage(self) -> Optional[int]: """AI confidence as percentage (0-100).""" if self.ai_confidence is None: return None return int(self.ai_confidence * 100) @property def is_ai_generated(self) -> bool: """Check if criterion was AI-generated.""" return self.source == CriterionSource.ai_generated class CriterionMatch(BaseModel): """Result of evaluating one criterion against one paper.""" criterion_id: int criterion_text: str criterion_category: str matched: bool reasoning: Optional[str] = None class TestResult(BaseModel): """ Result of testing eligibility criteria against a screened paper. Tracks ALL criterion matches for multi-criteria evaluation. """ __test__ = False # Not a pytest test class paper_id: int | str title: str abstract: str human_decision: str human_exclusion_reason: Optional[str] = None ai_decision: str = "include" criterion_matches: list[CriterionMatch] = Field(default_factory=list) @property def matched_criteria(self) -> list[CriterionMatch]: """Criteria that matched (triggered exclusion).""" return [m for m in self.criterion_matches if m.matched] @property def is_agreement(self) -> bool: """Check if human and AI decisions agree.""" return self.human_decision == self.ai_decision @property def is_mismatch(self) -> bool: return not self.is_agreement @property def is_gap(self) -> bool: """True when human excluded but AI included (no criterion matched).""" return self.human_decision == "exclude" and self.ai_decision == "include" @property def gap_categories(self) -> set[str]: """Categories that failed to match (for gap analysis).""" if not self.is_gap: return set() return {m.criterion_category for m in self.criterion_matches} class UndoAction(BaseModel): """Represents an undoable action in the criteria workspace.""" action_type: str # "activate", "deactivate", "add", "update", "delete" criterion_id: int previous_state: dict description: str # --- Consolidation & De-Duplicate Models --- class DuplicateGroupType(str, Enum): """Type of duplicate group detected.""" exact_duplicate = "exact_duplicate" near_duplicate = "near_duplicate" class DuplicateGroup(BaseModel): """ A group of duplicate or near-duplicate criteria detected by AI. The AI recommends a primary criterion to keep; others can be deactivated after user approval. """ group_type: DuplicateGroupType category: str criterion_ids: list[int] recommended_primary_id: int merge_rationale: str ai_confidence: float criteria_texts: list[str] = Field(default_factory=list) @property def confidence_percentage(self) -> int: return int(self.ai_confidence * 100) @property def duplicate_count(self) -> int: """Number of duplicate criteria (excluding primary).""" return len(self.criterion_ids) - 1 @property def ids_to_deactivate(self) -> list[int]: """IDs of criteria to deactivate (all except primary).""" return [cid for cid in self.criterion_ids if cid != self.recommended_primary_id] class ConsolidationProposal(BaseModel): """ A proposal to merge multiple granular criteria into a single broader criterion. Requires explicit user approval before applying. """ category: str criterion_ids: list[int] proposed_merged_criterion: str proposed_description: str proposed_type: CriterionType merge_rationale: str ai_confidence: float criteria_texts: list[str] = Field(default_factory=list) @property def confidence_percentage(self) -> int: return int(self.ai_confidence * 100) @property def criteria_count(self) -> int: return len(self.criterion_ids) def to_exclusion_criterion(self, project_id: int, criterion_id: int = 0) -> ExclusionCriterion: """Convert proposal to ExclusionCriterion.""" return ExclusionCriterion( id=criterion_id, project_id=project_id, category=self.category, text=self.proposed_merged_criterion, description=self.proposed_description, source=CriterionSource.ai_generated, is_active=True, criterion_type=self.proposed_type, ai_confidence=self.ai_confidence, ai_rationale=self.merge_rationale, notes=( f"Consolidates {self.criteria_count} criteria: " f"{'; '.join(self.criteria_texts[:3])}" f"{'...' if len(self.criteria_texts) > 3 else ''}" ), ) class ConsolidatedCriterion(BaseModel): """ Result of consolidating multiple criteria into one LLM-optimized criterion. """ text: str category: str criterion_type: CriterionType original_ids: list[int] = Field(default_factory=list) original_texts: list[str] = Field(default_factory=list) confidence: float = 0.8 rationale: Optional[str] = None @property def original_count(self) -> int: return len(self.original_ids) @property def confidence_percentage(self) -> int: return int(self.confidence * 100) def to_exclusion_criterion(self, project_id: int, criterion_id: int = 0) -> ExclusionCriterion: """Convert to ExclusionCriterion.""" return ExclusionCriterion( id=criterion_id, project_id=project_id, category=self.category, text=self.text, description=f"Consolidated from {self.original_count} criteria", source=CriterionSource.ai_generated, is_active=True, criterion_type=self.criterion_type, ai_confidence=self.confidence, ai_rationale=self.rationale, notes=( f"Replaces: {'; '.join(self.original_texts[:3])}" f"{'...' if len(self.original_texts) > 3 else ''}" ), ) class ConsolidationResult(BaseModel): """ Complete result of the consolidate & de-duplicate detection. """ duplicate_groups: list[DuplicateGroup] = Field(default_factory=list) consolidation_proposals: list[ConsolidationProposal] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) @property def has_duplicates(self) -> bool: return len(self.duplicate_groups) > 0 @property def has_proposals(self) -> bool: return len(self.consolidation_proposals) > 0 @property def has_warnings(self) -> bool: return len(self.warnings) > 0 @property def has_any_results(self) -> bool: return self.has_duplicates or self.has_proposals @property def total_duplicate_criteria(self) -> int: return sum(g.duplicate_count for g in self.duplicate_groups) @property def total_criteria_to_consolidate(self) -> int: return sum(p.criteria_count for p in self.consolidation_proposals)