| """ |
| 案件相关 Protocol 接口定义 |
| |
| 包含:ICaseService, ICaseSearchService, ICaseNumberService, ICaseLogService |
| """ |
|
|
| from typing import Any, Protocol |
|
|
| from apps.core.dto import CaseDTO, CasePartyDTO, CaseSearchResultDTO, CaseTemplateBindingDTO |
|
|
|
|
| class ICaseService(Protocol): |
| """ |
| 案件服务接口 |
| |
| 定义案件服务的公共方法,供其他模块使用 |
| """ |
|
|
| def get_case(self, case_id: int) -> CaseDTO | None: |
| """ |
| 获取案件信息 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案件 DTO,不存在时返回 None |
| """ |
| ... |
|
|
| def get_cases_by_contract(self, contract_id: int) -> list[CaseDTO]: |
| """ |
| 获取合同关联的案件 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 案件 DTO 列表 |
| """ |
| ... |
|
|
| def check_case_access(self, case_id: int, user_id: int) -> bool: |
| """ |
| 检查用户是否有权限访问案件 |
| |
| Args: |
| case_id: 案件 ID |
| user_id: 用户 ID |
| |
| Returns: |
| 是否有权限访问 |
| """ |
| ... |
|
|
| def get_cases_by_ids(self, case_ids: list[int]) -> list[CaseDTO]: |
| """ |
| 批量获取案件信息 |
| |
| Args: |
| case_ids: 案件 ID 列表 |
| |
| Returns: |
| 案件 DTO 列表 |
| """ |
| ... |
|
|
| def validate_case_active(self, case_id: int) -> bool: |
| """ |
| 验证案件是否有效(状态为 active) |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案件是否有效 |
| """ |
| ... |
|
|
| def get_case_current_stage(self, case_id: int) -> str | None: |
| """ |
| 获取案件的当前阶段 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 当前阶段,案件不存在时返回 None |
| """ |
| ... |
|
|
| def create_case(self, data: dict[str, Any]) -> CaseDTO: |
| """ |
| 创建案件 |
| |
| Args: |
| data: 案件数据字典,包含: |
| - name: 案件名称(必填) |
| - contract_id: 合同 ID(可选) |
| - is_archived: 是否已建档(可选,默认 False) |
| - case_type: 案件类型(可选) |
| - target_amount: 涉案金额(可选) |
| - cause_of_action: 案由(可选) |
| - current_stage: 当前阶段(可选) |
| |
| Returns: |
| 创建的案件 DTO |
| """ |
| ... |
|
|
| def create_case_assignment(self, case_id: int, lawyer_id: int) -> bool: |
| """ |
| 创建案件指派 |
| |
| Args: |
| case_id: 案件 ID |
| lawyer_id: 律师 ID |
| |
| Returns: |
| 是否创建成功 |
| """ |
| ... |
|
|
| def create_case_party(self, case_id: int, client_id: int, legal_status: str | None = None) -> bool: |
| """ |
| 创建案件当事人 |
| |
| Args: |
| case_id: 案件 ID |
| client_id: 客户 ID |
| legal_status: 诉讼地位(可选) |
| |
| Returns: |
| 是否创建成功 |
| """ |
| ... |
|
|
| def get_user_extra_case_access(self, user_id: int) -> list[int]: |
| """ |
| 获取用户的额外案件访问授权 |
| |
| Args: |
| user_id: 用户 ID |
| |
| Returns: |
| 用户有额外访问权限的案件 ID 列表 |
| """ |
| ... |
|
|
| def get_case_by_id_internal(self, case_id: int) -> CaseDTO | None: |
| """ |
| 内部方法:获取案件信息(无权限检查) |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案件 DTO,不存在时返回 None |
| """ |
| ... |
|
|
| def search_cases_by_party_internal(self, party_names: list[str], status: str | None = None) -> list[CaseDTO]: |
| """ |
| 内部方法:根据当事人名称搜索案件 |
| |
| Args: |
| party_names: 当事人名称列表 |
| status: 案件状态筛选(可选) |
| |
| Returns: |
| 匹配的案件 DTO 列表 |
| """ |
| ... |
|
|
| def get_case_numbers_by_case_internal(self, case_id: int) -> list[str]: |
| """ |
| 内部方法:获取案件的所有案号 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案号字符串列表 |
| """ |
| ... |
|
|
| def get_case_party_names_internal(self, case_id: int) -> list[str]: |
| """ |
| 内部方法:获取案件的所有当事人名称 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 当事人名称列表 |
| """ |
| ... |
|
|
| def search_cases_by_case_number_internal(self, case_number: str) -> list[CaseDTO]: |
| """ |
| 内部方法:根据案号搜索案件 |
| |
| Args: |
| case_number: 案号字符串 |
| |
| Returns: |
| 匹配的案件 DTO 列表 |
| """ |
| ... |
|
|
| def create_case_log_internal(self, case_id: int, content: str, user_id: int | None = None) -> int: |
| """ |
| 内部方法:创建案件日志,返回日志ID |
| |
| Args: |
| case_id: 案件 ID |
| content: 日志内容 |
| user_id: 用户 ID(可选) |
| |
| Returns: |
| 创建的日志 ID |
| |
| Raises: |
| NotFoundError: 案件不存在 |
| """ |
| ... |
|
|
| def add_case_log_attachment_internal(self, case_log_id: int, file_path: str, file_name: str) -> bool: |
| """ |
| 内部方法:添加案件日志附件 |
| |
| Args: |
| case_log_id: 案件日志 ID |
| file_path: 文件路径 |
| file_name: 文件名称 |
| |
| Returns: |
| 是否添加成功 |
| |
| Raises: |
| NotFoundError: 案件日志不存在 |
| """ |
| ... |
|
|
| def add_case_number_internal(self, case_id: int, case_number: str, user_id: int | None = None) -> bool: |
| """ |
| 内部方法:为案件添加案号(如果不存在) |
| |
| Args: |
| case_id: 案件 ID |
| case_number: 案号字符串 |
| user_id: 操作用户 ID(可选) |
| |
| Returns: |
| 是否添加成功(已存在也返回 True) |
| """ |
| ... |
|
|
| def update_case_log_reminder_internal(self, case_log_id: int, reminder_time: Any, reminder_type: str) -> bool: |
| """ |
| 内部方法:更新案件日志的提醒时间和类型 |
| |
| Args: |
| case_log_id: 案件日志 ID |
| reminder_time: 提醒时间(datetime) |
| reminder_type: 提醒类型(CaseLogReminderType 枚举值) |
| |
| Returns: |
| 是否更新成功 |
| """ |
| ... |
|
|
| def get_case_model_internal(self, case_id: int) -> Any | None: |
| """ |
| 内部方法:获取案件 Model 对象(用于外键赋值) |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| Case Model 对象,不存在时返回 None |
| """ |
| ... |
|
|
| def get_case_log_model_internal(self, case_log_id: int) -> Any | None: |
| """ |
| 内部方法:获取案件日志 Model 对象(用于外键赋值) |
| |
| Args: |
| case_log_id: 案件日志 ID |
| |
| Returns: |
| CaseLog Model 对象,不存在时返回 None |
| """ |
| ... |
|
|
| def get_case_with_details_internal(self, case_id: int) -> dict[str, Any] | None: |
| """ |
| 内部方法:获取案件详细信息(包含当事人、案号等关联数据) |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案件详细信息字典 |
| """ |
| ... |
|
|
| def unbind_cases_from_contract_internal(self, contract_id: int) -> int: |
| """ |
| 内部方法:解绑合同下的所有案件 |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 解绑的案件数量 |
| """ |
| ... |
|
|
| def count_cases_by_contract(self, contract_id: int) -> int: |
| """ |
| 内部方法:统计合同下的案件数（删除合同时用于日志记录） |
| |
| Args: |
| contract_id: 合同 ID |
| |
| Returns: |
| 案件数量 |
| """ |
| ... |
|
|
| def get_primary_lawyer_names_by_case_ids_internal(self, case_ids: list[int]) -> dict[int, str | None]: |
| """ |
| 内部方法:批量获取案件主办律师姓名(取案件指派列表的第一个律师) |
| """ |
| ... |
|
|
| def get_case_parties_by_legal_status_internal(self, case_id: int, legal_status: str) -> list[str]: |
| """ |
| 内部方法:根据诉讼地位获取当事人名称列表 |
| |
| Args: |
| case_id: 案件 ID |
| legal_status: 诉讼地位 (plaintiff/defendant 等) |
| |
| Returns: |
| 当事人名称列表 |
| """ |
| ... |
|
|
| def get_case_template_binding_internal(self, case_id: int) -> CaseTemplateBindingDTO | None: |
| """ |
| 内部方法:获取案件的模板绑定信息 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| CaseTemplateBindingDTO,不存在返回 None |
| """ |
| ... |
|
|
| def get_case_parties_internal(self, case_id: int, legal_status: str | None = None) -> list[CasePartyDTO]: |
| """ |
| 内部方法:获取案件当事人列表 |
| |
| Args: |
| case_id: 案件 ID |
| legal_status: 法律地位筛选(可选) |
| |
| Returns: |
| CasePartyDTO 列表 |
| """ |
| ... |
|
|
| def get_case_template_bindings_by_name_internal( |
| self, case_id: int, template_name: str |
| ) -> list[CaseTemplateBindingDTO]: |
| """ |
| 内部方法:根据模板名称获取案件模板绑定列表 |
| |
| Args: |
| case_id: 案件 ID |
| template_name: 模板名称 |
| |
| Returns: |
| CaseTemplateBindingDTO 列表 |
| """ |
| ... |
|
|
| def get_case_internal(self, case_id: int) -> CaseDTO | None: |
| """ |
| 内部方法:获取案件信息(无权限检查) |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案件 DTO,不存在时返回 None |
| """ |
| ... |
|
|
| def search_cases_for_binding_internal(self, search_term: str = "", limit: int = 20) -> list[dict[str, Any]]: |
| """ |
| 内部方法:搜索可绑定的案件(用于文书识别等跨模块场景) |
| |
| 支持按案件名称、案号、当事人搜索,返回包含案号和当事人的完整信息. |
| |
| Args: |
| search_term: 搜索关键词 |
| limit: 返回数量限制 |
| |
| Returns: |
| 案件信息字典列表,每个元素包含: |
| - id: 案件 ID |
| - name: 案件名称 |
| - case_numbers: 案号列表 |
| - parties: 当事人名称列表 |
| - created_at: 创建时间 ISO 格式字符串 |
| """ |
| ... |
|
|
|
|
| class ICaseSearchService(Protocol): |
| """ |
| 案件搜索服务接口 |
| |
| 定义案件搜索的核心方法,供跨模块调用. |
| 用于 API 层搜索案件,避免直接导入 Model 和使用 Q() 查询. |
| |
| Requirements: 1.1, 2.1 |
| """ |
|
|
| def search_cases(self, query: str, limit: int = 20) -> list[CaseSearchResultDTO]: |
| """ |
| 搜索案件 |
| |
| 支持按案件名称、案号、当事人搜索. |
| |
| Args: |
| query: 搜索关键词(案件名称、案号、当事人) |
| limit: 返回结果数量限制,默认20 |
| |
| Returns: |
| 匹配的案件搜索结果 DTO 列表 |
| """ |
| ... |
|
|
|
|
| class ICaseNumberService(Protocol): |
| """ |
| 案号服务接口 |
| |
| 定义案号管理的核心方法,供跨模块调用 |
| """ |
|
|
| def list_numbers_internal(self, case_id: int) -> list[Any]: |
| """ |
| 内部方法:获取案件的所有案号 |
| |
| Args: |
| case_id: 案件 ID |
| |
| Returns: |
| 案号对象列表 |
| """ |
| ... |
|
|
| def create_number_internal(self, case_id: int, number: str, remarks: str | None = None) -> Any: |
| """ |
| 内部方法:创建案号 |
| |
| Args: |
| case_id: 案件 ID |
| number: 案号 |
| remarks: 备注(可选) |
| |
| Returns: |
| 创建的案号对象 |
| """ |
| ... |
|
|
| def format_case_number(self, number: str) -> str: |
| """ |
| 格式化案号:统一括号、删除空格 |
| |
| 在保存前调用此方法，确保案号格式统一。 |
| |
| Args: |
| number: 原始案号 |
| |
| Returns: |
| 格式化后的案号 |
| """ |
| ... |
|
|
| def normalize_case_number(self, number: str) -> str: |
| """ |
| 规范化案号:统一括号、删除空格 |
| |
| .. deprecated:: |
| 使用 :meth:`format_case_number` 代替 |
| |
| Args: |
| number: 原始案号 |
| |
| Returns: |
| 规范化后的案号 |
| """ |
| ... |
|
|
|
|
| class ICaseFilingNumberService(Protocol): |
| def generate_case_filing_number_internal(self, case_id: int, case_type: str, created_year: int) -> str: ... |
|
|
|
|
| class ICaseLogService(Protocol): |
| """ |
| 案件日志服务接口 |
| |
| 定义案件日志管理的核心方法 |
| """ |
|
|
| def list_logs( |
| self, |
| case_id: int | None = None, |
| user: Any | None = None, |
| org_access: dict[str, Any] | None = None, |
| perm_open_access: bool = False, |
| ) -> Any: |
| """ |
| 获取日志列表 |
| |
| Args: |
| case_id: 案件 ID(可选,用于过滤) |
| user: 当前用户 |
| org_access: 组织访问策略 |
| perm_open_access: 是否有开放访问权限 |
| |
| Returns: |
| 日志查询集 |
| """ |
| ... |
|
|
| def get_log( |
| self, |
| log_id: int, |
| user: Any | None = None, |
| org_access: dict[str, Any] | None = None, |
| perm_open_access: bool = False, |
| ) -> Any: |
| """ |
| 获取单个日志 |
| |
| Args: |
| log_id: 日志 ID |
| user: 当前用户 |
| org_access: 组织访问策略 |
| perm_open_access: 是否有开放访问权限 |
| |
| Returns: |
| 日志对象 |
| |
| Raises: |
| NotFoundError: 日志不存在 |
| PermissionDenied: 无权限访问 |
| """ |
| ... |
|
|
| def create_log( |
| self, |
| case_id: int, |
| content: str, |
| user: Any | None = None, |
| reminder_type: str | None = None, |
| reminder_time: Any | None = None, |
| ) -> Any: |
| """ |
| 创建案件日志 |
| |
| Args: |
| case_id: 案件 ID |
| content: 日志内容 |
| user: 当前用户 |
| reminder_type: 提醒类型 |
| reminder_time: 提醒时间 |
| |
| Returns: |
| 创建的日志对象 |
| |
| Raises: |
| NotFoundError: 案件不存在 |
| """ |
| ... |
|
|
| def update_log( |
| self, |
| log_id: int, |
| data: dict[str, Any], |
| user: Any | None = None, |
| org_access: dict[str, Any] | None = None, |
| perm_open_access: bool = False, |
| ) -> Any: |
| """ |
| 更新案件日志 |
| |
| Args: |
| log_id: 日志 ID |
| data: 更新数据字典 |
| user: 当前用户 |
| org_access: 组织访问策略 |
| perm_open_access: 是否有开放访问权限 |
| |
| Returns: |
| 更新后的日志对象 |
| |
| Raises: |
| NotFoundError: 日志不存在 |
| PermissionDenied: 无权限修改 |
| """ |
| ... |
|
|
| def delete_log( |
| self, |
| log_id: int, |
| user: Any | None = None, |
| org_access: dict[str, Any] | None = None, |
| perm_open_access: bool = False, |
| ) -> dict[str, bool]: |
| """ |
| 删除案件日志 |
| |
| Args: |
| log_id: 日志 ID |
| user: 当前用户 |
| org_access: 组织访问策略 |
| perm_open_access: 是否有开放访问权限 |
| |
| Returns: |
| {"success": True} |
| |
| Raises: |
| NotFoundError: 日志不存在 |
| PermissionDenied: 无权限删除 |
| """ |
| ... |
|
|
| def upload_attachments( |
| self, |
| log_id: int, |
| files: list[Any], |
| user: Any | None = None, |
| org_access: dict[str, Any] | None = None, |
| perm_open_access: bool = False, |
| ) -> dict[str, int]: |
| """ |
| 上传日志附件 |
| |
| Args: |
| log_id: 日志 ID |
| files: 上传的文件列表 |
| user: 当前用户 |
| org_access: 组织访问策略 |
| perm_open_access: 是否有开放访问权限 |
| |
| Returns: |
| {"count": 上传数量} |
| |
| Raises: |
| NotFoundError: 日志不存在 |
| PermissionDenied: 无权限上传 |
| ValidationException: 文件验证失败 |
| """ |
| ... |
|
|
|
|
| class ILitigationFeeCalculatorService(Protocol): |
| def calculate_all_fees( |
| self, |
| target_amount: Any | None = None, |
| preservation_amount: Any | None = None, |
| case_type: str | None = None, |
| cause_of_action: str | None = None, |
| cause_of_action_id: int | None = None, |
| ) -> dict[str, Any]: ... |
|
|
