File size: 10,597 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 | """案件命令服务 - 负责所有案件写操作（创建、更新、删除）。"""
from __future__ import annotations
import logging
from typing import Any
from django.db import transaction
from django.utils.translation import gettext_lazy as _
from apps.cases.models import Case
from apps.core.config.business_config import business_config
from apps.core.exceptions import NotFoundError, ValidationException
from apps.core.interfaces import IContractService
from apps.core.security.access_context import AccessContext
from apps.core.security.permissions import PermissionMixin
from .case_access_policy import CaseAccessPolicy
from .case_queryset import get_case_queryset
logger = logging.getLogger("apps.cases")
class CaseCommandService(PermissionMixin):
"""案件命令服务，封装所有案件写操作业务逻辑。"""
def __init__(
self,
contract_service: IContractService | None = None,
access_policy: CaseAccessPolicy | None = None,
) -> None:
self._contract_service = contract_service
self._access_policy = access_policy or CaseAccessPolicy()
# ------------------------------------------------------------------
# 内部辅助
# ------------------------------------------------------------------
def _validate_stage(
self,
stage: str,
case_type: str | None,
representation_stages: list[str] | None = None,
) -> str:
"""验证案件阶段是否合法。"""
if case_type and not business_config.is_stage_valid_for_case_type(stage, case_type):
raise ValidationException(
_("该案件类型不支持此阶段"),
errors={"current_stage": str(_("阶段不适用于此案件类型"))},
)
if representation_stages and stage not in representation_stages:
raise ValidationException(
_("当前阶段必须属于代理阶段集合"),
errors={"current_stage": str(_("阶段不在代理范围内"))},
)
return stage
def _resolve_stage_from_contract(
self,
contract_id: int | None,
stage: str,
) -> str:
"""根据合同解析并验证阶段。"""
case_type: str | None = None
rep_stages: list[str] | None = None
if contract_id and self._contract_service:
contract = self._contract_service.get_contract(contract_id)
if contract:
case_type = contract.case_type
rep_stages = contract.representation_stages
return self._validate_stage(stage, case_type, rep_stages)
def _validate_contract(self, contract_id: int) -> None:
"""验证合同存在且处于激活状态。"""
if not self._contract_service:
return
contract = self._contract_service.get_contract(contract_id)
if not contract:
raise ValidationException(
message=_("合同不存在"),
code="CONTRACT_NOT_FOUND",
errors={"contract_id": f"无效的合同 ID: {contract_id}"},
)
if not self._contract_service.validate_contract_active(contract_id):
raise ValidationException(
message=_("合同未激活"),
code="CONTRACT_INACTIVE",
errors={"contract_id": str(_("合同状态不是 active"))},
)
# ------------------------------------------------------------------
# 公开命令方法
# ------------------------------------------------------------------
@transaction.atomic
def create_case(
self,
data: dict[str, Any],
*,
user: Any | None = None,
org_access: dict[str, Any] | None = None,
perm_open_access: bool = False,
) -> Case:
"""创建案件。
ForbiddenError: 用户未认证
ValidationException: 合同无效或阶段不合法
"""
ctx = AccessContext(user=user, org_access=org_access, perm_open_access=perm_open_access)
self.check_authenticated(ctx)
contract_id: int | None = data.get("contract_id")
if contract_id:
self._validate_contract(contract_id)
current_stage: str | None = data.get("current_stage")
if current_stage:
data["current_stage"] = self._resolve_stage_from_contract(contract_id, current_stage)
logger.info(
"创建案件",
extra={
"action": "create_case",
"case_name": data.get("name"),
"contract_id": contract_id,
"user_id": getattr(user, "id", None) if user else None,
},
)
return Case.objects.create(**data)
@transaction.atomic
def create_case_ctx(self, *, data: dict[str, Any], ctx: AccessContext) -> Case:
"""创建案件（AccessContext 版本）。"""
return self.create_case(
data,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
)
@transaction.atomic
def update_case(
self,
case_id: int,
data: dict[str, Any],
*,
user: Any | None = None,
org_access: dict[str, Any] | None = None,
perm_open_access: bool = False,
) -> Case:
"""更新案件。
NotFoundError: 案件不存在
ForbiddenError: 无权限
ValidationException: 合同无效或阶段不合法
"""
try:
case = get_case_queryset().select_related("contract").get(id=case_id)
except Case.DoesNotExist:
raise NotFoundError(_("案件 %(id)s 不存在") % {"id": case_id}) from None
ctx = AccessContext(user=user, org_access=org_access, perm_open_access=perm_open_access)
if not perm_open_access:
self.check_authenticated(ctx)
self._access_policy.ensure_access(
case_id=case.id,
user=user,
org_access=org_access,
perm_open_access=perm_open_access,
case=case,
message=_("无权限访问此案件"),
)
contract_id: int | None = data.get("contract_id")
if contract_id and self._contract_service:
contract = self._contract_service.get_contract(contract_id)
if not contract:
raise ValidationException(
message=_("合同不存在"),
code="CONTRACT_NOT_FOUND",
errors={"contract_id": f"无效的合同 ID: {contract_id}"},
)
current_stage: str | None = data.get("current_stage")
if current_stage:
check_contract_id = contract_id if contract_id else case.contract_id
data["current_stage"] = self._resolve_stage_from_contract(check_contract_id, current_stage)
for key, value in data.items():
setattr(case, key, value)
case.save()
logger.info(
"更新案件成功",
extra={
"action": "update_case",
"case_id": case_id,
"user_id": getattr(user, "id", None) if user else None,
},
)
return case
@transaction.atomic
def update_case_ctx(self, *, case_id: int, data: dict[str, Any], ctx: AccessContext) -> Case:
"""更新案件（AccessContext 版本）。"""
return self.update_case(
case_id,
data,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
)
@transaction.atomic
def delete_case(
self,
case_id: int,
*,
user: Any | None = None,
org_access: dict[str, Any] | None = None,
perm_open_access: bool = False,
) -> None:
"""删除案件。
NotFoundError: 案件不存在
ForbiddenError: 无权限
"""
try:
case = Case.objects.get(id=case_id)
except Case.DoesNotExist:
raise NotFoundError(_("案件 %(id)s 不存在") % {"id": case_id}) from None
ctx = AccessContext(user=user, org_access=org_access, perm_open_access=perm_open_access)
if not perm_open_access:
self.check_authenticated(ctx)
self._access_policy.ensure_access(
case_id=case.id,
user=user,
org_access=org_access,
perm_open_access=perm_open_access,
case=case,
message=_("无权限访问此案件"),
)
logger.info(
"删除案件",
extra={
"action": "delete_case",
"case_id": case_id,
"user_id": getattr(user, "id", None) if user else None,
},
)
from apps.cases.utils import fix_sqlite_orphan_contract_fk
fix_sqlite_orphan_contract_fk()
case.delete()
@transaction.atomic
def delete_case_ctx(self, *, case_id: int, ctx: AccessContext) -> None:
"""删除案件（AccessContext 版本）。"""
return self.delete_case(
case_id,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
)
def create_case_full(
self,
data: dict[str, Any],
actor_id: int | None = None,
user: Any | None = None,
) -> dict[str, Any]:
"""创建完整案件（包含当事人、指派、日志）。
ValidationException: 数据验证失败
ForbiddenError: 权限不足
"""
from .workflows.case_full_create_workflow import CaseFullCreateWorkflow
result = CaseFullCreateWorkflow(case_service=self).run(data=data, actor_id=actor_id, user=user)
if not isinstance(result, dict):
raise TypeError(f"CaseFullCreateWorkflow.run() 返回了非 dict 类型: {type(result)}")
return result
# ------------------------------------------------------------------
# Internal (cross-module) mutations
# ------------------------------------------------------------------
def unbind_cases_from_contract_internal(self, contract_id: int) -> int:
return int(Case.objects.filter(contract_id=contract_id).update(contract=None))
def count_cases_by_contract(self, contract_id: int) -> int:
return int(Case.objects.filter(contract_id=contract_id).count())
|