File size: 4,138 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 | """MessageSource Admin — 消息来源管理。"""
from __future__ import annotations
from typing import Any, ClassVar
from django.contrib import admin, messages
from django.http import HttpRequest, HttpResponse
from django.urls import path
from django.utils.html import format_html
from django.utils.safestring import SafeString
from django.utils.translation import gettext_lazy as _
from apps.message_hub.models import MessageSource, SyncStatus
@admin.register(MessageSource)
class MessageSourceAdmin(admin.ModelAdmin[MessageSource]):
list_display = [
"id",
"display_name",
"source_type_badge",
"credential",
"is_enabled",
"poll_interval_minutes",
"sync_status_badge",
"last_sync_at",
"refresh_button",
]
list_filter: ClassVar = ["source_type", "is_enabled", "last_sync_status"]
search_fields: ClassVar = ("display_name", "credential__account", "credential__site_name")
autocomplete_fields: ClassVar = ("credential",)
readonly_fields: ClassVar = ["last_sync_at", "last_sync_status", "last_sync_error", "last_synced_uid", "created_at"]
fieldsets: ClassVar = (
(
_("基本配置"),
{
"fields": (
"display_name",
"credential",
"source_type",
"is_enabled",
"poll_interval_minutes",
"sync_since",
)
},
),
(
_("发件人过滤"),
{
"fields": ("sender_whitelist", "sender_blacklist"),
"description": _("可输入邮箱地址或发件人名称，每行一个，大小写不敏感。白名单优先于黑名单。"),
},
),
(_("IMAP 配置"), {"fields": ("imap_host", "imap_account"), "classes": ("collapse",)}),
(
_("同步状态"),
{"fields": ("last_sync_at", "last_sync_status", "last_sync_error", "last_synced_uid", "created_at")},
),
)
def get_urls(self) -> list[Any]:
urls = super().get_urls()
custom = [
path("<int:pk>/sync/", self.admin_site.admin_view(self._sync_view), name="message_hub_messagesource_sync"),
]
return custom + urls
def _sync_view(self, request: HttpRequest, pk: int) -> HttpResponse:
from django.shortcuts import redirect
from django_q.tasks import async_task
async_task("apps.message_hub.tasks.sync_source_by_id", pk)
self.message_user(request, _("同步任务已提交，稍后刷新查看结果"), messages.SUCCESS)
return redirect("..")
@admin.display(description=_("来源类型"))
def source_type_badge(self, obj: MessageSource) -> SafeString:
colors = {"imap": "#007bff", "court_inbox": "#6f42c1", "court_schedule": "#e85d04"}
color = colors.get(obj.source_type, "#6c757d")
return format_html(
'<span style="background:{};color:#fff;padding:2px 8px;border-radius:4px;font-size:12px">{}</span>',
color,
obj.get_source_type_display(),
)
@admin.display(description=_("同步状态"))
def sync_status_badge(self, obj: MessageSource) -> SafeString:
colors = {SyncStatus.SUCCESS: "#28a745", SyncStatus.FAILED: "#dc3545", SyncStatus.PENDING: "#6c757d"}
color = colors.get(obj.last_sync_status, "#6c757d")
return format_html(
'<span style="background:{};color:#fff;padding:2px 8px;border-radius:4px;font-size:12px">{}</span>',
color,
obj.get_last_sync_status_display(),
)
@admin.display(description=_("操作"))
def refresh_button(self, obj: MessageSource) -> SafeString:
from django.urls import reverse
url = reverse("admin:message_hub_messagesource_sync", args=[obj.pk])
return format_html(
'<a class="button" href="{}" style="background:#17a2b8;color:#fff;padding:4px 10px;border-radius:4px;text-decoration:none;font-size:12px">立即同步</a>',
url,
)
|