ForgeFlow-ERP/backend/app/api/routes/production.py

from datetime import date, datetime, timedelta
from decimal import Decimal, ROUND_FLOOR
import re
from uuid import uuid4

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, aliased

from app.db.session import get_db
from app.models.document_archive import DocumentArchive
from app.models.master_data import BomItem, Item, Warehouse
from app.models.miniapp import MiniAppPersonnel, MiniAppProductionReport, MiniAppProductionReportItem
from app.models.operations import (
    CompletionReceipt,
    CompletionReceiptItem,
    Equipment,
    InventoryTxn,
    OperationReport,
    ProcessRouteOperation,
    ProductionBatchLedger,
    ProductionBatchLedgerTxn,
    ReturnDisposition,
    ReturnItem,
    ScrapRecord,
    StockLot,
    WorkOrder,
    WorkOrderMaterial,
    WorkOrderMaterialIssue,
    WorkOrderOperation,
)
from app.models.org import Department, Employee
from app.models.sales import SalesOrderItem
from app.schemas.operations import (
    CompletionReceiptCreate,
    CompletionReceiptItemRead,
    CompletionReceiptRead,
    MiniAppOperationReportRead,
    MiniAppOperationReportSyncRead,
    MiniAppOperationReportSyncStateRead,
    MiniAppScrapRecordCreate,
    OperationReportCreate,
    OperationReportRead,
    ProductionBatchLedgerRead,
    ProductionBatchLedgerInboundCreate,
    ProductionBatchLedgerInboundPreviewRead,
    ProductionBatchLedgerInboundRead,
    ProductionWorkOrderInboundCreate,
    ProductionWorkOrderInboundPreviewRead,
    ProductionWorkOrderInboundRead,
    ScrapRecordCreate,
    ScrapRecordRead,
    WorkOrderCreate,
    WorkOrderLedgerRead,
    WorkOrderMaterialIssueRead,
    WorkOrderMaterialRead,
    WorkOrderOperationRead,
    WorkOrderRead,
)
from app.schemas.document_archives import DocumentArchiveGenerateResult
from app.services.auth import AuthContext, require_authenticated_user
from app.services.auth import next_employee_code
from app.services.document_archives import (
    ARCHIVE_STATUS_MISSING,
    DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT,
    DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT,
    FILE_FORMAT_PDF,
    generate_document_archive,
    production_document_batch_no_from_text,
)
from app.services.operations import (
    build_raw_lot_product_work_order_no,
    build_yearly_product_work_order_no,
    create_inventory_txn,
    create_work_order_materials_and_operations,
    clamp_non_negative,
    get_completion_receipt_items_query,
    get_completion_receipts_query,
    get_default_bom,
    get_default_route,
    get_location_or_default,
    get_operation_reports_query,
    get_scrap_records_query,
    get_work_order_materials_query,
    get_work_order_material_issues_query,
    get_work_order_operations_query,
    get_work_orders_query,
    issue_selected_stock_lots,
    sync_recent_work_order_reference_qty,
    sync_work_order_reference_qty,
    sync_work_order_status,
    to_decimal,
    upsert_stock_balance,
)
from app.services.system_permissions import ensure_employee_has_permission
from app.services.system_config import (
    get_miniapp_operation_report_last_sync,
    lock_miniapp_operation_report_last_sync,
    upsert_miniapp_operation_report_last_sync,
)
from app.services.production_batch_ledger import (
    calculate_production_batch_ledger_snapshot,
    lock_production_batch_ledger,
    post_finished_inbound_to_ledger,
    post_scrap_inbound_to_ledger,
    post_surplus_inbound_to_ledger,
    record_production_issue_to_ledger,
)
from app.services.production_work_order_ledger import (
    INBOUND_TOLERANCE_MULTIPLIER,
    get_work_order_ledger_rows,
    get_work_order_limit_snapshot,
    validate_inbound_with_tolerance,
)
from app.services.production_work_order_inbound import (
    create_production_work_order_inbound,
    preview_production_work_order_inbound,
)
from app.services.stocktake import ensure_warehouses_unlocked
router = APIRouter(dependencies=[Depends(require_authenticated_user)])

_MINIAPP_WORK_ORDER_SHORT_CODE_RE = re.compile(r"^JH(\d{4})$")
PRODUCTION_LEDGER_INBOUND_TXN_TYPES = {"成品入库", "生产余料入库", "生产废料入库"}


def _archive_fields_from_result(
    result: DocumentArchiveGenerateResult | None,
    *,
    document_type: str,
    business_id: int | None,
) -> dict[str, object]:
    if result is None:
        return {
            "archive_status": "未生成",
            "archive_business_id": business_id,
            "archive_document_type": document_type if business_id else None,
            "archive_error_message": None,
        }
    return {
        "archive_status": result.archive_status,
        "archive_business_id": business_id,
        "archive_document_type": document_type,
        "archive_error_message": result.archive_error_message,
    }


def _failed_archive_result(document_type: str, business_id: int, document_no: str, exc: Exception) -> DocumentArchiveGenerateResult:
    _ = exc
    return DocumentArchiveGenerateResult(
        business_id=business_id,
        document_type=document_type,
        document_no=document_no,
        archive_status="归档失败",
        archive_version=None,
        archive_error_message="归档失败：PDF归档服务异常，请稍后重新生成",
    )


def _production_document_batch_no(now: datetime) -> str:
    return f"生产结算{now.strftime('%Y%m%d%H%M%S')}{uuid4().int % 1_000_000:06d}"


def _with_document_batch_no(remark: str, batch_no: str | None) -> str:
    if not batch_no:
        return remark
    return f"{remark}；单据批次号：{batch_no}"


def _latest_archive_map(db: Session, document_type: str, business_ids: list[int]) -> dict[int, DocumentArchive]:
    if not business_ids:
        return {}
    latest_version_subquery = (
        select(
            DocumentArchive.business_id.label("business_id"),
            func.max(DocumentArchive.archive_version).label("archive_version"),
        )
        .where(
            DocumentArchive.document_type == document_type,
            DocumentArchive.file_format == FILE_FORMAT_PDF,
            DocumentArchive.business_id.in_(business_ids),
        )
        .group_by(DocumentArchive.business_id)
        .subquery()
    )
    latest_id_subquery = (
        select(
            DocumentArchive.business_id.label("business_id"),
            func.max(DocumentArchive.id).label("archive_id"),
        )
        .join(
            latest_version_subquery,
            (latest_version_subquery.c.business_id == DocumentArchive.business_id)
            & (latest_version_subquery.c.archive_version == DocumentArchive.archive_version),
        )
        .where(
            DocumentArchive.document_type == document_type,
            DocumentArchive.file_format == FILE_FORMAT_PDF,
        )
        .group_by(DocumentArchive.business_id)
        .subquery()
    )
    rows = db.scalars(
        select(DocumentArchive).join(latest_id_subquery, latest_id_subquery.c.archive_id == DocumentArchive.id)
    ).all()
    return {int(row.business_id): row for row in rows}


def _code_token(value: str | None, fallback: str, max_length: int = 16) -> str:
    token = "".join(ch for ch in str(value or "").upper() if ch.isalnum())
    return (token or fallback)[-max_length:]


def _next_doc_no(db: Session, model, column_name: str, prefix: str) -> str:
    column = getattr(model, column_name)
    count = db.scalar(select(func.count(model.id)).where(column.like(f"{prefix}-%"))) or 0
    return f"{prefix}-{count + 1:03d}"


def _next_optional_sequence_doc_no(db: Session, model, column_name: str, prefix: str, max_length: int = 80) -> str:
    normalized_prefix = " ".join(str(prefix or "").split()).strip("- ") or "未命名"
    sequence_prefix = normalized_prefix[: max(max_length - 4, 1)]
    column = getattr(model, column_name)
    if not db.scalar(select(model.id).where(column == sequence_prefix).limit(1)):
        return sequence_prefix
    count = db.scalar(select(func.count(model.id)).where(column.like(f"{sequence_prefix}-%"))) or 0
    return f"{sequence_prefix}-{count + 1:03d}"


def _next_yearly_sequence_work_order_no(db: Session, product: Item, now: datetime | None = None) -> str:
    current_time = now or datetime.now()
    year_part = current_time.strftime("%Y")
    existing_nos = db.scalars(select(WorkOrder.work_order_no).where(WorkOrder.work_order_no.like(f"{year_part}____-%"))).all()
    used_sequences = [
        int(work_order_no[4:8])
        for work_order_no in existing_nos
        if len(work_order_no) > 8 and work_order_no[:4] == year_part and work_order_no[4:8].isdigit() and work_order_no[8] == "-"
    ]
    next_sequence = max(used_sequences, default=0) + 1
    product_name = " ".join(str(product.item_name or product.item_code or f"产品{product.id}").split()).strip("- ") or f"产品{product.id}"
    max_product_name_length = max(80 - len(year_part) - 4 - 1, 1)
    return f"{year_part}{next_sequence:04d}-{product_name[:max_product_name_length]}"


def _build_work_order_no(db: Session, product: Item) -> str:
    return build_yearly_product_work_order_no(db, product)


def _miniapp_work_order_short_code_sequence(value: str | None) -> int | None:
    match = _MINIAPP_WORK_ORDER_SHORT_CODE_RE.match(str(value or "").strip())
    if not match:
        return None
    return int(match.group(1))


def _is_miniapp_work_order_short_code(value: str | None) -> bool:
    return _miniapp_work_order_short_code_sequence(value) is not None


def _miniapp_syncable_work_order_status_filter():
    return func.upper(func.coalesce(WorkOrder.status, "")).notin_(("SETTLED", "CANCELED", "CANCELLED"))


def _resolve_work_order_by_miniapp_short_code(db: Session, short_code: str, row: dict) -> WorkOrder | None:
    sequence = _miniapp_work_order_short_code_sequence(short_code)
    if sequence is None:
        return None
    product_name = str(row.get("product_name") or "").strip()
    project_no = str(row.get("project_no") or "").strip()
    product_filters = []
    if product_name:
        product_filters.append(Item.item_name == product_name)
    if project_no:
        product_filters.append(Item.item_code == project_no)
    if not product_filters:
        return None

    year_part = datetime.now().strftime("%Y")
    work_order_prefix = f"{year_part}{sequence:04d}-"
    candidates = db.scalars(
        select(WorkOrder)
        .join(Item, Item.id == WorkOrder.product_item_id)
        .where(
            WorkOrder.work_order_no.like(f"{work_order_prefix}%"),
            _miniapp_syncable_work_order_status_filter(),
            or_(*product_filters),
        )
        .order_by(WorkOrder.id.desc())
    ).all()
    if len(candidates) == 1:
        return candidates[0]
    return None



def _build_scrap_no(db: Session, work_order: WorkOrder, disposal_type: str) -> str:
    date_part = date.today().strftime("%Y%m%d")
    type_part = "RWK" if disposal_type == "REWORK" else "SCRAP"
    work_order_part = _code_token(work_order.work_order_no, f"WO{work_order.id:04d}", 18)
    return _next_doc_no(db, ScrapRecord, "scrap_no", f"DSP-{type_part}-{date_part}-{work_order_part}")


def _build_completion_receipt_no(db: Session, work_order: WorkOrder) -> str:
    date_part = date.today().strftime("%Y%m%d")
    work_order_part = _code_token(work_order.work_order_no, f"WO{work_order.id:04d}", 18)
    return _next_doc_no(db, CompletionReceipt, "receipt_no", f"FGI-NB-{date_part}-{work_order_part}")


def _build_operation_report_no(db: Session, work_order: WorkOrder, operation: WorkOrderOperation, employee: Employee) -> str:
    date_part = date.today().strftime("%Y%m%d")
    work_order_part = _code_token(work_order.work_order_no, f"WO{work_order.id:04d}", 12)
    operation_part = f"OP{int(operation.seq_no or 0):02d}"
    employee_part = _code_token(employee.employee_code or employee.employee_name, f"EMP{employee.id:04d}", 8)
    return _next_doc_no(db, OperationReport, "report_no", f"RPT-NB-{date_part}-{work_order_part}-{operation_part}-{employee_part}")


def _production_ledger_txn_read(
    txn: ProductionBatchLedgerTxn,
    archive: DocumentArchive | None = None,
    archive_business_id: int | None = None,
) -> dict:
    return {
        "production_ledger_txn_id": txn.id,
        "production_ledger_id": txn.production_ledger_id,
        "txn_type": txn.txn_type,
        "qty_delta": float(to_decimal(txn.qty_delta)),
        "weight_delta_kg": float(to_decimal(txn.weight_delta_kg)),
        "outside_weight_after_kg": float(to_decimal(txn.outside_weight_after_kg)),
        "source_doc_type": txn.source_doc_type,
        "source_doc_id": txn.source_doc_id,
        "source_line_id": txn.source_line_id,
        "biz_time": txn.biz_time,
        "operator_user_id": txn.operator_user_id,
        "remark": txn.remark,
        "archive_status": archive.status if archive else (ARCHIVE_STATUS_MISSING if archive_business_id else None),
        "archive_business_id": archive_business_id,
        "archive_document_type": DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT if archive_business_id else None,
        "archive_error_message": archive.error_message if archive else None,
    }


def _build_finished_lot_no(db: Session, work_order: WorkOrder, product: Item, receipt_time: datetime, line_no: int) -> str:
    date_part = receipt_time.strftime("%Y%m%d")
    product_part = _code_token(product.item_code or product.item_name, f"FG{product.id:04d}", 12)
    work_order_part = _code_token(work_order.work_order_no, f"WO{work_order.id:04d}", 10)
    base_lot_no = f"FGL-NB-{date_part}-{product_part}-{work_order_part}-L{line_no:02d}"
    candidate = base_lot_no
    suffix = 1
    while db.scalar(select(func.count(StockLot.id)).where(StockLot.lot_no == candidate)):
        suffix += 1
        candidate = f"{base_lot_no}-{suffix:02d}"
    return candidate


