File size: 7,635 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 | """Module for chat."""
from __future__ import annotations
import json
from datetime import timedelta
from typing import TYPE_CHECKING, ClassVar
from django.db import models
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from apps.core.models.enums import ChatPlatform
from .case import Case
if TYPE_CHECKING:
from django.db.models.fields.related_descriptors import RelatedManager
class CaseChat(models.Model):
"""案件群聊"""
id: int
case = models.ForeignKey(Case, on_delete=models.CASCADE, related_name="chats", verbose_name=_("案件"))
platform = models.CharField(
max_length=32, choices=ChatPlatform.choices, default=ChatPlatform.FEISHU, verbose_name=_("平台")
)
chat_id = models.CharField(max_length=128, verbose_name=_("群聊ID"))
name = models.CharField(max_length=255, verbose_name=_("群名"))
is_active = models.BooleanField(default=True, verbose_name=_("是否有效"))
owner_id = models.CharField(
max_length=64,
null=True,
blank=True,
verbose_name=_("群主ID"),
help_text=_("飞书用户的open_id或其他平台的用户标识符"),
)
owner_verified = models.BooleanField(default=False, verbose_name=_("群主已验证"), help_text=_("群主设置是否已验证"))
owner_verified_at = models.DateTimeField(
null=True, blank=True, verbose_name=_("群主验证时间"), help_text=_("群主设置验证成功的时间")
)
creation_audit_log: dict[str, object] = models.JSONField(
default=dict, verbose_name=_("创建审计日志"), help_text=_("群聊创建过程的详细日志,包含群主设置信息")
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("创建时间"))
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("更新时间"))
if TYPE_CHECKING:
audit_logs: RelatedManager[ChatAuditLog]
class Meta:
verbose_name = _("群聊")
verbose_name_plural = _("群聊")
unique_together: ClassVar = [["case", "platform", "chat_id"]]
indexes: ClassVar = [
models.Index(fields=["case", "platform"]),
models.Index(fields=["chat_id"]),
models.Index(fields=["is_active"]),
models.Index(fields=["owner_id"]),
models.Index(fields=["owner_verified"]),
models.Index(fields=["owner_verified_at"]),
]
def __str__(self) -> str:
platform_display = self.get_platform_display()
return f"[{platform_display}] {self.name}"
def get_owner_display(self) -> str:
"""获取群主显示信息"""
if not self.owner_id:
return str(_("未设置群主"))
status = str(_("已验证")) if self.owner_verified else str(_("未验证"))
return f"{self.owner_id} ({status})"
def is_owner_verified_recently(self, hours: int = 24) -> bool:
"""检查群主是否在最近时间内验证过"""
if not self.owner_verified or not self.owner_verified_at:
return False
cutoff_time = timezone.now() - timedelta(hours=hours)
return bool(self.owner_verified_at >= cutoff_time)
def get_creation_summary(self) -> str:
"""获取创建摘要信息"""
summary_parts = [f"群聊: {self.name}"]
if self.owner_id:
summary_parts.append(f"群主: {self.owner_id}")
if self.owner_verified:
summary_parts.append("已验证")
return " | ".join(summary_parts)
class ChatAuditLog(models.Model):
"""群聊审计日志"""
id: int
ACTION_CHOICES: ClassVar[list[tuple[str, str]]] = [
("CREATE_START", _("开始创建")),
("CREATE_SUCCESS", _("创建成功")),
("CREATE_FAILED", _("创建失败")),
("OWNER_SET", _("设置群主")),
("OWNER_VERIFY", _("验证群主")),
("OWNER_RETRY", _("重试群主设置")),
("OWNER_SET_FAILED", _("群主设置失败")),
("CONFIG_ERROR", _("配置错误")),
]
chat = models.ForeignKey(
CaseChat,
on_delete=models.CASCADE,
related_name="audit_logs",
null=True,
blank=True,
verbose_name=_("关联群聊"),
help_text=_("关联的群聊,某些操作(如配置错误)可能没有关联群聊"),
)
case = models.ForeignKey(
Case,
on_delete=models.CASCADE,
related_name="chat_audit_logs",
null=True,
blank=True,
verbose_name=_("关联案件"),
help_text=_("关联的案件"),
)
action = models.CharField(
max_length=32, choices=ACTION_CHOICES, verbose_name=_("操作类型"), help_text=_("执行的操作类型")
)
details: dict[str, object] = models.JSONField(
default=dict, verbose_name=_("操作详情"), help_text=_("操作的详细信息,以JSON格式存储")
)
timestamp = models.DateTimeField(auto_now_add=True, verbose_name=_("时间戳"), help_text=_("操作发生的时间"))
success = models.BooleanField(default=True, verbose_name=_("操作成功"), help_text=_("操作是否成功执行"))
error_message = models.TextField(blank=True, verbose_name=_("错误信息"), help_text=_("操作失败时的错误信息"))
external_chat_id = models.CharField(
max_length=128, blank=True, verbose_name=_("群聊ID"), help_text=_("群聊的外部ID,便于查询")
)
platform = models.CharField(
max_length=32,
choices=ChatPlatform.choices,
default=ChatPlatform.FEISHU,
verbose_name=_("平台"),
help_text=_("群聊平台"),
)
audit_version = models.CharField(
max_length=16, default="1.0", verbose_name=_("审计版本"), help_text=_("审计日志格式版本")
)
class Meta:
verbose_name = _("群聊审计日志")
verbose_name_plural = _("群聊审计日志")
ordering: ClassVar = ["-timestamp"]
indexes: ClassVar = [
models.Index(fields=["chat", "-timestamp"]),
models.Index(fields=["case", "-timestamp"]),
models.Index(fields=["action", "-timestamp"]),
models.Index(fields=["success", "-timestamp"]),
models.Index(fields=["external_chat_id", "-timestamp"]),
models.Index(fields=["platform", "-timestamp"]),
models.Index(fields=["-timestamp"]),
]
def __str__(self) -> str:
chat_info = f"Chat:{self.external_chat_id}" if self.external_chat_id else f"ChatModel:{self.chat_id}"
case_info = f"Case:{self.case_id}" if self.case_id else "NoCase"
status = "SUCCESS" if self.success else "FAILED"
action_display = self.get_action_display()
return f"[{action_display}] {chat_info} | {case_info} | {status}"
@property
def formatted_details(self) -> str:
"""获取格式化的详情信息"""
try:
return str(json.dumps(self.details, ensure_ascii=False, indent=2))
except (TypeError, ValueError):
return str(self.details)
@property
def summary(self) -> str:
"""获取操作摘要"""
action_display = self.get_action_display()
summary_parts = [str(action_display)]
if self.external_chat_id:
summary_parts.append(f"群聊:{self.external_chat_id}")
if self.case_id:
summary_parts.append(f"案件:{self.case_id}")
if not self.success and self.error_message:
summary_parts.append(f"错误:{self.error_message[:50]}...")
return " | ".join(summary_parts)
|