| """ |
| 合同相关 Protocol 接口定义 |
| |
| 包含:IContractService, IContractPaymentService |
| """ |
|
|
| from typing import Any, Protocol |
|
|
| from apps.core.dto import ContractDTO, LawyerDTO, PartyRoleDTO, SupplementaryAgreementDTO |
| from apps.core.security.access_context import AccessContext |
|
|
|
|
| class IContractService(Protocol): |
| """ |
| 合同服务接口 |
| |
| 定义合同服务的公共方法,供其他模块使用 |
| """ |
|
|
| def get_contract(self, contract_id: int) -> ContractDTO | None: |
| """ |
| 获取合同信息 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 合同 DTO,不存在时返回 None |
| """ |
| ... |
|
|
| def get_contract_stages(self, contract_id: int) -> list[str]: |
| """ |
| 获取合同的代理阶段 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 代理阶段列表 |
| |
| Raises: |
| NotFoundError: 合同不存在 |
| """ |
| ... |
|
|
| def validate_contract_active(self, contract_id: int) -> bool: |
| """ |
| 验证合同是否有效(状态为 active) |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 合同是否有效 |
| """ |
| ... |
|
|
| def get_contracts_by_ids(self, contract_ids: list[int]) -> list[ContractDTO]: |
| """ |
| 批量获取合同信息 |
| |
| Args: |
| contract_ids: 合同 ID 列表 |
| |
| Returns: |
| 合同 DTO 列表 |
| """ |
| ... |
|
|
| def get_contract_assigned_lawyer_id(self, contract_id: int) -> int | None: |
| """ |
| 获取合同的主办律师 ID(使用 primary_lawyer) |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 主办律师 ID,合同不存在或无主办律师时返回 None |
| """ |
| ... |
|
|
| def get_contract_lawyers(self, contract_id: int) -> list[LawyerDTO]: |
| """ |
| 获取合同的所有律师 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 律师 DTO 列表,按 is_primary 降序、order 升序排列 |
| |
| Raises: |
| NotFoundError: 合同不存在 |
| """ |
| ... |
|
|
| def get_all_parties(self, contract_id: int) -> list[dict[str, Any]]: |
| """ |
| 获取合同及其补充协议的所有当事人 |
| |
| 聚合 ContractParty 和 SupplementaryAgreementParty 中的所有 Client, |
| 按 client_id 去重,返回包含来源标识的当事人列表. |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 当事人列表,每个元素包含: |
| - id: Client ID |
| - name: Client 名称 |
| - source: 来源 ("contract" 或 "supplementary") |
| |
| Raises: |
| NotFoundError: 合同不存在 |
| """ |
| ... |
|
|
| def get_contract_with_details_internal(self, contract_id: int) -> dict[str, Any] | None: |
| """ |
| 内部方法:获取合同详细信息(包含当事人、律师等关联数据) |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 合同详细信息字典,包含: |
| - 基本信息 |
| - contract_parties: 当事人列表 |
| - assignments: 律师指派列表 |
| """ |
| ... |
|
|
| def ensure_contract_access_ctx(self, contract_id: int, ctx: AccessContext) -> None: ... |
|
|
| def ensure_contract_access_ctx_internal(self, contract_id: int, ctx: AccessContext) -> None: ... |
|
|
| def get_party_roles_by_contract_internal(self, contract_id: int) -> list[PartyRoleDTO]: |
| """ |
| 内部方法:获取合同的所有当事人角色 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| PartyRoleDTO 列表 |
| """ |
| ... |
|
|
| def get_fee_mode_display_internal(self, fee_mode: str) -> str: |
| """ |
| 内部方法:获取收费模式显示名称 |
| |
| Args: |
| fee_mode: 收费模式代码 |
| |
| Returns: |
| 显示名称 |
| """ |
| ... |
|
|
| def get_opposing_parties_internal(self, contract_id: int) -> list[PartyRoleDTO]: |
| """ |
| 内部方法:获取对方当事人列表 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 对方当事人 PartyRoleDTO 列表 |
| """ |
| ... |
|
|
| def get_principals_internal(self, contract_id: int) -> list[PartyRoleDTO]: |
| """ |
| 内部方法:获取委托人列表 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 委托人 PartyRoleDTO 列表 |
| """ |
| ... |
|
|
| def get_supplementary_agreements_internal(self, contract_id: int) -> list[SupplementaryAgreementDTO]: |
| """ |
| 内部方法:获取合同的补充协议列表 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| SupplementaryAgreementDTO 列表 |
| """ |
| ... |
|
|
| def get_contract_model_internal(self, contract_id: int) -> Any | None: |
| """ |
| 内部方法:获取合同 Model 对象(用于外键赋值或占位符服务) |
| |
| .. deprecated:: |
| 此方法直接返回原始 Model 实例，破坏适配器层 DTO 封装边界。 |
| 请使用 ``get_contract_with_details_internal`` 获取字典格式数据。 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| Contract Model 对象,不存在时返回 None |
| """ |
| ... |
|
|
| def get_supplementary_agreement_model_internal(self, contract_id: int, agreement_id: int) -> Any | None: |
| """ |
| 内部方法:获取补充协议 Model 对象(用于占位符服务) |
| |
| Args: |
| contract_id: 合同 ID |
| agreement_id: 补充协议 ID |
| |
| Returns: |
| SupplementaryAgreement Model 对象,不存在时返回 None |
| """ |
| ... |
|
|
|
|
| class IContractAssignmentQueryService(Protocol): |
| def list_lawyer_ids_by_contract_internal(self, contract_id: int) -> list[int]: ... |
|
|
|
|
| class IContractPaymentService(Protocol): |
| """ |
| 合同收款服务接口 |
| |
| 定义合同收款管理的核心方法 |
| """ |
|
|
| def list_payments( |
| self, |
| contract_id: int | None = None, |
| invoice_status: str | None = None, |
| start_date: Any | None = None, |
| end_date: Any | None = None, |
| user: Any | None = None, |
| perm_open_access: bool = False, |
| ) -> Any: |
| """ |
| 获取收款列表 |
| |
| Args: |
| contract_id: 合同 ID(可选) |
| invoice_status: 开票状态筛选(可选) |
| start_date: 开始日期筛选(可选) |
| end_date: 结束日期筛选(可选) |
| user: 当前用户 |
| perm_open_access: 是否开放访问权限 |
| |
| Returns: |
| 收款记录查询集 |
| """ |
| ... |
|
|
| def get_payment( |
| self, |
| payment_id: int, |
| user: Any | None = None, |
| perm_open_access: bool = False, |
| ) -> Any: |
| """ |
| 获取单个收款记录 |
| |
| Args: |
| payment_id: 收款 ID |
| user: 当前用户 |
| perm_open_access: 是否开放访问权限 |
| |
| Returns: |
| 收款对象 |
| |
| Raises: |
| NotFoundError: 收款不存在 |
| """ |
| ... |
|
|
| def create_payment( |
| self, |
| contract_id: int, |
| amount: Any, |
| received_at: Any | None = None, |
| invoice_status: str | None = None, |
| invoiced_amount: Any | None = None, |
| note: str | None = None, |
| user: Any | None = None, |
| confirm: bool = False, |
| ) -> Any: |
| """ |
| 创建收款记录 |
| |
| Args: |
| contract_id: 合同 ID |
| amount: 收款金额 |
| received_at: 收款日期 |
| invoice_status: 开票状态 |
| invoiced_amount: 已开票金额 |
| note: 备注 |
| user: 当前用户 |
| confirm: 是否二次确认 |
| |
| Returns: |
| 创建的收款对象 |
| |
| Raises: |
| PermissionDenied: 无管理员权限 |
| ValidationException: 数据验证失败 |
| NotFoundError: 合同不存在 |
| """ |
| ... |
|
|
| def update_payment( |
| self, |
| payment_id: int, |
| data: dict[str, Any], |
| user: Any | None = None, |
| confirm: bool = False, |
| ) -> Any: |
| """ |
| 更新收款记录 |
| |
| Args: |
| payment_id: 收款 ID |
| data: 更新数据 |
| user: 当前用户 |
| confirm: 是否二次确认 |
| |
| Returns: |
| 更新后的收款对象 |
| |
| Raises: |
| PermissionDenied: 无管理员权限 |
| ValidationException: 数据验证失败 |
| NotFoundError: 收款不存在 |
| """ |
| ... |
|
|
| def delete_payment( |
| self, |
| payment_id: int, |
| user: Any | None = None, |
| confirm: bool = False, |
| ) -> dict[str, bool]: |
| """ |
| 删除收款记录 |
| |
| Args: |
| payment_id: 收款 ID |
| user: 当前用户 |
| confirm: 是否二次确认 |
| |
| Returns: |
| {"success": True} |
| |
| Raises: |
| PermissionDenied: 无管理员权限 |
| ValidationException: 未二次确认 |
| NotFoundError: 收款不存在 |
| """ |
| ... |
|
|
|
|
| class IContractFolderBindingService(Protocol): |
| def save_file_to_bound_folder( |
| self, |
| contract_id: int, |
| file_content: bytes, |
| file_name: str, |
| subdir_key: str = "contract_documents", |
| ) -> str | None: ... |
|
|
| def extract_zip_to_bound_folder(self, contract_id: int, zip_content: bytes) -> str | None: ... |
|
|