def _get_active_warehouse(db: Session, warehouse_type: str, operation_name: str) -> Warehouse:
    warehouse = db.scalar(
        select(Warehouse)
        .where(Warehouse.warehouse_type == warehouse_type, Warehouse.status == "ACTIVE")
        .order_by(Warehouse.id)
        .limit(1)
    )
    if not warehouse:
        warehouse_name = {"FINISHED": "成品库", "SCRAP": "废料库", "RAW": "原材料库"}.get(warehouse_type, "目标仓库")
        raise HTTPException(status_code=400, detail=f"{operation_name}缺少启用的{warehouse_name}")
    return warehouse


def _build_production_ledger_lot_no(db: Session, prefix: str, ledger: ProductionBatchLedger, item: Item, biz_time: datetime) -> str:
    item_part = _code_token(item.item_code or item.item_name, f"ITEM{item.id:04d}", 12)
    source_part = _code_token(ledger.material_lot_no, f"LOT{ledger.material_lot_id:04d}", 10)
    return _next_doc_no(db, StockLot, "lot_no", f"{prefix}-NB-{biz_time.strftime('%Y%m%d')}-{item_part}-{source_part}")


def _issue_raw_lot_to_production_ledger(
    db: Session,
    *,
    lot: StockLot,
    material_item_id: int,
    issued_weight: Decimal,
    production_ledger_id: int,
    operator_user_id: int | None,
    biz_time: datetime,
) -> InventoryTxn:
    if issued_weight <= 0:
        raise HTTPException(status_code=400, detail="出库重量必须大于0")
    if not lot or int(lot.item_id) != int(material_item_id):
        raise HTTPException(status_code=400, detail="库存批次不存在或物料不匹配")
    warehouse = db.get(Warehouse, lot.warehouse_id)
    if not warehouse or warehouse.warehouse_type != "RAW":
        raise HTTPException(status_code=400, detail="库存批次不属于原材料库")
    usable_weight = clamp_non_negative(to_decimal(lot.remaining_weight_kg) - to_decimal(lot.locked_weight_kg))
    if lot.quality_status != "PASS" or lot.status != "AVAILABLE" or usable_weight <= 0:
        raise HTTPException(status_code=400, detail=f"库存批次 {lot.lot_no} 不可用")
    if issued_weight > usable_weight:
        raise HTTPException(status_code=400, detail=f"库存批次 {lot.lot_no} 剩余重量不足")

    lot.remaining_qty = Decimal("0")
    lot.remaining_weight_kg = clamp_non_negative(to_decimal(lot.remaining_weight_kg) - issued_weight)
    if to_decimal(lot.remaining_weight_kg) <= 0:
        lot.status = "DEPLETED"
    db.add(lot)

    unit_cost = to_decimal(lot.unit_cost, "0.0001")
    upsert_stock_balance(
        db,
        item_id=material_item_id,
        warehouse_id=lot.warehouse_id,
        location_id=lot.location_id,
        qty_delta=Decimal("0"),
        weight_delta=-issued_weight,
        available_qty_delta=Decimal("0"),
        available_weight_delta=-issued_weight,
        unit_cost=unit_cost,
    )
    txn = create_inventory_txn(
        db,
        txn_type="MATERIAL_ISSUE",
        item_id=material_item_id,
        warehouse_id=lot.warehouse_id,
        location_id=lot.location_id,
        lot_id=lot.id,
        qty_change=Decimal("0"),
        weight_change=-issued_weight,
        unit_cost=unit_cost,
        source_doc_type="生产台账",
        source_doc_id=production_ledger_id,
        source_line_id=None,
        biz_time=biz_time,
        operator_user_id=operator_user_id,
        remark=f"生产出库：库存批次号 {lot.lot_no}",
        amount_basis="WEIGHT",
    )
    db.add(txn)
    db.flush()
    return txn


def _link_latest_issue_ledger_txn_to_inventory_txn(
    db: Session,
    *,
    production_ledger_id: int,
    inventory_txn: InventoryTxn,
) -> None:
    ledger_txn = db.scalar(
        select(ProductionBatchLedgerTxn)
        .where(
            ProductionBatchLedgerTxn.production_ledger_id == production_ledger_id,
            ProductionBatchLedgerTxn.txn_type == "生产出库",
        )
        .order_by(ProductionBatchLedgerTxn.id.desc())
        .limit(1)
    )
    if not ledger_txn:
        return
    ledger_txn.source_doc_type = "库存流水"
    ledger_txn.source_doc_id = inventory_txn.id
    ledger_txn.source_line_id = inventory_txn.id
    db.add(ledger_txn)


def _build_production_ledger_work_order_compat_read(
    *,
    production_ledger_id: int,
    material_lot_no: str,
    product: Item,
    route,
    bom,
    planned_qty: Decimal,
    payload: WorkOrderCreate,
    biz_time: datetime,
    status: str,
    operation_count: int,
) -> WorkOrderRead:
    return WorkOrderRead(
        work_order_id=production_ledger_id,
        work_order_no=material_lot_no,
        work_order_type="正常生产",
        source_sales_order_item_id=payload.source_sales_order_item_id,
        source_order_no=None,
        product_item_id=product.id,
        product_code=product.item_code,
        product_name=product.item_name,
        product_version_no=None,
        route_id=route.id,
        route_name=route.route_name,
        bom_id=bom.id,
        planned_qty=float(planned_qty),
        released_qty=float(planned_qty),
        finished_qty=0,
        scrap_qty=0,
        planned_start_time=payload.planned_start_time or biz_time,
        planned_end_time=payload.planned_end_time,
        actual_start_time=None,
        actual_end_time=None,
        priority_level=payload.priority_level or 3,
        status=status,
        remark=payload.remark,
        operation_count=operation_count,
        abnormal_operation_count=0,
        reported_good_qty=0,
        received_qty=0,
        max_receivable_qty=float(planned_qty),
        all_operations_done=False,
    )


def _calculate_actual_cycle_seconds(start_time: datetime, end_time: datetime, report_qty: Decimal) -> float | None:
    if report_qty <= 0 or end_time <= start_time:
        return None
    return round((end_time - start_time).total_seconds() / float(report_qty), 2)


def _build_operation_abnormal_reason(row: dict, actual_cycle_seconds: float | None) -> str | None:
    reasons: list[str] = []
    report_qty = to_decimal(row.get("report_qty"))
    scrap_qty = to_decimal(row.get("scrap_qty"))
    allowed_scrap_rate = to_decimal(row.get("allowed_scrap_rate"), "0.0001")
    ref_cycle_seconds = to_decimal(row.get("ref_cycle_seconds"), "0.01")

    if report_qty > 0 and allowed_scrap_rate >= 0:
        actual_scrap_rate = scrap_qty / report_qty if report_qty > 0 else Decimal("0")
        if actual_scrap_rate > allowed_scrap_rate:
            reasons.append(
                f"报废率超标 {round(float(actual_scrap_rate * 100), 2)}% > {round(float(allowed_scrap_rate * 100), 2)}%"
            )

    if actual_cycle_seconds is not None and ref_cycle_seconds > 0 and Decimal(str(actual_cycle_seconds)) > ref_cycle_seconds:
        reasons.append(f"实际节拍 {actual_cycle_seconds}s/件 慢于参考节拍 {float(ref_cycle_seconds):.2f}s/件")

    scrap_reason = row.get("scrap_reason")
    if scrap_reason:
        reasons.append(f"异常说明：{scrap_reason}")

    if not reasons:
        return None
    return "；".join(reasons)


def _build_operation_report_read(row: dict) -> OperationReportRead:
    payload = dict(row)
    actual_cycle_seconds = _calculate_actual_cycle_seconds(
        payload["start_time"],
        payload["end_time"],
        to_decimal(payload.get("report_qty")),
    )
    payload["ref_cycle_seconds"] = float(payload.get("ref_cycle_seconds") or 0)
    payload["actual_cycle_seconds"] = actual_cycle_seconds
    payload["abnormal_reason"] = _build_operation_abnormal_reason(payload, actual_cycle_seconds)
    return OperationReportRead.model_validate(payload)


def _get_average_issued_weight_cost(db: Session, work_order_id: int) -> Decimal:
    row = db.execute(
        select(
            func.coalesce(func.sum(InventoryTxn.amount), 0).label("total_amount"),
            func.coalesce(func.sum(InventoryTxn.weight_change_kg), 0).label("total_weight"),
        ).where(
            InventoryTxn.txn_type == "MATERIAL_ISSUE",
            InventoryTxn.source_doc_type == "WORK_ORDER",
            InventoryTxn.source_doc_id == work_order_id,
        )
    ).mappings().first()
    if not row:
        return Decimal("0")
    total_amount = to_decimal(row["total_amount"], "0.01")
    total_weight = abs(to_decimal(row["total_weight"]))
    if total_weight <= 0:
        return Decimal("0")
    return total_amount / total_weight


def _get_active_scrap_record(db: Session, report_id: int) -> ScrapRecord | None:
    return db.scalar(
        select(ScrapRecord)
        .where(
            ScrapRecord.report_id == report_id,
            ScrapRecord.status == "CONFIRMED",
        )
        .order_by(ScrapRecord.id.desc())
        .limit(1)
    )


def _revert_scrap_record_effect(work_order: WorkOrder, record: ScrapRecord) -> None:
    if record.disposal_type == "SCRAP":
        work_order.scrap_qty = max(Decimal("0"), to_decimal(work_order.scrap_qty) - to_decimal(record.scrap_qty))


MINIAPP_REPORT_STATUS_LABELS = {
    "pending": "待审核",
    "approved": "已审核",
    "rejected": "已驳回",
}


def _miniapp_report_no(report_id: int, item_id: int) -> str:
    return f"WRS-{report_id}-{item_id}"


def _miniapp_effective_scrap_qty(row: dict) -> Decimal:
    scrap_qty = to_decimal(row.get("miniapp_scrap_qty"))
    defect_qty = to_decimal(row.get("defect_qty"))
    return scrap_qty if scrap_qty > 0 else defect_qty


def _new_miniapp_resolve_cache() -> dict[str, dict]:
    return {
        "production_ledger": {},
        "work_order": {},
        "operation": {},
        "employee": {},
        "report": {},
    }


def _miniapp_report_item_rows(
    db: Session,
    *,
    limit: int | None = 300,
    source_item_id: int | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    updated_after: datetime | None = None,
    updated_until: datetime | None = None,
    order_by_updated_asc: bool = False,
) -> list[dict]:
    stmt = (
        select(
            MiniAppProductionReport.id.label("source_report_id"),
            MiniAppProductionReportItem.id.label("source_item_id"),
            func.coalesce(
                MiniAppProductionReportItem.attendance_point_name,
                MiniAppProductionReport.attendance_point_name,
            ).label("attendance_point_name"),
            MiniAppProductionReport.employee_phone.label("employee_phone"),
            MiniAppPersonnel.name.label("employee_name"),
            MiniAppProductionReport.report_date.label("report_date"),
            MiniAppProductionReport.start_at.label("start_time"),
            MiniAppProductionReport.end_at.label("end_time"),
            MiniAppProductionReport.duration_minutes.label("duration_minutes"),
            MiniAppProductionReport.effective_minutes.label("effective_minutes"),
            MiniAppProductionReport.status.label("status"),
            MiniAppProductionReport.reviewer_phone.label("reviewer_phone"),
            MiniAppProductionReport.reviewed_at.label("reviewed_at"),
            MiniAppProductionReport.reject_reason.label("reject_reason"),
            MiniAppProductionReportItem.device_no.label("device_no"),
            MiniAppProductionReportItem.project_no.label("project_no"),
            MiniAppProductionReportItem.product_name.label("product_name"),
            MiniAppProductionReportItem.material_code.label("material_code"),
            MiniAppProductionReportItem.material_name.label("material_name"),
            MiniAppProductionReportItem.raw_material_batch_no.label("raw_material_batch_no"),
            MiniAppProductionReportItem.process_name.label("process_name"),
            MiniAppProductionReportItem.stamping_method.label("stamping_method"),
            MiniAppProductionReportItem.standard_beat.label("standard_beat"),
            MiniAppProductionReportItem.standard_workload.label("standard_workload"),
            MiniAppProductionReportItem.good_qty.label("good_qty"),
            MiniAppProductionReportItem.defect_qty.label("defect_qty"),
            MiniAppProductionReportItem.scrap_qty.label("miniapp_scrap_qty"),
            MiniAppProductionReportItem.allocated_minutes.label("allocated_minutes"),
            MiniAppProductionReportItem.started_at.label("started_at"),
        )
        .join(MiniAppProductionReport, MiniAppProductionReport.id == MiniAppProductionReportItem.report_id)
        .outerjoin(MiniAppPersonnel, MiniAppPersonnel.phone == MiniAppProductionReport.employee_phone)
    )
    if source_item_id is not None:
        stmt = stmt.where(MiniAppProductionReportItem.id == source_item_id)
    if start_date is not None:
        stmt = stmt.where(MiniAppProductionReport.report_date >= start_date)
    if end_date is not None:
        stmt = stmt.where(MiniAppProductionReport.report_date <= end_date)
    if updated_after is not None:
        stmt = stmt.where(MiniAppProductionReport.updated_at > updated_after)
    if updated_until is not None:
        stmt = stmt.where(MiniAppProductionReport.updated_at <= updated_until)
    if order_by_updated_asc:
        stmt = stmt.order_by(MiniAppProductionReport.updated_at.asc(), MiniAppProductionReportItem.id.asc())
    else:
        stmt = stmt.order_by(MiniAppProductionReportItem.id.desc())
    if limit is not None:
        stmt = stmt.limit(limit)
    rows = db.execute(stmt).mappings().all()
    return [dict(row) for row in rows]


def _preload_miniapp_operation_reports(db: Session, report_nos: list[str], cache: dict[str, dict]) -> None:
    if not report_nos:
        return
    unique_report_nos = list(dict.fromkeys(report_nos))
    existing_reports = db.scalars(select(OperationReport).where(OperationReport.report_no.in_(unique_report_nos))).all()
    cache["report"].update({report_no: None for report_no in unique_report_nos})
    cache["report"].update({report.report_no: report for report in existing_reports})


