feat(dashboard): 人员工时台账——平铺明细+多维筛选+时间交集+动态工时+备注+CSV导出

backend/app/api/v1/endpoints/dashboard.py

@@ -1,6 +1,8 @@
 """Dashboard API — 上帝视角（全厂数据，无用户过滤）"""
+import io
 from datetime import datetime
 from fastapi import APIRouter, Depends, Query
+from fastapi.responses import StreamingResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.database import get_db
 from app.services.dashboard_service import (
@@ -8,6 +10,7 @@ from app.services.dashboard_service import (
     get_wip_tasks, WipTask,
     get_completed_tasks, CompletedTask,
     get_people_workload, PersonWorkload,
+    get_people_history, PersonHistoryRecord,
     search_product_messages, ProductMessageList,
 )
 
@@ -61,6 +64,76 @@ async def people_workload(
     return await get_people_workload(db)
 
 
+@router.get("/people-history", response_model=list[PersonHistoryRecord])
+async def people_history(
+    since: str | None = Query(None, description="起始日期 ISO"),
+    until: str | None = Query(None, description="截止日期 ISO"),
+    assignee_id: str | None = Query(None, description="负责人ID（精确）"),
+    spec_model: str | None = Query(None, description="规格型号（模糊）"),
+    product_sn: str | None = Query(None, description="身份证（模糊）"),
+    task_name: str | None = Query(None, description="任务名（模糊）"),
+    db: AsyncSession = Depends(get_db),
+):
+    """人员效能与工时台账 — 平铺 Task 明细，多维筛选 + 时间交集"""
+    since_dt = datetime.fromisoformat(since) if since else None
+    until_dt = datetime.fromisoformat(until) if until else None
+    return await get_people_history(
+        db, since=since_dt, until=until_dt,
+        assignee_id=assignee_id, spec_model=spec_model,
+        product_sn=product_sn, task_name=task_name,
+    )
+
+
+@router.get("/people-history/export")
+async def export_people_history(
+    since: str | None = Query(None, description="起始日期 ISO"),
+    until: str | None = Query(None, description="截止日期 ISO"),
+    assignee_id: str | None = Query(None, description="负责人ID（精确）"),
+    spec_model: str | None = Query(None, description="规格型号（模糊）"),
+    product_sn: str | None = Query(None, description="身份证（模糊）"),
+    task_name: str | None = Query(None, description="任务名（模糊）"),
+    db: AsyncSession = Depends(get_db),
+):
+    """导出工时台账为 Excel（与查询接口相同筛选条件）"""
+    since_dt = datetime.fromisoformat(since) if since else None
+    until_dt = datetime.fromisoformat(until) if until else None
+    records = await get_people_history(
+        db, since=since_dt, until=until_dt,
+        assignee_id=assignee_id, spec_model=spec_model,
+        product_sn=product_sn, task_name=task_name,
+    )
+
+    import csv
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["状态", "负责人", "身份证", "业务序列号", "产品名称", "规格型号", "任务名", "开始时间", "结束时间", "总耗时(小时)", "最新有效备注"])
+    status_label = {"WIP": "进行中", "PENDING": "待接收", "COMPLETED": "已完成"}
+    for r in records:
+        writer.writerow([
+            status_label.get(r.status, r.status),
+            r.assignee_name,
+            r.product_sn,
+            r.external_serial or "",
+            r.material_name,
+            r.spec_model,
+            r.task_name,
+            r.received_at or "",
+            r.completed_at or "进行中",
+            r.duration_hours,
+            r.latest_valid_remark or "",
+        ])
+
+    data = output.getvalue().encode("utf-8-sig")  # 带 BOM，Excel 正确识别中文
+    buf = io.BytesIO(data)
+    buf.seek(0)
+    return StreamingResponse(
+        buf,
+        media_type="text/csv; charset=utf-8",
+        headers={"Content-Disposition": "attachment; filename=people_history.csv"},
+    )
+
+
 @router.get("/messages", response_model=ProductMessageList)
 async def dashboard_messages(
     keyword: str = Query("", description="搜索: SN码/物料名/留言人/内容"),