test: 修复报工提交归属测试时间窗口

app/models.py

@@ -13,6 +13,7 @@ from sqlalchemy import (
     Numeric,
     String,
     Text,
+    text,
 )
 from sqlalchemy.dialects.mysql import JSON
 from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -249,6 +250,13 @@ class ProductionReport(Base):
     expected_workload: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
     pace_rate: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
     workload_rate: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
+    has_over_limit: Mapped[bool] = mapped_column(
+        Boolean,
+        nullable=False,
+        default=False,
+        server_default=text("0"),
+    )
+    over_limit_summary: Mapped[str | None] = mapped_column(String(500))
     status: Mapped[ReportStatus] = mapped_column(
         Enum(ReportStatus),
         nullable=False,
@@ -290,6 +298,7 @@ class ProductionReport(Base):
         Index("idx_reports_status_date", "status", "report_date"),
         Index("idx_reports_reviewer", "reviewer_phone", "reviewed_at"),
         Index("idx_reports_voided", "is_voided", "voided_at"),
+        Index("idx_reports_over_limit", "has_over_limit", "submitted_at"),
     )
 
 
@@ -315,6 +324,23 @@ class ProductionReportItem(Base):
     good_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
     defect_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
     scrap_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
+    over_limit_status: Mapped[str] = mapped_column(
+        String(32),
+        nullable=False,
+        default="not_checked",
+        server_default="not_checked",
+    )
+    over_limit_type: Mapped[str | None] = mapped_column(String(64))
+    over_limit_reason: Mapped[str | None] = mapped_column(String(500))
+    over_limit_checked_at: Mapped[datetime | None] = mapped_column(DateTime)
+    over_limit_check_source: Mapped[str | None] = mapped_column(String(32))
+    over_limit_batch_no: Mapped[str | None] = mapped_column(String(128))
+    over_limit_product_gross_weight_kg: Mapped[float | None] = mapped_column(Numeric(12, 4))
+    over_limit_issued_weight_kg: Mapped[float | None] = mapped_column(Numeric(18, 6))
+    over_limit_max_reportable_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
+    over_limit_previous_good_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
+    over_limit_current_cumulative_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
+    over_limit_current_report_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
     allocated_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
     started_at: Mapped[datetime | None] = mapped_column(DateTime)
     remark: Mapped[str | None] = mapped_column(String(500))
@@ -329,6 +355,15 @@ class ProductionReportItem(Base):
     __table_args__ = (
         Index("idx_report_items_report", "report_id"),
         Index("idx_report_items_product", "project_no", "product_name"),
+        Index("idx_report_items_over_limit_status", "over_limit_status"),
+        Index(
+            "idx_report_items_process_batch",
+            "attendance_point_name",
+            "project_no",
+            "product_name",
+            "raw_material_batch_no",
+            "process_name",
+        ),
     )
 
 
@@ -362,6 +397,13 @@ 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/dashboard.py

@@ -264,6 +264,11 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
                 "is_voided": False,
                 "review_remarks": [],
                 "review_remark_keys": set(),
+                "has_over_limit": False,
+                "over_limit_exceeded_keys": set(),
+                "over_limit_uncheckable_keys": set(),
+                "over_limit_reasons": [],
+                "over_limit_reason_keys": set(),
                 "correction_pairs": [],
                 "correction_pair_keys": set(),
             },
@@ -303,6 +308,17 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
         group["is_multi_person_assistant"] = (
             group["is_multi_person_assistant"] or bool(report.is_multi_person_assistant)
         )
+        over_limit_status = str(getattr(item, "over_limit_status", "") or "")
+        over_limit_reason = str(getattr(item, "over_limit_reason", "") or "").strip()
+        over_limit_item_key = getattr(item, "id", None) or id(item)
+        if over_limit_status == "exceeded":
+            group["has_over_limit"] = True
+            group["over_limit_exceeded_keys"].add(over_limit_item_key)
+        elif over_limit_status == "uncheckable":
+            group["over_limit_uncheckable_keys"].add(over_limit_item_key)
+        if over_limit_reason and over_limit_reason not in group["over_limit_reason_keys"]:
+            group["over_limit_reason_keys"].add(over_limit_reason)
+            group["over_limit_reasons"].append(over_limit_reason)
         if report.review_remark:
             review_remark = str(report.review_remark).strip()
             if review_remark and review_remark not in group["review_remark_keys"]:
@@ -333,6 +349,16 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
             "night": group["shift_night_minutes"],
         }
         source_report_dates = sorted(group["source_report_dates"])
+        exceeded_count = len(group["over_limit_exceeded_keys"])
+        uncheckable_count = len(group["over_limit_uncheckable_keys"])
+        if exceeded_count and uncheckable_count:
+            over_limit_summary = f"{exceeded_count}项超额，{uncheckable_count}项无法校验"
+        elif exceeded_count:
+            over_limit_summary = f"{exceeded_count}项超额"
+        elif uncheckable_count:
+            over_limit_summary = f"{uncheckable_count}项无法校验"
+        else:
+            over_limit_summary = None
         rows.append(
             DashboardRow(
                 id=group["id"],
@@ -380,6 +406,9 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
                 is_system_auto_submitted=group["is_system_auto_submitted"],
                 is_voided=group["is_voided"],
                 review_remark="；".join(group["review_remarks"]) or None,
+                has_over_limit=group["has_over_limit"],
+                over_limit_summary=over_limit_summary,
+                over_limit_reasons="；".join(group["over_limit_reasons"]) or None,
                 correction_pairs=group["correction_pairs"],
             )
         )

app/routers/reconciliation.py

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

app/routers/reports.py

@@ -41,6 +41,7 @@ from app.services.misc_work import is_misc_product
 from app.services.mold_locks import release_session_locks
 from app.services.report_allocations import refresh_report_allocations
 from app.services.report_lifecycle import purge_expired_voided_reports
+from app.services.report_over_limit import CHECK_SOURCE_SUBMIT, refresh_report_over_limit_snapshot
 from app.services.reporting_window import REPORTING_EXPIRED_DETAIL, is_reporting_expired
 from app.services.serializers import equipment_option, product_option, report_out
 from app.services.work_schedule import get_work_schedule_config
