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

import re
from datetime import datetime
from decimal import Decimal, InvalidOperation
from io import BytesIO
from typing import Any
from urllib.parse import quote

from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
from fastapi.responses import Response
from openpyxl import Workbook, load_workbook
from sqlalchemy.exc import IntegrityError
from sqlalchemy import and_, delete, func, or_, select
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.models.master_data import Bom, BomItem, Item, Material, Product, StockBalance, Warehouse
from app.models.miniapp import (
    MiniAppAttendancePoint,
    MiniAppPersonAttendancePoint,
    MiniAppPersonRole,
    MiniAppPersonnel,
    MiniAppProduct,
)
from app.models.operations import (
    CompletionReceipt,
    CompletionReceiptItem,
    CostAllocation,
    Delivery,
    DeliveryItem,
    InventoryTxn,
    MaintenanceOrder,
    OperationReport,
    Process,
    ProcessRoute,
    ProcessRouteOperation,
    PurchaseOrder,
    PurchaseOrderItem,
    PurchaseReceipt,
    PurchaseReceiptItem,
    ReturnItem,
    ReturnOrder,
    ScrapRecord,
    StockLot,
    Supplier,
    WarehouseLocation,
    WorkCenter,
    WorkOrder,
    WorkOrderMaterial,
    WorkOrderOperation,
)
from app.models.org import Department, Employee, Role, User, UserRole
from app.models.planning import MaterialDemand
from app.models.sales import SalesOrder, SalesOrderItem
from app.schemas.database import (
    BomRead,
    BomUpsert,
    DepartmentRead,
    EmployeeCreate,
    EmployeeRead,
    EmployeeUpdate,
    ItemRead,
    MaterialCreate,
    MaterialRead,
    MaterialUpdate,
    MiniAppPersonnelRead,
    MiniAppPersonnelUpsert,
    MiniAppMaterialWastePriceSync,
    MiniAppAttendancePointRead,
    MiniAppProductImportResult,
    MiniAppProductRead,
    MiniAppProductUpsert,
    ProcessRead,
    ProcessRouteRead,
    ProcessRouteUpsert,
    ProductCreate,
    ProductRead,
    ProductUpdate,
    WarehouseRead,
)
from app.schemas.operations import WarehouseLocationRead, WorkCenterRead
from app.schemas.system_permissions import EmployeeOption
from app.schemas.domain import MiniAppField, ProductSummary
from app.services.auth import require_authenticated_user
from app.services.auth import next_employee_code
from app.services.mock_data import get_mini_app_fields, get_product_overview_summary
from app.services.operations import build_supplier_code
from app.services.system_permissions import list_employee_options

router = APIRouter(dependencies=[Depends(require_authenticated_user)])

DEFAULT_PROCESS_CODE = "PROC-DEFAULT"
DEFAULT_PROCESS_NAME = "系统默认工序"
DEFAULT_WORK_CENTER_CODE = "WC-DEFAULT"
DEFAULT_WORK_CENTER_NAME = "系统默认工作中心"
DEFAULT_ORG_ROOT_CODE = "ORG_ROOT"


def _default_operation_department(db: Session) -> Department:
    department = db.scalar(select(Department).where(Department.dept_code == DEFAULT_ORG_ROOT_CODE).limit(1))
    if department:
        return department

    department = db.scalar(select(Department).order_by(Department.id.asc()).limit(1))
    if department:
        return department

    now = datetime.now()
    department = Department(
        dept_code=DEFAULT_ORG_ROOT_CODE,
        dept_name="总公司",
        parent_id=None,
        org_node_type="COMPANY",
        dept_type="ADMIN",
        manager_name=None,
        manager_employee_id=None,
        status="ACTIVE",
        sort_no=0,
        remark="系统自动创建的默认组织根节点",
        created_at=now,
        updated_at=now,
    )
    db.add(department)
    db.flush()
    return department


def _ensure_default_operation_foundation(db: Session) -> None:
    changed = False
    now = datetime.now()
    process_exists = db.scalar(select(Process.id).limit(1))
    if not process_exists:
        db.add(
            Process(
                process_code=DEFAULT_PROCESS_CODE,
                process_name=DEFAULT_PROCESS_NAME,
                process_type="INTERNAL",
                is_report_required=1,
                is_quality_gate=0,
                status="ACTIVE",
                remark="系统自动创建，用于产品需规快速建档时生成默认工艺路线",
                created_at=now,
                updated_at=now,
            )
        )
        changed = True

    work_center_exists = db.scalar(select(WorkCenter.id).limit(1))
    if not work_center_exists:
        department = _default_operation_department(db)
        db.add(
            WorkCenter(
                center_code=DEFAULT_WORK_CENTER_CODE,
                center_name=DEFAULT_WORK_CENTER_NAME,
                dept_id=department.id,
                center_type="INTERNAL",
                shift_pattern="默认班次",
                capacity_hours_per_day=Decimal("8.00"),
                status="ACTIVE",
                remark="系统自动创建，用于产品需规快速建档时生成默认工艺路线",
                created_at=now,
                updated_at=now,
            )
        )
        changed = True

    if changed:
        try:
            db.commit()
        except IntegrityError:
            db.rollback()

MATERIAL_EXCEL_HEADERS = ["原材料编码", "原材料名称", "材质", "规格", "废料单价", "性能要求", "状态"]
MATERIAL_EXCEL_REQUIRED_HEADERS = ["原材料名称", "材质", "规格"]
MATERIAL_STATUS_LABELS = {
    "ACTIVE": "启用",
    "INACTIVE": "停用",
}

MINIAPP_ROLE_LABELS = {
    "worker": "员工",
    "admin": "管理员",
    "manager": "经理",
}

_miniapp_master_sync_signature: tuple[object, ...] | None = None
MINIAPP_VALID_ROLES = set(MINIAPP_ROLE_LABELS)

PRODUCT_IMPORT_HEADERS = {
    "考勤点": "attendance_point_name",
    "项目号": "project_no",
    "型材号": "profile_no",
    "产品名称": "product_name",
    "物料编码": "material_code",
    "物料名称": "material_name",
    "供应商": "supplier",
    "产品净重(kg)": "product_net_weight_kg",
    "产品毛重(kg)": "product_gross_weight_kg",
    "允许报废率": "scrap_loss_rate",
    "废料单价(元/kg)": "waste_price_yuan_per_kg",
    "工序": "process_name",
    "冲压方式": "stamping_method",
    "操作人数": "operator_count",
    "标准节拍": "standard_beat",
}
PRODUCT_EXCEL_HEADERS = [
    "考勤点",
    "项目号",
    "型材号",
    "产品名称",
    "物料编码",
    "物料名称",
    "供应商",
    "工序",
    "冲压方式",
    "操作人数",
    "不包含物料转运节拍",
    "标准节拍",
    "产品净重(kg)",
    "产品毛重(kg)",
    "允许报废率",
    "废料单价(元/kg)",
]
PRODUCT_IMPORT_AUX_FIELDS = (
    ("product_net_weight_kg", "产品净重(kg)"),
    ("product_gross_weight_kg", "产品毛重(kg)"),
    ("scrap_loss_rate", "允许报废率"),
    ("waste_price_yuan_per_kg", "废料单价(元/kg)"),
)
MINIAPP_CLEANING_STAMPING_METHOD = "清洗"
MINIAPP_LEGACY_CLEANING_PROCESS_NAME = "清洗"
MINIAPP_MISC_PROCESS_NAME = "杂活"
MINIAPP_MISC_STAMPING_METHOD = "处理杂活"
MINIAPP_SYNC_REMARK = "小程序产品清单同步生成"
JIAHENG_WAREHOUSE_DEFINITIONS = (
    {
        "warehouse_code": "WH-RM-01",
        "warehouse_name": "原材料库",
        "warehouse_type": "RAW",
        "location_code": "RM-A-01",
        "location_name": "原料主库位",
        "zone_name": "A区",
        "remark": "自家料、客供料等生产主材库存",
    },
    {
        "warehouse_code": "WH-SEMI-01",
        "warehouse_name": "半成品库",
        "warehouse_type": "SEMI",
        "location_code": "SEMI-B-01",
        "location_name": "半成品暂存位",
        "zone_name": "B区",
        "remark": "加工到一半、尚未完成全部工艺的半成品暂存",
    },
    {
        "warehouse_code": "WH-FG-01",
        "warehouse_name": "成品库",
        "warehouse_type": "FINISHED",
        "location_code": "FG-C-01",
        "location_name": "成品主库位",
        "zone_name": "C区",
        "remark": "完工入库后可发货的成品库存",
    },
    {
        "warehouse_code": "WH-AUX-01",
        "warehouse_name": "辅料库",
        "warehouse_type": "AUX",
        "location_code": "AUX-D-01",
        "location_name": "辅料主库位",
        "zone_name": "D区",
        "remark": "手套、薄膜、纸箱、螺丝钉等生产消耗辅料库存",
    },
    {
        "warehouse_code": "WH-SCRAP-01",
        "warehouse_name": "废料库",
        "warehouse_type": "SCRAP",
        "location_code": "SCR-E-01",
        "location_name": "废料暂存位",
        "zone_name": "E区",
        "remark": "生产废料、边角料、委外废料、报废品等废料库存",
    },
    {
        "warehouse_code": "WH-RETURN-01",
        "warehouse_name": "退货库",
        "warehouse_type": "RETURN",
        "location_code": "RTN-F-01",
        "location_name": "退货暂存位",
        "zone_name": "F区",
        "remark": "客户退回且可返工产品的待处理库存",
    },
)


@router.get("/product-overview", response_model=ProductSummary)
def product_overview() -> ProductSummary:
    return get_product_overview_summary()


@router.get("/mini-app-fields", response_model=list[MiniAppField])
def mini_app_fields() -> list[MiniAppField]:
    return get_mini_app_fields()


@router.get("/departments", response_model=list[DepartmentRead])
def list_departments(
    limit: int = Query(default=50, ge=1, le=200),
    db: Session = Depends(get_db),
) -> list[DepartmentRead]:
    rows = db.scalars(select(Department).order_by(Department.sort_no, Department.id).limit(limit)).all()
    return [DepartmentRead.model_validate(row) for row in rows]


