fix: 加强使用统计归属查询一致性

app/routers/usage_stats.py

@@ -2,7 +2,7 @@ from datetime import date
 from math import ceil
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
-from sqlalchemy import select
+from sqlalchemy import and_, select
 from sqlalchemy.orm import Session, selectinload
 
 from app.database import get_db
@@ -12,6 +12,7 @@ from app.models import (
     Personnel,
     ProductionReport,
     ProductionReportAllocation,
+    ProductionReportItem,
     ReportStatus,
     Role,
 )
@@ -61,8 +62,14 @@ def _approved_allocations_query(
 ):
     query = (
         select(ProductionReportAllocation)
-        .join(ProductionReportAllocation.report)
+        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
-        .join(ProductionReportAllocation.item)
+        .join(
+            ProductionReportItem,
+            and_(
+                ProductionReportItem.id == ProductionReportAllocation.report_item_id,
+                ProductionReportItem.report_id == ProductionReportAllocation.report_id,
+            ),
+        )
         .options(
             selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee),
             selectinload(ProductionReportAllocation.item),
@@ -71,6 +78,8 @@ def _approved_allocations_query(
             ProductionReport.status == ReportStatus.approved,
             ProductionReport.is_voided.is_(False),
             ProductionReport.attendance_point_name.in_(point_names),
+            ProductionReportAllocation.attendance_point_name.in_(point_names),
+            ProductionReportItem.attendance_point_name.in_(point_names),
         )
     )
     if start_date:

tests/test_usage_stats.py

@@ -1,10 +1,20 @@
-from datetime import date
+from datetime import date, datetime
 from io import BytesIO
 from types import SimpleNamespace
 import json
 
 from openpyxl import load_workbook
+from sqlalchemy import BigInteger, create_engine
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.orm import sessionmaker
 
+from app.database import Base
+from app.models import (
+    ProductionReport,
+    ProductionReportAllocation,
+    ProductionReportItem,
+    ReportStatus,
+)
 from app.routers import usage_stats as usage_stats_router
 from app.routers.usage_stats import (
     _build_usage_stats_export_response,
@@ -21,6 +31,18 @@ from app.services.usage_stats import build_usage_stats, split_cleaning_device_no
 from app.services.usage_stats_export import export_usage_stats_rows
 
 
+@compiles(BigInteger, "sqlite")
+def _compile_big_integer_for_sqlite(type_, compiler, **kw):
+    return "INTEGER"
+
+
+def _sqlite_db():
+    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine, future=True)
+    return SessionLocal()
+
+
 def _report(*items):
     return SimpleNamespace(items=list(items))
 
@@ -40,6 +62,93 @@ def _item(**overrides):
     return SimpleNamespace(**values)
 
 
+def _db_report(
+    *,
+    id: int,
+    attendance_point_name: str = "嘉恒",
+    status: ReportStatus = ReportStatus.approved,
+    is_voided: bool = False,
+) -> ProductionReport:
+    return ProductionReport(
+        id=id,
+        session_id=id,
+        attendance_point_name=attendance_point_name,
+        employee_phone="13800000000",
+        report_date=date(2026, 7, 24),
+        start_at=datetime(2026, 7, 24, 20, 0),
+        end_at=datetime(2026, 7, 25, 2, 0),
+        duration_minutes=360,
+        break_minutes=0,
+        effective_minutes=360,
+        total_good_qty=100,
+        total_output_qty=100,
+        actual_beat=1,
+        standard_beat=1,
+        expected_workload=0,
+        pace_rate=0,
+        workload_rate=0,
+        status=status,
+        is_voided=is_voided,
+        submitted_at=datetime(2026, 7, 25, 2, 1),
+    )
+
+
+def _db_item(
+    *,
+    id: int,
+    report_id: int,
+    attendance_point_name: str = "嘉恒",
+    device_no: str = "28#",
+) -> ProductionReportItem:
+    return ProductionReportItem(
+        id=id,
+        report_id=report_id,
+        attendance_point_name=attendance_point_name,
+        device_no=device_no,
+        project_no="P1",
+        product_name="产品A",
+        process_name="1",
+        stamping_method="普通",
+        operator_count=1,
+        process_unit_price_yuan=1,
+        standard_beat=1,
+        standard_workload=0,
+        good_qty=100,
+        defect_qty=0,
+        scrap_qty=0,
+        allocated_minutes=120,
+    )
+
+
+def _db_allocation(
+    *,
+    id: int,
+    report_id: int,
+    report_item_id: int,
+    attendance_point_name: str = "嘉恒",
+    allocation_date: date = date(2026, 7, 25),
+    effective_minutes: float = 120,
+    good_qty: float = 100,
+) -> ProductionReportAllocation:
+    return ProductionReportAllocation(
+        id=id,
+        report_id=report_id,
+        report_item_id=report_item_id,
+        attendance_point_name=attendance_point_name,
+        employee_phone="13800000000",
+        allocation_date=allocation_date,
+        day_minutes=effective_minutes,
+        overtime_minutes=0,
+        night_minutes=0,
+        effective_minutes=effective_minutes,
+        good_qty=good_qty,
+        defect_qty=0,
+        scrap_qty=0,
+        changeover_count=0,
+        reference_wage=0,
+    )
+
+
 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) == []
@@ -728,3 +837,57 @@ def test_usage_stats_detail_rows_use_allocation_date_and_dedupe_daily_report_cou
         (row.report_id, row.report_date, row.value, row.report_count)
         for row in detail.report_rows
     ] == [(7, date(2026, 7, 25), 650, 1)]
+
+
+def test_approved_allocations_query_requires_approved_unvoided_matching_point_and_item_report():
+    db = _sqlite_db()
+    try:
+        db.add_all(
+            [
+                _db_report(id=1),
+                _db_item(id=101, report_id=1),
+                _db_allocation(id=1001, report_id=1, report_item_id=101),
+                _db_report(id=2, status=ReportStatus.pending),
+                _db_item(id=102, report_id=2),
+                _db_allocation(id=1002, report_id=2, report_item_id=102),
+                _db_report(id=3, is_voided=True),
+                _db_item(id=103, report_id=3),
+                _db_allocation(id=1003, report_id=3, report_item_id=103),
+                _db_report(id=4, attendance_point_name="二厂"),
+                _db_item(id=104, report_id=4, attendance_point_name="二厂"),
+                _db_allocation(
+                    id=1004,
+                    report_id=4,
+                    report_item_id=104,
+                    attendance_point_name="二厂",
+                ),
+                _db_allocation(
+                    id=1005,
+                    report_id=1,
+                    report_item_id=101,
+                    attendance_point_name="二厂",
+                ),
+                _db_allocation(id=1006, report_id=1, report_item_id=102),
+            ]
+        )
+        db.commit()
+
+        allocations = db.scalars(
+            usage_stats_router._approved_allocations_query(
+                ["嘉恒"],
+                date(2026, 7, 25),
+                date(2026, 7, 25),
+            )
+        ).all()
+
+        assert [allocation.id for allocation in allocations] == [1001]
+        rows = usage_stats_service.build_usage_stats_from_allocations(
+            allocations=allocations,
+            category="device",
+            equipment_type_by_key={("嘉恒", "28#"): "冲压设备"},
+        )
+        assert len(rows) == 1
+        assert rows[0].value == 120
+        assert rows[0].report_count == 1
+    finally:
+        db.close()