File size: 8,366 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 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 | """
爬虫任务 Admin
"""
import json
import logging
from pathlib import Path
from typing import Any, ClassVar
from django.contrib import admin
from django.http import HttpRequest
from django.utils.html import format_html, format_html_join
from django.utils.safestring import SafeData
from django.utils.translation import gettext_lazy as _
from apps.automation.models import ScraperTask
logger = logging.getLogger(__name__)
@admin.register(ScraperTask)
class ScraperTaskAdmin(admin.ModelAdmin[ScraperTask]):
"""爬虫任务管理"""
def get_model_perms(self, request: HttpRequest) -> dict[str, bool]:
"""隐藏后台入口（保留代码与直达地址能力）"""
return {}
list_display = (
"id",
"task_type",
"priority",
"status_colored",
"retry_info",
"url_short",
"case",
"created_at",
"duration",
)
list_filter = ("task_type", "status", "priority", "created_at")
search_fields = ("url", "error_message")
readonly_fields = (
"created_at",
"updated_at",
"started_at",
"finished_at",
"result_display",
"error_message",
"retry_count",
)
fieldsets = (
(_("基本信息"), {"fields": ("task_type", "status", "priority", "url", "case")}),
(_("重试配置"), {"fields": ("retry_count", "max_retries", "scheduled_at")}),
(_("配置"), {"fields": ("config",), "classes": ("collapse",)}),
(_("执行结果"), {"fields": ("result_display", "error_message")}),
(_("时间信息"), {"fields": ("created_at", "started_at", "finished_at", "updated_at")}),
)
@admin.display(description="状态")
def status_colored(self, obj: Any) -> SafeData:
"""带颜色的状态显示"""
colors = {
"pending": "#ffa500",
"running": "#007bff",
"success": "#28a745",
"failed": "#dc3545",
}
color = colors.get(obj.status, "#666")
return format_html(
'<span style="color: {}; font-weight: bold;">{}</span>',
color,
obj.get_status_display(),
)
@admin.display(description="URL")
def url_short(self, obj: Any) -> str:
"""缩短的 URL 显示"""
url: str = str(obj.url)
if len(url) > 50:
return url[:50] + "..."
return url
@admin.display(description="重试")
def retry_info(self, obj: Any) -> SafeData | str:
"""显示重试信息"""
if obj.retry_count > 0:
return format_html(
'<span style="color: #ffa500;">{}/{}</span>',
obj.retry_count,
obj.max_retries,
)
return f"0/{obj.max_retries}"
@admin.display(description="耗时")
def duration(self, obj: Any) -> str:
"""计算任务耗时"""
if obj.started_at and obj.finished_at:
delta = obj.finished_at - obj.started_at
seconds = delta.total_seconds()
if seconds < 60:
return f"{seconds:.1f}秒"
else:
minutes = seconds / 60
return f"{minutes:.1f}分钟"
return "-"
def _file_icon(self, filename: str) -> str:
"""根据文件扩展名返回图标"""
if filename.endswith(".pdf"):
return "📄"
if filename.endswith(".zip"):
return "📦"
if filename.endswith((".doc", ".docx")):
return "📝"
return "📎"
def _render_files_html(self, files: list[str]) -> SafeData:
"""渲染文件列表 HTML，返回安全的 format_html 结果"""
from django.conf import settings
def _file_item(f: str) -> SafeData:
filename = f.split("/")[-1] if "/" in f else f
try:
file_path = Path(f)
media_root = Path(settings.MEDIA_ROOT)
relative_path = file_path.relative_to(media_root)
file_url = settings.MEDIA_URL + str(relative_path)
icon = self._file_icon(filename)
return format_html(
'<li style="margin: 5px 0;">'
'<a href="{}" target="_blank" style="color: #0066cc; text-decoration: none;">'
"{} {}</a></li>",
file_url,
icon,
filename,
)
except Exception:
icon = self._file_icon(filename)
return format_html(
'<li style="margin: 5px 0;">{} {}</li>',
icon,
filename,
)
items = format_html_join("", "{}", ((_file_item(f),) for f in files))
return format_html(
'<div style="margin-top: 10px;"><strong>📁 下载的文件:</strong>'
'<ul style="list-style: none; padding-left: 0;">{}</ul></div>',
items,
)
def _render_screenshots_html(self, screenshots: list[str]) -> SafeData:
"""渲染截图列表 HTML，返回安全的 format_html 结果"""
from django.conf import settings
def _screenshot_item(ss: str) -> SafeData:
if ss.startswith(str(settings.MEDIA_ROOT)):
ss_url = ss.replace(str(settings.MEDIA_ROOT), settings.MEDIA_URL)
return format_html(
'<br><img src="{}" style="max-width: 600px; border: 1px solid #ddd; margin-top: 10px;">',
ss_url,
)
return format_html("{}", "")
return format_html_join("", "{}", ((_screenshot_item(ss),) for ss in screenshots))
@admin.display(description="执行结果")
def result_display(self, obj: Any) -> SafeData | str:
"""格式化显示结果"""
if not obj.result:
return "-"
result_json = json.dumps(obj.result, indent=2, ensure_ascii=False)
screenshot: str | None = obj.result.get("screenshot")
screenshots: list[str] = obj.result.get("screenshots", [])
files: list[str] = obj.result.get("files", [])
pre_block = format_html(
'<pre style="background: #f5f5f5; padding: 10px; border-radius: 4px;'
' max-height: 300px; overflow: auto;">{}</pre>',
result_json,
)
files_block = self._render_files_html(files) if files else format_html("{}", "")
if screenshot:
from django.conf import settings
if screenshot.startswith(str(settings.MEDIA_ROOT)):
screenshot_url = screenshot.replace(str(settings.MEDIA_ROOT), settings.MEDIA_URL)
single_screenshot_block = format_html(
'<br><img src="{}" style="max-width: 600px; border: 1px solid #ddd;">',
screenshot_url,
)
else:
single_screenshot_block = format_html("{}", "")
else:
single_screenshot_block = format_html("{}", "")
screenshots_block = self._render_screenshots_html(screenshots) if screenshots else format_html("{}", "")
return format_html(
"{}{}{}{}",
pre_block,
files_block,
single_screenshot_block,
screenshots_block,
)
@admin.action(description=_("立即执行选中的任务"))
def execute_tasks(self, request: Any, queryset: Any) -> None:
"""批量执行任务"""
from django_q.tasks import async_task
count = 0
for task in queryset:
if task.status in ["pending", "failed"]:
async_task("apps.automation.tasks.execute_scraper_task", task.id)
count += 1
logger.info("已提交 %d 个任务到后台队列", count)
self.message_user(request, _(f"已提交 {count} 个任务到后台队列"))
@admin.action(description=_("重置失败任务状态"))
def reset_failed_tasks(self, request: Any, queryset: Any) -> None:
"""重置失败任务，允许重新执行"""
count = queryset.filter(status="failed").update(status="pending", retry_count=0, error_message=None)
logger.info("已重置 %d 个失败任务", count)
self.message_user(request, _(f"已重置 {count} 个失败任务"))
|