76 lines
4.0 KiB
Python
76 lines
4.0 KiB
Python
"""效能分析 API — ECharts 数据源（个人能力图谱 / 设备流转对比 / 筛选选项）"""
|
||
from datetime import datetime
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.database import get_db
|
||
from app.services.analytics_service import (
|
||
get_capability_profile, CapabilityResponse,
|
||
get_flow_compare, FlowResponse,
|
||
get_analytics_options, AnalyticsOptions,
|
||
get_device_records, DeviceRecord,
|
||
)
|
||
|
||
router = APIRouter(prefix="/analytics", tags=["效能分析"])
|
||
|
||
|
||
@router.get("/capability", response_model=CapabilityResponse)
|
||
async def capability_profile(
|
||
assignee_ids: str | None = Query(None, description="负责人ID，逗号分隔"),
|
||
spec_models: str | None = Query(None, description="规格型号，逗号分隔（可选）"),
|
||
since: str | None = Query(None, description="起始日期 ISO"),
|
||
until: str | None = Query(None, description="截止日期 ISO"),
|
||
mode: str = Query("workdays", description="耗时口径: workdays(工作小时,默认) / natural(自然小时)"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""个人能力图谱 — X 轴=设备身份证，分组柱状图（单台设备总耗时）。"""
|
||
since_dt = datetime.fromisoformat(since) if since else None
|
||
until_dt = datetime.fromisoformat(until) if until else None
|
||
ids = [s.strip() for s in assignee_ids.split(",") if s.strip()] if assignee_ids else None
|
||
specs = [s.strip() for s in spec_models.split(",") if s.strip()] if spec_models else None
|
||
return await get_capability_profile(
|
||
db, assignee_ids=ids, spec_models=specs,
|
||
since=since_dt, until=until_dt, mode=mode,
|
||
)
|
||
|
||
|
||
@router.get("/flow", response_model=FlowResponse)
|
||
async def flow_compare(
|
||
product_sns: str | None = Query(None, description="身份证，逗号分隔"),
|
||
spec_models: str | None = Query(None, description="规格型号，逗号分隔（无 product_sns 时按型号取最近设备）"),
|
||
mode: str = Query("natural", description="时间口径: natural(自然小时) / workdays(工作小时,排除周末节假日)"),
|
||
since: str | None = Query(None, description="起始日期 ISO"),
|
||
until: str | None = Query(None, description="截止日期 ISO"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""设备流转对比 — 每台设备各操作人耗时（堆叠柱状，按人堆叠，识别瓶颈）。"""
|
||
sns = [s.strip() for s in product_sns.split(",") if s.strip()] if product_sns else None
|
||
specs = [s.strip() for s in spec_models.split(",") if s.strip()] if spec_models else None
|
||
since_dt = datetime.fromisoformat(since) if since else None
|
||
until_dt = datetime.fromisoformat(until) if until else None
|
||
return await get_flow_compare(db, product_sns=sns, spec_models=specs, mode=mode, since=since_dt, until=until_dt)
|
||
|
||
|
||
@router.get("/options", response_model=AnalyticsOptions)
|
||
async def analytics_options(
|
||
assignee_ids: str | None = Query(None, description="负责人ID，逗号分隔（联动过滤型号）"),
|
||
spec_models: str | None = Query(None, description="规格型号，逗号分隔（联动过滤人员）"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""顶部筛选栏选项 — 负责人 + 规格型号，支持动态联动。"""
|
||
ids = [s.strip() for s in assignee_ids.split(",") if s.strip()] if assignee_ids else None
|
||
specs = [s.strip() for s in spec_models.split(",") if s.strip()] if spec_models else None
|
||
return await get_analytics_options(db, assignee_ids=ids, spec_models=specs)
|
||
|
||
|
||
@router.get("/device-records", response_model=list[DeviceRecord])
|
||
async def device_records(
|
||
product_sn: str = Query(..., description="设备身份证"),
|
||
assignee_ids: str | None = Query(None, description="负责人ID，逗号分隔（可选，用于过滤）"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""某台设备的任务备注记录（含图片）；可选按负责人过滤。"""
|
||
ids = [s.strip() for s in assignee_ids.split(",") if s.strip()] if assignee_ids else None
|
||
return await get_device_records(db, product_sn, assignee_ids=ids)
|