| """Business logic services.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from datetime import date |
| from decimal import Decimal |
| from typing import TYPE_CHECKING, Any, cast |
|
|
| from apps.documents.utils.formatters import format_currency, format_date, format_percentage, get_choice_display |
|
|
| |
| _ROLE_PRINCIPAL = "PRINCIPAL" |
| _ROLE_BENEFICIARY = "BENEFICIARY" |
| _ROLE_OPPOSING = "OPPOSING" |
|
|
| if TYPE_CHECKING: |
| from apps.core.interfaces import IContractService |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class ContextBuilder: |
| """ |
| 上下文构建器 |
| |
| 从数据库提取数据并构建替换词上下文字典. |
| 优先使用 EnhancedContextBuilder 进行上下文构建, |
| 同时保留直接构建的能力以支持向后兼容. |
| """ |
|
|
| DEFAULT_DATE_FORMAT = "%Y年%m月%d日" |
|
|
| def __init__( |
| self, |
| date_format: str | None = None, |
| use_enhanced: bool = False, |
| contract_service: IContractService | None = None, |
| ) -> None: |
| """ |
| 初始化上下文构建器 |
| |
| Args: |
| date_format: 日期格式字符串 |
| use_enhanced: 是否使用 EnhancedContextBuilder(默认 True) |
| contract_service: 合同服务(可选,用于依赖注入) |
| """ |
| self.date_format = date_format or self.DEFAULT_DATE_FORMAT |
| self._use_enhanced = use_enhanced |
| self._enhanced_builder = None |
| self._contract_service = contract_service |
|
|
| @property |
| def contract_service(self) -> IContractService: |
| """延迟加载合同服务""" |
| if self._contract_service is None: |
| from apps.documents.services.infrastructure.wiring import get_contract_service |
|
|
| self._contract_service = get_contract_service() |
| return self._contract_service |
|
|
| @property |
| def enhanced_builder(self) -> Any: |
| """延迟加载 EnhancedContextBuilder""" |
| if self._enhanced_builder is None: |
| from apps.documents.services.placeholders.context_builder import EnhancedContextBuilder |
|
|
| self._enhanced_builder = EnhancedContextBuilder() |
| return self._enhanced_builder |
|
|
| def build_contract_context(self, contract_id: int) -> dict[str, Any]: |
| """ |
| 构建合同相关的完整上下文 |
| |
| 优先使用 EnhancedContextBuilder 通过占位符服务构建上下文, |
| 如果失败则回退到直接构建模式. |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 包含所有替换词的字典 |
| """ |
| if self._use_enhanced: |
| try: |
| context = self.enhanced_builder.build_contract_context(contract_id) |
| if context: |
| logger.debug("使用 EnhancedContextBuilder 构建合同上下文成功,合同ID: %s", contract_id) |
| return cast(dict[str, Any], context) |
| except Exception as e: |
| logger.warning( |
| "EnhancedContextBuilder 构建失败,回退到直接构建模式: %s", |
| e, |
| extra={"contract_id": contract_id}, |
| ) |
|
|
| return self._build_contract_context_directly(contract_id) |
|
|
| def _build_contract_context_directly(self, contract_id: int) -> dict[str, Any]: |
| """ |
| 直接从数据库构建合同上下文(向后兼容) |
| |
| 通过 ServiceLocator 获取合同数据,避免跨模块直接导入 Model. |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 包含所有替换词的字典 |
| """ |
| |
| |
| contract_dto = self.contract_service.get_contract_with_details_internal(contract_id) |
| if contract_dto is None: |
| logger.warning("合同不存在: %s", contract_id) |
| return {} |
|
|
| |
| context: dict[str, Any] = { |
| "contract_name": contract_dto.get("name") or "", |
| "contract_type": contract_dto.get("case_type_display") or "", |
| "contract_type_code": contract_dto.get("case_type") or "", |
| "contract_status": contract_dto.get("status_display") or "", |
| "contract_date": self._format_date(contract_dto.get("specified_date")), |
| "contract_start_date": self._format_date(contract_dto.get("start_date")), |
| "contract_end_date": self._format_date(contract_dto.get("end_date")), |
| "fee_mode": contract_dto.get("fee_mode_display") or "", |
| "fee_mode_code": contract_dto.get("fee_mode") or "", |
| "fixed_amount": self._format_currency(contract_dto.get("fixed_amount")), |
| "fixed_amount_raw": contract_dto.get("fixed_amount") or Decimal("0"), |
| "risk_rate": self._format_percentage(contract_dto.get("risk_rate")), |
| "risk_rate_raw": contract_dto.get("risk_rate") or Decimal("0"), |
| "custom_terms": contract_dto.get("custom_terms") or "", |
| "representation_stages": ", ".join(contract_dto.get("representation_stages") or []), |
| } |
|
|
| |
| parties = contract_dto.get("contract_parties") or [] |
| principals = [p for p in parties if p.get("role") == _ROLE_PRINCIPAL] |
| beneficiaries = [p for p in parties if p.get("role") == _ROLE_BENEFICIARY] |
| opposing = [p for p in parties if p.get("role") == _ROLE_OPPOSING] |
|
|
| def _get_client_field(party: dict[str, Any], field: str) -> str: |
| """兼容嵌套 client 字典和扁平结构两种 DTO 格式""" |
| nested = party.get("client") |
| if isinstance(nested, dict): |
| return nested.get(field) or "" |
| |
| if field == "name": |
| return party.get("client_name") or "" |
| return party.get(field) or "" |
|
|
| if principals: |
| context.update( |
| { |
| "principal_name": _get_client_field(principals[0], "name"), |
| "principal_id_number": _get_client_field(principals[0], "id_number"), |
| "principal_phone": _get_client_field(principals[0], "phone"), |
| "principal_address": _get_client_field(principals[0], "address"), |
| "all_principals": [ |
| { |
| "name": _get_client_field(p, "name"), |
| "id_number": _get_client_field(p, "id_number"), |
| } |
| for p in principals |
| ], |
| } |
| ) |
| else: |
| context.update( |
| { |
| "principal_name": "", |
| "principal_id_number": "", |
| "principal_phone": "", |
| "principal_address": "", |
| "all_principals": [], |
| } |
| ) |
|
|
| if beneficiaries: |
| context.update( |
| { |
| "beneficiary_name": _get_client_field(beneficiaries[0], "name"), |
| "beneficiary_id_number": _get_client_field(beneficiaries[0], "id_number"), |
| } |
| ) |
| else: |
| context.update({"beneficiary_name": "", "beneficiary_id_number": ""}) |
|
|
| if opposing: |
| context.update( |
| { |
| "opposing_party_name": _get_client_field(opposing[0], "name"), |
| "all_opposing_parties": [_get_client_field(p, "name") for p in opposing], |
| } |
| ) |
| else: |
| context.update({"opposing_party_name": "", "all_opposing_parties": []}) |
|
|
| |
| assignments = contract_dto.get("assignments") or [] |
|
|
| def _get_lawyer_field(assignment: dict[str, Any], field: str) -> str: |
| """兼容嵌套 lawyer 字典和扁平结构两种 DTO 格式""" |
| nested = assignment.get("lawyer") |
| if isinstance(nested, dict): |
| return nested.get(field) or "" |
| |
| if field in ("real_name", "username"): |
| return assignment.get("lawyer_name") or "" |
| if field == "phone": |
| return assignment.get("lawyer_phone") or "" |
| if field == "license_no": |
| return assignment.get("lawyer_license_no") or "" |
| return assignment.get(field) or "" |
|
|
| primary_assignment = next((a for a in assignments if a.get("is_primary")), None) |
|
|
| if primary_assignment: |
| context.update( |
| { |
| "primary_lawyer_name": _get_lawyer_field(primary_assignment, "real_name"), |
| "primary_lawyer_phone": _get_lawyer_field(primary_assignment, "phone"), |
| "primary_lawyer_license": _get_lawyer_field(primary_assignment, "license_no"), |
| } |
| ) |
| elif assignments: |
| context.update( |
| { |
| "primary_lawyer_name": _get_lawyer_field(assignments[0], "real_name"), |
| "primary_lawyer_phone": _get_lawyer_field(assignments[0], "phone"), |
| "primary_lawyer_license": _get_lawyer_field(assignments[0], "license_no"), |
| } |
| ) |
| else: |
| context.update({"primary_lawyer_name": "", "primary_lawyer_phone": "", "primary_lawyer_license": ""}) |
|
|
| context["all_lawyers"] = [ |
| { |
| "name": _get_lawyer_field(a, "real_name"), |
| "is_primary": bool(a.get("is_primary")), |
| } |
| for a in assignments |
| ] |
|
|
| return context |
|
|
| def _format_date(self, value: date | None) -> str: |
| return format_date(value, self.date_format) |
|
|
| def _format_currency(self, value: Decimal | None) -> str: |
| return format_currency(value) |
|
|
| def _format_percentage(self, value: Decimal | None) -> str: |
| return format_percentage(value) |
|
|
