| """Business logic services.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from typing import Any |
|
|
| from django.utils.translation import gettext_lazy as _ |
|
|
| from apps.core.exceptions import ValidationException |
|
|
| from .fallback import ( |
| PLACEHOLDER_FALLBACK_VALUE, |
| ensure_required_placeholders, |
| get_service_placeholder_keys, |
| normalize_service_result, |
| ) |
| from .registry import PlaceholderRegistry |
| from .types import PlaceholderContextData |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class EnhancedContextBuilder: |
| """增强的上下文构建器""" |
|
|
| def __init__(self, registry: PlaceholderRegistry | None = None) -> None: |
| """ |
| 初始化上下文构建器 |
| |
| Args: |
| registry: 占位符注册表实例,如果为 None 则使用默认实例 |
| """ |
| self.registry = registry or PlaceholderRegistry() |
|
|
| def build_context( |
| self, context_data: PlaceholderContextData, required_placeholders: list[str] | None = None |
| ) -> dict[str, Any]: |
| """ |
| 构建完整的替换词上下文 |
| |
| Args: |
| context_data: 原始数据(contract, clients 等) |
| required_placeholders: 可选,仅生成指定的占位符 |
| |
| Returns: |
| 完整的占位符上下文字典 |
| """ |
| if not context_data: |
| logger.warning("上下文数据为空") |
| return {} |
|
|
| context_data = self._normalize_context_data(context_data) |
| context: dict[str, Any] = {} |
| services = self._get_relevant_services(required_placeholders) |
|
|
| for service in services: |
| service_keys = get_service_placeholder_keys(service) |
| try: |
| service_result = service.generate(context_data) |
| normalized_result = normalize_service_result(service_result, expected_keys=service_keys) |
| if normalized_result: |
| context.update(normalized_result) |
| logger.debug("服务 %s 生成了 %s 个占位符", service.name, len(normalized_result)) |
| except Exception as e: |
| logger.error( |
| "占位符服务执行失败: %s", |
| service.name, |
| extra={ |
| "service_name": service.name, |
| "error": str(e), |
| "context_keys": list(context_data.keys()), |
| }, |
| exc_info=True, |
| ) |
| context.update(dict.fromkeys(service_keys, PLACEHOLDER_FALLBACK_VALUE)) |
| |
| continue |
|
|
| if context_data.get("supplementary_agreement"): |
| key_map = { |
| "补充协议委托人信息": "委托人信息", |
| "补充协议委托人签名盖章信息": "委托人签名盖章信息", |
| "补充协议委托人主体信息条款": "委托人主体信息条款", |
| "补充协议委托人数量": "委托人数量", |
| "补充协议对方当事人主体信息条款": "对方当事人主体信息条款", |
| } |
| for old_key, new_key in key_map.items(): |
| if old_key in context: |
| context[new_key] = context[old_key] |
|
|
| final_context = ensure_required_placeholders(context, required_placeholders) |
|
|
| logger.info("上下文构建完成,生成了 %s 个占位符", len(final_context)) |
| return final_context |
|
|
| def _normalize_context_data(self, context_data: PlaceholderContextData) -> PlaceholderContextData: |
| """标准化上下文,对常见缺失键进行兜底补全.""" |
| normalized: dict[str, Any] = dict(context_data) |
|
|
| if normalized.get("case_id") is None: |
| case_obj = normalized.get("case") |
| case_id = getattr(case_obj, "id", None) |
| if case_id is not None: |
| normalized["case_id"] = case_id |
|
|
| return normalized |
|
|
| def build_contract_context(self, contract_id: int) -> dict[str, Any]: |
| """ |
| 为合同构建上下文(便捷方法) |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 完整的占位符上下文字典 |
| |
| Raises: |
| ValidationException: 合同不存在或数据无效 |
| """ |
| try: |
| |
| |
| from apps.documents.services.infrastructure.wiring import get_contract_service |
|
|
| contract_service = get_contract_service() |
| contract_dto = contract_service.get_contract_internal(contract_id) |
|
|
| if not contract_dto: |
| raise ValidationException( |
| message=_("合同不存在"), |
| code="CONTRACT_NOT_FOUND", |
| errors={"contract_id": f"ID 为 {contract_id} 的合同不存在"}, |
| ) |
|
|
| contract = contract_service.get_contract_model_internal(contract_id) |
| if not contract: |
| raise ValidationException( |
| message=_("合同不存在"), |
| code="CONTRACT_NOT_FOUND", |
| errors={"contract_id": f"ID 为 {contract_id} 的合同不存在"}, |
| ) |
|
|
| |
| context_data: PlaceholderContextData = {"contract": contract, "contract_id": contract_id} |
|
|
| return self.build_context(context_data) |
|
|
| except Exception as e: |
| if isinstance(e, ValidationException): |
| raise |
|
|
| logger.error("构建合同上下文失败: %s", e, extra={"contract_id": contract_id}, exc_info=True) |
| raise ValidationException( |
| message=_("构建合同上下文失败"), |
| code="CONTEXT_BUILD_ERROR", |
| errors={"contract_id": f"合同 {contract_id} 上下文构建失败: {e!s}"}, |
| ) from e |
|
|
| def _get_relevant_services(self, required_placeholders: list[str] | None = None) -> Any: |
| """ |
| 获取相关的占位符服务 |
| |
| Args: required_placeholders: 需要的占位符键列表 |
| |
| Returns: |
| 相关的服务实例列表 |
| """ |
| if not required_placeholders: |
| |
| return self.registry.get_all_services() |
|
|
| |
| relevant_services: list[Any] = [] |
| for placeholder_key in required_placeholders: |
| service = self.registry.get_service_for_placeholder(placeholder_key) |
| if service and service not in relevant_services: |
| relevant_services.append(service) |
|
|
| return relevant_services |
|
|
| def get_available_placeholders(self) -> dict[str, list[str]]: |
| """ |
| 获取所有可用的占位符按分类分组 |
| |
| Returns: |
| 按分类分组的占位符字典 |
| """ |
| result: dict[str, Any] = {} |
| services = self.registry.get_all_services() |
|
|
| for service in services: |
| category = service.category |
| if category not in result: |
| result[category] = [] |
| result[category].extend(service.get_placeholder_keys()) |
|
|
| return result |
|
|
| def validate_placeholders(self, placeholder_keys: list[str]) -> dict[str, bool]: |
| """ |
| 验证占位符键是否可用 |
| |
| Args: |
| placeholder_keys: 要验证的占位符键列表 |
| |
| Returns: |
| 占位符键的可用性字典 |
| """ |
| result: dict[str, Any] = {} |
|
|
| for key in placeholder_keys: |
| service = self.registry.get_service_for_placeholder(key) |
| result[key] = service is not None |
|
|
| return result |
|
|