@router.get("/employees", response_model=list[EmployeeRead])
def list_employees(
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[EmployeeRead]:
    active_people = _active_miniapp_people(db)
    _sync_miniapp_personnel_to_employees(db, people=active_people, deactivate_missing=True)
    db.commit()
    miniapp_phones = [_normalize_text(person.phone) for person in active_people if _normalize_text(person.phone)]
    if miniapp_phones:
        employees = db.scalars(
            select(Employee)
            .where(Employee.mobile.in_(miniapp_phones), Employee.status == "ACTIVE")
            .order_by(Employee.employee_name.asc(), Employee.id.desc())
            .limit(limit)
        ).all()
    else:
        employees = []
    return _build_employee_reads(db, employees)


@router.get("/employee-options", response_model=list[EmployeeOption])
def list_business_employee_options(
    permission_code: str | None = Query(default=None),
    limit: int = Query(default=500, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[EmployeeOption]:
    return list_employee_options(db, permission_code=permission_code, limit=limit)


@router.get("/miniapp-personnel", response_model=list[MiniAppPersonnelRead])
def list_miniapp_personnel(
    limit: int = Query(default=200, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[MiniAppPersonnelRead]:
    people = _active_miniapp_people(db, limit=limit)
    _sync_miniapp_personnel_to_employees(db, people=people, deactivate_missing=True)
    db.commit()
    return _build_miniapp_personnel_reads(db, people)


@router.get("/miniapp-attendance-points", response_model=list[MiniAppAttendancePointRead])
def list_miniapp_attendance_points(
    include_inactive: bool = Query(default=False),
    db: Session = Depends(get_db),
) -> list[MiniAppAttendancePointRead]:
    stmt = select(MiniAppAttendancePoint).order_by(MiniAppAttendancePoint.is_active.desc(), MiniAppAttendancePoint.name.asc())
    if not include_inactive:
        stmt = stmt.where(MiniAppAttendancePoint.is_active.is_(True))
    rows = db.scalars(stmt).all()
    return [_miniapp_attendance_point_read(row) for row in rows]


@router.post("/miniapp-personnel", response_model=MiniAppPersonnelRead)
def create_miniapp_person(payload: MiniAppPersonnelUpsert, db: Session = Depends(get_db)) -> MiniAppPersonnelRead:
    phone = _normalize_phone(payload.phone)
    person = db.get(MiniAppPersonnel, phone)
    if person and _miniapp_person_has_roles(db, phone):
        raise HTTPException(status_code=400, detail="该手机号已在小程序人员清单中存在")
    if not person:
        person = MiniAppPersonnel(phone=phone)
    person.name = _normalize_text(payload.name)
    person.is_temporary = bool(payload.is_temporary)
    person.temporary_expires_at = payload.temporary_expires_at
    db.add(person)
    _replace_miniapp_person_roles(db, phone, payload.roles)
    _replace_miniapp_person_attendance_points(db, phone, payload.attendance_point_names)
    _sync_miniapp_personnel_to_employees(db, people=[person])
    db.commit()
    db.refresh(person)
    return _build_miniapp_personnel_reads(db, [person])[0]


@router.put("/miniapp-personnel/{phone}", response_model=MiniAppPersonnelRead)
def update_miniapp_person(phone: str, payload: MiniAppPersonnelUpsert, db: Session = Depends(get_db)) -> MiniAppPersonnelRead:
    current_phone = _normalize_phone(phone)
    next_phone = _normalize_phone(payload.phone)
    person = db.get(MiniAppPersonnel, current_phone)
    if not person:
        raise HTTPException(status_code=404, detail="小程序人员不存在")
    if next_phone != current_phone and db.get(MiniAppPersonnel, next_phone):
        raise HTTPException(status_code=400, detail="新的手机号已被其他人员使用")

    if next_phone != current_phone:
        # 小程序多张业务表以 phone 做外键，手机号不做原地改键，避免历史报工断链。
        raise HTTPException(status_code=400, detail="人员手机号是小程序身份主键，不能直接修改；请新增人员后停用旧人员")

    person.name = _normalize_text(payload.name)
    person.is_temporary = bool(payload.is_temporary)
    person.temporary_expires_at = payload.temporary_expires_at
    _replace_miniapp_person_roles(db, current_phone, payload.roles)
    _replace_miniapp_person_attendance_points(db, current_phone, payload.attendance_point_names)
    db.add(person)
    _sync_miniapp_personnel_to_employees(db, people=[person])
    db.commit()
    db.refresh(person)
    return _build_miniapp_personnel_reads(db, [person])[0]


@router.delete("/miniapp-personnel/{phone}")
def delete_miniapp_person(phone: str, db: Session = Depends(get_db)) -> dict[str, str]:
    normalized_phone = _normalize_phone(phone)
    person = db.get(MiniAppPersonnel, normalized_phone)
    if not person:
        raise HTTPException(status_code=404, detail="小程序人员不存在")
    try:
        db.execute(delete(MiniAppPersonAttendancePoint).where(MiniAppPersonAttendancePoint.phone == normalized_phone))
        db.execute(delete(MiniAppPersonRole).where(MiniAppPersonRole.phone == normalized_phone))
        db.delete(person)
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=400, detail="该人员已有小程序报工/通知等业务记录，不能删除") from exc
    return {"message": f"小程序人员 {person.name} 已删除"}


@router.get("/miniapp-products", response_model=list[MiniAppProductRead])
def list_miniapp_products(
    limit: int = Query(default=500, ge=1, le=1000),
    db: Session = Depends(get_db),
) -> list[MiniAppProductRead]:
    rows = db.scalars(
        select(MiniAppProduct)
        .order_by(MiniAppProduct.updated_at.desc(), MiniAppProduct.attendance_point_name.asc())
        .limit(limit)
    ).all()
    return [_miniapp_product_read(row) for row in rows]


@router.get("/miniapp-products/export-template")
def export_miniapp_products_template() -> Response:
    workbook = _build_miniapp_product_workbook([])
    return _excel_response(workbook, "基础资料导入模版.xlsx")


@router.get("/miniapp-products/export")
def export_miniapp_products(db: Session = Depends(get_db)) -> Response:
    rows = db.scalars(
        select(MiniAppProduct).order_by(
            MiniAppProduct.attendance_point_name.asc(),
            MiniAppProduct.project_no.asc(),
            MiniAppProduct.product_name.asc(),
            MiniAppProduct.process_name.asc(),
        )
    ).all()
    workbook = _build_miniapp_product_workbook(rows)
    return _excel_response(workbook, f"基础资料_{datetime.now().strftime('%Y%m%d%H%M%S')}.xlsx")


@router.post("/miniapp-products/import", response_model=MiniAppProductImportResult)
async def import_miniapp_products(
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
) -> MiniAppProductImportResult:
    content = await file.read()
    try:
        result = _import_miniapp_product_excel(db, content)
    except ValueError as exc:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return result


@router.post("/miniapp-materials/sync")
def sync_miniapp_materials(
    force: bool = Query(default=False),
    db: Session = Depends(get_db),
) -> dict[str, object]:
    global _miniapp_master_sync_signature

    current_signature = _get_miniapp_master_sync_signature(db)
    if not force and _miniapp_master_sync_signature == current_signature:
        return {
            "synced": False,
            "skipped_by_cache": True,
            "message": "小程序产品清单未变化，已跳过产品需规同步",
            "product_specs": {
                "matched": 0,
                "created": 0,
                "product_updated": 0,
                "bom_updated": 0,
                "route_updated": 0,
                "stale_deleted": 0,
                "stale_blocked": 0,
                "stale_deactivated": 0,
            },
        }

    product_result = _sync_miniapp_product_specs_to_master_data(db)
    db.commit()
    _miniapp_master_sync_signature = _get_miniapp_master_sync_signature(db)
    return {
        "created": 0,
        "linked": 0,
        "supplier_updated": 0,
        "skipped": 0,
        "synced": True,
        "skipped_by_cache": False,
        "message": "原材料名录已改为独立维护，本次仅同步产品需规",
        "product_specs": product_result,
    }


@router.post("/miniapp-material-waste-price/sync")
def sync_miniapp_material_waste_price(
    payload: MiniAppMaterialWastePriceSync,
    db: Session = Depends(get_db),
) -> dict[str, object]:
    return {"updated": 0, "message": "原材料名录已改为独立维护，不再从小程序同步废料单价"}


@router.post("/miniapp-products", response_model=MiniAppProductRead)
def save_miniapp_product(
    payload: MiniAppProductUpsert,
    sync: bool = Query(default=True),
    db: Session = Depends(get_db),
) -> MiniAppProductRead:
    row = _upsert_miniapp_product(db, payload)
    if sync:
        _sync_miniapp_product_specs_to_master_data(db)
    db.commit()
    if sync:
        _mark_miniapp_master_sync_fresh(db)
    db.refresh(row)
    return _miniapp_product_read(row)


@router.delete("/miniapp-products")
def delete_miniapp_product(
    attendance_point_name: str = Query(min_length=1),
    project_no: str = Query(min_length=1),
    product_name: str = Query(min_length=1),
    device_no: str = Query(default=""),
    process_name: str = Query(default=""),
    db: Session = Depends(get_db),
) -> dict[str, str]:
    key = {
        "attendance_point_name": _normalize_text(attendance_point_name),
        "project_no": _normalize_text(project_no),
        "product_name": _normalize_text(product_name),
        "device_no": _normalize_text(device_no),
        "process_name": _normalize_text(process_name),
    }
    row = db.get(MiniAppProduct, key)
    if not row:
        raise HTTPException(status_code=404, detail="小程序产品清单记录不存在")

    label = f"{row.project_no} / {row.product_name} / {row.process_name or '-'}"
    db.delete(row)
    db.flush()
    _sync_miniapp_product_specs_to_master_data(db)
    db.commit()
    _mark_miniapp_master_sync_fresh(db)
    return {"message": f"小程序产品清单 {label} 已删除，ERP 对应需规已同步停用"}


@router.post("/employees", response_model=EmployeeRead)
def create_employee(payload: EmployeeCreate, db: Session = Depends(get_db)) -> EmployeeRead:
    department = db.get(Department, payload.dept_id)
    if not department:
        raise HTTPException(status_code=404, detail="所属部门不存在")

    employee = Employee(
        employee_code=next_employee_code(db),
        employee_name=payload.employee_name,
        dept_id=payload.dept_id,
        mobile=payload.mobile,
        gender=payload.gender,
        hire_date=payload.hire_date,
        job_title=payload.job_title,
        shift_code=payload.shift_code,
        manager_employee_id=payload.manager_employee_id,
        is_operator=int(payload.is_operator),
        is_workshop_staff=int(payload.is_workshop_staff),
        status=payload.status,
        remark=payload.remark,
    )
    db.add(employee)
    db.commit()
    db.refresh(employee)
    return _build_employee_reads(db, [employee])[0]


@router.put("/employees/{employee_id}", response_model=EmployeeRead)
def update_employee(employee_id: int, payload: EmployeeUpdate, db: Session = Depends(get_db)) -> EmployeeRead:
    employee = db.get(Employee, employee_id)
    if not employee:
        raise HTTPException(status_code=404, detail="员工不存在")

    department = db.get(Department, payload.dept_id)
    if not department:
        raise HTTPException(status_code=404, detail="所属部门不存在")

    employee.employee_name = payload.employee_name
    employee.dept_id = payload.dept_id
    employee.mobile = payload.mobile
    employee.gender = payload.gender
    employee.hire_date = payload.hire_date
    employee.job_title = payload.job_title
    employee.shift_code = payload.shift_code
    employee.manager_employee_id = payload.manager_employee_id
    employee.is_operator = int(payload.is_operator)
    employee.is_workshop_staff = int(payload.is_workshop_staff)
    employee.status = payload.status
    employee.remark = payload.remark

    db.add(employee)
    db.commit()
    db.refresh(employee)
    return _build_employee_reads(db, [employee])[0]


@router.delete("/employees/{employee_id}")
def delete_employee(employee_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    employee = db.get(Employee, employee_id)
    if not employee:
        raise HTTPException(status_code=404, detail="员工不存在")

    _ensure_employee_can_delete(db, employee_id)
    employee_name = employee.employee_name
    db.delete(employee)
    db.commit()
    return {"message": f"员工 {employee_name} 已删除"}


@router.get("/warehouses", response_model=list[WarehouseRead])
def list_warehouses(
    limit: int = Query(default=50, ge=1, le=200),
    db: Session = Depends(get_db),
) -> list[WarehouseRead]:
    _ensure_jiaheng_warehouses(db)
    rows = db.scalars(select(Warehouse).order_by(Warehouse.id.desc()).limit(limit)).all()
    return [WarehouseRead.model_validate(row) for row in rows]


@router.get("/locations", response_model=list[WarehouseLocationRead])
def list_locations(
    warehouse_id: int | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WarehouseLocationRead]:
    _ensure_jiaheng_warehouses(db)
    stmt = (
        select(
            WarehouseLocation.id.label("id"),
            WarehouseLocation.warehouse_id.label("warehouse_id"),
            Warehouse.warehouse_code.label("warehouse_code"),
            Warehouse.warehouse_name.label("warehouse_name"),
            WarehouseLocation.location_code.label("location_code"),
            WarehouseLocation.location_name.label("location_name"),
            WarehouseLocation.zone_name.label("zone_name"),
            WarehouseLocation.is_default.label("is_default"),
            WarehouseLocation.is_locked.label("is_locked"),
            WarehouseLocation.status.label("status"),
        )
        .join(Warehouse, Warehouse.id == WarehouseLocation.warehouse_id)
        .order_by(WarehouseLocation.warehouse_id, WarehouseLocation.is_default.desc(), WarehouseLocation.id)
        .limit(limit)
    )
    if warehouse_id:
        stmt = stmt.where(WarehouseLocation.warehouse_id == warehouse_id)
    rows = db.execute(stmt).mappings().all()
    return [WarehouseLocationRead.model_validate(dict(row)) for row in rows]


@router.get("/work-centers", response_model=list[WorkCenterRead])
def list_work_centers(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[WorkCenterRead]:
    _ensure_default_operation_foundation(db)
    stmt = (
        select(
            WorkCenter.id.label("id"),
            WorkCenter.center_code.label("center_code"),
            WorkCenter.center_name.label("center_name"),
            WorkCenter.dept_id.label("dept_id"),
            Department.dept_name.label("dept_name"),
            WorkCenter.center_type.label("center_type"),
            WorkCenter.shift_pattern.label("shift_pattern"),
            WorkCenter.capacity_hours_per_day.label("capacity_hours_per_day"),
            WorkCenter.status.label("status"),
        )
        .join(Department, Department.id == WorkCenter.dept_id)
        .order_by(WorkCenter.id.desc())
        .limit(limit)
    )
    rows = db.execute(stmt).mappings().all()
    return [WorkCenterRead.model_validate(dict(row)) for row in rows]


@router.get("/items", response_model=list[ItemRead])
def list_items(
    item_type: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=200),
    db: Session = Depends(get_db),
) -> list[ItemRead]:
    stmt = select(Item).order_by(Item.id.desc()).limit(limit)
    if item_type:
        stmt = stmt.where(Item.item_type == item_type)
    rows = db.scalars(stmt).all()
    return [ItemRead.model_validate(row) for row in rows]


@router.get("/materials", response_model=list[MaterialRead])
def list_materials(
    limit: int = Query(default=50, ge=1, le=200),
    db: Session = Depends(get_db),
) -> list[MaterialRead]:
    stmt = _material_export_stmt().order_by(None).order_by(Material.id.desc()).limit(limit)
    rows = db.execute(stmt).mappings().all()
    return [MaterialRead.model_validate(_normalize_material_read_payload(dict(row))) for row in rows]


@router.get("/materials/export")
def export_materials(db: Session = Depends(get_db)) -> Response:
    rows = db.execute(_material_export_stmt()).mappings().all()
    workbook = _build_material_workbook(rows)
    return _excel_workbook_response(workbook, f"原材料名录_{datetime.now().strftime('%Y%m%d%H%M%S')}.xlsx", "materials.xlsx")


@router.post("/materials/import")
async def import_materials(
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
) -> dict[str, object]:
    content = await file.read()
    try:
        result = _import_material_excel(db, content)
    except ValueError as exc:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return result


@router.post("/materials", response_model=MaterialRead)
def create_material(payload: MaterialCreate, db: Session = Depends(get_db)) -> MaterialRead:
    payload = payload.model_copy(update={"item_code": _build_material_code(db, payload)})
    _validate_material_payload(payload)

    existing = db.scalar(select(Item).where(Item.item_code == payload.item_code))
    if existing:
        raise HTTPException(status_code=400, detail="原材料编码已存在")

    if payload.default_supplier_id and not db.get(Supplier, payload.default_supplier_id):
        raise HTTPException(status_code=404, detail="默认供应商不存在")

    item = Item(
        item_code=payload.item_code,
        item_name=payload.item_name,
        item_type="RAW_MATERIAL",
        specification=payload.specification,
        material_grade=payload.material_grade,
        unit_weight_kg=Decimal(str(payload.unit_weight_kg)),
        safety_stock_weight_kg=Decimal(str(payload.safety_stock_weight_kg)),
        scrap_sale_price=Decimal(str(payload.scrap_sale_price)),
        status=payload.status,
    )
    db.add(item)
    db.flush()

    material = Material(
        item_id=item.id,
        default_supplier_id=payload.default_supplier_id,
        purchase_calc_mode="BY_WEIGHT",
        min_purchase_qty=Decimal(str(payload.min_purchase_qty)),
        purchase_multiple_qty=Decimal("0"),
        purchase_multiple_weight_kg=Decimal(str(payload.purchase_multiple_weight_kg)),
        lead_time_days=payload.lead_time_days,
        quality_rule=_material_remark_value(payload),
    )
    db.add(material)
    db.commit()

    return _get_material_read(db, item.id)


@router.put("/materials/{item_id}", response_model=MaterialRead)
def update_material(item_id: int, payload: MaterialUpdate, db: Session = Depends(get_db)) -> MaterialRead:
    item = db.get(Item, item_id)
    if not item or item.item_type != "RAW_MATERIAL":
        raise HTTPException(status_code=404, detail="原材料不存在")

    payload = payload.model_copy(update={"item_code": item.item_code})
    _validate_material_payload(payload)

    material = db.scalar(select(Material).where(Material.item_id == item_id))
    if not material:
        raise HTTPException(status_code=404, detail="原材料扩展信息不存在")

    if payload.default_supplier_id and not db.get(Supplier, payload.default_supplier_id):
        raise HTTPException(status_code=404, detail="默认供应商不存在")

    item.item_name = payload.item_name
    item.specification = payload.specification
    item.material_grade = payload.material_grade
    item.unit_weight_kg = Decimal(str(payload.unit_weight_kg))
    item.safety_stock_weight_kg = Decimal(str(payload.safety_stock_weight_kg))
    item.scrap_sale_price = Decimal(str(payload.scrap_sale_price))
    item.status = payload.status

    material.default_supplier_id = payload.default_supplier_id
    material.purchase_calc_mode = "BY_WEIGHT"
    material.min_purchase_qty = Decimal(str(payload.min_purchase_qty))
    material.purchase_multiple_qty = Decimal("0")
    material.purchase_multiple_weight_kg = Decimal(str(payload.purchase_multiple_weight_kg))
    material.lead_time_days = payload.lead_time_days
    material.quality_rule = _material_remark_value(payload)

    db.add(item)
    db.add(material)
    db.commit()

    return _get_material_read(db, item_id)


@router.delete("/materials/{item_id}")
def delete_material(item_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    item = db.get(Item, item_id)
    if not item or item.item_type != "RAW_MATERIAL":
        raise HTTPException(status_code=404, detail="原材料不存在")

    material = db.scalar(select(Material).where(Material.item_id == item_id))
    if not material:
        raise HTTPException(status_code=404, detail="原材料扩展信息不存在")

    _ensure_material_can_delete(db, item_id)
    item_name = item.item_name
    db.delete(material)
    db.flush()
    db.delete(item)
    db.commit()
    return {"message": f"原材料 {item_name} 已删除"}


@router.get("/processes", response_model=list[ProcessRead])
def list_processes(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[ProcessRead]:
    _ensure_default_operation_foundation(db)
    rows = db.execute(
        select(
            Process.id.label("process_id"),
            Process.process_code.label("process_code"),
            Process.process_name.label("process_name"),
            Process.process_type.label("process_type"),
            Process.is_report_required.label("is_report_required"),
            Process.is_quality_gate.label("is_quality_gate"),
            Process.status.label("status"),
        )
        .order_by(Process.id)
        .limit(limit)
    ).mappings().all()
    return [ProcessRead.model_validate(dict(row)) for row in rows]


@router.get("/boms", response_model=list[BomRead])
def list_boms(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[BomRead]:
    return _list_bom_reads(db, limit=limit)


@router.post("/boms", response_model=BomRead)
def create_bom(payload: BomUpsert, db: Session = Depends(get_db)) -> BomRead:
    payload = payload.model_copy(update={"bom_code": _build_bom_code(db, payload.product_item_id)})
    _validate_bom_payload(db, payload)
    if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_bom(db, payload.product_item_id):
        raise HTTPException(status_code=400, detail="当前产品还没有默认BOM，首个生效版本必须设为默认")

    if _bom_code_version_exists(db, payload.bom_code, payload.version_no):
        raise HTTPException(status_code=400, detail="相同BOM编码和版本号已存在")

    if payload.is_default:
        _clear_default_bom(db, payload.product_item_id)

    bom = _create_versioned_master_record(db, lambda: _create_bom_record(db, payload))

    _commit_versioned_master_changes(db)
    return _get_bom_read(db, bom.id)


@router.put("/boms/{bom_id}", response_model=BomRead)
def update_bom(bom_id: int, payload: BomUpsert, db: Session = Depends(get_db)) -> BomRead:
    bom = db.get(Bom, bom_id)
    if not bom:
        raise HTTPException(status_code=404, detail="BOM不存在")
    _ensure_active_master_record(bom.status, "已停用的BOM只读，不允许修改")

    payload = payload.model_copy(update={"bom_code": bom.bom_code})
    _validate_bom_payload(db, payload)
    version_changed = _normalize_text(payload.version_no) != _normalize_text(bom.version_no)

    if version_changed:
        if _bom_code_version_exists(db, payload.bom_code, payload.version_no):
            raise HTTPException(status_code=400, detail="相同BOM编码和版本号已存在")
        if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_bom(db, payload.product_item_id):
            raise HTTPException(status_code=400, detail="当前产品至少需要保留一套默认BOM，请先建立新的默认版本")
        if payload.is_default:
            _clear_default_bom(db, payload.product_item_id)
        new_bom = _create_versioned_master_record(db, lambda: _create_bom_record(db, payload))
        _commit_versioned_master_changes(db)
        return _get_bom_read(db, new_bom.id)

    if _bom_code_version_exists(db, payload.bom_code, payload.version_no, exclude_bom_id=bom_id):
        raise HTTPException(status_code=400, detail="相同BOM编码和版本号已存在")

    if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_bom(
        db, payload.product_item_id, exclude_bom_id=bom_id
    ):
        raise HTTPException(status_code=400, detail="当前产品至少需要保留一套默认BOM，请先建立新的默认版本")

    if _bom_has_usage(db, bom_id):
        raise HTTPException(status_code=400, detail="该BOM已被MRP或工单引用，请修改版本号后保存为新版本")

    if payload.is_default:
        _clear_default_bom(db, payload.product_item_id, exclude_bom_id=bom_id)

    bom.bom_code = _normalize_text(payload.bom_code)
    bom.product_item_id = payload.product_item_id
    bom.version_no = _normalize_text(payload.version_no)
    bom.is_default = int(payload.is_default)
    bom.status = payload.status
    bom.effective_date = payload.effective_date
    bom.expire_date = payload.expire_date
    bom.remark = payload.remark
    db.add(bom)

    db.execute(delete(BomItem).where(BomItem.bom_id == bom_id))
    db.flush()

    for item in payload.items:
        db.add(
            BomItem(
                bom_id=bom.id,
                seq_no=item.seq_no,
                material_item_id=item.material_item_id,
                usage_type=_normalize_text(item.usage_type).upper(),
                qty_per_piece=Decimal(str(item.qty_per_piece)),
                net_weight_per_piece_kg=Decimal(str(item.net_weight_per_piece_kg)),
                gross_weight_per_piece_kg=Decimal(str(item.gross_weight_per_piece_kg)),
                loss_rate=Decimal(str(item.loss_rate)),
                issue_over_tolerance_rate=Decimal(str(item.issue_over_tolerance_rate)),
                remark=item.remark,
            )
        )

    _commit_versioned_master_changes(db)
    return _get_bom_read(db, bom_id)


@router.delete("/boms/{bom_id}")
def delete_bom(bom_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    bom = db.get(Bom, bom_id)
    if not bom:
        raise HTTPException(status_code=404, detail="BOM不存在")
    _ensure_active_master_record(bom.status, "已停用的BOM只读，不允许删除")

    _ensure_bom_can_delete(db, bom)
    bom_code = bom.bom_code
    db.execute(delete(BomItem).where(BomItem.bom_id == bom_id))
    db.flush()
    db.delete(bom)
    db.commit()
    return {"message": f"BOM {bom_code} 已删除"}


@router.get("/process-routes", response_model=list[ProcessRouteRead])
def list_process_routes(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[ProcessRouteRead]:
    return _list_process_route_reads(db, limit=limit)


@router.post("/process-routes", response_model=ProcessRouteRead)
def create_process_route(payload: ProcessRouteUpsert, db: Session = Depends(get_db)) -> ProcessRouteRead:
    payload = payload.model_copy(update={"route_code": _build_route_code(db, payload.product_item_id)})
    _validate_process_route_payload(db, payload)
    if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_route(db, payload.product_item_id):
        raise HTTPException(status_code=400, detail="当前产品还没有默认工艺路线，首个生效版本必须设为默认")

    if _route_code_version_exists(db, payload.route_code, payload.version_no):
        raise HTTPException(status_code=400, detail="相同工艺路线编码和版本号已存在")

    if payload.is_default:
        _clear_default_route(db, payload.product_item_id)

    route = _create_versioned_master_record(db, lambda: _create_process_route_record(db, payload))

    _commit_versioned_master_changes(db)
    return _get_process_route_read(db, route.id)


@router.put("/process-routes/{route_id}", response_model=ProcessRouteRead)
def update_process_route(route_id: int, payload: ProcessRouteUpsert, db: Session = Depends(get_db)) -> ProcessRouteRead:
    route = db.get(ProcessRoute, route_id)
    if not route:
        raise HTTPException(status_code=404, detail="工艺路线不存在")
    _ensure_active_master_record(route.status, "已停用的工艺路线只读，不允许修改")

    payload = payload.model_copy(update={"route_code": route.route_code})
    _validate_process_route_payload(db, payload)
    version_changed = _normalize_text(payload.version_no) != _normalize_text(route.version_no)

    if version_changed:
        if _route_code_version_exists(db, payload.route_code, payload.version_no):
            raise HTTPException(status_code=400, detail="相同工艺路线编码和版本号已存在")
        if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_route(db, payload.product_item_id):
            raise HTTPException(status_code=400, detail="当前产品至少需要保留一套默认工艺路线，请先建立新的默认版本")
        if payload.is_default:
            _clear_default_route(db, payload.product_item_id)
        new_route = _create_versioned_master_record(db, lambda: _create_process_route_record(db, payload))
        _commit_versioned_master_changes(db)
        return _get_process_route_read(db, new_route.id)

    if _route_code_version_exists(db, payload.route_code, payload.version_no, exclude_route_id=route_id):
        raise HTTPException(status_code=400, detail="相同工艺路线编码和版本号已存在")

    if payload.status == "ACTIVE" and not payload.is_default and not _has_active_default_route(
        db, payload.product_item_id, exclude_route_id=route_id
    ):
        raise HTTPException(status_code=400, detail="当前产品至少需要保留一套默认工艺路线，请先建立新的默认版本")

    if _route_has_usage(db, route_id):
        raise HTTPException(status_code=400, detail="该工艺路线已被工单引用，请修改版本号后保存为新版本")
    _ensure_route_payload_does_not_touch_inactive_operations(db, route_id, payload)

    if payload.is_default:
        _clear_default_route(db, payload.product_item_id, exclude_route_id=route_id)

    route.route_code = _normalize_text(payload.route_code)
    route.route_name = _normalize_text(payload.route_name)
    route.product_item_id = payload.product_item_id
    route.version_no = _normalize_text(payload.version_no)
    route.is_default = int(payload.is_default)
    route.status = payload.status
    route.effective_date = payload.effective_date
    route.expire_date = payload.expire_date
    route.remark = payload.remark
    db.add(route)

    db.execute(
        delete(ProcessRouteOperation).where(
            ProcessRouteOperation.route_id == route_id,
            ProcessRouteOperation.status != "INACTIVE",
        )
    )
    db.flush()

    for operation in payload.operations:
        db.add(
            ProcessRouteOperation(
                route_id=route.id,
                seq_no=operation.seq_no,
                process_id=operation.process_id,
                work_center_id=operation.work_center_id,
                operation_name=operation.operation_name,
                std_setup_minutes=Decimal(str(operation.std_setup_minutes)),
                std_cycle_seconds=Decimal(str(operation.std_cycle_seconds or 0)),
                std_labor_cost=Decimal("0"),
                std_machine_cost=Decimal("0"),
                std_outsource_cost=Decimal("0"),
                allowed_scrap_rate=Decimal(str(operation.allowed_scrap_rate)),
                report_required=int(operation.report_required),
                is_key_operation=int(operation.is_key_operation),
                status=operation.status,
                remark=operation.remark,
            )
        )

    _commit_versioned_master_changes(db)
    return _get_process_route_read(db, route_id)


@router.delete("/process-routes/{route_id}")
def delete_process_route(route_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    route = db.get(ProcessRoute, route_id)
    if not route:
        raise HTTPException(status_code=404, detail="工艺路线不存在")
    _ensure_active_master_record(route.status, "已停用的工艺路线只读，不允许删除")

    _ensure_route_can_delete(db, route)
    route_code = route.route_code
    db.execute(delete(ProcessRouteOperation).where(ProcessRouteOperation.route_id == route_id))
    db.flush()
    db.delete(route)
    db.commit()
    return {"message": f"工艺路线 {route_code} 已删除"}


@router.get("/products", response_model=list[ProductRead])
def list_products(
    limit: int = Query(default=50, ge=1, le=200),
    db: Session = Depends(get_db),
) -> list[ProductRead]:
    stmt = (
        select(
            Product.item_id.label("item_id"),
            Item.item_code.label("item_code"),
            Item.item_name.label("item_name"),
            Item.specification.label("specification"),
            Product.drawing_no.label("drawing_no"),
            Product.product_family.label("product_family"),
            Product.version_no.label("version_no"),
            Item.unit_weight_kg.label("unit_weight_kg"),
            Product.allowed_scrap_rate.label("allowed_scrap_rate"),
            Product.standard_output_qty.label("standard_output_qty"),
            Product.quality_standard.label("quality_standard"),
            Item.safety_stock_weight_kg.label("safety_stock_weight_kg"),
            Item.status.label("status"),
        )
        .join(Item, Item.id == Product.item_id)
        .order_by(Product.id.desc())
        .limit(limit)
    )
    rows = db.execute(stmt).mappings().all()
    return [ProductRead.model_validate(dict(row)) for row in rows]


@router.post("/products", response_model=ProductRead)
def create_product(payload: ProductCreate, db: Session = Depends(get_db)) -> ProductRead:
    payload = payload.model_copy(update={"item_code": _build_product_code(db, payload)})
    _validate_product_payload(payload)

    if _item_code_used_by_non_product(db, payload.item_code):
        raise HTTPException(status_code=400, detail="该产品编码已被其他物料占用")
    if _product_code_version_exists(db, payload.item_code, payload.version_no):
        raise HTTPException(status_code=400, detail="相同产品编码和版本号已存在")

    item_id = _create_versioned_master_record(db, lambda: _create_product_record(db, payload))
    _commit_versioned_master_changes(db)

    return _get_product_read(db, item_id)


@router.put("/products/{item_id}", response_model=ProductRead)
def update_product(item_id: int, payload: ProductUpdate, db: Session = Depends(get_db)) -> ProductRead:
    item = db.get(Item, item_id)
    if not item or item.item_type not in {"PRODUCT", "FINISHED_GOOD"}:
        raise HTTPException(status_code=404, detail="产品不存在")
    _ensure_active_master_record(item.status, "已停用的产品需规只读，不允许修改")

    payload = payload.model_copy(update={"item_code": item.item_code})
    _validate_product_payload(payload)

    product = db.scalar(select(Product).where(Product.item_id == item_id))
    if not product:
        raise HTTPException(status_code=404, detail="产品扩展信息不存在")

    if _item_code_used_by_non_product(db, payload.item_code, exclude_item_id=item_id):
        raise HTTPException(status_code=400, detail="该产品编码已被其他物料占用")

    version_changed = _normalize_text(payload.version_no) != _normalize_text(product.version_no)

    if version_changed:
        if _product_code_version_exists(db, payload.item_code, payload.version_no):
            raise HTTPException(status_code=400, detail="相同产品编码和版本号已存在")
        new_item_id = _create_versioned_master_record(db, lambda: _create_product_record(db, payload))
        _commit_versioned_master_changes(db)
        return _get_product_read(db, new_item_id)

    if _product_code_version_exists(db, payload.item_code, payload.version_no, exclude_item_id=item_id):
        raise HTTPException(status_code=400, detail="相同产品编码和版本号已存在")

    if _product_has_usage(db, item_id):
        raise HTTPException(status_code=400, detail="该产品已被业务引用，请修改版本号后保存为新版本")

    item.item_code = _normalize_text(payload.item_code)
    item.item_name = _normalize_text(payload.item_name)
    item.specification = payload.specification
    item.unit_weight_kg = Decimal(str(payload.unit_weight_kg))
    item.safety_stock_weight_kg = Decimal(str(payload.safety_stock_weight_kg))
    item.status = payload.status

    product.drawing_no = payload.drawing_no
    product.product_family = payload.product_family
    product.version_no = _normalize_text(payload.version_no)
    product.standard_output_qty = Decimal(str(payload.standard_output_qty))
    product.allowed_scrap_rate = Decimal(str(payload.allowed_scrap_rate))
    product.quality_standard = payload.quality_standard

    db.add(item)
    db.add(product)
    _commit_versioned_master_changes(db)

    return _get_product_read(db, item_id)


@router.delete("/products/{item_id}")
def delete_product(item_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    item = db.get(Item, item_id)
    if not item or item.item_type not in {"PRODUCT", "FINISHED_GOOD"}:
        raise HTTPException(status_code=404, detail="产品不存在")
    _ensure_active_master_record(item.status, "已停用的产品需规只读，不允许删除")

    product = db.scalar(select(Product).where(Product.item_id == item_id))
    if not product:
        raise HTTPException(status_code=404, detail="产品扩展信息不存在")

    _ensure_product_can_delete(db, item_id)
    item_name = item.item_name
    db.delete(product)
    db.flush()
    db.delete(item)
    db.commit()
    return {"message": f"产品 {item_name} 已删除"}


@router.delete("/product-specs/{item_id}")
def delete_product_spec(item_id: int, db: Session = Depends(get_db)) -> dict[str, str]:
    item = db.get(Item, item_id)
    if not item or item.item_type not in {"PRODUCT", "FINISHED_GOOD"}:
        raise HTTPException(status_code=404, detail="产品需规不存在")
    _ensure_active_master_record(item.status, "已停用的产品需规只读，不允许删除")

    product = db.scalar(select(Product).where(Product.item_id == item_id))
    if not product:
        raise HTTPException(status_code=404, detail="产品扩展信息不存在")

    _ensure_product_spec_can_delete(db, item_id)
    item_name = item.item_name

    _delete_product_spec_records(db, item, product)
    db.commit()
    return {"message": f"产品需规 {item_name} 已删除"}


def _normalize_phone(value: str | None) -> str:
    phone = _normalize_text(value)
    if not phone:
        raise HTTPException(status_code=400, detail="手机号不能为空")
    if len(phone) > 20:
        raise HTTPException(status_code=400, detail="手机号长度不能超过20位")
    return phone


def _clean_miniapp_roles(roles: list[str] | None) -> list[str]:
    cleaned: list[str] = []
    for role in roles or ["worker"]:
        normalized = _normalize_text(role)
        if normalized not in MINIAPP_VALID_ROLES:
            raise HTTPException(status_code=400, detail=f"小程序角色不支持：{normalized}")
        if normalized not in cleaned:
            cleaned.append(normalized)
    return cleaned or ["worker"]


def _replace_miniapp_person_roles(db: Session, phone: str, roles: list[str] | None) -> None:
    db.execute(delete(MiniAppPersonRole).where(MiniAppPersonRole.phone == phone))
    for role in _clean_miniapp_roles(roles):
        db.add(MiniAppPersonRole(phone=phone, role=role))


def _miniapp_person_has_roles(db: Session, phone: str) -> bool:
    return bool(db.scalar(select(func.count()).select_from(MiniAppPersonRole).where(MiniAppPersonRole.phone == phone)))


def _miniapp_attendance_point_read(row: MiniAppAttendancePoint) -> MiniAppAttendancePointRead:
    return MiniAppAttendancePointRead(
        name=row.name,
        latitude=_optional_float(row.latitude),
        longitude=_optional_float(row.longitude),
        radius_meters=int(row.radius_meters or 500),
        remark=row.remark,
        is_active=bool(row.is_active),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _default_miniapp_attendance_point_names(db: Session) -> list[str]:
    names = [
        name
        for name in db.scalars(
            select(MiniAppAttendancePoint.name)
            .where(MiniAppAttendancePoint.is_active.is_(True))
            .order_by(MiniAppAttendancePoint.name.asc())
        ).all()
        if _normalize_text(name)
    ]
    return names


def _clean_miniapp_attendance_point_names(db: Session, point_names: list[str] | None) -> list[str]:
    cleaned = []
    for name in point_names or _default_miniapp_attendance_point_names(db):
        point_name = _normalize_text(name)
        if point_name and point_name not in cleaned:
            cleaned.append(point_name)
    if not cleaned:
        return []
    existing = set(db.scalars(select(MiniAppAttendancePoint.name).where(MiniAppAttendancePoint.name.in_(cleaned))).all())
    missing = [name for name in cleaned if name not in existing]
    if missing:
        raise HTTPException(status_code=400, detail=f"小程序考勤点不存在：{'、'.join(missing)}")
    return cleaned


def _replace_miniapp_person_attendance_points(db: Session, phone: str, point_names: list[str] | None) -> None:
    db.execute(delete(MiniAppPersonAttendancePoint).where(MiniAppPersonAttendancePoint.phone == phone))
    for point_name in _clean_miniapp_attendance_point_names(db, point_names):
        db.add(MiniAppPersonAttendancePoint(phone=phone, attendance_point_name=point_name))


def _build_miniapp_personnel_reads(db: Session, people: list[MiniAppPersonnel]) -> list[MiniAppPersonnelRead]:
    if not people:
        return []

    phones = [person.phone for person in people]
    role_rows = db.execute(
        select(MiniAppPersonRole.phone, MiniAppPersonRole.role)
        .where(MiniAppPersonRole.phone.in_(phones))
        .order_by(MiniAppPersonRole.phone, MiniAppPersonRole.role)
    ).all()
    role_map: dict[str, list[str]] = {}
    for phone, role in role_rows:
        role_map.setdefault(phone, [])
        if role not in role_map[phone]:
            role_map[phone].append(role)
    point_rows = db.execute(
        select(MiniAppPersonAttendancePoint.phone, MiniAppPersonAttendancePoint.attendance_point_name)
        .where(MiniAppPersonAttendancePoint.phone.in_(phones))
        .order_by(MiniAppPersonAttendancePoint.phone, MiniAppPersonAttendancePoint.attendance_point_name)
    ).all()
    point_map: dict[str, list[str]] = {}
    for phone, point_name in point_rows:
        point_map.setdefault(phone, [])
        if point_name not in point_map[phone]:
            point_map[phone].append(point_name)

    now = datetime.now()
    result: list[MiniAppPersonnelRead] = []
    for person in people:
        roles = role_map.get(person.phone, [])
        role_names = [MINIAPP_ROLE_LABELS.get(role, role) for role in roles]
        result.append(
            MiniAppPersonnelRead(
                phone=person.phone,
                name=person.name,
                role=roles[0] if roles else None,
                role_name="、".join(role_names) if role_names else None,
                roles=roles,
                role_names=role_names,
                attendance_point_names=point_map.get(person.phone, []),
                is_temporary=bool(person.is_temporary),
                temporary_expires_at=person.temporary_expires_at,
                temporary_expired=bool(person.is_temporary and person.temporary_expires_at and person.temporary_expires_at <= now),
                created_at=person.created_at,
                updated_at=person.updated_at,
            )
        )
    return result


def _active_miniapp_people(db: Session, *, limit: int | None = None) -> list[MiniAppPersonnel]:
    role_phones = select(MiniAppPersonRole.phone).distinct()
    stmt = (
        select(MiniAppPersonnel)
        .where(MiniAppPersonnel.phone.in_(role_phones))
        .order_by(MiniAppPersonnel.updated_at.desc(), MiniAppPersonnel.phone.asc())
    )
    if limit is not None:
        stmt = stmt.limit(limit)
    return db.scalars(stmt).all()


def _deactivate_missing_miniapp_employees(db: Session, active_phones: set[str]) -> None:
    synced_employees = db.scalars(
        select(Employee).where(
            Employee.mobile.is_not(None),
            Employee.status == "ACTIVE",
            or_(
                Employee.remark == "由小程序人员清单同步生成",
                Employee.job_title.in_(["小程序人员", "小程序报工员工"]),
            ),
        )
    ).all()
    for employee in synced_employees:
        phone = _normalize_text(employee.mobile)
        if phone and phone not in active_phones:
            employee.status = "INACTIVE"
            employee.remark = "小程序人员清单已删除或停用"
            db.add(employee)


def _sync_miniapp_personnel_to_employees(
    db: Session,
    *,
    people: list[MiniAppPersonnel] | None = None,
    deactivate_missing: bool = False,
) -> None:
    miniapp_people = people if people is not None else _active_miniapp_people(db)
    active_phones = {_normalize_text(person.phone) for person in miniapp_people if _normalize_text(person.phone)}
    if deactivate_missing:
        _deactivate_missing_miniapp_employees(db, active_phones)
    if not miniapp_people:
        return

    department = db.scalar(select(Department).order_by(Department.id).limit(1))
    if not department:
        raise HTTPException(status_code=400, detail="ERP 未配置部门，无法把小程序人员映射到业务员工档案")

    phones = [_normalize_text(person.phone) for person in miniapp_people if _normalize_text(person.phone)]
    existing_employees = db.scalars(select(Employee).where(Employee.mobile.in_(phones))).all() if phones else []
    employee_by_phone = {_normalize_text(employee.mobile): employee for employee in existing_employees if _normalize_text(employee.mobile)}

    for person in miniapp_people:
        phone = _normalize_text(person.phone)
        if not phone:
            continue
        name = _normalize_text(person.name) or phone
        employee = employee_by_phone.get(phone)
        if not employee:
            employee = Employee(
                employee_code=next_employee_code(db),
                employee_name=name,
                dept_id=department.id,
                mobile=phone,
                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()
            employee_by_phone[phone] = employee
            continue

        changed = False
        if employee.employee_name != name:
            employee.employee_name = name
            changed = True
        if employee.dept_id is None:
            employee.dept_id = department.id
            changed = True
        if employee.status != "ACTIVE":
            employee.status = "ACTIVE"
            changed = True
        if employee.job_title in {None, "", "小程序报工员工"}:
            employee.job_title = "小程序人员"
            changed = True
        if not employee.is_operator:
            employee.is_operator = 1
            changed = True
        if not employee.is_workshop_staff:
            employee.is_workshop_staff = 1
            changed = True
        if changed:
            db.add(employee)


def _miniapp_product_key(payload: MiniAppProductUpsert, *, original: bool = False) -> dict[str, str]:
    if original:
        return {
            "attendance_point_name": _normalize_text(payload.original_attendance_point_name or payload.attendance_point_name),
            "project_no": _normalize_text(payload.original_project_no or payload.project_no),
            "product_name": _normalize_text(payload.original_product_name or payload.product_name),
            "device_no": _normalize_text(payload.original_device_no if payload.original_device_no is not None else payload.device_no),
            "process_name": _normalize_text(
                payload.original_process_name if payload.original_process_name is not None else payload.process_name
            ),
        }
    return {
        "attendance_point_name": _normalize_text(payload.attendance_point_name),
        "project_no": _normalize_text(payload.project_no),
        "product_name": _normalize_text(payload.product_name),
        "device_no": _normalize_text(payload.device_no),
        "process_name": _export_miniapp_process_name(payload.process_name),
    }


def _export_miniapp_process_name(value: object | None) -> str:
    text = _normalize_text(value)
    if not text or text == MINIAPP_LEGACY_CLEANING_PROCESS_NAME:
        return ""
    if text == MINIAPP_MISC_PROCESS_NAME:
        return MINIAPP_MISC_PROCESS_NAME
    legacy_match = re.fullmatch(r"0*(\d+)序", text)
    if legacy_match:
        return str(int(legacy_match.group(1)))
    number_match = re.fullmatch(r"0*(\d+)", text)
    if number_match:
        return str(int(number_match.group(1)))
    return text


def _validate_miniapp_process_name(value: object | None, stamping_method: object | None = None) -> str:
    raw_text = _normalize_text(value)
    stamping_text = _normalize_text(stamping_method)
    if raw_text == MINIAPP_LEGACY_CLEANING_PROCESS_NAME:
        raise ValueError("清洗请填写在冲压方式中，工序只能填写正整数")
    process_name = _export_miniapp_process_name(raw_text)
    if not process_name:
        raise ValueError("工序必须填写正整数")
    if process_name == MINIAPP_MISC_PROCESS_NAME:
        if stamping_text != MINIAPP_MISC_STAMPING_METHOD:
            raise ValueError("只有处理杂活的工序可以填写杂活")
        return process_name
    if stamping_text == MINIAPP_MISC_STAMPING_METHOD:
        raise ValueError("处理杂活的工序必须填写杂活")
    if not re.fullmatch(r"\d+", process_name) or int(process_name) <= 0:
        raise ValueError("工序必须填写正整数")
    return process_name


def _optional_decimal(value: object | None) -> Decimal | None:
    if value is None:
        return None
    text = str(value).strip()
    if text == "":
        return None
    return Decimal(text)


def _optional_float(value: object | None) -> float | None:
    if value is None:
        return None
    return float(value)


def _decimal_or_zero(value: object | None, *, scale: Decimal | None = None) -> Decimal:
    decimal_value = _optional_decimal(value) or Decimal("0")
    return decimal_value.quantize(scale) if scale else decimal_value


def _decimal_changed(current: object | None, next_value: object | None, *, scale: Decimal | None = None) -> bool:
    return _decimal_or_zero(current, scale=scale) != _decimal_or_zero(next_value, scale=scale)


def _set_attr_if_changed(target, attr_name: str, next_value) -> bool:
    if getattr(target, attr_name) == next_value:
        return False
    setattr(target, attr_name, next_value)
    return True


def _import_text(value: Any) -> str:
    return str(value or "").strip()


def _import_optional_float(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _import_number(value: Any, default: float = 0) -> float:
    parsed = _import_optional_float(value)
    return default if parsed is None else parsed


def _normalized_import_value(value: Any) -> str:
    text = _import_text(value)
    if not text:
        return ""
    try:
        return str(Decimal(text).normalize())
    except InvalidOperation:
        return text


def _build_miniapp_product_workbook(rows: list[MiniAppProduct]) -> Workbook:
    workbook = Workbook()
    worksheet = workbook.active
    worksheet.title = "产品清单"
    worksheet.freeze_panes = "A2"
    worksheet.append(PRODUCT_EXCEL_HEADERS)

    for row in rows:
        worksheet.append(
            [
                row.attendance_point_name,
                row.project_no,
                row.profile_no or "",
                row.product_name,
                row.material_code or "",
                row.material_name or "",
                row.supplier or "",
                _export_miniapp_process_name(row.process_name),
                row.stamping_method or "",
                _optional_float(row.operator_count) or 0,
                "",
                _optional_float(row.standard_beat) or 0,
                _optional_float(row.standard_workload) or 0,
                _optional_float(row.product_net_weight_kg) if row.product_net_weight_kg is not None else "",
                _optional_float(row.product_gross_weight_kg) if row.product_gross_weight_kg is not None else "",
                _optional_float(row.scrap_loss_rate) if row.scrap_loss_rate is not None else "",
                _optional_float(row.waste_price_yuan_per_kg) if row.waste_price_yuan_per_kg is not None else "",
            ]
        )

    column_widths = [18, 14, 16, 28, 18, 26, 22, 14, 14, 12, 20, 12, 14, 16, 16, 14, 18]
    for index, width in enumerate(column_widths, start=1):
        worksheet.column_dimensions[worksheet.cell(row=1, column=index).column_letter].width = width
    return workbook


def _excel_response(workbook: Workbook, filename: str) -> Response:
    output = BytesIO()
    workbook.save(output)
    encoded_filename = quote(filename)
    return Response(
        content=output.getvalue(),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename=\"master-data.xlsx\"; filename*=UTF-8''{encoded_filename}"},
    )


def _excel_workbook_response(workbook: Workbook, filename: str, ascii_filename: str = "master-data.xlsx") -> Response:
    output = BytesIO()
    workbook.save(output)
    encoded_filename = quote(filename)
    return Response(
        content=output.getvalue(),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename=\"{ascii_filename}\"; filename*=UTF-8''{encoded_filename}"},
    )


def _read_excel_rows(content: bytes) -> list[dict[str, Any]]:
    try:
        workbook = load_workbook(BytesIO(content), read_only=True, data_only=True)
    except Exception as exc:
        raise ValueError("Excel 文件读取失败，请确认上传的是小程序产品清单 xlsx 文件") from exc

    worksheet = workbook.active
    rows = list(worksheet.iter_rows(values_only=True))
    if not rows:
        return []

    headers = [_import_text(value) for value in rows[0]]
    mapped_rows: list[dict[str, Any]] = []
    for values in rows[1:]:
        mapped_rows.append({headers[index]: value for index, value in enumerate(values) if index < len(headers)})
    return mapped_rows


def _material_export_stmt():
    return (
        select(
            Material.item_id.label("item_id"),
            Item.item_code.label("item_code"),
            Item.item_name.label("item_name"),
            Item.specification.label("specification"),
            Item.material_grade.label("material_grade"),
            Item.unit_weight_kg.label("unit_weight_kg"),
            Item.safety_stock_weight_kg.label("safety_stock_weight_kg"),
            Item.scrap_sale_price.label("scrap_sale_price"),
            Material.default_supplier_id.label("default_supplier_id"),
            Supplier.supplier_name.label("supplier_name"),
            Material.purchase_calc_mode.label("purchase_calc_mode"),
            Material.min_purchase_qty.label("min_purchase_qty"),
            Material.purchase_multiple_qty.label("purchase_multiple_qty"),
            Material.purchase_multiple_weight_kg.label("purchase_multiple_weight_kg"),
            Material.lead_time_days.label("lead_time_days"),
            Material.quality_rule.label("quality_rule"),
            Material.quality_rule.label("remark"),
            Item.status.label("status"),
        )
        .join(Item, Item.id == Material.item_id)
        .outerjoin(Supplier, Supplier.id == Material.default_supplier_id)
        .order_by(Item.item_code.asc(), Item.id.asc())
    )


def _material_status_label(value: str | None) -> str:
    return MATERIAL_STATUS_LABELS.get(_import_text(value).upper(), _import_text(value) or "启用")


def _parse_material_status(value: object | None) -> str:
    text = _import_text(value)
    if not text:
        return "ACTIVE"
    normalized = text.upper()
    if text in {"启用", "有效", "正常"} or normalized == "ACTIVE":
        return "ACTIVE"
    if text in {"停用", "禁用", "无效"} or normalized == "INACTIVE":
        return "INACTIVE"
    raise ValueError(f"未知原材料状态：{text}，请填写启用或停用")


def _build_material_workbook(rows: list[Any]) -> Workbook:
    workbook = Workbook()
    worksheet = workbook.active
    worksheet.title = "原材料名录"
    worksheet.freeze_panes = "A2"
    worksheet.append(MATERIAL_EXCEL_HEADERS)
    for row in rows:
        payload = _normalize_material_read_payload(dict(row))
        worksheet.append(
            [
                payload.get("item_code") or "",
                payload.get("item_name") or "",
                payload.get("material_grade") or "",
                payload.get("specification") or "",
                float(payload.get("scrap_sale_price") or 0),
                payload.get("remark") or "",
                _material_status_label(payload.get("status")),
            ]
        )
    for index, width in enumerate([18, 28, 16, 22, 14, 38, 12], start=1):
        worksheet.column_dimensions[worksheet.cell(row=1, column=index).column_letter].width = width
    return workbook


def _read_material_excel_rows(content: bytes) -> list[dict[str, Any]]:
    try:
        workbook = load_workbook(BytesIO(content), read_only=True, data_only=True)
    except Exception as exc:
        raise ValueError("Excel 文件读取失败，请确认上传的是原材料名录 xlsx 文件") from exc

    worksheet = workbook.active
    rows = list(worksheet.iter_rows(values_only=True))
    if not rows:
        return []
    headers = [_import_text(value) for value in rows[0]]
    missing = [header for header in MATERIAL_EXCEL_REQUIRED_HEADERS if header not in headers]
    if missing:
        raise ValueError(f"Excel 缺少原材料名录必要列：{'、'.join(missing)}")
    return [{headers[index]: value for index, value in enumerate(row) if index < len(headers)} for row in rows[1:]]


def _material_scrap_price(row: dict[str, Any]) -> Decimal:
    value = row.get("废料单价")
    if value is None:
        value = row.get("废料单价(元/kg)")
    text = _import_text(value)
    if not text:
        return Decimal("0")
    try:
        return Decimal(text.replace("¥", "").replace(",", "")).quantize(Decimal("0.0001"))
    except InvalidOperation as exc:
        raise ValueError(f"废料单价格式不正确：{text}") from exc


def _find_material_item_by_natural_key(db: Session, *, item_name: str, material_grade: str, specification: str) -> Item | None:
    return db.scalar(
        select(Item)
        .join(Material, Material.item_id == Item.id)
        .where(
            Item.item_type == "RAW_MATERIAL",
            Item.item_name == item_name,
            Item.material_grade == material_grade,
            Item.specification == specification,
        )
        .order_by(Item.id.asc())
        .limit(1)
    )


def _import_material_excel(db: Session, content: bytes) -> dict[str, object]:
    rows = _read_material_excel_rows(content)
    if not rows:
        raise ValueError("Excel 中没有可导入的原材料数据")

    imported = 0
    updated = 0
    skipped = 0
    errors: list[str] = []

    for index, row in enumerate(rows, start=2):
        if not any(_import_text(value) for value in row.values()):
            continue
        item_name = _import_text(row.get("原材料名称"))
        material_grade = _import_text(row.get("材质"))
        specification = _import_text(row.get("规格"))
        if not item_name or not material_grade or not specification:
            skipped += 1
            errors.append(f"第{index}行缺少原材料名称、材质或规格")
            continue

        try:
            status = _parse_material_status(row.get("状态"))
            scrap_sale_price = _material_scrap_price(row)
        except ValueError as exc:
            skipped += 1
            errors.append(f"第{index}行{exc}")
            continue

        item_code = _import_text(row.get("原材料编码"))
        quality_rule = _import_text(row.get("性能要求")) or None
        item = db.scalar(
            select(Item)
            .where(Item.item_code == item_code, Item.item_type == "RAW_MATERIAL")
            .limit(1)
        ) if item_code else None
        if item is None:
            item = _find_material_item_by_natural_key(
                db,
                item_name=item_name,
                material_grade=material_grade,
                specification=specification,
            )

        if item is None:
            code_for_create = item_code or _build_material_code(db)
            duplicate = db.scalar(select(Item).where(Item.item_code == code_for_create).limit(1))
            if duplicate:
                skipped += 1
                errors.append(f"第{index}行原材料编码 {code_for_create} 已被其他物料占用")
                continue
            item = Item(
                item_code=code_for_create,
                item_name=item_name,
                item_type="RAW_MATERIAL",
                specification=specification,
                material_grade=material_grade,
                unit_weight_kg=Decimal("1"),
                safety_stock_weight_kg=Decimal("0"),
                scrap_sale_price=scrap_sale_price,
                status=status,
            )
            db.add(item)
            db.flush()
            db.add(
                Material(
                    item_id=item.id,
                    default_supplier_id=None,
                    purchase_calc_mode="BY_WEIGHT",
                    min_purchase_qty=Decimal("0"),
                    purchase_multiple_qty=Decimal("0"),
                    purchase_multiple_weight_kg=Decimal("0"),
                    lead_time_days=0,
                    quality_rule=quality_rule,
                )
            )
            db.flush()
            imported += 1
            continue

        material = db.scalar(select(Material).where(Material.item_id == item.id).limit(1))
        if material is None:
            material = Material(
                item_id=item.id,
                default_supplier_id=None,
                purchase_calc_mode="BY_WEIGHT",
                min_purchase_qty=Decimal("0"),
                purchase_multiple_qty=Decimal("0"),
                purchase_multiple_weight_kg=Decimal("0"),
                lead_time_days=0,
            )
            db.add(material)

        item.item_name = item_name
        item.specification = specification
        item.material_grade = material_grade
        item.unit_weight_kg = item.unit_weight_kg or Decimal("1")
        item.safety_stock_weight_kg = item.safety_stock_weight_kg or Decimal("0")
        item.scrap_sale_price = scrap_sale_price
        item.status = status
        material.purchase_calc_mode = "BY_WEIGHT"
        material.purchase_multiple_qty = Decimal("0")
        material.quality_rule = quality_rule
        db.add(item)
        db.add(material)
        updated += 1

    if imported == 0 and updated == 0 and skipped > 0:
        db.rollback()
        raise ValueError("Excel 中没有可导入的有效原材料数据")

    db.commit()
    return {
        "imported": imported,
        "updated": updated,
        "skipped": skipped,
        "errors": errors,
        "message": f"导入完成：新增 {imported} 条，更新 {updated} 条，跳过 {skipped} 条",
    }


def _validate_product_import_headers(rows: list[dict[str, Any]]) -> None:
    if not rows:
        raise ValueError("Excel 中没有可导入的数据")

    present_headers = set().union(*(row.keys() for row in rows))
    required_headers = {"项目号", "产品名称", "物料编码", "物料名称", "供应商", "工序"}
    missing = sorted(required_headers - present_headers)
    if missing:
        raise ValueError(f"Excel 缺少小程序产品清单必要列：{'、'.join(missing)}")


def _validate_product_import_aux_consistency(rows: list[dict[str, Any]], default_point_name: str) -> None:
    aux_headers = {label for _, label in PRODUCT_IMPORT_AUX_FIELDS}
    if not any(header in row for row in rows for header in aux_headers):
        return

    grouped: dict[str, tuple[int, str, str, str]] = {}
    errors: list[str] = []
    for index, raw in enumerate(rows, start=2):
        if not any(_import_text(value) for value in raw.values()):
            continue
        point_name = _import_text(raw.get("考勤点")) or default_point_name
        product_name = _import_text(raw.get("产品名称"))
        process_name = _import_text(raw.get("工序"))
        if not product_name:
            continue
        values = tuple(_normalized_import_value(raw.get(label)) for _, label in PRODUCT_IMPORT_AUX_FIELDS)
        raw_value = "|".join(values)
        text_value = "，".join(
            f"{label}={value or '空'}"
            for (_, label), value in zip(PRODUCT_IMPORT_AUX_FIELDS, values)
        )
        group_key = f"{point_name}||{product_name}"
        previous = grouped.get(group_key)
        if previous is None:
            grouped[group_key] = (index, process_name, text_value, raw_value)
            continue
        if previous[3] != raw_value:
            errors.append(
                f"考勤点【{point_name}】产品【{product_name}】不同工序的ERP辅助字段不一致："
                f"第{previous[0]}行({previous[1] or '未填写工序'})为【{previous[2]}】，"
                f"第{index}行({process_name or '未填写工序'})为【{text_value}】"
            )

    if errors:
        detail = "；".join(errors[:10])
        if len(errors) > 10:
            detail += f"；另有{len(errors) - 10}处不一致"
        raise ValueError(f"{detail}。本次导入已取消，数据库未写入。")


def _default_miniapp_product_import_point_name(db: Session) -> str:
    return (
        db.scalar(
            select(MiniAppAttendancePoint.name)
            .where(MiniAppAttendancePoint.is_active.is_(True))
            .order_by(MiniAppAttendancePoint.name.asc())
            .limit(1)
        )
        or "宁波百华智能科技有限公司"
    )


def _import_miniapp_product_excel(db: Session, content: bytes) -> MiniAppProductImportResult:
    rows = _read_excel_rows(content)
    _validate_product_import_headers(rows)
    default_point_name = _default_miniapp_product_import_point_name(db)
    _validate_product_import_aux_consistency(rows, default_point_name)

    point_names = [
        _import_text(raw.get("考勤点")) or default_point_name
        for raw in rows
        if any(_import_text(value) for value in raw.values())
    ]
    _clean_miniapp_attendance_point_names(db, point_names)

    imported = 0
    updated = 0
    skipped = 0
    errors: list[str] = []
    last_project_no = ""

    for index, raw in enumerate(rows, start=2):
        if not any(_import_text(value) for value in raw.values()):
            continue

        data = {field: raw[header] for header, field in PRODUCT_IMPORT_HEADERS.items() if header in raw}
        attendance_point_name = _import_text(data.get("attendance_point_name")) or default_point_name
        project_no = _import_text(data.get("project_no")) or last_project_no
        product_name = _import_text(data.get("product_name"))
        raw_process_name = _import_text(data.get("process_name"))
        stamping_method = _import_text(data.get("stamping_method"))
        standard_beat = _import_number(data.get("standard_beat"))
        is_cleaning = stamping_method == MINIAPP_CLEANING_STAMPING_METHOD

        if project_no:
            last_project_no = project_no
        try:
            process_name = _validate_miniapp_process_name(raw_process_name, stamping_method)
        except ValueError as exc:
            skipped += 1
            errors.append(f"第{index}行{exc}")
            continue
        if not project_no or not product_name or not process_name or (not is_cleaning and standard_beat <= 0):
            skipped += 1
            errors.append(f"第{index}行缺少项目号、产品名称、工序或标准节拍（冲压方式为清洗时标准节拍可空）")
            continue

        payload = MiniAppProductUpsert(
            attendance_point_name=attendance_point_name,
            project_no=project_no,
            product_name=product_name,
            profile_no=_import_text(data.get("profile_no")) or None,
            material_code=_import_text(data.get("material_code")) or None,
            material_name=_import_text(data.get("material_name")) or None,
            supplier=_import_text(data.get("supplier")) or None,
            product_net_weight_kg=_import_optional_float(data.get("product_net_weight_kg")),
            product_gross_weight_kg=_import_optional_float(data.get("product_gross_weight_kg")),
            allowed_scrap_rate=_import_optional_float(data.get("scrap_loss_rate")),
            scrap_loss_rate=_import_optional_float(data.get("scrap_loss_rate")),
            waste_price_yuan_per_kg=_import_optional_float(data.get("waste_price_yuan_per_kg")),
            device_no="",
            process_name=process_name,
            stamping_method=stamping_method or None,
            operator_count=_import_number(data.get("operator_count"), 1) or 1,
            standard_beat=0 if is_cleaning else standard_beat,
            standard_workload=0,
        )

        existing = db.get(MiniAppProduct, _miniapp_product_key(payload))
        _upsert_miniapp_product(db, payload)
        if existing:
            updated += 1
        else:
            imported += 1

    product_result = _sync_miniapp_product_specs_to_master_data(db)
    db.commit()
    _mark_miniapp_master_sync_fresh(db)
    return MiniAppProductImportResult(
        imported=imported,
        updated=updated,
        skipped=skipped,
        errors=errors,
        material_sync={"created": 0, "linked": 0, "supplier_updated": 0, "skipped": 0},
        product_specs=product_result,
        message=f"导入完成：新增 {imported} 条，更新 {updated} 条，跳过 {skipped} 条",
    )


def _sync_miniapp_waste_price_for_material(
    db: Session,
    *,
    material_code: str | None,
    material_name: str | None,
    waste_price: Decimal,
) -> int:
    conditions = []
    code = _normalize_text(material_code)
    name = _normalize_text(material_name)
    if code:
        conditions.append(MiniAppProduct.material_code == code)
    if name:
        conditions.append(MiniAppProduct.material_name == name)
    if not conditions:
        return 0

    rows = db.scalars(select(MiniAppProduct).where(or_(*conditions))).all()
    for row in rows:
        row.waste_price_yuan_per_kg = waste_price
        db.add(row)
    return len(rows)


def _sync_miniapp_supplier_for_material(
    db: Session,
    *,
    material_code: str | None,
    material_name: str | None,
    supplier_name: str | None,
) -> int:
    conditions = []
    code = _normalize_text(material_code)
    name = _normalize_text(material_name)
    if code:
        conditions.append(MiniAppProduct.material_code == code)
    if name:
        conditions.append(MiniAppProduct.material_name == name)
    if not conditions:
        return 0

    synced_supplier = _normalize_text(supplier_name) or None
    changed = 0
    rows = db.scalars(select(MiniAppProduct).where(or_(*conditions))).all()
    for row in rows:
        if (_normalize_text(row.supplier) or None) == synced_supplier:
            continue
        row.supplier = synced_supplier
        row.updated_at = datetime.now()
        db.add(row)
        changed += 1
    return changed


def _upsert_miniapp_product(db: Session, payload: MiniAppProductUpsert) -> MiniAppProduct:
    try:
        payload.process_name = _validate_miniapp_process_name(payload.process_name, payload.stamping_method)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    is_cleaning = _normalize_text(payload.stamping_method) == MINIAPP_CLEANING_STAMPING_METHOD
    is_misc = _normalize_text(payload.stamping_method) == MINIAPP_MISC_STAMPING_METHOD
    if not is_cleaning and not is_misc and float(payload.standard_beat or 0) <= 0:
        raise HTTPException(status_code=400, detail="非清洗工序必须填写标准节拍")
    if is_cleaning or is_misc:
        payload.standard_beat = 0
    payload.standard_workload = 0

    next_key = _miniapp_product_key(payload)
    if not next_key["attendance_point_name"] or not next_key["project_no"] or not next_key["product_name"]:
        raise HTTPException(status_code=400, detail="小程序产品清单必须填写考勤点、项目号和产品名称")
    _clean_miniapp_attendance_point_names(db, [next_key["attendance_point_name"]])

    original_key = _miniapp_product_key(payload, original=True)
    row = db.get(MiniAppProduct, original_key)
    if row is not None and original_key != next_key:
        if db.get(MiniAppProduct, next_key):
            raise HTTPException(status_code=409, detail="小程序产品清单中已存在相同考勤点、项目号、产品、设备号和工序")
        db.delete(row)
        db.flush()
        row = None
    if row is None:
        row = db.get(MiniAppProduct, next_key)
    if row is None:
        row = MiniAppProduct(**next_key)
        db.add(row)

    row.profile_no = payload.profile_no
    row.material_code = payload.material_code
    row.material_name = payload.material_name
    row.supplier = payload.supplier
    row.product_net_weight_kg = _optional_decimal(payload.product_net_weight_kg)
    row.product_gross_weight_kg = _optional_decimal(payload.product_gross_weight_kg)
    row.scrap_loss_rate = _optional_decimal(
        payload.allowed_scrap_rate if payload.allowed_scrap_rate is not None else payload.scrap_loss_rate
    )
    row.waste_price_yuan_per_kg = _optional_decimal(payload.waste_price_yuan_per_kg)
    row.stamping_method = payload.stamping_method
    row.operator_count = Decimal(str(payload.operator_count or 1))
    row.standard_beat = Decimal(str(payload.standard_beat or 0))
    row.standard_workload = Decimal("0")
    return row


def _miniapp_product_read(row: MiniAppProduct) -> MiniAppProductRead:
    process_name = _export_miniapp_process_name(row.process_name)
    display_name = f"{row.project_no} / {row.product_name} / {process_name or '-'}"
    return MiniAppProductRead(
        attendance_point_name=row.attendance_point_name,
        project_no=row.project_no,
        product_name=row.product_name,
        profile_no=row.profile_no,
        material_code=row.material_code,
        material_name=row.material_name,
        supplier=row.supplier,
        product_net_weight_kg=_optional_float(row.product_net_weight_kg),
        product_gross_weight_kg=_optional_float(row.product_gross_weight_kg),
        allowed_scrap_rate=_optional_float(row.scrap_loss_rate),
        scrap_loss_rate=_optional_float(row.scrap_loss_rate),
        waste_price_yuan_per_kg=_optional_float(row.waste_price_yuan_per_kg),
        device_no=row.device_no,
        process_name=process_name,
        stamping_method=row.stamping_method,
        operator_count=float(row.operator_count or 0),
        standard_beat=float(row.standard_beat or 0),
        standard_workload=float(row.standard_workload or 0),
        display_name=display_name,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _get_miniapp_master_sync_signature(db: Session) -> tuple[object, ...]:
    row_count, point_count, last_updated_at, net_sum, gross_sum, scrap_sum, waste_price_sum, supplier_count, supplier_max = db.execute(
        select(
            func.count(MiniAppProduct.project_no),
            func.count(func.distinct(MiniAppProduct.attendance_point_name)),
            func.max(MiniAppProduct.updated_at),
            func.sum(MiniAppProduct.product_net_weight_kg),
            func.sum(MiniAppProduct.product_gross_weight_kg),
            func.sum(MiniAppProduct.scrap_loss_rate),
            func.sum(MiniAppProduct.waste_price_yuan_per_kg),
            func.count(func.distinct(MiniAppProduct.supplier)),
            func.max(MiniAppProduct.supplier),
        )
    ).one()
    return (
        int(row_count or 0),
        int(point_count or 0),
        last_updated_at.isoformat() if last_updated_at else "",
        str(_decimal_or_zero(net_sum, scale=Decimal("0.0001"))),
        str(_decimal_or_zero(gross_sum, scale=Decimal("0.0001"))),
        str(_decimal_or_zero(scrap_sum, scale=Decimal("0.000001"))),
        str(_decimal_or_zero(waste_price_sum, scale=Decimal("0.0001"))),
        int(supplier_count or 0),
        _normalize_text(supplier_max),
    )


def _mark_miniapp_master_sync_fresh(db: Session) -> None:
    global _miniapp_master_sync_signature
    _miniapp_master_sync_signature = _get_miniapp_master_sync_signature(db)


def _get_or_create_miniapp_sync_supplier(db: Session) -> Supplier:
    supplier = db.scalar(select(Supplier).where(Supplier.supplier_name == "小程序物料同步供应商").limit(1))
    if supplier:
        return supplier
    supplier = Supplier(
        supplier_code=build_supplier_code(db),
        supplier_name="小程序物料同步供应商",
        short_name="小程序同步",
        contact_name=None,
        contact_phone=None,
        address=None,
        lead_time_days=0,
        default_tax_rate=Decimal("0"),
        status="ACTIVE",
        remark="系统自动创建，用于承接小程序产品清单中的物料编码/物料名称。请在02-1中改成真实供应商。",
    )
    db.add(supplier)
    db.flush()
    return supplier


def _get_or_create_supplier_by_name(
    db: Session,
    supplier_name: str | None,
    supplier_cache: dict[str, Supplier],
) -> Supplier:
    name = _normalize_text(supplier_name)
    if not name:
        return _get_or_create_miniapp_sync_supplier(db)
    if name in supplier_cache:
        return supplier_cache[name]

    supplier = db.scalar(
        select(Supplier)
        .where(or_(Supplier.supplier_name == name, Supplier.short_name == name))
        .order_by(Supplier.id.asc())
        .limit(1)
    )
    if not supplier:
        supplier = Supplier(
            supplier_code=build_supplier_code(db),
            supplier_name=name,
            short_name=name[:100],
            contact_name=None,
            contact_phone=None,
            address=None,
            lead_time_days=0,
            default_tax_rate=Decimal("0"),
            status="ACTIVE",
            remark="小程序产品清单供应商同步创建",
        )
        db.add(supplier)
        db.flush()

    supplier_cache[name] = supplier
    return supplier


def _sync_miniapp_materials_to_master_data(db: Session, *, commit: bool = True) -> dict[str, object]:
    miniapp_rows = db.execute(
        select(
            MiniAppProduct.material_code,
            MiniAppProduct.material_name,
            MiniAppProduct.supplier,
            MiniAppProduct.waste_price_yuan_per_kg,
            MiniAppProduct.updated_at,
        )
        .where(
            MiniAppProduct.material_code.is_not(None),
            MiniAppProduct.material_code != "",
        )
        .order_by(MiniAppProduct.material_code.asc(), MiniAppProduct.updated_at.desc())
    ).mappings().all()

    material_records: dict[str, dict[str, object]] = {}
    for row in miniapp_rows:
        code = _normalize_text(row["material_code"])
        if not code:
            continue
        record = material_records.setdefault(
            code,
            {
                "material_code": code,
                "material_name": "",
                "supplier": "",
                "supplier_names": set(),
                "waste_price_yuan_per_kg": None,
            },
        )
        material_name = _normalize_text(row["material_name"])
        supplier_name = _normalize_text(row["supplier"])
        waste_price = _optional_decimal(row["waste_price_yuan_per_kg"])
        if material_name and not record["material_name"]:
            record["material_name"] = material_name
        if supplier_name:
            supplier_names = record["supplier_names"]
            supplier_names.add(supplier_name)
            if not record["supplier"]:
                record["supplier"] = supplier_name
        if waste_price is not None:
            current_price = record["waste_price_yuan_per_kg"]
            if current_price is None or waste_price > current_price:
                record["waste_price_yuan_per_kg"] = waste_price

    material_rows = list(material_records.values())

    if not material_rows:
        return {
            "created": 0,
            "linked": 0,
            "supplier_updated": 0,
            "skipped": 0,
            "conflicts": [],
            "message": "小程序产品清单中暂无可同步物料",
        }

    supplier_cache: dict[str, Supplier] = {}
    material_codes = {_normalize_text(row["material_code"]) for row in material_rows if _normalize_text(row["material_code"])}
    material_names = {_normalize_text(row["material_name"]) for row in material_rows if _normalize_text(row["material_name"])}
    item_conditions = []
    if material_codes:
        item_conditions.append(Item.item_code.in_(material_codes))
    if material_names:
        item_conditions.append(Item.item_name.in_(material_names))
    existing_items = db.scalars(select(Item).where(or_(*item_conditions))).all() if item_conditions else []
    items_by_code = {_normalize_text(item.item_code): item for item in existing_items if _normalize_text(item.item_code)}
    items_by_name = {_normalize_text(item.item_name): item for item in existing_items if _normalize_text(item.item_name)}
    existing_materials = db.scalars(
        select(Material).where(Material.item_id.in_([item.id for item in existing_items]))
    ).all() if existing_items else []
    materials_by_item_id = {material.item_id: material for material in existing_materials}

    created = 0
    linked = 0
    supplier_updated = 0
    skipped = 0
    conflicts: list[str] = []

    for material_row in material_rows:
        code = _normalize_text(material_row["material_code"])
        if not code:
            continue
        name = _normalize_text(material_row["material_name"]) or code
        waste_price = _optional_decimal(material_row["waste_price_yuan_per_kg"])
        supplier_names = sorted(material_row["supplier_names"])
        if len(supplier_names) > 1:
            conflicts.append(f"{code} 存在多个小程序供应商，已采用最新记录：{material_row['supplier']}；全部值：{'、'.join(supplier_names)}")
        supplier = _get_or_create_supplier_by_name(db, _normalize_text(material_row["supplier"]), supplier_cache)

        item = items_by_code.get(code)
        if item and item.item_type != "RAW_MATERIAL":
            skipped += 1
            conflicts.append(f"{code} 已被非原材料物料占用")
            continue

        if not item:
            item = Item(
                item_code=code,
                item_name=name,
                item_type="RAW_MATERIAL",
                specification=name,
                material_grade="未维护",
                unit_weight_kg=Decimal("1"),
                safety_stock_weight_kg=Decimal("0"),
                scrap_sale_price=waste_price if waste_price is not None else Decimal("0"),
                status="ACTIVE",
            )
            db.add(item)
            db.flush()
            material = Material(
                item_id=item.id,
                default_supplier_id=supplier.id,
                purchase_calc_mode="BY_WEIGHT",
                min_purchase_qty=Decimal("0"),
                purchase_multiple_qty=Decimal("0"),
                purchase_multiple_weight_kg=Decimal("0"),
                lead_time_days=0,
                quality_rule="小程序产品清单同步，需在02-1补充真实规格、材质、重量和质检要求",
            )
            db.add(material)
            items_by_code[code] = item
            items_by_name[name] = item
            materials_by_item_id[item.id] = material
            created += 1
            continue

        material = materials_by_item_id.get(item.id)
        item_changed = False
        if waste_price is not None and _decimal_changed(item.scrap_sale_price, waste_price, scale=Decimal("0.0001")):
            item.scrap_sale_price = waste_price
            item_changed = True
        if name and item.item_name != name:
            item.item_name = name
            item_changed = True
        if item_changed:
            db.add(item)

        if not material:
            material = Material(
                item_id=item.id,
                default_supplier_id=supplier.id,
                purchase_calc_mode="BY_WEIGHT",
                min_purchase_qty=Decimal("0"),
                purchase_multiple_qty=Decimal("0"),
                purchase_multiple_weight_kg=Decimal("0"),
                lead_time_days=0,
                quality_rule="小程序产品清单同步，需在02-1补充真实规格、材质、重量和质检要求",
            )
            db.add(material)
            materials_by_item_id[item.id] = material
            linked += 1
        else:
            if material.default_supplier_id != supplier.id:
                material.default_supplier_id = supplier.id
                db.add(material)
                supplier_updated += 1
            else:
                skipped += 1

    if commit:
        db.commit()

    return {
        "created": created,
        "linked": linked,
        "supplier_updated": supplier_updated,
        "skipped": skipped,
        "conflicts": conflicts,
        "message": f"同步完成：新增原材料 {created} 条，补齐关联 {linked} 条，供应商更新 {supplier_updated} 条，已存在/跳过 {skipped} 条",
    }


def _sync_miniapp_product_specs_to_master_data(db: Session) -> dict[str, int]:
    rows = db.scalars(
        select(MiniAppProduct).order_by(MiniAppProduct.attendance_point_name.asc(), MiniAppProduct.updated_at.desc())
    ).all()
    context = _build_miniapp_product_sync_context(db, rows)
    matched = 0
    created = 0
    product_updated = 0
    bom_updated = 0
    route_updated = 0
    stale_deleted = 0
    stale_blocked = 0
    stale_deactivated = 0

    for group_rows in _group_miniapp_product_rows(rows):
        miniapp_product = group_rows[0]
        product_row = _find_product_for_miniapp_product_in_context(context, miniapp_product)
        if product_row:
            product, item = product_row
            matched += 1
        else:
            product, item = _create_product_from_miniapp_group(db, group_rows)
            if not product or not item:
                continue
            _register_product_in_miniapp_sync_context(context, product, item)
            created += 1
            matched += 1

        if _sync_miniapp_product_header_to_erp(db, product, item, group_rows):
            product_updated += 1
        if _sync_miniapp_bom_to_erp_fast(db, item, group_rows, context):
            bom_updated += 1
        if _sync_miniapp_route_to_erp_fast(db, item, group_rows, context):
            route_updated += 1

    stale_result = _deactivate_stale_miniapp_product_specs_from_master_data(db, rows)
    stale_deactivated = stale_result["deactivated"]

    return {
        "matched": matched,
        "created": created,
        "product_updated": product_updated,
        "bom_updated": bom_updated,
        "route_updated": route_updated,
        "stale_deleted": stale_deleted,
        "stale_blocked": stale_blocked,
        "stale_deactivated": stale_deactivated,
    }


def _build_miniapp_product_sync_context(db: Session, rows: list[MiniAppProduct]) -> dict[str, object]:
    product_names = {_normalize_text(row.product_name) for row in rows if _normalize_text(row.product_name)}
    project_nos = {_normalize_text(row.project_no) for row in rows if _normalize_text(row.project_no)}
    material_codes = {_normalize_text(row.material_code) for row in rows if _normalize_text(row.material_code)}
    material_names = {_normalize_text(row.material_name) for row in rows if _normalize_text(row.material_name)}
    process_names = {_export_miniapp_process_name(row.process_name) for row in rows if _export_miniapp_process_name(row.process_name)}

    product_conditions = []
    if product_names:
        product_conditions.append(Item.item_name.in_(product_names))
    if project_nos:
        product_conditions.append(Item.item_code.in_(project_nos))
    product_pairs = db.execute(
        select(Product, Item)
        .join(Item, Item.id == Product.item_id)
        .where(Item.item_type.in_(["PRODUCT", "FINISHED_GOOD"]), or_(*product_conditions))
    ).all() if product_conditions else []

    products_by_name: dict[str, tuple[Product, Item]] = {}
    products_by_code: dict[str, tuple[Product, Item]] = {}
    product_item_ids: list[int] = []
    for product, item in product_pairs:
        pair = (product, item)
        item_name = _normalize_text(item.item_name)
        item_code = _normalize_text(item.item_code)
        if item_name and item_name not in products_by_name:
            products_by_name[item_name] = pair
        if item_code and item_code not in products_by_code:
            products_by_code[item_code] = pair
        product_item_ids.append(item.id)

    material_conditions = []
    if material_codes:
        material_conditions.append(Item.item_code.in_(material_codes))
    if material_names:
        material_conditions.append(Item.item_name.in_(material_names))
    material_items = db.scalars(
        select(Item).where(Item.item_type == "RAW_MATERIAL", or_(*material_conditions))
    ).all() if material_conditions else []
    materials_by_code = {_normalize_text(item.item_code): item for item in material_items if _normalize_text(item.item_code)}
    materials_by_name = {_normalize_text(item.item_name): item for item in material_items if _normalize_text(item.item_name)}

    boms_by_product_id: dict[int, Bom] = {}
    bom_items_by_bom_id: dict[int, BomItem] = {}
    routes_by_product_id: dict[int, ProcessRoute] = {}
    route_ops_by_route_id: dict[int, list[ProcessRouteOperation]] = {}
    used_route_ids: set[int] = set()

    if product_item_ids:
        bom_rows = db.scalars(
            select(Bom)
            .where(Bom.product_item_id.in_(product_item_ids))
            .order_by(Bom.product_item_id, Bom.is_default.desc(), Bom.id.desc())
        ).all()
        for bom in bom_rows:
            boms_by_product_id.setdefault(bom.product_item_id, bom)

        bom_ids = [bom.id for bom in boms_by_product_id.values()]
        if bom_ids:
            bom_item_rows = db.scalars(
                select(BomItem)
                .where(BomItem.bom_id.in_(bom_ids))
                .order_by(BomItem.bom_id, BomItem.seq_no.asc(), BomItem.id.asc())
            ).all()
            for bom_item in bom_item_rows:
                bom_items_by_bom_id.setdefault(bom_item.bom_id, bom_item)

        route_rows = db.scalars(
            select(ProcessRoute)
            .where(ProcessRoute.product_item_id.in_(product_item_ids))
            .order_by(ProcessRoute.product_item_id, ProcessRoute.is_default.desc(), ProcessRoute.id.desc())
        ).all()
        for route in route_rows:
            routes_by_product_id.setdefault(route.product_item_id, route)

        route_ids = [route.id for route in routes_by_product_id.values()]
        if route_ids:
            route_op_rows = db.scalars(
                select(ProcessRouteOperation)
                .where(ProcessRouteOperation.route_id.in_(route_ids))
                .order_by(ProcessRouteOperation.route_id, ProcessRouteOperation.seq_no.asc(), ProcessRouteOperation.id.asc())
            ).all()
            for operation in route_op_rows:
                route_ops_by_route_id.setdefault(operation.route_id, []).append(operation)
            used_route_ids = set(
                db.scalars(select(WorkOrder.route_id).where(WorkOrder.route_id.in_(route_ids))).all()
            )

    process_rows = db.scalars(
        select(Process).where(Process.process_name.in_(process_names))
    ).all() if process_names else []
    processes_by_name = {_normalize_text(process.process_name): process for process in process_rows}
    first_process = db.scalar(select(Process).order_by(Process.id.asc()).limit(1))
    default_work_center = db.scalar(select(WorkCenter).order_by(WorkCenter.id.asc()).limit(1))

    return {
        "products_by_name": products_by_name,
        "products_by_code": products_by_code,
        "materials_by_code": materials_by_code,
        "materials_by_name": materials_by_name,
        "boms_by_product_id": boms_by_product_id,
        "bom_items_by_bom_id": bom_items_by_bom_id,
        "routes_by_product_id": routes_by_product_id,
        "route_ops_by_route_id": route_ops_by_route_id,
        "used_route_ids": used_route_ids,
        "processes_by_name": processes_by_name,
        "first_process": first_process,
        "default_work_center": default_work_center,
    }


def _find_product_for_miniapp_product_in_context(
    context: dict[str, object],
    miniapp_product: MiniAppProduct,
) -> tuple[Product, Item] | None:
    product_name = _normalize_text(miniapp_product.product_name)
    project_no = _normalize_text(miniapp_product.project_no)
    products_by_name = context["products_by_name"]
    products_by_code = context["products_by_code"]
    return products_by_name.get(product_name) or products_by_code.get(project_no)


def _register_product_in_miniapp_sync_context(context: dict[str, object], product: Product, item: Item) -> None:
    products_by_name = context["products_by_name"]
    products_by_code = context["products_by_code"]
    item_name = _normalize_text(item.item_name)
    item_code = _normalize_text(item.item_code)
    if item_name:
        products_by_name[item_name] = (product, item)
    if item_code:
        products_by_code[item_code] = (product, item)


def _deactivate_stale_miniapp_product_specs_from_master_data(db: Session, rows: list[MiniAppProduct]) -> dict[str, int]:
    active_product_names = {_normalize_text(row.product_name) for row in rows if _normalize_text(row.product_name)}
    active_project_nos = {_normalize_text(row.project_no) for row in rows if _normalize_text(row.project_no)}
    deactivated = 0

    stale_pairs = db.execute(
        select(Product, Item)
        .join(Item, Item.id == Product.item_id)
        .join(ProcessRoute, ProcessRoute.product_item_id == Item.id)
        .outerjoin(ProcessRouteOperation, ProcessRouteOperation.route_id == ProcessRoute.id)
        .where(
            Item.item_type.in_(["PRODUCT", "FINISHED_GOOD"]),
            or_(
                ProcessRoute.remark == MINIAPP_SYNC_REMARK,
                ProcessRouteOperation.remark == MINIAPP_SYNC_REMARK,
            ),
            ~Item.item_name.in_(active_product_names or {""}),
            ~Item.item_code.in_(active_project_nos or {""}),
        )
        .distinct()
    ).all()

    for product, item in stale_pairs:
        if _deactivate_product_spec_records(db, item):
            deactivated += 1

    return {"deactivated": deactivated}


def _deactivate_product_spec_records(db: Session, item: Item) -> bool:
    changed = False
    if item.status != "INACTIVE":
        item.status = "INACTIVE"
        db.add(item)
        changed = True

    boms = db.scalars(select(Bom).where(Bom.product_item_id == item.id)).all()
    for bom in boms:
        if bom.status != "INACTIVE":
            bom.status = "INACTIVE"
            db.add(bom)
            changed = True

    routes = db.scalars(select(ProcessRoute).where(ProcessRoute.product_item_id == item.id)).all()
    route_ids = [route.id for route in routes]
    for route in routes:
        if route.status != "INACTIVE":
            route.status = "INACTIVE"
            db.add(route)
            changed = True
    if route_ids:
        operations = db.scalars(
            select(ProcessRouteOperation).where(ProcessRouteOperation.route_id.in_(route_ids))
        ).all()
        for operation in operations:
            if operation.status != "INACTIVE":
                operation.status = "INACTIVE"
                db.add(operation)
                changed = True

    return changed


def _delete_product_spec_records(db: Session, item: Item, product: Product) -> None:
    route_ids = db.scalars(select(ProcessRoute.id).where(ProcessRoute.product_item_id == item.id)).all()
    if route_ids:
        db.execute(delete(ProcessRouteOperation).where(ProcessRouteOperation.route_id.in_(route_ids)))
        db.execute(delete(ProcessRoute).where(ProcessRoute.id.in_(route_ids)))

    bom_ids = db.scalars(select(Bom.id).where(Bom.product_item_id == item.id)).all()
    if bom_ids:
        db.execute(delete(BomItem).where(BomItem.bom_id.in_(bom_ids)))
        db.execute(delete(Bom).where(Bom.id.in_(bom_ids)))

    db.delete(product)
    db.flush()
    db.delete(item)


def _group_miniapp_product_rows(rows: list[MiniAppProduct]) -> list[list[MiniAppProduct]]:
    groups: dict[tuple[str, str, str], list[MiniAppProduct]] = {}
    for row in rows:
        key = (_normalize_text(row.attendance_point_name), _normalize_text(row.project_no), _normalize_text(row.product_name))
        if not key[0] or not key[1] or not key[2]:
            continue
        groups.setdefault(key, []).append(row)
    grouped_rows: list[list[MiniAppProduct]] = []
    for group in groups.values():
        deduped: dict[str, MiniAppProduct] = {}
        for row in _sort_miniapp_process_rows(group):
            process_name = _export_miniapp_process_name(row.process_name)
            if not process_name:
                continue
            existing = deduped.get(process_name)
            if existing is None or _normalize_text(row.process_name) == process_name:
                deduped[process_name] = row
        if deduped:
            grouped_rows.append(list(deduped.values()))
    return grouped_rows


def _sort_miniapp_process_rows(rows: list[MiniAppProduct]) -> list[MiniAppProduct]:
    def sort_key(row: MiniAppProduct) -> tuple[int, str]:
        process_name = _export_miniapp_process_name(row.process_name)
        match = re.search(r"\d+", process_name)
        return (int(match.group(0)) if match else 9999, process_name)

    return sorted(rows, key=sort_key)


def _first_decimal(rows: list[MiniAppProduct], attr_name: str) -> Decimal | None:
    for row in rows:
        value = _optional_decimal(getattr(row, attr_name, None))
        if value is not None:
            return value
    return None


def _first_text(rows: list[MiniAppProduct], attr_name: str) -> str:
    for row in rows:
        value = _normalize_text(getattr(row, attr_name, None))
        if value:
            return value
    return ""


def _create_product_from_miniapp_group(db: Session, rows: list[MiniAppProduct]) -> tuple[Product | None, Item | None]:
    first = rows[0]
    product_name = _normalize_text(first.product_name)
    if not product_name:
        return None, None

    net_weight = _first_decimal(rows, "product_net_weight_kg") or Decimal("1")
    allowed_scrap_rate = _first_decimal(rows, "scrap_loss_rate") or Decimal("0")
    item = Item(
        item_code=_build_product_code(db),
        item_name=product_name,
        item_type="FINISHED_GOOD",
        specification=None,
        material_grade=None,
        unit_weight_kg=net_weight,
        safety_stock_weight_kg=Decimal("0"),
        status="ACTIVE",
    )
    db.add(item)
    db.flush()

    product = Product(
        item_id=item.id,
        drawing_no=_normalize_text(first.project_no) or None,
        product_family=_normalize_text(first.project_no) or None,
        version_no="V1.0",
        standard_output_qty=Decimal("0"),
        allowed_scrap_rate=allowed_scrap_rate,
        quality_standard="小程序产品清单同步生成",
    )
    db.add(product)
    db.flush()
    return product, item


def _sync_miniapp_product_header_to_erp(db: Session, product: Product, item: Item, rows: list[MiniAppProduct]) -> bool:
    changed = False
    net_weight = _first_decimal(rows, "product_net_weight_kg")
    allowed_scrap_rate = _first_decimal(rows, "scrap_loss_rate")
    project_no = _normalize_text(rows[0].project_no)

    if net_weight is not None and _decimal_changed(item.unit_weight_kg, net_weight, scale=Decimal("0.000001")):
        item.unit_weight_kg = net_weight
        changed = True
    if allowed_scrap_rate is not None and _decimal_changed(product.allowed_scrap_rate, allowed_scrap_rate, scale=Decimal("0.0001")):
        product.allowed_scrap_rate = allowed_scrap_rate
        changed = True
    if project_no and product.drawing_no != project_no:
        product.drawing_no = project_no
        changed = True
    if project_no and product.product_family != project_no:
        product.product_family = project_no
        changed = True
    if item.status != "ACTIVE":
        item.status = "ACTIVE"
        changed = True
    if changed:
        db.add(item)
        db.add(product)
    return changed


def _sync_miniapp_bom_to_erp(db: Session, product_item: Item, rows: list[MiniAppProduct]) -> bool:
    material_item = _find_material_for_miniapp_rows(db, rows)
    if not material_item:
        return False

    net_weight = _first_decimal(rows, "product_net_weight_kg")
    gross_weight = _first_decimal(rows, "product_gross_weight_kg")
    if net_weight is None and gross_weight is None:
        return False

    bom = db.scalar(
        select(Bom)
        .where(Bom.product_item_id == product_item.id)
        .order_by(Bom.is_default.desc(), Bom.id.desc())
        .limit(1)
    )
    if not bom:
        bom = Bom(
            bom_code=_build_bom_code(db, product_item.id),
            product_item_id=product_item.id,
            version_no="V1.0",
            is_default=1,
            status="ACTIVE",
            effective_date=None,
            expire_date=None,
            remark="小程序产品清单同步生成",
        )
        db.add(bom)
        db.flush()
    elif bom.status != "ACTIVE":
        bom.status = "ACTIVE"
        db.add(bom)

    bom_item = db.scalar(
        select(BomItem)
        .where(BomItem.bom_id == bom.id)
        .order_by(BomItem.seq_no.asc(), BomItem.id.asc())
        .limit(1)
    )
    if not bom_item:
        bom_item = BomItem(
            bom_id=bom.id,
            seq_no=1,
            material_item_id=material_item.id,
            usage_type="WEIGHT",
            qty_per_piece=Decimal("1"),
            net_weight_per_piece_kg=Decimal("0"),
            gross_weight_per_piece_kg=Decimal("0"),
            loss_rate=Decimal("0"),
            issue_over_tolerance_rate=Decimal("0"),
            remark="小程序产品清单同步生成",
        )
        db.add(bom_item)

    effective_net_weight = net_weight if net_weight is not None else _optional_decimal(bom_item.net_weight_per_piece_kg)
    effective_gross_weight = gross_weight if gross_weight is not None else _optional_decimal(bom_item.gross_weight_per_piece_kg)
    bom_item.material_item_id = material_item.id
    if net_weight is not None:
        bom_item.net_weight_per_piece_kg = net_weight
    if gross_weight is not None:
        bom_item.gross_weight_per_piece_kg = gross_weight
    if effective_net_weight is not None and effective_gross_weight is not None:
        bom_item.loss_rate = _calculate_loss_rate(effective_net_weight, effective_gross_weight)
    db.add(bom)
    db.add(bom_item)
    return True


def _sync_miniapp_bom_to_erp_fast(
    db: Session,
    product_item: Item,
    rows: list[MiniAppProduct],
    context: dict[str, object],
) -> bool:
    material_item = _find_material_for_miniapp_rows_in_context(context, rows)
    if not material_item:
        return False

    net_weight = _first_decimal(rows, "product_net_weight_kg")
    gross_weight = _first_decimal(rows, "product_gross_weight_kg")
    if net_weight is None and gross_weight is None:
        return False

    boms_by_product_id = context["boms_by_product_id"]
    bom_items_by_bom_id = context["bom_items_by_bom_id"]
    bom = boms_by_product_id.get(product_item.id)
    changed = False
    if not bom:
        bom = Bom(
            bom_code=_build_bom_code(db, product_item.id),
            product_item_id=product_item.id,
            version_no="V1.0",
            is_default=1,
            status="ACTIVE",
            effective_date=None,
            expire_date=None,
            remark="小程序产品清单同步生成",
        )
        db.add(bom)
        db.flush()
        boms_by_product_id[product_item.id] = bom
        changed = True
    elif bom.status != "ACTIVE":
        bom.status = "ACTIVE"
        changed = True

    bom_item = bom_items_by_bom_id.get(bom.id)
    if not bom_item:
        bom_item = BomItem(
            bom_id=bom.id,
            seq_no=1,
            material_item_id=material_item.id,
            usage_type="WEIGHT",
            qty_per_piece=Decimal("1"),
            net_weight_per_piece_kg=Decimal("0"),
            gross_weight_per_piece_kg=Decimal("0"),
            loss_rate=Decimal("0"),
            issue_over_tolerance_rate=Decimal("0"),
            remark="小程序产品清单同步生成",
        )
        db.add(bom_item)
        bom_items_by_bom_id[bom.id] = bom_item
        changed = True

    effective_net_weight = net_weight if net_weight is not None else _optional_decimal(bom_item.net_weight_per_piece_kg)
    effective_gross_weight = gross_weight if gross_weight is not None else _optional_decimal(bom_item.gross_weight_per_piece_kg)
    if bom_item.material_item_id != material_item.id:
        bom_item.material_item_id = material_item.id
        changed = True
    if net_weight is not None and _decimal_changed(bom_item.net_weight_per_piece_kg, net_weight, scale=Decimal("0.000001")):
        bom_item.net_weight_per_piece_kg = net_weight
        changed = True
    if gross_weight is not None and _decimal_changed(bom_item.gross_weight_per_piece_kg, gross_weight, scale=Decimal("0.000001")):
        bom_item.gross_weight_per_piece_kg = gross_weight
        changed = True
    if effective_net_weight is not None and effective_gross_weight is not None:
        loss_rate = _calculate_loss_rate(effective_net_weight, effective_gross_weight)
        if _decimal_changed(bom_item.loss_rate, loss_rate, scale=Decimal("0.0001")):
            bom_item.loss_rate = loss_rate
            changed = True
    if changed:
        db.add(bom)
        db.add(bom_item)
    return changed


def _find_material_for_miniapp_rows(db: Session, rows: list[MiniAppProduct]) -> Item | None:
    material_code = _first_text(rows, "material_code")
    material_name = _first_text(rows, "material_name")
    if material_code:
        item = db.scalar(
            select(Item)
            .where(Item.item_type == "RAW_MATERIAL", Item.item_code == material_code)
            .limit(1)
        )
        if item:
            return item
    if material_name:
        return db.scalar(
            select(Item)
            .where(Item.item_type == "RAW_MATERIAL", Item.item_name == material_name)
            .limit(1)
        )
    return None


def _find_material_for_miniapp_rows_in_context(context: dict[str, object], rows: list[MiniAppProduct]) -> Item | None:
    material_code = _first_text(rows, "material_code")
    material_name = _first_text(rows, "material_name")
    materials_by_code = context["materials_by_code"]
    materials_by_name = context["materials_by_name"]
    return materials_by_code.get(material_code) or materials_by_name.get(material_name)


def _sync_miniapp_route_to_erp(db: Session, product_item: Item, rows: list[MiniAppProduct]) -> bool:
    process_rows = [row for row in rows if _export_miniapp_process_name(row.process_name)]
    if not process_rows:
        return False
    default_work_center = db.scalar(select(WorkCenter).order_by(WorkCenter.id.asc()).limit(1))
    first_process = db.scalar(select(Process).order_by(Process.id.asc()).limit(1))
    if not default_work_center or not first_process:
        return False

    route = db.scalar(
        select(ProcessRoute)
        .where(ProcessRoute.product_item_id == product_item.id)
        .order_by(ProcessRoute.is_default.desc(), ProcessRoute.id.desc())
        .limit(1)
    )
    allowed_scrap_rate = _first_decimal(rows, "scrap_loss_rate") or Decimal("0")

    if not route:
        route = ProcessRoute(
            route_code=_build_route_code(db, product_item.id),
            route_name=f"{product_item.item_name} 默认工艺路线",
            product_item_id=product_item.id,
            version_no="V1.0",
            is_default=1,
            status="ACTIVE",
            effective_date=None,
            expire_date=None,
            remark="小程序产品清单同步生成",
        )
        db.add(route)
        db.flush()
    elif route.status != "ACTIVE":
        route.status = "ACTIVE"
        db.add(route)

    processes_by_name = {
        _normalize_text(process.process_name): process
        for process in db.scalars(select(Process)).all()
    }
    desired_ops = []
    for index, miniapp_row in enumerate(process_rows, start=1):
        process_name = _export_miniapp_process_name(miniapp_row.process_name)
        process = processes_by_name.get(process_name) or first_process
        desired_ops.append(
            {
                "seq_no": index,
                "process_id": process.id,
                "work_center_id": default_work_center.id,
                "operation_name": process_name,
                "std_cycle_seconds": _decimal_or_zero(miniapp_row.standard_beat, scale=Decimal("0.01")),
                "allowed_scrap_rate": _decimal_or_zero(allowed_scrap_rate, scale=Decimal("0.0001")),
            }
        )

    existing_ops = db.scalars(
        select(ProcessRouteOperation)
        .where(ProcessRouteOperation.route_id == route.id)
        .order_by(ProcessRouteOperation.seq_no.asc(), ProcessRouteOperation.id.asc())
    ).all()
    op_changed = _sync_miniapp_route_operations(db, route, existing_ops, desired_ops)
    work_order_changed = _sync_work_order_operations_for_route(db, route.id, desired_ops) if _route_has_usage(db, route.id) else False
    return op_changed or work_order_changed


def _sync_miniapp_route_to_erp_fast(
    db: Session,
    product_item: Item,
    rows: list[MiniAppProduct],
    context: dict[str, object],
) -> bool:
    process_rows = [row for row in rows if _export_miniapp_process_name(row.process_name)]
    if not process_rows:
        return False

    default_work_center = context["default_work_center"]
    first_process = context["first_process"]
    if not default_work_center or not first_process:
        return False

    routes_by_product_id = context["routes_by_product_id"]
    route_ops_by_route_id = context["route_ops_by_route_id"]
    used_route_ids = context["used_route_ids"]
    processes_by_name = context["processes_by_name"]
    route = routes_by_product_id.get(product_item.id)
    changed = False
    if not route:
        route = ProcessRoute(
            route_code=_build_route_code(db, product_item.id),
            route_name=f"{product_item.item_name} 默认工艺路线",
            product_item_id=product_item.id,
            version_no="V1.0",
            is_default=1,
            status="ACTIVE",
            effective_date=None,
            expire_date=None,
            remark="小程序产品清单同步生成",
        )
        db.add(route)
        db.flush()
        routes_by_product_id[product_item.id] = route
        route_ops_by_route_id[route.id] = []
        changed = True

    allowed_scrap_rate = _first_decimal(rows, "scrap_loss_rate") or Decimal("0")
    existing_ops = route_ops_by_route_id.get(route.id, [])

    desired_ops = []
    for index, miniapp_row in enumerate(process_rows, start=1):
        process_name = _export_miniapp_process_name(miniapp_row.process_name)
        process = processes_by_name.get(process_name) or first_process
        desired_ops.append(
            {
                "seq_no": index,
                "process_id": process.id,
                "work_center_id": default_work_center.id,
                "operation_name": process_name,
                "std_cycle_seconds": _decimal_or_zero(miniapp_row.standard_beat, scale=Decimal("0.01")),
                "allowed_scrap_rate": _decimal_or_zero(allowed_scrap_rate, scale=Decimal("0.0001")),
            }
        )

    if route.status != "ACTIVE":
        route.status = "ACTIVE"
        changed = True
    if route.remark != MINIAPP_SYNC_REMARK:
        route.remark = MINIAPP_SYNC_REMARK
        changed = True

    op_changed = _sync_miniapp_route_operations(db, route, existing_ops, desired_ops)
    route_ops_by_route_id[route.id] = sorted(
        route_ops_by_route_id.get(route.id, existing_ops),
        key=lambda operation: (operation.seq_no, operation.id or 0),
    )
    work_order_changed = _sync_work_order_operations_for_route(db, route.id, desired_ops) if route.id in used_route_ids else False
    if changed:
        db.add(route)
    return changed or op_changed or work_order_changed


def _sync_miniapp_route_operations(
    db: Session,
    route: ProcessRoute,
    existing_ops: list[ProcessRouteOperation],
    desired_ops: list[dict[str, object]],
) -> bool:
    changed = False
    used_operation_ids: set[int] = set()
    existing_by_name: dict[str, ProcessRouteOperation] = {}
    for operation in existing_ops:
        operation_name = _normalize_text(operation.operation_name)
        if operation_name and operation_name not in existing_by_name:
            existing_by_name[operation_name] = operation

    matched_existing = {
        existing_by_name[_normalize_text(desired["operation_name"])].id
        for desired in desired_ops
        if _normalize_text(desired["operation_name"]) in existing_by_name
        and existing_by_name[_normalize_text(desired["operation_name"])].id
    }
    desired_seq_by_existing_id = {
        existing_by_name[_normalize_text(desired["operation_name"])].id: int(desired["seq_no"])
        for desired in desired_ops
        if _normalize_text(desired["operation_name"]) in existing_by_name
        and existing_by_name[_normalize_text(desired["operation_name"])].id
    }
    occupied_by_other = {
        operation.seq_no: operation.id
        for operation in existing_ops
        if operation.id
    }
    needs_temporary_resequence = any(
        _normalize_text(desired["operation_name"]) not in existing_by_name
        or (
            existing_by_name[_normalize_text(desired["operation_name"])].id
            and occupied_by_other.get(int(desired["seq_no"])) not in {None, existing_by_name[_normalize_text(desired["operation_name"])].id}
        )
        for desired in desired_ops
    )
    needs_temporary_resequence = needs_temporary_resequence or any(
        operation.id not in matched_existing and operation.seq_no <= len(desired_ops)
        for operation in existing_ops
        if operation.id
    )
    needs_temporary_resequence = needs_temporary_resequence or any(
        operation_id and occupied_by_other.get(seq_no) not in {None, operation_id}
        for operation_id, seq_no in desired_seq_by_existing_id.items()
    )

    if needs_temporary_resequence and existing_ops:
        temporary_seq = max(
            [int(operation.seq_no or 0) for operation in existing_ops] + [len(desired_ops)]
        ) + 1000
        for index, operation in enumerate(existing_ops):
            operation.seq_no = temporary_seq + index
            db.add(operation)
        db.flush()
        changed = True

    for desired in desired_ops:
        operation_name = _normalize_text(desired["operation_name"])
        operation = existing_by_name.get(operation_name)
        if not operation:
            operation = ProcessRouteOperation(
                route_id=route.id,
                seq_no=int(desired["seq_no"]),
                process_id=int(desired["process_id"]),
                work_center_id=int(desired["work_center_id"]),
                operation_name=operation_name,
                std_setup_minutes=Decimal("0"),
                std_cycle_seconds=desired["std_cycle_seconds"],
                std_labor_cost=Decimal("0"),
                std_machine_cost=Decimal("0"),
                std_outsource_cost=Decimal("0"),
                allowed_scrap_rate=desired["allowed_scrap_rate"],
                report_required=1,
                is_key_operation=0,
                status="ACTIVE",
                remark=MINIAPP_SYNC_REMARK,
            )
            db.add(operation)
            db.flush()
            existing_ops.append(operation)
            changed = True
        if operation.id:
            used_operation_ids.add(operation.id)
        operation_changed = False
        if operation.seq_no != desired["seq_no"]:
            operation.seq_no = int(desired["seq_no"])
            operation_changed = True
        if operation.process_id != desired["process_id"]:
            operation.process_id = int(desired["process_id"])
            operation_changed = True
        if operation.work_center_id != desired["work_center_id"]:
            operation.work_center_id = int(desired["work_center_id"])
            operation_changed = True
        if _normalize_text(operation.operation_name) != operation_name:
            operation.operation_name = operation_name
            operation_changed = True
        if _decimal_changed(operation.std_cycle_seconds, desired["std_cycle_seconds"], scale=Decimal("0.01")):
            operation.std_cycle_seconds = desired["std_cycle_seconds"]
            operation_changed = True
        if _decimal_changed(operation.allowed_scrap_rate, desired["allowed_scrap_rate"], scale=Decimal("0.0001")):
            operation.allowed_scrap_rate = desired["allowed_scrap_rate"]
            operation_changed = True
        if operation.status != "ACTIVE":
            operation.status = "ACTIVE"
            operation_changed = True
        if operation.remark != MINIAPP_SYNC_REMARK:
            operation.remark = MINIAPP_SYNC_REMARK
            operation_changed = True
        if operation_changed:
            db.add(operation)
            changed = True

    inactive_seq = len(desired_ops) + 1
    for operation in existing_ops:
        if operation.id in used_operation_ids:
            continue
        if _normalize_text(operation.remark) != MINIAPP_SYNC_REMARK:
            if operation.status == "INACTIVE":
                operation.status = "ACTIVE"
                db.add(operation)
                changed = True
            continue
        operation_changed = False
        if operation.status != "INACTIVE":
            operation.status = "INACTIVE"
            operation_changed = True
        if operation.seq_no < inactive_seq:
            operation.seq_no = inactive_seq
            inactive_seq += 1
            operation_changed = True
        else:
            inactive_seq = max(inactive_seq, int(operation.seq_no or 0) + 1)
        if operation_changed:
            db.add(operation)
            changed = True

    return changed


def _sync_work_order_operations_for_route(db: Session, route_id: int, desired_ops: list[dict[str, object]]) -> bool:
    desired_by_seq = {int(item["seq_no"]): item for item in desired_ops}
    if not desired_by_seq:
        return False

    changed = False
    work_order_ops = db.scalars(
        select(WorkOrderOperation)
        .join(WorkOrder, WorkOrder.id == WorkOrderOperation.work_order_id)
        .where(
            WorkOrder.route_id == route_id,
            WorkOrderOperation.seq_no.in_(desired_by_seq.keys()),
        )
    ).all()
    for operation in work_order_ops:
        desired = desired_by_seq.get(int(operation.seq_no))
        if not desired:
            continue
        if _normalize_text(operation.operation_name) != desired["operation_name"]:
            operation.operation_name = str(desired["operation_name"])
            changed = True
        if operation.work_center_id != desired["work_center_id"]:
            operation.work_center_id = int(desired["work_center_id"])
            changed = True
        if _decimal_changed(operation.allowed_scrap_rate, desired["allowed_scrap_rate"], scale=Decimal("0.0001")):
            operation.allowed_scrap_rate = desired["allowed_scrap_rate"]
            changed = True
        if changed:
            db.add(operation)
    return changed


def _miniapp_route_operations_match(
    existing_ops: list[ProcessRouteOperation],
    desired_ops: list[dict[str, object]],
) -> bool:
    if len(existing_ops) != len(desired_ops):
        return False
    for existing, desired in zip(existing_ops, desired_ops):
        if existing.seq_no != desired["seq_no"]:
            return False
        if existing.process_id != desired["process_id"]:
            return False
        if existing.work_center_id != desired["work_center_id"]:
            return False
        if _normalize_text(existing.operation_name) != desired["operation_name"]:
            return False
        if _decimal_changed(existing.std_cycle_seconds, desired["std_cycle_seconds"], scale=Decimal("0.01")):
            return False
        if _decimal_changed(existing.allowed_scrap_rate, desired["allowed_scrap_rate"], scale=Decimal("0.0001")):
            return False
    return True


def _find_product_for_miniapp_product(db: Session, miniapp_product: MiniAppProduct) -> tuple[Product, Item] | None:
    product_name = _normalize_text(miniapp_product.product_name)
    project_no = _normalize_text(miniapp_product.project_no)
    conditions = []
    if product_name:
        conditions.append(Item.item_name == product_name)
    if project_no:
        conditions.append(Item.item_code == project_no)
    if not conditions:
        return None

    row = db.execute(
        select(Product, Item)
        .join(Item, Item.id == Product.item_id)
        .where(Item.item_type.in_(["PRODUCT", "FINISHED_GOOD"]), or_(*conditions))
        .order_by(Product.id.desc())
        .limit(1)
    ).first()
    if not row:
        return None
    return row[0], row[1]


def _calculate_loss_rate(net_weight: Decimal, gross_weight: Decimal) -> Decimal:
    if gross_weight <= 0:
        return Decimal("0")
    loss_weight = gross_weight - net_weight
    if loss_weight <= 0:
        return Decimal("0")
    return loss_weight / gross_weight


def _build_employee_reads(db: Session, employees: list[Employee]) -> list[EmployeeRead]:
    if not employees:
        return []

    dept_map = {
        row.id: row.dept_name
        for row in db.scalars(select(Department).where(Department.id.in_({employee.dept_id for employee in employees}))).all()
    }

    employee_ids = [employee.id for employee in employees]
    user_rows = db.execute(
        select(User.employee_id, User.username, Role.role_name)
        .select_from(User)
        .outerjoin(UserRole, UserRole.user_id == User.id)
        .outerjoin(Role, Role.id == UserRole.role_id)
        .where(User.employee_id.in_(employee_ids))
        .order_by(User.employee_id, Role.id)
    ).all()

    user_map: dict[int, str | None] = {}
    role_map: dict[int, list[str]] = {}
    for employee_id, username, role_name in user_rows:
        if employee_id is None:
            continue
        if employee_id not in user_map:
            user_map[employee_id] = username
        if role_name:
            role_map.setdefault(employee_id, [])
            if role_name not in role_map[employee_id]:
                role_map[employee_id].append(role_name)

    return [
        EmployeeRead(
            id=employee.id,
            employee_code=employee.employee_code,
            employee_name=employee.employee_name,
            dept_id=employee.dept_id,
            dept_name=dept_map.get(employee.dept_id),
            mobile=employee.mobile,
            gender=employee.gender,
            hire_date=employee.hire_date,
            job_title=employee.job_title,
            shift_code=employee.shift_code,
            is_operator=bool(employee.is_operator),
            is_workshop_staff=bool(employee.is_workshop_staff),
            status=employee.status,
            remark=employee.remark,
            username=user_map.get(employee.id),
            role_names=role_map.get(employee.id, []),
        )
        for employee in employees
    ]


def _get_product_read(db: Session, item_id: int) -> ProductRead:
    row = db.execute(
        select(
            Product.item_id.label("item_id"),
            Item.item_code.label("item_code"),
            Item.item_name.label("item_name"),
            Item.specification.label("specification"),
            Product.drawing_no.label("drawing_no"),
            Product.product_family.label("product_family"),
            Product.version_no.label("version_no"),
            Item.unit_weight_kg.label("unit_weight_kg"),
            Product.allowed_scrap_rate.label("allowed_scrap_rate"),
            Product.standard_output_qty.label("standard_output_qty"),
            Product.quality_standard.label("quality_standard"),
            Item.safety_stock_weight_kg.label("safety_stock_weight_kg"),
            Item.status.label("status"),
        )
        .join(Item, Item.id == Product.item_id)
        .where(Product.item_id == item_id)
    ).mappings().first()
    if not row:
        raise HTTPException(status_code=404, detail="产品不存在")
    return ProductRead.model_validate(dict(row))


def _get_material_read(db: Session, item_id: int) -> MaterialRead:
    row = db.execute(
        select(
            Material.item_id.label("item_id"),
            Item.item_code.label("item_code"),
            Item.item_name.label("item_name"),
            Item.specification.label("specification"),
            Item.material_grade.label("material_grade"),
            Item.unit_weight_kg.label("unit_weight_kg"),
            Item.safety_stock_weight_kg.label("safety_stock_weight_kg"),
            Item.scrap_sale_price.label("scrap_sale_price"),
            Material.default_supplier_id.label("default_supplier_id"),
            Supplier.supplier_name.label("supplier_name"),
            Material.purchase_calc_mode.label("purchase_calc_mode"),
            Material.min_purchase_qty.label("min_purchase_qty"),
            Material.purchase_multiple_qty.label("purchase_multiple_qty"),
            Material.purchase_multiple_weight_kg.label("purchase_multiple_weight_kg"),
            Material.lead_time_days.label("lead_time_days"),
            Material.quality_rule.label("quality_rule"),
            Material.quality_rule.label("remark"),
            Item.status.label("status"),
        )
        .join(Item, Item.id == Material.item_id)
        .outerjoin(Supplier, Supplier.id == Material.default_supplier_id)
        .where(Material.item_id == item_id)
    ).mappings().first()
    if not row:
        raise HTTPException(status_code=404, detail="原材料不存在")
    return MaterialRead.model_validate(_normalize_material_read_payload(dict(row)))


def _ensure_jiaheng_warehouses(db: Session) -> None:
    existing_by_type = {
        str(row.warehouse_type or "").upper(): row
        for row in db.scalars(select(Warehouse).where(Warehouse.warehouse_type.in_(["RAW", "SEMI", "FINISHED", "AUX", "SCRAP", "RETURN"]))).all()
    }
    existing_by_code = {
        str(row.warehouse_code or "").upper(): row
        for row in db.scalars(select(Warehouse).where(Warehouse.warehouse_code.in_([item["warehouse_code"] for item in JIAHENG_WAREHOUSE_DEFINITIONS]))).all()
    }
    changed = False
    now = datetime.now()

    for definition in JIAHENG_WAREHOUSE_DEFINITIONS:
        warehouse_type = definition["warehouse_type"]
        warehouse = existing_by_code.get(definition["warehouse_code"]) or existing_by_type.get(warehouse_type)
        if not warehouse:
            warehouse = Warehouse(
                warehouse_code=definition["warehouse_code"],
                warehouse_name=definition["warehouse_name"],
                warehouse_type=warehouse_type,
                manager_employee_id=None,
                status="ACTIVE",
                remark=definition["remark"],
                created_at=now,
                updated_at=now,
            )
            db.add(warehouse)
            db.flush()
            changed = True
        else:
            warehouse_changed = False
            desired_values = {
                "warehouse_code": definition["warehouse_code"],
                "warehouse_name": definition["warehouse_name"],
                "warehouse_type": warehouse_type,
                "remark": definition["remark"],
                "status": "ACTIVE",
            }
            for field, value in desired_values.items():
                if getattr(warehouse, field) != value:
                    setattr(warehouse, field, value)
                    warehouse_changed = True
            if warehouse_changed:
                warehouse.updated_at = now
                db.add(warehouse)
                changed = True

        default_location = db.scalar(
            select(WarehouseLocation).where(
                WarehouseLocation.warehouse_id == warehouse.id,
                WarehouseLocation.location_code == definition["location_code"],
            )
        )
        if not default_location:
            db.add(
                WarehouseLocation(
                    warehouse_id=warehouse.id,
                    location_code=definition["location_code"],
                    location_name=definition["location_name"],
                    zone_name=definition["zone_name"],
                    is_default=1,
                    is_locked=0,
                    status="ACTIVE",
                    remark=definition["remark"],
                    created_at=now,
                    updated_at=now,
                )
            )
            changed = True

    if changed:
        db.commit()


def _normalize_material_read_payload(payload: dict) -> dict:
    normalized = dict(payload)
    normalized["purchase_calc_mode"] = "BY_WEIGHT"
    normalized["purchase_multiple_qty"] = 0
    normalized["scrap_sale_price"] = normalized.get("scrap_sale_price") or 0
    normalized["remark"] = normalized.get("remark") or normalized.get("quality_rule")
    normalized["quality_rule"] = normalized.get("quality_rule") or normalized.get("remark")
    return normalized


def _material_remark_value(payload: MaterialCreate | MaterialUpdate) -> str | None:
    return payload.remark if payload.remark is not None else payload.quality_rule


def _normalize_text(value: str | None) -> str:
    return str(value or "").strip()


_CHINESE_CODE_INITIALS = {
    "产": "C",
    "品": "P",
    "原": "Y",
    "料": "L",
    "本": "B",
    "体": "T",
    "垃": "L",
    "圾": "J",
    "桶": "T",
    "垫": "D",
    "电": "D",
    "脑": "N",
    "支": "Z",
    "架": "J",
    "钢": "G",
    "板": "B",
    "卷": "J",
    "铝": "L",
    "合": "H",
    "金": "J",
    "铁": "T",
    "铜": "T",
    "不": "B",
    "锈": "X",
    "冷": "L",
    "热": "R",
    "轧": "Z",
    "镀": "D",
    "锌": "X",
    "冲": "C",
    "压": "Y",
    "件": "J",
    "片": "P",
    "管": "G",
    "棒": "B",
    "圆": "Y",
    "方": "F",
    "型": "X",
    "材": "C",
    "外": "W",
    "壳": "K",
    "底": "D",
    "座": "Z",
    "盖": "G",
    "盒": "H",
    "箱": "X",
    "门": "M",
    "锁": "S",
    "扣": "K",
    "连": "L",
    "接": "J",
}


def _normalize_code_token(value: str | None, *, max_length: int = 24) -> str:
    text = _normalize_text(value).upper()
    if not text:
        return ""
    chunks: list[str] = []
    current = ""
    current_kind = ""

    def flush_current() -> None:
        nonlocal current, current_kind
        if current:
            chunks.append(current)
        current = ""
        current_kind = ""

    for char in text.replace("×", "X").replace("*", "X"):
        if char.isascii() and (char.isalnum() or char == "."):
            kind = "LATIN"
            token_char = char
        elif "\u4e00" <= char <= "\u9fff":
            mapped = _CHINESE_CODE_INITIALS.get(char)
            if mapped:
                kind = "CJK"
                token_char = mapped
            else:
                kind = "CJK_RAW"
                token_char = char
        else:
            flush_current()
            continue

        if current_kind and current_kind != kind:
            flush_current()
        current_kind = kind
        current += token_char

    flush_current()
    return "-".join(chunks)[:max_length].strip("-.")


def _build_code_prefix(*parts: str, max_length: int = 42) -> str:
    tokens = []
    for part in parts:
        token = _normalize_code_token(part)
        if token:
            tokens.append(token)
    if not tokens:
        return ""
    prefix = "-".join(tokens)
    return prefix[:max_length].strip("-.")


def _next_code_sequence(db: Session, model, column, prefix: str) -> int:
    if not prefix:
        return 1
    rows = db.scalars(select(column).where(column.like(f"{prefix}-%"))).all()
    max_no = 0
    for code in rows:
        code_text = _normalize_text(code)
        if not code_text.startswith(f"{prefix}-"):
            continue
        tail = code_text.rsplit("-", 1)[-1]
        if tail.isdigit():
            max_no = max(max_no, int(tail))
    return max_no + 1


def _build_sequenced_code(db: Session, model, column, prefix: str) -> str:
    safe_prefix = prefix[:42].strip("-.") or "AUTO"
    next_no = _next_code_sequence(db, model, column, safe_prefix)
    return f"{safe_prefix}-{next_no:03d}"


def _build_product_code(db: Session, payload: ProductCreate | ProductUpdate | None = None) -> str:
    prefix = "产品需规"
    pattern = re.compile(rf"{re.escape(prefix)}(\d{{5}})")
    max_no = 0
    rows = db.scalars(select(Item.item_code).where(Item.item_type.in_(["PRODUCT", "FINISHED_GOOD"]))).all()
    for code in rows:
        match = pattern.fullmatch(_normalize_text(code))
        if match:
            max_no = max(max_no, int(match.group(1)))

    next_no = max_no + 1
    while True:
        candidate = f"{prefix}{next_no:05d}"
        if not db.scalar(select(Item.id).where(Item.item_code == candidate).limit(1)):
            return candidate
        next_no += 1


def _build_material_code(db: Session, payload: MaterialCreate | MaterialUpdate | None = None) -> str:
    prefix = "原材料"
    pattern = re.compile(rf"{re.escape(prefix)}(\d{{5}})")
    max_no = 0
    rows = db.scalars(select(Item.item_code).where(Item.item_type == "RAW_MATERIAL")).all()
    for code in rows:
        match = pattern.fullmatch(_normalize_text(code))
        if match:
            max_no = max(max_no, int(match.group(1)))

    next_no = max_no + 1
    while True:
        candidate = f"{prefix}{next_no:05d}"
        if not db.scalar(select(Item.id).where(Item.item_code == candidate).limit(1)):
            return candidate
        next_no += 1


def _build_bom_code(db: Session, product_item_id: int) -> str:
    product_item = db.get(Item, product_item_id)
    if not product_item:
        raise HTTPException(status_code=404, detail="产品不存在")
    existing_code = db.scalar(
        select(Bom.bom_code)
        .where(Bom.product_item_id == product_item_id)
        .order_by(Bom.is_default.desc(), Bom.id.desc())
        .limit(1)
    )
    if existing_code:
        return _normalize_text(existing_code)
    product_code = _normalize_code_token(product_item.item_code, max_length=45)
    return f"BOM-{product_code}"[:50].strip("-.")


def _build_route_code(db: Session, product_item_id: int) -> str:
    product_item = db.get(Item, product_item_id)
    if not product_item:
        raise HTTPException(status_code=404, detail="产品不存在")
    existing_code = db.scalar(
        select(ProcessRoute.route_code)
        .where(ProcessRoute.product_item_id == product_item_id)
        .order_by(ProcessRoute.is_default.desc(), ProcessRoute.id.desc())
        .limit(1)
    )
    if existing_code:
        return _normalize_text(existing_code)
    product_code = _normalize_code_token(product_item.item_code, max_length=42)
    return f"ROUTE-{product_code}"[:50].strip("-.")


def _bom_code_version_exists(
    db: Session,
    bom_code: str,
    version_no: str,
    *,
    exclude_bom_id: int | None = None,
) -> bool:
    stmt = select(Bom.id).where(Bom.bom_code == _normalize_text(bom_code), Bom.version_no == _normalize_text(version_no))
    if exclude_bom_id is not None:
        stmt = stmt.where(Bom.id != exclude_bom_id)
    return db.scalar(stmt) is not None


def _route_code_version_exists(
    db: Session,
    route_code: str,
    version_no: str,
    *,
    exclude_route_id: int | None = None,
) -> bool:
    stmt = select(ProcessRoute.id).where(
        ProcessRoute.route_code == _normalize_text(route_code),
        ProcessRoute.version_no == _normalize_text(version_no),
    )
    if exclude_route_id is not None:
        stmt = stmt.where(ProcessRoute.id != exclude_route_id)
    return db.scalar(stmt) is not None


def _product_code_version_exists(
    db: Session,
    item_code: str,
    version_no: str,
    *,
    exclude_item_id: int | None = None,
) -> bool:
    stmt = (
        select(Product.item_id)
        .join(Item, Item.id == Product.item_id)
        .where(Item.item_code == _normalize_text(item_code), Product.version_no == _normalize_text(version_no))
    )
    if exclude_item_id is not None:
        stmt = stmt.where(Product.item_id != exclude_item_id)
    return db.scalar(stmt) is not None


def _item_code_used_by_non_product(db: Session, item_code: str, *, exclude_item_id: int | None = None) -> bool:
    stmt = select(Item.id).where(
        Item.item_code == _normalize_text(item_code),
        Item.item_type.notin_(("PRODUCT", "FINISHED_GOOD")),
    )
    if exclude_item_id is not None:
        stmt = stmt.where(Item.id != exclude_item_id)
    return db.scalar(stmt) is not None


def _product_has_usage(db: Session, item_id: int) -> bool:
    references = [
        db.scalar(select(func.count(SalesOrderItem.id)).where(SalesOrderItem.product_item_id == item_id)) or 0,
        db.scalar(select(func.count(MaterialDemand.id)).where(MaterialDemand.product_item_id == item_id)) or 0,
        db.scalar(select(func.count(WorkOrder.id)).where(WorkOrder.product_item_id == item_id)) or 0,
        db.scalar(select(func.count(CompletionReceiptItem.id)).where(CompletionReceiptItem.product_item_id == item_id)) or 0,
        db.scalar(select(func.count(DeliveryItem.id)).where(DeliveryItem.product_item_id == item_id)) or 0,
        db.scalar(select(func.count(ReturnItem.id)).where(ReturnItem.product_item_id == item_id)) or 0,
    ]
    return any(count > 0 for count in references)


def _create_bom_record(db: Session, payload: BomUpsert) -> Bom:
    bom = Bom(
        bom_code=_normalize_text(payload.bom_code),
        product_item_id=payload.product_item_id,
        version_no=_normalize_text(payload.version_no),
        is_default=int(payload.is_default),
        status=payload.status,
        effective_date=payload.effective_date,
        expire_date=payload.expire_date,
        remark=payload.remark,
    )
    db.add(bom)
    db.flush()

    for item in payload.items:
        db.add(
            BomItem(
                bom_id=bom.id,
                seq_no=item.seq_no,
                material_item_id=item.material_item_id,
                usage_type=_normalize_text(item.usage_type).upper(),
                qty_per_piece=Decimal(str(item.qty_per_piece)),
                net_weight_per_piece_kg=Decimal(str(item.net_weight_per_piece_kg)),
                gross_weight_per_piece_kg=Decimal(str(item.gross_weight_per_piece_kg)),
                loss_rate=Decimal(str(item.loss_rate)),
                issue_over_tolerance_rate=Decimal(str(item.issue_over_tolerance_rate)),
                remark=item.remark,
            )
        )

    db.flush()
    return bom


def _create_process_route_record(db: Session, payload: ProcessRouteUpsert) -> ProcessRoute:
    route = ProcessRoute(
        route_code=_normalize_text(payload.route_code),
        route_name=_normalize_text(payload.route_name),
        product_item_id=payload.product_item_id,
        version_no=_normalize_text(payload.version_no),
        is_default=int(payload.is_default),
        status=payload.status,
        effective_date=payload.effective_date,
        expire_date=payload.expire_date,
        remark=payload.remark,
    )
    db.add(route)
    db.flush()

    for operation in payload.operations:
        db.add(
            ProcessRouteOperation(
                route_id=route.id,
                seq_no=operation.seq_no,
                process_id=operation.process_id,
                work_center_id=operation.work_center_id,
                operation_name=_normalize_text(operation.operation_name),
                std_setup_minutes=Decimal(str(operation.std_setup_minutes)),
                std_cycle_seconds=Decimal(str(operation.std_cycle_seconds or 0)),
                std_labor_cost=Decimal("0"),
                std_machine_cost=Decimal("0"),
                std_outsource_cost=Decimal("0"),
                allowed_scrap_rate=Decimal(str(operation.allowed_scrap_rate)),
                report_required=int(operation.report_required),
                is_key_operation=int(operation.is_key_operation),
                status=operation.status,
                remark=operation.remark,
            )
        )

    db.flush()
    return route


def _create_product_record(db: Session, payload: ProductCreate | ProductUpdate) -> int:
    item = Item(
        item_code=_normalize_text(payload.item_code),
        item_name=_normalize_text(payload.item_name),
        item_type="FINISHED_GOOD",
        specification=payload.specification,
        material_grade=None,
        unit_weight_kg=Decimal(str(payload.unit_weight_kg)),
        safety_stock_weight_kg=Decimal(str(payload.safety_stock_weight_kg)),
        status=payload.status,
    )
    db.add(item)
    db.flush()

    db.add(
        Product(
            item_id=item.id,
            drawing_no=payload.drawing_no,
            product_family=payload.product_family,
            version_no=_normalize_text(payload.version_no),
            standard_output_qty=Decimal(str(payload.standard_output_qty)),
            allowed_scrap_rate=Decimal(str(payload.allowed_scrap_rate)),
            quality_standard=payload.quality_standard,
        )
    )
    db.flush()
    return item.id


def _commit_versioned_master_changes(db: Session) -> None:
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        _raise_versioned_master_integrity_error(exc)


def _create_versioned_master_record(db: Session, creator):
    try:
        return creator()
    except IntegrityError as exc:
        db.rollback()
        _raise_versioned_master_integrity_error(exc)


def _raise_versioned_master_integrity_error(exc: IntegrityError) -> None:
    error_text = str(exc.orig or exc)
    if any(token in error_text for token in ("item_code", "bom_code", "route_code", "Duplicate entry")):
        raise HTTPException(
            status_code=400,
            detail="数据库版本化约束尚未更新，请先执行 backend/sql/versioned_master_constraints_patch.sql",
        ) from exc
    raise exc


def _validate_material_payload(payload: MaterialCreate | MaterialUpdate) -> None:
    _ensure_required_text(payload.item_name, "原材料名称")
    _ensure_required_text(payload.specification, "原材料规格")
    _ensure_required_text(payload.material_grade, "材质牌号")

    if payload.default_supplier_id is not None and payload.default_supplier_id <= 0:
        raise HTTPException(status_code=400, detail="默认供应商ID不合法")
    if payload.unit_weight_kg <= 0:
        raise HTTPException(status_code=400, detail="原材料重量基准必须大于0")
    if payload.safety_stock_weight_kg < 0:
        raise HTTPException(status_code=400, detail="安全库存重量不能小于0")
    if payload.scrap_sale_price < 0:
        raise HTTPException(status_code=400, detail="废料单价不能小于0")
    if payload.min_purchase_qty < 0:
        raise HTTPException(status_code=400, detail="最小采购重量不能小于0")
    if payload.purchase_multiple_qty < 0:
        raise HTTPException(status_code=400, detail="采购倍数不能小于0")
    if payload.purchase_multiple_weight_kg < 0:
        raise HTTPException(status_code=400, detail="采购倍数重量不能小于0")
    if payload.lead_time_days < 0:
        raise HTTPException(status_code=400, detail="采购提前期不能小于0")


def _validate_product_payload(payload: ProductCreate | ProductUpdate) -> None:
    _ensure_required_text(payload.item_name, "产品名称")
    _ensure_required_text(payload.version_no, "版本号")

    if payload.unit_weight_kg <= 0:
        raise HTTPException(status_code=400, detail="产品单重必须大于0")
    if payload.allowed_scrap_rate < 0:
        raise HTTPException(status_code=400, detail="允许报废率不能小于0")
    if payload.standard_output_qty < 0:
        raise HTTPException(status_code=400, detail="标准产出数量不能小于0")
    if payload.safety_stock_weight_kg < 0:
        raise HTTPException(status_code=400, detail="安全库存重量不能小于0")


def _validate_bom_payload(db: Session, payload: BomUpsert) -> None:
    _ensure_required_text(payload.version_no, "BOM版本号")

    if payload.effective_date and payload.expire_date and payload.effective_date > payload.expire_date:
        raise HTTPException(status_code=400, detail="BOM失效日期不能早于生效日期")

    product = db.scalar(select(Product).where(Product.item_id == payload.product_item_id))
    if not product:
        raise HTTPException(status_code=404, detail="产品不存在")
    if not payload.items:
        raise HTTPException(status_code=400, detail="BOM至少需要一条明细")

    seq_set: set[int] = set()
    for item in payload.items:
        if item.seq_no <= 0:
            raise HTTPException(status_code=400, detail="BOM明细序号必须大于0")
        if item.seq_no in seq_set:
            raise HTTPException(status_code=400, detail="BOM明细序号不能重复")
        seq_set.add(item.seq_no)

        usage_type = (item.usage_type or "").strip().upper()
        if usage_type != "WEIGHT":
            raise HTTPException(status_code=400, detail="BOM明细原材料用量只支持按重量维护")
        if item.qty_per_piece < 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的单件用量不能小于0")
        if item.net_weight_per_piece_kg < 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的单件净重不能小于0")
        if item.gross_weight_per_piece_kg < 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的单件毛重不能小于0")
        if item.loss_rate < 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的损耗率不能小于0")
        if item.issue_over_tolerance_rate < 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的超发容差率不能小于0")
        if usage_type == "WEIGHT" and item.gross_weight_per_piece_kg <= 0:
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的单件毛重必须大于0")
        if (
            item.net_weight_per_piece_kg > 0
            and item.gross_weight_per_piece_kg > 0
            and item.gross_weight_per_piece_kg < item.net_weight_per_piece_kg
        ):
            raise HTTPException(status_code=400, detail=f"BOM序号 {item.seq_no} 的单件毛重不能小于单件净重")

        material = db.scalar(select(Material).where(Material.item_id == item.material_item_id))
        if not material:
            raise HTTPException(status_code=404, detail=f"原材料ID {item.material_item_id} 不存在")


def _clear_default_bom(db: Session, product_item_id: int, exclude_bom_id: int | None = None) -> None:
    existing_rows = db.scalars(select(Bom).where(Bom.product_item_id == product_item_id)).all()
    for row in existing_rows:
        if exclude_bom_id and row.id == exclude_bom_id:
            continue
        row.is_default = 0
        db.add(row)


def _has_active_default_bom(db: Session, product_item_id: int, exclude_bom_id: int | None = None) -> bool:
    stmt = select(func.count(Bom.id)).where(
        Bom.product_item_id == product_item_id,
        Bom.status == "ACTIVE",
        Bom.is_default == 1,
    )
    if exclude_bom_id is not None:
        stmt = stmt.where(Bom.id != exclude_bom_id)
    return bool(db.scalar(stmt) or 0)


def _bom_has_usage(db: Session, bom_id: int) -> bool:
    bom_item_ids = db.scalars(select(BomItem.id).where(BomItem.bom_id == bom_id)).all()
    mrp_count = 0
    material_count = 0
    if bom_item_ids:
        mrp_count = db.scalar(select(func.count(MaterialDemand.id)).where(MaterialDemand.bom_item_id.in_(bom_item_ids))) or 0
        material_count = (
            db.scalar(select(func.count(WorkOrderMaterial.id)).where(WorkOrderMaterial.bom_item_id.in_(bom_item_ids))) or 0
        )
    work_order_count = db.scalar(select(func.count(WorkOrder.id)).where(WorkOrder.bom_id == bom_id)) or 0
    return bool(mrp_count or work_order_count or material_count)


def _list_bom_reads(db: Session, limit: int) -> list[BomRead]:
    rows = db.execute(
        select(
            Bom.id.label("bom_id"),
            Bom.bom_code.label("bom_code"),
            Bom.product_item_id.label("product_item_id"),
            Item.item_code.label("product_code"),
            Item.item_name.label("product_name"),
            Bom.version_no.label("version_no"),
            Bom.is_default.label("is_default"),
            Bom.status.label("status"),
            Bom.effective_date.label("effective_date"),
            Bom.expire_date.label("expire_date"),
            Bom.remark.label("remark"),
        )
        .join(Item, Item.id == Bom.product_item_id)
        .order_by(Bom.is_default.desc(), Bom.id.desc())
        .limit(limit)
    ).mappings().all()
    if not rows:
        return []

    bom_ids = [row["bom_id"] for row in rows]
    item_rows = db.execute(
        select(
            BomItem.id.label("bom_item_id"),
            BomItem.bom_id.label("bom_id"),
            BomItem.seq_no.label("seq_no"),
            BomItem.material_item_id.label("material_item_id"),
            Item.item_code.label("material_code"),
            Item.item_name.label("material_name"),
            BomItem.usage_type.label("usage_type"),
            BomItem.qty_per_piece.label("qty_per_piece"),
            BomItem.net_weight_per_piece_kg.label("net_weight_per_piece_kg"),
            BomItem.gross_weight_per_piece_kg.label("gross_weight_per_piece_kg"),
            BomItem.loss_rate.label("loss_rate"),
            BomItem.issue_over_tolerance_rate.label("issue_over_tolerance_rate"),
            BomItem.remark.label("remark"),
        )
        .join(Item, Item.id == BomItem.material_item_id)
        .where(BomItem.bom_id.in_(bom_ids))
        .order_by(BomItem.bom_id, BomItem.seq_no)
    ).mappings().all()

    item_map: dict[int, list[dict]] = {}
    for row in item_rows:
        item_map.setdefault(row["bom_id"], []).append(dict(row))

    return [
        BomRead(
            bom_id=row["bom_id"],
            bom_code=row["bom_code"],
            product_item_id=row["product_item_id"],
            product_code=row["product_code"],
            product_name=row["product_name"],
            version_no=row["version_no"],
            is_default=bool(row["is_default"]),
            status=row["status"],
            effective_date=row["effective_date"],
            expire_date=row["expire_date"],
            remark=row["remark"],
            item_count=len(item_map.get(row["bom_id"], [])),
            items=item_map.get(row["bom_id"], []),
        )
        for row in rows
    ]


def _get_bom_read(db: Session, bom_id: int) -> BomRead:
    rows = _list_bom_reads(db, limit=500)
    for row in rows:
        if row.bom_id == bom_id:
            return row
    raise HTTPException(status_code=404, detail="BOM不存在")


def _ensure_active_master_record(status: str | None, message: str) -> None:
    if _normalize_text(status).upper() == "INACTIVE":
        raise HTTPException(status_code=400, detail=message)


def _validate_process_route_payload(db: Session, payload: ProcessRouteUpsert) -> None:
    _ensure_required_text(payload.route_name, "工艺路线名称")
    _ensure_required_text(payload.version_no, "工艺路线版本号")

    if payload.effective_date and payload.expire_date and payload.effective_date > payload.expire_date:
        raise HTTPException(status_code=400, detail="工艺路线失效日期不能早于生效日期")

    product = db.scalar(select(Product).where(Product.item_id == payload.product_item_id))
    if not product:
        raise HTTPException(status_code=404, detail="产品不存在")
    if not payload.operations:
        raise HTTPException(status_code=400, detail="工艺路线至少需要一条工序")

    seq_set: set[int] = set()
    for operation in payload.operations:
        if operation.seq_no <= 0:
            raise HTTPException(status_code=400, detail="工序序号必须大于0")
        if operation.seq_no in seq_set:
            raise HTTPException(status_code=400, detail="工序序号不能重复")
        seq_set.add(operation.seq_no)

        _ensure_required_text(operation.operation_name, f"工序序号 {operation.seq_no} 的工序名称")
        if operation.std_setup_minutes < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的准备时长不能小于0")
        if operation.std_cycle_seconds < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的参考节拍不能小于0")
        if operation.std_labor_cost < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的人工成本不能小于0")
        if operation.std_machine_cost < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的设备成本不能小于0")
        if operation.std_outsource_cost < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的委外成本不能小于0")
        if operation.allowed_scrap_rate < 0:
            raise HTTPException(status_code=400, detail=f"工序序号 {operation.seq_no} 的允许报废率不能小于0")

        process = db.get(Process, operation.process_id)
        if not process:
            raise HTTPException(status_code=404, detail=f"工序ID {operation.process_id} 不存在")

        work_center = db.get(WorkCenter, operation.work_center_id)
        if not work_center:
            raise HTTPException(status_code=404, detail=f"工作中心ID {operation.work_center_id} 不存在")


def _ensure_route_payload_does_not_touch_inactive_operations(
    db: Session,
    route_id: int,
    payload: ProcessRouteUpsert,
) -> None:
    inactive_operations = db.scalars(
        select(ProcessRouteOperation).where(
            ProcessRouteOperation.route_id == route_id,
            ProcessRouteOperation.status == "INACTIVE",
        )
    ).all()
    if not inactive_operations:
        return

    inactive_names = {
        _normalize_text(operation.operation_name)
        for operation in inactive_operations
        if _normalize_text(operation.operation_name)
    }
    inactive_seq_nos = {int(operation.seq_no) for operation in inactive_operations if operation.seq_no}
    payload_names = {
        _normalize_text(operation.operation_name)
        for operation in payload.operations
        if _normalize_text(operation.operation_name)
    }
    payload_seq_nos = {int(operation.seq_no) for operation in payload.operations if operation.seq_no}

    touched_names = inactive_names & payload_names
    if touched_names:
        raise HTTPException(
            status_code=400,
            detail=f"已停用工序 {', '.join(sorted(touched_names))} 只读，不允许修改；如需重新启用，请先在小程序产品清单重新添加该工序",
        )

    touched_seq_nos = inactive_seq_nos & payload_seq_nos
    if touched_seq_nos:
        seq_labels = ", ".join(str(seq_no) for seq_no in sorted(touched_seq_nos))
        raise HTTPException(
            status_code=400,
            detail=f"工序序号 {seq_labels} 已被停用历史工序占用，不允许覆盖修改",
        )


def _clear_default_route(db: Session, product_item_id: int, exclude_route_id: int | None = None) -> None:
    existing_rows = db.scalars(select(ProcessRoute).where(ProcessRoute.product_item_id == product_item_id)).all()
    for row in existing_rows:
        if exclude_route_id and row.id == exclude_route_id:
            continue
        row.is_default = 0
        db.add(row)


def _has_active_default_route(db: Session, product_item_id: int, exclude_route_id: int | None = None) -> bool:
    stmt = select(func.count(ProcessRoute.id)).where(
        ProcessRoute.product_item_id == product_item_id,
        ProcessRoute.status == "ACTIVE",
        ProcessRoute.is_default == 1,
    )
    if exclude_route_id is not None:
        stmt = stmt.where(ProcessRoute.id != exclude_route_id)
    return bool(db.scalar(stmt) or 0)


def _route_has_usage(db: Session, route_id: int) -> bool:
    work_order_count = db.scalar(select(func.count(WorkOrder.id)).where(WorkOrder.route_id == route_id)) or 0
    return bool(work_order_count)


def _ensure_required_text(value: str | None, field_name: str) -> None:
    if value is None or not str(value).strip():
        raise HTTPException(status_code=400, detail=f"{field_name}不能为空")


def _raise_delete_blocked(entity_name: str, references: list[tuple[str, int]]) -> None:
    used = [f"{label}({count})" for label, count in references if count > 0]
    if used:
        raise HTTPException(
            status_code=400,
            detail=f"{entity_name}已被业务数据引用，不能删除：{'、'.join(used[:6])}",
        )


def _ensure_material_can_delete(db: Session, item_id: int) -> None:
    references = [
        ("BOM明细", db.scalar(select(func.count(BomItem.id)).where(BomItem.material_item_id == item_id)) or 0),
        ("MRP需求", db.scalar(select(func.count(MaterialDemand.id)).where(MaterialDemand.material_item_id == item_id)) or 0),
        ("采购订单明细", db.scalar(select(func.count(PurchaseOrderItem.id)).where(PurchaseOrderItem.material_item_id == item_id)) or 0),
        ("到货入库明细", db.scalar(select(func.count(PurchaseReceiptItem.id)).where(PurchaseReceiptItem.material_item_id == item_id)) or 0),
        ("工单领料", db.scalar(select(func.count(WorkOrderMaterial.id)).where(WorkOrderMaterial.material_item_id == item_id)) or 0),
        ("库存批次", db.scalar(select(func.count(StockLot.id)).where(StockLot.item_id == item_id)) or 0),
        ("库存台账", db.scalar(select(func.count(StockBalance.id)).where(StockBalance.item_id == item_id)) or 0),
        ("库存流水", db.scalar(select(func.count(InventoryTxn.id)).where(InventoryTxn.item_id == item_id)) or 0),
        ("报废记录", db.scalar(select(func.count(ScrapRecord.id)).where(ScrapRecord.item_id == item_id)) or 0),
    ]
    _raise_delete_blocked("该原材料", references)


def _ensure_product_can_delete(db: Session, item_id: int) -> None:
    references = [
        ("BOM版本", db.scalar(select(func.count(Bom.id)).where(Bom.product_item_id == item_id)) or 0),
        ("工艺路线", db.scalar(select(func.count(ProcessRoute.id)).where(ProcessRoute.product_item_id == item_id)) or 0),
        ("订单明细", db.scalar(select(func.count(SalesOrderItem.id)).where(SalesOrderItem.product_item_id == item_id)) or 0),
        ("MRP需求", db.scalar(select(func.count(MaterialDemand.id)).where(MaterialDemand.product_item_id == item_id)) or 0),
        ("生产工单", db.scalar(select(func.count(WorkOrder.id)).where(WorkOrder.product_item_id == item_id)) or 0),
        ("成品入库明细", db.scalar(select(func.count(CompletionReceiptItem.id)).where(CompletionReceiptItem.product_item_id == item_id)) or 0),
        ("发货明细", db.scalar(select(func.count(DeliveryItem.id)).where(DeliveryItem.product_item_id == item_id)) or 0),
        ("退货明细", db.scalar(select(func.count(ReturnItem.id)).where(ReturnItem.product_item_id == item_id)) or 0),
        ("库存批次", db.scalar(select(func.count(StockLot.id)).where(StockLot.item_id == item_id)) or 0),
        ("库存台账", db.scalar(select(func.count(StockBalance.id)).where(StockBalance.item_id == item_id)) or 0),
        ("库存流水", db.scalar(select(func.count(InventoryTxn.id)).where(InventoryTxn.item_id == item_id)) or 0),
        ("报废记录", db.scalar(select(func.count(ScrapRecord.id)).where(ScrapRecord.item_id == item_id)) or 0),
        ("成本分摊", db.scalar(select(func.count(CostAllocation.id)).where(CostAllocation.product_item_id == item_id)) or 0),
    ]
    _raise_delete_blocked("该产品", references)


def _ensure_product_spec_can_delete(db: Session, item_id: int) -> None:
    bom_ids = db.scalars(select(Bom.id).where(Bom.product_item_id == item_id)).all()
    route_ids = db.scalars(select(ProcessRoute.id).where(ProcessRoute.product_item_id == item_id)).all()
    references = [
        ("已引用BOM版本", sum(1 for bom_id in bom_ids if _bom_has_usage(db, bom_id))),
        ("已引用工艺路线", sum(1 for route_id in route_ids if _route_has_usage(db, route_id))),
        ("订单明细", db.scalar(select(func.count(SalesOrderItem.id)).where(SalesOrderItem.product_item_id == item_id)) or 0),
        ("MRP需求", db.scalar(select(func.count(MaterialDemand.id)).where(MaterialDemand.product_item_id == item_id)) or 0),
        ("生产工单", db.scalar(select(func.count(WorkOrder.id)).where(WorkOrder.product_item_id == item_id)) or 0),
        ("成品入库明细", db.scalar(select(func.count(CompletionReceiptItem.id)).where(CompletionReceiptItem.product_item_id == item_id)) or 0),
        ("发货明细", db.scalar(select(func.count(DeliveryItem.id)).where(DeliveryItem.product_item_id == item_id)) or 0),
        ("退货明细", db.scalar(select(func.count(ReturnItem.id)).where(ReturnItem.product_item_id == item_id)) or 0),
        ("库存批次", db.scalar(select(func.count(StockLot.id)).where(StockLot.item_id == item_id)) or 0),
        ("库存台账", db.scalar(select(func.count(StockBalance.id)).where(StockBalance.item_id == item_id)) or 0),
        ("库存流水", db.scalar(select(func.count(InventoryTxn.id)).where(InventoryTxn.item_id == item_id)) or 0),
        ("报废记录", db.scalar(select(func.count(ScrapRecord.id)).where(ScrapRecord.item_id == item_id)) or 0),
        ("成本分摊", db.scalar(select(func.count(CostAllocation.id)).where(CostAllocation.product_item_id == item_id)) or 0),
    ]
    _raise_delete_blocked("该产品需规", references)


def _ensure_employee_can_delete(db: Session, employee_id: int) -> None:
    references = [
        ("系统账号", db.scalar(select(func.count(User.id)).where(User.employee_id == employee_id)) or 0),
        ("员工上下级关系", db.scalar(select(func.count(Employee.id)).where(Employee.manager_employee_id == employee_id)) or 0),
        ("仓库负责人", db.scalar(select(func.count(Warehouse.id)).where(Warehouse.manager_employee_id == employee_id)) or 0),
        ("保养工单负责人", db.scalar(select(func.count(MaintenanceOrder.id)).where(MaintenanceOrder.owner_employee_id == employee_id)) or 0),
        ("采购订单", db.scalar(select(func.count(PurchaseOrder.id)).where(PurchaseOrder.purchaser_employee_id == employee_id)) or 0),
        ("到货入库", db.scalar(select(func.count(PurchaseReceipt.id)).where(PurchaseReceipt.receiver_employee_id == employee_id)) or 0),
        ("订单", db.scalar(select(func.count(SalesOrder.id)).where(SalesOrder.sales_employee_id == employee_id)) or 0),
        ("工序报工", db.scalar(select(func.count(OperationReport.id)).where(OperationReport.employee_id == employee_id)) or 0),
        ("成品入库", db.scalar(select(func.count(CompletionReceipt.id)).where(CompletionReceipt.receiver_employee_id == employee_id)) or 0),
        ("发货单", db.scalar(select(func.count(Delivery.id)).where(Delivery.shipper_employee_id == employee_id)) or 0),
        ("退货处理", db.scalar(select(func.count(ReturnOrder.id)).where(ReturnOrder.handler_employee_id == employee_id)) or 0),
    ]
    _raise_delete_blocked("该员工", references)


def _ensure_bom_can_delete(db: Session, bom: Bom) -> None:
    if _bom_has_usage(db, bom.id):
        raise HTTPException(status_code=400, detail="该BOM已被MRP或工单引用，不能删除")
    if bom.is_default and bom.status == "ACTIVE" and not _has_active_default_bom(
        db, bom.product_item_id, exclude_bom_id=bom.id
    ):
        raise HTTPException(status_code=400, detail="当前是产品唯一默认BOM，请先建立新的默认版本后再删除")


def _ensure_route_can_delete(db: Session, route: ProcessRoute) -> None:
    if _route_has_usage(db, route.id):
        raise HTTPException(status_code=400, detail="该工艺路线已被工单引用，不能删除")
    if route.is_default and route.status == "ACTIVE" and not _has_active_default_route(
        db, route.product_item_id, exclude_route_id=route.id
    ):
        raise HTTPException(status_code=400, detail="当前是产品唯一默认工艺路线，请先建立新的默认版本后再删除")


def _list_process_route_reads(db: Session, limit: int) -> list[ProcessRouteRead]:
    rows = db.execute(
        select(
            ProcessRoute.id.label("route_id"),
            ProcessRoute.route_code.label("route_code"),
            ProcessRoute.route_name.label("route_name"),
            ProcessRoute.product_item_id.label("product_item_id"),
            Item.item_code.label("product_code"),
            Item.item_name.label("product_name"),
            ProcessRoute.version_no.label("version_no"),
            ProcessRoute.is_default.label("is_default"),
            ProcessRoute.status.label("status"),
            ProcessRoute.effective_date.label("effective_date"),
            ProcessRoute.expire_date.label("expire_date"),
            ProcessRoute.remark.label("remark"),
        )
        .join(Item, Item.id == ProcessRoute.product_item_id)
        .order_by(ProcessRoute.is_default.desc(), ProcessRoute.id.desc())
        .limit(limit)
    ).mappings().all()
    if not rows:
        return []

    route_ids = [row["route_id"] for row in rows]
    miniapp_stamping_subquery = (
        select(
            MiniAppProduct.project_no.label("project_no"),
            MiniAppProduct.product_name.label("product_name"),
            MiniAppProduct.process_name.label("process_name"),
            func.max(MiniAppProduct.stamping_method).label("stamping_method"),
        )
        .where(MiniAppProduct.stamping_method.is_not(None), MiniAppProduct.stamping_method != "")
        .group_by(MiniAppProduct.project_no, MiniAppProduct.product_name, MiniAppProduct.process_name)
        .subquery()
    )
    operation_rows = db.execute(
        select(
            ProcessRouteOperation.id.label("route_operation_id"),
            ProcessRouteOperation.route_id.label("route_id"),
            ProcessRouteOperation.seq_no.label("seq_no"),
            ProcessRouteOperation.process_id.label("process_id"),
            Process.process_code.label("process_code"),
            Process.process_name.label("process_name"),
            miniapp_stamping_subquery.c.stamping_method.label("stamping_method"),
            ProcessRouteOperation.work_center_id.label("work_center_id"),
            WorkCenter.center_code.label("work_center_code"),
            WorkCenter.center_name.label("work_center_name"),
            ProcessRouteOperation.operation_name.label("operation_name"),
            ProcessRouteOperation.std_setup_minutes.label("std_setup_minutes"),
            ProcessRouteOperation.std_cycle_seconds.label("std_cycle_seconds"),
            ProcessRouteOperation.std_labor_cost.label("std_labor_cost"),
            ProcessRouteOperation.std_machine_cost.label("std_machine_cost"),
            ProcessRouteOperation.std_outsource_cost.label("std_outsource_cost"),
            ProcessRouteOperation.allowed_scrap_rate.label("allowed_scrap_rate"),
            ProcessRouteOperation.report_required.label("report_required"),
            ProcessRouteOperation.is_key_operation.label("is_key_operation"),
            ProcessRouteOperation.status.label("status"),
            ProcessRouteOperation.remark.label("remark"),
        )
        .join(ProcessRoute, ProcessRoute.id == ProcessRouteOperation.route_id)
        .join(Item, Item.id == ProcessRoute.product_item_id)
        .join(Product, Product.item_id == Item.id)
        .join(Process, Process.id == ProcessRouteOperation.process_id)
        .join(WorkCenter, WorkCenter.id == ProcessRouteOperation.work_center_id)
        .outerjoin(
            miniapp_stamping_subquery,
            and_(
                or_(
                    miniapp_stamping_subquery.c.project_no == Product.drawing_no,
                    miniapp_stamping_subquery.c.project_no == Product.product_family,
                    miniapp_stamping_subquery.c.project_no == Item.item_code,
                ),
                miniapp_stamping_subquery.c.product_name == Item.item_name,
                miniapp_stamping_subquery.c.process_name == ProcessRouteOperation.operation_name,
            ),
        )
        .where(ProcessRouteOperation.route_id.in_(route_ids))
        .order_by(ProcessRouteOperation.route_id, ProcessRouteOperation.seq_no)
    ).mappings().all()

    operation_map: dict[int, list[dict]] = {}
    for row in operation_rows:
        payload = dict(row)
        payload["report_required"] = bool(payload["report_required"])
        payload["is_key_operation"] = bool(payload["is_key_operation"])
        operation_map.setdefault(payload["route_id"], []).append(payload)

    return [
        ProcessRouteRead(
            route_id=row["route_id"],
            route_code=row["route_code"],
            route_name=row["route_name"],
            product_item_id=row["product_item_id"],
            product_code=row["product_code"],
            product_name=row["product_name"],
            version_no=row["version_no"],
            is_default=bool(row["is_default"]),
            status=row["status"],
            effective_date=row["effective_date"],
            expire_date=row["expire_date"],
            remark=row["remark"],
            operation_count=len(operation_map.get(row["route_id"], [])),
            operations=operation_map.get(row["route_id"], []),
        )
        for row in rows
    ]


def _get_process_route_read(db: Session, route_id: int) -> ProcessRouteRead:
    rows = _list_process_route_reads(db, limit=500)
    for row in rows:
        if row.route_id == route_id:
            return row
    raise HTTPException(status_code=404, detail="工艺路线不存在")