refactor(admin): 收敛后台接口封装与页面状态逻辑

- 简化绑定和日志接口的查询、序列化与前端数据请求路径
- 统一登录流程与前端 API 调用层,补充后台图标依赖
- 抽取通用异步状态处理,减少多个管理页面的重复逻辑
This commit is contained in:
2026-03-04 00:18:47 +08:00
parent ab1bd90c65
commit 0a1eeb9ddf
12 changed files with 556 additions and 305 deletions

View File

@@ -34,6 +34,33 @@ def apply_log_filters(
return statement
def to_log_item(item: InterceptLog) -> InterceptLogItem:
return InterceptLogItem(
id=item.id,
token_display=item.token_display,
bound_ip=str(item.bound_ip),
attempt_ip=str(item.attempt_ip),
alerted=item.alerted,
intercepted_at=item.intercepted_at,
)
def write_log_csv(buffer: io.StringIO, logs: list[InterceptLog]) -> None:
writer = csv.writer(buffer)
writer.writerow(["id", "token_display", "bound_ip", "attempt_ip", "alerted", "intercepted_at"])
for item in logs:
writer.writerow(
[
item.id,
item.token_display,
str(item.bound_ip),
str(item.attempt_ip),
item.alerted,
item.intercepted_at.isoformat(),
]
)
@router.get("", response_model=LogListResponse)
async def list_logs(
page: int = Query(default=1, ge=1),
@@ -54,17 +81,7 @@ async def list_logs(
).all()
return LogListResponse(
items=[
InterceptLogItem(
id=item.id,
token_display=item.token_display,
bound_ip=str(item.bound_ip),
attempt_ip=str(item.attempt_ip),
alerted=item.alerted,
intercepted_at=item.intercepted_at,
)
for item in logs
],
items=[to_log_item(item) for item in logs],
total=total,
page=page,
page_size=page_size,
@@ -85,19 +102,7 @@ async def export_logs(
logs = (await session.scalars(statement)).all()
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(["id", "token_display", "bound_ip", "attempt_ip", "alerted", "intercepted_at"])
for item in logs:
writer.writerow(
[
item.id,
item.token_display,
str(item.bound_ip),
str(item.attempt_ip),
item.alerted,
item.intercepted_at.isoformat(),
]
)
write_log_csv(buffer, logs)
filename = f"sentinel-logs-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.csv"
return StreamingResponse(