"""API - 商品参数模板管理""" from typing import Optional, Tuple, Dict, Any from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app.core.database import get_db from app.core.auth import get_current_user from app.core.response import success_response, error_response from app.crud.product_parameter import product_parameter from app.schemas.product_parameter import ( ProductParameterCreate, ProductParameterUpdate, ProductParameterResponse, ProductParameterListResponse, ProductParameterSimpleResponse ) router = APIRouter(prefix="/web/admin/product-parameters", tags=["商品参数模板管理"]) def require_admin(current_user: dict = Depends(get_current_user)): """验证运营人员权限""" if current_user.get("role") not in {"admin", "super_admin"}: raise HTTPException(status_code=403, detail="需要运营人员权限") return current_user def _resolve_payload_with_crypto_fallback(request: dict, current_user: dict) -> Tuple[Dict[str, Any], Optional[dict]]: """兼容明文/密文请求，返回业务 payload。失败时返回 error_response。""" req_dict = request if isinstance(request, dict) else {} # 若是密文信封，优先解密。 if isinstance(req_dict, dict) and req_dict.get("enc") == 1 and req_dict.get("edata"): try: from app.core import api_crypto from app.core.config import settings ts = req_dict.get("ts") nonce = req_dict.get("nonce") edata = req_dict.get("edata") operator_user_id = current_user.get("user_id") if isinstance(current_user, dict) else None fixed_secret = getattr(settings, "API_ENCRYPT_FIXED_SECRET", "") or "" if not (fixed_secret and operator_user_id and ts is not None and nonce): return {}, error_response(400, "密文解析配置缺失") key = api_crypto.derive_key( fixed_secret=fixed_secret, user_id=str(operator_user_id), ts=int(ts), nonce=str(nonce), ) plain = api_crypto.decrypt_json(str(edata), key) payload = plain.get("data", plain) if isinstance(plain, dict) else {} if isinstance(payload, dict): return payload, None return {}, error_response(400, "密文解密后数据格式错误") except Exception as exc: return {}, error_response(400, f"密文解密失败: {exc}") payload = req_dict.get("data", req_dict) if isinstance(req_dict, dict) else {} return (payload if isinstance(payload, dict) else {}), None @router.get("", response_model=dict) def list_parameters( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[int] = Query(None, description="是否启用"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取参数模板列表""" items, total = product_parameter.get_multi_paged( db, skip=(page - 1) * pageSize, limit=pageSize, keyword=keyword, is_active=isActive ) return success_response({ "list": [{ "id": p.id, "name": p.name, "code": p.code, "paramType": p.param_type, "options": p.options, "placeholder": p.placeholder, "isRequired": p.is_required, "isActive": p.is_active, "sortOrder": p.sort_order, "remark": p.remark, "createAt": str(p.create_at) if p.create_at else None } for p in items], "total": total, "page": page, "pageSize": pageSize }) @router.get("/simple", response_model=dict) def list_parameters_simple( db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取参数模板简洁列表（用于下拉选择）""" items = product_parameter.get_active_all(db) return success_response({ "list": [{ "id": p.id, "name": p.name, "code": p.code, "paramType": p.param_type, "options": p.options } for p in items], "total": len(items) }) @router.get("/{parameter_id}", response_model=dict) def get_parameter( parameter_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取参数详情""" param = product_parameter.get(db, parameter_id) if not param: return error_response(404, "参数不存在") return success_response({ "id": param.id, "name": param.name, "code": param.code, "paramType": param.param_type, "options": param.options, "placeholder": param.placeholder, "isRequired": param.is_required, "isActive": param.is_active, "sortOrder": param.sort_order, "remark": param.remark, "createAt": str(param.create_at) if param.create_at else None }) @router.post("", response_model=dict) def create_parameter( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建参数模板""" payload, req_err = _resolve_payload_with_crypto_fallback(request, current_user) if req_err is not None: return req_err try: create_data = ProductParameterCreate.model_validate(payload) except Exception as exc: return error_response(400, f"参数错误: {exc}") # 检查编码是否已存在 if product_parameter.code_exists(db, create_data.code): return error_response(400, "参数编码已存在") creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} param = product_parameter.create(db, create_data.model_dump(), creator) return success_response({ "id": param.id, "name": param.name, "code": param.code }) @router.put("/{parameter_id}", response_model=dict) def update_parameter( parameter_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新参数模板""" payload, req_err = _resolve_payload_with_crypto_fallback(request, current_user) if req_err is not None: return req_err try: update_data = ProductParameterUpdate.model_validate(payload) except Exception as exc: return error_response(400, f"参数错误: {exc}") # 检查编码是否已存在（排除自身） if update_data.code: if product_parameter.code_exists(db, update_data.code, exclude_id=parameter_id): return error_response(400, "参数编码已存在") updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} param = product_parameter.update(db, parameter_id, update_data.model_dump(exclude_unset=True), updater) if not param: return error_response(404, "参数不存在") return success_response({ "id": param.id, "name": param.name, "code": param.code }) @router.delete("/{parameter_id}", response_model=dict) def delete_parameter( parameter_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除参数模板""" result = product_parameter.delete(db, parameter_id) if not result: return error_response(404, "参数不存在") return success_response({"message": "删除成功"})