app/models.py

@@ -362,13 +362,6 @@ class ProductionReportAllocation(Base):
     item: Mapped[ProductionReportItem | None] = relationship(back_populates="allocations")
 
     __table_args__ = (
-        Index(
-            "uq_report_allocations_report_item_date",
-            "report_id",
-            "report_item_id",
-            "allocation_date",
-            unique=True,
-        ),
         Index("idx_report_allocations_report", "report_id"),
         Index("idx_report_allocations_item", "report_item_id"),
         Index("idx_report_allocations_date", "allocation_date"),

app/routers/reconciliation.py

@@ -1,21 +1,12 @@
 from datetime import date
 
 from fastapi import APIRouter, Depends, HTTPException, Query
-from sqlalchemy import and_, extract, func, select
+from sqlalchemy import extract, func, select
 from sqlalchemy.orm import Session
 
 from app.database import get_db
 from app.deps import require_roles
-from app.models import (
+from app.models import Personnel, Product, ProductionReport, ProductionReportItem, ReconciliationLedgerEntry, ReportStatus, Role
-    Personnel,
-    Product,
-    ProductionReport,
-    ProductionReportAllocation,
-    ProductionReportItem,
-    ReconciliationLedgerEntry,
-    ReportStatus,
-    Role,
-)
 from app.schemas import (
     ReconciliationEntryUpdate,
     ReconciliationLedgerOut,
@@ -67,20 +58,6 @@ def _month_range(year: int, month: int) -> tuple[date, date]:
     return start, end
 
 
-def _fold_reported_good_quantity_rows(
-    rows,
-    latest_processes: dict[tuple[str, str], set[str]],
-) -> dict[tuple[str, str, int], float]:
-    totals: dict[tuple[str, str, int], float] = {}
-    for point_name, product_name, process_name, month, good_qty in rows:
-        product_key = (point_name, product_name)
-        if str(process_name or "").strip() not in latest_processes.get(product_key, set()):
-            continue
-        key = (point_name, product_name, int(month))
-        totals[key] = round2(totals.get(key, 0) + as_float(good_qty))
-    return totals
-
-
 def _reported_good_quantities(
     db: Session,
     *,
@@ -97,22 +74,14 @@ def _reported_good_quantities(
             ProductionReportItem.attendance_point_name,
             ProductionReportItem.product_name,
             ProductionReportItem.process_name,
-            extract("month", ProductionReportAllocation.allocation_date).label("month"),
+            extract("month", ProductionReport.report_date).label("month"),
-            func.sum(ProductionReportAllocation.good_qty).label("good_qty"),
+            func.sum(ProductionReportItem.good_qty).label("good_qty"),
         )
-        .select_from(ProductionReportAllocation)
+        .join(ProductionReport, ProductionReport.id == ProductionReportItem.report_id)
-        .join(
-            ProductionReportItem,
-            and_(
-                ProductionReportItem.id == ProductionReportAllocation.report_item_id,
-                ProductionReportItem.report_id == ProductionReportAllocation.report_id,
-            ),
-        )
-        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
         .where(
             ProductionReport.status == ReportStatus.approved,
             ProductionReport.is_voided.is_(False),
-            extract("year", ProductionReportAllocation.allocation_date) == year,
+            extract("year", ProductionReport.report_date) == year,
             ProductionReportItem.attendance_point_name.in_(point_names),
             ProductionReportItem.product_name.in_(product_names),
         )
@@ -120,10 +89,17 @@ def _reported_good_quantities(
             ProductionReportItem.attendance_point_name,
             ProductionReportItem.product_name,
             ProductionReportItem.process_name,
-            extract("month", ProductionReportAllocation.allocation_date),
+            extract("month", ProductionReport.report_date),
         )
     ).all()
-    return _fold_reported_good_quantity_rows(rows, latest_processes)
+    totals: dict[tuple[str, str, int], float] = {}
+    for point_name, product_name, process_name, month, good_qty in rows:
+        product_key = (point_name, product_name)
+        if str(process_name or "").strip() not in latest_processes.get(product_key, set()):
+            continue
+        key = (point_name, product_name, int(month))
+        totals[key] = round2(totals.get(key, 0) + as_float(good_qty))
+    return totals
 
 
 def _ledger_quantities(db: Session, *, year: int) -> dict[tuple[str, str, int], dict[str, float]]:
@@ -149,9 +125,7 @@ def list_reconciliation_years(
     current_year = today().year
     years = {current_year}
     report_years = db.execute(
-        select(extract("year", ProductionReportAllocation.allocation_date))
+        select(extract("year", ProductionReport.report_date))
-        .select_from(ProductionReportAllocation)
-        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
         .where(
             ProductionReport.status == ReportStatus.approved,
             ProductionReport.is_voided.is_(False),

app/routers/usage_stats.py

@@ -2,20 +2,12 @@ from datetime import date
 from math import ceil
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
-from sqlalchemy import and_, select
+from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
 
 from app.database import get_db
 from app.deps import require_roles
-from app.models import (
+from app.models import Equipment, Personnel, ProductionReport, ReportStatus, Role
-    Equipment,
-    Personnel,
-    ProductionReport,
-    ProductionReportAllocation,
-    ProductionReportItem,
-    ReportStatus,
-    Role,
-)
 from app.schemas import (
     PageResponse,
     UsageStatsDailyRow,
@@ -26,7 +18,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_allocations
+from app.services.usage_stats import UsageStatRow, build_usage_stats
 from app.services.usage_stats_export import export_usage_stats_rows
 
 router = APIRouter(prefix="/api/usage-stats", tags=["usage-stats"])
@@ -55,42 +47,21 @@ 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_allocations_query(
+def _approved_reports_query(point_names: list[str], start_date: date | None, end_date: date | None):
-    point_names: list[str],
-    start_date: date | None,
-    end_date: date | None,
-):
     query = (
-        select(ProductionReportAllocation)
+        select(ProductionReport)
-        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
+        .options(selectinload(ProductionReport.employee), selectinload(ProductionReport.items))
-        .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),
-        )
         .where(
             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:
-        query = query.where(ProductionReportAllocation.allocation_date >= start_date)
+        query = query.where(ProductionReport.report_date >= start_date)
     if end_date:
-        query = query.where(ProductionReportAllocation.allocation_date <= end_date)
+        query = query.where(ProductionReport.report_date <= end_date)
-    return query.order_by(
+    return query.order_by(ProductionReport.report_date.desc(), ProductionReport.id.desc())
-        ProductionReportAllocation.allocation_date.desc(),
-        ProductionReportAllocation.report_id.desc(),
-        ProductionReportAllocation.id.desc(),
-    )
 
 
 def _sort_usage_rows(rows: list[UsageStatRow], sort_by: str) -> list[UsageStatRow]:
@@ -203,93 +174,6 @@ 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)$"),
@@ -305,11 +189,11 @@ def usage_stats_summary(
 ) -> PageResponse:
     purge_expired_voided_reports(db)
     point_names = _requested_points(db, user, attendance_point_name)
-    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
+    reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
     rows = _sort_usage_rows(
         _filter_usage_rows(
-            build_usage_stats_from_allocations(
+            build_usage_stats(
-                allocations=allocations,
+                reports=reports,
                 category=category,
                 equipment_type_by_key=_equipment_type_map(db, point_names),
             ),
@@ -343,11 +227,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)
-    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
+    reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
     device_rows = _sort_usage_rows(
         _filter_usage_rows(
-            build_usage_stats_from_allocations(
+            build_usage_stats(
-                allocations=allocations,
+                reports=reports,
                 category="device",
                 equipment_type_by_key=equipment_type_by_key,
             ),
@@ -357,8 +241,8 @@ def export_usage_stats_excel(
     )
     mold_rows = _sort_usage_rows(
         _filter_usage_rows(
-            build_usage_stats_from_allocations(
+            build_usage_stats(
-                allocations=allocations,
+                reports=reports,
                 category="mold",
                 equipment_type_by_key=equipment_type_by_key,
             ),
@@ -387,7 +271,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)
-    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
+    reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
     target_args = {
         "id": _clean(id),
         "category": category,
@@ -398,9 +282,53 @@ def usage_stats_detail(
         "stamping_method": _clean(stamping_method),
         "metric_kind": _clean(metric_kind),
     }
-    return _build_usage_stats_detail_from_allocations(
+    rows = build_usage_stats(
-        allocations=allocations,
+        reports=reports,
         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),
+    )

app/services/report_allocations.py

@@ -1,13 +1,11 @@
 from collections import defaultdict
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from datetime import date, datetime, timedelta
-from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
-from typing import Any
 
-from sqlalchemy import delete, select
+from sqlalchemy import delete
 from sqlalchemy.orm import Session
 
-from app.models import ProductionReport, ProductionReportAllocation
+from app.models import ProductionReportAllocation
 from app.services.common import as_float, round2
 from app.services.metrics import (
     SHIFT_BUCKETS,
@@ -26,8 +24,6 @@ from app.services.metrics import (
 )
 from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig
 
-SPLIT_QUANT = Decimal("0.01")
-
 
 @dataclass(frozen=True)
 class ReportAllocationDraft:
@@ -130,34 +126,8 @@ def _report_devices(report) -> list | None:
     return list(devices) if devices is not None else None
 
 
-def _decimal_value(value: Any) -> Decimal:
+def _item_reference_wage(item) -> float:
-    if value is None:
+    return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0))
-        return Decimal("0")
-    if isinstance(value, Decimal):
-        return value
-    try:
-        return Decimal(str(value))
-    except (InvalidOperation, TypeError, ValueError):
-        return Decimal("0")
-
-
-def _round_split_value(value: Any) -> float:
-    return float(_decimal_value(value).quantize(SPLIT_QUANT, rounding=ROUND_HALF_UP))
-
-
-def _split_cents(value: Any) -> int:
-    rounded = _decimal_value(value).quantize(SPLIT_QUANT, rounding=ROUND_HALF_UP)
-    return int(rounded * 100)
-
-
-def _scaled_split_value(value: Any, ratio: float) -> float:
-    return _round_split_value(_decimal_value(value) * _decimal_value(ratio))
-
-
-def _item_reference_wage(item) -> Decimal:
-    return _decimal_value(getattr(item, "good_qty", 0)) * _decimal_value(
-        getattr(item, "process_unit_price_yuan", 0)
-    )
 
 
 def _allocation_targets(report, items: list, schedule: WorkScheduleConfig | None = None):
@@ -428,11 +398,11 @@ def _fallback_row(report, item) -> ReportAllocationDraft:
         attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""),
         employee_phone=str(getattr(report, "employee_phone", "") or ""),
         allocation_date=getattr(report, "report_date"),
-        good_qty=_round_split_value(getattr(item, "good_qty", 0)),
+        good_qty=round2(getattr(item, "good_qty", 0)),
-        defect_qty=_round_split_value(getattr(item, "defect_qty", 0)),
+        defect_qty=round2(getattr(item, "defect_qty", 0)),
-        scrap_qty=_round_split_value(getattr(item, "scrap_qty", 0)),
+        scrap_qty=round2(getattr(item, "scrap_qty", 0)),
-        changeover_count=_round_split_value(getattr(item, "changeover_count", 0)),
+        changeover_count=round2(getattr(item, "changeover_count", 0)),
-        reference_wage=_round_split_value(_item_reference_wage(item)),
+        reference_wage=round2(_item_reference_wage(item)),
     )
 
 
@@ -451,99 +421,14 @@ def _draft_from_minutes(report, item, allocation_date: date, minutes: dict[str,
         overtime_minutes=overtime_minutes,
         night_minutes=night_minutes,
         effective_minutes=round2(day_minutes + overtime_minutes + night_minutes),
-        good_qty=_scaled_split_value(getattr(item, "good_qty", 0), ratio),
+        good_qty=round2(as_float(getattr(item, "good_qty", 0)) * ratio),
-        defect_qty=_scaled_split_value(getattr(item, "defect_qty", 0), ratio),
+        defect_qty=round2(as_float(getattr(item, "defect_qty", 0)) * ratio),
-        scrap_qty=_scaled_split_value(getattr(item, "scrap_qty", 0), ratio),
+        scrap_qty=round2(as_float(getattr(item, "scrap_qty", 0)) * ratio),
-        changeover_count=_scaled_split_value(getattr(item, "changeover_count", 0), ratio),
+        changeover_count=round2(as_float(getattr(item, "changeover_count", 0)) * ratio),
-        reference_wage=_round_split_value(reference_wage * _decimal_value(ratio)),
+        reference_wage=round2(reference_wage * ratio),
     )
 
 
-SPLIT_RESIDUAL_FIELDS = (
-    "good_qty",
-    "defect_qty",
-    "scrap_qty",
-    "changeover_count",
-    "reference_wage",
-)
-
-
-def _item_split_targets(item) -> dict[str, float]:
-    return {
-        "good_qty": _round_split_value(getattr(item, "good_qty", 0)),
-        "defect_qty": _round_split_value(getattr(item, "defect_qty", 0)),
-        "scrap_qty": _round_split_value(getattr(item, "scrap_qty", 0)),
-        "changeover_count": _round_split_value(getattr(item, "changeover_count", 0)),
-        "reference_wage": _round_split_value(_item_reference_wage(item)),
-    }
-
-
-def _balance_item_split_residuals(
-    drafts: list[ReportAllocationDraft],
-    targets: dict[str, float],
-) -> list[ReportAllocationDraft]:
-    if not drafts:
-        return drafts
-
-    mutable_values = [
-        {field: _split_cents(getattr(draft, field)) for field in SPLIT_RESIDUAL_FIELDS}
-        for draft in drafts
-    ]
-
-    for field in SPLIT_RESIDUAL_FIELDS:
-        target_cents = _split_cents(targets[field])
-        current_cents = sum(values[field] for values in mutable_values)
-        residual = target_cents - current_cents
-        if residual == 0:
-            continue
-
-        if residual > 0:
-            ordered_indexes = sorted(
-                range(len(drafts)),
-                key=lambda index: (
-                    -as_float(drafts[index].effective_minutes),
-                    drafts[index].allocation_date.isoformat(),
-                    index,
-                ),
-            )
-        else:
-            ordered_indexes = sorted(
-                range(len(drafts)),
-                key=lambda index: (
-                    -mutable_values[index][field],
-                    -as_float(drafts[index].effective_minutes),
-                    drafts[index].allocation_date.isoformat(),
-                    index,
-                ),
-            )
-
-        step = 1 if residual > 0 else -1
-        remaining = abs(residual)
-        while remaining > 0:
-            changed = False
-            for index in ordered_indexes:
-                if step < 0 and mutable_values[index][field] <= 0:
-                    continue
-                mutable_values[index][field] += step
-                remaining -= 1
-                changed = True
-                if remaining == 0:
-                    break
-            if not changed:
-                break
-
-    return [
-        replace(
-            draft,
-            **{
-                field: mutable_values[index][field] / 100
-                for field in SPLIT_RESIDUAL_FIELDS
-            },
-        )
-        for index, draft in enumerate(drafts)
-    ]
-
-
 def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None = None) -> list[ReportAllocationDraft]:
     items = list(getattr(report, "items", []) or [])
     if not items:
@@ -567,14 +452,12 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None =
         if total_effective <= 0:
             rows.append(_fallback_row(report, item))
             continue
-        item_rows: list[ReportAllocationDraft] = []
         for allocation_date in sorted(item_minutes):
             minutes = item_minutes[allocation_date]
             effective = sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
             if effective <= 0:
                 continue
-            item_rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective))
+            rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective))
-        rows.extend(_balance_item_split_residuals(item_rows, _item_split_targets(item)))
     return rows
 
 
@@ -606,14 +489,7 @@ def refresh_report_allocations(
     commit: bool = False,
 ) -> list[ProductionReportAllocation]:
     """Rebuild allocation rows with metrics semantics, updating item allocation-derived fields."""
-    report_id = getattr(report, "id")
+    db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == getattr(report, "id")))
-    if isinstance(report_id, int):
-        db.execute(
-            select(ProductionReport.id)
-            .where(ProductionReport.id == report_id)
-            .with_for_update()
-        ).scalar_one_or_none()
-    db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report_id))
     rows = [
         ProductionReportAllocation(
             report_id=draft.report_id,

app/services/usage_stats.py

@@ -1,7 +1,6 @@
 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
@@ -34,7 +33,6 @@ 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 = ""
@@ -60,7 +58,6 @@ 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:
@@ -89,10 +86,6 @@ class UsageStatAccumulator:
             row.object_type = object_type
 
         row.value += as_float(value)
-        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:
@@ -126,42 +119,12 @@ 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, report_id in _iter_report_items(reports):
+    for item in _iter_report_items(reports):
         if is_misc_item(item):
             continue
 
@@ -181,7 +144,6 @@ def _build_device_usage_stats(
                     value=value,
                     tags=[TAG_CLEANING],
                     attendance_point_name=point_name,
-                    report_id=report_id,
                 )
             continue
 
@@ -195,14 +157,13 @@ 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, report_id in _iter_report_items(reports):
+    for item in _iter_report_items(reports):
         if is_misc_item(item):
             continue
 
@@ -226,17 +187,13 @@ 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:
-        report_id = getattr(report, "id", None)
+        yield from getattr(report, "items", []) or []
-        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]:

scripts/migrate_report_allocations.py

@@ -2,16 +2,12 @@ from pathlib import Path
 import re
 import sys
 
-from sqlalchemy import select, text
+from sqlalchemy import text
-from sqlalchemy.orm import selectinload
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT))
 
-from app.database import SessionLocal, engine  # noqa: E402
+from app.database import engine  # noqa: E402
-from app.models import AttendancePoint, ProductionReport, WorkSession  # noqa: E402
-from app.services.report_allocations import refresh_report_allocations  # noqa: E402
-from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig  # noqa: E402
 
 
 TABLE_NAME = "production_report_allocations"
@@ -49,21 +45,11 @@ REQUIRED_COLUMNS = {
     },
 }
 REQUIRED_INDEXES = {
-    "uq_report_allocations_report_item_date": {
+    "idx_report_allocations_report": ("report_id",),
-        "columns": ("report_id", "report_item_id", "allocation_date"),
+    "idx_report_allocations_item": ("report_item_id",),
-        "non_unique": 0,
+    "idx_report_allocations_date": ("allocation_date",),
-    },
+    "idx_report_allocations_point_date": ("attendance_point_name", "allocation_date"),
-    "idx_report_allocations_report": {"columns": ("report_id",), "non_unique": 1},
+    "idx_report_allocations_employee_date": ("employee_phone", "allocation_date"),
-    "idx_report_allocations_item": {"columns": ("report_item_id",), "non_unique": 1},
-    "idx_report_allocations_date": {"columns": ("allocation_date",), "non_unique": 1},
-    "idx_report_allocations_point_date": {
-        "columns": ("attendance_point_name", "allocation_date"),
-        "non_unique": 1,
-    },
-    "idx_report_allocations_employee_date": {
-        "columns": ("employee_phone", "allocation_date"),
-        "non_unique": 1,
-    },
 }
 REQUIRED_FOREIGN_KEYS = {
     "report_id": ("production_reports", "id", "CASCADE"),
@@ -163,51 +149,22 @@ def existing_foreign_keys(conn) -> dict[str, tuple[str, str, str]]:
     return {row[0]: (row[1], row[2], row[3]) for row in rows}
 
 
-def required_index_columns(index_name: str) -> tuple[str, ...]:
+def find_equivalent_index(indexes: dict[str, IndexMetadata], columns: tuple[str, ...]) -> str | None:
-    return REQUIRED_INDEXES[index_name]["columns"]
-
-
-def required_index_non_unique(index_name: str) -> int:
-    return int(REQUIRED_INDEXES[index_name]["non_unique"])
-
-
-def find_equivalent_index(
-    indexes: dict[str, IndexMetadata],
-    columns: tuple[str, ...],
-    non_unique: int,
-) -> str | None:
     for index_name, metadata in indexes.items():
-        if (
+        if index_name != "PRIMARY" and metadata["columns"] == columns and metadata["non_unique"] == 1:
-            index_name != "PRIMARY"
-            and metadata["columns"] == columns
-            and metadata["non_unique"] == non_unique
-        ):
             return index_name
     return None
 
 
 def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None:
-    if index_name not in REQUIRED_INDEXES or required_index_columns(index_name) != columns:
+    if index_name not in REQUIRED_INDEXES or REQUIRED_INDEXES[index_name] != columns:
         raise ValueError(f"unexpected index definition: {index_name} {columns}")
 
-    non_unique = required_index_non_unique(index_name)
     indexes = existing_indexes(conn)
-    if (
-        index_name in indexes
-        and indexes[index_name]["columns"] == columns
-        and indexes[index_name]["non_unique"] == non_unique
-    ):
-        return
     if index_name in indexes:
-        conn.execute(
+        return
-            text(
-                f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
-                f"DROP INDEX {quote_identifier(index_name)}"
-            )
-        )
-        indexes = existing_indexes(conn)
 
-    equivalent_index = find_equivalent_index(indexes, columns, non_unique)
+    equivalent_index = find_equivalent_index(indexes, columns)
     if equivalent_index:
         conn.execute(
             text(
@@ -218,51 +175,14 @@ def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None:
         return
 
     column_sql = ", ".join(quote_identifier(column) for column in columns)
-    unique_sql = "UNIQUE " if non_unique == 0 else ""
     conn.execute(
         text(
-            f"CREATE {unique_sql}INDEX {quote_identifier(index_name)} "
+            f"CREATE INDEX {quote_identifier(index_name)} "
             f"ON {quote_identifier(TABLE_NAME)} ({column_sql})"
         )
     )
 
 
-def cleanup_duplicate_allocation_keys(conn) -> None:
-    if not table_exists(conn):
-        return
-    conn.execute(
-        text(
-            """
-            DELETE target
-            FROM production_report_allocations target
-            JOIN (
-                SELECT id
-                FROM (
-                    SELECT allocation.id
-                    FROM production_report_allocations allocation
-                    JOIN (
-                        SELECT
-                            report_id,
-                            report_item_id,
-                            allocation_date,
-                            MIN(id) AS keep_id
-                        FROM production_report_allocations
-                        WHERE report_item_id IS NOT NULL
-                        GROUP BY report_id, report_item_id, allocation_date
-                        HAVING COUNT(*) > 1
-                    ) duplicate_key
-                      ON duplicate_key.report_id = allocation.report_id
-                     AND duplicate_key.report_item_id = allocation.report_item_id
-                     AND duplicate_key.allocation_date = allocation.allocation_date
-                    WHERE allocation.id <> duplicate_key.keep_id
-                ) duplicate_rows
-            ) doomed
-              ON doomed.id = target.id
-            """
-        )
-    )
-
-
 def cleanup_duplicate_fk_indexes(conn) -> None:
     indexes = existing_indexes(conn)
     duplicates = [
@@ -311,8 +231,6 @@ def ensure_schema(conn) -> None:
                 created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                 updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                 PRIMARY KEY (id),
-                UNIQUE KEY uq_report_allocations_report_item_date
-                    (report_id, report_item_id, allocation_date),
                 KEY idx_report_allocations_report (report_id),
                 KEY idx_report_allocations_item (report_item_id),
                 KEY idx_report_allocations_date (allocation_date),
@@ -329,9 +247,8 @@ def ensure_schema(conn) -> None:
         )
     )
 
-    cleanup_duplicate_allocation_keys(conn)
+    for index_name, columns in REQUIRED_INDEXES.items():
-    for index_name in REQUIRED_INDEXES:
+        create_index(conn, index_name, columns)
-        create_index(conn, index_name, required_index_columns(index_name))
     cleanup_duplicate_fk_indexes(conn)
 
 
@@ -398,21 +315,19 @@ def verify_schema(conn) -> None:
 
     wrong_indexes = [
         f"{index_name}({', '.join(indexes[index_name]['columns'])})"
-        for index_name in REQUIRED_INDEXES
+        for index_name, columns in REQUIRED_INDEXES.items()
-        if indexes.get(index_name, {}).get("columns") != required_index_columns(index_name)
+        if indexes.get(index_name, {}).get("columns") != columns
     ]
     if wrong_indexes:
         raise RuntimeError(f"{TABLE_NAME} incompatible indexes: {', '.join(wrong_indexes)}")
 
-    wrong_index_uniqueness = [
+    unique_required_indexes = [
         index_name
         for index_name in REQUIRED_INDEXES
-        if indexes[index_name]["non_unique"] != required_index_non_unique(index_name)
+        if indexes[index_name]["non_unique"] != 1
     ]
-    if wrong_index_uniqueness:
+    if unique_required_indexes:
-        raise RuntimeError(
+        raise RuntimeError(f"{TABLE_NAME} required indexes are unique: {', '.join(unique_required_indexes)}")
-            f"{TABLE_NAME} incompatible index uniqueness: {', '.join(wrong_index_uniqueness)}"
-        )
 
     duplicate_single_column_indexes = [
         index_name
@@ -459,72 +374,12 @@ def verify_schema(conn) -> None:
         raise RuntimeError(f"{TABLE_NAME} missing foreign keys: {', '.join(missing_foreign_keys)}")
 
 
-def _read_work_schedule_config(db, attendance_point_name: str | None) -> WorkScheduleConfig:
-    point_name = str(attendance_point_name or "").strip()
-    if not point_name:
-        return DEFAULT_WORK_SCHEDULE_CONFIG
-
-    point = db.get(AttendancePoint, point_name)
-    if point is None:
-        return DEFAULT_WORK_SCHEDULE_CONFIG
-
-    return WorkScheduleConfig(
-        day_start=point.day_start or DEFAULT_WORK_SCHEDULE_CONFIG.day_start,
-        day_end=point.day_end or DEFAULT_WORK_SCHEDULE_CONFIG.day_end,
-        lunch_start=point.lunch_start or DEFAULT_WORK_SCHEDULE_CONFIG.lunch_start,
-        lunch_end=point.lunch_end or DEFAULT_WORK_SCHEDULE_CONFIG.lunch_end,
-        dinner_start=point.dinner_start or DEFAULT_WORK_SCHEDULE_CONFIG.dinner_start,
-        dinner_end=point.dinner_end or DEFAULT_WORK_SCHEDULE_CONFIG.dinner_end,
-        overtime_start=point.overtime_start or DEFAULT_WORK_SCHEDULE_CONFIG.overtime_start,
-        overtime_end=point.overtime_end or DEFAULT_WORK_SCHEDULE_CONFIG.overtime_end,
-        night_start=point.night_start or DEFAULT_WORK_SCHEDULE_CONFIG.night_start,
-        night_end=point.night_end or DEFAULT_WORK_SCHEDULE_CONFIG.night_end,
-    )
-
-
-def _backfill(batch_size: int = 200) -> tuple[int, int]:
-    processed = 0
-    skipped = 0
-    last_id = 0
-
-    with SessionLocal() as db:
-        while True:
-            query = (
-                select(ProductionReport)
-                .where(ProductionReport.id > last_id)
-                .options(
-                    selectinload(ProductionReport.items),
-                    selectinload(ProductionReport.session).selectinload(WorkSession.devices),
-                )
-                .order_by(ProductionReport.id.asc())
-                .limit(batch_size)
-            )
-            reports = db.scalars(query).all()
-            if not reports:
-                break
-
-            for report in reports:
-                if report is None:
-                    skipped += 1
-                    continue
-
-                last_id = report.id
-                schedule = _read_work_schedule_config(db, report.attendance_point_name)
-                refresh_report_allocations(db, report, schedule=schedule, commit=False)
-                processed += 1
-
-            db.commit()
-
-    return processed, skipped
-
-
 def main() -> None:
     with engine.begin() as conn:
         ensure_schema(conn)
         verify_schema(conn)
 
-    processed, skipped = _backfill()
+    print("report allocations schema migrated")
-    print(f"report allocations migrated, processed={processed}, skipped={skipped}")
 
 
 if __name__ == "__main__":

tests/test_dashboard_allocations.py

@@ -60,7 +60,7 @@ def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc():
             defect_qty=0,
             scrap_qty=0,
             changeover_count=1,
-            reference_wage=700,
+            reference_wage=350,
         ),
         SimpleNamespace(
             report=report,
@@ -74,7 +74,7 @@ def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc():
             defect_qty=0,
             scrap_qty=0,
             changeover_count=1,
-            reference_wage=600,
+            reference_wage=300,
         ),
     ]
 
@@ -94,7 +94,7 @@ def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc():
     assert current.total_defect_qty == 0
     assert current.total_output_qty == 300
     assert current.changeover_count == 1
-    assert current.reference_wage == 600
+    assert current.reference_wage == 300
 
     assert previous.effective_minutes == 350
     assert previous.shift_day_minutes == 0
@@ -104,7 +104,7 @@ def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc():
     assert previous.total_defect_qty == 0
     assert previous.total_output_qty == 350
     assert previous.changeover_count == 1
-    assert previous.reference_wage == 700
+    assert previous.reference_wage == 350
 
 
 def test_export_dashboard_rows_writes_allocation_and_source_dates_with_warning_columns():

tests/test_reconciliation_allocations.py

@@ -1,223 +0,0 @@
-from datetime import date, datetime
-
-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 reconciliation
-
-
-@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(
-    *,
-    id: int,
-    status: ReportStatus = ReportStatus.approved,
-    is_voided: bool = False,
-):
-    return ProductionReport(
-        id=id,
-        session_id=id,
-        attendance_point_name="嘉恒",
-        employee_phone="13800000000",
-        report_date=date(2026, 7, 31),
-        start_at=datetime(2026, 7, 31, 20, 0),
-        end_at=datetime(2026, 8, 1, 2, 0),
-        duration_minutes=360,
-        break_minutes=0,
-        effective_minutes=360,
-        total_good_qty=650,
-        total_output_qty=650,
-        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, 8, 1, 2, 1),
-    )
-
-
-def _item(
-    *,
-    id: int,
-    report_id: int,
-    process_name: str = "2序",
-):
-    return ProductionReportItem(
-        id=id,
-        report_id=report_id,
-        attendance_point_name="嘉恒",
-        device_no="A-01",
-        project_no="P1",
-        product_name="产品A",
-        process_name=process_name,
-        operator_count=1,
-        process_unit_price_yuan=1,
-        standard_beat=1,
-        standard_workload=0,
-        good_qty=650,
-        defect_qty=0,
-        scrap_qty=0,
-        allocated_minutes=360,
-    )
-
-
-def _allocation(
-    *,
-    id: int,
-    report_id: int,
-    report_item_id: int,
-    allocation_date: date,
-    good_qty: float,
-):
-    return ProductionReportAllocation(
-        id=id,
-        report_id=report_id,
-        report_item_id=report_item_id,
-        attendance_point_name="嘉恒",
-        employee_phone="13800000000",
-        allocation_date=allocation_date,
-        day_minutes=0,
-        overtime_minutes=0,
-        night_minutes=0,
-        effective_minutes=0,
-        good_qty=good_qty,
-        defect_qty=0,
-        scrap_qty=0,
-        changeover_count=0,
-        reference_wage=0,
-    )
-
-
-def test_fold_reported_good_quantity_rows_uses_allocation_month_for_latest_process_only():
-    rows = [
-        ("嘉恒", "产品A", "1序", 7, 999),
-        ("嘉恒", "产品A", "2序", 7, 350),
-        ("嘉恒", "产品A", "2序", 8, 300),
-    ]
-    latest_processes = {("嘉恒", "产品A"): {"2序"}}
-
-    totals = reconciliation._fold_reported_good_quantity_rows(rows, latest_processes)
-
-    assert totals[("嘉恒", "产品A", 7)] == 350
-    assert totals[("嘉恒", "产品A", 8)] == 300
-
-
-def test_reported_good_quantities_groups_by_allocation_month_for_approved_unvoided_latest_process():
-    db = _sqlite_db()
-    try:
-        approved = _report(id=1)
-        pending = _report(id=2, status=ReportStatus.pending)
-        voided = _report(id=3, is_voided=True)
-        latest_item = _item(id=1, report_id=approved.id, process_name="2序")
-        earlier_item = _item(id=2, report_id=approved.id, process_name="1序")
-        pending_item = _item(id=3, report_id=pending.id, process_name="2序")
-        voided_item = _item(id=4, report_id=voided.id, process_name="2序")
-        db.add_all([
-            approved,
-            pending,
-            voided,
-            latest_item,
-            earlier_item,
-            pending_item,
-            voided_item,
-            _allocation(
-                id=1,
-                report_id=approved.id,
-                report_item_id=latest_item.id,
-                allocation_date=date(2026, 7, 31),
-                good_qty=350,
-            ),
-            _allocation(
-                id=2,
-                report_id=approved.id,
-                report_item_id=latest_item.id,
-                allocation_date=date(2026, 8, 1),
-                good_qty=300,
-            ),
-            _allocation(
-                id=3,
-                report_id=approved.id,
-                report_item_id=earlier_item.id,
-                allocation_date=date(2026, 7, 31),
-                good_qty=999,
-            ),
-            _allocation(
-                id=4,
-                report_id=pending.id,
-                report_item_id=pending_item.id,
-                allocation_date=date(2026, 7, 31),
-                good_qty=111,
-            ),
-            _allocation(
-                id=5,
-                report_id=voided.id,
-                report_item_id=voided_item.id,
-                allocation_date=date(2026, 8, 1),
-                good_qty=222,
-            ),
-        ])
-        db.commit()
-
-        totals = reconciliation._reported_good_quantities(
-            db,
-            year=2026,
-            product_keys=[("嘉恒", "产品A")],
-            latest_processes={("嘉恒", "产品A"): {"2序"}},
-        )
-
-        assert totals[("嘉恒", "产品A", 7)] == 350
-        assert totals[("嘉恒", "产品A", 8)] == 300
-    finally:
-        db.close()
-
-
-def test_reported_good_quantities_ignores_mismatched_allocation_report_item_rows():
-    db = _sqlite_db()
-    try:
-        approved = _report(id=1)
-        other_report = _report(id=2, status=ReportStatus.pending)
-        other_item = _item(id=2, report_id=other_report.id, process_name="2序")
-        db.add_all([
-            approved,
-            other_report,
-            other_item,
-            _allocation(
-                id=1,
-                report_id=approved.id,
-                report_item_id=other_item.id,
-                allocation_date=date(2026, 7, 31),
-                good_qty=123,
-            ),
-        ])
-        db.commit()
-
-        totals = reconciliation._reported_good_quantities(
-            db,
-            year=2026,
-            product_keys=[("嘉恒", "产品A")],
-            latest_processes={("嘉恒", "产品A"): {"2序"}},
-        )
-
-        assert totals == {}
-    finally:
-        db.close()

tests/test_report_allocations.py

@@ -2,10 +2,8 @@ from datetime import date, datetime
 from decimal import Decimal
 from types import SimpleNamespace
 
-import pytest
+from sqlalchemy import BigInteger, create_engine, select
-from sqlalchemy import BigInteger, create_engine, inspect, select
 from sqlalchemy.ext.compiler import compiles
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import raiseload, selectinload, sessionmaker
 
 from app.database import Base
@@ -205,51 +203,6 @@ def test_short_shared_report_rounding_residual_keeps_aggregate_night_total():
     assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows)
 
 
