"""Service tính trạng thái field và readiness cho generation workflow pilot.""" from __future__ import annotations from collections.abc import Mapping from typing import Any from app.models.dossier_schemas import ( BienBanBanGiaoMatBangGenerationPayload, GenerationFieldStatus, GenerationReadinessStatus, ) from app.modules.agents.application.generation_template_registry import get_placeholder_definitions PILOT_DOCUMENT_TYPE = "bien_ban_ban_giao_mat_bang" PAYLOAD_TO_PLACEHOLDER_PATHS: dict[str, str] = { "document_title": "template.template_name", "document_number": "dossier_code", "project_name": "project_name", "site_name": "site_description", "handover_date": "handover_date_text", "handover_location": "handover_location", "transferor_org_name": "handing_over_organization.name", "transferor_org_address": "handing_over_organization.address", "transferor_signatory_full_name": "handing_over_signatory.full_name", "transferor_signatory_title": "handing_over_signatory.title", "recipient_org_name": "receiving_organization.name", "recipient_org_address": "receiving_organization.address", "recipient_signatory_full_name": "receiving_signatory.full_name", "recipient_signatory_title": "receiving_signatory.title", "handover_scope_summary": "site_description", "handover_notes": "extra_context.handover_notes", "legal_basis_summary": "extra_context.legal_basis_summary", } class GenerationReadinessService: """Tính trạng thái field và readiness cho pilot generation.""" def build_field_statuses_for_bien_ban_ban_giao_mat_bang( self, payload: BienBanBanGiaoMatBangGenerationPayload | Mapping[str, Any], ) -> list[GenerationFieldStatus]: """Tạo danh sách trạng thái field cho biên bản bàn giao mặt bằng.""" payload_data = self._normalize_payload(payload) field_statuses: list[GenerationFieldStatus] = [] for definition in get_placeholder_definitions(PILOT_DOCUMENT_TYPE): field_name = definition["name"] payload_path = PAYLOAD_TO_PLACEHOLDER_PATHS.get(field_name) value = ( self._get_nested_value(payload_data, payload_path) if payload_path else None ) confidence = self._infer_confidence(value, definition["minimum_confidence"]) is_resolved = self._is_field_resolved(value) notes: str | None = None if definition["required"] and not is_resolved: notes = "Thiếu dữ liệu bắt buộc, cần bổ sung trước khi render." elif confidence == "suggested_from_similar_template": notes = "Giá trị đang ở mức gợi ý, cần reviewer xác nhận." elif confidence == "manual_input_required": notes = "Chưa có dữ liệu nguồn, cần nhập tay." field_statuses.append( GenerationFieldStatus( field_name=field_name, label=definition["label"], value=self._normalize_field_value(value), confidence=confidence, is_required=definition["required"], is_resolved=is_resolved, source_paths=[payload_path] if payload_path else [], notes=notes, ) ) return field_statuses def build_readiness_status( self, field_statuses: list[GenerationFieldStatus], ) -> GenerationReadinessStatus: """Tính readiness từ danh sách trạng thái field.""" missing_required_fields = [ item.field_name for item in field_statuses if item.is_required and not item.is_resolved ] unresolved_review_fields = [ item.field_name for item in field_statuses if item.is_resolved and item.confidence != "verified_from_source" ] if missing_required_fields: return GenerationReadinessStatus( status="not_ready", missing_required_fields=missing_required_fields, unresolved_review_fields=unresolved_review_fields, summary="Còn thiếu trường bắt buộc nên chưa thể tạo bản nháp.", ) if unresolved_review_fields: if any( item.confidence == "manual_input_required" for item in field_statuses if item.field_name in unresolved_review_fields ): status = "draft_ready" summary = "Đủ trường bắt buộc để tạo nháp nhưng còn dữ liệu nhập tay cần rà soát." else: status = "review_required" summary = "Đủ dữ liệu để tạo hồ sơ nhưng còn trường gợi ý cần reviewer xác nhận." return GenerationReadinessStatus( status=status, missing_required_fields=[], unresolved_review_fields=unresolved_review_fields, summary=summary, ) return GenerationReadinessStatus( status="ready_for_render", missing_required_fields=[], unresolved_review_fields=[], summary="Tất cả trường bắt buộc đã được xác minh, có thể render.", ) def build_generation_payload_status( self, payload: BienBanBanGiaoMatBangGenerationPayload | Mapping[str, Any], ) -> tuple[list[GenerationFieldStatus], GenerationReadinessStatus]: """Tính đồng thời field statuses và readiness cho payload pilot.""" field_statuses = self.build_field_statuses_for_bien_ban_ban_giao_mat_bang( payload ) readiness = self.build_readiness_status(field_statuses) return field_statuses, readiness def _normalize_payload( self, payload: BienBanBanGiaoMatBangGenerationPayload | Mapping[str, Any], ) -> dict[str, Any]: """Chuẩn hóa payload về dict để truy cập thống nhất.""" if isinstance(payload, BienBanBanGiaoMatBangGenerationPayload): return payload.model_dump() return dict(payload) def _get_nested_value(self, payload_data: Mapping[str, Any], path: str) -> Any: """Lấy giá trị theo đường dẫn lồng nhau.""" current: Any = payload_data for part in path.split("."): if not isinstance(current, Mapping): return None current = current.get(part) if current is None: return None return current def _infer_confidence(self, value: Any, minimum_confidence: str) -> str: """Suy ra confidence từ giá trị hiện có và policy placeholder.""" if not self._has_meaningful_value(value): return "manual_input_required" if isinstance(value, str) and value.strip().startswith("[[suggested]]"): return "suggested_from_similar_template" return minimum_confidence def _is_field_resolved(self, value: Any) -> bool: """Xác định field đã có dữ liệu hữu ích hay chưa.""" return self._has_meaningful_value(value) def _has_meaningful_value(self, value: Any) -> bool: """Kiểm tra giá trị có thực sự hữu ích hay không.""" if value is None: return False if isinstance(value, str): return bool(value.strip()) if isinstance(value, list): return any(self._has_meaningful_value(item) for item in value) return True def _normalize_field_value( self, value: Any ) -> str | int | float | bool | list[str] | None: """Chuẩn hóa value về đúng kiểu hỗ trợ của schema.""" if value is None: return None if isinstance(value, list): return [str(item) for item in value if self._has_meaningful_value(item)] if isinstance(value, (str, int, float, bool)): return value return str(value) def build_bien_ban_ban_giao_mat_bang_field_statuses( payload: BienBanBanGiaoMatBangGenerationPayload | Mapping[str, Any], ) -> list[GenerationFieldStatus]: """Hàm tiện ích tính field statuses cho pilot.""" service = GenerationReadinessService() return service.build_field_statuses_for_bien_ban_ban_giao_mat_bang(payload) def build_generation_readiness_status( field_statuses: list[GenerationFieldStatus], ) -> GenerationReadinessStatus: """Hàm tiện ích tính readiness từ field statuses.""" service = GenerationReadinessService() return service.build_readiness_status(field_statuses) def build_bien_ban_ban_giao_mat_bang_generation_status( payload: BienBanBanGiaoMatBangGenerationPayload | Mapping[str, Any], ) -> tuple[list[GenerationFieldStatus], GenerationReadinessStatus]: """Hàm tiện ích tính trọn bộ trạng thái generation cho pilot.""" service = GenerationReadinessService() return service.build_generation_payload_status(payload)