From 37f2b2dcb22d46940efd1e8a1f7f26ea530f397b Mon Sep 17 00:00:00 2001 From: souplearn Date: Sat, 25 Jul 2026 05:28:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AE=BE=E5=A4=87=E6=A8=A1=E5=85=B7?= =?UTF-8?q?=E7=BB=9F=E8=AE=A1=E6=8C=89=E5=BD=92=E5=B1=9E=E6=97=A5=E6=9C=9F?= =?UTF-8?q?=E6=B1=87=E6=80=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/routers/usage_stats.py | 191 ++++++++++++++++++++++++------------ app/services/usage_stats.py | 51 +++++++++- tests/test_usage_stats.py | 82 ++++++++++++++++ 3 files changed, 256 insertions(+), 68 deletions(-) diff --git a/app/routers/usage_stats.py b/app/routers/usage_stats.py index 824ef14..69c9e18 100644 --- a/app/routers/usage_stats.py +++ b/app/routers/usage_stats.py @@ -7,7 +7,14 @@ from sqlalchemy.orm import Session, selectinload from app.database import get_db from app.deps import require_roles -from app.models import Equipment, Personnel, ProductionReport, ReportStatus, Role +from app.models import ( + Equipment, + Personnel, + ProductionReport, + ProductionReportAllocation, + ReportStatus, + Role, +) from app.schemas import ( PageResponse, UsageStatsDailyRow, @@ -18,7 +25,7 @@ from app.schemas import ( from app.services.attendance_points import accessible_point_names, require_attendance_point_access from app.services.common import round2 from app.services.report_lifecycle import purge_expired_voided_reports -from app.services.usage_stats import UsageStatRow, build_usage_stats +from app.services.usage_stats import UsageStatRow, build_usage_stats_from_allocations from app.services.usage_stats_export import export_usage_stats_rows router = APIRouter(prefix="/api/usage-stats", tags=["usage-stats"]) @@ -47,10 +54,19 @@ def _equipment_type_map(db: Session, point_names: list[str]) -> dict[tuple[str, return {(_clean(row.attendance_point_name), _clean(row.device_no)): row.device_type for row in rows} -def _approved_reports_query(point_names: list[str], start_date: date | None, end_date: date | None): +def _approved_allocations_query( + point_names: list[str], + start_date: date | None, + end_date: date | None, +): query = ( - select(ProductionReport) - .options(selectinload(ProductionReport.employee), selectinload(ProductionReport.items)) + select(ProductionReportAllocation) + .join(ProductionReportAllocation.report) + .join(ProductionReportAllocation.item) + .options( + selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee), + selectinload(ProductionReportAllocation.item), + ) .where( ProductionReport.status == ReportStatus.approved, ProductionReport.is_voided.is_(False), @@ -58,10 +74,14 @@ def _approved_reports_query(point_names: list[str], start_date: date | None, end ) ) if start_date: - query = query.where(ProductionReport.report_date >= start_date) + query = query.where(ProductionReportAllocation.allocation_date >= start_date) if end_date: - query = query.where(ProductionReport.report_date <= end_date) - return query.order_by(ProductionReport.report_date.desc(), ProductionReport.id.desc()) + query = query.where(ProductionReportAllocation.allocation_date <= end_date) + return query.order_by( + ProductionReportAllocation.allocation_date.desc(), + ProductionReportAllocation.report_id.desc(), + ProductionReportAllocation.id.desc(), + ) def _sort_usage_rows(rows: list[UsageStatRow], sort_by: str) -> list[UsageStatRow]: @@ -174,6 +194,93 @@ def _build_usage_stats_export_response(device_rows: list[UsageStatRow], mold_row ) +def _build_usage_stats_detail_from_allocations( + *, + allocations: list, + category: str, + target_args: dict, + equipment_type_by_key: dict[tuple[str, str], str], +) -> UsageStatsDetailOut: + rows = build_usage_stats_from_allocations( + allocations=allocations, + category=category, + equipment_type_by_key=equipment_type_by_key, + ) + target = next((row for row in rows if _matches_target(row, **target_args)), None) + if target is None: + return UsageStatsDetailOut( + id=target_args["id"], + category=category, + attendance_point_name=target_args["attendance_point_name"], + name=target_args["name"], + product_name=target_args["product_name"], + process_name=target_args["process_name"], + stamping_method=target_args["stamping_method"], + metric_kind=target_args["metric_kind"], + ) + + grouped_allocations: dict[tuple[date, int], list] = {} + group_reports: dict[tuple[date, int], object | None] = {} + for allocation in allocations: + allocation_date = getattr(allocation, "allocation_date") + report = getattr(allocation, "report", None) + report_id = getattr(allocation, "report_id", None) + if report_id is None and report is not None: + report_id = getattr(report, "id", None) + group_key = (allocation_date, report_id if report_id is not None else id(allocation)) + grouped_allocations.setdefault(group_key, []).append(allocation) + group_reports.setdefault(group_key, report) + + daily_rows: dict[date, UsageStatsDailyRow] = {} + report_rows: list[UsageStatsReportRow] = [] + for (allocation_date, report_key), group in grouped_allocations.items(): + group_matches = build_usage_stats_from_allocations( + allocations=group, + category=category, + equipment_type_by_key=equipment_type_by_key, + ) + match = next((row for row in group_matches if _matches_target(row, **target_args)), None) + if match is None: + continue + + day = daily_rows.setdefault( + allocation_date, + UsageStatsDailyRow(report_date=allocation_date), + ) + day.value = round2(day.value + match.value) + day.report_count += match.report_count + + report = group_reports[(allocation_date, report_key)] + employee = getattr(report, "employee", None) if report is not None else None + report_id = ( + getattr(report, "id", None) + if report is not None + else getattr(group[0], "report_id", None) + ) + employee_phone = getattr(report, "employee_phone", "") if report is not None else "" + report_rows.append( + UsageStatsReportRow( + report_id=report_id or 0, + report_date=allocation_date, + employee_phone=employee_phone, + employee_name=employee.name if employee else "", + display_name=target.name, + value=match.value, + report_count=match.report_count, + ) + ) + + return _to_detail_out( + target, + daily_rows=sorted(daily_rows.values(), key=lambda row: row.report_date), + report_rows=sorted( + report_rows, + key=lambda row: (row.report_date, row.report_id), + reverse=True, + ), + ) + + @router.get("/summary", response_model=PageResponse) def usage_stats_summary( category: str = Query("device", pattern="^(device|mold)$"), @@ -189,11 +296,11 @@ def usage_stats_summary( ) -> PageResponse: purge_expired_voided_reports(db) point_names = _requested_points(db, user, attendance_point_name) - reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all() + allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() rows = _sort_usage_rows( _filter_usage_rows( - build_usage_stats( - reports=reports, + build_usage_stats_from_allocations( + allocations=allocations, category=category, equipment_type_by_key=_equipment_type_map(db, point_names), ), @@ -227,11 +334,11 @@ def export_usage_stats_excel( purge_expired_voided_reports(db) point_names = _requested_points(db, user, attendance_point_name) equipment_type_by_key = _equipment_type_map(db, point_names) - reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all() + allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() device_rows = _sort_usage_rows( _filter_usage_rows( - build_usage_stats( - reports=reports, + build_usage_stats_from_allocations( + allocations=allocations, category="device", equipment_type_by_key=equipment_type_by_key, ), @@ -241,8 +348,8 @@ def export_usage_stats_excel( ) mold_rows = _sort_usage_rows( _filter_usage_rows( - build_usage_stats( - reports=reports, + build_usage_stats_from_allocations( + allocations=allocations, category="mold", equipment_type_by_key=equipment_type_by_key, ), @@ -271,7 +378,7 @@ def usage_stats_detail( purge_expired_voided_reports(db) point_names = _requested_points(db, user, attendance_point_name) equipment_type_by_key = _equipment_type_map(db, point_names) - reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all() + allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() target_args = { "id": _clean(id), "category": category, @@ -282,53 +389,9 @@ def usage_stats_detail( "stamping_method": _clean(stamping_method), "metric_kind": _clean(metric_kind), } - rows = build_usage_stats( - reports=reports, + return _build_usage_stats_detail_from_allocations( + allocations=allocations, category=category, + target_args=target_args, equipment_type_by_key=equipment_type_by_key, ) - target = next((row for row in rows if _matches_target(row, **target_args)), None) - if target is None: - return UsageStatsDetailOut( - id=target_args["id"], - category=category, - attendance_point_name=target_args["attendance_point_name"], - name=target_args["name"], - product_name=target_args["product_name"], - process_name=target_args["process_name"], - stamping_method=target_args["stamping_method"], - metric_kind=target_args["metric_kind"], - ) - - daily_rows: dict[date, UsageStatsDailyRow] = {} - report_rows: list[UsageStatsReportRow] = [] - for report in reports: - report_matches = build_usage_stats( - reports=[report], - category=category, - equipment_type_by_key=equipment_type_by_key, - ) - match = next((row for row in report_matches if _matches_target(row, **target_args)), None) - if match is None: - continue - - day = daily_rows.setdefault(report.report_date, UsageStatsDailyRow(report_date=report.report_date)) - day.value = round2(day.value + match.value) - day.report_count += match.report_count - report_rows.append( - UsageStatsReportRow( - report_id=report.id, - report_date=report.report_date, - employee_phone=report.employee_phone, - employee_name=report.employee.name if report.employee else "", - display_name=target.name, - value=match.value, - report_count=match.report_count, - ) - ) - - return _to_detail_out( - target, - daily_rows=sorted(daily_rows.values(), key=lambda row: row.report_date), - report_rows=sorted(report_rows, key=lambda row: (row.report_date, row.report_id), reverse=True), - ) diff --git a/app/services/usage_stats.py b/app/services/usage_stats.py index 9593389..9968751 100644 --- a/app/services/usage_stats.py +++ b/app/services/usage_stats.py @@ -1,6 +1,7 @@ import json import re from dataclasses import dataclass, field +from types import SimpleNamespace from typing import Any from app.services.cleaning import is_cleaning_item @@ -33,6 +34,7 @@ class UsageStatRow: object_type: str value: float = 0 report_count: int = 0 + report_ids: set[int] = field(default_factory=set, repr=False) tags: list[str] = field(default_factory=list) attendance_point_name: str = "" product_name: str = "" @@ -58,6 +60,7 @@ class UsageStatAccumulator: product_name: str = "", process_name: str = "", stamping_method: str = "", + report_id: int | None = None, ) -> None: row = self._rows.get(key) if row is None: @@ -86,7 +89,11 @@ class UsageStatAccumulator: row.object_type = object_type row.value += as_float(value) - row.report_count += 1 + if report_id is None: + row.report_count += 1 + elif report_id not in row.report_ids: + row.report_ids.add(report_id) + row.report_count += 1 for tag in tags or []: if tag and tag not in row.tags: row.tags.append(tag) @@ -119,12 +126,42 @@ def build_usage_stats( raise ValueError("category must be device or mold") +def build_usage_stats_from_allocations( + *, + allocations: list, + category: str, + equipment_type_by_key: dict[tuple[str, str], str], +) -> list[UsageStatRow]: + reports = [] + for allocation in allocations: + item = allocation.item + if item is None: + continue + proxy = SimpleNamespace( + attendance_point_name=item.attendance_point_name, + product_name=item.product_name, + process_name=item.process_name, + stamping_method=item.stamping_method, + operator_count=item.operator_count, + device_no=item.device_no, + allocated_minutes=allocation.effective_minutes, + good_qty=allocation.good_qty, + report_id=allocation.report_id, + ) + reports.append(SimpleNamespace(items=[proxy])) + return build_usage_stats( + reports=reports, + category=category, + equipment_type_by_key=equipment_type_by_key, + ) + + def _build_device_usage_stats( reports: list, equipment_type_by_key: dict[tuple[str, str], str], ) -> list[UsageStatRow]: acc = UsageStatAccumulator(category="device") - for item in _iter_report_items(reports): + for item, report_id in _iter_report_items(reports): if is_misc_item(item): continue @@ -144,6 +181,7 @@ def _build_device_usage_stats( value=value, tags=[TAG_CLEANING], attendance_point_name=point_name, + report_id=report_id, ) continue @@ -157,13 +195,14 @@ def _build_device_usage_stats( object_type=equipment_type_by_key.get((point_name, device_no), DEVICE_TYPE_STAMPING), value=as_float(getattr(item, "allocated_minutes", 0)), attendance_point_name=point_name, + report_id=report_id, ) return acc.rows() def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]: acc = UsageStatAccumulator(category="mold") - for item in _iter_report_items(reports): + for item, report_id in _iter_report_items(reports): if is_misc_item(item): continue @@ -187,13 +226,17 @@ def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]: product_name=product_name, process_name=process_name, stamping_method=stamping_method, + report_id=report_id, ) return acc.rows() def _iter_report_items(reports: list): for report in reports: - yield from getattr(report, "items", []) or [] + report_id = getattr(report, "id", None) + for item in getattr(report, "items", []) or []: + item_report_id = getattr(item, "report_id", None) + yield item, item_report_id if item_report_id is not None else report_id def _item_tags(item) -> list[str]: diff --git a/tests/test_usage_stats.py b/tests/test_usage_stats.py index 4086473..89e5faf 100644 --- a/tests/test_usage_stats.py +++ b/tests/test_usage_stats.py @@ -1,9 +1,11 @@ +from datetime import date from io import BytesIO from types import SimpleNamespace import json from openpyxl import load_workbook +from app.routers import usage_stats as usage_stats_router from app.routers.usage_stats import ( _build_usage_stats_export_response, _filter_usage_rows, @@ -13,6 +15,7 @@ from app.routers.usage_stats import ( _to_schema_row, ) from app.schemas import UsageStatsDetailOut, UsageStatsRow +from app.services import usage_stats as usage_stats_service from app.services.usage_stats import UsageStatRow from app.services.usage_stats import build_usage_stats, split_cleaning_device_nos from app.services.usage_stats_export import export_usage_stats_rows @@ -314,6 +317,31 @@ def test_build_usage_stats_for_cleaning_device_defaults_object_type_to_cleaning_ assert rows[0].object_type == "清洗设备" +def test_usage_stats_dedupes_report_count_for_split_allocations(): + allocation_a = SimpleNamespace( + report_id=1, + effective_minutes=350, + good_qty=350, + item=_item(device_no="28#", allocated_minutes=0, good_qty=0), + ) + allocation_b = SimpleNamespace( + report_id=1, + effective_minutes=300, + good_qty=300, + item=_item(device_no="28#", allocated_minutes=0, good_qty=0), + ) + + rows = usage_stats_service.build_usage_stats_from_allocations( + allocations=[allocation_a, allocation_b], + category="device", + equipment_type_by_key={("嘉恒", "28#"): "冲压设备"}, + ) + + assert len(rows) == 1 + assert rows[0].value == 650 + assert rows[0].report_count == 1 + + def test_filter_usage_rows_searches_all_visible_fields(): rows = [ UsageStatRow( @@ -646,3 +674,57 @@ def test_matches_target_uses_id_before_ambiguous_device_fields(): assert _matches_target(minutes_row, **target_args) assert not _matches_target(quantity_row, **target_args) + + +def test_usage_stats_detail_rows_use_allocation_date_and_dedupe_daily_report_count(): + report = SimpleNamespace( + id=7, + report_date=date(2026, 7, 24), + employee_phone="13800000000", + employee=SimpleNamespace(name="张三"), + ) + item = _item(device_no="28#", allocated_minutes=0, good_qty=0) + allocations = [ + SimpleNamespace( + report_id=7, + allocation_date=date(2026, 7, 25), + effective_minutes=350, + good_qty=350, + report=report, + item=item, + ), + SimpleNamespace( + report_id=7, + allocation_date=date(2026, 7, 25), + effective_minutes=300, + good_qty=300, + report=report, + item=item, + ), + ] + + detail = usage_stats_router._build_usage_stats_detail_from_allocations( + allocations=allocations, + category="device", + target_args={ + "id": "", + "category": "device", + "attendance_point_name": "", + "name": "28#", + "product_name": "", + "process_name": "", + "stamping_method": "", + "metric_kind": "minutes", + }, + equipment_type_by_key={("嘉恒", "28#"): "冲压设备"}, + ) + + assert detail.value == 650 + assert detail.report_count == 1 + assert [(row.report_date, row.value, row.report_count) for row in detail.daily_rows] == [ + (date(2026, 7, 25), 650, 1) + ] + assert [ + (row.report_id, row.report_date, row.value, row.report_count) + for row in detail.report_rows + ] == [(7, date(2026, 7, 25), 650, 1)]