-def test_item_split_fields_keep_original_totals_after_rounding_residual():
-    item = _item(
-        good_qty=1,
-        defect_qty=1,
-        scrap_qty=1,
-        changeover_count=1,
-        process_unit_price_yuan=1,
-    )
-    report = _report(datetime(2026, 7, 25, 6, 0), datetime(2026, 7, 28, 6, 0), item)
-
-    rows = build_report_allocation_drafts(report)
-
-    assert [row.allocation_date for row in rows] == [
-        date(2026, 7, 25),
-        date(2026, 7, 26),
-        date(2026, 7, 27),
-    ]
-    assert [row.effective_minutes for row in rows] == [1340, 1340, 1340]
-    assert round(sum(row.good_qty for row in rows), 2) == 1
-    assert round(sum(row.defect_qty for row in rows), 2) == 1
-    assert round(sum(row.scrap_qty for row in rows), 2) == 1
-    assert round(sum(row.changeover_count for row in rows), 2) == 1
-    assert round(sum(row.reference_wage for row in rows), 2) == 1
-
-
-def test_reference_wage_uses_half_up_rounding_for_allocation_totals():
-    item = _item(good_qty=115, process_unit_price_yuan=Decimal("0.867"))
-    report = _report(datetime(2026, 7, 25, 8, 0), datetime(2026, 7, 25, 10, 0), item)
-
-    rows = build_report_allocation_drafts(report)
-
-    assert len(rows) == 1
-    assert rows[0].reference_wage == 99.71
-
-
-def test_split_reference_wage_residual_keeps_half_up_item_total():
-    item = _item(good_qty=115, process_unit_price_yuan=Decimal("0.867"))
-    report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item)
-
-    rows = build_report_allocation_drafts(report)
-
-    assert [row.allocation_date for row in rows] == [date(2026, 7, 24), date(2026, 7, 25)]
-    assert round(sum(row.reference_wage for row in rows), 2) == 99.71
-
-
 def test_items_sharing_same_started_at_split_the_segment_minutes():
     started_at = datetime(2026, 7, 25, 8, 0)
     first_item = _item(id=1, good_qty=10, started_at=started_at)
