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, _matches_target, _sort_usage_rows, _to_detail_out, _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 def _report(*items): return SimpleNamespace(items=list(items)) def _item(**overrides): values = { "attendance_point_name": "嘉恒", "product_name": "产品A", "process_name": "1", "stamping_method": "普通", "operator_count": 1, "device_no": "28#", "allocated_minutes": 0, "good_qty": 0, } values.update(overrides) return SimpleNamespace(**values) def test_split_cleaning_device_nos_supports_chinese_and_ascii_commas(): assert split_cleaning_device_nos("清洗机A、清洗机B, 清洗机C") == ["清洗机A", "清洗机B", "清洗机C"] assert split_cleaning_device_nos(None) == [] assert split_cleaning_device_nos("") == [] def test_split_cleaning_device_nos_deduplicates_in_order(): assert split_cleaning_device_nos("A、A、B") == ["A", "B"] def test_build_usage_stats_for_stamping_device_uses_allocated_minutes(): rows = build_usage_stats( reports=[_report(_item(device_no="28#", allocated_minutes=75, good_qty=12))], category="device", equipment_type_by_key={("嘉恒", "28#"): "冲压设备"}, ) assert len(rows) == 1 row = rows[0] assert row.name == "28#" assert row.metric_kind == "minutes" assert row.value == 75 assert row.report_count == 1 assert row.object_type == "冲压设备" assert row.id assert "device" in row.id assert "嘉恒" in row.id assert "28#" in row.id assert "minutes" in row.id def test_build_usage_stats_for_cleaning_device_splits_quantity_evenly(): rows = build_usage_stats( reports=[ _report( _item( device_no="清洗机A、清洗机B", stamping_method="清洗", good_qty=100, allocated_minutes=0, ) ) ], category="device", equipment_type_by_key={("嘉恒", "清洗机A"): "清洗设备", ("嘉恒", "清洗机B"): "清洗设备"}, ) assert [(row.name, row.metric_kind, row.value, row.report_count) for row in rows] == [ ("清洗机A", "quantity", 50, 1), ("清洗机B", "quantity", 50, 1), ] assert [row.object_type for row in rows] == ["清洗设备", "清洗设备"] def test_build_usage_stats_for_cleaning_device_deduplicates_before_splitting_quantity(): rows = build_usage_stats( reports=[ _report( _item( device_no="清洗机A、清洗机A、清洗机B", stamping_method="清洗", good_qty=100, ) ) ], category="device", equipment_type_by_key={}, ) assert [(row.name, row.value, row.report_count) for row in rows] == [ ("清洗机A", 50, 1), ("清洗机B", 50, 1), ] def test_build_usage_stats_for_device_skips_misc_items(): rows = build_usage_stats( reports=[ _report( _item( product_name="处理杂活", process_name="杂活", stamping_method="处理杂活", device_no="28#", allocated_minutes=30, ) ) ], category="device", equipment_type_by_key={("嘉恒", "28#"): "冲压设备"}, ) assert rows == [] def test_build_usage_stats_for_non_cleaning_device_ignores_cleaning_equipment_type(): rows = build_usage_stats( reports=[_report(_item(device_no="清洗机A", allocated_minutes=75, good_qty=12))], category="device", equipment_type_by_key={("嘉恒", "清洗机A"): "清洗设备"}, ) assert len(rows) == 1 assert rows[0].name == "清洗机A" assert rows[0].metric_kind == "minutes" assert rows[0].value == 75 assert rows[0].object_type == "清洗设备" def test_build_usage_stats_for_cleaning_device_marks_cleaning_tag(): rows = build_usage_stats( reports=[ _report( _item( device_no="清洗机A", stamping_method="清洗", good_qty=100, allocated_minutes=0, ) ) ], category="device", equipment_type_by_key={("嘉恒", "清洗机A"): "清洗设备"}, ) assert len(rows) == 1 assert "清洗" in rows[0].tags assert "清洗设备" not in rows[0].tags assert rows[0].object_type == "清洗设备" def test_build_usage_stats_for_mold_merges_by_point_product_process_and_stamping_method(): rows = build_usage_stats( reports=[ _report( _item(stamping_method="连续模", allocated_minutes=30), _item(stamping_method="连续模", allocated_minutes=45), ) ], category="mold", equipment_type_by_key={}, ) assert len(rows) == 1 row = rows[0] assert row.attendance_point_name == "嘉恒" assert row.product_name == "产品A" assert row.process_name == "1" assert row.stamping_method == "连续模" assert row.metric_kind == "minutes" assert row.value == 75 assert row.report_count == 2 assert row.object_type == "连续模" assert "连续模" in row.tags def test_build_usage_stats_for_mold_ids_distinguish_point_and_process(): rows = build_usage_stats( reports=[ _report( _item(attendance_point_name="嘉恒", process_name="1", allocated_minutes=30), _item(attendance_point_name="二厂", process_name="1", allocated_minutes=40), _item(attendance_point_name="嘉恒", process_name="2", allocated_minutes=50), ) ], category="mold", equipment_type_by_key={}, ) ids = {row.id for row in rows} assert len(rows) == 3 assert len(ids) == 3 def test_build_usage_stats_preserves_numeric_zero_identity_fields(): rows = build_usage_stats( reports=[_report(_item(product_name=0, process_name=0, allocated_minutes=10))], category="mold", equipment_type_by_key={}, ) assert len(rows) == 1 assert rows[0].product_name == "0" assert rows[0].process_name == "0" assert json.loads(rows[0].id)[3] == "0" assert json.loads(rows[0].id)[4] == "0" def test_build_usage_stats_uses_structured_ids_when_fields_contain_delimiter(): rows = build_usage_stats( reports=[ _report( _item(attendance_point_name="A||B", product_name="C", process_name="1", allocated_minutes=10), _item(attendance_point_name="A", product_name="B||C", process_name="1", allocated_minutes=20), ) ], category="mold", equipment_type_by_key={}, ) ids = {row.id for row in rows} assert len(ids) == 2 assert all(json.loads(row_id) for row_id in ids) assert {tuple(json.loads(row_id)[:5]) for row_id in ids} == { ("mold", "A||B", "A||B / C / 1 / 普通", "C", "1"), ("mold", "A", "A / B||C / 1 / 普通", "B||C", "1"), } def test_build_usage_stats_for_cleaning_mold_uses_good_quantity(): rows = build_usage_stats( reports=[_report(_item(stamping_method="清洗", allocated_minutes=80, good_qty=64))], category="mold", equipment_type_by_key={}, ) assert len(rows) == 1 assert rows[0].metric_kind == "quantity" assert rows[0].value == 64 assert rows[0].object_type == "清洗模具" assert "清洗" in rows[0].tags def test_build_usage_stats_for_mold_skips_misc_and_marks_multi_person(): rows = build_usage_stats( reports=[ _report( _item( product_name="处理杂活", process_name="杂活", stamping_method="处理杂活", allocated_minutes=30, ), _item(operator_count=2, allocated_minutes=40), ) ], category="mold", equipment_type_by_key={}, ) assert len(rows) == 1 assert rows[0].product_name == "产品A" assert rows[0].value == 40 assert rows[0].object_type == "多人工序" assert "多人协作" in rows[0].tags def test_build_usage_stats_for_non_cleaning_device_defaults_object_type_to_stamping_device(): rows = build_usage_stats( reports=[_report(_item(device_no="29#", allocated_minutes=25))], category="device", equipment_type_by_key={}, ) assert len(rows) == 1 assert rows[0].object_type == "冲压设备" def test_build_usage_stats_for_cleaning_device_defaults_object_type_to_cleaning_device(): rows = build_usage_stats( reports=[ _report( _item( device_no="清洗机Z", stamping_method="清洗", good_qty=10, ) ) ], category="device", equipment_type_by_key={}, ) assert len(rows) == 1 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( id="device-1", category="device", name="28#", metric_kind="minutes", object_type="冲压设备", value=360, report_count=2, attendance_point_name="嘉恒", ), UsageStatRow( id="mold-1", category="mold", name="旭升 / 产品B / 2 / 连续模", metric_kind="minutes", object_type="连续模", value=45, report_count=9, attendance_point_name="旭升", product_name="产品B", process_name="2", stamping_method="连续模", tags=["连续模", "多人协作"], ), ] assert _filter_usage_rows(rows, "") == rows assert [row.id for row in _filter_usage_rows(rows, "嘉恒")] == ["device-1"] assert [row.id for row in _filter_usage_rows(rows, "冲压")] == ["device-1"] assert [row.id for row in _filter_usage_rows(rows, "产品B")] == ["mold-1"] assert [row.id for row in _filter_usage_rows(rows, "多人协作")] == ["mold-1"] assert [row.id for row in _filter_usage_rows(rows, "360")] == ["device-1"] assert [row.id for row in _filter_usage_rows(rows, "9")] == ["mold-1"] def test_export_usage_stats_rows_writes_device_and_mold_sheets(): device_rows = [ UsageStatRow( id="device-minutes", category="device", name="28#", metric_kind="minutes", object_type="冲压设备", value=360, report_count=2, attendance_point_name="嘉恒", ), UsageStatRow( id="device-quantity", category="device", name="清洗机A", metric_kind="quantity", object_type="清洗设备", value=30, report_count=1, attendance_point_name="嘉恒", ), ] mold_rows = [ UsageStatRow( id="mold-minutes", category="mold", name="嘉恒 / 产品A / 1 / 普通", metric_kind="minutes", object_type="普通模具", value=45, report_count=3, attendance_point_name="嘉恒", product_name="产品A", process_name="1", stamping_method="普通", ), UsageStatRow( id="mold-quantity", category="mold", name="嘉恒 / 产品B / 2 / 清洗", metric_kind="quantity", object_type="清洗模具", value=18, report_count=1, attendance_point_name="嘉恒", product_name="产品B", process_name="2", stamping_method="清洗", ), ] workbook = load_workbook(BytesIO(export_usage_stats_rows(device_rows, mold_rows))) assert workbook.sheetnames == ["设备统计", "模具统计"] device_sheet = workbook["设备统计"] mold_sheet = workbook["模具统计"] assert [cell.value for cell in device_sheet[1]] == ["考勤点", "设备号", "设备类型", "使用时长(分钟)", "使用时长(小时)", "使用时长(天)", "清洗数量", "报工次数"] assert [cell.value for cell in mold_sheet[1]] == ["考勤点", "产品名称", "工序", "冲压方式", "使用时长(分钟)", "使用时长(小时)", "使用时长(天)", "清洗数量", "报工次数"] assert [cell.value for cell in device_sheet[2]] == ["嘉恒", "28#", "冲压设备", 360, 6, 0.25, 0, 2] assert [cell.value for cell in device_sheet[3]] == ["嘉恒", "清洗机A", "清洗设备", 0, 0, 0, 30, 1] assert [cell.value for cell in mold_sheet[2]] == ["嘉恒", "产品A", "1", "普通", 45, 0.75, 0.03, 0, 3] assert [cell.value for cell in mold_sheet[3]] == ["嘉恒", "产品B", "2", "清洗", 0, 0, 0, 18, 1] def test_build_usage_stats_export_response_returns_xlsx_attachment(): device_rows = [ UsageStatRow( id="device-minutes", category="device", name="28#", metric_kind="minutes", object_type="冲压设备", value=75, report_count=2, attendance_point_name="嘉恒", ) ] mold_rows = [ UsageStatRow( id="mold-minutes", category="mold", name="嘉恒 / 产品A / 1 / 普通", metric_kind="minutes", object_type="普通模具", value=45, report_count=3, attendance_point_name="嘉恒", product_name="产品A", process_name="1", stamping_method="普通", ) ] response = _build_usage_stats_export_response(device_rows, mold_rows) assert response.media_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" assert "usage-stats.xlsx" in response.headers["Content-Disposition"] workbook = load_workbook(BytesIO(response.body)) assert workbook.sheetnames == ["设备统计", "模具统计"] def test_sort_usage_rows_by_value_desc_then_report_count_desc_then_name(): rows = [ UsageStatRow(id="b", category="device", name="B", metric_kind="minutes", object_type="冲压设备", value=10, report_count=1), UsageStatRow(id="c", category="device", name="C", metric_kind="minutes", object_type="冲压设备", value=20, report_count=1), UsageStatRow(id="a", category="device", name="A", metric_kind="minutes", object_type="冲压设备", value=20, report_count=3), ] sorted_rows = _sort_usage_rows(rows, "value") assert [row.name for row in sorted_rows] == ["A", "C", "B"] def test_sort_usage_rows_by_count_desc_then_value_desc_then_name(): rows = [ UsageStatRow(id="b", category="device", name="B", metric_kind="minutes", object_type="冲压设备", value=20, report_count=2), UsageStatRow(id="c", category="device", name="C", metric_kind="minutes", object_type="冲压设备", value=30, report_count=1), UsageStatRow(id="a", category="device", name="A", metric_kind="minutes", object_type="冲压设备", value=30, report_count=2), ] sorted_rows = _sort_usage_rows(rows, "count") assert [row.name for row in sorted_rows] == ["A", "B", "C"] def test_to_schema_row_maps_usage_stat_fields(): row = UsageStatRow( id="row-1", category="mold", name="嘉恒 / 产品A / 1 / 普通", metric_kind="minutes", object_type="普通模具", value=12.34, report_count=5, tags=["多人协作"], attendance_point_name="嘉恒", product_name="产品A", process_name="1", stamping_method="普通", ) schema_row = _to_schema_row(row) assert isinstance(schema_row, UsageStatsRow) assert schema_row.model_dump() == { "id": "row-1", "category": "mold", "name": "嘉恒 / 产品A / 1 / 普通", "metric_kind": "minutes", "object_type": "普通模具", "value": 12.34, "report_count": 5, "tags": ["多人协作"], "attendance_point_name": "嘉恒", "product_name": "产品A", "process_name": "1", "stamping_method": "普通", } def test_to_detail_out_maps_target_identity_fields(): row = UsageStatRow( id="target-1", category="mold", name="嘉恒 / 产品A / 1 / 连续模", metric_kind="minutes", object_type="连续模", value=88, report_count=4, tags=["连续模"], attendance_point_name="嘉恒", product_name="产品A", process_name="1", stamping_method="连续模", ) detail = _to_detail_out(row) assert isinstance(detail, UsageStatsDetailOut) assert detail.model_dump() == { "id": "target-1", "category": "mold", "attendance_point_name": "嘉恒", "name": "嘉恒 / 产品A / 1 / 连续模", "object_type": "连续模", "metric_kind": "minutes", "value": 88, "report_count": 4, "product_name": "产品A", "process_name": "1", "stamping_method": "连续模", "tags": ["连续模"], "daily_rows": [], "report_rows": [], } def test_usage_stats_detail_out_identity_fields_have_empty_defaults(): detail = UsageStatsDetailOut(category="device") assert detail.id == "" assert detail.product_name == "" assert detail.process_name == "" assert detail.stamping_method == "" assert detail.tags == [] def test_matches_target_allows_accessible_point_when_attendance_point_name_is_empty(): row = UsageStatRow( id="row-1", category="device", name="28#", metric_kind="minutes", object_type="冲压设备", attendance_point_name="嘉恒", ) assert _matches_target( row, id="", category="device", attendance_point_name="", name="28#", product_name="", process_name="", stamping_method="", metric_kind="", ) def test_matches_target_uses_metric_kind_to_disambiguate_same_device_name(): minutes_row = UsageStatRow( id="minutes-id", category="device", name="清洗机A", metric_kind="minutes", object_type="清洗设备", attendance_point_name="嘉恒", ) quantity_row = UsageStatRow( id="quantity-id", category="device", name="清洗机A", metric_kind="quantity", object_type="清洗设备", attendance_point_name="嘉恒", ) target_args = { "id": "", "category": "device", "attendance_point_name": "", "name": "清洗机A", "product_name": "", "process_name": "", "stamping_method": "", "metric_kind": "quantity", } assert not _matches_target(minutes_row, **target_args) assert _matches_target(quantity_row, **target_args) def test_matches_target_uses_id_before_ambiguous_device_fields(): minutes_row = UsageStatRow( id="minutes-id", category="device", name="清洗机A", metric_kind="minutes", object_type="清洗设备", attendance_point_name="嘉恒", ) quantity_row = UsageStatRow( id="quantity-id", category="device", name="清洗机A", metric_kind="quantity", object_type="清洗设备", attendance_point_name="嘉恒", ) target_args = { "id": "minutes-id", "category": "device", "attendance_point_name": "", "name": "清洗机A", "product_name": "", "process_name": "", "stamping_method": "", "metric_kind": "quantity", } 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)]