def _get_miniapp_report_item_row(db: Session, source_item_id: int) -> dict:
    rows = _miniapp_report_item_rows(db, limit=1, source_item_id=source_item_id)
    if rows:
        return rows[0]
    raise HTTPException(status_code=404, detail="小程序报工明细不存在")


def _miniapp_report_matches_target(
    report: OperationReport | None,
    work_order: WorkOrder | None,
    operation: WorkOrderOperation | None,
) -> bool:
    return bool(
        report
        and work_order
        and operation
        and int(report.work_order_id) == int(work_order.id)
        and int(report.work_order_operation_id) == int(operation.id)
    )


def _miniapp_report_matches_production_ledger(
    report: OperationReport | None,
    production_ledger: ProductionBatchLedger | None,
) -> bool:
    return bool(
        report
        and production_ledger
        and report.production_ledger_id
        and int(report.production_ledger_id) == int(production_ledger.id)
    )


def _unlink_stale_miniapp_report(
    db: Session,
    report: OperationReport | None,
    *,
    cache: dict[str, dict] | None = None,
) -> None:
    if not report:
        return

    scrap_count = db.scalar(select(func.count(ScrapRecord.id)).where(ScrapRecord.report_id == report.id)) or 0
    if scrap_count:
        # Do not delete a report already used by disposal records. It will be hidden from miniapp sync state
        # by strict matching, and manual correction can handle the dependent disposal data if needed.
        return

    operation = db.get(WorkOrderOperation, report.work_order_operation_id) if report.work_order_operation_id else None
    work_order_id = int(report.work_order_id) if report.work_order_id else None
    if operation:
        operation.report_qty = max(Decimal("0"), to_decimal(operation.report_qty) - to_decimal(report.report_qty))
        operation.good_qty = max(Decimal("0"), to_decimal(operation.good_qty) - to_decimal(report.good_qty))
        operation.scrap_qty = max(Decimal("0"), to_decimal(operation.scrap_qty) - to_decimal(report.scrap_qty))
        operation.rework_qty = max(Decimal("0"), to_decimal(operation.rework_qty) - to_decimal(report.rework_qty))
        report_qty = to_decimal(operation.report_qty)
        planned_qty = to_decimal(operation.planned_qty)
        if planned_qty > 0 and report_qty >= planned_qty:
            operation.status = "DONE"
        elif report_qty > 0:
            operation.status = "IN_PROGRESS"
        else:
            operation.status = "PENDING"
        operation.is_abnormal = 1 if report_qty > 0 and (
            to_decimal(operation.scrap_qty) / report_qty
        ) > to_decimal(operation.allowed_scrap_rate, "0.0001") else 0
        db.add(operation)

    report_no = report.report_no
    db.delete(report)
    db.flush()
    if cache is not None:
        cache["report"][report_no] = None
    if work_order_id:
        sync_work_order_status(db, work_order_id)


def _select_miniapp_issue_candidate(
    db: Session,
    candidates: list[WorkOrderMaterialIssue],
    row: dict,
) -> WorkOrderMaterialIssue | None:
    unsettled_work_order_ids = {
        work_order_id
        for (work_order_id,) in db.execute(
            select(WorkOrder.id).where(
                WorkOrder.id.in_({issue.work_order_id for issue in candidates}),
                _miniapp_syncable_work_order_status_filter(),
            )
        ).all()
    }
    latest_by_work_order: dict[int, WorkOrderMaterialIssue] = {}
    for issue in candidates:
        if issue.work_order_id not in unsettled_work_order_ids:
            continue
        latest_by_work_order.setdefault(issue.work_order_id, issue)
    if len(latest_by_work_order) == 1:
        return next(iter(latest_by_work_order.values()))

    product_name = str(row.get("product_name") or "").strip()
    project_no = str(row.get("project_no") or "").strip()
    if not product_name and not project_no:
        return None

    matched_work_order_ids: set[int] = set()
    product_rows = db.execute(
        select(WorkOrder.id, Item.item_name, Item.item_code)
        .join(Item, Item.id == WorkOrder.product_item_id)
        .where(
            WorkOrder.id.in_(latest_by_work_order),
            _miniapp_syncable_work_order_status_filter(),
        )
    ).all()
    for work_order_id, item_name, item_code in product_rows:
        if product_name and str(item_name or "").strip() == product_name:
            matched_work_order_ids.add(work_order_id)
        if project_no and str(item_code or "").strip() == project_no:
            matched_work_order_ids.add(work_order_id)
    if len(matched_work_order_ids) == 1:
        return latest_by_work_order[next(iter(matched_work_order_ids))]
    return None


def _resolve_work_order_issue_by_inventory_lot_no(
    db: Session,
    lot_no: str,
    row: dict,
) -> tuple[WorkOrderMaterialIssue | None, bool]:
    candidates = db.scalars(
        select(WorkOrderMaterialIssue)
        .join(WorkOrder, WorkOrder.id == WorkOrderMaterialIssue.work_order_id)
        .join(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id)
        .where(
            StockLot.lot_no == lot_no,
            _miniapp_syncable_work_order_status_filter(),
        )
        .order_by(WorkOrderMaterialIssue.id.desc())
    ).all()
    if candidates:
        return _select_miniapp_issue_candidate(db, list(candidates), row), True
    has_filtered_source_lot_candidate = bool(db.scalar(
        select(func.count(WorkOrderMaterialIssue.id))
        .join(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id)
        .where(StockLot.lot_no == lot_no)
    ) or 0)
    candidates = db.scalars(
        select(WorkOrderMaterialIssue)
        .join(WorkOrder, WorkOrder.id == WorkOrderMaterialIssue.work_order_id)
        .where(
            WorkOrderMaterialIssue.material_sub_batch_no == lot_no,
            _miniapp_syncable_work_order_status_filter(),
        )
        .order_by(WorkOrderMaterialIssue.id.desc())
    ).all()
    if candidates:
        return _select_miniapp_issue_candidate(db, list(candidates), row), True
    has_filtered_sub_batch_candidate = bool(db.scalar(
        select(func.count(WorkOrderMaterialIssue.id)).where(WorkOrderMaterialIssue.material_sub_batch_no == lot_no)
    ) or 0)
    if has_filtered_source_lot_candidate or has_filtered_sub_batch_candidate:
        return None, True
    return None, False


def _find_work_order_for_miniapp_item(db: Session, row: dict, cache: dict[str, dict] | None = None) -> WorkOrder | None:
    batch_no = str(row.get("raw_material_batch_no") or "").strip()
    product_name = str(row.get("product_name") or "").strip()
    project_no = str(row.get("project_no") or "").strip()
    cache_key = ("batch", batch_no, product_name, project_no)
    if cache is not None and cache_key in cache["work_order"]:
        return cache["work_order"][cache_key]

    if not batch_no:
        if cache is not None:
            cache["work_order"][cache_key] = None
        return None

    issue, has_issue_candidates = _resolve_work_order_issue_by_inventory_lot_no(db, batch_no, row)
    work_order = db.get(WorkOrder, issue.work_order_id) if issue else None
    if not work_order and _is_miniapp_work_order_short_code(batch_no):
        work_order = _resolve_work_order_by_miniapp_short_code(db, batch_no, row)
    if not work_order and not has_issue_candidates:
        work_order = db.scalar(
            select(WorkOrder)
            .where(
                WorkOrder.work_order_no == batch_no,
                _miniapp_syncable_work_order_status_filter(),
            )
            .limit(1)
        )
    if work_order:
        if cache is not None:
            cache["work_order"][cache_key] = work_order
        return work_order
    if cache is not None:
        cache["work_order"][cache_key] = work_order
    return work_order


def _find_production_ledger_for_miniapp_item(
    db: Session,
    row: dict,
    cache: dict[str, dict] | None = None,
) -> ProductionBatchLedger | None:
    batch_no = str(row.get("raw_material_batch_no") or "").strip()
    product_name = str(row.get("product_name") or "").strip()
    project_no = str(row.get("project_no") or "").strip()
    cache_key = ("production_ledger", batch_no, product_name, project_no)
    if cache is not None and cache_key in cache["production_ledger"]:
        return cache["production_ledger"][cache_key]
    if not batch_no or (not product_name and not project_no):
        if cache is not None:
            cache["production_ledger"][cache_key] = None
        return None
    product_filters = []
    if product_name:
        product_filters.append(Item.item_name == product_name)
    if project_no:
        product_filters.append(Item.item_code == project_no)
    ledger = db.scalar(
        select(ProductionBatchLedger)
        .join(Item, Item.id == ProductionBatchLedger.product_item_id)
        .where(
            ProductionBatchLedger.material_lot_no == batch_no,
            ProductionBatchLedger.status == "在生产",
            ProductionBatchLedger.miniapp_selectable == 1,
            or_(*product_filters),
        )
        .order_by(ProductionBatchLedger.id.desc())
        .limit(1)
    )
    if cache is not None:
        cache["production_ledger"][cache_key] = ledger
    return ledger


def _find_work_order_operation_for_miniapp_item(
    db: Session,
    work_order_id: int,
    process_name: str | None,
    cache: dict[str, dict] | None = None,
    *,
    allow_fallback: bool = True,
) -> WorkOrderOperation | None:
    normalized_process = str(process_name or "").strip()
    cache_key = (work_order_id, normalized_process, allow_fallback)
    if cache is not None and cache_key in cache["operation"]:
        return cache["operation"][cache_key]
    operation = None
    if normalized_process:
        operation = db.scalar(
            select(WorkOrderOperation)
            .where(
                WorkOrderOperation.work_order_id == work_order_id,
                WorkOrderOperation.operation_name == normalized_process,
            )
            .order_by(WorkOrderOperation.seq_no)
            .limit(1)
        )
        if operation:
            if cache is not None:
                cache["operation"][cache_key] = operation
            return operation
    if not allow_fallback:
        if cache is not None:
            cache["operation"][cache_key] = None
        return None
    operation = db.scalar(
        select(WorkOrderOperation)
        .where(WorkOrderOperation.work_order_id == work_order_id)
        .order_by(WorkOrderOperation.seq_no)
        .limit(1)
    )
    if cache is not None:
        cache["operation"][cache_key] = operation
    return operation


def _get_or_create_miniapp_employee(db: Session, row: dict, cache: dict[str, dict] | None = None) -> Employee:
    phone = str(row.get("employee_phone") or "").strip()
    employee_name = str(row.get("employee_name") or "").strip() or phone or "小程序员工"
    cache_key = phone or f"name:{employee_name}"
    if cache is not None and cache_key in cache["employee"]:
        return cache["employee"][cache_key]
    if phone:
        employee = db.scalar(select(Employee).where(Employee.mobile == phone).order_by(Employee.id.desc()).limit(1))
        if employee:
            if employee.employee_name != employee_name:
                employee.employee_name = employee_name
                db.add(employee)
            if cache is not None:
                cache["employee"][cache_key] = employee
            return employee

    department = db.scalar(select(Department).order_by(Department.id).limit(1))
    if not department:
        raise HTTPException(status_code=400, detail="ERP 未配置部门，无法把小程序报工员工映射到 ERP 员工档案")
    employee = Employee(
        employee_code=next_employee_code(db),
        employee_name=employee_name,
        dept_id=department.id,
        mobile=phone or None,
        gender=None,
        hire_date=None,
        job_title="小程序报工员工",
        shift_code=None,
        manager_employee_id=None,
        is_operator=1,
        is_workshop_staff=1,
        status="ACTIVE",
        remark="由小程序报工同步生成",
    )
    db.add(employee)
    db.flush()
    if cache is not None:
        cache["employee"][cache_key] = employee
    return employee


