File size: 6,338 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 | """Module for log."""
from __future__ import annotations
import logging
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar
from django.core.files.uploadedfile import UploadedFile
from django.db import models
from django.utils.translation import gettext_lazy as _
from apps.cases.utils import CASE_LOG_ALLOWED_EXTENSIONS, CASE_LOG_MAX_FILE_SIZE
from apps.core.filesystem.storage import KeepOriginalNameStorage
_SENTINEL = object()
from .case import Case
if TYPE_CHECKING:
from django.db.models.fields.related_descriptors import RelatedManager
# 案件日志附件存储
case_log_storage = KeepOriginalNameStorage()
logger = logging.getLogger(__name__)
def validate_log_attachment(file: UploadedFile) -> None:
"""验证日志附件"""
name = str(getattr(file, "name", ""))
size: int = int(getattr(file, "size", 0) or 0)
ext = Path(name).suffix.lower()
if ext not in CASE_LOG_ALLOWED_EXTENSIONS:
from django.core.exceptions import ValidationError
raise ValidationError(_("不支持的文件类型"))
if size and size > CASE_LOG_MAX_FILE_SIZE:
from django.core.exceptions import ValidationError
raise ValidationError(_("文件大小超过50MB限制"))
class CaseLog(models.Model):
id: int
case = models.ForeignKey(Case, on_delete=models.CASCADE, related_name="logs", verbose_name=_("案件"))
content = models.TextField(verbose_name=_("日志内容"))
actor = models.ForeignKey(
"organization.Lawyer", on_delete=models.PROTECT, related_name="case_logs", verbose_name=_("操作人")
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("创建日期"))
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("修改日期"))
if TYPE_CHECKING:
attachments: RelatedManager[CaseLogAttachment]
versions: RelatedManager[CaseLogVersion]
class Meta:
verbose_name = _("日志")
verbose_name_plural = _("日志")
indexes: ClassVar = [
models.Index(fields=["case", "-created_at"]),
models.Index(fields=["actor"]),
]
def __str__(self) -> str:
return f"{self.case_id}-{self.actor_id}-{self.created_at}"
def _exported_reminders(self) -> list[dict[str, Any]]:
cached = getattr(self, "_cached_exported_reminders", _SENTINEL)
if cached is not _SENTINEL:
return cached
if not getattr(self, "id", None):
self._cached_exported_reminders = []
return []
reminders: list[dict[str, Any]]
try:
from apps.core.interfaces import ServiceLocator
reminder_service = ServiceLocator.get_reminder_service()
reminders = reminder_service.export_case_log_reminders_internal(case_log_id=int(self.id))
except Exception:
logger.exception("case_log_export_reminders_failed", extra={"case_log_id": int(self.id)})
reminders = []
self._cached_exported_reminders = reminders
return reminders
@property
def reminder_entries(self) -> list[dict[str, Any]]:
return self._exported_reminders()
@property
def has_reminders(self) -> bool:
return bool(self._exported_reminders())
@property
def reminder_count(self) -> int:
return len(self._exported_reminders())
@property
def _latest_reminder(self) -> Any | None:
"""缓存最近的提醒记录，避免重复查询。"""
cached = getattr(self, "_cached_latest_reminder", _SENTINEL)
if cached is not _SENTINEL:
return cached
if not getattr(self, "id", None):
self._cached_latest_reminder = None
return None
reminder: dict[str, Any] | None = None
try:
from apps.core.interfaces import ServiceLocator
reminder_service = ServiceLocator.get_reminder_service()
reminder = reminder_service.get_latest_case_log_reminder_internal(case_log_id=int(self.id))
except Exception:
logger.exception("case_log_latest_reminder_failed", extra={"case_log_id": int(self.id)})
reminders = self._exported_reminders()
reminder = reminders[-1] if reminders else None
self._cached_latest_reminder = reminder
return reminder
@property
def reminder_type(self) -> str | None:
"""获取最近的提醒类型。"""
reminder = self._latest_reminder
if reminder is None:
return None
if isinstance(reminder, dict):
return str(reminder.get("reminder_type") or "")
return str(getattr(reminder, "reminder_type", ""))
@property
def reminder_time(self) -> datetime | None:
"""获取最近的提醒时间。"""
reminder = self._latest_reminder
if reminder is None:
return None
due_at = reminder.get("due_at") if isinstance(reminder, dict) else getattr(reminder, "due_at", None)
return due_at if isinstance(due_at, datetime) else None
class CaseLogAttachment(models.Model):
id: int
log = models.ForeignKey(CaseLog, on_delete=models.CASCADE, related_name="attachments", verbose_name=_("日志"))
file = models.FileField(
upload_to="case_logs/",
storage=case_log_storage,
validators=[validate_log_attachment],
verbose_name=_("相关文书"),
)
uploaded_at = models.DateTimeField(auto_now_add=True, verbose_name=_("上传时间"))
class Meta:
verbose_name = _("日志附件")
verbose_name_plural = _("日志附件")
class CaseLogVersion(models.Model):
id: int
log = models.ForeignKey(CaseLog, on_delete=models.CASCADE, related_name="versions", verbose_name=_("日志"))
content = models.TextField(verbose_name=_("历史内容"))
version_at = models.DateTimeField(auto_now_add=True, verbose_name=_("版本时间"))
actor = models.ForeignKey(
"organization.Lawyer", on_delete=models.PROTECT, related_name="case_log_versions", verbose_name=_("操作者")
)
class Meta:
verbose_name = _("案件日志版本")
verbose_name_plural = _("案件日志版本")
def __str__(self) -> str:
return f"{self.log_id}-{self.version_at}"
|