50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
|
||
class RecentDateTimeFilter:
|
||
"""
|
||
最近时间筛选（避免依赖 DateTime 的 OperationColumnFilter 支持情况）：
|
||
- all: 全部
|
||
- 1h/24h/7d/30d: 最近 N
|
||
"""
|
||
|
||
def __init__(self, column: Any, *, title: str, parameter_name: str) -> None:
|
||
self.column = column
|
||
self.title = title
|
||
self.parameter_name = parameter_name
|
||
|
||
def lookups(self, request, model) -> list[tuple[str, str]]: # noqa: ARG002 (framework signature)
|
||
return [
|
||
("all", "全部"),
|
||
("1h", "最近 1 小时"),
|
||
("24h", "最近 24 小时"),
|
||
("7d", "最近 7 天"),
|
||
("30d", "最近 30 天"),
|
||
]
|
||
|
||
def get_filtered_query(self, query, value: str):
|
||
if not value or value == "all":
|
||
return query
|
||
|
||
now = datetime.utcnow()
|
||
if value == "1h":
|
||
threshold = now - timedelta(hours=1)
|
||
elif value == "24h":
|
||
threshold = now - timedelta(hours=24)
|
||
elif value == "7d":
|
||
threshold = now - timedelta(days=7)
|
||
elif value == "30d":
|
||
threshold = now - timedelta(days=30)
|
||
else:
|
||
return query
|
||
|
||
cond = self.column >= threshold
|
||
# SQLAlchemy Select 在不同版本下可能是 where/filter；这里兼容两者
|
||
if hasattr(query, "where"):
|
||
return query.where(cond)
|
||
return query.filter(cond)
|
||
|