feat(base): 产品管理接口（类别校验/组合唯一/启停）

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-13 10:35:03 +08:00 · 2026-10-13 10:35:03 +08:00 · 7eeed85c96
commit 7eeed85c96
parent a694570dd1
4 changed files with 642 additions and 0 deletions
--- a/backend/app/api/v1/base/products.py
+++ b/backend/app/api/v1/base/products.py
@ -0,0 +1,114 @@
+"""产品管理接口（需求 2.1.2：产品主数据维护）。
+
+产品是折算阈值的行级锚点（product_category 决定 30%/20% 折算阈值），
+类别值域由后端校验 —— 前端下拉只是展示约束，后端校验是唯一防线。
+
+**数据提取（POST /products/extract）暂不提供**：它依赖业务明细表
+（V5 迁移），待业务管理任务完成后补充。
+
+基础数据属业务操作，登录用户即可访问（区别于系统管理模块的 AdminUser）；
+写操作经 RequestMetaDep 带入请求元信息，由 service 层写入操作日志。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from app.api.deps import DbSession, RequestMetaDep, get_current_user
+from app.core.response import ApiResponse, PageResult, ok
+from app.models import User
+from app.schemas.product import (
+    ProductCreateIn,
+    ProductOut,
+    ProductQuery,
+    ProductStatusIn,
+    ProductUpdateIn,
+)
+from app.services import product_service
+
+CurrentUser = Annotated[User, Depends(get_current_user)]
+
+router = APIRouter(prefix="/products", tags=["产品管理"])
+
+
+@router.get("", response_model=ApiResponse[PageResult[ProductOut]], summary="产品列表")
+def list_products(
+    session: DbSession,
+    _: CurrentUser,
+    query: Annotated[ProductQuery, Query()],
+) -> ApiResponse[PageResult[ProductOut]]:
+    """分页查询产品（含停用）。keyword 对名称或规格模糊匹配。"""
+    return ok(product_service.list_products(session, query))
+
+
+@router.post("", response_model=ApiResponse[ProductOut], summary="新增产品")
+def create_product(
+    session: DbSession,
+    current_user: CurrentUser,
+    payload: ProductCreateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[ProductOut]:
+    """新增产品。类别须在合法值域内；新产品默认启用。"""
+    return ok(
+        product_service.create_product(session, payload, operator=current_user, meta=meta),
+        message="产品已创建",
+    )
+
+
+@router.get("/{product_id}", response_model=ApiResponse[ProductOut], summary="产品详情")
+def get_product(
+    session: DbSession,
+    _: CurrentUser,
+    product_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[ProductOut]:
+    """按主键查询产品详情。"""
+    return ok(product_service.get_product(session, product_id))
+
+
+@router.put("/{product_id}", response_model=ApiResponse[ProductOut], summary="编辑产品")
+def update_product(
+    session: DbSession,
+    current_user: CurrentUser,
+    product_id: Annotated[int, Path(ge=1)],
+    payload: ProductUpdateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[ProductOut]:
+    """编辑产品基本信息（全量替换，不含状态；启用/停用走独立端点）。"""
+    return ok(
+        product_service.update_product(
+            session, product_id, payload, operator=current_user, meta=meta
+        ),
+        message="已保存",
+    )
+
+
+@router.delete("/{product_id}", response_model=ApiResponse[None], summary="删除产品")
+def delete_product(
+    session: DbSession,
+    current_user: CurrentUser,
+    product_id: Annotated[int, Path(ge=1)],
+    meta: RequestMetaDep,
+) -> ApiResponse[None]:
+    """物理删除产品（仅限误录数据，常规下架用停用）。
+
+    被业务明细引用的保护检查待业务数据表（V5）就位后补充。
+    """
+    product_service.delete_product(session, product_id, operator=current_user, meta=meta)
+    return ok(message="产品已删除")
+
+
+@router.put(
+    "/{product_id}/status", response_model=ApiResponse[ProductOut], summary="启用/停用产品"
+)
+def set_status(
+    session: DbSession,
+    current_user: CurrentUser,
+    product_id: Annotated[int, Path(ge=1)],
+    payload: ProductStatusIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[ProductOut]:
+    """启用或停用产品（1启用 0停用）。"""
+    result = product_service.set_status(
+        session, product_id, payload.status, operator=current_user, meta=meta
+    )
+    return ok(result, message="已启用" if payload.status == 1 else "已停用")
--- a/backend/app/repositories/product_repository.py
+++ b/backend/app/repositories/product_repository.py
@ -0,0 +1,118 @@
+"""产品主数据数据访问。
+
+本层只 flush 不 commit —— 事务边界由 services 层控制。
+"""
+
+from sqlalchemy import Select, func, or_, select
+from sqlalchemy.orm import Session
+
+from app.models import Product
+
+# ==================== 读取 ====================
+
+
+def get_by_id(session: Session, product_id: int, *, for_update: bool = False) -> Product | None:
+    """按主键查询。删除校验走 ``for_update`` 锁行（同 business_role_repository
+    的理由：校验与删除之间若发生并发引用，锁行使并发写入可见地失败，
+    而非引用被静默悬空）。"""
+    if for_update:
+        stmt = select(Product).where(Product.id == product_id).with_for_update()
+        return session.execute(stmt).scalar_one_or_none()
+    return session.get(Product, product_id)
+
+
+def get_by_name_spec(
+    session: Session, product_name: str, spec_model: str
+) -> Product | None:
+    """按 (名称, 规格型号) 组合查询。
+
+    组合是 uk_products_name_spec 唯一键的键值，先查用于给出即时业务提示；
+    并发窗口仍存在，最终由 IntegrityError 兜底（见 product_service）。
+    """
+    stmt = select(Product).where(
+        Product.product_name == product_name,
+        Product.spec_model == spec_model,
+    )
+    return session.execute(stmt).scalar_one_or_none()
+
+
+def list_products(
+    session: Session,
+    *,
+    offset: int,
+    limit: int,
+    keyword: str | None = None,
+    product_category: str | None = None,
+    status: int | None = None,
+) -> list[Product]:
+    """分页查询产品（含停用）。按 id 升序，与 M1 岗位列表的次序口径一致。"""
+    stmt = _apply_filters(
+        select(Product),
+        keyword=keyword,
+        product_category=product_category,
+        status=status,
+    )
+    stmt = stmt.order_by(Product.id).offset(offset).limit(limit)
+    return list(session.execute(stmt).scalars())
+
+
+def count_products(
+    session: Session,
+    *,
+    keyword: str | None = None,
+    product_category: str | None = None,
+    status: int | None = None,
+) -> int:
+    """统计符合条件的产品总数，供分页使用。"""
+    stmt = _apply_filters(
+        select(func.count(Product.id)),
+        keyword=keyword,
+        product_category=product_category,
+        status=status,
+    )
+    return int(session.execute(stmt).scalar_one())
+
+
+def _apply_filters(
+    stmt: Select,
+    *,
+    keyword: str | None,
+    product_category: str | None,
+    status: int | None,
+) -> Select:
+    """应用通用筛选条件。列表与计数共用，避免两处口径不一致。
+
+    关键词对名称/规格两列取 OR 命中。转义与 query_utils.fuzzy 同一机制
+    （``contains(autoescape=True)`` 转义 LIKE 通配符并附加 ESCAPE 子句）；
+    因 fuzzy 是单列 where 语义（两次调用会叠加成 AND），双列 OR 在此直接
+    写 contains —— M1 business_role_repository 的双列关键词同款写法。
+    规格型号常含 ``_``（如 DN300_II），不转义会当单字符通配执行。
+    """
+    if keyword:
+        stmt = stmt.where(
+            or_(
+                Product.product_name.contains(keyword, autoescape=True),
+                Product.spec_model.contains(keyword, autoescape=True),
+            )
+        )
+    if product_category:
+        stmt = stmt.where(Product.product_category == product_category)
+    if status is not None:
+        stmt = stmt.where(Product.status == status)
+    return stmt
+
+
+# ==================== 写入 ====================
+
+
+def create(session: Session, product: Product) -> Product:
+    """新增产品。只 flush 以取得自增 id。"""
+    session.add(product)
+    session.flush()
+    return product
+
+
+def delete(session: Session, product: Product) -> None:
+    """物理删除产品。调用方须先完成引用保护校验（V5 就位后生效）。"""
+    session.delete(product)
+    session.flush()
--- a/backend/app/schemas/product.py
+++ b/backend/app/schemas/product.py
@ -0,0 +1,104 @@
+"""产品管理的请求/响应模型。
+
+数量与单价（stock_quantity/sale_price）入参出参一律用**字符串**承载：
+前端 JS 的 number 无法精确表示全部十进制小数，字符串是金额红线的入口闸门，
+Decimal 转换统一收敛在 service 层。字段命名与 user/role/param 一致采用
+蛇形直出（前端 TS 接口同构）。
+"""
+
+from datetime import datetime
+from decimal import Decimal, InvalidOperation
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.core.response import PageQuery
+
+
+def _parse_non_negative(value: str, *, label: str) -> str:
+    """非负数字字符串校验：必须可被 Decimal 解析、为有限数且 ≥ 0。
+
+    ``NaN`` 参与比较会抛 InvalidOperation、``Infinity`` 会在后续运算溢出，
+    均须在入口用 ``is_finite()`` 拦下（与 schemas/param.py 的 _parse_amount
+    同一口径）。
+    """
+    try:
+        amount = Decimal(value.strip())
+    except InvalidOperation as exc:
+        raise ValueError(f"{label}必须是有效数字：{value}") from exc
+    if not amount.is_finite():
+        raise ValueError(f"{label}必须为有限数字：{value}")
+    if amount < 0:
+        raise ValueError(f"{label}不得为负数：{value}")
+    return value
+
+
+class ProductOut(BaseModel):
+    """产品响应（列表与详情共用）。数量/单价按红线以字符串出参。"""
+
+    id: int
+    product_name: str
+    spec_model: str
+    unit: str | None = None
+    stock_quantity: str | None = None
+    sale_price: str | None = None
+    product_category: str
+    status: int = 1
+    remark: str | None = None
+    create_time: datetime
+    update_time: datetime
+
+
+class ProductCreateIn(BaseModel):
+    """新增产品。
+
+    spec_model 允许缺省/空：service 层归一为空串（组合唯一键
+    uk_products_name_spec 对 NULL 不去重，归 NULL 会放行重名同规格数据）。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    product_name: str = Field(min_length=1, max_length=128, description="产品名称")
+    spec_model: str | None = Field(default=None, max_length=128, description="规格型号，空归一为空串")
+    unit: str | None = Field(default=None, max_length=16, description="计量单位")
+    stock_quantity: str | None = Field(default=None, description="数量，非负数字字符串")
+    sale_price: str | None = Field(default=None, description="销售单价（元），非负数字字符串")
+    product_category: str = Field(min_length=1, max_length=32, description="产品类别")
+    remark: str | None = Field(default=None, max_length=255)
+
+    @field_validator("stock_quantity")
+    @classmethod
+    def _check_stock_quantity(cls, value: str | None) -> str | None:
+        """空串与 None 同义（可空列），归一为 None；非空则须为非负数字。"""
+        if not value:
+            return None
+        return _parse_non_negative(value, label="数量")
+
+    @field_validator("sale_price")
+    @classmethod
+    def _check_sale_price(cls, value: str | None) -> str | None:
+        """空串与 None 同义（可空列），归一为 None；非空则须为非负数字。"""
+        if not value:
+            return None
+        return _parse_non_negative(value, label="销售单价")
+
+
+class ProductUpdateIn(ProductCreateIn):
+    """编辑产品。
+
+    前端表单全量提交，故为全量替换语义；**不含 status** —— 启用/停用是
+    独立操作（走 PUT /{id}/status），避免编辑表单误触停用。
+    """
+
+
+class ProductStatusIn(BaseModel):
+    """启用/停用产品。"""
+
+    status: int = Field(ge=0, le=1, description="1启用 0停用")
+
+
+class ProductQuery(PageQuery):
+    """产品列表查询条件。"""
+
+    keyword: str | None = Field(default=None, max_length=64, description="名称或规格，模糊匹配")
+    product_category: str | None = Field(default=None, max_length=32, description="按类别筛选")
+    status: int | None = Field(default=None, ge=0, le=1, description="状态筛选")
--- a/backend/app/services/product_service.py
+++ b/backend/app/services/product_service.py
@ -0,0 +1,306 @@
+"""产品管理业务编排。
+
+产品主数据的两条硬约束（违反会算错钱）：
+
+- ``product_category`` 是**折算阈值的行级锚点**：混合业务按产品逐条计算
+  （对照表 #14），阈值由 M1 参数 ``product.threshold`` 按 PRODUCT_CATEGORY
+  维度承担。类别游离于合法值域之外时，核算按类别回溯参数会取不到阈值，
+  故值域校验是业务规则而非格式校验
+- 数量/单价全程 ``Decimal``：schema 层已拦格式与非负（金额红线入口），
+  service 层负责字符串到 Decimal 的转换与落库
+
+另：``spec_model`` 归一为**空串而非 NULL** —— 它参与 uk_products_name_spec
+组合唯一键，MySQL 组合索引对 NULL 不去重，归 NULL 会放行重名同规格数据。
+删除为物理删除，被业务明细引用的保护检查待业务数据表（V5）就位后补充。
+"""
+
+import logging
+from decimal import Decimal
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core.constants import (
+    ACTION_CREATE,
+    ACTION_DELETE,
+    ACTION_DISABLE,
+    ACTION_ENABLE,
+    ACTION_UPDATE,
+    MODULE_PRODUCT,
+    PRODUCT_CATEGORIES,
+)
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.core.response import PageResult
+from app.models import Product, User
+from app.repositories import product_repository
+from app.schemas.product import (
+    ProductCreateIn,
+    ProductOut,
+    ProductQuery,
+    ProductUpdateIn,
+)
+from app.services import audit_service
+from app.services.audit_service import AuditRecord, RequestMeta
+
+logger = logging.getLogger(__name__)
+
+# 唯一键冲突的 MySQL 错误码（pymysql 原始异常 args 首位）
+_ERR_DUP_ENTRY = 1062
+
+
+def _to_out(product: Product) -> ProductOut:
+    """ORM 对象转响应模型。禁止把 ORM 对象直接当 response_model。"""
+    return ProductOut(
+        id=product.id,
+        product_name=product.product_name,
+        spec_model=product.spec_model,
+        unit=product.unit,
+        stock_quantity=str(product.stock_quantity) if product.stock_quantity is not None else None,
+        sale_price=str(product.sale_price) if product.sale_price is not None else None,
+        product_category=product.product_category,
+        status=product.status,
+        remark=product.remark,
+        create_time=product.create_time,
+        update_time=product.update_time,
+    )
+
+
+def _to_decimal(value: str | None) -> Decimal | None:
+    """数字字符串转 Decimal（可空列的空串/None 透传 None）。
+
+    入参已 str 化再进 Decimal：即使上游口径变化传入数字类型，也不经 float
+    二进制近似，金额红线在最后一道转换处仍然成立。schema 已保证可解析，
+    此处不再重复校验。
+    """
+    if not value:
+        return None
+    return Decimal(str(value))
+
+
+def _ensure_category(product_category: str) -> None:
+    """类别必须 ∈ PRODUCT_CATEGORIES（前端下拉仅是展示约束，后端校验是唯一防线）。"""
+    if product_category not in PRODUCT_CATEGORIES:
+        raise BadRequestError("产品类别不合法")
+
+
+def _is_duplicate_entry(exc: IntegrityError) -> bool:
+    """判断 IntegrityError 是否为 MySQL 1062（唯一键冲突）。
+
+    只有 1062 是「已存在」语义，可转业务提示；其余完整性错误（外键、
+    数据溢出等）语义未知，必须原样上抛交由全局处理器记 500，不能吞成 400。
+    """
+    orig_args = getattr(exc.orig, "args", ())
+    return bool(orig_args) and orig_args[0] == _ERR_DUP_ENTRY
+
+
+def _raise_if_duplicate(exc: IntegrityError) -> None:
+    """唯一键冲突统一转业务提示；其余完整性错误原样重抛。"""
+    if _is_duplicate_entry(exc):
+        raise BadRequestError("已存在同名称同规格的产品") from exc
+    raise exc
+
+
+def list_products(session: Session, query: ProductQuery) -> PageResult[ProductOut]:
+    """分页查询产品（含停用）。"""
+    products = product_repository.list_products(
+        session,
+        offset=query.offset,
+        limit=query.page_size,
+        keyword=query.keyword,
+        product_category=query.product_category,
+        status=query.status,
+    )
+    total = product_repository.count_products(
+        session,
+        keyword=query.keyword,
+        product_category=query.product_category,
+        status=query.status,
+    )
+    return PageResult[ProductOut](
+        records=[_to_out(product) for product in products],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_product(session: Session, product_id: int) -> ProductOut:
+    """产品详情。
+
+    :raises NotFoundError: 产品不存在
+    """
+    product = product_repository.get_by_id(session, product_id)
+    if product is None:
+        raise NotFoundError("产品不存在")
+    return _to_out(product)
+
+
+def create_product(
+    session: Session,
+    payload: ProductCreateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> ProductOut:
+    """新增产品。类别须在合法值域内，新产品固定启用。
+
+    :raises BadRequestError: 产品类别不合法 / 已存在同名称同规格的产品
+    """
+    _ensure_category(payload.product_category)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_PRODUCT,
+            action=ACTION_CREATE,
+            summary=f"新增产品 {payload.product_name}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        spec_model = payload.spec_model or ""  # 归一空串：组合唯一键对 NULL 不去重
+        if product_repository.get_by_name_spec(session, payload.product_name, spec_model) is not None:
+            raise BadRequestError("已存在同名称同规格的产品")
+
+        product = Product(
+            product_name=payload.product_name,
+            spec_model=spec_model,
+            unit=payload.unit,
+            stock_quantity=_to_decimal(payload.stock_quantity),
+            sale_price=_to_decimal(payload.sale_price),
+            product_category=payload.product_category,
+            status=1,
+            remark=payload.remark,
+        )
+        try:
+            product_repository.create(session, product)
+            # 响应与后置动作全部在 commit 之前完成（见 business_role_service 同处说明）
+            out = _to_out(product)
+            session.commit()
+        except IntegrityError as exc:
+            # 先查后写不是原子操作：并发提交同名称同规格会撞 uk_products_name_spec
+            session.rollback()
+            _raise_if_duplicate(exc)
+
+        logger.info(
+            "新增产品 | product_id=%s | category=%s", product.id, payload.product_category
+        )
+        return out
+
+
+def update_product(
+    session: Session,
+    product_id: int,
+    payload: ProductUpdateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> ProductOut:
+    """编辑产品基本信息（全量替换，不含状态；启停走独立端点）。
+
+    :raises NotFoundError: 产品不存在
+    :raises BadRequestError: 产品类别不合法 / 已存在同名称同规格的产品
+    """
+    product = product_repository.get_by_id(session, product_id)
+    _ensure_category(payload.product_category)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_PRODUCT,
+            action=ACTION_UPDATE,
+            summary=f"修改产品 {product.product_name if product else f'#{product_id}'}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if product is None:
+            raise NotFoundError("产品不存在")
+
+        product.product_name = payload.product_name
+        product.spec_model = payload.spec_model or ""
+        product.unit = payload.unit
+        product.stock_quantity = _to_decimal(payload.stock_quantity)
+        product.sale_price = _to_decimal(payload.sale_price)
+        product.product_category = payload.product_category
+        product.remark = payload.remark
+        out = _to_out(product)
+        try:
+            session.commit()
+        except IntegrityError as exc:
+            # 改名/改规格可能撞 uk_products_name_spec（本行改值与他人既有数据重合）
+            session.rollback()
+            _raise_if_duplicate(exc)
+
+        logger.info("修改产品 | product_id=%s | category=%s", product_id, payload.product_category)
+        return out
+
+
+def set_status(
+    session: Session,
+    product_id: int,
+    status: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> ProductOut:
+    """启用或停用产品。停用为主（模型约定），物理删除仅限误录数据。
+
+    :raises NotFoundError: 产品不存在
+    """
+    product = product_repository.get_by_id(session, product_id)
+    verb = "启用" if status == 1 else "停用"
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_PRODUCT,
+            action=ACTION_ENABLE if status == 1 else ACTION_DISABLE,
+            summary=f"{verb}产品 {product.product_name if product else f'#{product_id}'}",
+            request_data={"status": status},
+        ),
+        meta=meta,
+    ):
+        if product is None:
+            raise NotFoundError("产品不存在")
+
+        product.status = status
+        out = _to_out(product)
+        session.commit()
+        logger.info("变更产品状态 | product_id=%s | status=%s", product_id, status)
+        return out
+
+
+def delete_product(
+    session: Session,
+    product_id: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> None:
+    """物理删除产品。
+
+    **被业务明细（发货/回款）引用的保护检查待业务数据表（V5）就位后补充**；
+    当前先以 ``FOR UPDATE`` 锁行删除 —— V5 接入后，校验与删除之间的并发
+    引用会因锁可见地失败，而非引用被静默悬空。
+
+    :raises NotFoundError: 产品不存在
+    """
+    product = product_repository.get_by_id(session, product_id, for_update=True)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_PRODUCT,
+            action=ACTION_DELETE,
+            summary=f"删除产品 {product.product_name if product else f'#{product_id}'}",
+        ),
+        meta=meta,
+    ):
+        if product is None:
+            raise NotFoundError("产品不存在")
+
+        product_name = product.product_name  # 固化：对象随删除失效
+        product_repository.delete(session, product)
+        session.commit()
+        logger.info("删除产品 | product_name=%s", product_name)