def _sync_miniapp_item_to_erp(
    db: Session,
    source_item_id: int,
    *,
    required: bool = False,
    row: dict | None = None,
    cache: dict[str, dict] | None = None,
    sync_status: bool = True,
) -> OperationReport | None:
    if row is None:
        row = _get_miniapp_report_item_row(db, source_item_id)
    report_no = _miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"]))
    if cache is not None and report_no in cache["report"]:
        report = cache["report"][report_no]
    else:
        report = db.scalar(select(OperationReport).where(OperationReport.report_no == report_no).limit(1))
        if cache is not None:
            cache["report"][report_no] = report
    if str(row.get("status") or "") == "rejected":
        _unlink_stale_miniapp_report(db, report, cache=cache)
        if required:
            raise HTTPException(status_code=400, detail="该小程序报工已驳回，不能同步为 ERP 报工")
        return None

    production_ledger = _find_production_ledger_for_miniapp_item(db, row, cache)
    work_order = None if production_ledger else _find_work_order_for_miniapp_item(db, row, cache)
    if not work_order:
        if not production_ledger:
            _unlink_stale_miniapp_report(db, report, cache=cache)
            if required:
                raise HTTPException(status_code=400, detail="小程序报工缺少有效库存批次号，不能同步为 ERP 报工")
            return None
    operation = None
    if work_order:
        operation = _find_work_order_operation_for_miniapp_item(
            db,
            work_order.id,
            row.get("process_name"),
            cache,
            allow_fallback=not _is_miniapp_work_order_short_code(row.get("raw_material_batch_no")),
        )
        if not operation:
            _unlink_stale_miniapp_report(db, report, cache=cache)
            if required:
                raise HTTPException(status_code=400, detail="该库存批次缺少可匹配的小程序报工工序")
            return None

    employee = _get_or_create_miniapp_employee(db, row, cache)
    start_time = row.get("started_at") or row.get("start_time") or datetime.now()
    end_time = row.get("end_time") or start_time + timedelta(minutes=float(row.get("allocated_minutes") or 0) or 1)
    if end_time <= start_time:
        end_time = start_time + timedelta(minutes=1)

    good_qty = to_decimal(row.get("good_qty"))
    scrap_qty = _miniapp_effective_scrap_qty(row)
    report_qty = good_qty + scrap_qty
    if report_qty <= 0:
        if required:
            raise HTTPException(status_code=400, detail="该小程序报工没有有效产出或报废数量")
        return None

    if production_ledger:
        if report and not _miniapp_report_matches_production_ledger(report, production_ledger):
            _unlink_stale_miniapp_report(db, report, cache=cache)
            report = None
    elif report and not _miniapp_report_matches_target(report, work_order, operation):
        _unlink_stale_miniapp_report(db, report, cache=cache)
        report = None
    old_report_qty = to_decimal(report.report_qty) if report else Decimal("0")
    old_good_qty = to_decimal(report.good_qty) if report else Decimal("0")
    old_scrap_qty = to_decimal(report.scrap_qty) if report else Decimal("0")

    actual_cycle_seconds = _calculate_actual_cycle_seconds(start_time, end_time, report_qty)
    ref_cycle_seconds = to_decimal(row.get("standard_beat"), "0.01")
    abnormal = 0
    if operation and report_qty > 0 and (scrap_qty / report_qty) > to_decimal(operation.allowed_scrap_rate, "0.0001"):
        abnormal = 1
    if actual_cycle_seconds is not None and ref_cycle_seconds > 0 and Decimal(str(actual_cycle_seconds)) > ref_cycle_seconds:
        abnormal = 1

    if not report:
        report = OperationReport(
            report_no=report_no,
            work_order_id=work_order.id if work_order else None,
            work_order_operation_id=operation.id if operation else None,
            production_ledger_id=production_ledger.id if production_ledger else None,
            employee_id=employee.id,
            equipment_id=None,
            report_source="MINIAPP",
            shift_code=None,
            start_time=start_time,
            end_time=end_time,
            report_qty=report_qty,
            good_qty=good_qty,
            scrap_qty=scrap_qty,
            rework_qty=to_decimal(0),
            scrap_reason=None,
            downtime_minutes=to_decimal(0, "0.01"),
            extra_cost_amount=to_decimal(0, "0.01"),
            is_abnormal=abnormal,
            approved_by=None,
            approved_at=None,
            remark=(
                f"小程序报工同步：考勤点={row.get('attendance_point_name') or '-'} "
                f"report_id={row['source_report_id']} item_id={row['source_item_id']}"
            ),
        )
    else:
        report.work_order_id = work_order.id if work_order else None
        report.work_order_operation_id = operation.id if operation else None
        report.production_ledger_id = production_ledger.id if production_ledger else None
        report.employee_id = employee.id
        report.report_source = "MINIAPP"
        report.start_time = start_time
        report.end_time = end_time
        report.report_qty = report_qty
        report.good_qty = good_qty
        report.scrap_qty = scrap_qty
        report.rework_qty = to_decimal(0)
        report.is_abnormal = abnormal
        report.remark = (
            f"小程序报工同步：考勤点={row.get('attendance_point_name') or '-'} "
            f"report_id={row['source_report_id']} item_id={row['source_item_id']}"
        )
    db.add(report)

    if operation and work_order:
        delta_report_qty = report_qty - old_report_qty
        delta_good_qty = good_qty - old_good_qty
        delta_scrap_qty = scrap_qty - old_scrap_qty
        operation.report_qty = max(Decimal("0"), to_decimal(operation.report_qty) + delta_report_qty)
        operation.good_qty = max(Decimal("0"), to_decimal(operation.good_qty) + delta_good_qty)
        operation.scrap_qty = max(Decimal("0"), to_decimal(operation.scrap_qty) + delta_scrap_qty)
        if abnormal:
            operation.is_abnormal = 1
        if to_decimal(operation.report_qty) >= to_decimal(operation.planned_qty):
            operation.status = "DONE"
        elif to_decimal(operation.report_qty) > 0:
            operation.status = "IN_PROGRESS"
        db.add(operation)

        if work_order.actual_start_time is None:
            work_order.actual_start_time = start_time
        db.add(work_order)
        if sync_status:
            sync_work_order_status(db, work_order.id)
    db.flush()
    if cache is not None:
        cache["report"][report_no] = report
    return report


def _build_miniapp_operation_report_read(
    db: Session,
    row: dict,
    report: OperationReport | None = None,
    cache: dict[str, dict] | None = None,
    resolve_targets: bool = True,
) -> MiniAppOperationReportRead:
    if not report:
        report_no = _miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"]))
        if cache is not None and report_no in cache["report"]:
            report = cache["report"][report_no]
        else:
            report = db.scalar(select(OperationReport).where(OperationReport.report_no == report_no).limit(1))
            if cache is not None:
                cache["report"][report_no] = report
    erp_scrap_qty = _miniapp_effective_scrap_qty(row)
    status = str(row.get("status") or "pending")
    if not resolve_targets:
        return MiniAppOperationReportRead(
            source_report_id=int(row["source_report_id"]),
            source_item_id=int(row["source_item_id"]),
            report_no=_miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"])),
            attendance_point_name=str(row.get("attendance_point_name") or ""),
            employee_phone=str(row.get("employee_phone") or ""),
            employee_name=str(row.get("employee_name") or ""),
            report_date=row["report_date"],
            start_time=row["start_time"],
            end_time=row["end_time"],
            duration_minutes=float(row.get("duration_minutes") or 0),
            effective_minutes=float(row.get("effective_minutes") or 0),
            status=status,
            status_name=MINIAPP_REPORT_STATUS_LABELS.get(status, status),
            reviewer_phone=row.get("reviewer_phone"),
            reviewed_at=row.get("reviewed_at"),
            reject_reason=row.get("reject_reason"),
            device_no=str(row.get("device_no") or ""),
            project_no=str(row.get("project_no") or ""),
            product_name=str(row.get("product_name") or ""),
            material_code=row.get("material_code"),
            material_name=row.get("material_name"),
            raw_material_batch_no=row.get("raw_material_batch_no"),
            process_name=row.get("process_name"),
            standard_beat=float(row.get("standard_beat") or 0),
            standard_workload=float(row.get("standard_workload") or 0),
            good_qty=float(row.get("good_qty") or 0),
            defect_qty=float(row.get("defect_qty") or 0),
            scrap_qty=float(row.get("miniapp_scrap_qty") or 0),
            erp_scrap_qty=float(erp_scrap_qty),
            allocated_minutes=float(row.get("allocated_minutes") or 0),
            started_at=row.get("started_at"),
            work_order_id=report.work_order_id if report else None,
            work_order_no=row.get("raw_material_batch_no"),
            work_order_operation_id=report.work_order_operation_id if report else None,
            operation_name=row.get("process_name"),
            erp_report_id=report.id if report else None,
            erp_report_no=report.report_no if report else None,
            is_synced_to_erp=bool(report),
            abnormal_reason=None,
        )
    production_ledger = _find_production_ledger_for_miniapp_item(db, row, cache)
    work_order = None if production_ledger else _find_work_order_for_miniapp_item(db, row, cache)
    operation = (
        _find_work_order_operation_for_miniapp_item(
            db,
            work_order.id,
            row.get("process_name"),
            cache,
            allow_fallback=not _is_miniapp_work_order_short_code(row.get("raw_material_batch_no")),
        )
        if work_order
        else None
    )
    synced_report = (
        report
        if (
            _miniapp_report_matches_production_ledger(report, production_ledger)
            or _miniapp_report_matches_target(report, work_order, operation)
        )
        else None
    )
    report_qty = to_decimal(row.get("good_qty")) + erp_scrap_qty
    abnormal_reason = None
    if report_qty > 0 and operation:
        scrap_rate = erp_scrap_qty / report_qty
        if scrap_rate > to_decimal(operation.allowed_scrap_rate, "0.0001"):
            abnormal_reason = (
                f"报废率超标 {round(float(scrap_rate * 100), 2)}% > "
                f"{round(float(to_decimal(operation.allowed_scrap_rate, '0.0001') * 100), 2)}%"
            )
    return MiniAppOperationReportRead(
        source_report_id=int(row["source_report_id"]),
        source_item_id=int(row["source_item_id"]),
        report_no=_miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"])),
        attendance_point_name=str(row.get("attendance_point_name") or ""),
        employee_phone=str(row.get("employee_phone") or ""),
        employee_name=str(row.get("employee_name") or ""),
        report_date=row["report_date"],
        start_time=row["start_time"],
        end_time=row["end_time"],
        duration_minutes=float(row.get("duration_minutes") or 0),
        effective_minutes=float(row.get("effective_minutes") or 0),
        status=status,
        status_name=MINIAPP_REPORT_STATUS_LABELS.get(status, status),
        reviewer_phone=row.get("reviewer_phone"),
        reviewed_at=row.get("reviewed_at"),
        reject_reason=row.get("reject_reason"),
        device_no=str(row.get("device_no") or ""),
        project_no=str(row.get("project_no") or ""),
        product_name=str(row.get("product_name") or ""),
        material_code=row.get("material_code"),
        material_name=row.get("material_name"),
        raw_material_batch_no=row.get("raw_material_batch_no"),
        process_name=row.get("process_name"),
        standard_beat=float(row.get("standard_beat") or 0),
        standard_workload=float(row.get("standard_workload") or 0),
        good_qty=float(row.get("good_qty") or 0),
        defect_qty=float(row.get("defect_qty") or 0),
        scrap_qty=float(row.get("miniapp_scrap_qty") or 0),
        erp_scrap_qty=float(erp_scrap_qty),
        allocated_minutes=float(row.get("allocated_minutes") or 0),
        started_at=row.get("started_at"),
        work_order_id=work_order.id if work_order else None,
        work_order_no=work_order.work_order_no if work_order else row.get("raw_material_batch_no"),
        work_order_operation_id=operation.id if operation else None,
        operation_name=operation.operation_name if operation else row.get("process_name"),
        erp_report_id=synced_report.id if synced_report else None,
        erp_report_no=synced_report.report_no if synced_report else None,
        is_synced_to_erp=bool(synced_report),
        abnormal_reason=abnormal_reason,
    )


@router.get("/work-order-ledger", response_model=list[WorkOrderLedgerRead])
def list_work_order_ledger(
    work_order_id: int | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WorkOrderLedgerRead]:
    if work_order_id and sync_work_order_reference_qty(db, work_order_id):
        db.commit()
    rows = get_work_order_ledger_rows(db, limit=limit, work_order_id=work_order_id)
    return [WorkOrderLedgerRead.model_validate(row.model_dump()) for row in rows]


