46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.database import GateAccessLog, get_db
|
|
from core.dependencies import require_manager
|
|
from core.schemas import StatsPage
|
|
|
|
router = APIRouter(prefix="/api/admin/stats", tags=["admin-stats"])
|
|
|
|
|
|
@router.get("", response_model=StatsPage)
|
|
async def get_stats(
|
|
db: Session = Depends(get_db),
|
|
_: dict = Depends(require_manager),
|
|
gate_id: Optional[int] = Query(None),
|
|
keypass_code: Optional[str] = Query(None),
|
|
success: Optional[bool] = Query(None),
|
|
date_from: Optional[datetime] = Query(None),
|
|
date_to: Optional[datetime] = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
):
|
|
q = db.query(GateAccessLog)
|
|
if gate_id is not None:
|
|
q = q.filter(GateAccessLog.gate_id == gate_id)
|
|
if keypass_code:
|
|
q = q.filter(GateAccessLog.keypass_code.ilike(f"%{keypass_code}%"))
|
|
if success is not None:
|
|
q = q.filter(GateAccessLog.success == success)
|
|
if date_from is not None:
|
|
q = q.filter(GateAccessLog.timestamp >= date_from)
|
|
if date_to is not None:
|
|
q = q.filter(GateAccessLog.timestamp <= date_to)
|
|
|
|
total = q.count()
|
|
items = (
|
|
q.order_by(GateAccessLog.timestamp.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
return StatsPage(total=total, page=page, page_size=page_size, items=items)
|