@@ -707,51 +660,6 @@ def _sqlite_db():
     return SessionLocal()
 
 
-def test_report_allocation_schema_has_unique_report_item_date_key():
-    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
-    Base.metadata.create_all(engine)
-    inspector = inspect(engine)
-
-    unique_column_sets = [
-        tuple(constraint["column_names"])
-        for constraint in inspector.get_unique_constraints("production_report_allocations")
-    ]
-    unique_column_sets.extend(
-        tuple(index["column_names"])
-        for index in inspector.get_indexes("production_report_allocations")
-        if index.get("unique")
-    )
-
-    assert ("report_id", "report_item_id", "allocation_date") in unique_column_sets
-
-
-def test_report_allocation_rejects_duplicate_report_item_date_key():
-    db = _sqlite_db()
-    try:
-        first = ProductionReportAllocation(
-            report_id=1,
-            report_item_id=1,
-            attendance_point_name="总厂",
-            employee_phone="13800000000",
-            allocation_date=date(2026, 7, 25),
-        )
-        duplicate = ProductionReportAllocation(
-            report_id=1,
-            report_item_id=1,
-            attendance_point_name="总厂",
-            employee_phone="13800000000",
-            allocation_date=date(2026, 7, 25),
-        )
-        db.add(first)
-        db.flush()
-        db.add(duplicate)
-
-        with pytest.raises(IntegrityError):
-            db.flush()
-    finally:
-        db.close()
-
-
 def test_report_out_skips_unloaded_allocations_without_lazy_load():
     db = _sqlite_db()
     try:

tests/test_usage_stats.py

@@ -1,21 +1,9 @@
-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,
     _filter_usage_rows,
@@ -25,24 +13,11 @@ 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
 
 
-@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))
 
@@ -62,93 +37,6 @@ 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) == []
@@ -426,31 +314,6 @@ 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(
@@ -783,121 +646,3 @@ 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)]
-
-
-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_report(id=5),
-                _db_item(id=105, report_id=5),
-                _db_allocation(
-                    id=1005,
-                    report_id=5,
-                    report_item_id=105,
-                    attendance_point_name="二厂",
-                ),
-                _db_allocation(id=1006, report_id=1, report_item_id=102),
-                _db_allocation(
-                    id=1007,
-                    report_id=1,
-                    report_item_id=101,
-                    allocation_date=date(2026, 7, 26),
-                    effective_minutes=999,
-                    good_qty=999,
-                ),
-            ]
-        )
-        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()