File size: 9,574 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 | """
合同相关 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: ...
|