diff --git a/backend/curve_client.py b/backend/curve_client.py
new file mode 100644
index 0000000..71e4de0
--- /dev/null
+++ b/backend/curve_client.py
@@ -0,0 +1,148 @@
+"""Luxsin 曲线 API：拉取、自定义 Base64 解码与 parametric_eq 校验。"""
+import base64
+import json
+import logging
+from typing import Any, Optional, Tuple
+from urllib.parse import urlencode
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+LUXSIN_CURVE_API_BASE = "https://api.luxsin.com.cn/audio/getCurve"
+
+CUSTOM_CHARS = "KLMPQRSTUVWXYZABCGHdefIJjkNOlmnopqrstuvwxyzabcghiDEF34501289+67/"
+STANDARD_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
+
+TARGET_OVER_EAR = "Harman over-ear 2018"
+TARGET_IN_EAR = "Harman in-ear 2019"
+
+FORM_TARGET_MAP = {
+ "over-ear": TARGET_OVER_EAR,
+ "in-ear": TARGET_IN_EAR,
+}
+
+
+def curve_target_for_form(form: Optional[str]) -> Optional[str]:
+ if not form:
+ return None
+ return FORM_TARGET_MAP.get(form.strip().lower())
+
+
+def custom_base64_to_string(encoded: str) -> str:
+ """将 Luxsin 自定义 Base64 字母表映射为标准 Base64 后解码为 UTF-8 字符串。"""
+ if not encoded or not isinstance(encoded, str):
+ raise ValueError("曲线数据为空")
+
+ standard_b64 = []
+ for c in encoded:
+ idx = CUSTOM_CHARS.find(c)
+ if idx != -1:
+ standard_b64.append(STANDARD_CHARS[idx])
+ else:
+ standard_b64.append(c)
+
+ try:
+ raw = base64.b64decode("".join(standard_b64), validate=False)
+ except Exception as e:
+ raise ValueError(f"Base64 解码失败：{e}") from e
+
+ try:
+ return raw.decode("utf-8")
+ except UnicodeDecodeError as e:
+ raise ValueError(f"UTF-8 解码失败：{e}") from e
+
+
+def is_valid_parametric_eq_payload(data: Any) -> bool:
+ if not isinstance(data, dict):
+ return False
+ peq = data.get("parametric_eq")
+ if not isinstance(peq, dict):
+ return False
+ filters = peq.get("filters")
+ return isinstance(filters, list) and len(filters) == 10
+
+
+def extract_encoded_payload(response: requests.Response) -> str:
+ """从 getCurve 响应中提取待解码字符串。"""
+ text = (response.text or "").strip()
+ if not text:
+ raise ValueError("曲线接口响应为空")
+
+ try:
+ body = response.json()
+ except ValueError:
+ return text
+
+ if isinstance(body, str):
+ return body.strip()
+
+ if not isinstance(body, dict):
+ raise ValueError("曲线接口响应格式异常")
+
+ for key in ("data", "curve", "result", "content", "body"):
+ val = body.get(key)
+ if isinstance(val, str) and val.strip():
+ return val.strip()
+
+ nested = body.get("data")
+ if isinstance(nested, dict):
+ for key in ("curve", "data", "content", "encoded"):
+ val = nested.get(key)
+ if isinstance(val, str) and val.strip():
+ return val.strip()
+
+ if isinstance(nested, str) and nested.strip():
+ return nested.strip()
+
+ raise ValueError("曲线接口响应中未找到可解码数据")
+
+
+def fetch_and_validate_curve(
+ brand: str,
+ name: str,
+ form: str,
+ *,
+ timeout: float = 20.0,
+) -> Tuple[bool, str]:
+ """
+ 拉取并校验曲线。返回 (是否通过, 失败原因)；通过时原因为空字符串。
+ """
+ brand = (brand or "").strip()
+ name = (name or "").strip()
+ if not brand or not name:
+ return False, "品牌名称或型号名称为空"
+
+ target = curve_target_for_form(form)
+ if not target:
+ return False, "佩戴方式须为入耳式(in-ear)或头戴式(over-ear)才能校验曲线"
+
+ query = urlencode({"brand": brand, "name": name, "target": target})
+ url = f"{LUXSIN_CURVE_API_BASE}?{query}"
+
+ try:
+ resp = requests.get(url, timeout=timeout)
+ except requests.RequestException as e:
+ logger.warning("getCurve request failed: %s %s", url, e)
+ return False, f"无法连接曲线接口：{e}"
+
+ if resp.status_code != 200:
+ return False, f"曲线接口返回 HTTP {resp.status_code}"
+
+ try:
+ encoded = extract_encoded_payload(resp)
+ except ValueError as e:
+ return False, str(e)
+
+ try:
+ decoded_text = custom_base64_to_string(encoded)
+ payload = json.loads(decoded_text)
+ except json.JSONDecodeError:
+ return False, "解码后的数据不是合法 JSON"
+ except ValueError as e:
+ return False, str(e)
+
+ if not is_valid_parametric_eq_payload(payload):
+ return False, "曲线数据异常"
+
+ return True, ""
diff --git a/backend/routes/brands.py b/backend/routes/brands.py
index 51e2931..02301e5 100644
--- a/backend/routes/brands.py
+++ b/backend/routes/brands.py
@@ -65,14 +65,16 @@ def get_brand(brand_id: int, db: Session = Depends(get_db)):
def create_brand(brand: BrandCreate, db: Session = Depends(get_db)):
"""创建新品牌"""
try:
- logger.info(f"Creating brand: name={brand.name}")
- # 检查是否已存在
- existing = db.query(Brand).filter(Brand.name == brand.name).first()
+ name = (brand.name or "").strip()
+ if not name:
+ return ApiResponse.error(msg="品牌名称不能为空", code=0)
+ logger.info(f"Creating brand: name={name}")
+ existing = db.query(Brand).filter(Brand.name == name).first()
if existing:
- logger.warning(f"Brand already exists: name={brand.name}")
+ logger.warning(f"Brand already exists: name={name}")
return ApiResponse.error(msg="品牌名称已存在", code=0)
-
- db_brand = Brand(name=brand.name)
+
+ db_brand = Brand(name=name)
db.add(db_brand)
db.commit()
db.refresh(db_brand)
@@ -88,19 +90,33 @@ def create_brand(brand: BrandCreate, db: Session = Depends(get_db)):
def update_brand(brand_id: int, brand: BrandUpdate, db: Session = Depends(get_db)):
"""更新品牌"""
try:
- logger.info(f"Updating brand: id={brand_id}, data={brand.dict()}")
+ logger.info(f"Updating brand: id={brand_id}, data={brand.model_dump()}")
db_brand = db.query(Brand).filter(Brand.id == brand_id).first()
if not db_brand:
logger.warning(f"Brand not found: id={brand_id}")
return ApiResponse.no_data(msg="品牌不存在")
-
- # 如果更新名称，检查是否冲突
- if brand.name and brand.name != db_brand.name:
- existing = db.query(Brand).filter(Brand.name == brand.name).first()
+
+ if brand.name is None:
+ db.commit()
+ db.refresh(db_brand)
+ return ApiResponse.success(data=db_brand.to_dict(), msg="品牌更新成功")
+
+ new_name = brand.name.strip()
+ if not new_name:
+ return ApiResponse.error(msg="品牌名称不能为空", code=0)
+
+ current_name = (db_brand.name or "").strip()
+ if new_name != current_name:
+ # 排除当前记录；MySQL 大小写不敏感时，仅改大小写也会命中自身
+ existing = (
+ db.query(Brand)
+ .filter(Brand.name == new_name, Brand.id != brand_id)
+ .first()
+ )
if existing:
- logger.warning(f"Brand name already exists: name={brand.name}")
+ logger.warning(f"Brand name already exists: name={new_name}")
return ApiResponse.error(msg="品牌名称已存在", code=0)
- db_brand.name = brand.name
+ db_brand.name = new_name
db.commit()
db.refresh(db_brand)
diff --git a/backend/routes/models.py b/backend/routes/models.py
index e5e9b46..cb8f151 100644
--- a/backend/routes/models.py
+++ b/backend/routes/models.py
@@ -1,4 +1,5 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Body, UploadFile, File, Form
+from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session
from sqlalchemy import asc, desc
from typing import List, Optional
@@ -6,6 +7,7 @@ import logging
from database import get_db
from models.model import Model
from response import ApiResponse, PageData
+from curve_client import fetch_and_validate_curve
import requests
import os
import shutil
@@ -16,6 +18,15 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/models", tags=["models"])
+
+class PushToSearchBody(BaseModel):
+ """推送 Meilisearch 请求体；字段名 model_ids 需关闭 protected_namespaces 避免 Pydantic 警告。"""
+
+ model_config = ConfigDict(protected_namespaces=())
+
+ model_ids: List[int] = Field(..., min_length=1, description="型号 ID 列表")
+
+
# Meilisearch 配置（从环境变量读取）
MEILISEARCH_URL = os.getenv("MEILISEARCH_URL", "http://localhost:7700")
MEILISEARCH_API_KEY = os.getenv("MEILISEARCH_API_KEY", "")
@@ -260,69 +271,120 @@ def delete_model(model_id: int, db: Session = Depends(get_db)):
return ApiResponse(code=0, msg="error", data=None)
-@router.post("/push-to-search", response_model=ApiResponse)
-def push_to_search(
- model_ids: List[int] = Body(..., embed=True, description="型号 ID 列表"),
- db: Session = Depends(get_db)
-):
- """推送型号数据到 Meilisearch"""
- try:
- if not model_ids:
- return ApiResponse(code=0, msg="请选择要推送的型号", data=None)
-
- # 查询选中的型号
- models = db.query(Model).filter(Model.id.in_(model_ids)).all()
-
- if not models:
- return ApiResponse(code=0, msg="未找到选中的型号数据", data=None)
-
- # 准备推送数据
- push_data = []
- for model in models:
- model_dict = {
- 'id': model.id,
- 'brand_name': model.brand_name,
- 'name': model.name,
- 'rig': model.rig,
- 'form': model.form,
- 'source': model.source
- }
- # 只包含有值的字段
- push_data.append({k: v for k, v in model_dict.items() if v is not None})
-
- # 推送到 Meilisearch
- headers = {
- "Authorization": f"Bearer {MEILISEARCH_API_KEY}",
- "Content-Type": "application/json"
- }
-
- # 如果索引不存在，先创建索引
- try:
- # 更新或添加文档
- response = requests.post(
- f"{MEILISEARCH_URL}/indexes/{MEILISEARCH_INDEX}/documents",
- json=push_data,
- headers=headers,
- timeout=30
- )
-
- if response.status_code not in [200, 202]:
- return ApiResponse(code=0, msg=f"推送到 Meilisearch 失败：{response.text}", data=None)
-
- task_info = response.json()
-
- return ApiResponse(
- code=1,
- msg="success",
- data={
- "pushed_count": len(push_data),
- "task_uid": task_info.get("taskUid"),
- "models": push_data
+def _get_models_by_ids(model_ids: List[int], db: Session):
+ if not model_ids:
+ return None, ApiResponse(code=0, msg="请选择要推送的型号", data=None)
+ models = db.query(Model).filter(Model.id.in_(model_ids)).all()
+ if not models:
+ return None, ApiResponse(code=0, msg="未找到选中的型号数据", data=None)
+ return models, None
+
+
+def _validate_models_curve(models) -> List[dict]:
+ validation_errors = []
+ for model in models:
+ ok, reason = fetch_and_validate_curve(
+ model.brand_name,
+ model.name,
+ model.form or "",
+ )
+ if not ok:
+ validation_errors.append(
+ {
+ "id": model.id,
+ "brand_name": model.brand_name,
+ "name": model.name,
+ "form": model.form,
+ "reason": reason,
}
)
-
- except requests.exceptions.RequestException as e:
- return ApiResponse(code=0, msg=f"连接 Meilisearch 失败：{str(e)}", data=None)
-
+ return validation_errors
+
+
+def _build_meilisearch_documents(models) -> List[dict]:
+ push_data = []
+ for model in models:
+ model_dict = {
+ "id": model.id,
+ "brand_name": model.brand_name,
+ "name": model.name,
+ "rig": model.rig,
+ "form": model.form,
+ "source": model.source,
+ }
+ push_data.append({k: v for k, v in model_dict.items() if v is not None})
+ return push_data
+
+
+def _push_documents_to_meilisearch(push_data: List[dict]) -> ApiResponse:
+ headers = {
+ "Authorization": f"Bearer {MEILISEARCH_API_KEY}",
+ "Content-Type": "application/json",
+ }
+ try:
+ response = requests.post(
+ f"{MEILISEARCH_URL}/indexes/{MEILISEARCH_INDEX}/documents",
+ json=push_data,
+ headers=headers,
+ timeout=30,
+ )
+ if response.status_code not in [200, 202]:
+ return ApiResponse(code=0, msg=f"推送到 Meilisearch 失败：{response.text}", data=None)
+
+ task_info = response.json()
+ return ApiResponse(
+ code=1,
+ msg="success",
+ data={
+ "pushed_count": len(push_data),
+ "task_uid": task_info.get("taskUid"),
+ "models": push_data,
+ },
+ )
+ except requests.exceptions.RequestException as e:
+ return ApiResponse(code=0, msg=f"连接 Meilisearch 失败：{str(e)}", data=None)
+
+
+@router.post("/push-to-search/validate", response_model=ApiResponse)
+def validate_push_to_search(body: PushToSearchBody, db: Session = Depends(get_db)):
+ """推送前校验：拉取并验证曲线 parametric_eq 数据"""
+ try:
+ models, err = _get_models_by_ids(body.model_ids, db)
+ if err:
+ return err
+
+ validation_errors = _validate_models_curve(models)
+ if validation_errors:
+ names = "、".join(
+ f"{e['brand_name']} {e['name']}" for e in validation_errors[:5]
+ )
+ suffix = " 等" if len(validation_errors) > 5 else ""
+ return ApiResponse(
+ code=0,
+ msg=f"曲线数据校验未通过：{names}{suffix}",
+ data={"errors": validation_errors, "validated_count": 0},
+ )
+
+ return ApiResponse(
+ code=1,
+ msg="success",
+ data={"validated_count": len(models)},
+ )
except Exception as e:
- return ApiResponse(code=0, msg="error", data=None)
+ logger.error("validate_push_to_search failed: %s", e, exc_info=True)
+ return ApiResponse(code=0, msg=f"校验失败：{str(e)}", data=None)
+
+
+@router.post("/push-to-search", response_model=ApiResponse)
+def push_to_search(body: PushToSearchBody, db: Session = Depends(get_db)):
+ """推送型号数据到 Meilisearch（需先通过 validate 接口）"""
+ try:
+ models, err = _get_models_by_ids(body.model_ids, db)
+ if err:
+ return err
+
+ push_data = _build_meilisearch_documents(models)
+ return _push_documents_to_meilisearch(push_data)
+ except Exception as e:
+ logger.error("push_to_search failed: %s", e, exc_info=True)
+ return ApiResponse(code=0, msg=f"推送失败：{str(e)}", data=None)
diff --git a/backend/test.py b/backend/test.py
deleted file mode 100644
index f8dd9a8..0000000
--- a/backend/test.py
+++ /dev/null
@@ -1,12 +0,0 @@
-from database import get_db
-from models import Brand
-
-if __name__ == "__main__":
- db_gen = get_db()
- db = next(db_gen)
- try:
- brands = db.query(Brand).all()
- for brand in brands:
- print(brand.name)
- finally:
- db_gen.close() # 这会触发 finally 块中的 db.close()
\ No newline at end of file
diff --git a/backend/test_api.py b/backend/test_api.py
deleted file mode 100644
index 2353c5c..0000000
--- a/backend/test_api.py
+++ /dev/null
@@ -1,17 +0,0 @@
-import requests
-import json
-
-try:
- response = requests.get("http://localhost:8001/api/brands/", params={"skip": 0, "limit": 10})
- print(f"Status Code: {response.status_code}")
- print(f"Content-Type: {response.headers.get('content-type')}")
- print(f"Response: {response.text[:500]}")
- if response.status_code == 200:
- data = response.json()
- print(f"\nSuccess! Got {len(data)} brands")
- for brand in data[:5]:
- print(f" - {brand}")
-except Exception as e:
- print(f"Error: {e}")
- import traceback
- traceback.print_exc()
diff --git a/backend/test_api_simple.py b/backend/test_api_simple.py
deleted file mode 100644
index 988103c..0000000
--- a/backend/test_api_simple.py
+++ /dev/null
@@ -1,52 +0,0 @@
-import requests
-import json
-
-BASE_URL = "http://localhost:8002"
-
-def test_api(endpoint, method="GET", data=None, params=None):
- """测试 API 接口"""
- url = f"{BASE_URL}{endpoint}"
- try:
- if method == "GET":
- response = requests.get(url, params=params)
- elif method == "POST":
- response = requests.post(url, json=data, params=params)
- elif method == "PUT":
- response = requests.put(url, json=data)
- elif method == "DELETE":
- response = requests.delete(url)
-
- result = response.json()
- print(f"\n{method} {endpoint}")
- print(f"Status: {response.status_code}")
- print(f"Response: {json.dumps(result, ensure_ascii=False, indent=2)}")
- return result
- except Exception as e:
- print(f"\n{method} {endpoint} - Error: {e}")
- import traceback
- traceback.print_exc()
- return None
-
-# 测试品牌接口
-print("=" * 60)
-print("测试品牌接口")
-print("=" * 60)
-
-# 1. 获取品牌列表（成功，有数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 5})
-
-# 2. 模糊查询品牌（成功，有数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 10, "name": "Audio"})
-
-# 3. 模糊查询品牌（无数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 10, "name": "不存在的品牌 XYZ"})
-
-# 4. 获取单个品牌（成功）
-test_api("/api/brands/1")
-
-# 5. 获取单个品牌（不存在）
-test_api("/api/brands/999999")
-
-print("\n" + "=" * 60)
-print("测试完成！")
-print("=" * 60)
diff --git a/backend/test_response.py b/backend/test_response.py
deleted file mode 100644
index e1ad996..0000000
--- a/backend/test_response.py
+++ /dev/null
@@ -1,65 +0,0 @@
-import requests
-import json
-
-BASE_URL = "http://localhost:8002"
-
-def test_api(endpoint, method="GET", data=None, params=None):
- """测试 API 接口"""
- url = f"{BASE_URL}{endpoint}"
- try:
- if method == "GET":
- response = requests.get(url, params=params)
- elif method == "POST":
- response = requests.post(url, json=data, params=params)
- elif method == "PUT":
- response = requests.put(url, json=data)
- elif method == "DELETE":
- response = requests.delete(url)
-
- result = response.json()
- print(f"\n{method} {endpoint}")
- print(f"Status: {response.status_code}")
- print(f"Response: {json.dumps(result, ensure_ascii=False, indent=2)}")
- return result
- except Exception as e:
- print(f"\n{method} {endpoint} - Error: {e}")
- return None
-
-# 测试品牌接口
-print("=" * 60)
-print("测试品牌接口")
-print("=" * 60)
-
-# 1. 获取品牌列表（成功，有数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 5})
-
-# 2. 模糊查询品牌（成功，有数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 10, "name": "Audio"})
-
-# 3. 模糊查询品牌（无数据）
-test_api("/api/brands/", params={"skip": 0, "limit": 10, "name": "不存在的品牌XYZ"})
-
-# 4. 获取单个品牌（成功）
-test_api("/api/brands/1")
-
-# 5. 获取单个品牌（不存在）
-test_api("/api/brands/999999")
-
-# 6. 创建品牌（成功）
-test_api("/api/brands/", method="POST", data={"name": "Test Brand 测试"})
-
-# 7. 创建品牌（重复）
-test_api("/api/brands/", method="POST", data={"name": "1MORE"})
-
-# 8. 更新品牌（成功）
-test_api("/api/brands/1", method="PUT", data={"name": "1MORE Updated"})
-
-# 9. 删除品牌（成功）
-test_api("/api/brands/623", method="DELETE")
-
-# 10. 删除品牌（不存在）
-test_api("/api/brands/999999", method="DELETE")
-
-print("\n" + "=" * 60)
-print("测试完成！")
-print("=" * 60)
diff --git a/frontend/index.html b/frontend/index.html
index da54b45..3821548 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -2,7 +2,7 @@
-
+
diff --git a/frontend/public/favicon.svg b/frontend/public/favicon.svg
new file mode 100644
index 0000000..f06ddef
--- /dev/null
+++ b/frontend/public/favicon.svg
@@ -0,0 +1,3 @@
+
diff --git a/frontend/src/api/model.js b/frontend/src/api/model.js
index ec90162..06d046e 100644
--- a/frontend/src/api/model.js
+++ b/frontend/src/api/model.js
@@ -72,13 +72,29 @@ export function deleteModel(id) {
}
/**
- * 推送到 Meilisearch
+ * 推送前校验曲线数据
+ * @param {Array} modelIds - 型号 ID 列表
+ */
+export function validatePushToMeilisearch(modelIds) {
+ return request({
+ url: '/models/push-to-search/validate',
+ method: 'post',
+ data: { model_ids: modelIds },
+ timeout: 120000,
+ skipErrorToast: true
+ })
+}
+
+/**
+ * 推送到 Meilisearch（需先通过校验）
* @param {Array} modelIds - 型号 ID 列表
*/
export function pushToMeilisearch(modelIds) {
return request({
url: '/models/push-to-search',
method: 'post',
- data: { model_ids: modelIds }
+ data: { model_ids: modelIds },
+ timeout: 60000,
+ skipErrorToast: true
})
}
diff --git a/frontend/src/utils/request.js b/frontend/src/utils/request.js
index 6506b66..da4ce5b 100644
--- a/frontend/src/utils/request.js
+++ b/frontend/src/utils/request.js
@@ -29,10 +29,14 @@ request.interceptors.response.use(
response => {
const res = response.data
- // 如果响应码为 0，表示错误
+ // 如果响应码为 0，表示错误（可由调用方自行提示，如推送进度弹窗）
if (res.code === 0) {
- ElMessage.error(res.msg || '请求失败')
- return Promise.reject(new Error(res.msg || '请求失败'))
+ if (!response.config.skipErrorToast) {
+ ElMessage.error(res.msg || '请求失败')
+ }
+ const err = new Error(res.msg || '请求失败')
+ err.responseData = res
+ return Promise.reject(err)
}
return res
diff --git a/frontend/src/views/model/index.vue b/frontend/src/views/model/index.vue
index 4c20d42..29c8feb 100644
--- a/frontend/src/views/model/index.vue
+++ b/frontend/src/views/model/index.vue
@@ -45,9 +45,8 @@