niuniu/JhHardwareWRS_BackPoint
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: niuniu:codex/report-time-allocation
Branches Tags
niuniu:main
niuniu:codex/report-time-allocation
..
compare: niuniu:main
Branches Tags
niuniu:codex/report-time-allocation
niuniu:main

No commits in common. "codex/report-time-allocation" and "main" have entirely different histories.
codex/repo ... main

24 changed files with 226 additions and 5752 deletions
Show Stats Expand all files Collapse all files

88
app/models.py
View File

@ -13,7 +13,6 @@ from sqlalchemy import (
Numeric, Numeric,
String, String,
Text, Text,
text,
) )
from sqlalchemy.dialects.mysql import JSON from sqlalchemy.dialects.mysql import JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship 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) 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) 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) 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( status: Mapped[ReportStatus] = mapped_column(
Enum(ReportStatus), Enum(ReportStatus),
nullable=False, nullable=False,
@ -285,10 +277,6 @@ class ProductionReport(Base):
back_populates="report", back_populates="report",
cascade="all, delete-orphan", cascade="all, delete-orphan",
) )
allocations: Mapped[list["ProductionReportAllocation"]] = relationship(
back_populates="report",
cascade="all, delete-orphan",
)
audit_logs: Mapped[list["ReportAuditLog"]] = relationship( audit_logs: Mapped[list["ReportAuditLog"]] = relationship(
order_by="ReportAuditLog.created_at", order_by="ReportAuditLog.created_at",
) )
@ -298,7 +286,6 @@ class ProductionReport(Base):
Index("idx_reports_status_date", "status", "report_date"), Index("idx_reports_status_date", "status", "report_date"),
Index("idx_reports_reviewer", "reviewer_phone", "reviewed_at"), Index("idx_reports_reviewer", "reviewer_phone", "reviewed_at"),
Index("idx_reports_voided", "is_voided", "voided_at"), Index("idx_reports_voided", "is_voided", "voided_at"),
Index("idx_reports_over_limit", "has_over_limit", "submitted_at"),
) )
@ -324,91 +311,16 @@ class ProductionReportItem(Base):
good_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) 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) 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) 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) allocated_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
started_at: Mapped[datetime | None] = mapped_column(DateTime) started_at: Mapped[datetime | None] = mapped_column(DateTime)
remark: Mapped[str | None] = mapped_column(String(500)) remark: Mapped[str | None] = mapped_column(String(500))
created_at: Mapped[datetime] = mapped_column(DateTime, default=now, nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now, nullable=False)
report: Mapped[ProductionReport] = relationship(back_populates="items") report: Mapped[ProductionReport] = relationship(back_populates="items")
allocations: Mapped[list["ProductionReportAllocation"]] = relationship(
back_populates="item",
cascade="all, delete-orphan",
)
__table_args__ = ( __table_args__ = (
Index("idx_report_items_report", "report_id"), Index("idx_report_items_report", "report_id"),
Index("idx_report_items_product", "project_no", "product_name"), 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",
),
)
class ProductionReportAllocation(Base):
__tablename__ = "production_report_allocations"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
report_id: Mapped[int] = mapped_column(
ForeignKey("production_reports.id", ondelete="CASCADE"),
nullable=False,
)
report_item_id: Mapped[int | None] = mapped_column(
ForeignKey("production_report_items.id", ondelete="CASCADE"),
)
attendance_point_name: Mapped[str] = mapped_column(String(128), nullable=False, default="")
employee_phone: Mapped[str] = mapped_column(String(20), nullable=False, default="")
allocation_date: Mapped[date] = mapped_column(Date, nullable=False)
day_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
overtime_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
night_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
effective_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0)
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)
changeover_count: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
reference_wage: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime, default=now, nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=now, onupdate=now, nullable=False)
report: Mapped[ProductionReport] = relationship(back_populates="allocations")
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"),
Index("idx_report_allocations_point_date", "attendance_point_name", "allocation_date"),
Index("idx_report_allocations_employee_date", "employee_phone", "allocation_date"),
) )

201
app/routers/dashboard.py
View File

