File size: 6,937 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 | from __future__ import annotations
from typing import Any
from django.contrib import admin, messages
from django.db.models import QuerySet
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
from django.urls import reverse
from django.utils.html import format_html
from django.utils.safestring import SafeString
from django.utils.translation import gettext_lazy as _
from apps.cases.admin.base_admin import BaseTabularInline
from apps.cases.admin.mixins import CaseAdminServiceMixin
from apps.cases.models import CaseChat
@admin.register(CaseChat)
class CaseChatAdmin(CaseAdminServiceMixin, admin.ModelAdmin[CaseChat]):
"""案件群聊管理"""
list_display = ("name", "chat_id_display", "platform_display", "case_link", "status_display", "created_at")
list_filter = ("platform", "is_active", "created_at")
search_fields = ("name", "chat_id", "case__name")
readonly_fields = ("chat_id", "created_at", "updated_at")
fields = ("case", "platform", "chat_id", "name", "is_active", "created_at", "updated_at")
ordering = ("-created_at",)
actions = ["unbind_selected_chats"]
change_form_template = "admin/cases/casechat/change_form.html"
def chat_id_display(self, obj: CaseChat) -> str:
"""显示群聊ID（截断显示）"""
chat_id: str = obj.chat_id
if chat_id:
if len(chat_id) > 20:
return f"{chat_id[:20]}..."
return chat_id
return "-"
chat_id_display.short_description = _("群聊ID") # type: ignore[attr-defined]
def platform_display(self, obj: CaseChat) -> str:
"""显示平台（带图标）"""
platform_icons = {"feishu": "🚀", "dingtalk": "📱", "wechat_work": "💬", "telegram": "✈️", "slack": "💼"}
platform_labels = {"feishu": "飞书", "dingtalk": "钉钉", "wechat_work": "企业微信", "telegram": "Telegram", "slack": "Slack"}
icon = platform_icons.get(obj.platform, "📢")
display = platform_labels.get(obj.platform, obj.platform)
return f"{icon} {display}"
platform_display.short_description = _("平台") # type: ignore[attr-defined]
def case_link(self, obj: CaseChat) -> SafeString:
"""案件链接"""
case_id = getattr(obj, "case_id", None)
if case_id:
url = reverse("admin:cases_case_change", args=[case_id])
case_name = getattr(obj.case, "name", str(case_id))
return format_html('<a href="{}" target="_blank">{}</a>', url, case_name)
return format_html("<span>{}</span>", "-")
case_link.short_description = _("关联案件") # type: ignore[attr-defined]
def status_display(self, obj: CaseChat) -> SafeString:
"""状态显示"""
if obj.is_active:
return format_html('<span style="color: green;">●</span> {}', "有效")
else:
return format_html('<span style="color: red;">●</span> {}', "已解绑")
status_display.short_description = _("状态") # type: ignore[attr-defined]
def unbind_selected_chats(self, request: HttpRequest, queryset: QuerySet[CaseChat, CaseChat]) -> None:
"""批量解除绑定群聊"""
service = self._get_case_chat_service()
success_count = 0
for chat in queryset.filter(is_active=True):
try:
if service.unbind_chat(chat.id):
success_count += 1
except Exception as e:
msg = _("解除绑定群聊 %(chat)s 失败: %(error)s") % {
"chat": chat.name,
"error": str(e),
}
messages.error(request, msg)
if success_count > 0:
messages.success(request, _("成功解除绑定 %d 个群聊") % success_count)
unbind_selected_chats.short_description = _("解除绑定选中的群聊") # type: ignore[attr-defined]
def has_add_permission(self, request: HttpRequest) -> bool:
"""禁止直接添加群聊记录"""
return False
def has_delete_permission(self, request: HttpRequest, obj: CaseChat | None = None) -> bool:
"""禁止删除群聊记录"""
return False
def response_change(self, request: HttpRequest, obj: CaseChat) -> HttpResponse:
"""处理自定义操作"""
if "_unbind_chat" in request.POST:
service = self._get_case_chat_service()
try:
if service.unbind_chat(obj.id):
messages.success(request, _("成功解除绑定群聊: %s") % obj.name)
else:
messages.error(request, _("解除绑定群聊失败: %s") % obj.name)
except Exception as e:
messages.error(request, _("解除绑定群聊时发生错误: %s") % str(e))
return HttpResponseRedirect(reverse("admin:cases_casechat_changelist"))
return super().response_change(request, obj)
class CaseChatInline(BaseTabularInline):
"""案件群聊内联管理"""
model = CaseChat
extra = 0
fields = ("platform_display", "name", "chat_id_display", "status_display", "created_at")
readonly_fields = ("platform_display", "name", "chat_id_display", "status_display", "created_at")
ordering = ("platform", "-created_at")
def platform_display(self, obj: CaseChat) -> str:
"""显示平台（带图标）"""
if not obj.pk:
return ""
platform_icons = {"feishu": "🚀", "dingtalk": "📱", "wechat_work": "💬", "telegram": "✈️", "slack": "💼"}
icon = platform_icons.get(obj.platform, "📢")
display = obj.get_platform_display()
return f"{icon} {display}"
platform_display.short_description = _("平台") # type: ignore[attr-defined]
def chat_id_display(self, obj: CaseChat) -> str:
"""显示群聊ID（截断显示）"""
if not obj.pk or not obj.chat_id:
return ""
chat_id: str = obj.chat_id
if len(chat_id) > 15:
return f"{chat_id[:15]}..."
return chat_id
chat_id_display.short_description = _("群聊ID") # type: ignore[attr-defined]
def status_display(self, obj: CaseChat) -> SafeString:
"""状态显示"""
if not obj.pk:
return format_html("<span>{}</span>", "")
if obj.is_active:
return format_html('<span style="color: green;">●</span> {}', "有效")
else:
return format_html('<span style="color: red;">●</span> {}', "已解绑")
status_display.short_description = _("状态") # type: ignore[attr-defined]
def has_add_permission(self, request: HttpRequest, obj: Any = None) -> bool:
"""禁止直接添加群聊记录"""
return False
def has_delete_permission(self, request: HttpRequest, obj: Any = None) -> bool:
"""禁止删除群聊记录"""
return False
|