From 4c0cbc981531d312c91c7be14bd0ce1978945431 Mon Sep 17 00:00:00 2001 From: souplearn Date: Sat, 25 Jul 2026 06:33:12 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=B9=B3=E8=A1=A1=E6=8A=A5=E5=B7=A5?= =?UTF-8?q?=E5=BD=92=E5=B1=9E=E6=8B=86=E5=88=86=E5=B9=B6=E7=BA=A6=E6=9D=9F?= =?UTF-8?q?=E5=94=AF=E4=B8=80=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/models.py | 7 ++ app/services/report_allocations.py | 104 ++++++++++++++++++++-- scripts/migrate_report_allocations.py | 119 ++++++++++++++++++++++---- tests/test_report_allocations.py | 74 +++++++++++++++- tests/test_usage_stats.py | 6 +- 5 files changed, 283 insertions(+), 27 deletions(-) diff --git a/app/models.py b/app/models.py index f008bb3..8839532 100644 --- a/app/models.py +++ b/app/models.py @@ -362,6 +362,13 @@ class ProductionReportAllocation(Base): item: Mapped[ProductionReportItem | None] = relationship(back_populates="allocations") __table_args__ = ( + Index( + "uq_report_allocations_report_item_date", + "report_id", + "report_item_id", + "allocation_date", + unique=True, + ), Index("idx_report_allocations_report", "report_id"), Index("idx_report_allocations_item", "report_item_id"), Index("idx_report_allocations_date", "allocation_date"), diff --git a/app/services/report_allocations.py b/app/services/report_allocations.py index 3c2d3a0..87141de 100644 --- a/app/services/report_allocations.py +++ b/app/services/report_allocations.py @@ -1,11 +1,11 @@ from collections import defaultdict -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import date, datetime, timedelta -from sqlalchemy import delete +from sqlalchemy import delete, select from sqlalchemy.orm import Session -from app.models import ProductionReportAllocation +from app.models import ProductionReport, ProductionReportAllocation from app.services.common import as_float, round2 from app.services.metrics import ( SHIFT_BUCKETS, @@ -429,6 +429,91 @@ def _draft_from_minutes(report, item, allocation_date: date, minutes: dict[str, ) +SPLIT_RESIDUAL_FIELDS = ( + "good_qty", + "defect_qty", + "scrap_qty", + "changeover_count", + "reference_wage", +) + + +def _item_split_targets(item) -> dict[str, float]: + return { + "good_qty": round2(getattr(item, "good_qty", 0)), + "defect_qty": round2(getattr(item, "defect_qty", 0)), + "scrap_qty": round2(getattr(item, "scrap_qty", 0)), + "changeover_count": round2(getattr(item, "changeover_count", 0)), + "reference_wage": round2(_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: int(round(as_float(getattr(draft, field)) * 100)) for field in SPLIT_RESIDUAL_FIELDS} + for draft in drafts + ] + + for field in SPLIT_RESIDUAL_FIELDS: + target_cents = int(round(round2(targets[field]) * 100)) + current_cents = sum(values[field] for values in mutable_values) + residual = target_cents - current_cents + if residual == 0: + continue + + if residual > 0: + ordered_indexes = sorted( + range(len(drafts)), + key=lambda index: ( + -as_float(drafts[index].effective_minutes), + drafts[index].allocation_date.isoformat(), + index, + ), + ) + else: + ordered_indexes = sorted( + range(len(drafts)), + key=lambda index: ( + -mutable_values[index][field], + -as_float(drafts[index].effective_minutes), + drafts[index].allocation_date.isoformat(), + index, + ), + ) + + step = 1 if residual > 0 else -1 + remaining = abs(residual) + while remaining > 0: + changed = False + for index in ordered_indexes: + if step < 0 and mutable_values[index][field] <= 0: + continue + mutable_values[index][field] += step + remaining -= 1 + changed = True + if remaining == 0: + break + if not changed: + break + + return [ + replace( + draft, + **{ + field: mutable_values[index][field] / 100 + for field in SPLIT_RESIDUAL_FIELDS + }, + ) + for index, draft in enumerate(drafts) + ] + + def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None = None) -> list[ReportAllocationDraft]: items = list(getattr(report, "items", []) or []) if not items: @@ -452,12 +537,14 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None = if total_effective <= 0: rows.append(_fallback_row(report, item)) continue + item_rows: list[ReportAllocationDraft] = [] for allocation_date in sorted(item_minutes): minutes = item_minutes[allocation_date] effective = sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) if effective <= 0: continue - rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective)) + item_rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective)) + rows.extend(_balance_item_split_residuals(item_rows, _item_split_targets(item))) return rows @@ -489,7 +576,14 @@ def refresh_report_allocations( commit: bool = False, ) -> list[ProductionReportAllocation]: """Rebuild allocation rows with metrics semantics, updating item allocation-derived fields.""" - db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == getattr(report, "id"))) + report_id = getattr(report, "id") + if isinstance(report_id, int): + db.execute( + select(ProductionReport.id) + .where(ProductionReport.id == report_id) + .with_for_update() + ).scalar_one_or_none() + db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report_id)) rows = [ ProductionReportAllocation( report_id=draft.report_id, diff --git a/scripts/migrate_report_allocations.py b/scripts/migrate_report_allocations.py index 8ed9509..acf3e3f 100644 --- a/scripts/migrate_report_allocations.py +++ b/scripts/migrate_report_allocations.py @@ -49,11 +49,21 @@ REQUIRED_COLUMNS = { }, } REQUIRED_INDEXES = { - "idx_report_allocations_report": ("report_id",), - "idx_report_allocations_item": ("report_item_id",), - "idx_report_allocations_date": ("allocation_date",), - "idx_report_allocations_point_date": ("attendance_point_name", "allocation_date"), - "idx_report_allocations_employee_date": ("employee_phone", "allocation_date"), + "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"), @@ -153,22 +163,51 @@ def existing_foreign_keys(conn) -> dict[str, tuple[str, str, str]]: return {row[0]: (row[1], row[2], row[3]) for row in rows} -def find_equivalent_index(indexes: dict[str, IndexMetadata], columns: tuple[str, ...]) -> str | None: +def required_index_columns(index_name: str) -> tuple[str, ...]: + return REQUIRED_INDEXES[index_name]["columns"] + + +def required_index_non_unique(index_name: str) -> int: + return int(REQUIRED_INDEXES[index_name]["non_unique"]) + + +def find_equivalent_index( + indexes: dict[str, IndexMetadata], + columns: tuple[str, ...], + non_unique: int, +) -> str | None: for index_name, metadata in indexes.items(): - if index_name != "PRIMARY" and metadata["columns"] == columns and metadata["non_unique"] == 1: + if ( + index_name != "PRIMARY" + and metadata["columns"] == columns + and metadata["non_unique"] == non_unique + ): return index_name return None def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None: - if index_name not in REQUIRED_INDEXES or REQUIRED_INDEXES[index_name] != columns: + if index_name not in REQUIRED_INDEXES or required_index_columns(index_name) != columns: raise ValueError(f"unexpected index definition: {index_name} {columns}") + non_unique = required_index_non_unique(index_name) indexes = existing_indexes(conn) - if index_name in indexes: + if ( + index_name in indexes + and indexes[index_name]["columns"] == columns + and indexes[index_name]["non_unique"] == non_unique + ): return + if index_name in indexes: + conn.execute( + text( + f"ALTER TABLE {quote_identifier(TABLE_NAME)} " + f"DROP INDEX {quote_identifier(index_name)}" + ) + ) + indexes = existing_indexes(conn) - equivalent_index = find_equivalent_index(indexes, columns) + equivalent_index = find_equivalent_index(indexes, columns, non_unique) if equivalent_index: conn.execute( text( @@ -179,14 +218,51 @@ def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None: return column_sql = ", ".join(quote_identifier(column) for column in columns) + unique_sql = "UNIQUE " if non_unique == 0 else "" conn.execute( text( - f"CREATE INDEX {quote_identifier(index_name)} " + f"CREATE {unique_sql}INDEX {quote_identifier(index_name)} " f"ON {quote_identifier(TABLE_NAME)} ({column_sql})" ) ) +def cleanup_duplicate_allocation_keys(conn) -> None: + if not table_exists(conn): + return + conn.execute( + text( + """ + DELETE target + FROM production_report_allocations target + JOIN ( + SELECT id + FROM ( + SELECT allocation.id + FROM production_report_allocations allocation + JOIN ( + SELECT + report_id, + report_item_id, + allocation_date, + MIN(id) AS keep_id + FROM production_report_allocations + WHERE report_item_id IS NOT NULL + GROUP BY report_id, report_item_id, allocation_date + HAVING COUNT(*) > 1 + ) duplicate_key + ON duplicate_key.report_id = allocation.report_id + AND duplicate_key.report_item_id = allocation.report_item_id + AND duplicate_key.allocation_date = allocation.allocation_date + WHERE allocation.id <> duplicate_key.keep_id + ) duplicate_rows + ) doomed + ON doomed.id = target.id + """ + ) + ) + + def cleanup_duplicate_fk_indexes(conn) -> None: indexes = existing_indexes(conn) duplicates = [ @@ -235,6 +311,8 @@ def ensure_schema(conn) -> None: created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), + UNIQUE KEY uq_report_allocations_report_item_date + (report_id, report_item_id, allocation_date), KEY idx_report_allocations_report (report_id), KEY idx_report_allocations_item (report_item_id), KEY idx_report_allocations_date (allocation_date), @@ -251,8 +329,9 @@ def ensure_schema(conn) -> None: ) ) - for index_name, columns in REQUIRED_INDEXES.items(): - create_index(conn, index_name, columns) + 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) @@ -319,19 +398,21 @@ def verify_schema(conn) -> None: wrong_indexes = [ f"{index_name}({', '.join(indexes[index_name]['columns'])})" - for index_name, columns in REQUIRED_INDEXES.items() - if indexes.get(index_name, {}).get("columns") != 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)}") - unique_required_indexes = [ + wrong_index_uniqueness = [ index_name for index_name in REQUIRED_INDEXES - if indexes[index_name]["non_unique"] != 1 + if indexes[index_name]["non_unique"] != required_index_non_unique(index_name) ] - if unique_required_indexes: - raise RuntimeError(f"{TABLE_NAME} required indexes are unique: {', '.join(unique_required_indexes)}") + if wrong_index_uniqueness: + raise RuntimeError( + f"{TABLE_NAME} incompatible index uniqueness: {', '.join(wrong_index_uniqueness)}" + ) duplicate_single_column_indexes = [ index_name diff --git a/tests/test_report_allocations.py b/tests/test_report_allocations.py index a5be109..ceee5bf 100644 --- a/tests/test_report_allocations.py +++ b/tests/test_report_allocations.py @@ -2,8 +2,10 @@ from datetime import date, datetime from decimal import Decimal from types import SimpleNamespace -from sqlalchemy import BigInteger, create_engine, select +import pytest +from sqlalchemy import BigInteger, create_engine, inspect, select from sqlalchemy.ext.compiler import compiles +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import raiseload, selectinload, sessionmaker from app.database import Base @@ -203,6 +205,31 @@ def test_short_shared_report_rounding_residual_keeps_aggregate_night_total(): assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows) +def test_item_split_fields_keep_original_totals_after_rounding_residual(): + item = _item( + good_qty=1, + defect_qty=1, + scrap_qty=1, + changeover_count=1, + process_unit_price_yuan=1, + ) + report = _report(datetime(2026, 7, 25, 6, 0), datetime(2026, 7, 28, 6, 0), item) + + rows = build_report_allocation_drafts(report) + + assert [row.allocation_date for row in rows] == [ + date(2026, 7, 25), + date(2026, 7, 26), + date(2026, 7, 27), + ] + assert [row.effective_minutes for row in rows] == [1340, 1340, 1340] + assert round(sum(row.good_qty for row in rows), 2) == 1 + assert round(sum(row.defect_qty for row in rows), 2) == 1 + assert round(sum(row.scrap_qty for row in rows), 2) == 1 + assert round(sum(row.changeover_count for row in rows), 2) == 1 + assert round(sum(row.reference_wage for row in rows), 2) == 1 + + def test_items_sharing_same_started_at_split_the_segment_minutes(): started_at = datetime(2026, 7, 25, 8, 0) first_item = _item(id=1, good_qty=10, started_at=started_at) @@ -660,6 +687,51 @@ def _sqlite_db(): return SessionLocal() +def test_report_allocation_schema_has_unique_report_item_date_key(): + engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + inspector = inspect(engine) + + unique_column_sets = [ + tuple(constraint["column_names"]) + for constraint in inspector.get_unique_constraints("production_report_allocations") + ] + unique_column_sets.extend( + tuple(index["column_names"]) + for index in inspector.get_indexes("production_report_allocations") + if index.get("unique") + ) + + assert ("report_id", "report_item_id", "allocation_date") in unique_column_sets + + +def test_report_allocation_rejects_duplicate_report_item_date_key(): + db = _sqlite_db() + try: + first = ProductionReportAllocation( + report_id=1, + report_item_id=1, + attendance_point_name="总厂", + employee_phone="13800000000", + allocation_date=date(2026, 7, 25), + ) + duplicate = ProductionReportAllocation( + report_id=1, + report_item_id=1, + attendance_point_name="总厂", + employee_phone="13800000000", + allocation_date=date(2026, 7, 25), + ) + db.add(first) + db.flush() + db.add(duplicate) + + with pytest.raises(IntegrityError): + db.flush() + finally: + db.close() + + def test_report_out_skips_unloaded_allocations_without_lazy_load(): db = _sqlite_db() try: diff --git a/tests/test_usage_stats.py b/tests/test_usage_stats.py index 9b0a543..63da15e 100644 --- a/tests/test_usage_stats.py +++ b/tests/test_usage_stats.py @@ -861,10 +861,12 @@ def test_approved_allocations_query_requires_approved_unvoided_matching_point_an report_item_id=104, attendance_point_name="二厂", ), + _db_report(id=5), + _db_item(id=105, report_id=5), _db_allocation( id=1005, - report_id=1, - report_item_id=101, + report_id=5, + report_item_id=105, attendance_point_name="二厂", ), _db_allocation(id=1006, report_id=1, report_item_id=102),