app/models.py

@@ -13,7 +13,6 @@ from sqlalchemy import (
     Numeric,
     String,
     Text,
-    text,
 )
 from sqlalchemy.dialects.mysql import JSON
 from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -250,13 +249,6 @@ 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,
@@ -298,7 +290,6 @@ 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"),
     )
 
 
@@ -324,23 +315,6 @@ 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))
@@ -355,15 +329,6 @@ 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",
-        ),
     )
 
 

app/routers/dashboard.py

@@ -264,11 +264,6 @@ 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(),
             },
@@ -308,17 +303,6 @@ 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"]:
@@ -349,16 +333,6 @@ 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"],
@@ -406,9 +380,6 @@ 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/reports.py

@@ -41,7 +41,6 @@ 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
@@ -461,7 +460,6 @@ 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,7 +36,6 @@ 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
@@ -116,14 +115,6 @@ 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),
@@ -406,7 +397,6 @@ 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)
@@ -452,14 +442,11 @@ def approve_report(
             if correction.raw_material_batch_no is not None
             else item.raw_material_batch_no
         )
-        product_identity_changed = (
-            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
+            or next_project_no != item.project_no
+            or next_product_name != item.product_name
+            or next_process_name != item.process_name
         ):
             product = db.scalar(
                 select(Product).where(
@@ -493,13 +480,10 @@ 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)
@@ -514,15 +498,12 @@ 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:
@@ -551,8 +532,6 @@ 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/schemas.py

@@ -491,18 +491,6 @@ 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)
 
@@ -542,8 +530,6 @@ 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
@@ -658,9 +644,6 @@ 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,9 +133,6 @@ def export_dashboard_rows(rows: list) -> bytes:
         "操作人数",
         "工序单价",
         "参考工资",
-        "是否超额",
-        "超额摘要",
-        "超额原因",
         "报工标记",
         "校正项",
         "备注",
@@ -187,9 +184,6 @@ 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_over_limit.py

@@ -1,437 +0,0 @@
-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,10 +52,6 @@ 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,
@@ -127,6 +123,7 @@ 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,
@@ -135,10 +132,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,
@@ -413,18 +410,6 @@ 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 {},
     )
@@ -520,8 +505,6 @@ 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),

scripts/migrate_report_over_limit_snapshot.py

@@ -1,177 +0,0 @@
-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

@@ -156,237 +156,3 @@ 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_report_over_limit.py

@@ -1,945 +0,0 @@
-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