backend/app/core/db_errors.py

@@ -0,0 +1,18 @@
+"""数据库完整性错误的统一识别与翻译口径。
+
+判断依据：pymysql 原生异常 ``orig.args`` 首位是 MySQL 错误码 —— 只有
+1062（ER_DUP_ENTRY，唯一键冲突）是「已存在」语义，可安全转 400 业务
+提示；其余完整性错误（外键 RESTRICT、数据截断等）原样上抛，由全局
+异常处理器收敛 500（程序缺陷或数据异常，均不应伪装成用户错误）。
+"""
+
+from sqlalchemy.exc import IntegrityError
+
+# MySQL 唯一键冲突错误码（pymysql 原生异常 args 首位）
+ERR_DUP_ENTRY = 1062
+
+
+def is_duplicate_entry(exc: IntegrityError) -> bool:
+    """判断 IntegrityError 是否为 MySQL 唯一键冲突（错误码 1062）。"""
+    args = getattr(exc.orig, "args", ())
+    return bool(args) and args[0] == ERR_DUP_ENTRY

backend/app/services/business_service.py

@@ -30,6 +30,7 @@ from app.core.constants import (
     MODULE_BUSINESS,
     NO_PREFIX_BUSINESS,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageQuery, PageResult
 from app.models import (
@@ -221,8 +222,7 @@ def _ensure_customer(session: Session, customer_id: int) -> Customer:
 
 
 def _is_duplicate_entry(exc: IntegrityError) -> bool:
-    args = getattr(exc.orig, "args", ())
-    return bool(args) and args[0] == 1062
+    return is_duplicate_entry(exc)
 
 
 def list_businesses(session: Session, query: BusinessQuery) -> PageResult[BusinessOut]:

backend/app/services/customer_service.py

@@ -26,6 +26,7 @@ from app.core.constants import (
     CUSTOMER_TYPES,
     MODULE_CUSTOMER,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageQuery, PageResult
 from app.models import Customer, OwnershipChangeLog, Personnel, User
@@ -108,14 +109,13 @@ def _ensure_owner_personnel(session: Session, personnel_id: int) -> Personnel:
 
 
 def _is_duplicate_entry(exc: IntegrityError) -> bool:
-    """判断 IntegrityError 是否为 MySQL 唯一键冲突（错误码 1062）。
+    """判断 IntegrityError 是否为 MySQL 唯一键冲突（1062）。
 
     customers 上还有归属人员外键（RESTRICT），并发窗口内删除人员等场景
     会抛 1452 —— 不能一并翻译成「已存在」，否则用户会被错误提示带偏；
     非 1062 原样上抛，由全局处理器按 500 收敛。
     """
-    args = getattr(exc.orig, "args", ())
-    return bool(args) and args[0] == 1062
+    return is_duplicate_entry(exc)
 
 
 def list_customers(session: Session, query: CustomerQuery) -> PageResult[CustomerOut]:

backend/app/services/product_service.py

@@ -29,6 +29,7 @@ from app.core.constants import (
     MODULE_PRODUCT,
     PRODUCT_CATEGORIES,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageResult
 from app.models import Product, User
@@ -44,9 +45,6 @@ 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。"""
@@ -89,8 +87,7 @@ def _is_duplicate_entry(exc: IntegrityError) -> bool:
     只有 1062 是「已存在」语义，可转业务提示；其余完整性错误（外键、
     数据溢出等）语义未知，必须原样上抛交由全局处理器记 500，不能吞成 400。
     """
-    orig_args = getattr(exc.orig, "args", ())
-    return bool(orig_args) and orig_args[0] == _ERR_DUP_ENTRY
+    return is_duplicate_entry(exc)
 
 
 def _raise_if_duplicate(exc: IntegrityError) -> None:

backend/app/services/rating_service.py

@@ -31,6 +31,7 @@ from app.core.constants import (
     ACTION_UPDATE,
     MODULE_RATING,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.models import CustomerRatingGrade, User
 from app.repositories import rating_repository
@@ -44,9 +45,6 @@ logger = logging.getLogger(__name__)
 _SCORE_FLOOR = Decimal("0")
 _SCORE_CEIL = Decimal("100")
 
-# 唯一键冲突的 MySQL 错误码（pymysql 原始异常 args 首位）
-_ERR_DUP_ENTRY = 1062
-
 
 def _to_out(grade: CustomerRatingGrade) -> RatingGradeOut:
     """ORM 对象转响应模型。分值 Decimal → str（分值红线出参口径）。"""
@@ -110,8 +108,7 @@ def _is_duplicate_entry(exc: IntegrityError) -> bool:
     只有 1062 是「已存在」语义，可转业务提示；其余完整性错误（外键、
     数据溢出等）语义未知，必须原样上抛交由全局处理器记 500，不能吞成 400。
     """
-    orig_args = getattr(exc.orig, "args", ())
-    return bool(orig_args) and orig_args[0] == _ERR_DUP_ENTRY
+    return is_duplicate_entry(exc)
 
 
 def _raise_if_duplicate(exc: IntegrityError) -> None:

backend/app/services/receipt_service.py

@@ -35,6 +35,7 @@ from app.core.constants import (
     MODULE_RECEIPT,
     NO_PREFIX_RECEIPT,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageResult
 from app.models import Receipt
@@ -342,5 +343,4 @@ def delete_receipt(
 
 def _is_duplicate_entry(exc: IntegrityError) -> bool:
     """识别 MySQL 唯一键冲突（1062），与 business_service 同口径。"""
-    args = getattr(exc.orig, "args", ())
-    return bool(args) and args[0] == 1062
+    return is_duplicate_entry(exc)

backend/app/services/shipment_service.py

@@ -35,6 +35,7 @@ from app.core.constants import (
     MODULE_SHIPMENT,
     NO_PREFIX_SHIPMENT,
 )
+from app.core.db_errors import is_duplicate_entry
 from app.core.exceptions import BadRequestError, NotFoundError
 from app.core.response import PageResult
 from app.models import Business, Shipment, ShipmentItem, User
@@ -144,8 +145,7 @@ def _ensure_business_writable(session: Session, business_id: int, action: str) -
 
 def _is_duplicate_entry(exc: IntegrityError) -> bool:
     """识别 MySQL 唯一键冲突（1062），与 business/receipt_service 同口径。"""
-    args = getattr(exc.orig, "args", ())
-    return bool(args) and args[0] == 1062
+    return is_duplicate_entry(exc)
 
 
 def list_shipments(session: Session, query: ShipmentQuery) -> PageResult[ShipmentOut]: