From b0be6609996523af4eb2d27be50d9a397ca3d149 Mon Sep 17 00:00:00 2001 From: souplearn Date: Sat, 25 Jul 2026 03:00:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=E6=8A=A5=E5=B7=A5?= =?UTF-8?q?=E5=B7=A5=E6=97=B6=E5=BD=92=E5=B1=9E=E6=8B=86=E5=88=86=E8=AE=A1?= =?UTF-8?q?=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/metrics.py | 73 ++++++++ app/services/report_allocations.py | 260 +++++++++++++++++++++++++++++ tests/test_metrics.py | 13 ++ tests/test_report_allocations.py | 202 ++++++++++++++++++++++ 4 files changed, 548 insertions(+) create mode 100644 app/services/report_allocations.py create mode 100644 tests/test_report_allocations.py diff --git a/app/services/metrics.py b/app/services/metrics.py index 25293a1..d91a6e0 100644 --- a/app/services/metrics.py +++ b/app/services/metrics.py @@ -1,4 +1,5 @@ from collections import defaultdict +from dataclasses import dataclass from datetime import date, datetime, time, timedelta from typing import Protocol @@ -29,6 +30,13 @@ class ReportDeviceLike(Protocol): sort_order: int +@dataclass(frozen=True) +class ItemTimeRange: + item: ReportItemLike + start_at: datetime + end_at: datetime + + SHIFT_BUCKETS = ("day", "overtime", "night") SHIFT_LABELS = { "day": "白班", @@ -389,6 +397,71 @@ 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] + for item in grouped_items[item_start]: + ranges.append(ItemTimeRange(item=item, start_at=item_start, end_at=item_end)) + 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: + return [ItemTimeRange(item=item, start_at=start_at, end_at=end_at) 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() + for device_no, device_ranges in segment_ranges.items(): + device_items = items_by_device.get(device_no, []) + if not device_items: + 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: + ranges.extend(ItemTimeRange(item=item, start_at=start_at, end_at=end_at) for item in unmatched_items) + return ranges + + def _allocate_item_minutes( start_at: datetime, end_at: datetime, diff --git a/app/services/report_allocations.py b/app/services/report_allocations.py new file mode 100644 index 0000000..a994c19 --- /dev/null +++ b/app/services/report_allocations.py @@ -0,0 +1,260 @@ +from collections import defaultdict +from dataclasses import dataclass +from datetime import date, datetime, time, timedelta + +from sqlalchemy import delete +from sqlalchemy.orm import Session + +from app.models import ProductionReportAllocation +from app.services.common import as_float, round2 +from app.services.metrics import ( + SHIFT_BUCKETS, + SHIFT_LABELS, + _blank_overtime_intervals, + _covered_meal_overlap_minutes, + _iter_shift_base_dates, + _overlap_minutes, + _schedule_interval, + format_hours, + item_time_ranges, + minutes_between, +) +from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig + + +@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 _midnight(value: date) -> datetime: + return datetime.combine(value, time.min) + + +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 _add_night_minutes( + rows: dict[date, dict[str, float]], + start_at: datetime, + end_at: datetime, + base_date: date, +) -> None: + midnight = _midnight(base_date + timedelta(days=1)) + before_midnight = _overlap_minutes(start_at, end_at, _midnight(base_date), midnight) + after_midnight = _overlap_minutes(start_at, end_at, midnight, midnight + timedelta(days=1)) + _add_minutes(rows, base_date, "night", before_midnight + after_midnight) + + +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_night_minutes(rows, overlap_start, overlap_end, current_date) + 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( + _covered_meal_overlap_minutes(start_at, end_at, meal_start, meal_end) + for meal_start, meal_end in meal_intervals + ) + 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 _item_reference_wage(item) -> float: + return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0)) + + +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=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 _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=round2(as_float(getattr(item, "good_qty", 0)) * ratio), + defect_qty=round2(as_float(getattr(item, "defect_qty", 0)) * ratio), + scrap_qty=round2(as_float(getattr(item, "scrap_qty", 0)) * ratio), + changeover_count=round2(as_float(getattr(item, "changeover_count", 0)) * ratio), + reference_wage=round2(reference_wage * ratio), + ) + + +def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None = None) -> list[ReportAllocationDraft]: + items = list(getattr(report, "items", []) or []) + if not items: + return [] + + 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(lambda: {key: 0.0 for key in SHIFT_BUCKETS})) + for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items(): + for key in SHIFT_BUCKETS: + target[allocation_date][key] += minutes.get(key, 0.0) + + 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 + 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)) + 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]: + db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == getattr(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 diff --git a/tests/test_metrics.py b/tests/test_metrics.py index b1ab1e7..d5dfa1c 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -310,3 +310,16 @@ def test_dinner_blank_interval_counts_as_overtime(): 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 diff --git a/tests/test_report_allocations.py b/tests/test_report_allocations.py new file mode 100644 index 0000000..11a294d --- /dev/null +++ b/tests/test_report_allocations.py @@ -0,0 +1,202 @@ +from datetime import date, datetime +from types import SimpleNamespace + +from sqlalchemy import BigInteger, create_engine, select +from sqlalchemy.ext.compiler import compiles +from sqlalchemy.orm import sessionmaker + +from app.database import Base +from app.models import ProductionReport, ProductionReportAllocation, ProductionReportItem +from app.services.report_allocations import ( + allocation_summary_text, + build_report_allocation_drafts, + refresh_report_allocations, +) + + +@compiles(BigInteger, "sqlite") +def _compile_big_integer_for_sqlite(type_, compiler, **kw): + return "INTEGER" + + +def _item( + *, + id: int = 1, + good_qty: float = 0, + defect_qty: float = 0, + scrap_qty: float = 0, + changeover_count: float = 0, + process_unit_price_yuan: float = 0, + started_at: datetime | None = None, +): + return SimpleNamespace( + id=id, + device_no="23#", + product_name="23#", + process_name="冲压", + started_at=started_at, + standard_beat=1, + standard_workload=0, + stamping_method=None, + good_qty=good_qty, + defect_qty=defect_qty, + scrap_qty=scrap_qty, + changeover_count=changeover_count, + process_unit_price_yuan=process_unit_price_yuan, + allocated_minutes=0, + ) + + +def _report(start_at: datetime, end_at: datetime, item, *, id: int = 1, report_date: date | None = None): + return SimpleNamespace( + id=id, + attendance_point_name="总厂", + employee_phone="13800000000", + report_date=report_date or start_at.date(), + start_at=start_at, + end_at=end_at, + items=[item], + session=SimpleNamespace(devices=[]), + ) + + +def test_cross_day_after_midnight_allocates_previous_night_and_current_overtime_day(): + item = _item( + good_qty=650, + defect_qty=65, + scrap_qty=6.5, + changeover_count=13, + process_unit_price_yuan=1, + ) + report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) + + rows = build_report_allocation_drafts(report) + + assert len(rows) == 2 + previous, current = rows + assert previous.allocation_date == date(2026, 7, 24) + assert previous.night_minutes == 350 + assert previous.day_minutes == 0 + assert previous.overtime_minutes == 0 + assert previous.effective_minutes == 350 + assert previous.good_qty == 350 + assert previous.defect_qty == 35 + assert previous.scrap_qty == 3.5 + assert previous.changeover_count == 7 + assert previous.reference_wage == 350 + + assert current.allocation_date == date(2026, 7, 25) + assert current.night_minutes == 0 + assert current.overtime_minutes == 120 + assert current.day_minutes == 180 + assert current.effective_minutes == 300 + assert current.good_qty == 300 + assert current.defect_qty == 30 + assert current.scrap_qty == 3 + assert current.changeover_count == 6 + assert current.reference_wage == 300 + + +def test_one_minute_before_night_end_allocates_to_previous_date(): + item = _item(good_qty=301, process_unit_price_yuan=2) + report = _report(datetime(2026, 7, 25, 5, 59), datetime(2026, 7, 25, 11, 0), item) + + rows = build_report_allocation_drafts(report) + + assert [(row.allocation_date, row.night_minutes, row.overtime_minutes, row.day_minutes) for row in rows] == [ + (date(2026, 7, 24), 1, 0, 0), + (date(2026, 7, 25), 0, 120, 180), + ] + assert rows[0].good_qty == 1 + assert rows[0].reference_wage == 2 + assert rows[1].good_qty == 300 + assert rows[1].reference_wage == 600 + + +def test_zero_effective_minutes_falls_back_to_original_report_date(): + item = _item(good_qty=5, defect_qty=1, scrap_qty=0.5, changeover_count=2, process_unit_price_yuan=3) + report = _report( + datetime(2026, 7, 25, 9, 0), + datetime(2026, 7, 25, 9, 0), + item, + report_date=date(2026, 7, 25), + ) + + rows = build_report_allocation_drafts(report) + + assert len(rows) == 1 + row = rows[0] + assert row.allocation_date == date(2026, 7, 25) + assert row.effective_minutes == 0 + assert row.day_minutes == 0 + assert row.overtime_minutes == 0 + assert row.night_minutes == 0 + assert row.good_qty == 5 + assert row.defect_qty == 1 + assert row.scrap_qty == 0.5 + assert row.changeover_count == 2 + assert row.reference_wage == 15 + + +def test_allocation_summary_text_groups_dates_and_shift_kinds(): + item = _item(good_qty=650, process_unit_price_yuan=1) + report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) + + rows = build_report_allocation_drafts(report) + + assert allocation_summary_text(rows) == "2026-07-24 夜班5.83小时；2026-07-25 白班3小时、加班2小时" + + +def test_refresh_report_allocations_replaces_existing_rows_and_flushes(): + engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + SessionLocal = sessionmaker(bind=engine, future=True) + db = SessionLocal() + try: + report = ProductionReport( + id=1, + session_id=1, + attendance_point_name="总厂", + employee_phone="13800000000", + report_date=date(2026, 7, 25), + start_at=datetime(2026, 7, 25, 5, 59), + end_at=datetime(2026, 7, 25, 11, 0), + ) + item = ProductionReportItem( + id=1, + report_id=1, + attendance_point_name="总厂", + device_no="23#", + project_no="P1", + product_name="23#", + process_name="冲压", + process_unit_price_yuan=2, + good_qty=301, + defect_qty=0, + scrap_qty=0, + changeover_count=0, + standard_beat=1, + standard_workload=0, + ) + stale = ProductionReportAllocation( + id=99, + report_id=1, + report_item_id=1, + attendance_point_name="总厂", + employee_phone="13800000000", + allocation_date=date(2026, 7, 20), + effective_minutes=1, + ) + report.items.append(item) + db.add_all([report, stale]) + db.flush() + + rows = refresh_report_allocations(db, report) + + assert len(rows) == 2 + assert all(row.id is not None for row in rows) + stored = db.scalars(select(ProductionReportAllocation).order_by(ProductionReportAllocation.allocation_date)).all() + assert [row.allocation_date for row in stored] == [date(2026, 7, 24), date(2026, 7, 25)] + assert [row.good_qty for row in stored] == [1, 300] + finally: + db.close()