116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from openpyxl import Workbook
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import require_roles
|
|
from app.core.paths import data_path
|
|
from app.db.session import get_db
|
|
from app.models import AdminUser, OperationLog
|
|
from app.schemas.log import OperationLogOut
|
|
from app.services.logs import HIGH_RISK_LOG_RETENTION_DAYS, LOG_RETENTION_DAYS, cleanup_expired_logs, format_log
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=list[OperationLogOut])
|
|
def list_logs(
|
|
action: str | None = Query(default=None),
|
|
object_type: str | None = Query(default=None),
|
|
risk: str | None = Query(default=None),
|
|
keyword: str | None = Query(default=None),
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=100),
|
|
db: Session = Depends(get_db),
|
|
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin", "readonly")),
|
|
) -> dict[str, object]:
|
|
items = _query_logs(db, action, object_type, risk, keyword)
|
|
total = len(items)
|
|
total_pages = max(1, (total + page_size - 1) // page_size)
|
|
page = min(page, total_pages)
|
|
start = (page - 1) * page_size
|
|
return {
|
|
"items": items[start:start + page_size],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total_pages": total_pages,
|
|
}
|
|
|
|
|
|
@router.get("/export")
|
|
def export_logs(
|
|
action: str | None = Query(default=None),
|
|
object_type: str | None = Query(default=None),
|
|
risk: str | None = Query(default=None),
|
|
keyword: str | None = Query(default=None),
|
|
db: Session = Depends(get_db),
|
|
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin", "readonly")),
|
|
) -> StreamingResponse:
|
|
items = _query_logs(db, action, object_type, risk, keyword)
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "\u64cd\u4f5c\u65e5\u5fd7"
|
|
sheet.append(["\u65f6\u95f4", "\u64cd\u4f5c\u4ebaID", "\u64cd\u4f5c", "\u5bf9\u8c61", "\u5bf9\u8c61ID", "\u98ce\u9669\u7b49\u7ea7", "\u6458\u8981", "\u8be6\u60c5"])
|
|
for item in items:
|
|
sheet.append([
|
|
item.get("created_at"),
|
|
item.get("admin_user_id"),
|
|
item.get("action_label"),
|
|
item.get("object_label"),
|
|
item.get("object_id"),
|
|
item.get("risk_label"),
|
|
item.get("summary"),
|
|
item.get("detail_json"),
|
|
])
|
|
for column in "ABCDEFGH":
|
|
sheet.column_dimensions[column].width = 22 if column != "H" else 60
|
|
export_path = data_path("exports") / "operation-logs.xlsx"
|
|
workbook.save(export_path)
|
|
file_handle = export_path.open("rb")
|
|
return StreamingResponse(
|
|
file_handle,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": 'attachment; filename="operation-logs.xlsx"'},
|
|
)
|
|
|
|
|
|
@router.post("/cleanup")
|
|
def cleanup_logs(
|
|
db: Session = Depends(get_db),
|
|
_: AdminUser = Depends(require_roles("system_admin", "certificate_admin")),
|
|
) -> dict[str, int]:
|
|
removed = cleanup_expired_logs(db)
|
|
db.commit()
|
|
return {
|
|
"removed": removed,
|
|
"normal_retention_days": LOG_RETENTION_DAYS,
|
|
"high_risk_retention_days": HIGH_RISK_LOG_RETENTION_DAYS,
|
|
}
|
|
|
|
|
|
def _query_logs(
|
|
db: Session,
|
|
action: str | None,
|
|
object_type: str | None,
|
|
risk: str | None,
|
|
keyword: str | None,
|
|
) -> list[dict[str, object]]:
|
|
query = db.query(OperationLog).order_by(OperationLog.id.desc())
|
|
cleanup_expired_logs(db)
|
|
db.commit()
|
|
if action:
|
|
query = query.filter(OperationLog.action == action)
|
|
if object_type:
|
|
query = query.filter(OperationLog.object_type == object_type)
|
|
items = [format_log(item) for item in query.limit(500).all()]
|
|
if risk:
|
|
items = [item for item in items if item["risk"] == risk]
|
|
if keyword:
|
|
word = keyword.strip().lower()
|
|
items = [
|
|
item for item in items
|
|
if word in " ".join(str(item.get(key, "")) for key in ["object_id", "summary", "detail_json", "action_label", "object_label"]).lower()
|
|
]
|
|
return items[:200]
|