update

app/admin/filters.py

@@ -6,9 +6,12 @@ from typing import Any
 
 class RecentDateTimeFilter:
     """
-    最近时间筛选（避免依赖 DateTime 的 OperationColumnFilter 支持情况）：
+    最近时间范围筛选（用于 started_at/created_at 等 DateTime 列）：
-    - all: 全部
+    - all / 1h / 24h / 7d / 30d
-    - 1h/24h/7d/30d: 最近 N
+
+    SQLAdmin 自定义 ColumnFilter 约定：
+    - 必须提供 title / parameter_name
+    - 必须实现 lookups(request, model) 与 get_filtered_query(query, value)
     """
 
     def __init__(self, column: Any, *, title: str, parameter_name: str) -> None:
@@ -16,7 +19,7 @@ class RecentDateTimeFilter:
         self.title = title
         self.parameter_name = parameter_name
 
-    def lookups(self, request, model) -> list[tuple[str, str]]:  # noqa: ARG002 (framework signature)
+    def lookups(self, request, model) -> list[tuple[str, str]]:  # noqa: ARG002
         return [
             ("all", "全部"),
             ("1h", "最近1小时"),
@@ -25,25 +28,30 @@ class RecentDateTimeFilter:
             ("30d", "最近30天"),
         ]
 
-    def get_filtered_query(self, query, value: str):
+    def get_filtered_query(self, query, value):  # type: ignore[no-untyped-def]
         if not value or value == "all":
             return query
 
         now = datetime.utcnow()
+        delta: timedelta | None = None
         if value == "1h":
-            threshold = now - timedelta(hours=1)
+            delta = timedelta(hours=1)
         elif value == "24h":
-            threshold = now - timedelta(hours=24)
+            delta = timedelta(hours=24)
         elif value == "7d":
-            threshold = now - timedelta(days=7)
+            delta = timedelta(days=7)
         elif value == "30d":
-            threshold = now - timedelta(days=30)
+            delta = timedelta(days=30)
-        else:
+
+        if delta is None:
             return query
 
-        cond = self.column >= threshold
+        threshold = now - delta
-        # SQLAlchemy Select 在不同版本下可能是 where/filter；这里兼容两者
+
-        if hasattr(query, "where"):
+        # 兼容 SQLAlchemy Query / Select 两种风格
-            return query.where(cond)
+        if hasattr(query, "where"):
-        return query.filter(cond)
+            return query.where(self.column >= threshold)
+        if hasattr(query, "filter"):
+            return query.filter(self.column >= threshold)
+        return query
 

app/admin/views.py

@@ -77,6 +77,16 @@ class JobAdmin(ModelView, model=Job):
     # 列表页模板：加入每行 Run Now
     list_template = "job_list.html"
 
+    # 搜索
+    column_searchable_list = [Job.id, Job.handler_path, Job.cron_expr]
+
+    # 筛选
+    column_filters = [
+        BooleanFilter(Job.enabled),
+        OperationColumnFilter(Job.handler_path),
+        RecentDateTimeFilter(Job.created_at, title="创建时间", parameter_name="created_at_recent"),
+    ]
+
     # 编辑页排除 secret_cfg，避免回显密文；由自定义模板额外渲染一个空输入框
     # 注意：SQLAdmin 这里需要字段名字符串（不是 SQLAlchemy Column 对象）
     form_edit_rules = ["id", "enabled", "cron_expr", "handler_path", "public_cfg"]
@@ -105,16 +115,6 @@ class JobAdmin(ModelView, model=Job):
         Job.last_run_at: lambda m, a: _fmt_dt_seconds(m.last_run_at),
     }
 
-    # 搜索：顶部 Search 输入框
-    column_searchable_list = [Job.id, Job.handler_path, Job.cron_expr]
-
-    # 筛选：右侧 Filters 栏
-    column_filters = [
-        BooleanFilter(Job.enabled),
-        OperationColumnFilter(Job.handler_path),
-        RecentDateTimeFilter(Job.created_at, title="创建时间", parameter_name="created_at_recent"),
-    ]
-
     @action(
         name="run_now",
         label="立即运行",
@@ -226,6 +226,25 @@ class JobLogAdmin(ModelView, model=JobLog):
     # 为 JobLog 详情页单独指定模板（用于加入 Retry 按钮）
     details_template = "joblog_details.html"
 
+    # 搜索
+    column_searchable_list = [JobLog.job_id, JobLog.message, JobLog.celery_task_id]
+
+    # 筛选
+    column_filters = [
+        StaticValuesFilter(
+            JobLog.status,
+            values=[
+                ("RUNNING", "运行中"),
+                ("SUCCESS", "成功"),
+                ("FAILURE", "失败"),
+                ("RETRY", "重试"),
+            ],
+            title="状态",
+        ),
+        OperationColumnFilter(JobLog.attempt),
+        RecentDateTimeFilter(JobLog.started_at, title="开始时间", parameter_name="started_at_recent"),
+    ]
+
     column_labels = {
         "id": "日志ID",
         "job_id": "任务ID",
@@ -240,25 +259,6 @@ class JobLogAdmin(ModelView, model=JobLog):
         "finished_at": "结束时间",
     }
 
-    # 搜索：任务ID / message / celery_task_id
-    column_searchable_list = [JobLog.job_id, JobLog.message, JobLog.celery_task_id]
-
-    # 筛选：状态 / 重试次数 / 最近开始时间
-    column_filters = [
-        StaticValuesFilter(
-            JobLog.status,
-            options=(
-                ("RUNNING", "运行中"),
-                ("SUCCESS", "成功"),
-                ("FAILURE", "失败"),
-                ("RETRY", "重试"),
-            ),
-            title="状态",
-        ),
-        OperationColumnFilter(JobLog.attempt),
-        RecentDateTimeFilter(JobLog.started_at, title="开始时间", parameter_name="started_at_recent"),
-    ]
-
     column_formatters = {
         JobLog.started_at: lambda m, a: _fmt_dt_seconds(m.started_at),
         JobLog.finished_at: lambda m, a: _fmt_dt_seconds(m.finished_at),