Samir Boulahtit
6d6eba75bf
Some checks failed
Migrates scanning pipeline from marketing-.lu-domains app into Orion module. Supports digital (domain scan) and offline (manual capture) lead channels with enrichment, scoring, campaign management, and interaction tracking. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
# app/modules/prospecting/routes/api/admin_stats.py
|
|
"""
|
|
Admin API routes for dashboard statistics.
|
|
"""
|
|
|
|
import logging
|
|
from math import ceil
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_admin_api
|
|
from app.core.database import get_db
|
|
from app.modules.prospecting.schemas.scan_job import (
|
|
ScanJobListResponse,
|
|
ScanJobResponse,
|
|
)
|
|
from app.modules.prospecting.services.stats_service import stats_service
|
|
from app.modules.tenancy.schemas.auth import UserContext
|
|
|
|
router = APIRouter(prefix="/stats")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@router.get("")
|
|
def get_overview_stats(
|
|
db: Session = Depends(get_db),
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
):
|
|
"""Get overview statistics for the dashboard."""
|
|
return stats_service.get_overview(db)
|
|
|
|
|
|
@router.get("/jobs", response_model=ScanJobListResponse)
|
|
def list_scan_jobs(
|
|
page: int = Query(1, ge=1),
|
|
per_page: int = Query(20, ge=1, le=100),
|
|
status: str | None = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_admin: UserContext = Depends(get_current_admin_api),
|
|
):
|
|
"""Get paginated scan jobs."""
|
|
jobs, total = stats_service.get_scan_jobs(db, page=page, per_page=per_page, status=status)
|
|
return ScanJobListResponse(
|
|
items=[ScanJobResponse.model_validate(j) for j in jobs],
|
|
total=total,
|
|
page=page,
|
|
per_page=per_page,
|
|
pages=ceil(total / per_page) if per_page else 0,
|
|
)
|