75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""Admin view and pause control for historical audit-data cleanup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from time import time
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Request
|
|
from pydantic import BaseModel
|
|
|
|
from app.gateway.admin_audit_retention_control import (
|
|
is_paused,
|
|
read_status,
|
|
set_paused,
|
|
write_status,
|
|
)
|
|
from app.gateway.routers.admin_users import _require_admin
|
|
from deerflow.config.runtime_paths import runtime_home
|
|
from deerflow.persistence.types import BEIJING_TZ
|
|
|
|
router = APIRouter(prefix="/api/admin/audit-retention", tags=["admin-audit-retention"])
|
|
|
|
|
|
class RetentionControlRequest(BaseModel):
|
|
paused: bool
|
|
|
|
|
|
def _status_payload() -> dict[str, Any]:
|
|
root = runtime_home()
|
|
value = read_status(root)
|
|
paused = is_paused(root)
|
|
value["paused"] = paused
|
|
value["pause_requested"] = paused and value.get("status") == "running"
|
|
if value["pause_requested"]:
|
|
value["display_status"] = "pausing"
|
|
elif paused:
|
|
value["display_status"] = "paused"
|
|
else:
|
|
value["display_status"] = value.get("status", "scheduled")
|
|
return value
|
|
|
|
|
|
def _apply_control(paused: bool) -> dict[str, Any]:
|
|
root = runtime_home()
|
|
set_paused(root, paused)
|
|
value = read_status(root)
|
|
if paused and value.get("status") != "running":
|
|
value.update(status="paused", current_source=None, next_run_at=None)
|
|
elif not paused and value.get("status") == "paused":
|
|
value.update(status="scheduled", next_run_at=time())
|
|
write_status(root, value)
|
|
return _status_payload()
|
|
|
|
|
|
@router.get("")
|
|
async def get_retention_status(request: Request) -> dict[str, Any]:
|
|
await _require_admin(request)
|
|
return await asyncio.to_thread(_status_payload)
|
|
|
|
|
|
@router.put("/control")
|
|
async def update_retention_control(
|
|
payload: RetentionControlRequest,
|
|
request: Request,
|
|
) -> dict[str, Any]:
|
|
await _require_admin(request)
|
|
value = await asyncio.to_thread(_apply_control, payload.paused)
|
|
value["control_updated_at"] = datetime.now(BEIJING_TZ).isoformat(timespec="seconds")
|
|
return value
|
|
|
|
|
|
__all__ = ["router"]
|