"""Shared service guide profile models. These models are reused by both API responses and core extraction / ingest code so that typed contracts do not depend on the API layer. """ from __future__ import annotations from typing import Any from pydantic import BaseModel, ConfigDict, Field, ValidationError class ServiceGuideSchemaModel(BaseModel): """Base model for guide schemas. 首批强类型只收紧稳定字段，同时允许保留未来扩展字段，避免阻塞抽取器迭代。 """ model_config = ConfigDict(extra="allow") class ServiceGuideTimeLimit(ServiceGuideSchemaModel): raw_text: str = "" duration: str = "" unit: str = "" is_working_day: bool | None = None class ServiceGuideDocumentInfo(ServiceGuideSchemaModel): title: str = "" normalized_title: str = "" doc_type: str = "" knowledge_category: str = "" publish_date: str = "" effective_date: str = "" expiry_date: str = "" issuing_org: str = "" status: str = "" class ServiceGuideMatterIdentity(ServiceGuideSchemaModel): matter_name: str = "" colloquial_names: list[str] = Field(default_factory=list) matter_type: str = "" basic_code: str = "" implementation_code: str = "" business_item_code: str = "" matter_version: str = "" implementing_subject: str = "" subject_nature: str = "" delegated_department: str = "" class ServiceGuideBasicInfo(ServiceGuideSchemaModel): service_object: list[str] = Field(default_factory=list) promised_time_limit: ServiceGuideTimeLimit = Field(default_factory=ServiceGuideTimeLimit) legal_time_limit: ServiceGuideTimeLimit = Field(default_factory=ServiceGuideTimeLimit) visit_count_to_hall: int | None = None must_onsite: bool | None = None must_onsite_reason: str = "" case_type: str = "" notified_commitment_enabled: bool | None = None hall_required: bool | None = None express_supported: bool | None = None reservation_supported: bool | None = None reservation_url: str = "" service_modes: list[str] = Field(default_factory=list) online_depth: str = "" linked_agencies: list[str] = Field(default_factory=list) class ServiceGuideCrossRegion(ServiceGuideSchemaModel): service_scope_type: str = "" regions_summary: list[str] = Field(default_factory=list) regions_detail: list[str] = Field(default_factory=list) regions_truncated: bool = False service_modes: list[str] = Field(default_factory=list) notes: str = "" raw_text: str = "" class ServiceGuideProcessInfo(ServiceGuideSchemaModel): summary: str = "" step_titles: list[str] = Field(default_factory=list) raw_text: str = "" notes: str = "" needs_review: bool = False class ServiceGuideMaterial(ServiceGuideSchemaModel): guide_material_id: str = "" material_name: str = "" linked_material_id: str = "" requirement_level: str = "" original_count: int | None = None copy_count: int | None = None form_types: list[str] = Field(default_factory=list) paper_spec: str = "" electronic_license_linked: bool | None = None exempt_submission: bool | None = None reusable_previous_submission: bool | None = None material_type: str = "" source_channel: str = "" fill_instructions: str = "" notes: str = "" applicable_conditions: list[str] = Field(default_factory=list) blank_form_available: bool | None = None sample_available: bool | None = None download_hint: str = "" raw_row_text: str = "" class ServiceGuideFee(ServiceGuideSchemaModel): fee_name: str = "" amount_text: str = "" amount_value: float | None = None currency: str = "" charging_body: str = "" charging_method: str = "" reducible: bool | None = None notes: str = "" class ServiceGuideLegalBasis(ServiceGuideSchemaModel): law_name: str = "" document_no: str = "" article_no: str = "" issuing_body: str = "" effective_date: str | None = None article_content: str = "" class ServiceGuideConsultation(ServiceGuideSchemaModel): consultation_phones: list[str] = Field(default_factory=list) consultation_urls: list[str] = Field(default_factory=list) complaint_phones: list[str] = Field(default_factory=list) complaint_urls: list[str] = Field(default_factory=list) class ServiceGuideServiceWindow(ServiceGuideSchemaModel): window_name: str = "" location: str = "" office_phone: str = "" office_hours: str = "" navigation: str = "" scope: str = "" class ServiceGuideReviewInfo(ServiceGuideSchemaModel): power_level: str = "" power_source: str = "" service_forms: list[str] = Field(default_factory=list) business_system: str = "" class ServiceGuideAcceptanceInfo(ServiceGuideSchemaModel): service_targets: list[str] = Field(default_factory=list) natural_person_topics: list[str] = Field(default_factory=list) legal_person_topics: list[str] = Field(default_factory=list) local_feature_topics: list[str] = Field(default_factory=list) application_scope: str = "" acceptance_conditions: list[str] = Field(default_factory=list) class ServiceGuideRightsAndObligations(ServiceGuideSchemaModel): rights: list[str] = Field(default_factory=list) obligations: list[str] = Field(default_factory=list) class ServiceGuideOrganizationBinding(ServiceGuideSchemaModel): name: str = "" organization_id: str = "" role: str = "" class ServiceGuideBindings(ServiceGuideSchemaModel): matter_ids: list[str] = Field(default_factory=list) organization_bindings: list[ServiceGuideOrganizationBinding] = Field(default_factory=list) region_bindings: list[dict[str, Any]] = Field(default_factory=list) class ServiceGuideQuality(ServiceGuideSchemaModel): completeness_score: float = 0.0 confidence_score: float = 0.0 missing_fields: list[str] = Field(default_factory=list) needs_review: bool = False warnings: list[str] = Field(default_factory=list) class StandardServiceGuideProfile(ServiceGuideSchemaModel): schema_version: str = "" profile_id: str = "" doc_id: str = "" content_hash: str = "" acl_ids: list[str] = Field(default_factory=list) scene_type: str = "" source: str = "" source_url: str = "" is_current: bool = True guide_version: str = "" extractor_version: str = "" extracted_at: str = "" updated_at: str = "" matter_name: str = "" colloquial_names: list[str] = Field(default_factory=list) matter_type: str = "" implementation_code: str = "" basic_code: str = "" business_item_code: str = "" matter_version: str = "" implementing_subject: str = "" implementing_subject_nature: str = "" delegated_department: str = "" service_objects: list[str] = Field(default_factory=list) service_modes: list[str] = Field(default_factory=list) online_depth: str = "" hall_required: bool | None = None express_supported: bool | None = None reservation_supported: bool | None = None must_onsite: bool | None = None must_onsite_reason: str = "" visit_count_to_hall: int | None = None promised_time_limit_days: int | None = None legal_time_limit_days: int | None = None handled_org_names: list[str] = Field(default_factory=list) region_names: list[str] = Field(default_factory=list) linked_matter_ids: list[str] = Field(default_factory=list) material_names: list[str] = Field(default_factory=list) fee_names: list[str] = Field(default_factory=list) window_names: list[str] = Field(default_factory=list) legal_basis_names: list[str] = Field(default_factory=list) cross_region_scope: str = "" cross_region_summary: list[str] = Field(default_factory=list) guide_search_text: str = "" needs_review: bool = False completeness_score: float = 0.0 confidence_score: float = 0.0 document_info: ServiceGuideDocumentInfo = Field(default_factory=ServiceGuideDocumentInfo) matter_identity: ServiceGuideMatterIdentity = Field(default_factory=ServiceGuideMatterIdentity) basic_info: ServiceGuideBasicInfo = Field(default_factory=ServiceGuideBasicInfo) cross_region_service: list[ServiceGuideCrossRegion] = Field(default_factory=list) review_info: ServiceGuideReviewInfo = Field(default_factory=ServiceGuideReviewInfo) result_info: list[dict[str, Any]] = Field(default_factory=list) acceptance_info: ServiceGuideAcceptanceInfo = Field(default_factory=ServiceGuideAcceptanceInfo) process_info: ServiceGuideProcessInfo = Field(default_factory=ServiceGuideProcessInfo) materials: list[ServiceGuideMaterial] = Field(default_factory=list) fees: list[ServiceGuideFee] = Field(default_factory=list) legal_basis: list[ServiceGuideLegalBasis] = Field(default_factory=list) rights_and_obligations: ServiceGuideRightsAndObligations = Field( default_factory=ServiceGuideRightsAndObligations ) remedies: dict[str, Any] = Field(default_factory=dict) consultation_and_supervision: ServiceGuideConsultation = Field( default_factory=ServiceGuideConsultation ) service_windows: list[ServiceGuideServiceWindow] = Field(default_factory=list) bindings: ServiceGuideBindings = Field(default_factory=ServiceGuideBindings) quality: ServiceGuideQuality = Field(default_factory=ServiceGuideQuality) raw_sections: dict[str, str] = Field(default_factory=dict) def _coerce_string(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, (int, float)) and not isinstance(value, bool): return str(value) return "" def _coerce_string_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): return [value] if value else [] if isinstance(value, (list, tuple, set)): values: list[str] = [] for item in value: coerced = _coerce_string(item).strip() if coerced: values.append(coerced) return values coerced = _coerce_string(value).strip() return [coerced] if coerced else [] def _coerce_bool(value: Any) -> bool | None: return value if isinstance(value, bool) else None def _coerce_int(value: Any) -> int | None: if isinstance(value, bool) or value is None: return None if isinstance(value, int): return value if isinstance(value, float): return int(value) if isinstance(value, str) and value.strip().isdigit(): return int(value.strip()) return None def _coerce_float(value: Any) -> float | None: if isinstance(value, bool) or value is None: return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): try: return float(value.strip()) except ValueError: return None return None def _coerce_dict_list(value: Any) -> list[dict[str, Any]]: if isinstance(value, dict): return [value] if isinstance(value, list): return [item for item in value if isinstance(item, dict)] return [] def _sanitize_time_limit(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "raw_text": _coerce_string(source.get("raw_text")), "duration": _coerce_string(source.get("duration")), "unit": _coerce_string(source.get("unit")), "is_working_day": _coerce_bool(source.get("is_working_day")), } def _sanitize_document_info(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "title": _coerce_string(source.get("title")), "normalized_title": _coerce_string(source.get("normalized_title")), "doc_type": _coerce_string(source.get("doc_type")), "knowledge_category": _coerce_string(source.get("knowledge_category")), "publish_date": _coerce_string(source.get("publish_date")), "effective_date": _coerce_string(source.get("effective_date")), "expiry_date": _coerce_string(source.get("expiry_date")), "issuing_org": _coerce_string(source.get("issuing_org")), "status": _coerce_string(source.get("status")), } def _sanitize_matter_identity(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "matter_name": _coerce_string(source.get("matter_name")), "colloquial_names": _coerce_string_list(source.get("colloquial_names")), "matter_type": _coerce_string(source.get("matter_type")), "basic_code": _coerce_string(source.get("basic_code")), "implementation_code": _coerce_string(source.get("implementation_code")), "business_item_code": _coerce_string(source.get("business_item_code")), "matter_version": _coerce_string(source.get("matter_version")), "implementing_subject": _coerce_string(source.get("implementing_subject")), "subject_nature": _coerce_string(source.get("subject_nature")), "delegated_department": _coerce_string(source.get("delegated_department")), } def _sanitize_basic_info(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "service_object": _coerce_string_list(source.get("service_object")), "promised_time_limit": _sanitize_time_limit(source.get("promised_time_limit")), "legal_time_limit": _sanitize_time_limit(source.get("legal_time_limit")), "visit_count_to_hall": _coerce_int(source.get("visit_count_to_hall")), "must_onsite": _coerce_bool(source.get("must_onsite")), "must_onsite_reason": _coerce_string(source.get("must_onsite_reason")), "case_type": _coerce_string(source.get("case_type")), "notified_commitment_enabled": _coerce_bool(source.get("notified_commitment_enabled")), "hall_required": _coerce_bool(source.get("hall_required")), "express_supported": _coerce_bool(source.get("express_supported")), "reservation_supported": _coerce_bool(source.get("reservation_supported")), "reservation_url": _coerce_string(source.get("reservation_url")), "service_modes": _coerce_string_list(source.get("service_modes")), "online_depth": _coerce_string(source.get("online_depth")), "linked_agencies": _coerce_string_list(source.get("linked_agencies")), } def _sanitize_process_info(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "summary": _coerce_string(source.get("summary")), "step_titles": _coerce_string_list(source.get("step_titles")), "raw_text": _coerce_string(source.get("raw_text")), "notes": _coerce_string(source.get("notes")), "needs_review": source.get("needs_review") if isinstance(source.get("needs_review"), bool) else False, } def _sanitize_review_info(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "power_level": _coerce_string(source.get("power_level")), "power_source": _coerce_string(source.get("power_source")), "service_forms": _coerce_string_list(source.get("service_forms")), "business_system": _coerce_string(source.get("business_system")), } def _sanitize_acceptance_info(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "service_targets": _coerce_string_list(source.get("service_targets")), "natural_person_topics": _coerce_string_list(source.get("natural_person_topics")), "legal_person_topics": _coerce_string_list(source.get("legal_person_topics")), "local_feature_topics": _coerce_string_list(source.get("local_feature_topics")), "application_scope": _coerce_string(source.get("application_scope")), "acceptance_conditions": _coerce_string_list(source.get("acceptance_conditions")), } def _sanitize_consultation(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "consultation_phones": _coerce_string_list(source.get("consultation_phones")), "consultation_urls": _coerce_string_list(source.get("consultation_urls")), "complaint_phones": _coerce_string_list(source.get("complaint_phones")), "complaint_urls": _coerce_string_list(source.get("complaint_urls")), } def _sanitize_rights_and_obligations(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "rights": _coerce_string_list(source.get("rights")), "obligations": _coerce_string_list(source.get("obligations")), } def _sanitize_bindings(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "matter_ids": _coerce_string_list(source.get("matter_ids")), "organization_bindings": _coerce_dict_list(source.get("organization_bindings")), "region_bindings": _coerce_dict_list(source.get("region_bindings")), } def _sanitize_quality(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "completeness_score": _coerce_float(source.get("completeness_score")) or 0.0, "confidence_score": _coerce_float(source.get("confidence_score")) or 0.0, "missing_fields": _coerce_string_list(source.get("missing_fields")), "needs_review": source.get("needs_review") if isinstance(source.get("needs_review"), bool) else False, "warnings": _coerce_string_list(source.get("warnings")), } def sanitize_service_guide_profile_payload(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} return { "schema_version": _coerce_string(source.get("schema_version")), "profile_id": _coerce_string(source.get("profile_id")), "doc_id": _coerce_string(source.get("doc_id")), "content_hash": _coerce_string(source.get("content_hash")), "acl_ids": _coerce_string_list(source.get("acl_ids")), "scene_type": _coerce_string(source.get("scene_type")), "source": _coerce_string(source.get("source")), "source_url": _coerce_string(source.get("source_url")), "is_current": source.get("is_current") if isinstance(source.get("is_current"), bool) else True, "guide_version": _coerce_string(source.get("guide_version")), "extractor_version": _coerce_string(source.get("extractor_version")), "extracted_at": _coerce_string(source.get("extracted_at")), "updated_at": _coerce_string(source.get("updated_at")), "matter_name": _coerce_string(source.get("matter_name")), "colloquial_names": _coerce_string_list(source.get("colloquial_names")), "matter_type": _coerce_string(source.get("matter_type")), "implementation_code": _coerce_string(source.get("implementation_code")), "basic_code": _coerce_string(source.get("basic_code")), "business_item_code": _coerce_string(source.get("business_item_code")), "matter_version": _coerce_string(source.get("matter_version")), "implementing_subject": _coerce_string(source.get("implementing_subject")), "implementing_subject_nature": _coerce_string(source.get("implementing_subject_nature")), "delegated_department": _coerce_string(source.get("delegated_department")), "service_objects": _coerce_string_list(source.get("service_objects")), "service_modes": _coerce_string_list(source.get("service_modes")), "online_depth": _coerce_string(source.get("online_depth")), "hall_required": _coerce_bool(source.get("hall_required")), "express_supported": _coerce_bool(source.get("express_supported")), "reservation_supported": _coerce_bool(source.get("reservation_supported")), "must_onsite": _coerce_bool(source.get("must_onsite")), "must_onsite_reason": _coerce_string(source.get("must_onsite_reason")), "visit_count_to_hall": _coerce_int(source.get("visit_count_to_hall")), "promised_time_limit_days": _coerce_int(source.get("promised_time_limit_days")), "legal_time_limit_days": _coerce_int(source.get("legal_time_limit_days")), "handled_org_names": _coerce_string_list(source.get("handled_org_names")), "region_names": _coerce_string_list(source.get("region_names")), "linked_matter_ids": _coerce_string_list(source.get("linked_matter_ids")), "material_names": _coerce_string_list(source.get("material_names")), "fee_names": _coerce_string_list(source.get("fee_names")), "window_names": _coerce_string_list(source.get("window_names")), "legal_basis_names": _coerce_string_list(source.get("legal_basis_names")), "cross_region_scope": _coerce_string(source.get("cross_region_scope")), "cross_region_summary": _coerce_string_list(source.get("cross_region_summary")), "guide_search_text": _coerce_string(source.get("guide_search_text")), "needs_review": source.get("needs_review") if isinstance(source.get("needs_review"), bool) else False, "completeness_score": _coerce_float(source.get("completeness_score")) or 0.0, "confidence_score": _coerce_float(source.get("confidence_score")) or 0.0, "document_info": _sanitize_document_info(source.get("document_info")), "matter_identity": _sanitize_matter_identity(source.get("matter_identity")), "basic_info": _sanitize_basic_info(source.get("basic_info")), "cross_region_service": _coerce_dict_list(source.get("cross_region_service")), "review_info": _sanitize_review_info(source.get("review_info")), "result_info": _coerce_dict_list(source.get("result_info")), "acceptance_info": _sanitize_acceptance_info(source.get("acceptance_info")), "process_info": _sanitize_process_info(source.get("process_info")), "materials": _coerce_dict_list(source.get("materials")), "fees": _coerce_dict_list(source.get("fees")), "legal_basis": _coerce_dict_list(source.get("legal_basis")), "rights_and_obligations": _sanitize_rights_and_obligations(source.get("rights_and_obligations")), "remedies": source.get("remedies") if isinstance(source.get("remedies"), dict) else {}, "consultation_and_supervision": _sanitize_consultation(source.get("consultation_and_supervision")), "service_windows": _coerce_dict_list(source.get("service_windows")), "bindings": _sanitize_bindings(source.get("bindings")), "quality": _sanitize_quality(source.get("quality")), "raw_sections": source.get("raw_sections") if isinstance(source.get("raw_sections"), dict) else {}, } def build_compatible_service_guide_profile(value: Any) -> tuple[StandardServiceGuideProfile, bool]: if isinstance(value, StandardServiceGuideProfile): return value, False if not isinstance(value, dict): return StandardServiceGuideProfile(), True try: return StandardServiceGuideProfile.model_validate(value), False except ValidationError: sanitized = sanitize_service_guide_profile_payload(value) return StandardServiceGuideProfile.model_validate(sanitized), True