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