from __future__ import annotations """Case items extraction and domain projection.""" from typing import Any from app.models.schemas import ( DossierCase, OperationsEscalationItem, OperationsEvidenceItem, ) class CasesBuilderMixin: """Case items extraction and domain projection mixin.""" def _build_cases_from_session(self, session: dict[str, Any]) -> list[DossierCase]: """Map one review session into operational cases.""" items: list[DossierCase] = [] project_code = str(session.get("project_code") or "") review_id = str(session.get("review_id") or "") for finding in self._ensure_list(session.get("findings")): case_id = self._build_case_id(review_id=review_id, finding=finding) assignments = self._ensure_list(finding.get("assignments")) primary_assignment = assignments[0] if assignments else {} items.append( DossierCase( case_id=case_id, case_code=str(finding.get("finding_code") or case_id), project_code=project_code, review_session_id=review_id or None, finding_id=str(finding.get("finding_id") or "") or None, finding_code=str(finding.get("finding_code") or "") or None, case_type=str(finding.get("finding_type") or "dossier_gap"), title=str(finding.get("title") or "Dossier case"), description=finding.get("description"), dossier_stage=finding.get("dossier_stage"), department_code=finding.get("responsible_department_code") or primary_assignment.get("assigned_department_code"), employee_code=primary_assignment.get("assigned_employee_code"), priority=self._derive_priority( finding.get("severity"), finding.get("impact_level"), finding.get("is_blocking"), ), severity=str(finding.get("severity") or "medium"), is_blocking=bool(finding.get("is_blocking") or False), current_status=str(finding.get("status") or "open"), opened_at=finding.get("detected_at"), closed_at=finding.get("resolved_at"), source_system="dossier_review", source_record_type="finding", source_record_id=str(finding.get("finding_id") or "") or None, readiness_impact=str(finding.get("impact_level") or "") or None, operational_impact=self._derive_operational_impact(finding), confidence_score=self._derive_confidence_score(finding), metadata=self._merge_metadata( finding.get("metadata"), { "review_code": session.get("review_code"), "project_name": session.get("project_name"), "supplement_status": finding.get("supplement_status"), "assignment_count": len(assignments), "post_close_actions": self._collect_post_close_actions( session=session, finding=finding ), "post_verification_actions": self._collect_post_verification_actions( session=session, finding=finding, ), }, ), evidences=self._build_evidences( project_code=project_code, review_id=review_id, case_id=case_id, finding=finding, ), escalations=self._build_escalations( project_code=project_code, review_id=review_id, case_id=case_id, finding=finding, session_actions=self._ensure_list(session.get("actions")), ), ) ) return items def _build_evidences( self, *, project_code: str, review_id: str, case_id: str, finding: dict[str, Any], ) -> list[OperationsEvidenceItem]: """Map dossier finding evidences into operations evidence records.""" items: list[OperationsEvidenceItem] = [] finding_code = str(finding.get("finding_code") or "") or None for index, evidence in enumerate( self._ensure_list(finding.get("evidences")), start=1 ): items.append( OperationsEvidenceItem( evidence_id=f"{case_id}:evidence:{index}", case_id=case_id, project_code=project_code, review_session_id=review_id or None, finding_code=finding_code, evidence_type=str( evidence.get("evidence_type") or "document_evidence" ), source_system="dossier_review", source_ref=evidence.get("source_uri") or evidence.get("source_name"), title=str( evidence.get("source_name") or evidence.get("rule_code") or "Evidence" ), excerpt=evidence.get("excerpt"), availability="available", strength=self._derive_evidence_strength( evidence.get("confidence_score") ), supports_claim=True, metadata=self._merge_metadata( evidence.get("evidence_payload"), { "source_page": evidence.get("source_page"), "rationale": evidence.get("rationale"), "confidence_score": evidence.get("confidence_score"), }, ), ) ) return items def _build_escalations( self, *, project_code: str, review_id: str, case_id: str, finding: dict[str, Any], session_actions: list[dict[str, Any]], ) -> list[OperationsEscalationItem]: """Map escalation-related review actions into operations escalation records.""" items: list[OperationsEscalationItem] = [] finding_code = str(finding.get("finding_code") or "") or None for index, action in enumerate(session_actions, start=1): action_payload = self._ensure_dict(action.get("action_payload")) target_finding_code = str(action_payload.get("finding_code") or "") or None reference_finding_id = str(action_payload.get("finding_id") or "") or None if str(action.get("action_type") or "") != "escalation_triggered": continue if target_finding_code and target_finding_code != finding_code: continue if reference_finding_id and reference_finding_id != str( finding.get("finding_id") or "" ): continue items.append( OperationsEscalationItem( escalation_id=f"{case_id}:escalation:{index}", case_id=case_id, action_id=str(action.get("action_id") or "") or None, project_code=project_code, review_session_id=review_id or None, target_department_code=action_payload.get("target_department_code"), target_employee_code=action_payload.get("target_employee_code"), target_agent_code=action_payload.get("target_agent_code") or action.get("actor_agent_code"), reason=action.get("action_summary") or "Escalation triggered", trigger_type="workflow_action", priority=str( action_payload.get("priority") or finding.get("severity") or "medium" ), status="queued", suggested_deadline_hours=action_payload.get( "suggested_deadline_hours" ), handoff_context=self._merge_metadata( action.get("before_state"), { "after_state": self._ensure_dict(action.get("after_state")), "action_payload": action_payload, }, ), created_at=action.get("created_at"), resolved_at=None, metadata={}, ) ) return items