from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status from fastapi.responses import Response from sqlalchemy import or_, select from sqlalchemy.orm import Session from app.database import get_db from app.deps import require_roles from app.models import DeviceQRCode, MoldLockFeedback, Personnel, Product, ProductionReportItem, Role, WorkSessionDevice from app.schemas import ImportResult, MoldNameOut, PageResponse, ProductCreate, ProductOut from app.services.attendance_points import accessible_point_names, require_attendance_point_access from app.services.common import normalize_device_no, paginate from app.services.excel_import import ProductImportValidationError, import_products from app.services.excel_export import export_products from app.services.cleaning import LEGACY_CLEANING_PROCESS_NAME, is_cleaning_stamping_method from app.services.continuous_die import is_continuous_die_product from app.services.display_names import mold_process_display_name from app.services.misc_work import ensure_misc_work_products, is_misc_stamping_method from app.services.multi_person import is_multi_person_product from app.services.process_names import MISC_PROCESS_NAME, normalize_process_name from app.services.serializers import product_out router = APIRouter(prefix="/api/products", tags=["products"]) def _cascade_product_key_change(db: Session, original_key: dict, next_key: dict) -> None: if original_key == next_key: return old_point = original_key["attendance_point_name"] old_product = original_key["product_name"] old_process = original_key["process_name"] next_point = next_key["attendance_point_name"] next_product = next_key["product_name"] next_process = next_key["process_name"] old_qr = db.get(DeviceQRCode, { "attendance_point_name": old_point, "device_no": old_product, "process_name": old_process, }) if old_qr is not None: qr_target = db.get(DeviceQRCode, { "attendance_point_name": next_point, "device_no": next_product, "process_name": next_process, }) if qr_target is not None: raise HTTPException(status_code=409, detail="目标产品/工序二维码已存在，不能合并覆盖") old_qr.attendance_point_name = next_point old_qr.device_no = next_product old_qr.process_name = next_process for device in db.scalars(select(WorkSessionDevice).where( WorkSessionDevice.attendance_point_name == old_point, WorkSessionDevice.device_no == old_product, WorkSessionDevice.process_name == old_process, )).all(): device.attendance_point_name = next_point device.device_no = next_product device.process_name = next_process for item in db.scalars(select(ProductionReportItem).where( ProductionReportItem.attendance_point_name == old_point, ProductionReportItem.project_no == original_key["project_no"], ProductionReportItem.product_name == old_product, ProductionReportItem.process_name == old_process, )).all(): item.attendance_point_name = next_point item.project_no = next_key["project_no"] item.product_name = next_product item.process_name = next_process for feedback in db.scalars(select(MoldLockFeedback).where( MoldLockFeedback.attendance_point_name == old_point, MoldLockFeedback.mold_name == old_product, MoldLockFeedback.process_name == old_process, )).all(): feedback.attendance_point_name = next_point feedback.mold_name = next_product feedback.process_name = next_process @router.get("/molds", response_model=list[MoldNameOut]) def list_mold_names( keyword: str = "", limit: int = Query(100, ge=1, le=500), user: Personnel = Depends(require_roles(Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> list[MoldNameOut]: point_names = accessible_point_names(db, user) ensure_misc_work_products(db, point_names) db.commit() query = ( select(Product) .where( Product.attendance_point_name.in_(point_names), Product.product_name != "", Product.process_name != "", Product.device_no == "", ) .distinct() .order_by(Product.attendance_point_name, Product.product_name, Product.process_name) ) if keyword: like = f"%{keyword}%" query = query.where(or_(Product.product_name.like(like), Product.process_name.like(like), Product.stamping_method.like(like))) rows = db.scalars(query.limit(limit)).all() return [ MoldNameOut( attendance_point_name=product.attendance_point_name, name=product.product_name, process_name=product.process_name or "", stamping_method=product.stamping_method, operator_count=product.operator_count, is_cleaning=is_cleaning_stamping_method(product.stamping_method), is_misc=is_misc_stamping_method(product.stamping_method), is_continuous_die=is_continuous_die_product(product), is_multi_person=is_multi_person_product(product), display_name=mold_process_display_name(product.product_name, product.process_name, product.stamping_method), ) for product in rows ] @router.get("", response_model=PageResponse) def list_products( attendance_point_name: str = "", keyword: str = "", device_no: str = "", product_name: str = "", page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100), user: Personnel = Depends(require_roles(Role.worker, Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> PageResponse: point_names = accessible_point_names(db, user) ensure_misc_work_products(db, point_names) db.commit() query = ( select(Product) .where(Product.device_no == "", Product.attendance_point_name.in_(point_names)) .order_by( Product.updated_at.desc(), Product.attendance_point_name.asc(), Product.project_no.asc(), Product.product_name.asc(), Product.process_name.asc(), ) ) if attendance_point_name: try: point_name = require_attendance_point_access(db, user, attendance_point_name) except (ValueError, PermissionError) as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc query = query.where(Product.attendance_point_name == point_name) if product_name: query = query.where(Product.product_name == product_name) if keyword: like = f"%{keyword}%" query = query.where( or_( Product.project_no.like(like), Product.product_name.like(like), Product.attendance_point_name.like(like), Product.material_code.like(like), Product.material_name.like(like), Product.supplier.like(like), Product.process_name.like(like), Product.stamping_method.like(like), ) ) result = paginate(db, query, page, page_size) return PageResponse(**{**result, "rows": [product_out(row) for row in result["rows"]]}) @router.post("", response_model=ProductOut) def save_product( payload: ProductCreate, user: Personnel = Depends(require_roles(Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> ProductOut: try: point_name = require_attendance_point_access(db, user, payload.attendance_point_name) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except PermissionError as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc raw_process_name = str(payload.process_name or "").strip() if raw_process_name == LEGACY_CLEANING_PROCESS_NAME: raise HTTPException(status_code=400, detail="清洗请填写在冲压方式中，工序只能填写数字，处理杂活除外") try: process_name = normalize_process_name(raw_process_name) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc is_cleaning = is_cleaning_stamping_method(payload.stamping_method) is_misc = is_misc_stamping_method(payload.stamping_method) if is_misc and process_name != MISC_PROCESS_NAME: raise HTTPException(status_code=400, detail="处理杂活的工序必须填写杂活") if not is_misc and process_name == MISC_PROCESS_NAME: raise HTTPException(status_code=400, detail="只有处理杂活的工序可以填写杂活") if not is_cleaning and not is_misc and payload.standard_beat <= 0: raise HTTPException(status_code=400, detail="非清洗工序必须填写标准节拍") next_key = { "attendance_point_name": point_name, "project_no": payload.project_no, "product_name": payload.product_name, "device_no": "", "process_name": process_name, } original_key = { "attendance_point_name": payload.original_attendance_point_name or point_name, "project_no": payload.original_project_no or payload.project_no, "product_name": payload.original_product_name or payload.product_name, "device_no": normalize_device_no(payload.original_device_no if payload.original_device_no is not None else ""), "process_name": payload.original_process_name if payload.original_process_name is not None else process_name, } product = db.get(Product, original_key) if product is not None and original_key != next_key: existed = db.get(Product, next_key) if existed is not None: raise HTTPException(status_code=409, detail="目标产品/工序已存在，不能合并覆盖") else: _cascade_product_key_change(db, original_key, next_key) product.project_no = next_key["project_no"] product.product_name = next_key["product_name"] product.device_no = next_key["device_no"] product.process_name = next_key["process_name"] elif product is None: product = db.get(Product, next_key) if product is None: product = Product( attendance_point_name=next_key["attendance_point_name"], project_no=next_key["project_no"], product_name=next_key["product_name"], device_no=next_key["device_no"], process_name=next_key["process_name"], ) db.add(product) aux_values = ( payload.product_net_weight_kg, payload.product_gross_weight_kg, payload.scrap_loss_rate, payload.waste_price_yuan_per_kg, ) if all(value is None for value in aux_values): sibling = next( ( item for item in db.scalars( select(Product).where(Product.product_name == next_key["product_name"], Product.device_no == "") .where(Product.attendance_point_name == point_name) ).all() if item.product_net_weight_kg is not None and item.product_gross_weight_kg is not None and item.scrap_loss_rate is not None and item.waste_price_yuan_per_kg is not None ), None, ) if sibling is not None: aux_values = ( sibling.product_net_weight_kg, sibling.product_gross_weight_kg, sibling.scrap_loss_rate, sibling.waste_price_yuan_per_kg, ) product.profile_no = payload.profile_no product.attendance_point_name = next_key["attendance_point_name"] product.material_code = payload.material_code product.material_name = payload.material_name product.supplier = payload.supplier product.product_net_weight_kg = aux_values[0] product.product_gross_weight_kg = aux_values[1] product.scrap_loss_rate = aux_values[2] product.waste_price_yuan_per_kg = aux_values[3] product.device_no = next_key["device_no"] product.process_name = next_key["process_name"] product.stamping_method = payload.stamping_method product.operator_count = payload.operator_count or 1 product.process_unit_price_yuan = payload.process_unit_price_yuan or 0 product.standard_beat = 0 if (is_cleaning or is_misc) else payload.standard_beat product.standard_workload = 0 for sibling in db.scalars(select(Product).where( Product.attendance_point_name == point_name, Product.product_name == next_key["product_name"], Product.device_no == "", )).all(): sibling.product_net_weight_kg = product.product_net_weight_kg sibling.product_gross_weight_kg = product.product_gross_weight_kg sibling.scrap_loss_rate = product.scrap_loss_rate sibling.waste_price_yuan_per_kg = product.waste_price_yuan_per_kg db.commit() db.refresh(product) return product_out(product) @router.delete("", status_code=204) def delete_product( attendance_point_name: str, project_no: str, product_name: str, device_no: str = "", process_name: str = "", user: Personnel = Depends(require_roles(Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> None: try: point_name = require_attendance_point_access(db, user, attendance_point_name) except (ValueError, PermissionError) as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc product = db.get( Product, { "attendance_point_name": point_name, "project_no": project_no, "product_name": product_name, "device_no": "", "process_name": process_name, }, ) if product is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="产品不存在") qrcode = db.get(DeviceQRCode, { "attendance_point_name": point_name, "device_no": product.product_name, "process_name": product.process_name or "", }) if qrcode is not None: db.delete(qrcode) db.delete(product) db.commit() @router.post("/import", response_model=ImportResult) async def import_product_excel( file: UploadFile = File(...), user: Personnel = Depends(require_roles(Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> ImportResult: content = await file.read() try: result = import_products(db, content, user=user) except ProductImportValidationError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return ImportResult(**result.as_dict()) @router.get("/export") def export_product_excel( user: Personnel = Depends(require_roles(Role.admin, Role.manager)), db: Session = Depends(get_db), ) -> Response: content = export_products(db, user=user) return Response( content=content, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": 'attachment; filename="products.xlsx"'}, )