@router.get("/production-ledger", response_model=list[ProductionBatchLedgerRead])
def list_production_batch_ledger(
    production_ledger_id: int | None = Query(default=None),
    status: str | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[ProductionBatchLedgerRead]:
    material_item = aliased(Item)
    product_item = aliased(Item)
    query = (
        select(
            ProductionBatchLedger,
            material_item.item_code.label("material_code"),
            material_item.item_name.label("material_name"),
            product_item.item_code.label("product_code"),
            product_item.item_name.label("product_name"),
        )
        .join(material_item, material_item.id == ProductionBatchLedger.material_item_id, isouter=True)
        .join(product_item, product_item.id == ProductionBatchLedger.product_item_id, isouter=True)
        .order_by(ProductionBatchLedger.updated_at.desc(), ProductionBatchLedger.id.desc())
        .limit(limit)
    )
    if production_ledger_id:
        query = query.where(ProductionBatchLedger.id == production_ledger_id)
    if status:
        query = query.where(ProductionBatchLedger.status == status)

    rows = db.execute(query).all()
    ledger_ids = [row.ProductionBatchLedger.id for row in rows]
    material_out_archives = _latest_archive_map(db, DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT, [int(ledger_id) for ledger_id in ledger_ids])
    txns_by_ledger_id: dict[int, list[ProductionBatchLedgerTxn]] = {ledger_id: [] for ledger_id in ledger_ids}
    settlement_archives: dict[int, DocumentArchive] = {}
    settlement_business_id_by_txn_id: dict[int, int] = {}
    if ledger_ids:
        txn_rows = db.scalars(
            select(ProductionBatchLedgerTxn)
            .where(ProductionBatchLedgerTxn.production_ledger_id.in_(ledger_ids))
            .order_by(ProductionBatchLedgerTxn.biz_time.desc(), ProductionBatchLedgerTxn.id.desc())
        ).all()
        for txn in txn_rows:
            txns_by_ledger_id.setdefault(txn.production_ledger_id, []).append(txn)
        settlement_group_order = {"成品入库": 1, "生产余料入库": 2, "生产废料入库": 3}
        settlement_grouped_txns: dict[tuple[int, object], list[ProductionBatchLedgerTxn]] = {}
        for txn in txn_rows:
            if txn.source_doc_type != "库存流水" or not txn.source_doc_id or txn.txn_type not in PRODUCTION_LEDGER_INBOUND_TXN_TYPES:
                continue
            batch_no = production_document_batch_no_from_text(txn.remark)
            group_key = (int(txn.production_ledger_id), f"批次:{batch_no}" if batch_no else txn.biz_time)
            settlement_grouped_txns.setdefault(group_key, []).append(txn)
        for grouped_txns in settlement_grouped_txns.values():
            main_txn = sorted(
                grouped_txns,
                key=lambda item: (settlement_group_order.get(str(item.txn_type or ""), 99), int(item.source_doc_id or 0)),
            )[0]
            main_business_id = int(main_txn.source_doc_id)
            for txn in grouped_txns:
                settlement_business_id_by_txn_id[int(txn.id)] = main_business_id
        settlement_business_ids = list(set(settlement_business_id_by_txn_id.values()))
        settlement_archives = _latest_archive_map(db, DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, settlement_business_ids)

    result: list[ProductionBatchLedgerRead] = []
    for row in rows:
        ledger = row.ProductionBatchLedger
        material_archive = material_out_archives.get(int(ledger.id))
        result.append(
            ProductionBatchLedgerRead(
                production_ledger_id=ledger.id,
                material_lot_id=ledger.material_lot_id,
                material_lot_no=ledger.material_lot_no,
                material_item_id=ledger.material_item_id,
                material_code=row.material_code,
                material_name=row.material_name,
                product_item_id=ledger.product_item_id,
                product_code=row.product_code,
                product_name=row.product_name,
                status=ledger.status,
                miniapp_selectable=bool(ledger.miniapp_selectable),
                total_issued_weight_kg=float(to_decimal(ledger.total_issued_weight_kg)),
                outside_weight_kg=float(to_decimal(ledger.outside_weight_kg)),
                finished_inbound_qty=float(to_decimal(ledger.finished_inbound_qty)),
                surplus_inbound_weight_kg=float(to_decimal(ledger.surplus_inbound_weight_kg)),
                scrap_inbound_weight_kg=float(to_decimal(ledger.scrap_inbound_weight_kg)),
                writeoff_weight_kg=float(to_decimal(ledger.writeoff_weight_kg)),
                first_issue_time=ledger.first_issue_time,
                last_issue_time=ledger.last_issue_time,
                locked_at=ledger.locked_at,
                reopened_at=ledger.reopened_at,
                lock_count=int(ledger.lock_count or 0),
                remark=ledger.remark,
                material_out_archive_status=material_archive.status if material_archive else ARCHIVE_STATUS_MISSING,
                material_out_archive_business_id=ledger.id,
                material_out_archive_document_type=DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT,
                material_out_archive_error_message=material_archive.error_message if material_archive else None,
                txns=[
                    _production_ledger_txn_read(
                        txn,
                        settlement_archives.get(settlement_business_id_by_txn_id[int(txn.id)])
                        if int(txn.id) in settlement_business_id_by_txn_id
                        else None,
                        settlement_business_id_by_txn_id.get(int(txn.id)),
                    )
                    for txn in txns_by_ledger_id.get(ledger.id, [])
                ],
            )
        )
    return result


@router.get("/work-order-inbounds/preview", response_model=ProductionWorkOrderInboundPreviewRead)
def preview_work_order_inbound(
    work_order_id: int = Query(...),
    finished_qty: float | None = Query(default=None),
    db: Session = Depends(get_db),
) -> ProductionWorkOrderInboundPreviewRead:
    return preview_production_work_order_inbound(db, work_order_id=work_order_id, finished_qty=finished_qty)


@router.post("/work-order-inbounds", response_model=ProductionWorkOrderInboundRead)
def create_work_order_inbound(
    payload: ProductionWorkOrderInboundCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> ProductionWorkOrderInboundRead:
    return create_production_work_order_inbound(db, payload=payload, operator_user_id=context.user.id)


@router.get("/production-ledger-inbounds/preview", response_model=ProductionBatchLedgerInboundPreviewRead)
def preview_production_batch_ledger_inbound(
    production_ledger_id: int = Query(...),
    finished_qty: float = Query(default=0),
    db: Session = Depends(get_db),
) -> ProductionBatchLedgerInboundPreviewRead:
    snapshot = calculate_production_batch_ledger_snapshot(
        db,
        production_ledger_id,
        smart_reporting_enabled=False,
        selected_finished_qty=to_decimal(finished_qty),
    )
    material = db.get(Item, snapshot.material_item_id)
    product = db.get(Item, snapshot.product_item_id)
    return ProductionBatchLedgerInboundPreviewRead(
        production_ledger_id=snapshot.production_ledger_id,
        material_lot_no=snapshot.material_lot_no,
        product_item_id=snapshot.product_item_id,
        product_code=product.item_code if product else None,
        product_name=product.item_name if product else None,
        material_item_id=snapshot.material_item_id,
        material_code=material.item_code if material else None,
        material_name=material.item_name if material else None,
        gross_weight_per_piece_kg=float(snapshot.gross_weight_per_piece_kg),
        net_weight_per_piece_kg=float(snapshot.net_weight_per_piece_kg),
        total_issued_weight_kg=float(snapshot.total_issued_weight_kg),
        outside_weight_kg=float(snapshot.outside_weight_kg),
        finished_inbound_qty=float(snapshot.finished_inbound_qty),
        surplus_inbound_weight_kg=float(snapshot.surplus_inbound_weight_kg),
        scrap_inbound_weight_kg=float(snapshot.scrap_inbound_weight_kg),
        writeoff_weight_kg=float(snapshot.writeoff_weight_kg),
        reference_finished_qty=float(snapshot.reference_finished_qty),
        used_material_weight_kg=float(snapshot.used_material_weight_kg),
        theoretical_surplus_weight_kg=float(snapshot.theoretical_surplus_weight_kg),
        theoretical_scrap_weight_kg=float(snapshot.theoretical_scrap_weight_kg),
    )


@router.post("/production-ledger-inbounds", response_model=ProductionBatchLedgerInboundRead)
def create_production_batch_ledger_inbound(
    payload: ProductionBatchLedgerInboundCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> ProductionBatchLedgerInboundRead:
    finished_qty = to_decimal(payload.finished_qty)
    surplus_weight = to_decimal(payload.surplus_weight_kg)
    scrap_weight = to_decimal(payload.scrap_weight_kg)
    if finished_qty <= 0 and surplus_weight <= 0 and scrap_weight <= 0 and not payload.lock_material_batch:
        raise HTTPException(status_code=400, detail="请至少填写一种入库数量或勾选该批材料结单")

    ledger: ProductionBatchLedger | None = db.get(ProductionBatchLedger, payload.production_ledger_id)
    if not ledger:
        raise HTTPException(status_code=404, detail="生产台账不存在")
    source_lot = db.get(StockLot, ledger.material_lot_id)
    material = db.get(Item, ledger.material_item_id)
    product = db.get(Item, ledger.product_item_id)
    if not source_lot or not material or not product:
        raise HTTPException(status_code=400, detail="生产台账来源批次、原材料或产品数据不完整")
    snapshot = calculate_production_batch_ledger_snapshot(
        db,
        payload.production_ledger_id,
        smart_reporting_enabled=False,
        selected_finished_qty=finished_qty,
    )
    deviation_remark = str(payload.deviation_remark or "").strip()
    total_finished_qty = to_decimal(snapshot.finished_inbound_qty) + finished_qty
    reference_finished_qty = to_decimal(snapshot.reference_finished_qty)
    if reference_finished_qty <= 0:
        finished_exceeded = total_finished_qty > 0
    else:
        finished_exceeded = total_finished_qty > reference_finished_qty * Decimal("1.15")

    def _deviation_exceeded(actual: Decimal, theoretical: Decimal) -> bool:
        actual = to_decimal(actual)
        theoretical = to_decimal(theoretical)
        if theoretical == 0:
            return actual != 0
        return abs(actual - theoretical) / abs(theoretical) > Decimal("0.15")

    if finished_exceeded and not deviation_remark:
        raise HTTPException(status_code=400, detail="成品入库数量超过本生产台账参考生产数量上限15%，请填写偏差说明")
    if (
        _deviation_exceeded(surplus_weight, to_decimal(snapshot.theoretical_surplus_weight_kg))
        or _deviation_exceeded(scrap_weight, to_decimal(snapshot.theoretical_scrap_weight_kg))
    ) and not deviation_remark:
        raise HTTPException(status_code=400, detail="余料或废料入库重量与理论值偏差超过上下15%，请填写偏差说明")
    gross_weight = to_decimal(snapshot.gross_weight_per_piece_kg)
    net_weight = to_decimal(snapshot.net_weight_per_piece_kg)
    if net_weight <= 0:
        net_weight = to_decimal(product.unit_weight_kg)
    if net_weight <= 0:
        net_weight = gross_weight
    if gross_weight <= 0 and finished_qty > 0:
        raise HTTPException(status_code=400, detail="产品需用清单缺少单件毛重，不能进行生产台账入库")

    now = datetime.now()
    locked_check_warehouse_ids: set[int] = set()
    finished_warehouse = _get_active_warehouse(db, "FINISHED", "生产台账入库") if finished_qty > 0 else None
    scrap_warehouse = _get_active_warehouse(db, "SCRAP", "生产台账入库") if scrap_weight > 0 else None
    if finished_warehouse:
        locked_check_warehouse_ids.add(int(finished_warehouse.id))
    if surplus_weight > 0:
        locked_check_warehouse_ids.add(int(source_lot.warehouse_id))
    if scrap_warehouse:
        locked_check_warehouse_ids.add(int(scrap_warehouse.id))
    ensure_warehouses_unlocked(db, locked_check_warehouse_ids, "生产台账入库")

    finished_lot_id = surplus_lot_id = scrap_lot_id = None
    finished_txn_id = surplus_txn_id = scrap_txn_id = None
    base_remark = payload.remark or "生产台账入库"
    if payload.deviation_remark:
        base_remark = f"{base_remark}；偏差说明：{payload.deviation_remark}"
    document_batch_no = _production_document_batch_no(now) if (finished_qty > 0 or surplus_weight > 0 or scrap_weight > 0) else None

    if finished_qty > 0:
        receipt_weight = to_decimal(finished_qty * net_weight)
        unit_cost = to_decimal(to_decimal(source_lot.unit_cost) * gross_weight, "0.0001")
        lot_no = _build_production_ledger_lot_no(db, "FGL", ledger, product, now)
        source_summary = f"{material.item_name}：{ledger.material_lot_no}"
        finished_lot = StockLot(
            lot_no=lot_no,
            parent_lot_id=source_lot.id,
            lot_role="FINISHED_FROM_RAW_BATCH",
            material_sub_batch_no=None,
            item_id=product.id,
            warehouse_id=finished_warehouse.id,
            location_id=None,
            source_doc_type="PRODUCTION_LEDGER_IN",
            source_doc_id=ledger.id,
            source_line_id=None,
            source_material_lot_id=source_lot.id,
            source_material_sub_batch_no=ledger.material_lot_no,
            source_material_summary=source_summary,
            inbound_qty=finished_qty,
            inbound_weight_kg=receipt_weight,
            remaining_qty=finished_qty,
            remaining_weight_kg=receipt_weight,
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=unit_cost,
            production_date=now.date(),
            expire_date=None,
            quality_status="PASS",
            status="AVAILABLE",
            remark=_with_document_batch_no(f"{base_remark}；成品入库；来源库存批次已记录", document_batch_no),
        )
        db.add(finished_lot)
        db.flush()
        upsert_stock_balance(
            db,
            item_id=product.id,
            warehouse_id=finished_warehouse.id,
            location_id=None,
            qty_delta=finished_qty,
            weight_delta=receipt_weight,
            available_qty_delta=finished_qty,
            available_weight_delta=receipt_weight,
            unit_cost=unit_cost,
        )
        finished_txn = create_inventory_txn(
            db,
            txn_type="FG_IN",
            item_id=product.id,
            warehouse_id=finished_warehouse.id,
            location_id=None,
            lot_id=finished_lot.id,
            qty_change=finished_qty,
            weight_change=receipt_weight,
            unit_cost=unit_cost,
            source_doc_type="生产台账",
            source_doc_id=ledger.id,
            source_line_id=finished_lot.id,
            biz_time=now,
            operator_user_id=context.user.id,
            remark=_with_document_batch_no(f"{base_remark}；成品入库", document_batch_no),
            amount_basis="QTY",
        )
        finished_lot_id = finished_lot.id
        finished_txn_id = finished_txn.id
        ledger = post_finished_inbound_to_ledger(
            db,
            production_ledger_id=payload.production_ledger_id,
            finished_qty=finished_qty,
            operator_user_id=context.user.id,
            source_doc_type="库存流水",
            source_doc_id=finished_txn.id,
            source_line_id=finished_lot.id,
            biz_time=now,
            remark=_with_document_batch_no(f"{base_remark}；成品入库联动生产台账", document_batch_no),
        )
    if surplus_weight > 0:
        source_lot.remaining_weight_kg = to_decimal(source_lot.remaining_weight_kg) + surplus_weight
        source_lot.status = "AVAILABLE"
        db.add(source_lot)
        upsert_stock_balance(
            db,
            item_id=material.id,
            warehouse_id=source_lot.warehouse_id,
            location_id=source_lot.location_id,
            qty_delta=Decimal("0"),
            weight_delta=surplus_weight,
            available_qty_delta=Decimal("0"),
            available_weight_delta=surplus_weight,
            unit_cost=to_decimal(source_lot.unit_cost),
        )
        surplus_txn = create_inventory_txn(
            db,
            txn_type="PRODUCTION_SURPLUS_IN",
            item_id=material.id,
            warehouse_id=source_lot.warehouse_id,
            location_id=source_lot.location_id,
            lot_id=source_lot.id,
            qty_change=Decimal("0"),
            weight_change=surplus_weight,
            unit_cost=to_decimal(source_lot.unit_cost),
            source_doc_type="生产台账",
            source_doc_id=ledger.id,
            source_line_id=source_lot.id,
            biz_time=now,
            operator_user_id=context.user.id,
            remark=_with_document_batch_no(f"{base_remark}；生产余料入库", document_batch_no),
            amount_basis="WEIGHT",
        )
        surplus_lot_id = source_lot.id
        surplus_txn_id = surplus_txn.id
        ledger = post_surplus_inbound_to_ledger(
            db,
            production_ledger_id=payload.production_ledger_id,
            surplus_weight_kg=surplus_weight,
            operator_user_id=context.user.id,
            source_doc_type="库存流水",
            source_doc_id=surplus_txn.id,
            source_line_id=source_lot.id,
            biz_time=now,
            remark=_with_document_batch_no(f"{base_remark}；生产余料入库联动生产台账", document_batch_no),
        )
    if scrap_weight > 0:
        scrap_lot_no = _build_production_ledger_lot_no(db, "SCRAPI", ledger, material, now)
        scrap_lot = StockLot(
            lot_no=scrap_lot_no,
            parent_lot_id=source_lot.id,
            lot_role="PRODUCTION_SCRAP",
            material_sub_batch_no=None,
            item_id=material.id,
            warehouse_id=scrap_warehouse.id,
            location_id=None,
            source_doc_type="PRODUCTION_SCRAP_IN",
            source_doc_id=ledger.id,
            source_line_id=None,
            source_material_lot_id=source_lot.id,
            source_material_sub_batch_no=ledger.material_lot_no,
            source_material_summary=f"{material.item_name}：{ledger.material_lot_no}",
            inbound_qty=Decimal("0"),
            inbound_weight_kg=scrap_weight,
            remaining_qty=Decimal("0"),
            remaining_weight_kg=scrap_weight,
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=to_decimal(source_lot.unit_cost),
            production_date=now.date(),
            expire_date=None,
            quality_status="SCRAP",
            status="AVAILABLE",
            remark=_with_document_batch_no(f"{base_remark}；生产废料入库；来源库存批次已记录", document_batch_no),
        )
        db.add(scrap_lot)
        db.flush()
        upsert_stock_balance(
            db,
            item_id=material.id,
            warehouse_id=scrap_warehouse.id,
            location_id=None,
            qty_delta=Decimal("0"),
            weight_delta=scrap_weight,
            available_qty_delta=Decimal("0"),
            available_weight_delta=scrap_weight,
            unit_cost=to_decimal(source_lot.unit_cost),
        )
        scrap_txn = create_inventory_txn(
            db,
            txn_type="PRODUCTION_SCRAP_IN",
            item_id=material.id,
            warehouse_id=scrap_warehouse.id,
            location_id=None,
            lot_id=scrap_lot.id,
            qty_change=Decimal("0"),
            weight_change=scrap_weight,
            unit_cost=to_decimal(source_lot.unit_cost),
            source_doc_type="生产台账",
            source_doc_id=ledger.id,
            source_line_id=scrap_lot.id,
            biz_time=now,
            operator_user_id=context.user.id,
            remark=_with_document_batch_no(f"{base_remark}；生产废料入库", document_batch_no),
            amount_basis="WEIGHT",
        )
        scrap_lot_id = scrap_lot.id
        scrap_txn_id = scrap_txn.id
        ledger = post_scrap_inbound_to_ledger(
            db,
            production_ledger_id=payload.production_ledger_id,
            scrap_weight_kg=scrap_weight,
            operator_user_id=context.user.id,
            source_doc_type="库存流水",
            source_doc_id=scrap_txn.id,
            source_line_id=scrap_lot.id,
            biz_time=now,
            remark=_with_document_batch_no(f"{base_remark}；生产废料入库联动生产台账", document_batch_no),
        )
    if payload.lock_material_batch:
        if not payload.lock_confirmed:
            raise HTTPException(status_code=400, detail="请确认该批材料结单后再提交")
        ledger = lock_production_batch_ledger(
            db,
            production_ledger_id=payload.production_ledger_id,
            operator_user_id=context.user.id,
            biz_time=now,
            remark=payload.deviation_remark or payload.remark or "生产台账入库后该批材料结单",
        )
    response_ledger_id = int(ledger.id)
    response_material_lot_no = str(ledger.material_lot_no)
    response_status = str(ledger.status)
    main_txn_id = finished_txn_id or surplus_txn_id or scrap_txn_id
    db.commit()

    archive_result: DocumentArchiveGenerateResult | None = None
    if main_txn_id:
        try:
            archive_result = generate_document_archive(
                db,
                DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT,
                int(main_txn_id),
                created_by=context.user.id,
            )
        except Exception as exc:
            archive_result = _failed_archive_result(
                DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT,
                int(main_txn_id),
                f"库存流水#{main_txn_id}",
                exc,
            )

    response = ProductionBatchLedgerInboundRead(
        production_ledger_id=response_ledger_id,
        material_lot_no=response_material_lot_no,
        finished_qty=float(finished_qty),
        surplus_weight_kg=float(surplus_weight),
        scrap_weight_kg=float(scrap_weight),
        finished_lot_id=finished_lot_id,
        surplus_lot_id=surplus_lot_id,
        scrap_lot_id=scrap_lot_id,
        finished_txn_id=finished_txn_id,
        surplus_txn_id=surplus_txn_id,
        scrap_txn_id=scrap_txn_id,
        locked=bool(payload.lock_material_batch),
        status=response_status,
        remark=payload.remark,
    )
    for key, value in _archive_fields_from_result(
        archive_result,
        document_type=DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT,
        business_id=int(main_txn_id) if main_txn_id else None,
    ).items():
        setattr(response, key, value)
    return response


@router.get("/work-orders", response_model=list[WorkOrderRead])
def list_work_orders(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WorkOrderRead]:
    if sync_recent_work_order_reference_qty(db, limit=limit):
        db.commit()
    rows = db.execute(get_work_orders_query(limit=limit)).mappings().all()
    return [WorkOrderRead.model_validate(dict(row)) for row in rows]


@router.get("/work-order-materials", response_model=list[WorkOrderMaterialRead])
def list_work_order_materials(
    work_order_id: int | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WorkOrderMaterialRead]:
    rows = db.execute(get_work_order_materials_query(limit=limit, work_order_id=work_order_id)).mappings().all()
    return [WorkOrderMaterialRead.model_validate(dict(row)) for row in rows]


@router.get("/work-order-material-issues", response_model=list[WorkOrderMaterialIssueRead])
def list_work_order_material_issues(
    work_order_id: int | None = Query(default=None),
    limit: int = Query(default=300, ge=1, le=800),
    db: Session = Depends(get_db),
) -> list[WorkOrderMaterialIssueRead]:
    rows = db.execute(get_work_order_material_issues_query(limit=limit, work_order_id=work_order_id)).mappings().all()
    return [WorkOrderMaterialIssueRead.model_validate(dict(row)) for row in rows]


@router.get("/work-order-operations", response_model=list[WorkOrderOperationRead])
def list_work_order_operations(
    work_order_id: int | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WorkOrderOperationRead]:
    if work_order_id and sync_work_order_reference_qty(db, work_order_id):
        db.commit()
    rows = db.execute(get_work_order_operations_query(limit=limit, work_order_id=work_order_id)).mappings().all()
    return [WorkOrderOperationRead.model_validate(dict(row)) for row in rows]


@router.post("/work-orders", response_model=WorkOrderRead)
def create_work_order(
    payload: WorkOrderCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> WorkOrderRead:
    product = db.get(Item, payload.product_item_id)
    if not product:
        raise HTTPException(status_code=404, detail="产品不存在")
    if payload.source_sales_order_item_id:
        source_order_item = db.get(SalesOrderItem, payload.source_sales_order_item_id)
        if not source_order_item:
            raise HTTPException(status_code=404, detail="来源订单明细不存在")

    selected_lots = [
        {
            "source_lot_id": int(row.source_lot_id),
            "material_item_id": int(row.material_item_id) if row.material_item_id else None,
            "issued_weight_kg": to_decimal(row.issued_weight_kg),
        }
        for row in payload.selected_stock_lots
    ]
    if not selected_lots:
        raise HTTPException(status_code=400, detail="生产出库必须选择库存批次号")
    if len(selected_lots) != 1:
        raise HTTPException(status_code=400, detail="一次生产出库只能选择一个材料库存批次号")
    route = get_default_route(db, payload.product_item_id)
    bom = get_default_bom(db, payload.product_item_id)
    bom_items = db.scalars(select(BomItem).where(BomItem.bom_id == bom.id).order_by(BomItem.seq_no)).all()
    if not bom_items:
        raise HTTPException(status_code=400, detail="产品需规缺少默认用料，无法按库存批次出库")

    selected_lot_payload = selected_lots[0]
    selected_lot = db.scalar(
        select(StockLot)
        .where(StockLot.id == int(selected_lot_payload["source_lot_id"]))
        .with_for_update()
    )
    if not selected_lot:
        raise HTTPException(status_code=400, detail="库存批次不存在")
    selected_material_item_id = int(selected_lot_payload["material_item_id"] or selected_lot.item_id)
    if selected_material_item_id != int(selected_lot.item_id):
        raise HTTPException(status_code=400, detail=f"库存批次 {selected_lot.lot_no} 与选择的原材料不匹配")
    selected_lot_payload["material_item_id"] = selected_material_item_id
    bom_item_by_material_id = {int(item.material_item_id): item for item in bom_items}
    selected_bom_item = bom_item_by_material_id.get(selected_material_item_id)
    if not selected_bom_item:
        raise HTTPException(status_code=400, detail=f"库存批次 {selected_lot.lot_no} 不属于该产品需规默认用料")
    ensure_warehouses_unlocked(db, {int(selected_lot.warehouse_id)}, "生产出库")

    issue_weight = to_decimal(selected_lot_payload["issued_weight_kg"])
    if issue_weight <= 0:
        raise HTTPException(status_code=400, detail="出库重量必须大于0")

    planned_qty = to_decimal(payload.planned_qty)
    if issue_weight > 0:
        gross_weight = to_decimal(selected_bom_item.gross_weight_per_piece_kg)
        if gross_weight <= 0:
            gross_weight = to_decimal(product.unit_weight_kg)
        if gross_weight <= 0:
            raise HTTPException(status_code=400, detail="产品需用清单缺少单件毛重/净重，无法按领料重量计算参考生产数量")
        planned_qty = (issue_weight / gross_weight).to_integral_value(rounding=ROUND_FLOOR)
    if planned_qty <= 0:
        raise HTTPException(status_code=400, detail="参考生产数量必须大于0")
    biz_time = datetime.now()
    ledger = record_production_issue_to_ledger(
        db,
        material_lot_id=selected_lot.id,
        product_item_id=payload.product_item_id,
        issued_weight_kg=issue_weight,
        operator_user_id=context.user.id,
        source_doc_type="库存流水",
        source_doc_id=None,
        source_line_id=None,
        biz_time=biz_time,
        remark=payload.remark or "原材料库生产出库形成生产台账",
    )
    inventory_txn = _issue_raw_lot_to_production_ledger(
        db,
        lot=selected_lot,
        material_item_id=selected_material_item_id,
        issued_weight=issue_weight,
        production_ledger_id=ledger.id,
        operator_user_id=context.user.id,
        biz_time=biz_time,
    )
    _link_latest_issue_ledger_txn_to_inventory_txn(db, production_ledger_id=ledger.id, inventory_txn=inventory_txn)
    production_ledger_id = int(ledger.id)
    production_material_lot_no = str(ledger.material_lot_no)
    production_ledger_status = str(ledger.status)
    operation_count = db.scalar(select(func.count(ProcessRouteOperation.id)).where(ProcessRouteOperation.route_id == route.id)) or 0
    response = _build_production_ledger_work_order_compat_read(
        production_ledger_id=production_ledger_id,
        material_lot_no=production_material_lot_no,
        product=product,
        route=route,
        bom=bom,
        planned_qty=planned_qty,
        payload=payload,
        biz_time=biz_time,
        status=production_ledger_status,
        operation_count=int(operation_count),
    )
    db.commit()

    archive_result: DocumentArchiveGenerateResult | None = None
    try:
        archive_result = generate_document_archive(
            db,
            DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT,
            production_ledger_id,
            created_by=context.user.id,
        )
    except Exception as exc:
        archive_result = _failed_archive_result(
            DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT,
            production_ledger_id,
            f"生产台账#{production_ledger_id}",
            exc,
        )
    for key, value in _archive_fields_from_result(
        archive_result,
        document_type=DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT,
        business_id=production_ledger_id,
    ).items():
        setattr(response, key, value)
    return response


@router.get("/miniapp-operation-reports/sync-state", response_model=MiniAppOperationReportSyncStateRead)
def get_miniapp_operation_report_sync_state(
    db: Session = Depends(get_db),
) -> MiniAppOperationReportSyncStateRead:
    return MiniAppOperationReportSyncStateRead(last_synced_at=get_miniapp_operation_report_last_sync(db))


@router.post("/miniapp-operation-reports/sync", response_model=MiniAppOperationReportSyncRead)
def sync_miniapp_operation_reports(
    db: Session = Depends(get_db),
) -> MiniAppOperationReportSyncRead:
    lock_miniapp_operation_report_last_sync(db)
    sync_from = get_miniapp_operation_report_last_sync(db)
    sync_to = datetime.now()
    rows = _miniapp_report_item_rows(
        db,
        limit=None,
        updated_after=sync_from,
        updated_until=sync_to,
        order_by_updated_asc=True,
    )
    cache = _new_miniapp_resolve_cache()
    report_nos = [_miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"])) for row in rows]
    _preload_miniapp_operation_reports(db, report_nos, cache)

    touched_work_order_ids: set[int] = set()
    synced = 0
    skipped = 0
    for row in rows:
        synced_report = _sync_miniapp_item_to_erp(
            db,
            int(row["source_item_id"]),
            required=False,
            row=row,
            cache=cache,
            sync_status=False,
        )
        if synced_report:
            synced += 1
            if synced_report.work_order_id:
                touched_work_order_ids.add(int(synced_report.work_order_id))
        else:
            skipped += 1

    for work_order_id in touched_work_order_ids:
        sync_work_order_status(db, work_order_id)
    row = upsert_miniapp_operation_report_last_sync(db, sync_to)
    db.commit()
    db.refresh(row)
    return MiniAppOperationReportSyncRead(
        sync_from=sync_from,
        sync_to=sync_to,
        last_synced_at=sync_to,
        total=len(rows),
        synced=synced,
        skipped=skipped,
        failed=0,
    )


@router.get("/miniapp-operation-reports", response_model=list[MiniAppOperationReportRead])
def list_miniapp_operation_reports(
    limit: int = Query(default=300, ge=1, le=800),
    sync_to_erp: bool = Query(default=True),
    start_date: date | None = Query(default=None),
    end_date: date | None = Query(default=None),
    db: Session = Depends(get_db),
) -> list[MiniAppOperationReportRead]:
    if start_date and end_date and start_date > end_date:
        raise HTTPException(status_code=400, detail="开始日期不能晚于结束日期")
    rows = _miniapp_report_item_rows(db, limit=limit, start_date=start_date, end_date=end_date)
    cache = _new_miniapp_resolve_cache()
    report_nos = [_miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"])) for row in rows]
    _preload_miniapp_operation_reports(db, report_nos, cache)
    result: list[MiniAppOperationReportRead] = []
    touched_work_order_ids: set[int] = set()
    for row in rows:
        synced_report = None
        if sync_to_erp:
            synced_report = _sync_miniapp_item_to_erp(
                db,
                int(row["source_item_id"]),
                required=False,
                row=row,
                cache=cache,
                sync_status=False,
            )
            if synced_report and synced_report.work_order_id:
                touched_work_order_ids.add(int(synced_report.work_order_id))
        result.append(_build_miniapp_operation_report_read(db, row, synced_report, cache=cache, resolve_targets=sync_to_erp))
    if sync_to_erp:
        for work_order_id in touched_work_order_ids:
            sync_work_order_status(db, work_order_id)
        db.commit()
    return result


@router.get("/operation-reports", response_model=list[OperationReportRead])
def list_operation_reports(
    work_order_id: int | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[OperationReportRead]:
    rows = db.execute(get_operation_reports_query(limit=limit, work_order_id=work_order_id)).mappings().all()
    return [_build_operation_report_read(dict(row)) for row in rows]


@router.post("/miniapp-scrap-records", response_model=ScrapRecordRead)
def create_miniapp_scrap_record(payload: MiniAppScrapRecordCreate, db: Session = Depends(get_db)) -> ScrapRecordRead:
    report = _sync_miniapp_item_to_erp(db, payload.source_item_id, required=True)
    if not report:
        raise HTTPException(status_code=400, detail="小程序报工无法同步为 ERP 报工")
    work_order = db.get(WorkOrder, report.work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="库存批次号不存在")
    source_row = _get_miniapp_report_item_row(db, payload.source_item_id)
    source_scrap_qty = _miniapp_effective_scrap_qty(source_row)
    scrap_qty = to_decimal(payload.scrap_qty) if to_decimal(payload.scrap_qty) > 0 else source_scrap_qty
    scrap_payload = ScrapRecordCreate(
        report_id=report.id,
        work_order_id=report.work_order_id,
        work_order_operation_id=report.work_order_operation_id,
        item_id=work_order.product_item_id,
        scrap_qty=float(scrap_qty),
        scrap_weight_kg=payload.scrap_weight_kg,
        scrap_reason=payload.scrap_reason,
        disposal_type="SCRAP",
        saleable_scrap_weight_kg=payload.saleable_scrap_weight_kg,
        scrap_cost_amount=payload.scrap_cost_amount,
    )
    return create_scrap_record(scrap_payload, db)


@router.post("/operation-reports", response_model=OperationReportRead)
def create_operation_report(
    payload: OperationReportCreate,
    db: Session = Depends(get_db),
) -> OperationReportRead:
    work_order = db.get(WorkOrder, payload.work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="工单不存在")
    operation = db.get(WorkOrderOperation, payload.work_order_operation_id)
    if not operation or operation.work_order_id != payload.work_order_id:
        raise HTTPException(status_code=400, detail="工单工序不存在或不属于当前工单")
    employee = db.get(Employee, payload.employee_id)
    if not employee:
        raise HTTPException(status_code=404, detail="员工不存在")
    if payload.equipment_id:
        equipment = db.get(Equipment, payload.equipment_id)
        if not equipment:
            raise HTTPException(status_code=404, detail="设备不存在")
    if payload.end_time <= payload.start_time:
        raise HTTPException(status_code=400, detail="报工结束时间必须晚于开始时间")

    report_qty = to_decimal(payload.report_qty)
    good_qty = to_decimal(payload.good_qty)
    scrap_qty = to_decimal(payload.scrap_qty)
    rework_qty = to_decimal(payload.rework_qty)
    if report_qty <= 0:
        raise HTTPException(status_code=400, detail="报工数量必须大于0")
    if good_qty < 0 or scrap_qty < 0 or rework_qty < 0:
        raise HTTPException(status_code=400, detail="报工数量、合格数、报废数、返工数不能小于0")
    if good_qty + scrap_qty + rework_qty > report_qty:
        raise HTTPException(status_code=400, detail="合格数、报废数、返工数合计不能超过报工数量")

    route_operation = db.get(ProcessRouteOperation, operation.route_operation_id) if operation.route_operation_id else None
    ref_cycle_seconds = to_decimal(route_operation.std_cycle_seconds if route_operation else 0, "0.01")
    actual_cycle_seconds = _calculate_actual_cycle_seconds(payload.start_time, payload.end_time, report_qty)
    abnormal_reasons: list[str] = []
    if report_qty > 0 and (scrap_qty / report_qty) > to_decimal(operation.allowed_scrap_rate, "0.0001"):
        abnormal_reasons.append("报废率超出工序允许值")
    if actual_cycle_seconds is not None and ref_cycle_seconds > 0 and Decimal(str(actual_cycle_seconds)) > ref_cycle_seconds:
        abnormal_reasons.append("实际节拍慢于参考节拍")
    abnormal = 1 if abnormal_reasons else 0

    report = OperationReport(
        report_no=_build_operation_report_no(db, work_order, operation, employee),
        work_order_id=payload.work_order_id,
        work_order_operation_id=payload.work_order_operation_id,
        employee_id=payload.employee_id,
        equipment_id=payload.equipment_id,
        report_source=payload.report_source,
        shift_code=payload.shift_code,
        start_time=payload.start_time,
        end_time=payload.end_time,
        report_qty=report_qty,
        good_qty=good_qty,
        scrap_qty=scrap_qty,
        rework_qty=rework_qty,
        scrap_reason=payload.scrap_reason,
        downtime_minutes=to_decimal(payload.downtime_minutes, "0.01"),
        extra_cost_amount=to_decimal(payload.extra_cost_amount, "0.01"),
        is_abnormal=abnormal,
        approved_by=None,
        approved_at=None,
        remark=payload.remark,
    )
    db.add(report)

    operation.report_qty = to_decimal(operation.report_qty) + report_qty
    operation.good_qty = to_decimal(operation.good_qty) + good_qty
    operation.scrap_qty = to_decimal(operation.scrap_qty) + scrap_qty
    operation.rework_qty = to_decimal(operation.rework_qty) + rework_qty
    if abnormal:
        operation.is_abnormal = 1
    if operation.report_qty > 0 and (to_decimal(operation.scrap_qty) / to_decimal(operation.report_qty)) > to_decimal(
        operation.allowed_scrap_rate, "0.0001"
    ):
        operation.is_abnormal = 1
    if to_decimal(operation.report_qty) >= to_decimal(operation.planned_qty):
        operation.status = "DONE"
    else:
        operation.status = "IN_PROGRESS"
    db.add(operation)

    if work_order.actual_start_time is None:
        work_order.actual_start_time = payload.start_time
    db.add(work_order)
    sync_work_order_status(db, work_order.id)
    db.commit()

    row = db.execute(get_operation_reports_query(limit=1).where(OperationReport.id == report.id)).mappings().first()
    if not row:
        raise HTTPException(status_code=500, detail="报工记录创建后读取失败")
    return _build_operation_report_read(dict(row))


@router.get("/scrap-records", response_model=list[ScrapRecordRead])
def list_scrap_records(
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[ScrapRecordRead]:
    rows = db.execute(get_scrap_records_query(limit=limit)).mappings().all()
    return [ScrapRecordRead.model_validate(dict(row)) for row in rows]


@router.post("/scrap-records", response_model=ScrapRecordRead)
def create_scrap_record(payload: ScrapRecordCreate, db: Session = Depends(get_db)) -> ScrapRecordRead:
    report = db.get(OperationReport, payload.report_id)
    if not report:
        raise HTTPException(status_code=404, detail="报工记录不存在")
    if report.work_order_id != payload.work_order_id or report.work_order_operation_id != payload.work_order_operation_id:
        raise HTTPException(status_code=400, detail="处置记录与来源报工不匹配")
    work_order = db.get(WorkOrder, payload.work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="工单不存在")
    operation = db.get(WorkOrderOperation, payload.work_order_operation_id)
    if not operation or operation.work_order_id != payload.work_order_id:
        raise HTTPException(status_code=404, detail="工序不存在")
    item = db.get(Item, payload.item_id)
    if not item:
        raise HTTPException(status_code=404, detail="物料不存在")
    disposal_type = str(payload.disposal_type or "SCRAP").strip().upper()
    if disposal_type != "SCRAP":
        raise HTTPException(status_code=400, detail="生产过程只登记报废；客户退货返工请在发货概览模块处理")

    existing_active_record = _get_active_scrap_record(db, payload.report_id)
    if existing_active_record:
        _revert_scrap_record_effect(work_order, existing_active_record)
        existing_active_record.status = "OVERRIDDEN"
        db.add(existing_active_record)

    source_qty = to_decimal(report.rework_qty if disposal_type == "REWORK" and to_decimal(report.rework_qty) > 0 else report.scrap_qty)
    scrap_qty = to_decimal(payload.scrap_qty) if to_decimal(payload.scrap_qty) > 0 else source_qty
    if scrap_qty <= 0:
        raise HTTPException(status_code=400, detail="当前来源报工没有可处置的报废/返工数量")
    scrap_weight = to_decimal(payload.scrap_weight_kg)
    if scrap_weight <= 0:
        scrap_weight = scrap_qty * to_decimal(item.unit_weight_kg)

    saleable_scrap_weight = Decimal("0")
    scrap_cost_amount = Decimal("0")
    if disposal_type == "SCRAP":
        saleable_scrap_weight = (
            to_decimal(payload.saleable_scrap_weight_kg)
            if to_decimal(payload.saleable_scrap_weight_kg) > 0
            else scrap_weight
        )
        scrap_cost_amount = to_decimal(payload.scrap_cost_amount, "0.01")
        if scrap_cost_amount <= 0 and saleable_scrap_weight > 0:
            scrap_cost_amount = to_decimal(_get_average_issued_weight_cost(db, payload.work_order_id) * saleable_scrap_weight, "0.01")
    else:
        scrap_cost_amount = to_decimal(scrap_qty * to_decimal(operation.std_labor_cost, "0.0001"), "0.01")

    record = ScrapRecord(
        scrap_no=_build_scrap_no(db, work_order, disposal_type),
        report_id=payload.report_id,
        work_order_id=payload.work_order_id,
        work_order_operation_id=payload.work_order_operation_id,
        item_id=payload.item_id,
        scrap_qty=scrap_qty,
        scrap_weight_kg=scrap_weight,
        scrap_reason=payload.scrap_reason or report.scrap_reason,
        disposal_type=disposal_type,
        saleable_scrap_weight_kg=saleable_scrap_weight,
        scrap_cost_amount=scrap_cost_amount,
        status="CONFIRMED",
    )
    db.add(record)

    if disposal_type == "SCRAP":
        work_order.scrap_qty = to_decimal(work_order.scrap_qty) + scrap_qty
    db.add(work_order)
    sync_work_order_status(db, work_order.id)
    db.commit()

    row = db.execute(get_scrap_records_query(limit=1).where(ScrapRecord.id == record.id)).mappings().first()
    if not row:
        raise HTTPException(status_code=500, detail="报废/返工记录创建后读取失败")
    return ScrapRecordRead.model_validate(dict(row))


@router.get("/completion-receipts", response_model=list[CompletionReceiptRead])
def list_completion_receipts(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[CompletionReceiptRead]:
    rows = db.execute(get_completion_receipts_query(limit=limit)).mappings().all()
    return [CompletionReceiptRead.model_validate(dict(row)) for row in rows]


@router.get("/completion-items", response_model=list[CompletionReceiptItemRead])
def list_completion_receipt_items(
    receipt_id: int | None = Query(default=None),
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[CompletionReceiptItemRead]:
    rows = db.execute(get_completion_receipt_items_query(limit=limit, receipt_id=receipt_id)).mappings().all()
    return [CompletionReceiptItemRead.model_validate(dict(row)) for row in rows]


def _build_work_order_source_material_summary(db: Session, work_order_id: int) -> dict[str, object | None]:
    rows = db.execute(get_work_order_material_issues_query(limit=500, work_order_id=work_order_id)).mappings().all()
    if not rows:
        return {
            "source_material_lot_id": None,
            "source_material_sub_batch_no": None,
            "source_material_summary": None,
            "source_material_weight_kg": Decimal("0"),
            "source_material_amount": Decimal("0"),
            "source_material_unit_cost": Decimal("0"),
        }
    first = rows[0]
    parts: list[str] = []
    total_weight = Decimal("0")
    total_amount = Decimal("0")
    for row in rows:
        weight = to_decimal(row["issued_weight_kg"])
        amount = to_decimal(row["amount"], "0.01")
        source_lot_no = row["source_lot_no"] or row["material_sub_batch_no"]
        parts.append(f"{source_lot_no} / {row['material_name']} / {weight}kg / ¥{amount}")
        if weight > 0:
            total_weight += weight
            total_amount += amount
    first_lot_no = first["source_lot_no"] or first["material_sub_batch_no"]
    material_unit_cost = (total_amount / total_weight) if total_weight > 0 else Decimal("0")
    return {
        "source_material_lot_id": first["source_lot_id"],
        "source_material_sub_batch_no": first_lot_no,
        "source_material_summary": "；".join(parts)[:1000],
        "source_material_weight_kg": to_decimal(total_weight, "0.000001"),
        "source_material_amount": to_decimal(total_amount, "0.01"),
        "source_material_unit_cost": to_decimal(material_unit_cost, "0.0001"),
    }


def _is_rework_work_order(work_order: WorkOrder) -> bool:
    return str(work_order.work_order_type or "").upper() == "REWORK"


def _build_rework_source_material_summary(db: Session, work_order_id: int) -> dict[str, object | None]:
    row = db.execute(
        select(
            ReturnDisposition.id.label("return_disposition_id"),
            ReturnItem.source_lot_id.label("source_lot_id"),
        )
        .join(ReturnItem, ReturnItem.id == ReturnDisposition.return_item_id)
        .where(ReturnDisposition.rework_work_order_id == work_order_id)
        .limit(1)
    ).mappings().first()
    if not row:
        raise HTTPException(status_code=400, detail="该返工工单缺少退货返工来源，不能返工入库")

    finished_lot = db.get(StockLot, row["source_lot_id"]) if row["source_lot_id"] else None
    raw_lot = db.get(StockLot, finished_lot.source_material_lot_id) if finished_lot and finished_lot.source_material_lot_id else None
    source_material_lot_id = finished_lot.source_material_lot_id if finished_lot else None
    source_material_sub_batch_no = (
        finished_lot.source_material_sub_batch_no
        if finished_lot and finished_lot.source_material_sub_batch_no
        else raw_lot.lot_no if raw_lot else None
    )
    source_material_summary = (
        finished_lot.source_material_summary
        if finished_lot and finished_lot.source_material_summary
        else raw_lot.lot_no if raw_lot else "退货返工来源"
    )
    source_material_unit_cost = to_decimal(raw_lot.unit_cost if raw_lot else finished_lot.unit_cost if finished_lot else 0, "0.0001")
    return {
        "source_material_lot_id": source_material_lot_id,
        "source_material_sub_batch_no": source_material_sub_batch_no or "退货返工来源批次未指定",
        "source_material_summary": f"退货返工；{source_material_summary}",
        "source_material_weight_kg": Decimal("0"),
        "source_material_amount": Decimal("0"),
        "source_material_unit_cost": source_material_unit_cost,
    }


def _get_work_order_gross_weight_per_piece(db: Session, work_order: WorkOrder, product: Item) -> Decimal:
    gross_weight = to_decimal(
        db.scalar(
            select(BomItem.gross_weight_per_piece_kg)
            .where(BomItem.bom_id == work_order.bom_id, BomItem.gross_weight_per_piece_kg > 0)
            .order_by(BomItem.seq_no, BomItem.id)
            .limit(1)
        ),
        "0.000001",
    )
    if gross_weight <= 0:
        gross_weight = to_decimal(product.unit_weight_kg, "0.000001")
    return gross_weight


def _calculate_finished_material_cost_per_piece(
    source_material: dict[str, object | None],
    gross_weight_per_piece: Decimal,
) -> Decimal:
    material_unit_cost = to_decimal(source_material.get("source_material_unit_cost"), "0.0001")
    if material_unit_cost <= 0 or gross_weight_per_piece <= 0:
        return Decimal("0")
    return to_decimal(material_unit_cost * gross_weight_per_piece, "0.0001")


def _append_finished_work_order_settle_remark(work_order: WorkOrder, settle_remark: str | None) -> None:
    existing = str(work_order.remark or "").strip()
    parts = [part.strip() for part in existing.replace(";", "；").split("；") if part.strip()]
    if "生产入库已结单" not in parts:
        parts.append("生产入库已结单")
    remark = str(settle_remark or "").strip()
    if remark:
        parts.append(f"生产入库结单说明：{remark}")
    unique_parts: list[str] = []
    for part in parts:
        if part not in unique_parts:
            unique_parts.append(part)
    work_order.remark = "；".join(unique_parts)[-255:]


def _rework_finished_inbound_qty(db: Session, work_order_id: int) -> Decimal:
    return to_decimal(
        db.scalar(
            select(func.coalesce(func.sum(InventoryTxn.qty_change), 0)).where(
                InventoryTxn.txn_type == "REWORK_FINISHED_IN",
                InventoryTxn.source_doc_type == "WORK_ORDER",
                InventoryTxn.source_doc_id == work_order_id,
            )
        ),
        "0.000001",
    )


def _rework_scrap_inbound_qty(db: Session, work_order_id: int) -> Decimal:
    return to_decimal(
        db.scalar(
            select(func.coalesce(func.sum(StockLot.inbound_qty), 0)).where(
                StockLot.lot_role == "REWORK_SCRAP",
                StockLot.source_doc_type == "WORK_ORDER",
                StockLot.source_doc_id == work_order_id,
            )
        ),
        "0.000001",
    )


def _settle_rework_work_order_if_ready(
    db: Session,
    work_order: WorkOrder,
    *,
    incoming_finished_qty: Decimal = Decimal("0"),
    settle_remark: str | None = None,
) -> None:
    planned_qty = to_decimal(work_order.planned_qty)
    processed_qty = _rework_finished_inbound_qty(db, work_order.id) + incoming_finished_qty + _rework_scrap_inbound_qty(db, work_order.id)
    if processed_qty < planned_qty:
        raise HTTPException(
            status_code=400,
            detail=f"返工工单尚未闭环：返工出库 {planned_qty} 件，已回库/入废料 {processed_qty} 件",
        )
    work_order.status = "SETTLED"
    work_order.actual_end_time = work_order.actual_end_time or datetime.now()
    work_order.updated_at = datetime.now()
    parts = [part.strip() for part in str(work_order.remark or "").replace(";", "；").split("；") if part.strip()]
    if "返工工单已结单" not in parts:
        parts.append("返工工单已结单")
    remark = str(settle_remark or "").strip()
    if remark:
        parts.append(f"返工结单说明：{remark}")
    work_order.remark = "；".join(dict.fromkeys(parts))[-255:]
    db.add(work_order)


def _validate_finished_settle_request(
    payload: CompletionReceiptCreate,
    *,
    receipt_qty: Decimal,
    expected_receipt_qty: Decimal,
) -> None:
    if not payload.settle_work_order:
        return
    lower_limit = expected_receipt_qty * Decimal("0.85")
    upper_limit = expected_receipt_qty * INBOUND_TOLERANCE_MULTIPLIER
    if lower_limit <= receipt_qty <= upper_limit:
        return
    if not str(payload.settle_remark or "").strip():
        raise HTTPException(status_code=400, detail="该工单结单入库成品数量超出标准成品数量上下15%，请填写结单说明")


@router.post("/completion-receipts", response_model=CompletionReceiptRead)
def create_completion_receipt(
    payload: CompletionReceiptCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> CompletionReceiptRead:
    work_order = db.get(WorkOrder, payload.work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="工单不存在")
    if str(work_order.status or "").upper() == "SETTLED":
        raise HTTPException(status_code=400, detail="该生产工单已结单，不能继续成品入库")
    warehouse = db.get(Warehouse, payload.warehouse_id)
    if not warehouse:
        raise HTTPException(status_code=404, detail="仓库不存在")
    ensure_warehouses_unlocked(db, [payload.warehouse_id], "成品入库")
    if not payload.items:
        raise HTTPException(status_code=400, detail="成品入库至少需要一条明细")

    is_rework = _is_rework_work_order(work_order)
    limit_snapshot = get_work_order_limit_snapshot(db, payload.work_order_id)
    max_receivable_qty = to_decimal(limit_snapshot.max_finished_inbound_qty)
    if max_receivable_qty <= 0:
        detail = "返工工单当前没有可返工入库数量" if is_rework else "该工单当前没有可入库的完工数量"
        raise HTTPException(status_code=400, detail=detail)

    total_receipt_qty = Decimal("0")
    for item_payload in payload.items:
        product = db.get(Item, item_payload.product_item_id)
        if not product:
            raise HTTPException(status_code=404, detail=f"产品不存在: {item_payload.product_item_id}")
        if item_payload.product_item_id != work_order.product_item_id:
            raise HTTPException(status_code=400, detail="成品入库产品必须与工单产品一致")
        receipt_qty = to_decimal(item_payload.receipt_qty)
        if receipt_qty <= 0:
            raise HTTPException(status_code=400, detail="入库数量必须大于0")
        total_receipt_qty += receipt_qty
    if is_rework:
        if total_receipt_qty > max_receivable_qty:
            raise HTTPException(status_code=400, detail=f"返工入库数量不能超过当前可返工入库数量 {max_receivable_qty}")
    else:
        validate_inbound_with_tolerance("成品入库数量", total_receipt_qty, max_receivable_qty)
    if not is_rework:
        _validate_finished_settle_request(
            payload,
            receipt_qty=total_receipt_qty,
            expected_receipt_qty=max_receivable_qty,
        )
    elif payload.settle_work_order:
        _settle_rework_work_order_if_ready(
            db,
            work_order,
            incoming_finished_qty=total_receipt_qty,
            settle_remark=payload.settle_remark,
        )

    if is_rework:
        source_material = _build_rework_source_material_summary(db, payload.work_order_id)
    else:
        source_material = _build_work_order_source_material_summary(db, payload.work_order_id)
        if not source_material["source_material_sub_batch_no"]:
            raise HTTPException(status_code=400, detail="该工单没有原材料库存批次领用记录，不能进行成品入库")
    ensure_employee_has_permission(
        db,
        payload.receiver_employee_id,
        "MENU_INVENTORY_LEDGER",
        "接收人",
        required=False,
    )
    completion = CompletionReceipt(
        receipt_no=_build_completion_receipt_no(db, work_order),
        work_order_id=payload.work_order_id,
        warehouse_id=payload.warehouse_id,
        receipt_time=payload.receipt_time or datetime.now(),
        receiver_employee_id=payload.receiver_employee_id,
        status="POSTED",
        remark=payload.remark,
    )
    db.add(completion)
    db.flush()

    remaining_receivable_qty = max_receivable_qty
    for index, item_payload in enumerate(payload.items, start=1):
        product = db.get(Item, item_payload.product_item_id)
        if not product:
            raise HTTPException(status_code=404, detail=f"产品不存在: {item_payload.product_item_id}")
        if item_payload.product_item_id != work_order.product_item_id:
            raise HTTPException(status_code=400, detail="成品入库产品必须与工单产品一致")
        location = get_location_or_default(db, payload.warehouse_id, item_payload.location_id)
        receipt_qty = to_decimal(item_payload.receipt_qty)
        if receipt_qty <= 0:
            raise HTTPException(status_code=400, detail="入库数量必须大于0")
        gross_weight = _get_work_order_gross_weight_per_piece(db, work_order, product)
        if gross_weight <= 0:
            raise HTTPException(status_code=400, detail="产品需用清单缺少单件毛重，无法计算成品入库材料成本")
        receipt_weight = receipt_qty * to_decimal(product.unit_weight_kg)
        unit_cost = _calculate_finished_material_cost_per_piece(source_material, gross_weight)
        lot_no = _build_finished_lot_no(db, work_order, product, completion.receipt_time, index)

        receipt_item = CompletionReceiptItem(
            completion_receipt_id=completion.id,
            line_no=index,
            product_item_id=item_payload.product_item_id,
            location_id=location.id if location else None,
            lot_no=lot_no,
            source_material_lot_id=source_material["source_material_lot_id"],
            source_material_sub_batch_no=source_material["source_material_sub_batch_no"],
            source_material_summary=source_material["source_material_summary"],
            receipt_qty=receipt_qty,
            receipt_weight_kg=receipt_weight,
            unit_cost=unit_cost,
            status="POSTED",
            remark=item_payload.remark,
        )
        db.add(receipt_item)
        db.flush()

        lot = StockLot(
            lot_no=lot_no,
            parent_lot_id=source_material["source_material_lot_id"],
            lot_role="FINISHED_FROM_RAW_BATCH",
            material_sub_batch_no=None,
            item_id=item_payload.product_item_id,
            warehouse_id=payload.warehouse_id,
            location_id=location.id if location else None,
            source_doc_type="REWORK_RECEIPT" if is_rework else "FG_RECEIPT",
            source_doc_id=completion.id,
            source_line_id=receipt_item.id,
            source_material_lot_id=source_material["source_material_lot_id"],
            source_material_sub_batch_no=source_material["source_material_sub_batch_no"],
            source_material_summary=source_material["source_material_summary"],
            inbound_qty=receipt_qty,
            inbound_weight_kg=receipt_weight,
            remaining_qty=receipt_qty,
            remaining_weight_kg=receipt_weight,
            locked_qty=to_decimal(0),
            locked_weight_kg=to_decimal(0),
            unit_cost=unit_cost,
            production_date=completion.receipt_time.date(),
            expire_date=None,
            quality_status="PASS",
            status="AVAILABLE",
            remark="返工入库；来源库存批次已记录" if is_rework else "完工入库；来源库存批次已记录",
        )
        db.add(lot)
        db.flush()

        upsert_stock_balance(
            db,
            item_id=item_payload.product_item_id,
            warehouse_id=payload.warehouse_id,
            location_id=location.id if location else None,
            qty_delta=receipt_qty,
            weight_delta=receipt_weight,
            available_qty_delta=receipt_qty,
            available_weight_delta=receipt_weight,
            unit_cost=unit_cost,
        )
        create_inventory_txn(
            db,
            txn_type="REWORK_FINISHED_IN" if is_rework else "FG_IN",
            item_id=item_payload.product_item_id,
            warehouse_id=payload.warehouse_id,
            location_id=location.id if location else None,
            lot_id=lot.id,
            qty_change=receipt_qty,
            weight_change=receipt_weight,
            unit_cost=unit_cost,
            source_doc_type="WORK_ORDER" if is_rework else "FG_RECEIPT",
            source_doc_id=work_order.id if is_rework else completion.id,
            source_line_id=receipt_item.id,
            biz_time=completion.receipt_time,
            operator_user_id=context.user.id,
            remark="返工入库" if is_rework else "完工入库",
            amount_basis="QTY",
        )

        work_order.finished_qty = to_decimal(work_order.finished_qty) + receipt_qty
        db.add(work_order)
        remaining_receivable_qty -= receipt_qty

    if payload.settle_work_order:
        if not is_rework:
            work_order.status = "SETTLED"
            work_order.actual_end_time = work_order.actual_end_time or datetime.now()
            work_order.updated_at = datetime.now()
            _append_finished_work_order_settle_remark(work_order, payload.settle_remark)
            db.add(work_order)
    else:
        sync_work_order_status(db, work_order.id)
    db.commit()

    row = db.execute(get_completion_receipts_query(limit=1).where(CompletionReceipt.id == completion.id)).mappings().first()
    if not row:
        raise HTTPException(status_code=500, detail="完工入库单创建后读取失败")
    return CompletionReceiptRead.model_validate(dict(row))