@@ -460,6 +461,7 @@ def submit_report(
     db.add(report)
     db.flush()
     refresh_report_allocations(db, report, schedule=schedule, commit=False)
+    refresh_report_over_limit_snapshot(db, report, source=CHECK_SOURCE_SUBMIT, checked_at=submitted_at)
     db.commit()
 
     saved = _get_report_or_404(db, report.id)

app/routers/reviews.py

@@ -36,6 +36,7 @@ from app.services.report_lifecycle import (
     purge_expired_voided_reports,
     restore_voided_report,
 )
+from app.services.report_over_limit import CHECK_SOURCE_REVIEW, refresh_report_over_limit_snapshot
 from app.services.serializers import report_out
 from app.services.work_schedule import WorkScheduleConfig, get_work_schedule_config
 from app.timezone import now
@@ -115,6 +116,14 @@ def _validate_normal_report_devices(db: Session, report: ProductionReport) -> No
             raise HTTPException(status_code=400, detail=f"设备 {device_no} 不是该考勤点的冲压设备")
 
 
+def _missing_over_limit_snapshot(report: ProductionReport) -> bool:
+    return any(
+        item.over_limit_checked_at is None and not item.over_limit_check_source
+        for item in report.items
+        if not is_cleaning_item(item)
+    )
+
+
 def _report_query():
     return select(ProductionReport).options(
         selectinload(ProductionReport.allocations),
@@ -397,6 +406,7 @@ def approve_report(
     schedule = get_work_schedule_config(db, report.attendance_point_name)
     item_map = {item.id: item for item in report.items}
     has_correction = False
+    over_limit_input_changed = False
 
     next_start_at = _effective_review_time(payload.start_at, report.start_at)
     next_end_at = _effective_review_time(payload.end_at, report.end_at)
@@ -442,11 +452,14 @@ def approve_report(
             if correction.raw_material_batch_no is not None
             else item.raw_material_batch_no
         )
-        if (
+        product_identity_changed = (
-            next_device_no != item.device_no
+            next_project_no != item.project_no
-            or next_project_no != item.project_no
             or next_product_name != item.product_name
             or next_process_name != item.process_name
+        )
+        if (
+            next_device_no != item.device_no
+            or product_identity_changed
         ):
             product = db.scalar(
                 select(Product).where(
@@ -480,10 +493,13 @@ def approve_report(
             item.standard_beat = as_float(product.standard_beat)
             item.standard_workload = 0
             has_correction = True
+            if product_identity_changed:
+                over_limit_input_changed = True
 
         if next_raw_material_batch_no != item.raw_material_batch_no:
             item.raw_material_batch_no = next_raw_material_batch_no
             has_correction = True
+            over_limit_input_changed = True
 
         if correction.changeover_count is not None:
             next_changeover_count = as_float(correction.changeover_count)
@@ -498,12 +514,15 @@ def approve_report(
         if correction.good_qty is not None and as_float(correction.good_qty) != as_float(item.good_qty):
             item.good_qty = correction.good_qty
             has_correction = True
+            over_limit_input_changed = True
         if correction.defect_qty is not None and as_float(correction.defect_qty) != as_float(item.defect_qty):
             item.defect_qty = correction.defect_qty
             has_correction = True
+            over_limit_input_changed = True
         if correction.scrap_qty is not None and as_float(correction.scrap_qty) != as_float(item.scrap_qty):
             item.scrap_qty = correction.scrap_qty
             has_correction = True
+            over_limit_input_changed = True
         if item_is_misc and correction.remark is not None:
             next_remark = str(correction.remark or "").strip() or None
             if next_remark != item.remark:
@@ -532,6 +551,8 @@ def approve_report(
     _apply_metrics(report, schedule)
     refresh_report_allocations(db, report, schedule=schedule, commit=False)
     reviewed_at = now()
+    if not is_cleaning and (over_limit_input_changed or _missing_over_limit_snapshot(report)):
+        refresh_report_over_limit_snapshot(db, report, source=CHECK_SOURCE_REVIEW, checked_at=reviewed_at)
     was_pending = report.status == ReportStatus.pending
     report.status = ReportStatus.approved
     report.reviewer_phone = user.phone

app/routers/usage_stats.py

@@ -2,12 +2,20 @@ 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
 from app.deps import require_roles
-from app.models import Equipment, Personnel, ProductionReport, ReportStatus, Role
+from app.models import (
+    Equipment,
+    Personnel,
+    ProductionReport,
+    ProductionReportAllocation,
+    ProductionReportItem,
+    ReportStatus,
+    Role,
+)
 from app.schemas import (
     PageResponse,
     UsageStatsDailyRow,
@@ -18,7 +26,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,21 +55,42 @@ 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)
+        select(ProductionReportAllocation)
-        .options(selectinload(ProductionReport.employee), selectinload(ProductionReport.items))
+        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
+        .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(ProductionReport.report_date >= start_date)
+        query = query.where(ProductionReportAllocation.allocation_date >= start_date)
     if end_date:
-        query = query.where(ProductionReport.report_date <= end_date)
+        query = query.where(ProductionReportAllocation.allocation_date <= end_date)
-    return query.order_by(ProductionReport.report_date.desc(), ProductionReport.id.desc())
+    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 +203,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 +305,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(
+            build_usage_stats_from_allocations(
-                reports=reports,
+                allocations=allocations,
                 category=category,
                 equipment_type_by_key=_equipment_type_map(db, point_names),
             ),
@@ -227,11 +343,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(
+            build_usage_stats_from_allocations(
-                reports=reports,
+                allocations=allocations,
                 category="device",
                 equipment_type_by_key=equipment_type_by_key,
             ),
@@ -241,8 +357,8 @@ def export_usage_stats_excel(
     )
     mold_rows = _sort_usage_rows(
         _filter_usage_rows(
-            build_usage_stats(
+            build_usage_stats_from_allocations(
-                reports=reports,
+                allocations=allocations,
                 category="mold",
                 equipment_type_by_key=equipment_type_by_key,
             ),
@@ -271,7 +387,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 +398,9 @@ def usage_stats_detail(
         "stamping_method": _clean(stamping_method),
         "metric_kind": _clean(metric_kind),
     }
-    rows = build_usage_stats(
+    return _build_usage_stats_detail_from_allocations(
-        reports=reports,
+        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),
-    )

app/schemas.py

@@ -491,6 +491,18 @@ class ReportItemOut(BaseModel):
     is_continuous_die: bool = False
     is_multi_person: bool = False
     remark: str | None = None
+    over_limit_status: str = "not_checked"
+    over_limit_type: str | None = None
+    over_limit_reason: str | None = None
+    over_limit_checked_at: datetime | None = None
+    over_limit_check_source: str | None = None
+    over_limit_batch_no: str | None = None
+    over_limit_product_gross_weight_kg: float | None = None
+    over_limit_issued_weight_kg: float | None = None
+    over_limit_max_reportable_qty: float | None = None
+    over_limit_previous_good_qty: float | None = None
+    over_limit_current_cumulative_qty: float | None = None
+    over_limit_current_report_qty: float | None = None
     allocations: list[ReportAllocationOut] = Field(default_factory=list)
     corrections: dict[str, Any] = Field(default_factory=dict)
 
@@ -530,6 +542,8 @@ class ReportOut(BaseModel):
     submitted_at: datetime
     is_system_auto_submitted: bool = False
     auto_submit_reason: str | None = None
+    has_over_limit: bool = False
+    over_limit_summary: str | None = None
     is_multi_person_assistant: bool = False
     multi_person_source_report_id: int | None = None
     is_voided: bool = False
@@ -644,6 +658,9 @@ class DashboardRow(BaseModel):
     is_system_auto_submitted: bool = False
     is_voided: bool = False
     review_remark: str | None = None
+    has_over_limit: bool = False
+    over_limit_summary: str | None = None
+    over_limit_reasons: str | None = None
     correction_pairs: list[dict[str, Any]] = Field(default_factory=list)
 
 

app/services/excel_export.py

@@ -133,6 +133,9 @@ def export_dashboard_rows(rows: list) -> bytes:
         "操作人数",
         "工序单价",
         "参考工资",
+        "是否超额",
+        "超额摘要",
+        "超额原因",
         "报工标记",
         "校正项",
         "备注",
@@ -184,6 +187,9 @@ def export_dashboard_rows(rows: list) -> bytes:
             row.operator_count,
             row.process_unit_price_yuan,
             row.reference_wage,
+            "是" if getattr(row, "has_over_limit", False) else "否",
+            getattr(row, "over_limit_summary", None),
+            getattr(row, "over_limit_reasons", None),
             "、".join(report_flags),
             correction_text,
             getattr(row, "review_remark", None),

app/services/report_allocations.py

@@ -1,11 +1,13 @@
 from collections import defaultdict
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import date, datetime, timedelta
+from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+from typing import Any
 
-from sqlalchemy import delete
+from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
-from app.models import ProductionReportAllocation
+from app.models import ProductionReport, ProductionReportAllocation
 from app.services.common import as_float, round2
 from app.services.metrics import (
     SHIFT_BUCKETS,
@@ -24,6 +26,8 @@ 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:
@@ -126,8 +130,34 @@ def _report_devices(report) -> list | None:
     return list(devices) if devices is not None else None
 
 
-def _item_reference_wage(item) -> float:
+def _decimal_value(value: Any) -> Decimal:
-    return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0))
+    if value is None:
+        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):
@@ -398,11 +428,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=round2(getattr(item, "good_qty", 0)),
+        good_qty=_round_split_value(getattr(item, "good_qty", 0)),
-        defect_qty=round2(getattr(item, "defect_qty", 0)),
+        defect_qty=_round_split_value(getattr(item, "defect_qty", 0)),
-        scrap_qty=round2(getattr(item, "scrap_qty", 0)),
+        scrap_qty=_round_split_value(getattr(item, "scrap_qty", 0)),
-        changeover_count=round2(getattr(item, "changeover_count", 0)),
+        changeover_count=_round_split_value(getattr(item, "changeover_count", 0)),
-        reference_wage=round2(_item_reference_wage(item)),
+        reference_wage=_round_split_value(_item_reference_wage(item)),
     )
 
 
@@ -421,14 +451,99 @@ 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=round2(as_float(getattr(item, "good_qty", 0)) * ratio),
+        good_qty=_scaled_split_value(getattr(item, "good_qty", 0), ratio),
-        defect_qty=round2(as_float(getattr(item, "defect_qty", 0)) * ratio),
+        defect_qty=_scaled_split_value(getattr(item, "defect_qty", 0), ratio),
-        scrap_qty=round2(as_float(getattr(item, "scrap_qty", 0)) * ratio),
+        scrap_qty=_scaled_split_value(getattr(item, "scrap_qty", 0), ratio),
-        changeover_count=round2(as_float(getattr(item, "changeover_count", 0)) * ratio),
+        changeover_count=_scaled_split_value(getattr(item, "changeover_count", 0), ratio),
-        reference_wage=round2(reference_wage * ratio),
+        reference_wage=_round_split_value(reference_wage * _decimal_value(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:
@@ -452,12 +567,14 @@ 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
-            rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective))
+            item_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
 
 
@@ -489,7 +606,14 @@ def refresh_report_allocations(
     commit: bool = False,
 ) -> list[ProductionReportAllocation]:
     """Rebuild allocation rows with metrics semantics, updating item allocation-derived fields."""
-    db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == getattr(report, "id")))
+    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/report_over_limit.py

@@ -0,0 +1,437 @@
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from datetime import datetime
+from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+
+from sqlalchemy import func, select, text
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
+from app.services.cleaning import is_cleaning_item
+from app.services.misc_work import is_misc_item
+from app.timezone import now
+
+
+logger = logging.getLogger(__name__)
+
+
+class OverLimitDataSourceError(RuntimeError):
+    pass
+
+
+OVER_LIMIT_STATUS_NOT_CHECKED = "not_checked"
+OVER_LIMIT_STATUS_OK = "ok"
+OVER_LIMIT_STATUS_EXCEEDED = "exceeded"
+OVER_LIMIT_STATUS_UNCHECKABLE = "uncheckable"
+
+OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL = "first_process_material"
+OVER_LIMIT_TYPE_PREVIOUS_PROCESS = "previous_process"
+OVER_LIMIT_TYPE_UNCHECKABLE = "uncheckable"
+
+CHECK_SOURCE_SUBMIT = "submit"
+CHECK_SOURCE_REVIEW = "review"
+
+_TWO_PLACES = Decimal("0.01")
+
+
+@dataclass(frozen=True)
+class OverLimitResult:
+    status: str
+    over_limit_type: str | None = None
+    reason: str | None = None
+    batch_no: str | None = None
+    product_gross_weight_kg: Decimal | None = None
+    issued_weight_kg: Decimal | None = None
+    max_reportable_qty: Decimal | None = None
+    previous_good_qty: Decimal | None = None
+    current_cumulative_qty: Decimal | None = None
+    current_report_qty: Decimal | None = None
+
+
+def _to_decimal(value) -> Decimal:
+    if value is None:
+        return Decimal("0")
+    if isinstance(value, Decimal):
+        return value
+    try:
+        return Decimal(str(value))
+    except (InvalidOperation, ValueError):
+        return Decimal("0")
+
+
+def _money_qty(value: Decimal) -> Decimal:
+    return value.quantize(_TWO_PLACES, rounding=ROUND_HALF_UP)
+
+
+def _positive_decimal(value) -> Decimal | None:
+    number = _to_decimal(value)
+    return number if number > 0 else None
+
+
+def _clean_text(value) -> str:
+    return str(value or "").strip()
+
+
+def _batch_no(value) -> str:
+    return _clean_text(value)
+
+
+def parse_process_order(process_name: str | None) -> int | None:
+    text_value = _clean_text(process_name)
+    match = re.fullmatch(r"0*(\d+)(?:\s*序)?", text_value)
+    if match is None:
+        return None
+    order = int(match.group(1))
+    return order if order > 0 else None
+
+
+def current_report_qty(item) -> Decimal:
+    total = _to_decimal(getattr(item, "good_qty", None))
+    total += _to_decimal(getattr(item, "defect_qty", None))
+    total += _to_decimal(getattr(item, "scrap_qty", None))
+    return _money_qty(total)
+
+
+def read_batch_issued_weight_kg(
+    db: Session,
+    batch_no: str | None,
+    project_no: str | None,
+    product_name: str | None,
+) -> Decimal | None:
+    normalized_batch_no = _batch_no(batch_no)
+    normalized_project_no = _clean_text(project_no)
+    normalized_product_name = _clean_text(product_name)
+    if not normalized_batch_no:
+        return None
+
+    try:
+        value = db.scalar(
+            text(
+                """
+                select sum(ledger.total_issued_weight_kg)
+                from pp_production_batch_ledger ledger
+                join md_item product on product.id = ledger.product_item_id
+                where ledger.material_lot_no = :batch_no
+                  and coalesce(ledger.miniapp_selectable, 1) = 1
+                  and (
+                    (:product_name <> '' and :project_no <> '' and product.item_name = :product_name and product.item_code = :project_no)
+                    or (:product_name <> '' and :project_no = '' and product.item_name = :product_name)
+                    or (:product_name = '' and :project_no <> '' and product.item_code = :project_no)
+                  )
+                """
+            ),
+            {
+                "batch_no": normalized_batch_no,
+                "project_no": normalized_project_no,
+                "product_name": normalized_product_name,
+            },
+        )
+    except SQLAlchemyError as exc:
+        logger.exception(
+            "failed to read production batch issued weight for batch=%s project=%s product=%s",
+            normalized_batch_no,
+            normalized_project_no,
+            normalized_product_name,
+        )
+        raise OverLimitDataSourceError("failed to read batch issued weight") from exc
+    weight = _positive_decimal(value)
+    return weight
+
+
+def _product_catalog_row(db: Session, item: ProductionReportItem) -> Product | None:
+    return db.get(
+        Product,
+        {
+            "attendance_point_name": getattr(item, "attendance_point_name", "") or "",
+            "project_no": getattr(item, "project_no", "") or "",
+            "product_name": getattr(item, "product_name", "") or "",
+            "device_no": "",
+            "process_name": getattr(item, "process_name", "") or "",
+        },
+    )
+
+
+def _process_orders_for_product(db: Session, item: ProductionReportItem) -> list[int]:
+    rows = db.scalars(
+        select(Product.process_name).where(
+            Product.attendance_point_name == (getattr(item, "attendance_point_name", "") or ""),
+            Product.project_no == (getattr(item, "project_no", "") or ""),
+            Product.product_name == (getattr(item, "product_name", "") or ""),
+            Product.device_no == "",
+        )
+    ).all()
+    return sorted({order for name in rows if (order := parse_process_order(name)) is not None})
+
+
+def _previous_process_order(db: Session, item: ProductionReportItem, current_order: int) -> int | None:
+    previous_orders = [order for order in _process_orders_for_product(db, item) if order < current_order]
+    return max(previous_orders) if previous_orders else None
+
+
+def _process_name_candidates(process_order: int) -> tuple[str, ...]:
+    values: set[str] = set()
+    for width in range(1, 5):
+        value = str(process_order).zfill(width)
+        values.add(value)
+        values.add(f"{value}序")
+        values.add(f"{value} 序")
+    return tuple(sorted(values))
+
+
+def _history_items(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> list[ProductionReportItem]:
+    query = (
+        select(ProductionReportItem)
+        .join(ProductionReport)
+        .where(
+            ProductionReport.status.in_([ReportStatus.pending, ReportStatus.approved]),
+            ProductionReport.is_voided.is_(False),
+            ProductionReportItem.attendance_point_name == (getattr(item, "attendance_point_name", "") or ""),
+            ProductionReportItem.project_no == (getattr(item, "project_no", "") or ""),
+            ProductionReportItem.product_name == (getattr(item, "product_name", "") or ""),
+            func.trim(ProductionReportItem.raw_material_batch_no) == _batch_no(getattr(item, "raw_material_batch_no", "")),
+            func.trim(ProductionReportItem.process_name).in_(_process_name_candidates(process_order)),
+        )
+    )
+    if exclude_report_id is not None:
+        query = query.where(ProductionReport.id != exclude_report_id)
+    return list(db.scalars(query).all())
+
+
+def _historical_consumed_qty(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> Decimal:
+    total = Decimal("0")
+    for history_item in _history_items(db, item, process_order, exclude_report_id):
+        total += current_report_qty(history_item)
+    return _money_qty(total)
+
+
+def _historical_good_qty(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> Decimal:
+    total = Decimal("0")
+    for history_item in _history_items(db, item, process_order, exclude_report_id):
+        total += _to_decimal(history_item.good_qty)
+    return _money_qty(total)
+
+
+def _format_process(order: int) -> str:
+    return f"{order}序"
+
+
+def _same_scope(left: ProductionReportItem, right: ProductionReportItem) -> bool:
+    return (
+        (getattr(left, "attendance_point_name", "") or "") == (getattr(right, "attendance_point_name", "") or "")
+        and (getattr(left, "project_no", "") or "") == (getattr(right, "project_no", "") or "")
+        and (getattr(left, "product_name", "") or "") == (getattr(right, "product_name", "") or "")
+        and _batch_no(getattr(left, "raw_material_batch_no", "")) == _batch_no(getattr(right, "raw_material_batch_no", ""))
+    )
+
+
+def _current_report_consumed_qty(
+    report: ProductionReport,
+    item: ProductionReportItem,
+    process_order: int,
+) -> Decimal:
+    total = Decimal("0")
+    for report_item in list(getattr(report, "items", []) or []):
+        if _same_scope(item, report_item) and parse_process_order(getattr(report_item, "process_name", None)) == process_order:
+            total += current_report_qty(report_item)
+    return _money_qty(total)
+
+
+def _current_report_good_qty(
+    report: ProductionReport,
+    item: ProductionReportItem,
+    process_order: int,
+) -> Decimal:
+    total = Decimal("0")
+    for report_item in list(getattr(report, "items", []) or []):
+        if _same_scope(item, report_item) and parse_process_order(getattr(report_item, "process_name", None)) == process_order:
+            total += _to_decimal(getattr(report_item, "good_qty", 0))
+    return _money_qty(total)
+
+
+def evaluate_report_item_over_limit(
+    db: Session,
+    report: ProductionReport,
+    item: ProductionReportItem,
+    *,
+    exclude_report_id: int | None = None,
+) -> OverLimitResult:
+    if exclude_report_id is None:
+        exclude_report_id = getattr(report, "id", None)
+
+    batch_no = _batch_no(getattr(item, "raw_material_batch_no", "")) or None
+    report_qty = current_report_qty(item)
+
+    if is_cleaning_item(item) or is_misc_item(item) or str(getattr(item, "process_name", "") or "").strip() == "清洗":
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_NOT_CHECKED,
+            batch_no=batch_no,
+            current_report_qty=report_qty,
+        )
+
+    current_order = parse_process_order(getattr(item, "process_name", None))
+    if current_order is None:
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_NOT_CHECKED,
+            batch_no=batch_no,
+            current_report_qty=report_qty,
+        )
+
+    if not batch_no:
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_UNCHECKABLE,
+            over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+            reason="缺少材料批次，无法校验超额",
+            current_report_qty=report_qty,
+        )
+
+    historical_qty = _historical_consumed_qty(db, item, current_order, exclude_report_id)
+    current_cumulative_qty = _money_qty(historical_qty + _current_report_consumed_qty(report, item, current_order))
+    previous_order = _previous_process_order(db, item, current_order)
+
+    if previous_order is None:
+        product = _product_catalog_row(db, item)
+        gross_weight = _positive_decimal(getattr(product, "product_gross_weight_kg", None))
+        if gross_weight is None:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="缺少产品毛重，无法校验超额",
+                batch_no=batch_no,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+
+        try:
+            issued_weight = read_batch_issued_weight_kg(db, batch_no, item.project_no, item.product_name)
+        except OverLimitDataSourceError:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="读取批次投入重量失败，无法校验超额",
+                batch_no=batch_no,
+                product_gross_weight_kg=gross_weight,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+        if issued_weight is None:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="缺少批次投入重量，无法校验超额",
+                batch_no=batch_no,
+                product_gross_weight_kg=gross_weight,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+
+        max_reportable_qty = _money_qty(issued_weight / gross_weight)
+        status = OVER_LIMIT_STATUS_EXCEEDED if current_cumulative_qty > max_reportable_qty else OVER_LIMIT_STATUS_OK
+        reason = None
+        if status == OVER_LIMIT_STATUS_EXCEEDED:
+            reason = (
+                f"{_format_process(current_order)}累计报工数量超过本批次材料可支撑数量："
+                f"{current_cumulative_qty} > {max_reportable_qty}"
+            )
+        return OverLimitResult(
+            status=status,
+            over_limit_type=OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL if status == OVER_LIMIT_STATUS_EXCEEDED else None,
+            reason=reason,
+            batch_no=batch_no,
+            product_gross_weight_kg=gross_weight,
+            issued_weight_kg=issued_weight,
+            max_reportable_qty=max_reportable_qty,
+            current_cumulative_qty=current_cumulative_qty,
+            current_report_qty=report_qty,
+        )
+
+    previous_good_qty = _money_qty(
+        _historical_good_qty(db, item, previous_order, exclude_report_id)
+        + _current_report_good_qty(report, item, previous_order)
+    )
+    status = OVER_LIMIT_STATUS_EXCEEDED if current_cumulative_qty > previous_good_qty else OVER_LIMIT_STATUS_OK
+    reason = None
+    if status == OVER_LIMIT_STATUS_EXCEEDED:
+        reason = (
+            f"{_format_process(current_order)}累计报工数量超过{_format_process(previous_order)}累计成品数量："
+            f"{current_cumulative_qty} > {previous_good_qty}"
+        )
+    return OverLimitResult(
+        status=status,
+        over_limit_type=OVER_LIMIT_TYPE_PREVIOUS_PROCESS if status == OVER_LIMIT_STATUS_EXCEEDED else None,
+        reason=reason,
+        batch_no=batch_no,
+        previous_good_qty=previous_good_qty,
+        current_cumulative_qty=current_cumulative_qty,
+        current_report_qty=report_qty,
+    )
+
+
+def _apply_result_to_item(
+    item: ProductionReportItem,
+    result: OverLimitResult,
+    *,
+    source: str,
+    checked_at: datetime,
+) -> None:
+    item.over_limit_status = result.status
+    item.over_limit_type = result.over_limit_type
+    item.over_limit_reason = result.reason
+    item.over_limit_checked_at = checked_at
+    item.over_limit_check_source = source
+    item.over_limit_batch_no = result.batch_no
+    item.over_limit_product_gross_weight_kg = result.product_gross_weight_kg
+    item.over_limit_issued_weight_kg = result.issued_weight_kg
+    item.over_limit_max_reportable_qty = result.max_reportable_qty
+    item.over_limit_previous_good_qty = result.previous_good_qty
+    item.over_limit_current_cumulative_qty = result.current_cumulative_qty
+    item.over_limit_current_report_qty = result.current_report_qty
+
+
+def refresh_report_over_limit_snapshot(
+    db: Session,
+    report: ProductionReport,
+    source: str,
+    *,
+    checked_at: datetime | None = None,
+) -> None:
+    timestamp = checked_at or now()
+    exceeded_count = 0
+    uncheckable_count = 0
+
+    for item in list(getattr(report, "items", []) or []):
+        result = evaluate_report_item_over_limit(db, report, item)
+        _apply_result_to_item(item, result, source=source, checked_at=timestamp)
+        if result.status == OVER_LIMIT_STATUS_EXCEEDED:
+            exceeded_count += 1
+        elif result.status == OVER_LIMIT_STATUS_UNCHECKABLE:
+            uncheckable_count += 1
+
+    report.has_over_limit = exceeded_count > 0
+    if exceeded_count and uncheckable_count:
+        report.over_limit_summary = f"{exceeded_count}项超额，{uncheckable_count}项无法校验"
+    elif exceeded_count:
+        report.over_limit_summary = f"{exceeded_count}项超额"
+    elif uncheckable_count:
+        report.over_limit_summary = f"{uncheckable_count}项无法校验"
+    else:
+        report.over_limit_summary = None
+    db.flush()

app/services/serializers.py

@@ -52,6 +52,10 @@ def temporary_expired(person: Personnel) -> bool:
     )
 
 
+def _optional_float(value) -> float | None:
+    return None if value is None else as_float(value)
+
+
 def attendance_point_out(point: AttendancePoint) -> AttendancePointOut:
     return AttendancePointOut(
         name=point.name,
@@ -123,7 +127,6 @@ def personnel_summary_out(person: Personnel) -> dict:
 
 
 def product_out(product: Product) -> ProductOut:
-    optional_float = lambda value: None if value is None else as_float(value)
     return ProductOut(
         attendance_point_name=product.attendance_point_name,
         project_no=product.project_no,
@@ -132,10 +135,10 @@ def product_out(product: Product) -> ProductOut:
         material_code=product.material_code,
         material_name=product.material_name,
         supplier=product.supplier,
-        product_net_weight_kg=optional_float(product.product_net_weight_kg),
+        product_net_weight_kg=_optional_float(product.product_net_weight_kg),
-        product_gross_weight_kg=optional_float(product.product_gross_weight_kg),
+        product_gross_weight_kg=_optional_float(product.product_gross_weight_kg),
-        scrap_loss_rate=optional_float(product.scrap_loss_rate),
+        scrap_loss_rate=_optional_float(product.scrap_loss_rate),
-        waste_price_yuan_per_kg=optional_float(product.waste_price_yuan_per_kg),
+        waste_price_yuan_per_kg=_optional_float(product.waste_price_yuan_per_kg),
         device_no=product.device_no,
         process_name=product.process_name,
         stamping_method=product.stamping_method,
@@ -410,6 +413,18 @@ def report_item_out(item: ProductionReportItem, corrections: dict | None = None)
         is_continuous_die=is_continuous_die_item(item),
         is_multi_person=is_multi_person_item(item),
         remark=item.remark,
+        over_limit_status=item.over_limit_status or "not_checked",
+        over_limit_type=item.over_limit_type,
+        over_limit_reason=item.over_limit_reason,
+        over_limit_checked_at=item.over_limit_checked_at,
+        over_limit_check_source=item.over_limit_check_source,
+        over_limit_batch_no=item.over_limit_batch_no,
+        over_limit_product_gross_weight_kg=_optional_float(item.over_limit_product_gross_weight_kg),
+        over_limit_issued_weight_kg=_optional_float(item.over_limit_issued_weight_kg),
+        over_limit_max_reportable_qty=_optional_float(item.over_limit_max_reportable_qty),
+        over_limit_previous_good_qty=_optional_float(item.over_limit_previous_good_qty),
+        over_limit_current_cumulative_qty=_optional_float(item.over_limit_current_cumulative_qty),
+        over_limit_current_report_qty=_optional_float(item.over_limit_current_report_qty),
         allocations=[report_allocation_out(row) for row in allocation_rows],
         corrections=corrections or {},
     )
@@ -505,6 +520,8 @@ def report_out(report: ProductionReport, schedule: WorkScheduleConfig | None = N
         submitted_at=report.submitted_at,
         is_system_auto_submitted=bool(report.is_system_auto_submitted),
         auto_submit_reason=report.auto_submit_reason,
+        has_over_limit=bool(report.has_over_limit),
+        over_limit_summary=report.over_limit_summary,
         is_multi_person_assistant=bool(report.is_multi_person_assistant),
         multi_person_source_report_id=report.multi_person_source_report_id,
         is_voided=bool(report.is_voided),

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,6 +89,10 @@ 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:
@@ -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]:

scripts/migrate_report_allocations.py

@@ -2,12 +2,16 @@ from pathlib import Path
 import re
 import sys
 
-from sqlalchemy import text
+from sqlalchemy import select, text
+from sqlalchemy.orm import selectinload
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT))
 
-from app.database import engine  # noqa: E402
+from app.database import SessionLocal, 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"
@@ -45,11 +49,21 @@ REQUIRED_COLUMNS = {
     },
 }
 REQUIRED_INDEXES = {
-    "idx_report_allocations_report": ("report_id",),
+    "uq_report_allocations_report_item_date": {
-    "idx_report_allocations_item": ("report_item_id",),
+        "columns": ("report_id", "report_item_id", "allocation_date"),
-    "idx_report_allocations_date": ("allocation_date",),
+        "non_unique": 0,
-    "idx_report_allocations_point_date": ("attendance_point_name", "allocation_date"),
+    },
-    "idx_report_allocations_employee_date": ("employee_phone", "allocation_date"),
+    "idx_report_allocations_report": {"columns": ("report_id",), "non_unique": 1},
+    "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"),
@@ -149,22 +163,51 @@ 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 find_equivalent_index(indexes: dict[str, IndexMetadata], columns: tuple[str, ...]) -> str | None:
+def required_index_columns(index_name: str) -> tuple[str, ...]:
+    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 index_name != "PRIMARY" and metadata["columns"] == columns and metadata["non_unique"] == 1:
+        if (
+            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_INDEXES[index_name] != columns:
+    if index_name not in REQUIRED_INDEXES or required_index_columns(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:
+    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(
+            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)
+    equivalent_index = find_equivalent_index(indexes, columns, non_unique)
     if equivalent_index:
         conn.execute(
             text(
@@ -175,14 +218,51 @@ 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 INDEX {quote_identifier(index_name)} "
+            f"CREATE {unique_sql}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 = [
@@ -231,6 +311,8 @@ 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),
@@ -247,8 +329,9 @@ def ensure_schema(conn) -> None:
         )
     )
 
-    for index_name, columns in REQUIRED_INDEXES.items():
+    cleanup_duplicate_allocation_keys(conn)
-        create_index(conn, index_name, columns)
+    for index_name in REQUIRED_INDEXES:
+        create_index(conn, index_name, required_index_columns(index_name))
     cleanup_duplicate_fk_indexes(conn)
 
 
@@ -315,19 +398,21 @@ def verify_schema(conn) -> None:
 
     wrong_indexes = [
         f"{index_name}({', '.join(indexes[index_name]['columns'])})"
-        for index_name, columns in REQUIRED_INDEXES.items()
+        for index_name in REQUIRED_INDEXES
-        if indexes.get(index_name, {}).get("columns") != columns
+        if indexes.get(index_name, {}).get("columns") != required_index_columns(index_name)
     ]
     if wrong_indexes:
         raise RuntimeError(f"{TABLE_NAME} incompatible indexes: {', '.join(wrong_indexes)}")
 
-    unique_required_indexes = [
+    wrong_index_uniqueness = [
         index_name
         for index_name in REQUIRED_INDEXES
-        if indexes[index_name]["non_unique"] != 1
+        if indexes[index_name]["non_unique"] != required_index_non_unique(index_name)
     ]
-    if unique_required_indexes:
+    if wrong_index_uniqueness:
-        raise RuntimeError(f"{TABLE_NAME} required indexes are unique: {', '.join(unique_required_indexes)}")
+        raise RuntimeError(
+            f"{TABLE_NAME} incompatible index uniqueness: {', '.join(wrong_index_uniqueness)}"
+        )
 
     duplicate_single_column_indexes = [
         index_name
@@ -374,12 +459,72 @@ 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)
 
-    print("report allocations schema migrated")
+    processed, skipped = _backfill()
+    print(f"report allocations migrated, processed={processed}, skipped={skipped}")
 
 
 if __name__ == "__main__":

scripts/migrate_report_over_limit_snapshot.py

@@ -0,0 +1,177 @@
+from pathlib import Path
+import re
+import sys
+
+from sqlalchemy import text
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from app.database import engine  # noqa: E402
+
+
+IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+REQUIRED_COLUMNS = {
+    "production_reports": {
+        "has_over_limit": "BOOLEAN NOT NULL DEFAULT FALSE AFTER workload_rate",
+        "over_limit_summary": "VARCHAR(500) NULL AFTER has_over_limit",
+    },
+    "production_report_items": {
+        "over_limit_status": "VARCHAR(32) NOT NULL DEFAULT 'not_checked' AFTER scrap_qty",
+        "over_limit_type": "VARCHAR(64) NULL AFTER over_limit_status",
+        "over_limit_reason": "VARCHAR(500) NULL AFTER over_limit_type",
+        "over_limit_checked_at": "DATETIME NULL AFTER over_limit_reason",
+        "over_limit_check_source": "VARCHAR(32) NULL AFTER over_limit_checked_at",
+        "over_limit_batch_no": "VARCHAR(128) NULL AFTER over_limit_check_source",
+        "over_limit_product_gross_weight_kg": "DECIMAL(12, 4) NULL AFTER over_limit_batch_no",
+        "over_limit_issued_weight_kg": "DECIMAL(18, 6) NULL AFTER over_limit_product_gross_weight_kg",
+        "over_limit_max_reportable_qty": "DECIMAL(12, 2) NULL AFTER over_limit_issued_weight_kg",
+        "over_limit_previous_good_qty": "DECIMAL(12, 2) NULL AFTER over_limit_max_reportable_qty",
+        "over_limit_current_cumulative_qty": "DECIMAL(12, 2) NULL AFTER over_limit_previous_good_qty",
+        "over_limit_current_report_qty": "DECIMAL(12, 2) NULL AFTER over_limit_current_cumulative_qty",
+    },
+}
+
+REQUIRED_INDEXES = {
+    "production_reports": {
+        "idx_reports_over_limit": ("has_over_limit", "submitted_at"),
+    },
+    "production_report_items": {
+        "idx_report_items_over_limit_status": ("over_limit_status",),
+        "idx_report_items_process_batch": (
+            "attendance_point_name",
+            "project_no",
+            "product_name",
+            "raw_material_batch_no",
+            "process_name",
+        ),
+    },
+}
+
+
+def quote_identifier(identifier: str) -> str:
+    if not IDENTIFIER_RE.fullmatch(identifier):
+        raise ValueError(f"unexpected SQL identifier: {identifier}")
+    return f"`{identifier}`"
+
+
+def table_exists(conn, table_name: str) -> bool:
+    return bool(
+        conn.execute(
+            text(
+                """
+                SELECT COUNT(1)
+                FROM information_schema.tables
+                WHERE table_schema = DATABASE()
+                  AND table_name = :table_name
+                """
+            ),
+            {"table_name": table_name},
+        ).scalar_one()
+    )
+
+
+def existing_columns(conn, table_name: str) -> set[str]:
+    rows = conn.execute(
+        text(
+            """
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_schema = DATABASE()
+              AND table_name = :table_name
+            """
+        ),
+        {"table_name": table_name},
+    ).all()
+    return {row[0] for row in rows}
+
+
+def existing_indexes(conn, table_name: str) -> dict[str, tuple[str, ...]]:
+    rows = conn.execute(
+        text(
+            """
+            SELECT index_name, column_name
+            FROM information_schema.statistics
+            WHERE table_schema = DATABASE()
+              AND table_name = :table_name
+            ORDER BY index_name, seq_in_index
+            """
+        ),
+        {"table_name": table_name},
+    ).all()
+    indexes: dict[str, list[str]] = {}
+    for index_name, column_name in rows:
+        indexes.setdefault(index_name, []).append(column_name)
+    return {index_name: tuple(columns) for index_name, columns in indexes.items()}
+
+
+def add_missing_columns(conn, table_name: str) -> None:
+    columns = existing_columns(conn, table_name)
+    for column_name, column_definition in REQUIRED_COLUMNS[table_name].items():
+        if column_name in columns:
+            continue
+        conn.execute(
+            text(
+                f"ALTER TABLE {quote_identifier(table_name)} "
+                f"ADD COLUMN {quote_identifier(column_name)} {column_definition}"
+            )
+        )
+        columns.add(column_name)
+
+
+def create_missing_indexes(conn, table_name: str) -> None:
+    indexes = existing_indexes(conn, table_name)
+    for index_name, columns in REQUIRED_INDEXES[table_name].items():
+        existing_columns_for_index = indexes.get(index_name)
+        if existing_columns_for_index == columns:
+            continue
+        if existing_columns_for_index is not None:
+            raise RuntimeError(
+                f"{table_name}.{index_name} exists with columns "
+                f"{existing_columns_for_index}, expected {columns}; "
+                "please drop or rename the conflicting index before rerunning"
+            )
+        column_sql = ", ".join(quote_identifier(column) for column in columns)
+        conn.execute(
+            text(
+                f"CREATE INDEX {quote_identifier(index_name)} "
+                f"ON {quote_identifier(table_name)} ({column_sql})"
+            )
+        )
+
+
+def verify_schema(conn) -> None:
+    missing: list[str] = []
+    for table_name, columns in REQUIRED_COLUMNS.items():
+        if not table_exists(conn, table_name):
+            missing.append(f"{table_name} table")
+            continue
+
+        existing_column_names = existing_columns(conn, table_name)
+        for column_name in columns:
+            if column_name not in existing_column_names:
+                missing.append(f"{table_name}.{column_name}")
+
+        indexes = existing_indexes(conn, table_name)
+        for index_name, index_columns in REQUIRED_INDEXES[table_name].items():
+            if indexes.get(index_name) != index_columns:
+                missing.append(f"{table_name}.{index_name}")
+
+    if missing:
+        raise RuntimeError(f"missing report over limit snapshot objects: {', '.join(missing)}")
+
+
+def main() -> None:
+    with engine.begin() as conn:
+        for table_name in REQUIRED_COLUMNS:
+            if not table_exists(conn, table_name):
+                raise RuntimeError(f"missing required table: {table_name}")
+            add_missing_columns(conn, table_name)
+            create_missing_indexes(conn, table_name)
+        verify_schema(conn)
+    print("report over limit snapshot migrated")
+
+
+if __name__ == "__main__":
+    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=350,
+            reference_wage=700,
         ),
         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=300,
+            reference_wage=600,
         ),
     ]
 
@@ -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 == 300
+    assert current.reference_wage == 600
 
     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 == 350
+    assert previous.reference_wage == 700
 
 
 def test_export_dashboard_rows_writes_allocation_and_source_dates_with_warning_columns():
@@ -156,3 +156,237 @@ def test_export_dashboard_rows_writes_allocation_and_source_dates_with_warning_c
     assert sheet.cell(row=2, column=3).value == "2026-07-23、2026-07-25"
     assert sheet.cell(row=2, column=12).fill.fgColor.rgb == "FFFF6666"
     assert sheet.cell(row=2, column=15).fill.fgColor.rgb == "FFFF6666"
+
+
+def test_dashboard_rows_include_over_limit_summary_from_reports():
+    report = SimpleNamespace(
+        id=1,
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        report_date=date(2026, 7, 25),
+        has_over_limit=True,
+        over_limit_summary="1项超额",
+        review_remark=None,
+        employee=SimpleNamespace(name="张三", is_temporary=False),
+        items=[],
+        audit_logs=[],
+        session=None,
+        is_system_auto_submitted=False,
+        is_voided=False,
+        is_multi_person_assistant=False,
+    )
+    item = SimpleNamespace(
+        id=1,
+        report=report,
+        report_id=1,
+        attendance_point_name="总厂",
+        device_no="A",
+        project_no="P1",
+        product_name="产品A",
+        material_code="M1",
+        material_name="材料A",
+        raw_material_batch_no="LOT-1",
+        process_name="1",
+        remark=None,
+        stamping_method="冲压",
+        operator_count=1,
+        process_unit_price_yuan=1,
+        standard_beat=1,
+        good_qty=60,
+        defect_qty=0,
+        scrap_qty=0,
+        changeover_count=0,
+        allocated_minutes=60,
+        over_limit_status="exceeded",
+        over_limit_reason="1序累计报工数量超过本批次材料可支撑数量",
+    )
+    report.items = [item]
+    allocation = SimpleNamespace(
+        report=report,
+        item=item,
+        report_id=1,
+        allocation_date=date(2026, 7, 25),
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        good_qty=60,
+        defect_qty=0,
+        scrap_qty=0,
+        effective_minutes=60,
+        day_minutes=60,
+        overtime_minutes=0,
+        night_minutes=0,
+        reference_wage=60,
+        changeover_count=0,
+    )
+
+    rows = dashboard._dashboard_rows_from_allocations([allocation])
+
+    assert rows[0].has_over_limit is True
+    assert rows[0].over_limit_summary == "1项超额"
+    assert rows[0].over_limit_reasons == "1序累计报工数量超过本批次材料可支撑数量"
+
+
+def test_export_dashboard_rows_writes_over_limit_columns():
+    row = SimpleNamespace(
+        attendance_point_name="总厂",
+        report_date=date(2026, 7, 24),
+        source_report_dates=[date(2026, 7, 24)],
+        source_report_date=None,
+        employee_name="张三",
+        employee_phone="13800000000",
+        worker_type="正式工",
+        project_no="P1",
+        product_name="23#",
+        process_name="1",
+        effective_minutes=60,
+        shift_distribution_text="白班1小时",
+        total_good_qty=50,
+        total_defect_qty=2,
+        expected_workload=100,
+        actual_beat=20,
+        standard_beat=10,
+        profile_no="PF-1",
+        material_code="M-1",
+        material_name="铝材",
+        raw_material_batch_no="BATCH-1",
+        device_no="A-01",
+        stamping_method="单冲",
+        changeover_count=1,
+        operator_count=1,
+        process_unit_price_yuan=2,
+        reference_wage=100,
+        has_over_limit=True,
+        over_limit_summary="1项超额",
+        over_limit_reasons="1序累计报工数量超过本批次材料可支撑数量",
+        is_voided=False,
+        is_multi_person=False,
+        is_continuous_die=False,
+        correction_pairs=[],
+        review_remark="复核通过",
+    )
+
+    workbook = load_workbook(BytesIO(export_dashboard_rows([row])))
+    sheet = workbook["报工看板"]
+    headers = [sheet.cell(row=1, column=index).value for index in range(1, sheet.max_column + 1)]
+
+    assert "是否超额" in headers
+    assert "超额摘要" in headers
+    assert "超额原因" in headers
+    assert sheet.cell(row=2, column=headers.index("是否超额") + 1).value == "是"
+    assert sheet.cell(row=2, column=headers.index("超额摘要") + 1).value == "1项超额"
+    assert sheet.cell(row=2, column=headers.index("超额原因") + 1).value == "1序累计报工数量超过本批次材料可支撑数量"
+
+
+def test_dashboard_over_limit_does_not_spread_to_other_items_in_same_report():
+    report = SimpleNamespace(
+        id=1,
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        report_date=date(2026, 7, 25),
+        has_over_limit=True,
+        over_limit_summary="1项超额",
+        review_remark=None,
+        employee=SimpleNamespace(name="张三", is_temporary=False),
+        items=[],
+        audit_logs=[],
+        session=None,
+        is_system_auto_submitted=False,
+        is_voided=False,
+        is_multi_person_assistant=False,
+    )
+    exceeded_item = SimpleNamespace(
+        id=1,
+        report=report,
+        report_id=1,
+        attendance_point_name="总厂",
+        device_no="A",
+        project_no="P1",
+        product_name="产品A",
+        material_code="M1",
+        material_name="材料A",
+        raw_material_batch_no="LOT-1",
+        process_name="1",
+        remark=None,
+        stamping_method="冲压",
+        operator_count=1,
+        process_unit_price_yuan=1,
+        standard_beat=1,
+        good_qty=60,
+        defect_qty=0,
+        scrap_qty=0,
+        changeover_count=0,
+        allocated_minutes=60,
+        over_limit_status="exceeded",
+        over_limit_reason="1序累计报工数量超过本批次材料可支撑数量",
+    )
+    normal_item = SimpleNamespace(
+        id=2,
+        report=report,
+        report_id=1,
+        attendance_point_name="总厂",
+        device_no="B",
+        project_no="P1",
+        product_name="产品A",
+        material_code="M1",
+        material_name="材料A",
+        raw_material_batch_no="LOT-1",
+        process_name="2",
+        remark=None,
+        stamping_method="冲压",
+        operator_count=1,
+        process_unit_price_yuan=1,
+        standard_beat=1,
+        good_qty=40,
+        defect_qty=0,
+        scrap_qty=0,
+        changeover_count=0,
+        allocated_minutes=40,
+        over_limit_status="ok",
+        over_limit_reason=None,
+    )
+    report.items = [exceeded_item, normal_item]
+    allocations = [
+        SimpleNamespace(
+            report=report,
+            item=exceeded_item,
+            report_id=1,
+            allocation_date=date(2026, 7, 25),
+            attendance_point_name="总厂",
+            employee_phone="13800000000",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            effective_minutes=60,
+            day_minutes=60,
+            overtime_minutes=0,
+            night_minutes=0,
+            reference_wage=60,
+            changeover_count=0,
+        ),
+        SimpleNamespace(
+            report=report,
+            item=normal_item,
+            report_id=1,
+            allocation_date=date(2026, 7, 25),
+            attendance_point_name="总厂",
+            employee_phone="13800000000",
+            good_qty=40,
+            defect_qty=0,
+            scrap_qty=0,
+            effective_minutes=40,
+            day_minutes=40,
+            overtime_minutes=0,
+            night_minutes=0,
+            reference_wage=40,
+            changeover_count=0,
+        ),
+    ]
+
+    rows = dashboard._dashboard_rows_from_allocations(allocations)
+    rows_by_process = {row.process_name: row for row in rows}
+
+    assert rows_by_process["1"].has_over_limit is True
+    assert rows_by_process["1"].over_limit_reasons == "1序累计报工数量超过本批次材料可支撑数量"
+    assert rows_by_process["2"].has_over_limit is False
+    assert rows_by_process["2"].over_limit_summary is None
+    assert rows_by_process["2"].over_limit_reasons is None

tests/test_reconciliation_allocations.py

@@ -0,0 +1,223 @@
+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,8 +2,10 @@ from datetime import date, datetime
 from decimal import Decimal
 from types import SimpleNamespace
 
-from sqlalchemy import BigInteger, create_engine, select
+import pytest
+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
@@ -33,6 +35,7 @@ from app.schemas import (
     ReportSubmitRequest,
 )
 from app.services import auto_submit as auto_submit_service
+from app.services import reporting_window as reporting_window_service
 from app.services.report_lifecycle import purge_expired_voided_reports
 from app.services.metrics import calculate_report_metrics
 from app.services.report_allocations import (
@@ -203,6 +206,51 @@ 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)
@@ -660,6 +708,51 @@ 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:
@@ -804,6 +897,7 @@ def test_submit_report_creates_allocations(monkeypatch):
         db.add_all([session, _press_equipment(), _product()])
         db.commit()
         monkeypatch.setattr(reports_router, "report_out", lambda report, schedule=None: report)
+        monkeypatch.setattr(reporting_window_service, "now", lambda: datetime(2026, 7, 25, 11, 5))
 
         report = reports_router.submit_report(
             ReportSubmitRequest(

tests/test_report_over_limit.py

@@ -0,0 +1,945 @@
+from datetime import date, datetime, timedelta
+from decimal import Decimal
+from types import SimpleNamespace
+
+from sqlalchemy import BigInteger, create_engine, inspect
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.orm import sessionmaker
+
+from app.database import Base
+from app.models import Equipment, Product, ProductionReport, ProductionReportItem, ReportStatus, Role, SessionStatus, WorkSession, WorkSessionDevice
+from app.routers import reports as reports_router
+from app.routers import reviews as reviews_router
+from app.schemas import (
+    ApproveReportRequest,
+    ReportItemCorrection,
+    ReportSubmitBlock,
+    ReportSubmitItem,
+    ReportSubmitRequest,
+)
+from app.services.report_over_limit import (
+    OVER_LIMIT_STATUS_EXCEEDED,
+    OVER_LIMIT_STATUS_NOT_CHECKED,
+    OVER_LIMIT_TYPE_UNCHECKABLE,
+    current_report_qty,
+    evaluate_report_item_over_limit,
+    parse_process_order,
+    refresh_report_over_limit_snapshot,
+)
+from app.services.serializers import report_out
+from app.timezone import now
+
+
+@compiles(BigInteger, "sqlite")
+def _compile_big_integer_for_sqlite(type_, compiler, **kw):
+    return "INTEGER"
+
+
+def test_report_over_limit_columns_exist_on_sqlite_metadata():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    inspector = inspect(engine)
+
+    report_column_rows = inspector.get_columns("production_reports")
+    item_column_rows = inspector.get_columns("production_report_items")
+    report_columns = {column["name"] for column in report_column_rows}
+    item_columns = {column["name"] for column in item_column_rows}
+    report_column_by_name = {column["name"]: column for column in report_column_rows}
+    item_column_by_name = {column["name"]: column for column in item_column_rows}
+    report_indexes = {index["name"]: tuple(index["column_names"]) for index in inspector.get_indexes("production_reports")}
+    item_indexes = {
+        index["name"]: tuple(index["column_names"])
+        for index in inspector.get_indexes("production_report_items")
+    }
+
+    assert {"has_over_limit", "over_limit_summary"}.issubset(report_columns)
+    assert {
+        "over_limit_status",
+        "over_limit_type",
+        "over_limit_reason",
+        "over_limit_checked_at",
+        "over_limit_check_source",
+        "over_limit_batch_no",
+        "over_limit_product_gross_weight_kg",
+        "over_limit_issued_weight_kg",
+        "over_limit_max_reportable_qty",
+        "over_limit_previous_good_qty",
+        "over_limit_current_cumulative_qty",
+        "over_limit_current_report_qty",
+    }.issubset(item_columns)
+
+    assert report_indexes["idx_reports_over_limit"] == ("has_over_limit", "submitted_at")
+    assert item_indexes["idx_report_items_over_limit_status"] == ("over_limit_status",)
+    assert item_indexes["idx_report_items_process_batch"] == (
+        "attendance_point_name",
+        "project_no",
+        "product_name",
+        "raw_material_batch_no",
+        "process_name",
+    )
+    assert report_column_by_name["has_over_limit"]["nullable"] is False
+    assert str(report_column_by_name["has_over_limit"]["default"]).strip("'\"") in {"0", "false", "FALSE"}
+    assert item_column_by_name["over_limit_status"]["nullable"] is False
+    assert str(item_column_by_name["over_limit_status"]["default"]).strip("'\"") == "not_checked"
+
+
+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 test_parse_process_order_accepts_numeric_process_names():
+    assert parse_process_order("1") == 1
+    assert parse_process_order("1序") == 1
+    assert parse_process_order(" 02 序 ") == 2
+
+
+def test_parse_process_order_rejects_special_and_text_process_names():
+    assert parse_process_order("杂活") is None
+    assert parse_process_order("清洗") is None
+    assert parse_process_order("第一序") is None
+
+
+def test_current_report_qty_uses_good_defect_and_scrap():
+    item = SimpleNamespace(good_qty=Decimal("90"), defect_qty=Decimal("5"), scrap_qty=Decimal("5"))
+
+    assert current_report_qty(item) == Decimal("100.00")
+
+
+def _report(report_id: int, status: ReportStatus = ReportStatus.pending) -> ProductionReport:
+    return ProductionReport(
+        id=report_id,
+        session_id=report_id,
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        report_date=datetime(2026, 7, 25).date(),
+        start_at=datetime(2026, 7, 25, 8),
+        end_at=datetime(2026, 7, 25, 10),
+        status=status,
+    )
+
+
+def _numeric_product(*, gross_weight: Decimal = Decimal("2")) -> Product:
+    return Product(
+        attendance_point_name="总厂",
+        project_no="P1",
+        product_name="产品A",
+        device_no="",
+        process_name="1",
+        product_gross_weight_kg=gross_weight,
+        standard_beat=1,
+        standard_workload=0,
+    )
+
+
+def _press_equipment() -> Equipment:
+    return Equipment(attendance_point_name="总厂", device_no="E1", device_type="冲压设备")
+
+
+def _numeric_session(*, status: SessionStatus = SessionStatus.reporting) -> WorkSession:
+    end_at = now() - timedelta(minutes=1)
+    start_at = end_at - timedelta(hours=2)
+    return WorkSession(
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        start_at=start_at,
+        end_at=end_at,
+        status=status,
+        devices=[
+            WorkSessionDevice(
+                attendance_point_name="总厂",
+                device_no="产品A",
+                process_name="1",
+                scanned_at=start_at,
+                sort_order=0,
+            )
+        ],
+    )
+
+
+def test_first_process_exceeds_material_limit_with_submit_snapshot(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert result.max_reportable_qty == Decimal("50.00")
+        assert result.current_report_qty == Decimal("60.00")
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert "1序累计报工数量超过本批次材料可支撑数量" in result.reason
+    finally:
+        db.close()
+
+
+def test_evaluate_defaults_to_excluding_current_report(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=50,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("50.00")
+        assert result.status != OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_current_report_sibling_items_contribute_to_current_cumulative(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        first_item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        second_item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no=" LOT-1 ",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, first_item, second_item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, first_item)
+
+        assert result.current_report_qty == Decimal("30.00")
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_history_batch_numbers_are_compared_after_trim(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no=" LOT-1 ",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        report = _report(2)
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, previous_report, previous_item, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_erp_weight_lookup_sql_error_has_distinct_uncheckable_reason():
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.over_limit_type == OVER_LIMIT_TYPE_UNCHECKABLE
+        assert result.reason == "读取批次投入重量失败，无法校验超额"
+    finally:
+        db.close()
+
+
+def test_evaluate_report_item_over_limit_does_not_write_snapshot_fields(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert item.over_limit_status == OVER_LIMIT_STATUS_NOT_CHECKED
+        assert item.over_limit_type is None
+        assert item.over_limit_reason is None
+    finally:
+        db.close()
+
+
+def test_later_process_exceeds_previous_good_quantity():
+    db = _sqlite_db()
+    try:
+        db.add_all(
+            [
+                Product(
+                    attendance_point_name="总厂",
+                    project_no="P1",
+                    product_name="产品A",
+                    device_no="",
+                    process_name="1",
+                    standard_beat=1,
+                ),
+                Product(
+                    attendance_point_name="总厂",
+                    project_no="P1",
+                    product_name="产品A",
+                    device_no="",
+                    process_name="3",
+                    standard_beat=1,
+                ),
+            ]
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=100,
+            defect_qty=10,
+            scrap_qty=5,
+            standard_beat=1,
+        )
+        report = _report(2)
+        report.employee_phone = "13800000001"
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="3",
+            good_qty=98,
+            defect_qty=3,
+            scrap_qty=2,
+            standard_beat=1,
+        )
+        db.add_all([previous_report, previous_item, report, item])
+        db.flush()
+
+        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert result.previous_good_qty == Decimal("100.00")
+        assert result.current_report_qty == Decimal("103.00")
+        assert result.current_cumulative_qty == Decimal("103.00")
+        assert "3序累计报工数量超过1序累计成品数量" in result.reason
+    finally:
+        db.close()
+
+
+def test_history_process_names_are_compared_after_trim(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name=" 01序 ",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        report = _report(2)
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, previous_report, previous_item, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_refresh_report_over_limit_snapshot_writes_item_and_report_summary(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no=" LOT-1 ",
+            process_name="1",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        refresh_report_over_limit_snapshot(
+            db,
+            report,
+            source="submit",
+            checked_at=datetime(2026, 7, 25, 10),
+        )
+
+        assert report.has_over_limit is True
+        assert report.over_limit_summary == "1项超额"
+        assert item.over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
+        assert item.over_limit_check_source == "submit"
+        assert item.over_limit_checked_at == datetime(2026, 7, 25, 10)
+        assert item.over_limit_batch_no == "LOT-1"
+        assert item.over_limit_max_reportable_qty == Decimal("50.00")
+    finally:
+        db.close()
+
+
+def test_refresh_report_over_limit_snapshot_skips_non_numeric_process(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="冲压",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="冲压",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        refresh_report_over_limit_snapshot(
+            db,
+            report,
+            source="submit",
+            checked_at=datetime(2026, 7, 25, 10),
+        )
+
+        assert report.has_over_limit is False
+        assert report.over_limit_summary is None
+        assert item.over_limit_status == OVER_LIMIT_STATUS_NOT_CHECKED
+        assert item.over_limit_type is None
+        assert item.over_limit_reason is None
+    finally:
+        db.close()
+
+
+def test_refresh_report_over_limit_snapshot_accepts_source_positional_argument(monkeypatch):
+    db = _sqlite_db()
+    try:
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="冲压",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([report, item])
+        db.flush()
+
+        refresh_report_over_limit_snapshot(db, report, "submit", checked_at=datetime(2026, 7, 25, 10))
+
+        assert item.over_limit_check_source == "submit"
+    finally:
+        db.close()
+
+
+def test_submit_report_refreshes_over_limit_snapshot(monkeypatch):
+    db = _sqlite_db()
+    try:
+        user = SimpleNamespace(phone="13800000000")
+        session = _numeric_session()
+        db.add_all([_press_equipment(), _numeric_product(), session])
+        db.commit()
+        monkeypatch.setattr(reports_router, "report_out", lambda report, schedule=None: report)
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        report = reports_router.submit_report(
+            ReportSubmitRequest(
+                session_id=session.id,
+                device_blocks=[
+                    ReportSubmitBlock(
+                        device_no="产品A",
+                        process_name="1",
+                        items=[
+                            ReportSubmitItem(
+                                device_no="E1",
+                                project_no="P1",
+                                product_name="产品A",
+                                process_name="1",
+                                standard_beat=1,
+                                raw_material_batch_no="LOT-1",
+                                good_qty=60,
+                            )
+                        ],
+                    )
+                ],
+            ),
+            user=user,
+            db=db,
+        )
+
+        assert report.has_over_limit is True
+        assert report.over_limit_summary == "1项超额"
+        assert report.items[0].over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
+        assert report.items[0].over_limit_check_source == "submit"
+        assert report.items[0].over_limit_current_report_qty == Decimal("60.00")
+    finally:
+        db.close()
+
+
+def test_approve_report_refreshes_over_limit_snapshot_after_correction(monkeypatch):
+    db = _sqlite_db()
+    try:
+        user = SimpleNamespace(phone="13900000000", role=Role.admin)
+        session = _numeric_session(status=SessionStatus.submitted)
+        report = ProductionReport(
+            attendance_point_name="总厂",
+            session=session,
+            employee_phone="13800000000",
+            report_date=date(2026, 7, 25),
+            start_at=session.start_at,
+            end_at=session.end_at,
+            duration_minutes=120,
+            effective_minutes=120,
+            total_good_qty=40,
+            total_output_qty=40,
+            actual_beat=180,
+            standard_beat=1,
+            expected_workload=0,
+            pace_rate=0,
+            workload_rate=0,
+            status=ReportStatus.pending,
+            submitted_at=session.end_at,
+            items=[
+                ProductionReportItem(
+                    attendance_point_name="总厂",
+                    device_no="E1",
+                    project_no="P1",
+                    product_name="产品A",
+                    raw_material_batch_no="LOT-1",
+                    process_name="1",
+                    standard_beat=1,
+                    standard_workload=0,
+                    good_qty=40,
+                    defect_qty=0,
+                    scrap_qty=0,
+                )
+            ],
+        )
+        db.add_all([_press_equipment(), _numeric_product(), report])
+        db.commit()
+        monkeypatch.setattr(reviews_router, "accessible_point_names", lambda _db, _user: ["总厂"])
+        monkeypatch.setattr(reviews_router, "report_out", lambda report, schedule=None: report)
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        saved = reviews_router.approve_report(
+            report.id,
+            ApproveReportRequest(
+                item_corrections=[
+                    ReportItemCorrection(id=report.items[0].id, good_qty=60)
+                ]
+            ),
+            user=user,
+            db=db,
+        )
+
+        assert saved.has_over_limit is True
+        assert saved.over_limit_summary == "1项超额"
+        assert saved.items[0].good_qty == 60
+        assert saved.items[0].over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
+        assert saved.items[0].over_limit_check_source == "review"
+        assert saved.items[0].over_limit_current_report_qty == Decimal("60.00")
+    finally:
+        db.close()
+
+
+def test_approve_report_without_over_limit_input_changes_preserves_submit_snapshot(monkeypatch):
+    db = _sqlite_db()
+    try:
+        user = SimpleNamespace(phone="13900000000", role=Role.admin)
+        session = _numeric_session(status=SessionStatus.submitted)
+        report = ProductionReport(
+            attendance_point_name="总厂",
+            session=session,
+            employee_phone="13800000000",
+            report_date=date(2026, 7, 25),
+            start_at=session.start_at,
+            end_at=session.end_at,
+            duration_minutes=120,
+            effective_minutes=120,
+            total_good_qty=60,
+            total_output_qty=60,
+            actual_beat=120,
+            standard_beat=1,
+            expected_workload=0,
+            pace_rate=0,
+            workload_rate=0,
+            status=ReportStatus.pending,
+            submitted_at=session.end_at,
+            has_over_limit=True,
+            over_limit_summary="1项超额",
+            items=[
+                ProductionReportItem(
+                    attendance_point_name="总厂",
+                    device_no="E1",
+                    project_no="P1",
+                    product_name="产品A",
+                    raw_material_batch_no="LOT-1",
+                    process_name="1",
+                    standard_beat=1,
+                    standard_workload=0,
+                    good_qty=60,
+                    defect_qty=0,
+                    scrap_qty=0,
+                    over_limit_status=OVER_LIMIT_STATUS_EXCEEDED,
+                    over_limit_type="first_process_material",
+                    over_limit_reason="提交时已超额",
+                    over_limit_checked_at=datetime(2026, 7, 25, 10),
+                    over_limit_check_source="submit",
+                    over_limit_batch_no="LOT-1",
+                    over_limit_product_gross_weight_kg=Decimal("2.0000"),
+                    over_limit_issued_weight_kg=Decimal("100.000000"),
+                    over_limit_max_reportable_qty=Decimal("50.00"),
+                    over_limit_current_cumulative_qty=Decimal("60.00"),
+                    over_limit_current_report_qty=Decimal("60.00"),
+                )
+            ],
+        )
+        db.add_all([_press_equipment(), _numeric_product(), report])
+        db.commit()
+        monkeypatch.setattr(reviews_router, "accessible_point_names", lambda _db, _user: ["总厂"])
+        monkeypatch.setattr(reviews_router, "report_out", lambda report, schedule=None: report)
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("200"),
+        )
+
+        saved = reviews_router.approve_report(
+            report.id,
+            ApproveReportRequest(),
+            user=user,
+            db=db,
+        )
+
+        assert saved.has_over_limit is True
+        assert saved.over_limit_summary == "1项超额"
+        assert saved.items[0].over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
+        assert saved.items[0].over_limit_check_source == "submit"
+        assert saved.items[0].over_limit_reason == "提交时已超额"
+        assert saved.items[0].over_limit_issued_weight_kg == Decimal("100.000000")
+        assert saved.items[0].over_limit_max_reportable_qty == Decimal("50.00")
+    finally:
+        db.close()
+
+
+def test_report_out_exposes_over_limit_snapshot_fields():
+    report = ProductionReport(
+        id=1,
+        session_id=1,
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        report_date=datetime(2026, 7, 25).date(),
+        start_at=datetime(2026, 7, 25, 8),
+        end_at=datetime(2026, 7, 25, 10),
+        status=ReportStatus.pending,
+        submitted_at=datetime(2026, 7, 25, 10),
+        has_over_limit=True,
+        over_limit_summary="1项超额",
+    )
+    item = ProductionReportItem(
+        id=1,
+        report=report,
+        attendance_point_name="总厂",
+        device_no="A",
+        project_no="P1",
+        product_name="产品A",
+        raw_material_batch_no="LOT-1",
+        process_name="1",
+        good_qty=60,
+        defect_qty=0,
+        scrap_qty=0,
+        standard_beat=1,
+        over_limit_status=OVER_LIMIT_STATUS_EXCEEDED,
+        over_limit_type="first_process_material",
+        over_limit_reason="1序累计报工数量超过本批次材料可支撑数量",
+        over_limit_checked_at=datetime(2026, 7, 25, 10),
+        over_limit_check_source="submit",
+        over_limit_batch_no="LOT-1",
+        over_limit_product_gross_weight_kg=Decimal("2.0000"),
+        over_limit_issued_weight_kg=Decimal("100.000000"),
+        over_limit_max_reportable_qty=Decimal("50.00"),
+        over_limit_previous_good_qty=None,
+        over_limit_current_cumulative_qty=Decimal("60.00"),
+        over_limit_current_report_qty=Decimal("60.00"),
+    )
+    report.items = [item]
+    report.allocations = []
+    report.audit_logs = []
+
+    output = report_out(report)
+
+    assert output.has_over_limit is True
+    assert output.over_limit_summary == "1项超额"
+    assert output.items[0].over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
+    assert output.items[0].over_limit_reason == "1序累计报工数量超过本批次材料可支撑数量"
+    assert output.items[0].over_limit_max_reportable_qty == 50

tests/test_usage_stats.py

@@ -1,9 +1,21 @@
+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,
@@ -13,11 +25,24 @@ 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))
 
@@ -37,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) == []
@@ -314,6 +426,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 +783,121 @@ 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()