| """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) |
|
|