@ -12,7 +12,6 @@ from app.models import (
Personnel, Personnel,
Product, Product,
ProductionReport, ProductionReport,
ProductionReportAllocation,
ProductionReportItem, ProductionReportItem,
ReportStatus, ReportStatus,
Role, Role,
@ -23,11 +22,12 @@ from app.services.cleaning import is_cleaning_item, is_cleaning_product
from app.services.common import as_float, round1, round2 from app.services.common import as_float, round1, round2
from app.services.continuous_die import is_continuous_die_item, is_continuous_die_product from app.services.continuous_die import is_continuous_die_item, is_continuous_die_product
from app.services.excel_export import export_dashboard_rows from app.services.excel_export import export_dashboard_rows
from app.services.metrics import shift_distribution_text from app.services.metrics import calculate_report_metrics, shift_distribution_text
from app.services.misc_work import is_misc_item, is_misc_product from app.services.misc_work import is_misc_item, is_misc_product
from app.services.multi_person import is_multi_person_item from app.services.multi_person import is_multi_person_item
from app.services.report_lifecycle import purge_expired_voided_reports from app.services.report_lifecycle import purge_expired_voided_reports
from app.services.serializers import report_correction_maps from app.services.serializers import report_correction_maps
from app.services.work_schedule import get_work_schedule_config
router = APIRouter(prefix="/api/dashboard", tags=["dashboard"]) router = APIRouter(prefix="/api/dashboard", tags=["dashboard"])
@ -123,85 +123,57 @@ def _dashboard_rows(
include_voided: bool = False, include_voided: bool = False,
) -> list[DashboardRow]: ) -> list[DashboardRow]:
query = ( query = (
select(ProductionReportAllocation) select(ProductionReport)
.join(ProductionReportAllocation.report)
.join(ProductionReportAllocation.item)
.options( .options(
selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee), selectinload(ProductionReport.employee),
selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.items), selectinload(ProductionReport.items),
selectinload(ProductionReportAllocation.report).selectinload( selectinload(ProductionReport.audit_logs),
ProductionReport.audit_logs selectinload(ProductionReport.session).selectinload(WorkSession.devices),
),
selectinload(ProductionReportAllocation.report)
.selectinload(ProductionReport.session)
.selectinload(WorkSession.devices),
selectinload(ProductionReportAllocation.item),
) )
.where(ProductionReport.status == ReportStatus.approved) .where(ProductionReport.status == ReportStatus.approved)
) )
if not include_voided: if not include_voided:
query = query.where(ProductionReport.is_voided.is_(False)) query = query.where(ProductionReport.is_voided.is_(False))
if start_date: if start_date:
query = query.where(ProductionReportAllocation.allocation_date >= start_date) query = query.where(ProductionReport.report_date >= start_date)
if end_date: if end_date:
query = query.where(ProductionReportAllocation.allocation_date <= end_date) query = query.where(ProductionReport.report_date <= end_date)
allocations = db.scalars(query).all() reports = db.scalars(query).all()
product_cache: dict[tuple[str, str, str, str, str], Product | None] = {}
for allocation in allocations:
item = allocation.item
if item is not None:
setattr(item, "_dashboard_product", _product_for_item(db, product_cache, item))
return _dashboard_rows_from_allocations(allocations)
def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
groups: dict[tuple, dict] = {} groups: dict[tuple, dict] = {}
correction_cache: dict[int, tuple[dict, dict[int, dict], dict[int, dict]]] = {} product_cache: dict[tuple[str, str, str, str, str], Product | None] = {}
for allocation in allocations: schedule_cache = {}
report = getattr(allocation, "report", None) for report in reports:
item = getattr(allocation, "item", None) if report.attendance_point_name not in schedule_cache:
if report is None or item is None: schedule_cache[report.attendance_point_name] = get_work_schedule_config(db, report.attendance_point_name)
continue schedule = schedule_cache[report.attendance_point_name]
report_id = getattr(report, "id", None) report_corrections, item_corrections, _device_corrections = report_correction_maps(report)
report_key = report_id if report_id is not None else id(report) report_metrics = calculate_report_metrics(
if report_key not in correction_cache: report.start_at,
correction_cache[report_key] = report_correction_maps(report) report.end_at,
report_corrections, item_corrections, _device_corrections = correction_cache[report_key] list(report.items),
devices=report.session.devices if report.session else None,
schedule=schedule,
)
item_count = max(1, len(report.items))
fallback_minutes = as_float(report.effective_minutes) / item_count
employee_name = report.employee.name if report.employee else "" employee_name = report.employee.name if report.employee else ""
worker_type = ( worker_type = "临时工" if (report.employee and report.employee.is_temporary) else "正式工"
"临时工" if (report.employee and report.employee.is_temporary) else "正式工" for item in report.items:
)
item_misc = is_misc_item(item) item_misc = is_misc_item(item)
process_name = ( process_name = (item.remark or item.process_name or "") if item_misc else (item.process_name or "")
(item.remark or item.process_name or "") if item_misc else (item.process_name or "") product = _product_for_item(db, product_cache, item)
)
product = getattr(item, "_dashboard_product", None)
is_cleaning = is_cleaning_item(item) or is_cleaning_product(product) is_cleaning = is_cleaning_item(item) or is_cleaning_product(product)
is_misc = item_misc or is_misc_product(product) is_misc = item_misc or is_misc_product(product)
is_continuous_die = is_continuous_die_item(item) or is_continuous_die_product(product) is_continuous_die = is_continuous_die_item(item) or is_continuous_die_product(product)
item_standard_beat = getattr(item, "standard_beat", 0) standard_beat = 0 if (is_cleaning or is_misc) else as_float(product.standard_beat if product else item.standard_beat)
standard_beat = ( process_unit_price = as_float(item.process_unit_price_yuan)
0 stamping_method = item.stamping_method or (product.stamping_method if product else None)
if (is_cleaning or is_misc)
else as_float(product.standard_beat if product else item_standard_beat)
)
process_unit_price = as_float(getattr(item, "process_unit_price_yuan", 0))
stamping_method = getattr(item, "stamping_method", None) or (
product.stamping_method if product else None
)
allocation_date = getattr(allocation, "allocation_date")
attendance_point_name = getattr(
report, "attendance_point_name", getattr(allocation, "attendance_point_name", "")
)
employee_phone = getattr(
report, "employee_phone", getattr(allocation, "employee_phone", "")
)
key = ( key = (
attendance_point_name, report.attendance_point_name,
allocation_date, report.report_date,
employee_name, employee_name,
employee_phone, report.employee_phone,
item.project_no, item.project_no,
item.product_name, item.product_name,
process_name, process_name,
@ -211,16 +183,15 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
key, key,
{ {
"id": "__".join(str(part) for part in key), "id": "__".join(str(part) for part in key),
"attendance_point_name": attendance_point_name, "attendance_point_name": report.attendance_point_name,
"report_date": allocation_date, "report_date": report.report_date,
"employee_phone": employee_phone, "employee_phone": report.employee_phone,
"employee_name": employee_name, "employee_name": employee_name,
"worker_type": worker_type, "worker_type": worker_type,
"project_no": item.project_no, "project_no": item.project_no,
"product_name": item.product_name, "product_name": item.product_name,
"process_name": process_name, "process_name": process_name,
"report_ids": set(), "report_ids": set(),
"source_report_dates": set(),
"effective_minutes": 0.0, "effective_minutes": 0.0,
"shift_day_minutes": 0.0, "shift_day_minutes": 0.0,
"shift_overtime_minutes": 0.0, "shift_overtime_minutes": 0.0,
@ -233,29 +204,13 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
"changeover_count": 0.0, "changeover_count": 0.0,
"reference_wage": 0.0, "reference_wage": 0.0,
"expected_workload": 0.0, "expected_workload": 0.0,
"profile_no": ( "profile_no": product.profile_no if product else None,
getattr(product, "profile_no", getattr(item, "profile_no", None)) "material_code": product.material_code if product else item.material_code,
if product "material_name": product.material_name if product else item.material_name,
else getattr(item, "profile_no", None)
),
"material_code": (
getattr(product, "material_code", getattr(item, "material_code", None))
if product
else getattr(item, "material_code", None)
),
"material_name": (
getattr(product, "material_name", getattr(item, "material_name", None))
if product
else getattr(item, "material_name", None)
),
"raw_material_batch_numbers": set(), "raw_material_batch_numbers": set(),
"device_no": item.device_no, "device_no": item.device_no,
"stamping_method": stamping_method, "stamping_method": stamping_method,
"operator_count": ( "operator_count": as_float(product.operator_count) if product else as_float(item.operator_count),
as_float(product.operator_count)
if product
else as_float(getattr(item, "operator_count", 1))
),
"is_continuous_die": is_continuous_die, "is_continuous_die": is_continuous_die,
"is_multi_person": is_multi_person_item(item), "is_multi_person": is_multi_person_item(item),
"is_multi_person_assistant": False, "is_multi_person_assistant": False,
@ -264,39 +219,25 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
"is_voided": False, "is_voided": False,
"review_remarks": [], "review_remarks": [],
"review_remark_keys": set(), "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_pairs": [],
"correction_pair_keys": set(), "correction_pair_keys": set(),
}, },
) )
_append_dashboard_corrections( _append_dashboard_corrections(group, report, item, report_corrections, item_corrections.get(item.id, {}))
group, good_qty = as_float(item.good_qty)
report, defect_qty = as_float(item.defect_qty) + as_float(item.scrap_qty)
item,
report_corrections,
item_corrections.get(getattr(item, "id", None), {}),
)
good_qty = as_float(getattr(allocation, "good_qty", 0))
defect_qty = as_float(getattr(allocation, "defect_qty", 0)) + as_float(
getattr(allocation, "scrap_qty", 0)
)
output_qty = good_qty + defect_qty output_qty = good_qty + defect_qty
effective_minutes = as_float(getattr(allocation, "effective_minutes", 0)) allocated_minutes = 0 if is_cleaning else (as_float(item.allocated_minutes) or fallback_minutes)
expected_workload = effective_minutes * 60 / standard_beat if standard_beat > 0 else 0 expected_workload = allocated_minutes * 60 / standard_beat if standard_beat > 0 else 0
group["report_ids"].add(report_key) group["report_ids"].add(report.id)
group["source_report_dates"].add(report.report_date) group["effective_minutes"] += allocated_minutes
group["effective_minutes"] += effective_minutes
group["expected_workload"] += expected_workload group["expected_workload"] += expected_workload
group["shift_day_minutes"] += as_float(getattr(allocation, "day_minutes", 0)) group["shift_day_minutes"] += as_float(getattr(item, "_shift_day_minutes", 0))
group["shift_overtime_minutes"] += as_float(getattr(allocation, "overtime_minutes", 0)) group["shift_overtime_minutes"] += as_float(getattr(item, "_shift_overtime_minutes", 0))
group["shift_night_minutes"] += as_float(getattr(allocation, "night_minutes", 0)) group["shift_night_minutes"] += as_float(getattr(item, "_shift_night_minutes", 0))
group["total_good_qty"] += good_qty group["total_good_qty"] += good_qty
group["reference_wage"] += as_float(getattr(allocation, "reference_wage", 0)) group["reference_wage"] += good_qty * process_unit_price
group["changeover_count"] += as_float(getattr(allocation, "changeover_count", 0)) group["changeover_count"] += as_float(item.changeover_count)
group["total_defect_qty"] += defect_qty group["total_defect_qty"] += defect_qty
group["total_output_qty"] += output_qty group["total_output_qty"] += output_qty
group["is_system_auto_submitted"] = ( group["is_system_auto_submitted"] = (
@ -308,25 +249,13 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
group["is_multi_person_assistant"] = ( group["is_multi_person_assistant"] = (
group["is_multi_person_assistant"] or bool(report.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: if report.review_remark:
review_remark = str(report.review_remark).strip() review_remark = str(report.review_remark).strip()
if review_remark and review_remark not in group["review_remark_keys"]: if review_remark and review_remark not in group["review_remark_keys"]:
group["review_remark_keys"].add(review_remark) group["review_remark_keys"].add(review_remark)
group["review_remarks"].append(review_remark) group["review_remarks"].append(review_remark)
raw_material_batch_no = getattr(item, "raw_material_batch_no", None) if item.raw_material_batch_no:
if raw_material_batch_no: group["raw_material_batch_numbers"].add(item.raw_material_batch_no)
group["raw_material_batch_numbers"].add(raw_material_batch_no)
rows: list[DashboardRow] = [] rows: list[DashboardRow] = []
for group in groups.values(): for group in groups.values():
@ -348,24 +277,11 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
"overtime": group["shift_overtime_minutes"], "overtime": group["shift_overtime_minutes"],
"night": group["shift_night_minutes"], "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( rows.append(
DashboardRow( DashboardRow(
id=group["id"], id=group["id"],
attendance_point_name=group["attendance_point_name"], attendance_point_name=group["attendance_point_name"],
report_date=group["report_date"], report_date=group["report_date"],
source_report_date=source_report_dates[-1] if source_report_dates else group["report_date"],
source_report_dates=source_report_dates,
employee_phone=group["employee_phone"], employee_phone=group["employee_phone"],
employee_name=group["employee_name"], employee_name=group["employee_name"],
worker_type=group["worker_type"], worker_type=group["worker_type"],
@ -406,9 +322,6 @@ def _dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]:
is_system_auto_submitted=group["is_system_auto_submitted"], is_system_auto_submitted=group["is_system_auto_submitted"],
is_voided=group["is_voided"], is_voided=group["is_voided"],
review_remark="；".join(group["review_remarks"]) or None, 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"], correction_pairs=group["correction_pairs"],
) )
) )

58
app/routers/reconciliation.py
View File

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

13
app/routers/reports.py
View File

@ -39,9 +39,7 @@ from app.services.display_names import mold_process_display_name
from app.services.metrics import calculate_report_metrics, minutes_between, release_at_caps_work_time from app.services.metrics import calculate_report_metrics, minutes_between, release_at_caps_work_time
from app.services.misc_work import is_misc_product from app.services.misc_work import is_misc_product
from app.services.mold_locks import release_session_locks 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_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.reporting_window import REPORTING_EXPIRED_DETAIL, is_reporting_expired
from app.services.serializers import equipment_option, product_option, report_out from app.services.serializers import equipment_option, product_option, report_out
from app.services.work_schedule import get_work_schedule_config from app.services.work_schedule import get_work_schedule_config
@ -75,8 +73,7 @@ def _cleaning_device_nos(item: CleaningReportSubmitItem) -> list[str]:
def _report_query(): def _report_query():
return select(ProductionReport).options( return select(ProductionReport).options(
selectinload(ProductionReport.allocations), selectinload(ProductionReport.items),
selectinload(ProductionReport.items).selectinload(ProductionReportItem.allocations),
selectinload(ProductionReport.audit_logs), selectinload(ProductionReport.audit_logs),
selectinload(ProductionReport.employee), selectinload(ProductionReport.employee),
selectinload(ProductionReport.reviewer), selectinload(ProductionReport.reviewer),
@ -459,9 +456,6 @@ def submit_report(
release_session_locks(session, user.phone, "报工提交释放占用", submitted_at) release_session_locks(session, user.phone, "报工提交释放占用", submitted_at)
session.status = SessionStatus.submitted session.status = SessionStatus.submitted
db.add(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() db.commit()
saved = _get_report_or_404(db, report.id) saved = _get_report_or_404(db, report.id)
@ -608,13 +602,10 @@ def submit_cleaning_report(
items=report_items, items=report_items,
) )
db.add(report) db.add(report)
schedule = get_work_schedule_config(db, point_name)
db.flush()
refresh_report_allocations(db, report, schedule=schedule, commit=False)
db.commit() db.commit()
saved = _get_report_or_404(db, report.id) saved = _get_report_or_404(db, report.id)
return report_out(saved, schedule) return report_out(saved, get_work_schedule_config(db, saved.attendance_point_name))
@router.get("/mine", response_model=PageResponse) @router.get("/mine", response_model=PageResponse)

32
app/routers/reviews.py
View File

@ -29,14 +29,12 @@ from app.services.continuous_die import is_continuous_die_item, is_continuous_di
from app.services.display_names import mold_process_display_name from app.services.display_names import mold_process_display_name
from app.services.metrics import calculate_report_metrics from app.services.metrics import calculate_report_metrics
from app.services.misc_work import is_misc_item, is_misc_product from app.services.misc_work import is_misc_item, is_misc_product
from app.services.report_allocations import refresh_report_allocations
from app.services.report_lifecycle import ( from app.services.report_lifecycle import (
can_edit_report, can_edit_report,
mark_report_voided, mark_report_voided,
purge_expired_voided_reports, purge_expired_voided_reports,
restore_voided_report, 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.serializers import report_out
from app.services.work_schedule import WorkScheduleConfig, get_work_schedule_config from app.services.work_schedule import WorkScheduleConfig, get_work_schedule_config
from app.timezone import now from app.timezone import now
@ -116,18 +114,9 @@ def _validate_normal_report_devices(db: Session, report: ProductionReport) -> No
raise HTTPException(status_code=400, detail=f"设备 {device_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(): def _report_query():
return select(ProductionReport).options( return select(ProductionReport).options(
selectinload(ProductionReport.allocations), selectinload(ProductionReport.items),
selectinload(ProductionReport.items).selectinload(ProductionReportItem.allocations),
selectinload(ProductionReport.audit_logs), selectinload(ProductionReport.audit_logs),
selectinload(ProductionReport.employee), selectinload(ProductionReport.employee),
selectinload(ProductionReport.reviewer), selectinload(ProductionReport.reviewer),
@ -406,7 +395,6 @@ def approve_report(
schedule = get_work_schedule_config(db, report.attendance_point_name) schedule = get_work_schedule_config(db, report.attendance_point_name)
item_map = {item.id: item for item in report.items} item_map = {item.id: item for item in report.items}
has_correction = False has_correction = False
over_limit_input_changed = False
next_start_at = _effective_review_time(payload.start_at, report.start_at) next_start_at = _effective_review_time(payload.start_at, report.start_at)
next_end_at = _effective_review_time(payload.end_at, report.end_at) next_end_at = _effective_review_time(payload.end_at, report.end_at)
@ -452,14 +440,11 @@ def approve_report(
if correction.raw_material_batch_no is not None if correction.raw_material_batch_no is not None
else item.raw_material_batch_no 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 ( if (
next_device_no != item.device_no 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( product = db.scalar(
select(Product).where( select(Product).where(
@ -493,13 +478,10 @@ def approve_report(
item.standard_beat = as_float(product.standard_beat) item.standard_beat = as_float(product.standard_beat)
item.standard_workload = 0 item.standard_workload = 0
has_correction = True has_correction = True
if product_identity_changed:
over_limit_input_changed = True
if next_raw_material_batch_no != item.raw_material_batch_no: if next_raw_material_batch_no != item.raw_material_batch_no:
item.raw_material_batch_no = next_raw_material_batch_no item.raw_material_batch_no = next_raw_material_batch_no
has_correction = True has_correction = True
over_limit_input_changed = True
if correction.changeover_count is not None: if correction.changeover_count is not None:
next_changeover_count = as_float(correction.changeover_count) next_changeover_count = as_float(correction.changeover_count)
@ -514,15 +496,12 @@ def approve_report(
if correction.good_qty is not None and as_float(correction.good_qty) != as_float(item.good_qty): if correction.good_qty is not None and as_float(correction.good_qty) != as_float(item.good_qty):
item.good_qty = correction.good_qty item.good_qty = correction.good_qty
has_correction = True 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): if correction.defect_qty is not None and as_float(correction.defect_qty) != as_float(item.defect_qty):
item.defect_qty = correction.defect_qty item.defect_qty = correction.defect_qty
has_correction = True 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): if correction.scrap_qty is not None and as_float(correction.scrap_qty) != as_float(item.scrap_qty):
item.scrap_qty = correction.scrap_qty item.scrap_qty = correction.scrap_qty
has_correction = True has_correction = True
over_limit_input_changed = True
if item_is_misc and correction.remark is not None: if item_is_misc and correction.remark is not None:
next_remark = str(correction.remark or "").strip() or None next_remark = str(correction.remark or "").strip() or None
if next_remark != item.remark: if next_remark != item.remark:
@ -549,10 +528,7 @@ def approve_report(
_validate_normal_report_devices(db, report) _validate_normal_report_devices(db, report)
_apply_metrics(report, schedule) _apply_metrics(report, schedule)
refresh_report_allocations(db, report, schedule=schedule, commit=False)
reviewed_at = now() 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 was_pending = report.status == ReportStatus.pending
report.status = ReportStatus.approved report.status = ReportStatus.approved
report.reviewer_phone = user.phone report.reviewer_phone = user.phone

202
app/routers/usage_stats.py
View File

@ -2,20 +2,12 @@ from datetime import date
from math import ceil from math import ceil
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import and_, select from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload from sqlalchemy.orm import Session, selectinload
from app.database import get_db from app.database import get_db
from app.deps import require_roles from app.deps import require_roles
from app.models import ( from app.models import Equipment, Personnel, ProductionReport, ReportStatus, Role
Equipment,
Personnel,
ProductionReport,
ProductionReportAllocation,
ProductionReportItem,
ReportStatus,
Role,
)
from app.schemas import ( from app.schemas import (
PageResponse, PageResponse,
UsageStatsDailyRow, UsageStatsDailyRow,
@ -26,7 +18,7 @@ from app.schemas import (
from app.services.attendance_points import accessible_point_names, require_attendance_point_access from app.services.attendance_points import accessible_point_names, require_attendance_point_access
from app.services.common import round2 from app.services.common import round2
from app.services.report_lifecycle import purge_expired_voided_reports from app.services.report_lifecycle import purge_expired_voided_reports
from app.services.usage_stats import UsageStatRow, build_usage_stats_from_allocations from app.services.usage_stats import UsageStatRow, build_usage_stats
from app.services.usage_stats_export import export_usage_stats_rows from app.services.usage_stats_export import export_usage_stats_rows
router = APIRouter(prefix="/api/usage-stats", tags=["usage-stats"]) router = APIRouter(prefix="/api/usage-stats", tags=["usage-stats"])
@ -55,42 +47,21 @@ def _equipment_type_map(db: Session, point_names: list[str]) -> dict[tuple[str,
return {(_clean(row.attendance_point_name), _clean(row.device_no)): row.device_type for row in rows} return {(_clean(row.attendance_point_name), _clean(row.device_no)): row.device_type for row in rows}
def _approved_allocations_query( def _approved_reports_query(point_names: list[str], start_date: date | None, end_date: date | None):
point_names: list[str],
start_date: date | None,
end_date: date | None,
):
query = ( query = (
select(ProductionReportAllocation) select(ProductionReport)
.join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id) .options(selectinload(ProductionReport.employee), selectinload(ProductionReport.items))
.join(
ProductionReportItem,
and_(
ProductionReportItem.id == ProductionReportAllocation.report_item_id,
ProductionReportItem.report_id == ProductionReportAllocation.report_id,
),
)
.options(
selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee),
selectinload(ProductionReportAllocation.item),
)
.where( .where(
ProductionReport.status == ReportStatus.approved, ProductionReport.status == ReportStatus.approved,
ProductionReport.is_voided.is_(False), ProductionReport.is_voided.is_(False),
ProductionReport.attendance_point_name.in_(point_names), ProductionReport.attendance_point_name.in_(point_names),
ProductionReportAllocation.attendance_point_name.in_(point_names),
ProductionReportItem.attendance_point_name.in_(point_names),
) )
) )
if start_date: if start_date:
query = query.where(ProductionReportAllocation.allocation_date >= start_date) query = query.where(ProductionReport.report_date >= start_date)
if end_date: if end_date:
query = query.where(ProductionReportAllocation.allocation_date <= end_date) query = query.where(ProductionReport.report_date <= end_date)
return query.order_by( return query.order_by(ProductionReport.report_date.desc(), ProductionReport.id.desc())
ProductionReportAllocation.allocation_date.desc(),
ProductionReportAllocation.report_id.desc(),
ProductionReportAllocation.id.desc(),
)
def _sort_usage_rows(rows: list[UsageStatRow], sort_by: str) -> list[UsageStatRow]: def _sort_usage_rows(rows: list[UsageStatRow], sort_by: str) -> list[UsageStatRow]:
@ -203,93 +174,6 @@ def _build_usage_stats_export_response(device_rows: list[UsageStatRow], mold_row
) )
def _build_usage_stats_detail_from_allocations(
*,
allocations: list,
category: str,
target_args: dict,
equipment_type_by_key: dict[tuple[str, str], str],
) -> UsageStatsDetailOut:
rows = build_usage_stats_from_allocations(
allocations=allocations,
category=category,
equipment_type_by_key=equipment_type_by_key,
)
target = next((row for row in rows if _matches_target(row, **target_args)), None)
if target is None:
return UsageStatsDetailOut(
id=target_args["id"],
category=category,
attendance_point_name=target_args["attendance_point_name"],
name=target_args["name"],
product_name=target_args["product_name"],
process_name=target_args["process_name"],
stamping_method=target_args["stamping_method"],
metric_kind=target_args["metric_kind"],
)
grouped_allocations: dict[tuple[date, int], list] = {}
group_reports: dict[tuple[date, int], object | None] = {}
for allocation in allocations:
allocation_date = getattr(allocation, "allocation_date")
report = getattr(allocation, "report", None)
report_id = getattr(allocation, "report_id", None)
if report_id is None and report is not None:
report_id = getattr(report, "id", None)
group_key = (allocation_date, report_id if report_id is not None else id(allocation))
grouped_allocations.setdefault(group_key, []).append(allocation)
group_reports.setdefault(group_key, report)
daily_rows: dict[date, UsageStatsDailyRow] = {}
report_rows: list[UsageStatsReportRow] = []
for (allocation_date, report_key), group in grouped_allocations.items():
group_matches = build_usage_stats_from_allocations(
allocations=group,
category=category,
equipment_type_by_key=equipment_type_by_key,
)
match = next((row for row in group_matches if _matches_target(row, **target_args)), None)
if match is None:
continue
day = daily_rows.setdefault(
allocation_date,
UsageStatsDailyRow(report_date=allocation_date),
)
day.value = round2(day.value + match.value)
day.report_count += match.report_count
report = group_reports[(allocation_date, report_key)]
employee = getattr(report, "employee", None) if report is not None else None
report_id = (
getattr(report, "id", None)
if report is not None
else getattr(group[0], "report_id", None)
)
employee_phone = getattr(report, "employee_phone", "") if report is not None else ""
report_rows.append(
UsageStatsReportRow(
report_id=report_id or 0,
report_date=allocation_date,
employee_phone=employee_phone,
employee_name=employee.name if employee else "",
display_name=target.name,
value=match.value,
report_count=match.report_count,
)
)
return _to_detail_out(
target,
daily_rows=sorted(daily_rows.values(), key=lambda row: row.report_date),
report_rows=sorted(
report_rows,
key=lambda row: (row.report_date, row.report_id),
reverse=True,
),
)
@router.get("/summary", response_model=PageResponse) @router.get("/summary", response_model=PageResponse)
def usage_stats_summary( def usage_stats_summary(
category: str = Query("device", pattern="^(device|mold)$"), category: str = Query("device", pattern="^(device|mold)$"),
@ -305,11 +189,11 @@ def usage_stats_summary(
) -> PageResponse: ) -> PageResponse:
purge_expired_voided_reports(db) purge_expired_voided_reports(db)
point_names = _requested_points(db, user, attendance_point_name) point_names = _requested_points(db, user, attendance_point_name)
allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
rows = _sort_usage_rows( rows = _sort_usage_rows(
_filter_usage_rows( _filter_usage_rows(
build_usage_stats_from_allocations( build_usage_stats(
allocations=allocations, reports=reports,
category=category, category=category,
equipment_type_by_key=_equipment_type_map(db, point_names), equipment_type_by_key=_equipment_type_map(db, point_names),
), ),
@ -343,11 +227,11 @@ def export_usage_stats_excel(
purge_expired_voided_reports(db) purge_expired_voided_reports(db)
point_names = _requested_points(db, user, attendance_point_name) point_names = _requested_points(db, user, attendance_point_name)
equipment_type_by_key = _equipment_type_map(db, point_names) equipment_type_by_key = _equipment_type_map(db, point_names)
allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
device_rows = _sort_usage_rows( device_rows = _sort_usage_rows(
_filter_usage_rows( _filter_usage_rows(
build_usage_stats_from_allocations( build_usage_stats(
allocations=allocations, reports=reports,
category="device", category="device",
equipment_type_by_key=equipment_type_by_key, equipment_type_by_key=equipment_type_by_key,
), ),
@ -357,8 +241,8 @@ def export_usage_stats_excel(
) )
mold_rows = _sort_usage_rows( mold_rows = _sort_usage_rows(
_filter_usage_rows( _filter_usage_rows(
build_usage_stats_from_allocations( build_usage_stats(
allocations=allocations, reports=reports,
category="mold", category="mold",
equipment_type_by_key=equipment_type_by_key, equipment_type_by_key=equipment_type_by_key,
), ),
@ -387,7 +271,7 @@ def usage_stats_detail(
purge_expired_voided_reports(db) purge_expired_voided_reports(db)
point_names = _requested_points(db, user, attendance_point_name) point_names = _requested_points(db, user, attendance_point_name)
equipment_type_by_key = _equipment_type_map(db, point_names) equipment_type_by_key = _equipment_type_map(db, point_names)
allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all() reports = db.scalars(_approved_reports_query(point_names, start_date, end_date)).all()
target_args = { target_args = {
"id": _clean(id), "id": _clean(id),
"category": category, "category": category,
@ -398,9 +282,53 @@ def usage_stats_detail(
"stamping_method": _clean(stamping_method), "stamping_method": _clean(stamping_method),
"metric_kind": _clean(metric_kind), "metric_kind": _clean(metric_kind),
} }
return _build_usage_stats_detail_from_allocations( rows = build_usage_stats(
allocations=allocations, reports=reports,
category=category, category=category,
target_args=target_args,
equipment_type_by_key=equipment_type_by_key, 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),
)

34
app/schemas.py
View File

@ -453,19 +453,6 @@ class ReportMetrics(BaseModel):
workload_rate: float workload_rate: float
class ReportAllocationOut(BaseModel):
allocation_date: date
day_minutes: float = 0
overtime_minutes: float = 0
night_minutes: float = 0
effective_minutes: float = 0
good_qty: float = 0
defect_qty: float = 0
scrap_qty: float = 0
changeover_count: float = 0
reference_wage: float = 0
class ReportItemOut(BaseModel): class ReportItemOut(BaseModel):
id: int | None = None id: int | None = None
attendance_point_name: str = "" attendance_point_name: str = ""
@ -491,19 +478,6 @@ class ReportItemOut(BaseModel):
is_continuous_die: bool = False is_continuous_die: bool = False
is_multi_person: bool = False is_multi_person: bool = False
remark: str | None = None 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) corrections: dict[str, Any] = Field(default_factory=dict)
@ -542,8 +516,6 @@ class ReportOut(BaseModel):
submitted_at: datetime submitted_at: datetime
is_system_auto_submitted: bool = False is_system_auto_submitted: bool = False
auto_submit_reason: str | None = None auto_submit_reason: str | None = None
has_over_limit: bool = False
over_limit_summary: str | None = None
is_multi_person_assistant: bool = False is_multi_person_assistant: bool = False
multi_person_source_report_id: int | None = None multi_person_source_report_id: int | None = None
is_voided: bool = False is_voided: bool = False
@ -559,7 +531,6 @@ class ReportOut(BaseModel):
items: list[ReportItemOut] items: list[ReportItemOut]
metrics: ReportMetrics metrics: ReportMetrics
result_text: str result_text: str
allocation_summary_text: str = ""
corrections: dict[str, Any] = Field(default_factory=dict) corrections: dict[str, Any] = Field(default_factory=dict)
@ -616,8 +587,6 @@ class DashboardRow(BaseModel):
id: str id: str
attendance_point_name: str = "" attendance_point_name: str = ""
report_date: date report_date: date
source_report_date: date | None = None
source_report_dates: list[date] = Field(default_factory=list)
employee_phone: str employee_phone: str
employee_name: str employee_name: str
project_no: str project_no: str
@ -658,9 +627,6 @@ class DashboardRow(BaseModel):
is_system_auto_submitted: bool = False is_system_auto_submitted: bool = False
is_voided: bool = False is_voided: bool = False
review_remark: str | None = None 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) correction_pairs: list[dict[str, Any]] = Field(default_factory=list)

3
app/services/auto_submit.py
View File

@ -18,7 +18,6 @@ from app.models import (
from app.services.metrics import calculate_report_metrics, minutes_between, release_at_caps_work_time from app.services.metrics import calculate_report_metrics, minutes_between, release_at_caps_work_time
from app.services.misc_work import is_misc_product from app.services.misc_work import is_misc_product
from app.services.mold_locks import release_session_locks from app.services.mold_locks import release_session_locks
from app.services.report_allocations import refresh_report_allocations
from app.services.work_schedule import DEFAULT_AUTO_SUBMIT_HOURS, get_work_schedule, get_work_schedule_config from app.services.work_schedule import DEFAULT_AUTO_SUBMIT_HOURS, get_work_schedule, get_work_schedule_config
from app.timezone import now from app.timezone import now
@ -172,8 +171,6 @@ def auto_submit_session(db: Session, session: WorkSession, submitted_at: datetim
items=items, items=items,
) )
db.add(report) db.add(report)
db.flush()
refresh_report_allocations(db, report, schedule=schedule, commit=False)
return report return report

25
app/services/excel_export.py
View File

@ -108,8 +108,7 @@ def export_dashboard_rows(rows: list) -> bytes:
void_fill = PatternFill(fill_type="solid", fgColor="FFD9D9D9") void_fill = PatternFill(fill_type="solid", fgColor="FFD9D9D9")
sheet.append([ sheet.append([
"考勤点", "考勤点",
"统计归属日期", "报工日期",
"原始报工日期",
"员工姓名", "员工姓名",
"员工电话", "员工电话",
"工种", "工种",
@ -133,22 +132,12 @@ def export_dashboard_rows(rows: list) -> bytes:
"操作人数", "操作人数",
"工序单价", "工序单价",
"参考工资", "参考工资",
"是否超额",
"超额摘要",
"超额原因",
"报工标记", "报工标记",
"校正项", "校正项",
"备注", "备注",
]) ])
wage_summary: dict[tuple[str, str, str], dict[str, float | str]] = {} wage_summary: dict[tuple[str, str, str], dict[str, float | str]] = {}
for row_index, row in enumerate(rows, start=2): for row_index, row in enumerate(rows, start=2):
source_dates = getattr(row, "source_report_dates", []) or [
getattr(row, "source_report_date", None) or row.report_date
]
source_dates = sorted({item for item in source_dates if item})
source_date_text = "、".join(
item.isoformat() for item in source_dates
)
report_flags = [] report_flags = []
if getattr(row, "is_voided", False): if getattr(row, "is_voided", False):
report_flags.append("作废") report_flags.append("作废")
@ -163,7 +152,6 @@ def export_dashboard_rows(rows: list) -> bytes:
sheet.append([ sheet.append([
row.attendance_point_name, row.attendance_point_name,
row.report_date.isoformat(), row.report_date.isoformat(),
source_date_text,
row.employee_name, row.employee_name,
row.employee_phone, row.employee_phone,
row.worker_type, row.worker_type,
@ -187,9 +175,6 @@ def export_dashboard_rows(rows: list) -> bytes:
row.operator_count, row.operator_count,
row.process_unit_price_yuan, row.process_unit_price_yuan,
row.reference_wage, 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), "、".join(report_flags),
correction_text, correction_text,
getattr(row, "review_remark", None), getattr(row, "review_remark", None),
@ -214,11 +199,9 @@ def export_dashboard_rows(rows: list) -> bytes:
total_good_qty = as_float(row.total_good_qty) total_good_qty = as_float(row.total_good_qty)
expected_workload = as_float(row.expected_workload) expected_workload = as_float(row.expected_workload)
if standard_beat > 0 and (actual_beat > standard_beat or actual_beat < standard_beat * 0.7): if standard_beat > 0 and (actual_beat > standard_beat or actual_beat < standard_beat * 0.7):
sheet.cell(row=row_index, column=15).fill = warn_fill sheet.cell(row=row_index, column=14).fill = warn_fill
if expected_workload > 0 and ( if expected_workload > 0 and (total_good_qty < expected_workload or total_good_qty > expected_workload * 1.3):
total_good_qty < expected_workload or total_good_qty > expected_workload * 1.3 sheet.cell(row=row_index, column=11).fill = warn_fill
):
sheet.cell(row=row_index, column=12).fill = warn_fill
if getattr(row, "is_voided", False): if getattr(row, "is_voided", False):
for cell in sheet[row_index]: for cell in sheet[row_index]:
cell.fill = void_fill cell.fill = void_fill

183
app/services/metrics.py
View File

@ -1,5 +1,4 @@
from collections import defaultdict from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta from datetime import date, datetime, time, timedelta
from typing import Protocol from typing import Protocol
@ -30,14 +29,6 @@ class ReportDeviceLike(Protocol):
sort_order: int sort_order: int
@dataclass(frozen=True)
class ItemTimeRange:
item: ReportItemLike
start_at: datetime
end_at: datetime
allocation_weight: float = 1.0
SHIFT_BUCKETS = ("day", "overtime", "night") SHIFT_BUCKETS = ("day", "overtime", "night")
SHIFT_LABELS = { SHIFT_LABELS = {
"day": "白班", "day": "白班",
@ -80,75 +71,6 @@ def _overlap_minutes(
return minutes_between(start, end) if end > start else 0.0 return minutes_between(start, end) if end > start else 0.0
def _clip_interval(
interval_start: datetime,
interval_end: datetime,
boundary_start: datetime,
boundary_end: datetime,
) -> tuple[datetime, datetime] | None:
start = max(interval_start, boundary_start)
end = min(interval_end, boundary_end)
if end <= start:
return None
return start, end
def _merge_intervals(intervals: list[tuple[datetime, datetime]]) -> list[tuple[datetime, datetime]]:
ordered = sorted(intervals, key=lambda item: item[0])
merged: list[tuple[datetime, datetime]] = []
for start, end in ordered:
if not merged or start > merged[-1][1]:
merged.append((start, end))
elif end > merged[-1][1]:
merged[-1] = (merged[-1][0], end)
return merged
def _subtract_intervals(
base_start: datetime,
base_end: datetime,
occupied: list[tuple[datetime, datetime]],
) -> list[tuple[datetime, datetime]]:
gaps: list[tuple[datetime, datetime]] = []
cursor = base_start
for start, end in _merge_intervals(occupied):
if start > cursor:
gaps.append((cursor, start))
if end > cursor:
cursor = end
if cursor < base_end:
gaps.append((cursor, base_end))
return gaps
def _calendar_dates_between(start_at: datetime, end_at: datetime):
current = start_at.date()
last = end_at.date()
while current <= last:
yield current
current += timedelta(days=1)
def _blank_overtime_intervals(
current_date: date,
schedule: WorkScheduleConfig,
) -> list[tuple[datetime, datetime]]:
day_start = datetime.combine(current_date, time.min)
day_end = day_start + timedelta(days=1)
occupied: list[tuple[datetime, datetime]] = []
raw_intervals = [
_schedule_interval(current_date, schedule.day_start, schedule.day_end),
_schedule_interval(current_date, schedule.overtime_start, schedule.overtime_end),
_schedule_interval(current_date, schedule.night_start, schedule.night_end),
_schedule_interval(current_date - timedelta(days=1), schedule.night_start, schedule.night_end),
]
for interval_start, interval_end in raw_intervals:
clipped = _clip_interval(interval_start, interval_end, day_start, day_end)
if clipped is not None:
occupied.append(clipped)
return _subtract_intervals(day_start, day_end, occupied)
def _covered_meal_overlap_minutes( def _covered_meal_overlap_minutes(
start_at: datetime, start_at: datetime,
end_at: datetime, end_at: datetime,
@ -221,21 +143,6 @@ def _raw_shift_minutes(
) )
) )
minutes[bucket] += max(0.0, raw_minutes - meal_minutes) minutes[bucket] += max(0.0, raw_minutes - meal_minutes)
for blank_start, blank_end in _blank_overtime_intervals(current_date, active_schedule):
raw_minutes = _overlap_minutes(start_at, end_at, blank_start, blank_end)
blank_start_at = max(start_at, blank_start)
blank_end_at = min(end_at, blank_end)
meal_minutes = sum(
_overlap_minutes(blank_start_at, blank_end_at, meal_start, meal_end)
for meal_start, meal_end in meal_intervals
if (
start_at <= meal_start
and end_at >= meal_end
and end_at > meal_end + timedelta(minutes=MEAL_END_GRACE_MINUTES)
)
)
minutes["overtime"] += max(0.0, raw_minutes - meal_minutes)
return minutes return minutes
@ -398,96 +305,6 @@ def _allocate_device_item_minutes(
) )
def _device_item_time_ranges(
device_items: list[ReportItemLike],
segment_start: datetime,
segment_end: datetime,
) -> list[ItemTimeRange]:
if not device_items:
return []
grouped_items: dict[datetime, list[ReportItemLike]] = defaultdict(list)
for item in device_items:
item_start = getattr(item, "started_at", None) or segment_start
grouped_items[_clamp_datetime(item_start, segment_start, segment_end)].append(item)
first_item_start = min(grouped_items)
if first_item_start > segment_start:
grouped_items[segment_start].extend(grouped_items.pop(first_item_start))
ranges: list[ItemTimeRange] = []
ordered_starts = sorted(grouped_items)
for index, item_start in enumerate(ordered_starts):
item_end = segment_end
if index + 1 < len(ordered_starts):
item_end = ordered_starts[index + 1]
items_at_start = grouped_items[item_start]
allocation_weight = 1 / len(items_at_start)
for item in items_at_start:
ranges.append(
ItemTimeRange(
item=item,
start_at=item_start,
end_at=item_end,
allocation_weight=allocation_weight,
)
)
return ranges
def item_time_ranges(
start_at: datetime,
end_at: datetime,
items: list[ReportItemLike],
devices: list[ReportDeviceLike] | None = None,
) -> list[ItemTimeRange]:
if not items or end_at <= start_at:
return []
segment_ranges = _device_segment_ranges(start_at, end_at, devices)
if not segment_ranges:
allocation_weight = 1 / len(items)
return [
ItemTimeRange(item=item, start_at=start_at, end_at=end_at, allocation_weight=allocation_weight)
for item in items
]
items_by_device: dict[str, list[ReportItemLike]] = defaultdict(list)
for item in items:
mold_key = _mold_key(
getattr(item, "product_name", "") or item.device_no,
getattr(item, "process_name", ""),
)
items_by_device[mold_key].append(item)
ranges: list[ItemTimeRange] = []
allocated_item_ids: set[int] = set()
unassigned_ranges: list[tuple[datetime, datetime]] = []
for device_no, device_ranges in segment_ranges.items():
device_items = items_by_device.get(device_no, [])
if not device_items:
unassigned_ranges.extend(device_ranges)
continue
for segment_start, segment_end in device_ranges:
ranges.extend(_device_item_time_ranges(device_items, segment_start, segment_end))
for item in device_items:
allocated_item_ids.add(id(item))
unmatched_items = [item for item in items if id(item) not in allocated_item_ids]
if unmatched_items and unassigned_ranges:
allocation_weight = 1 / len(unmatched_items)
for segment_start, segment_end in unassigned_ranges:
ranges.extend(
ItemTimeRange(
item=item,
start_at=segment_start,
end_at=segment_end,
allocation_weight=allocation_weight,
)
for item in unmatched_items
)
return ranges
def _allocate_item_minutes( def _allocate_item_minutes(
start_at: datetime, start_at: datetime,
end_at: datetime, end_at: datetime,

640
app/services/report_allocations.py
View File

@ -1,640 +0,0 @@
from collections import defaultdict
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, select
from sqlalchemy.orm import Session
from app.models import ProductionReport, ProductionReportAllocation
from app.services.common import as_float, round2
from app.services.metrics import (
SHIFT_BUCKETS,
SHIFT_LABELS,
_blank_overtime_intervals,
_device_item_time_ranges,
_device_segment_ranges,
_iter_shift_base_dates,
_mold_key,
_overlap_minutes,
_schedule_interval,
calculate_report_metrics,
format_hours,
item_time_ranges,
minutes_between,
)
from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig
SPLIT_QUANT = Decimal("0.01")
@dataclass(frozen=True)
class ReportAllocationDraft:
report_id: int | None
report_item_id: int | None
attendance_point_name: str
employee_phone: str
allocation_date: date
day_minutes: float = 0
overtime_minutes: float = 0
night_minutes: float = 0
effective_minutes: float = 0
good_qty: float = 0
defect_qty: float = 0
scrap_qty: float = 0
changeover_count: float = 0
reference_wage: float = 0
def _add_minutes(
rows: dict[date, dict[str, float]],
allocation_date: date,
bucket: str,
minutes: float,
) -> None:
if minutes <= 0:
return
rows[allocation_date][bucket] += minutes
def _range_allocation_minutes(
start_at: datetime,
end_at: datetime,
schedule: WorkScheduleConfig | None = None,
) -> dict[date, dict[str, float]]:
active_schedule = schedule or DEFAULT_WORK_SCHEDULE_CONFIG
rows: dict[date, dict[str, float]] = defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS})
if end_at <= start_at:
return rows
for current_date in _iter_shift_base_dates(start_at, end_at):
meal_intervals = [
_schedule_interval(current_date, active_schedule.lunch_start, active_schedule.lunch_end),
_schedule_interval(current_date, active_schedule.dinner_start, active_schedule.dinner_end),
]
shift_intervals = {
"day": [_schedule_interval(current_date, active_schedule.day_start, active_schedule.day_end)],
"overtime": [
_schedule_interval(current_date, active_schedule.overtime_start, active_schedule.overtime_end)
],
"night": [_schedule_interval(current_date, active_schedule.night_start, active_schedule.night_end)],
}
for bucket, intervals in shift_intervals.items():
for interval_start, interval_end in intervals:
overlap_start = max(start_at, interval_start)
overlap_end = min(end_at, interval_end)
if overlap_end <= overlap_start:
continue
meal_minutes = sum(
_overlap_minutes(overlap_start, overlap_end, meal_start, meal_end)
for meal_start, meal_end in meal_intervals
if (
start_at <= meal_start
and end_at >= meal_end
and end_at > meal_end + timedelta(minutes=5)
)
)
minutes = max(0.0, minutes_between(overlap_start, overlap_end) - meal_minutes)
if bucket == "night":
_add_minutes(rows, current_date, "night", minutes)
else:
_add_minutes(rows, overlap_start.date(), bucket, minutes)
for blank_start, blank_end in _blank_overtime_intervals(current_date, active_schedule):
overlap_start = max(start_at, blank_start)
overlap_end = min(end_at, blank_end)
if overlap_end <= overlap_start:
continue
meal_minutes = sum(
_overlap_minutes(overlap_start, overlap_end, meal_start, meal_end)
for meal_start, meal_end in meal_intervals
if (
start_at <= meal_start
and end_at >= meal_end
and end_at > meal_end + timedelta(minutes=5)
)
)
minutes = max(0.0, minutes_between(overlap_start, overlap_end) - meal_minutes)
_add_minutes(rows, overlap_start.date(), "overtime", minutes)
return rows
def _report_devices(report) -> list | None:
session = getattr(report, "session", None)
devices = getattr(session, "devices", None)
if devices is not None:
return list(devices)
devices = getattr(report, "devices", None)
return list(devices) if devices is not None else None
def _decimal_value(value: Any) -> Decimal:
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):
metrics = calculate_report_metrics(
getattr(report, "start_at"),
getattr(report, "end_at"),
items,
break_minutes=as_float(getattr(report, "break_minutes", 0)),
devices=_report_devices(report),
schedule=schedule,
)
return (
{id(item): as_float(getattr(item, "allocated_minutes", 0)) for item in items},
{
"day": as_float(metrics.get("shift_day_minutes", 0)),
"overtime": as_float(metrics.get("shift_overtime_minutes", 0)),
"night": as_float(metrics.get("shift_night_minutes", 0)),
},
)
def _minutes_total(rows: dict[date, dict[str, float]]) -> float:
return sum(sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) for minutes in rows.values())
def _empty_shift_minutes() -> dict[str, float]:
return {key: 0.0 for key in SHIFT_BUCKETS}
def _add_date_bucket_minutes(
target: dict[date, dict[str, float]],
source: dict[date, dict[str, float]],
*,
scale: float = 1.0,
) -> None:
for allocation_date, minutes in source.items():
row = target.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
row[key] += minutes.get(key, 0.0) * scale
def _bucket_totals(rows_by_date: dict[date, dict[str, float]]) -> dict[str, float]:
return {
key: sum(minutes.get(key, 0.0) for minutes in rows_by_date.values())
for key in SHIFT_BUCKETS
}
def _item_bucket_totals(minutes_by_item: dict[int, dict[date, dict[str, float]]]) -> dict[str, float]:
totals = _empty_shift_minutes()
for rows_by_date in minutes_by_item.values():
for key, value in _bucket_totals(rows_by_date).items():
totals[key] += value
return totals
def _allocate_unassigned_minutes_to_existing_items(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
unassigned_minutes: dict[date, dict[str, float]],
) -> None:
unassigned_total = _minutes_total(unassigned_minutes)
if unassigned_total <= 0 or not minutes_by_item:
return
raw_item_totals = {
item_id: _minutes_total(rows_by_date)
for item_id, rows_by_date in minutes_by_item.items()
}
raw_total = sum(raw_item_totals.values())
item_count = len(minutes_by_item)
for item_id, rows_by_date in minutes_by_item.items():
weight = raw_item_totals[item_id] / raw_total if raw_total > 0 else 1 / item_count
_add_date_bucket_minutes(rows_by_date, unassigned_minutes, scale=weight)
def _scale_item_minutes_to_bucket_targets(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
target_buckets: dict[str, float],
) -> None:
raw_buckets = _item_bucket_totals(minutes_by_item)
for key in SHIFT_BUCKETS:
raw_value = raw_buckets.get(key, 0.0)
target_value = target_buckets.get(key, 0.0)
if raw_value <= 0 or abs(raw_value - target_value) <= 0.000001:
continue
scale = target_value / raw_value
for rows_by_date in minutes_by_item.values():
for minutes in rows_by_date.values():
minutes[key] = minutes.get(key, 0.0) * scale
def _scale_bucket_minutes(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
bucket: str,
scale: float,
) -> None:
for rows_by_date in minutes_by_item.values():
for minutes in rows_by_date.values():
minutes[bucket] = minutes.get(bucket, 0.0) * scale
def _balance_minutes_to_targets(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
target_minutes_by_item: dict[int, float],
target_buckets: dict[str, float],
) -> None:
for _ in range(100):
for item_id, rows_by_date in minutes_by_item.items():
raw_total = _minutes_total(rows_by_date)
target_total = target_minutes_by_item.get(item_id, raw_total)
if raw_total <= 0:
continue
scale = target_total / raw_total
if abs(scale - 1) <= 0.000000001:
continue
for minutes in rows_by_date.values():
for key in SHIFT_BUCKETS:
minutes[key] = minutes.get(key, 0.0) * scale
raw_buckets = _item_bucket_totals(minutes_by_item)
for key in SHIFT_BUCKETS:
raw_value = raw_buckets.get(key, 0.0)
target_value = target_buckets.get(key, 0.0)
if raw_value <= 0:
continue
scale = target_value / raw_value
if abs(scale - 1) <= 0.000000001:
continue
_scale_bucket_minutes(minutes_by_item, key, scale)
def _round_bucket_cells_to_targets(
minutes_by_item: dict[int, dict[date, dict[str, float]]],
target_buckets: dict[str, float],
item_order: dict[int, int],
) -> None:
for bucket in SHIFT_BUCKETS:
cells: list[tuple[float, int, int, date]] = []
for item_id, rows_by_date in minutes_by_item.items():
for allocation_date, minutes in rows_by_date.items():
value = minutes.get(bucket, 0.0)
if value > 0:
cents_value = max(0.0, value * 100)
floor_cents = int(cents_value + 0.000000001)
cells.append((cents_value - floor_cents, floor_cents, item_order.get(item_id, 0), allocation_date))
minutes[bucket] = floor_cents / 100
else:
minutes[bucket] = 0.0
target_cents = int(round(round2(target_buckets.get(bucket, 0.0)) * 100))
floor_total = sum(cell[1] for cell in cells)
residual = target_cents - floor_total
if not cells or residual == 0:
continue
ordered = sorted(
cells,
key=lambda cell: (-cell[0], cell[2], cell[3].isoformat()),
)
if residual < 0:
ordered = sorted(
cells,
key=lambda cell: (cell[0], cell[2], cell[3].isoformat()),
)
cents_by_cell: dict[tuple[int, date], int] = {
(cell[2], cell[3]): cell[1] for cell in cells
}
steps = abs(residual)
for index in range(steps):
_, _, item_index, allocation_date = ordered[index % len(ordered)]
key = (item_index, allocation_date)
if residual > 0:
cents_by_cell[key] += 1
elif cents_by_cell[key] > 0:
cents_by_cell[key] -= 1
item_id_by_order = {order: item_id for item_id, order in item_order.items()}
for (item_index, allocation_date), cents in cents_by_cell.items():
item_id = item_id_by_order[item_index]
minutes_by_item[item_id][allocation_date][bucket] = cents / 100
def _items_by_device(items: list) -> dict[str, list]:
grouped: dict[str, list] = defaultdict(list)
for item in items:
mold_key = _mold_key(
getattr(item, "product_name", "") or item.device_no,
getattr(item, "process_name", ""),
)
grouped[mold_key].append(item)
return grouped
def _device_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None):
start_at = getattr(report, "start_at")
end_at = getattr(report, "end_at")
segment_ranges = _device_segment_ranges(start_at, end_at, _report_devices(report))
if not segment_ranges:
return None
items_by_device = _items_by_device(items)
minutes_by_item: dict[int, dict[date, dict[str, float]]] = {}
target_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes)
unassigned_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes)
allocated_item_ids: set[int] = set()
for device_no, ranges in segment_ranges.items():
device_items = items_by_device.get(device_no, [])
for segment_start, segment_end in ranges:
segment_minutes = _range_allocation_minutes(segment_start, segment_end, schedule)
_add_date_bucket_minutes(target_minutes, segment_minutes)
if not device_items:
_add_date_bucket_minutes(unassigned_minutes, segment_minutes)
continue
for item_range in _device_item_time_ranges(device_items, segment_start, segment_end):
item_minutes = minutes_by_item.setdefault(id(item_range.item), defaultdict(_empty_shift_minutes))
range_minutes = _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule)
_add_date_bucket_minutes(
item_minutes,
range_minutes,
scale=as_float(getattr(item_range, "allocation_weight", 1.0)),
)
for item in device_items:
allocated_item_ids.add(id(item))
unmatched_items = [item for item in items if id(item) not in allocated_item_ids]
if unmatched_items and _minutes_total(unassigned_minutes) > 0:
target_buckets = _bucket_totals(target_minutes)
matched_buckets = _item_bucket_totals(minutes_by_item)
unassigned_buckets = _bucket_totals(unassigned_minutes)
for unmatched_item in unmatched_items:
item_minutes = minutes_by_item.setdefault(id(unmatched_item), defaultdict(_empty_shift_minutes))
for allocation_date, minutes in unassigned_minutes.items():
row = item_minutes.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
remaining = max(0.0, target_buckets.get(key, 0.0) - matched_buckets.get(key, 0.0))
raw_unassigned = unassigned_buckets.get(key, 0.0)
if remaining <= 0 or raw_unassigned <= 0:
continue
row[key] += minutes.get(key, 0.0) * (remaining / raw_unassigned) / len(unmatched_items)
return minutes_by_item
_allocate_unassigned_minutes_to_existing_items(minutes_by_item, unassigned_minutes)
_scale_item_minutes_to_bucket_targets(minutes_by_item, _bucket_totals(target_minutes))
return minutes_by_item
def _range_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None):
ranges = item_time_ranges(getattr(report, "start_at"), getattr(report, "end_at"), items, _report_devices(report))
minutes_by_item: dict[int, dict[date, dict[str, float]]] = {}
for item_range in ranges:
item_key = id(item_range.item)
target = minutes_by_item.setdefault(item_key, defaultdict(_empty_shift_minutes))
for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items():
row = target.setdefault(allocation_date, _empty_shift_minutes())
for key in SHIFT_BUCKETS:
row[key] += minutes.get(key, 0.0) * as_float(
getattr(item_range, "allocation_weight", 1.0)
)
return minutes_by_item
def _fallback_row(report, item) -> ReportAllocationDraft:
return ReportAllocationDraft(
report_id=getattr(report, "id", None),
report_item_id=getattr(item, "id", None),
attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""),
employee_phone=str(getattr(report, "employee_phone", "") or ""),
allocation_date=getattr(report, "report_date"),
good_qty=_round_split_value(getattr(item, "good_qty", 0)),
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 _draft_from_minutes(report, item, allocation_date: date, minutes: dict[str, float], ratio: float) -> ReportAllocationDraft:
reference_wage = _item_reference_wage(item)
day_minutes = round2(minutes.get("day", 0))
overtime_minutes = round2(minutes.get("overtime", 0))
night_minutes = round2(minutes.get("night", 0))
return ReportAllocationDraft(
report_id=getattr(report, "id", None),
report_item_id=getattr(item, "id", None),
attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""),
employee_phone=str(getattr(report, "employee_phone", "") or ""),
allocation_date=allocation_date,
day_minutes=day_minutes,
overtime_minutes=overtime_minutes,
night_minutes=night_minutes,
effective_minutes=round2(day_minutes + overtime_minutes + night_minutes),
good_qty=_scaled_split_value(getattr(item, "good_qty", 0), ratio),
defect_qty=_scaled_split_value(getattr(item, "defect_qty", 0), ratio),
scrap_qty=_scaled_split_value(getattr(item, "scrap_qty", 0), ratio),
changeover_count=_scaled_split_value(getattr(item, "changeover_count", 0), 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:
return []
minutes_by_item = _device_minutes_by_item(report, items, schedule)
if minutes_by_item is None:
minutes_by_item = _range_minutes_by_item(report, items, schedule)
item_targets, bucket_targets = _allocation_targets(report, items, schedule)
_balance_minutes_to_targets(minutes_by_item, item_targets, bucket_targets)
_round_bucket_cells_to_targets(
minutes_by_item,
bucket_targets,
{id(item): index for index, item in enumerate(items)},
)
rows: list[ReportAllocationDraft] = []
for item in items:
item_minutes = minutes_by_item.get(id(item), {})
total_effective = sum(sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) for minutes in item_minutes.values())
if total_effective <= 0:
rows.append(_fallback_row(report, item))
continue
item_rows: list[ReportAllocationDraft] = []
for allocation_date in sorted(item_minutes):
minutes = item_minutes[allocation_date]
effective = sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS)
if effective <= 0:
continue
item_rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective))
rows.extend(_balance_item_split_residuals(item_rows, _item_split_targets(item)))
return rows
def allocation_summary_text(rows) -> str:
grouped: dict[date, dict[str, float]] = defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS})
for row in rows:
allocation_date = getattr(row, "allocation_date")
grouped[allocation_date]["day"] += as_float(getattr(row, "day_minutes", 0))
grouped[allocation_date]["overtime"] += as_float(getattr(row, "overtime_minutes", 0))
grouped[allocation_date]["night"] += as_float(getattr(row, "night_minutes", 0))
parts: list[str] = []
for allocation_date in sorted(grouped):
shift_text = "、".join(
f"{SHIFT_LABELS[key]}{format_hours(grouped[allocation_date].get(key, 0))}小时"
for key in SHIFT_BUCKETS
if round2(grouped[allocation_date].get(key, 0)) > 0
)
if shift_text:
parts.append(f"{allocation_date.isoformat()} {shift_text}")
return "；".join(parts)
def refresh_report_allocations(
db: Session,
report,
schedule: WorkScheduleConfig | None = None,
*,
commit: bool = False,
) -> list[ProductionReportAllocation]:
"""Rebuild allocation rows with metrics semantics, updating item allocation-derived fields."""
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,
report_item_id=draft.report_item_id,
attendance_point_name=draft.attendance_point_name,
employee_phone=draft.employee_phone,
allocation_date=draft.allocation_date,
day_minutes=draft.day_minutes,
overtime_minutes=draft.overtime_minutes,
night_minutes=draft.night_minutes,
effective_minutes=draft.effective_minutes,
good_qty=draft.good_qty,
defect_qty=draft.defect_qty,
scrap_qty=draft.scrap_qty,
changeover_count=draft.changeover_count,
reference_wage=draft.reference_wage,
)
for draft in build_report_allocation_drafts(report, schedule)
]
db.add_all(rows)
db.flush()
if commit:
db.commit()
return rows

2
app/services/report_lifecycle.py
View File

@ -8,7 +8,6 @@ from app.models import (
MoldLockFeedback, MoldLockFeedback,
Personnel, Personnel,
ProductionReport, ProductionReport,
ProductionReportAllocation,
ProductionReportItem, ProductionReportItem,
ReportAuditLog, ReportAuditLog,
WorkSession, WorkSession,
@ -87,7 +86,6 @@ def purge_expired_voided_reports(db: Session) -> int:
) )
) )
db.execute(delete(ReportAuditLog).where(ReportAuditLog.report_id == report.id)) db.execute(delete(ReportAuditLog).where(ReportAuditLog.report_id == report.id))
db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report.id))
db.execute(delete(ProductionReportItem).where(ProductionReportItem.report_id == report.id)) db.execute(delete(ProductionReportItem).where(ProductionReportItem.report_id == report.id))
db.delete(report) db.delete(report)
if session is not None: if session is not None:

437
app/services/report_over_limit.py
View File

@ -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()

77
app/services/serializers.py
View File

@ -1,9 +1,7 @@
from datetime import date, datetime from datetime import datetime
from sqlalchemy import inspect, select from sqlalchemy import select
from sqlalchemy.exc import NoInspectionAvailable
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import NO_VALUE
from app.models import ( from app.models import (
AttendancePoint, AttendancePoint,
@ -25,7 +23,6 @@ from app.schemas import (
PersonnelOut, PersonnelOut,
ProductOption, ProductOption,
ProductOut, ProductOut,
ReportAllocationOut,
ReportDeviceSegmentOut, ReportDeviceSegmentOut,
ReportItemOut, ReportItemOut,
ReportMetrics, ReportMetrics,
@ -39,7 +36,6 @@ from app.services.metrics import build_result_text, calculate_report_metrics
from app.services.misc_work import is_misc_item, is_misc_product from app.services.misc_work import is_misc_item, is_misc_product
from app.services.multi_person import is_multi_person_item, is_multi_person_product from app.services.multi_person import is_multi_person_item, is_multi_person_product
from app.services.process_names import export_process_name from app.services.process_names import export_process_name
from app.services.report_allocations import allocation_summary_text
from app.services.report_lifecycle import can_edit_report, can_unvoid_report from app.services.report_lifecycle import can_edit_report, can_unvoid_report
from app.services.work_schedule import WorkScheduleConfig from app.services.work_schedule import WorkScheduleConfig
from app.timezone import now from app.timezone import now
@ -52,10 +48,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: def attendance_point_out(point: AttendancePoint) -> AttendancePointOut:
return AttendancePointOut( return AttendancePointOut(
name=point.name, name=point.name,
@ -127,6 +119,7 @@ def personnel_summary_out(person: Personnel) -> dict:
def product_out(product: Product) -> ProductOut: def product_out(product: Product) -> ProductOut:
optional_float = lambda value: None if value is None else as_float(value)
return ProductOut( return ProductOut(
attendance_point_name=product.attendance_point_name, attendance_point_name=product.attendance_point_name,
project_no=product.project_no, project_no=product.project_no,
@ -135,10 +128,10 @@ def product_out(product: Product) -> ProductOut:
material_code=product.material_code, material_code=product.material_code,
material_name=product.material_name, material_name=product.material_name,
supplier=product.supplier, 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, device_no=product.device_no,
process_name=product.process_name, process_name=product.process_name,
stamping_method=product.stamping_method, stamping_method=product.stamping_method,
@ -348,46 +341,7 @@ def report_device_segment_out(
) )
def report_allocation_out(row) -> ReportAllocationOut:
return ReportAllocationOut(
allocation_date=row.allocation_date,
day_minutes=as_float(row.day_minutes),
overtime_minutes=as_float(row.overtime_minutes),
night_minutes=as_float(row.night_minutes),
effective_minutes=as_float(row.effective_minutes),
good_qty=as_float(row.good_qty),
defect_qty=as_float(row.defect_qty),
scrap_qty=as_float(row.scrap_qty),
changeover_count=as_float(row.changeover_count),
reference_wage=as_float(row.reference_wage),
)
def _loaded_relationship_rows(obj, relationship_name: str) -> list:
try:
state = inspect(obj)
except NoInspectionAvailable:
return list(getattr(obj, relationship_name, []) or [])
if relationship_name not in state.attrs:
return list(getattr(obj, relationship_name, []) or [])
relationship_state = state.attrs[relationship_name]
if relationship_state.loaded_value is NO_VALUE:
return []
return list(relationship_state.value or [])
def _sorted_allocation_rows(rows) -> list:
return sorted(
list(rows or []),
key=lambda row: (
getattr(row, "allocation_date", None) or date.min,
getattr(row, "id", None) or 0,
),
)
def report_item_out(item: ProductionReportItem, corrections: dict | None = None) -> ReportItemOut: def report_item_out(item: ProductionReportItem, corrections: dict | None = None) -> ReportItemOut:
allocation_rows = _sorted_allocation_rows(_loaded_relationship_rows(item, "allocations"))
return ReportItemOut( return ReportItemOut(
id=item.id, id=item.id,
attendance_point_name=item.attendance_point_name, attendance_point_name=item.attendance_point_name,
@ -413,19 +367,6 @@ def report_item_out(item: ProductionReportItem, corrections: dict | None = None)
is_continuous_die=is_continuous_die_item(item), is_continuous_die=is_continuous_die_item(item),
is_multi_person=is_multi_person_item(item), is_multi_person=is_multi_person_item(item),
remark=item.remark, 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 {}, corrections=corrections or {},
) )
@ -464,7 +405,6 @@ def _cleaning_report_metrics(report: ProductionReport) -> dict:
def report_out(report: ProductionReport, schedule: WorkScheduleConfig | None = None) -> ReportOut: def report_out(report: ProductionReport, schedule: WorkScheduleConfig | None = None) -> ReportOut:
allocation_rows = _sorted_allocation_rows(_loaded_relationship_rows(report, "allocations"))
is_cleaning_report = _report_is_cleaning(report) is_cleaning_report = _report_is_cleaning(report)
calculated_metrics = ( calculated_metrics = (
_cleaning_report_metrics(report) _cleaning_report_metrics(report)
@ -520,8 +460,6 @@ def report_out(report: ProductionReport, schedule: WorkScheduleConfig | None = N
submitted_at=report.submitted_at, submitted_at=report.submitted_at,
is_system_auto_submitted=bool(report.is_system_auto_submitted), is_system_auto_submitted=bool(report.is_system_auto_submitted),
auto_submit_reason=report.auto_submit_reason, 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), is_multi_person_assistant=bool(report.is_multi_person_assistant),
multi_person_source_report_id=report.multi_person_source_report_id, multi_person_source_report_id=report.multi_person_source_report_id,
is_voided=bool(report.is_voided), is_voided=bool(report.is_voided),
@ -548,6 +486,5 @@ def report_out(report: ProductionReport, schedule: WorkScheduleConfig | None = N
if is_cleaning_report if is_cleaning_report
else ("处理杂活已提交，等待管理员审核" if _report_is_misc_only(report) else build_result_text(metrics)) else ("处理杂活已提交，等待管理员审核" if _report_is_misc_only(report) else build_result_text(metrics))
), ),
allocation_summary_text=allocation_summary_text(allocation_rows),
corrections=report_corrections, corrections=report_corrections,
) )

49
app/services/usage_stats.py
View File

@ -1,7 +1,6 @@
import json import json
import re import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from types import SimpleNamespace
from typing import Any from typing import Any
from app.services.cleaning import is_cleaning_item from app.services.cleaning import is_cleaning_item
@ -34,7 +33,6 @@ class UsageStatRow:
object_type: str object_type: str
value: float = 0 value: float = 0
report_count: int = 0 report_count: int = 0
report_ids: set[int] = field(default_factory=set, repr=False)
tags: list[str] = field(default_factory=list) tags: list[str] = field(default_factory=list)
attendance_point_name: str = "" attendance_point_name: str = ""
product_name: str = "" product_name: str = ""
@ -60,7 +58,6 @@ class UsageStatAccumulator:
product_name: str = "", product_name: str = "",
process_name: str = "", process_name: str = "",
stamping_method: str = "", stamping_method: str = "",
report_id: int | None = None,
) -> None: ) -> None:
row = self._rows.get(key) row = self._rows.get(key)
if row is None: if row is None:
@ -89,10 +86,6 @@ class UsageStatAccumulator:
row.object_type = object_type row.object_type = object_type
row.value += as_float(value) 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 row.report_count += 1
for tag in tags or []: for tag in tags or []:
if tag and tag not in row.tags: if tag and tag not in row.tags:
@ -126,42 +119,12 @@ def build_usage_stats(
raise ValueError("category must be device or mold") 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( def _build_device_usage_stats(
reports: list, reports: list,
equipment_type_by_key: dict[tuple[str, str], str], equipment_type_by_key: dict[tuple[str, str], str],
) -> list[UsageStatRow]: ) -> list[UsageStatRow]:
acc = UsageStatAccumulator(category="device") acc = UsageStatAccumulator(category="device")
for item, report_id in _iter_report_items(reports): for item in _iter_report_items(reports):
if is_misc_item(item): if is_misc_item(item):
continue continue
@ -181,7 +144,6 @@ def _build_device_usage_stats(
value=value, value=value,
tags=[TAG_CLEANING], tags=[TAG_CLEANING],
attendance_point_name=point_name, attendance_point_name=point_name,
report_id=report_id,
) )
continue continue
@ -195,14 +157,13 @@ def _build_device_usage_stats(
object_type=equipment_type_by_key.get((point_name, device_no), DEVICE_TYPE_STAMPING), object_type=equipment_type_by_key.get((point_name, device_no), DEVICE_TYPE_STAMPING),
value=as_float(getattr(item, "allocated_minutes", 0)), value=as_float(getattr(item, "allocated_minutes", 0)),
attendance_point_name=point_name, attendance_point_name=point_name,
report_id=report_id,
) )
return acc.rows() return acc.rows()
def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]: def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]:
acc = UsageStatAccumulator(category="mold") acc = UsageStatAccumulator(category="mold")
for item, report_id in _iter_report_items(reports): for item in _iter_report_items(reports):
if is_misc_item(item): if is_misc_item(item):
continue continue
@ -226,17 +187,13 @@ def _build_mold_usage_stats(reports: list) -> list[UsageStatRow]:
product_name=product_name, product_name=product_name,
process_name=process_name, process_name=process_name,
stamping_method=stamping_method, stamping_method=stamping_method,
report_id=report_id,
) )
return acc.rows() return acc.rows()
def _iter_report_items(reports: list): def _iter_report_items(reports: list):
for report in reports: for report in reports:
report_id = getattr(report, "id", None) yield from getattr(report, "items", []) or []
for item in getattr(report, "items", []) or []:
item_report_id = getattr(item, "report_id", None)
yield item, item_report_id if item_report_id is not None else report_id
def _item_tags(item) -> list[str]: def _item_tags(item) -> list[str]:

531
scripts/migrate_report_allocations.py
View File

@ -1,531 +0,0 @@
from pathlib import Path
import re
import sys
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 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"
IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
REQUIRED_COLUMNS = {
"id": {"column_type": "bigint", "is_nullable": "NO", "extra_contains": ("auto_increment",)},
"report_id": {"column_type": "bigint", "is_nullable": "NO"},
"report_item_id": {"column_type": "bigint", "is_nullable": "YES"},
"attendance_point_name": {
"column_type": "varchar(128)",
"is_nullable": "NO",
"default": "empty_string",
},
"employee_phone": {"column_type": "varchar(20)", "is_nullable": "NO", "default": "empty_string"},
"allocation_date": {"column_type": "date", "is_nullable": "NO"},
"day_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"overtime_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"night_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"effective_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"good_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"defect_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"scrap_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"changeover_count": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"reference_wage": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"created_at": {
"column_type": "datetime",
"is_nullable": "NO",
"default": "current_timestamp",
},
"updated_at": {
"column_type": "datetime",
"is_nullable": "NO",
"default": "current_timestamp",
"extra_contains": ("on update current_timestamp",),
},
}
REQUIRED_INDEXES = {
"uq_report_allocations_report_item_date": {
"columns": ("report_id", "report_item_id", "allocation_date"),
"non_unique": 0,
},
"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"),
"report_item_id": ("production_report_items", "id", "CASCADE"),
}
IndexMetadata = dict[str, tuple[str, ...] | int]
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) -> 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) -> dict[str, dict[str, str | None]]:
rows = conn.execute(
text(
"""
SELECT column_name, column_type, is_nullable, column_default, extra
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = :table_name
"""
),
{"table_name": TABLE_NAME},
).all()
return {
row[0]: {
"column_type": row[1],
"is_nullable": row[2],
"column_default": row[3],
"extra": row[4],
}
for row in rows
}
def existing_indexes(conn) -> dict[str, IndexMetadata]:
rows = conn.execute(
text(
"""
SELECT index_name, non_unique, 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, dict[str, object]] = {}
for index_name, non_unique, column_name in rows:
metadata = indexes.setdefault(index_name, {"columns": [], "non_unique": int(non_unique)})
metadata["columns"].append(column_name)
return {
index_name: {
"columns": tuple(metadata["columns"]),
"non_unique": metadata["non_unique"],
}
for index_name, metadata in indexes.items()
}
def existing_foreign_keys(conn) -> dict[str, tuple[str, str, str]]:
rows = conn.execute(
text(
"""
SELECT k.column_name, k.referenced_table_name, k.referenced_column_name, r.delete_rule
FROM information_schema.key_column_usage k
JOIN information_schema.referential_constraints r
ON r.constraint_schema = k.constraint_schema
AND r.constraint_name = k.constraint_name
WHERE k.table_schema = DATABASE()
AND k.table_name = :table_name
AND k.referenced_table_name IS NOT NULL
ORDER BY k.constraint_name, k.ordinal_position
"""
),
{"table_name": TABLE_NAME},
).all()
return {row[0]: (row[1], row[2], row[3]) for row in rows}
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"] == non_unique
):
return index_name
return None
def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None:
if index_name not in REQUIRED_INDEXES or required_index_columns(index_name) != columns:
raise ValueError(f"unexpected index definition: {index_name} {columns}")
non_unique = required_index_non_unique(index_name)
indexes = existing_indexes(conn)
if (
index_name in indexes
and indexes[index_name]["columns"] == columns
and indexes[index_name]["non_unique"] == non_unique
):
return
if index_name in indexes:
conn.execute(
text(
f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
f"DROP INDEX {quote_identifier(index_name)}"
)
)
indexes = existing_indexes(conn)
equivalent_index = find_equivalent_index(indexes, columns, non_unique)
if equivalent_index:
conn.execute(
text(
f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
f"RENAME INDEX {quote_identifier(equivalent_index)} TO {quote_identifier(index_name)}"
)
)
return
column_sql = ", ".join(quote_identifier(column) for column in columns)
unique_sql = "UNIQUE " if non_unique == 0 else ""
conn.execute(
text(
f"CREATE {unique_sql}INDEX {quote_identifier(index_name)} "
f"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 = [
index_name
for index_name, metadata in indexes.items()
if (
metadata["non_unique"] == 1
and (
(metadata["columns"] == ("report_id",) and index_name != "idx_report_allocations_report")
or (
metadata["columns"] == ("report_item_id",)
and index_name != "idx_report_allocations_item"
)
)
)
]
for index_name in duplicates:
conn.execute(
text(
f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
f"DROP INDEX {quote_identifier(index_name)}"
)
)
def ensure_schema(conn) -> None:
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS production_report_allocations (
id BIGINT NOT NULL AUTO_INCREMENT,
report_id BIGINT NOT NULL,
report_item_id BIGINT NULL,
attendance_point_name VARCHAR(128) NOT NULL DEFAULT '',
employee_phone VARCHAR(20) NOT NULL DEFAULT '',
allocation_date DATE NOT NULL,
day_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
overtime_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
night_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
effective_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
good_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
defect_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
scrap_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
changeover_count DECIMAL(12, 2) NOT NULL DEFAULT 0,
reference_wage DECIMAL(12, 2) NOT NULL DEFAULT 0,
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),
KEY idx_report_allocations_point_date (attendance_point_name, allocation_date),
KEY idx_report_allocations_employee_date (employee_phone, allocation_date),
CONSTRAINT fk_report_allocations_report
FOREIGN KEY (report_id) REFERENCES production_reports (id)
ON DELETE CASCADE,
CONSTRAINT fk_report_allocations_item
FOREIGN KEY (report_item_id) REFERENCES production_report_items (id)
ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
"""
)
)
cleanup_duplicate_allocation_keys(conn)
for index_name in REQUIRED_INDEXES:
create_index(conn, index_name, required_index_columns(index_name))
cleanup_duplicate_fk_indexes(conn)
def normalized_default(value: str | None) -> str | None:
if value is None:
return None
lowered = str(value).strip().lower()
if lowered in {"current_timestamp()", "current_timestamp"}:
return "current_timestamp"
try:
return "zero" if float(lowered) == 0 else lowered
except ValueError:
return lowered
def column_matches(column: dict[str, str | None], expected: dict[str, object]) -> list[str]:
mismatches: list[str] = []
actual_type = str(column["column_type"]).lower()
actual_nullable = str(column["is_nullable"]).upper()
actual_default = normalized_default(column["column_default"])
actual_extra = str(column["extra"] or "").lower()
if actual_type != expected["column_type"]:
mismatches.append(f"column_type={column['column_type']}")
if actual_nullable != expected["is_nullable"]:
mismatches.append(f"is_nullable={column['is_nullable']}")
expected_default = expected.get("default")
if expected_default == "empty_string" and actual_default != "":
mismatches.append(f"column_default={column['column_default']!r}")
elif expected_default == "zero" and actual_default != "zero":
mismatches.append(f"column_default={column['column_default']!r}")
elif expected_default == "current_timestamp" and actual_default != "current_timestamp":
mismatches.append(f"column_default={column['column_default']!r}")
for extra_text in expected.get("extra_contains", ()):
if extra_text not in actual_extra:
mismatches.append(f"extra={column['extra']!r}")
return mismatches
def verify_schema(conn) -> None:
if not table_exists(conn):
raise RuntimeError(f"{TABLE_NAME} table is missing")
columns = existing_columns(conn)
missing_columns = sorted(set(REQUIRED_COLUMNS) - set(columns))
if missing_columns:
raise RuntimeError(f"{TABLE_NAME} missing columns: {', '.join(missing_columns)}")
incompatible_columns = [
f"{column_name}({', '.join(mismatches)})"
for column_name, expected in REQUIRED_COLUMNS.items()
if (mismatches := column_matches(columns[column_name], expected))
]
if incompatible_columns:
raise RuntimeError(f"{TABLE_NAME} incompatible columns: {', '.join(incompatible_columns)}")
indexes = existing_indexes(conn)
missing_indexes = [index_name for index_name in REQUIRED_INDEXES if index_name not in indexes]
if missing_indexes:
raise RuntimeError(f"{TABLE_NAME} missing indexes: {', '.join(missing_indexes)}")
wrong_indexes = [
f"{index_name}({', '.join(indexes[index_name]['columns'])})"
for index_name in REQUIRED_INDEXES
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)}")
wrong_index_uniqueness = [
index_name
for index_name in REQUIRED_INDEXES
if indexes[index_name]["non_unique"] != required_index_non_unique(index_name)
]
if wrong_index_uniqueness:
raise RuntimeError(
f"{TABLE_NAME} incompatible index uniqueness: {', '.join(wrong_index_uniqueness)}"
)
duplicate_single_column_indexes = [
index_name
for index_name, metadata in indexes.items()
if (
metadata["non_unique"] == 1
and (
(metadata["columns"] == ("report_id",) and index_name != "idx_report_allocations_report")
or (
metadata["columns"] == ("report_item_id",)
and index_name != "idx_report_allocations_item"
)
)
)
]
if duplicate_single_column_indexes:
raise RuntimeError(
f"{TABLE_NAME} duplicate FK indexes: {', '.join(sorted(duplicate_single_column_indexes))}"
)
unique_single_column_indexes = [
index_name
for index_name, metadata in indexes.items()
if (
metadata["non_unique"] != 1
and (
metadata["columns"] == ("report_id",)
or metadata["columns"] == ("report_item_id",)
)
)
]
if unique_single_column_indexes:
raise RuntimeError(
f"{TABLE_NAME} unique FK indexes are incompatible: {', '.join(sorted(unique_single_column_indexes))}"
)
foreign_keys = existing_foreign_keys(conn)
missing_foreign_keys = [
column
for column, definition in REQUIRED_FOREIGN_KEYS.items()
if foreign_keys.get(column) != definition
]
if missing_foreign_keys:
raise RuntimeError(f"{TABLE_NAME} missing foreign keys: {', '.join(missing_foreign_keys)}")
def _read_work_schedule_config(db, attendance_point_name: str | None) -> WorkScheduleConfig:
point_name = str(attendance_point_name or "").strip()
if not point_name:
return DEFAULT_WORK_SCHEDULE_CONFIG
point = db.get(AttendancePoint, point_name)
if point is None:
return DEFAULT_WORK_SCHEDULE_CONFIG
return WorkScheduleConfig(
day_start=point.day_start or DEFAULT_WORK_SCHEDULE_CONFIG.day_start,
day_end=point.day_end or DEFAULT_WORK_SCHEDULE_CONFIG.day_end,
lunch_start=point.lunch_start or DEFAULT_WORK_SCHEDULE_CONFIG.lunch_start,
lunch_end=point.lunch_end or DEFAULT_WORK_SCHEDULE_CONFIG.lunch_end,
dinner_start=point.dinner_start or DEFAULT_WORK_SCHEDULE_CONFIG.dinner_start,
dinner_end=point.dinner_end or DEFAULT_WORK_SCHEDULE_CONFIG.dinner_end,
overtime_start=point.overtime_start or DEFAULT_WORK_SCHEDULE_CONFIG.overtime_start,
overtime_end=point.overtime_end or DEFAULT_WORK_SCHEDULE_CONFIG.overtime_end,
night_start=point.night_start or DEFAULT_WORK_SCHEDULE_CONFIG.night_start,
night_end=point.night_end or DEFAULT_WORK_SCHEDULE_CONFIG.night_end,
)
def _backfill(batch_size: int = 200) -> tuple[int, int]:
processed = 0
skipped = 0
last_id = 0
with SessionLocal() as db:
while True:
query = (
select(ProductionReport)
.where(ProductionReport.id > last_id)
.options(
selectinload(ProductionReport.items),
selectinload(ProductionReport.session).selectinload(WorkSession.devices),
)
.order_by(ProductionReport.id.asc())
.limit(batch_size)
)
reports = db.scalars(query).all()
if not reports:
break
for report in reports:
if report is None:
skipped += 1
continue
last_id = report.id
schedule = _read_work_schedule_config(db, report.attendance_point_name)
refresh_report_allocations(db, report, schedule=schedule, commit=False)
processed += 1
db.commit()
return processed, skipped
def main() -> None:
with engine.begin() as conn:
ensure_schema(conn)
verify_schema(conn)
processed, skipped = _backfill()
print(f"report allocations migrated, processed={processed}, skipped={skipped}")
if __name__ == "__main__":
main()

177
scripts/migrate_report_over_limit_snapshot.py
View File

@ -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()

392
tests/test_dashboard_allocations.py
View File

@ -1,392 +0,0 @@
from datetime import date
from io import BytesIO
from types import SimpleNamespace
from openpyxl import load_workbook
from app.routers import dashboard
from app.services.excel_export import export_dashboard_rows
def test_dashboard_rows_from_allocations_groups_by_allocation_date_desc():
report_date = date(2026, 7, 25)
report = SimpleNamespace(
id=1,
attendance_point_name="总厂",
report_date=report_date,
employee_phone="13800000000",
employee=SimpleNamespace(name="张三", is_temporary=False),
items=[],
audit_logs=[],
session=None,
is_system_auto_submitted=False,
is_voided=False,
is_multi_person_assistant=False,
review_remark="复核通过",
)
item = SimpleNamespace(
id=10,
attendance_point_name="总厂",
device_no="A-01",
project_no="P1",
product_name="23#",
process_name="冲压",
remark=None,
raw_material_batch_no="BATCH-1",
material_code="M-1",
material_name="铝材",
profile_no="PF-1",
stamping_method="单冲",
operator_count=1,
process_unit_price_yuan=2,
standard_beat=1,
good_qty=650,
defect_qty=0,
scrap_qty=0,
changeover_count=2,
allocated_minutes=650,
)
report.items = [item]
allocations = [
SimpleNamespace(
report=report,
item=item,
allocation_date=date(2026, 7, 24),
night_minutes=350,
overtime_minutes=0,
day_minutes=0,
effective_minutes=350,
good_qty=350,
defect_qty=0,
scrap_qty=0,
changeover_count=1,
reference_wage=700,
),
SimpleNamespace(
report=report,
item=item,
allocation_date=date(2026, 7, 25),
night_minutes=0,
overtime_minutes=60,
day_minutes=240,
effective_minutes=300,
good_qty=300,
defect_qty=0,
scrap_qty=0,
changeover_count=1,
reference_wage=600,
),
]
rows = dashboard._dashboard_rows_from_allocations(allocations)
assert [row.report_date for row in rows] == [date(2026, 7, 25), date(2026, 7, 24)]
assert [row.source_report_dates for row in rows] == [[report_date], [report_date]]
assert [row.source_report_date for row in rows] == [report_date, report_date]
assert [row.report_count for row in rows] == [1, 1]
current, previous = rows
assert current.effective_minutes == 300
assert current.shift_day_minutes == 240
assert current.shift_overtime_minutes == 60
assert current.shift_night_minutes == 0
assert current.total_good_qty == 300
assert current.total_defect_qty == 0
assert current.total_output_qty == 300
assert current.changeover_count == 1
assert current.reference_wage == 600
assert previous.effective_minutes == 350
assert previous.shift_day_minutes == 0
assert previous.shift_overtime_minutes == 0
assert previous.shift_night_minutes == 350
assert previous.total_good_qty == 350
assert previous.total_defect_qty == 0
assert previous.total_output_qty == 350
assert previous.changeover_count == 1
assert previous.reference_wage == 700
def test_export_dashboard_rows_writes_allocation_and_source_dates_with_warning_columns():
row = SimpleNamespace(
attendance_point_name="总厂",
report_date=date(2026, 7, 24),
source_report_dates=[date(2026, 7, 25), date(2026, 7, 23)],
source_report_date=None,
employee_name="张三",
employee_phone="13800000000",
worker_type="正式工",
project_no="P1",
product_name="23#",
process_name="冲压",
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,
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["报工看板"]
assert [sheet.cell(row=1, column=index).value for index in range(1, 5)] == [
"考勤点",
"统计归属日期",
"原始报工日期",
"员工姓名",
]
assert sheet.cell(row=2, column=2).value == "2026-07-24"
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

123
tests/test_dashboard_source_dates.py
View File

@ -1,123 +0,0 @@
from datetime import date, datetime
from sqlalchemy import BigInteger, create_engine
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import (
AttendancePoint,
Personnel,
Product,
ProductionReport,
ProductionReportAllocation,
ProductionReportItem,
ReportStatus,
SessionStatus,
WorkSession,
)
from app.routers.dashboard import _dashboard_rows
@compiles(BigInteger, "sqlite")
def _compile_big_integer_for_sqlite(type_, compiler, **kw):
return "INTEGER"
def _db_session():
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine, future=True)
return SessionLocal()
def test_dashboard_rows_populate_source_report_dates_from_original_reports():
db = _db_session()
try:
point_name = "总厂"
report_date = date(2026, 7, 25)
person = Personnel(phone="13800000000", name="张三")
product = Product(
attendance_point_name=point_name,
project_no="P1",
product_name="23#",
device_no="",
process_name="冲压",
operator_count=1,
process_unit_price_yuan=2,
standard_beat=1,
standard_workload=0,
)
db.add_all([
AttendancePoint(name=point_name),
person,
product,
])
for index, start_hour in enumerate([8, 10], start=1):
session = WorkSession(
id=index,
attendance_point_name=point_name,
employee_phone=person.phone,
start_at=datetime(2026, 7, 25, start_hour, 0),
end_at=datetime(2026, 7, 25, start_hour + 1, 0),
status=SessionStatus.submitted,
)
report = ProductionReport(
id=index,
session_id=session.id,
attendance_point_name=point_name,
employee_phone=person.phone,
report_date=report_date,
start_at=session.start_at,
end_at=session.end_at,
duration_minutes=60,
break_minutes=0,
effective_minutes=60,
status=ReportStatus.approved,
submitted_at=session.end_at,
)
item = ProductionReportItem(
id=index,
report_id=report.id,
attendance_point_name=point_name,
device_no="23#",
project_no=product.project_no,
product_name=product.product_name,
process_name=product.process_name,
operator_count=1,
process_unit_price_yuan=2,
standard_beat=1,
standard_workload=0,
good_qty=10,
defect_qty=0,
scrap_qty=0,
allocated_minutes=60,
)
allocation = ProductionReportAllocation(
id=index,
report_id=report.id,
report_item_id=item.id,
attendance_point_name=point_name,
employee_phone=person.phone,
allocation_date=report_date,
day_minutes=60,
overtime_minutes=0,
night_minutes=0,
effective_minutes=60,
good_qty=10,
defect_qty=0,
scrap_qty=0,
changeover_count=0,
reference_wage=20,
)
db.add_all([session, report, item, allocation])
db.commit()
rows = _dashboard_rows(db, report_date, report_date)
assert len(rows) == 1
assert rows[0].report_count == 2
assert rows[0].source_report_date == report_date
assert rows[0].source_report_dates == [report_date]
finally:
db.close()

43
tests/test_metrics.py
View File

@ -280,46 +280,3 @@ def test_admin_forced_release_still_caps_work_time():
assert metrics["duration_minutes"] == 60 assert metrics["duration_minutes"] == 60
assert metrics["effective_minutes"] == 60 assert metrics["effective_minutes"] == 60
assert item.allocated_minutes == 60 assert item.allocated_minutes == 60
def test_morning_blank_interval_counts_as_overtime():
start_at = datetime(2026, 7, 25, 6, 0, 0)
end_at = datetime(2026, 7, 25, 8, 0, 0)
item = Item(good_qty=10)
metrics = calculate_report_metrics(start_at, end_at, [item])
assert metrics["duration_minutes"] == 120
assert metrics["effective_minutes"] == 120
assert metrics["shift_day_minutes"] == 0
assert metrics["shift_overtime_minutes"] == 120
assert metrics["shift_night_minutes"] == 0
assert item.allocated_minutes == 120
def test_dinner_blank_interval_counts_as_overtime():
start_at = datetime(2026, 7, 25, 17, 20, 0)
end_at = datetime(2026, 7, 25, 18, 0, 0)
item = Item(good_qty=10)
metrics = calculate_report_metrics(start_at, end_at, [item])
assert metrics["duration_minutes"] == 40
assert metrics["effective_minutes"] == 40
assert metrics["shift_day_minutes"] == 0
assert metrics["shift_overtime_minutes"] == 40
assert metrics["shift_night_minutes"] == 0
assert item.allocated_minutes == 40
def test_midnight_to_night_end_remains_night_not_blank_overtime():
start_at = datetime(2026, 7, 25, 0, 0, 0)
end_at = datetime(2026, 7, 25, 6, 0, 0)
item = Item(good_qty=10)
metrics = calculate_report_metrics(start_at, end_at, [item])
assert metrics["effective_minutes"] == 360
assert metrics["shift_night_minutes"] == 360
assert metrics["shift_overtime_minutes"] == 0
assert metrics["shift_day_minutes"] == 0

223
tests/test_reconciliation_allocations.py
View File

@ -1,223 +0,0 @@
from datetime import date, datetime
from sqlalchemy import BigInteger, create_engine
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import (
ProductionReport,
ProductionReportAllocation,
ProductionReportItem,
ReportStatus,
)
from app.routers import reconciliation
@compiles(BigInteger, "sqlite")
def _compile_big_integer_for_sqlite(type_, compiler, **kw):
return "INTEGER"
def _sqlite_db():
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine, future=True)
return SessionLocal()
def _report(
*,
id: int,
status: ReportStatus = ReportStatus.approved,
is_voided: bool = False,
):
return ProductionReport(
id=id,
session_id=id,
attendance_point_name="嘉恒",
employee_phone="13800000000",
report_date=date(2026, 7, 31),
start_at=datetime(2026, 7, 31, 20, 0),
end_at=datetime(2026, 8, 1, 2, 0),
duration_minutes=360,
break_minutes=0,
effective_minutes=360,
total_good_qty=650,
total_output_qty=650,
actual_beat=1,
standard_beat=1,
expected_workload=0,
pace_rate=0,
workload_rate=0,
status=status,
is_voided=is_voided,
submitted_at=datetime(2026, 8, 1, 2, 1),
)
def _item(
*,
id: int,
report_id: int,
process_name: str = "2序",
):
return ProductionReportItem(
id=id,
report_id=report_id,
attendance_point_name="嘉恒",
device_no="A-01",
project_no="P1",
product_name="产品A",
process_name=process_name,
operator_count=1,
process_unit_price_yuan=1,
standard_beat=1,
standard_workload=0,
good_qty=650,
defect_qty=0,
scrap_qty=0,
allocated_minutes=360,
)
def _allocation(
*,
id: int,
report_id: int,
report_item_id: int,
allocation_date: date,
good_qty: float,
):
return ProductionReportAllocation(
id=id,
report_id=report_id,
report_item_id=report_item_id,
attendance_point_name="嘉恒",
employee_phone="13800000000",
allocation_date=allocation_date,
day_minutes=0,
overtime_minutes=0,
night_minutes=0,
effective_minutes=0,
good_qty=good_qty,
defect_qty=0,
scrap_qty=0,
changeover_count=0,
reference_wage=0,
)
def test_fold_reported_good_quantity_rows_uses_allocation_month_for_latest_process_only():
rows = [
("嘉恒", "产品A", "1序", 7, 999),
("嘉恒", "产品A", "2序", 7, 350),
("嘉恒", "产品A", "2序", 8, 300),
]
latest_processes = {("嘉恒", "产品A"): {"2序"}}
totals = reconciliation._fold_reported_good_quantity_rows(rows, latest_processes)
assert totals[("嘉恒", "产品A", 7)] == 350
assert totals[("嘉恒", "产品A", 8)] == 300
def test_reported_good_quantities_groups_by_allocation_month_for_approved_unvoided_latest_process():
db = _sqlite_db()
try:
approved = _report(id=1)
pending = _report(id=2, status=ReportStatus.pending)
voided = _report(id=3, is_voided=True)
latest_item = _item(id=1, report_id=approved.id, process_name="2序")
earlier_item = _item(id=2, report_id=approved.id, process_name="1序")
pending_item = _item(id=3, report_id=pending.id, process_name="2序")
voided_item = _item(id=4, report_id=voided.id, process_name="2序")
db.add_all([
approved,
pending,
voided,
latest_item,
earlier_item,
pending_item,
voided_item,
_allocation(
id=1,
report_id=approved.id,
report_item_id=latest_item.id,
allocation_date=date(2026, 7, 31),
good_qty=350,
),
_allocation(
id=2,
report_id=approved.id,
report_item_id=latest_item.id,
allocation_date=date(2026, 8, 1),
good_qty=300,
),
_allocation(
id=3,
report_id=approved.id,
report_item_id=earlier_item.id,
allocation_date=date(2026, 7, 31),
good_qty=999,
),
_allocation(
id=4,
report_id=pending.id,
report_item_id=pending_item.id,
allocation_date=date(2026, 7, 31),
good_qty=111,
),
_allocation(
id=5,
report_id=voided.id,
report_item_id=voided_item.id,
allocation_date=date(2026, 8, 1),
good_qty=222,
),
])
db.commit()
totals = reconciliation._reported_good_quantities(
db,
year=2026,
product_keys=[("嘉恒", "产品A")],
latest_processes={("嘉恒", "产品A"): {"2序"}},
)
assert totals[("嘉恒", "产品A", 7)] == 350
assert totals[("嘉恒", "产品A", 8)] == 300
finally:
db.close()
def test_reported_good_quantities_ignores_mismatched_allocation_report_item_rows():
db = _sqlite_db()
try:
approved = _report(id=1)
other_report = _report(id=2, status=ReportStatus.pending)
other_item = _item(id=2, report_id=other_report.id, process_name="2序")
db.add_all([
approved,
other_report,
other_item,
_allocation(
id=1,
report_id=approved.id,
report_item_id=other_item.id,
allocation_date=date(2026, 7, 31),
good_qty=123,
),
])
db.commit()
totals = reconciliation._reported_good_quantities(
db,
year=2026,
product_keys=[("嘉恒", "产品A")],
latest_processes={("嘉恒", "产品A"): {"2序"}},
)
assert totals == {}
finally:
db.close()

1109
tests/test_report_allocations.py
View File

File diff suppressed because it is too large Load Diff

945
tests/test_report_over_limit.py
View File

@ -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

255
tests/test_usage_stats.py
View File

@ -1,21 +1,9 @@
from datetime import date, datetime
from io import BytesIO from io import BytesIO
from types import SimpleNamespace from types import SimpleNamespace
import json import json
from openpyxl import load_workbook 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 ( from app.routers.usage_stats import (
_build_usage_stats_export_response, _build_usage_stats_export_response,
_filter_usage_rows, _filter_usage_rows,
@ -25,24 +13,11 @@ from app.routers.usage_stats import (
_to_schema_row, _to_schema_row,
) )
from app.schemas import UsageStatsDetailOut, UsageStatsRow 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 UsageStatRow
from app.services.usage_stats import build_usage_stats, split_cleaning_device_nos from app.services.usage_stats import build_usage_stats, split_cleaning_device_nos
from app.services.usage_stats_export import export_usage_stats_rows 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): def _report(*items):
return SimpleNamespace(items=list(items)) return SimpleNamespace(items=list(items))
@ -62,93 +37,6 @@ def _item(**overrides):
return SimpleNamespace(**values) 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(): 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("清洗机A、清洗机B, 清洗机C") == ["清洗机A", "清洗机B", "清洗机C"]
assert split_cleaning_device_nos(None) == [] assert split_cleaning_device_nos(None) == []
@ -426,31 +314,6 @@ def test_build_usage_stats_for_cleaning_device_defaults_object_type_to_cleaning_
assert rows[0].object_type == "清洗设备" 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(): def test_filter_usage_rows_searches_all_visible_fields():
rows = [ rows = [
UsageStatRow( UsageStatRow(
@ -783,121 +646,3 @@ def test_matches_target_uses_id_before_ambiguous_device_fields():
assert _matches_target(minutes_row, **target_args) assert _matches_target(minutes_row, **target_args)
assert not _matches_target(quantity_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()