diff --git a/backend/app/api/v1/endpoints/dashboard.py b/backend/app/api/v1/endpoints/dashboard.py index a38adbf..1a7781b 100644 --- a/backend/app/api/v1/endpoints/dashboard.py +++ b/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码/物料名/留言人/内容"), diff --git a/backend/app/services/dashboard_service.py b/backend/app/services/dashboard_service.py index 5773f40..d3f2427 100644 --- a/backend/app/services/dashboard_service.py +++ b/backend/app/services/dashboard_service.py @@ -55,6 +55,8 @@ class PersonDevice(BaseModel): material_name: str # 产品名称 spec_model: str # 规格型号 task_status: str # 该设备名下的状态 WIP/PENDING + duration_hours: float # 滞留时长（在当前人手上多久，小时） + received_at: str | None # 接收时间（北京时间 MM-DD HH:mm） class PersonWorkload(BaseModel): @@ -64,6 +66,23 @@ class PersonWorkload(BaseModel): devices: list[PersonDevice] +class PersonHistoryRecord(BaseModel): + task_id: str + task_name: str # 工序/任务名 + assignee_id: str # 负责人ID + assignee_name: str # 负责人姓名 + product_sn: str # 16位身份证 + external_serial: str | None # 业务序列号 + material_name: str # 产品名称 + spec_model: str # 规格型号 + status: str # 状态 WIP/PENDING/COMPLETED + received_at: str | None # 开始时间 MM-DD HH:mm + completed_at: str | None # 结束时间 MM-DD HH:mm（进行中为 None） + duration_hours: float # 总耗时（小时） + latest_valid_remark: str | None = None # 最新有效备注（排除系统转交类） + record_count: int = 0 # 记录总数（含系统备注） + + class ProductMessageItem(BaseModel): id: str content: str @@ -269,16 +288,17 @@ async def get_completed_tasks( # ============================================================ async def get_people_workload(db: AsyncSession) -> list[PersonWorkload]: - """上帝视角 — 按负责人聚合当前在制品设备（WIP/PENDING 任务，product 去重）。""" + """上帝视角 — 按负责人聚合当前在制品设备（WIP/PENDING 任务，product 去重，含滞留时长）。""" from app.models.task import Task, TASK_STATUS_PENDING, TASK_STATUS_WIP from app.models.product import Product + from app.core.time_utils import get_beijing_time, BEIJING_TZ stmt = ( select( Task.assignee_id, Product.id, Product.serial_number, Product.external_serial, Product.material_name, Product.spec_model, - Task.status, + Task.status, Task.received_at, Task.created_at, ) .join(Product, Task.product_id == Product.id) .where( @@ -290,46 +310,239 @@ async def get_people_workload(db: AsyncSession) -> list[PersonWorkload]: result = await db.execute(stmt) rows = result.all() - # 按 assignee 聚合，product 去重；状态优先级 WIP > PENDING - by_assignee: dict[str, dict[str, PersonDevice]] = {} + def _to_bj(dt): + if not dt: + return None + if dt.tzinfo is None: + from datetime import timezone as dt_timezone + dt = dt.replace(tzinfo=dt_timezone.utc).astimezone(BEIJING_TZ) + else: + dt = dt.astimezone(BEIJING_TZ) + return dt + + now = get_beijing_time() + + # 中间聚合：by_assignee[assignee][product_id] -> 设备快照 + 最早接手时间 + agg: dict[str, dict[str, dict]] = {} for row in rows: assignee = row[0] product_id = str(row[1]) status = row[6] or "" - devices = by_assignee.setdefault(assignee, {}) - if product_id in devices: - # 已有该设备：若新状态为 WIP 则提升（更活跃） - if status == "WIP": - devices[product_id].task_status = "WIP" - continue - devices[product_id] = PersonDevice( - product_id=product_id, - serial_number=row[2] or "", - external_serial=row[3] or None, - material_name=row[4] or "", - spec_model=row[5] or "", - task_status=status, - ) + received_dt = _to_bj(row[7] or row[8]) + entry = agg.setdefault(assignee, {}).setdefault(product_id, { + "status": "", + "earliest_dt": None, + "serial": row[2] or "", + "ext": row[3] or None, + "mat": row[4] or "", + "spec": row[5] or "", + }) + # 状态优先级 WIP > PENDING + if status == "WIP": + entry["status"] = "WIP" + elif not entry["status"]: + entry["status"] = status + # 取最早接手时间 + if received_dt and (entry["earliest_dt"] is None or received_dt < entry["earliest_dt"]): + entry["earliest_dt"] = received_dt - raw_ids = list(by_assignee.keys()) + raw_ids = list(agg.keys()) name_map: dict[str, str] = {} if raw_ids: from app.services.mom_cache import get_display_names name_map = get_display_names(raw_ids) - workloads = [ - PersonWorkload( + workloads: list[PersonWorkload] = [] + for assignee, prods in agg.items(): + devices: list[PersonDevice] = [] + for product_id, e in prods.items(): + dt = e["earliest_dt"] + hours = round((now - dt).total_seconds() / 3600, 1) if dt else 0.0 + received_str = dt.strftime("%m-%d %H:%M") if dt else None + devices.append(PersonDevice( + product_id=product_id, + serial_number=e["serial"], + external_serial=e["ext"], + material_name=e["mat"], + spec_model=e["spec"], + task_status=e["status"], + duration_hours=hours, + received_at=received_str, + )) + workloads.append(PersonWorkload( assignee_id=assignee, assignee_name=name_map.get(assignee, assignee), device_count=len(devices), - devices=list(devices.values()), - ) - for assignee, devices in by_assignee.items() - ] + devices=devices, + )) workloads.sort(key=lambda w: w.device_count, reverse=True) return workloads +# ============================================================ +# 历史人员看板（按人聚合历史备注记录，按时段过滤） +# ============================================================ + +async def get_people_history( + db: AsyncSession, + since: datetime | None = None, + until: datetime | None = None, + assignee_id: str | None = None, + spec_model: str | None = None, + product_sn: str | None = None, + task_name: str | None = None, +) -> list[PersonHistoryRecord]: + """上帝视角 — 人员效能与工时台账（平铺 Task 明细，含 WIP/PENDING/COMPLETED）。""" + from app.models.task import Task, TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED + from app.models.product import Product + from app.core.time_utils import get_beijing_time, BEIJING_TZ + from sqlalchemy import func + + now = get_beijing_time() + + # ── 关联 TaskRecord：最新有效备注 + 记录总数 ── + from app.models.task import TaskRecord + from sqlalchemy import and_, or_ + + # 系统自动备注关键字（转交/移交/撤回/派发/驳回等） + sys_remark = or_( + TaskRecord.remark.ilike("%转交%"), + TaskRecord.remark.ilike("%移交%"), + TaskRecord.remark.ilike("%撤回%"), + TaskRecord.remark.ilike("%重新接手%"), + TaskRecord.remark.ilike("%派发%"), + TaskRecord.remark.ilike("%分配%"), + TaskRecord.remark.ilike("%驳回%"), + TaskRecord.remark.ilike("%返工%"), + TaskRecord.remark.ilike("%完工%"), + ) + + # 最新有效备注（排除系统备注，按时间倒序取第一条） + valid_remark_subq = ( + select( + TaskRecord.task_id, + TaskRecord.remark.label("latest_valid_remark"), + func.row_number().over( + partition_by=TaskRecord.task_id, + order_by=TaskRecord.created_at.desc(), + ).label("rn"), + ) + .where( + TaskRecord.remark.isnot(None), + func.trim(TaskRecord.remark) != "", + ~sys_remark, + ) + ).subquery("vr") + + # 记录总数（含系统备注） + record_count_subq = ( + select( + TaskRecord.task_id, + func.count(TaskRecord.id).label("record_count"), + ) + .group_by(TaskRecord.task_id) + ).subquery("rc") + + stmt = ( + select( + Task.id, Task.task_name, Task.assignee_id, Task.status, + Task.received_at, Task.created_at, Task.completed_at, + Product.serial_number, Product.external_serial, + Product.material_name, Product.spec_model, + valid_remark_subq.c.latest_valid_remark, + func.coalesce(record_count_subq.c.record_count, 0).label("record_count"), + ) + .join(Product, Task.product_id == Product.id) + .outerjoin( + valid_remark_subq, + and_(valid_remark_subq.c.task_id == Task.id, valid_remark_subq.c.rn == 1), + ) + .outerjoin(record_count_subq, record_count_subq.c.task_id == Task.id) + .where( + Task.status.in_([TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED]), + Task.assignee_id.isnot(None), + ) + ) + + # 多维筛选（模糊/精确匹配） + if assignee_id: + stmt = stmt.where(Task.assignee_id == assignee_id) + if spec_model: + stmt = stmt.where(Product.spec_model.ilike(f"%{spec_model.strip()}%")) + if product_sn: + stmt = stmt.where(Product.serial_number.ilike(f"%{product_sn.strip()}%")) + if task_name: + stmt = stmt.where(Task.task_name.ilike(f"%{task_name.strip()}%")) + + # 时间交集：start = COALESCE(received_at, created_at)，end = COALESCE(completed_at, now) + start_expr = func.coalesce(Task.received_at, Task.created_at) + end_expr = func.coalesce(Task.completed_at, now) + if until: + stmt = stmt.where(start_expr <= until) + if since: + stmt = stmt.where(end_expr >= since) + + # 排序：先进行中（WIP/PENDING），后已完成（COMPLETED），状态内按开始时间降序 + from sqlalchemy import case + status_prio = case( + (Task.status == TASK_STATUS_WIP, 0), + (Task.status == TASK_STATUS_PENDING, 1), + (Task.status == TASK_STATUS_COMPLETED, 2), + else_=3, + ) + stmt = stmt.order_by( + status_prio.asc(), + func.coalesce(Task.received_at, Task.created_at).desc(), + ) + result = await db.execute(stmt) + rows = result.all() + + def _to_bj(dt): + if not dt: + return None + if dt.tzinfo is None: + from datetime import timezone as dt_timezone + dt = dt.replace(tzinfo=dt_timezone.utc).astimezone(BEIJING_TZ) + else: + dt = dt.astimezone(BEIJING_TZ) + return dt + + raw_ids = list({r[2] for r in rows if r[2]}) + name_map: dict[str, str] = {} + if raw_ids: + from app.services.mom_cache import get_display_names + name_map = get_display_names(raw_ids) + + records: list[PersonHistoryRecord] = [] + for row in rows: + received_dt = _to_bj(row[4] or row[5]) # received_at or created_at + completed_dt = _to_bj(row[6]) # completed_at（进行中为 None） + if completed_dt: + end_dt = completed_dt + completed_str = completed_dt.strftime("%m-%d %H:%M") + else: + end_dt = now + completed_str = None + hours = round((end_dt - received_dt).total_seconds() / 3600, 1) if received_dt else 0.0 + records.append(PersonHistoryRecord( + task_id=str(row[0]), + task_name=row[1] or "", + assignee_id=row[2] or "", + assignee_name=name_map.get(row[2] or "", row[2] or ""), + product_sn=row[7] or "", + external_serial=row[8] or None, + material_name=row[9] or "", + spec_model=row[10] or "", + status=row[3] or "", + received_at=received_dt.strftime("%m-%d %H:%M") if received_dt else None, + completed_at=completed_str, + duration_hours=hours, + latest_valid_remark=row[11], + record_count=row[12] or 0, + )) + return records + + # ============================================================ # 协同留言搜索（上帝视角 — 全厂） # ============================================================